文章

Flask - 用戶認證基礎

目標

  • 安裝並配置 Flask-HTTPAuth
  • 實現基本的用戶名/密碼認證
  • 保護 API 端點

步驟

  1. 準備環境

    • 繼續使用 flask_api/ 項目結構,激活虛擬環境:
      1
      2
      
      # Windows: flask_api_env\Scripts\activate
      # macOS/Linux: source flask_api_env/bin/activate
      
    • 安裝 Flask-HTTPAuth:
      1
      
      pip install flask-httpauth
      
  2. 更新模型

    • 修改 app/models.py,為 User 添加密碼字段(暫時明文存儲,後續會加密):

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      
      from . import db
      from datetime import datetime
      
      class User(db.Model):
          __tablename__ = 'users'
          id = db.Column(db.Integer, primary_key=True)
          username = db.Column(db.String(50), unique=True, nullable=False)
          password = db.Column(db.String(50), nullable=False)  # 新增密碼字段
          posts = db.relationship('Post', backref='user', lazy=True)
      
      class Post(db.Model):
          __tablename__ = 'posts'
          id = db.Column(db.Integer, primary_key=True)
          title = db.Column(db.String(100), nullable=False)
          content = db.Column(db.Text, nullable=False)
          created_at = db.Column(db.DateTime, default=datetime.utcnow)
          user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
          category = db.Column(db.String(50), default='general')
      
  3. 配置認證

    • 修改 app/init.py,初始化 Flask-HTTPAuth:

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      39
      40
      41
      42
      43
      44
      45
      46
      47
      48
      49
      50
      51
      52
      53
      54
      
      from flask import Flask, jsonify
      from flask_sqlalchemy import SQLAlchemy
      from flask_marshmallow import Marshmallow
      from flask_httpauth import HTTPBasicAuth  # 新增
      from .routes.v1.todos import todos_bp as todos_v1_bp
      from .routes.v1.users import users_bp as users_v1_bp
      from .routes.v1.posts import posts_bp as posts_v1_bp
      from .routes.v2.todos import todos_bp as todos_v2_bp
      from .routes.v2.posts import posts_bp as posts_v2_bp
      from .config import config_map
      import os
      
      db = SQLAlchemy()
      ma = Marshmallow()
      auth = HTTPBasicAuth()  # 初始化認證
      
      @auth.verify_password
      def verify_password(username, password):
          from .models import User  # 避免循環導入
          user = User.query.filter_by(username=username).first()
          if user and user.password == password:  # 簡單明文比較
              return user
          return None
      
      def create_app():
          app = Flask(__name__)
          env = os.getenv('FLASK_ENV', 'development')
          app.config.from_object(config_map[env])
          db.init_app(app)
          ma.init_app(app)
          auth.init_app(app)  # 初始化認證
      
          app.register_blueprint(todos_v1_bp, url_prefix='/api/v1')
          app.register_blueprint(users_v1_bp, url_prefix='/api/v1')
          app.register_blueprint(posts_v1_bp, url_prefix='/api/v1')
          app.register_blueprint(todos_v2_bp, url_prefix='/api/v2')
          app.register_blueprint(posts_v2_bp, url_prefix='/api/v2')
      
          @app.errorhandler(404)
          def not_found(error):
              return jsonify({'error': 'Not Found', 'message': str(error)}), 404
      
          @app.errorhandler(400)
          def bad_request(error):
              return jsonify({'error': 'Bad Request', 'message': str(error)}), 400
      
          @app.errorhandler(500)
          def internal_error(error):
              return jsonify({'error': 'Internal Server Error', 'message': 'Something went wrong on our end'}), 500
      
          with app.app_context():
              db.create_all()
      
          return app
      
  4. 保護路由

    • 修改 app/routes/v1/posts.py,添加認證:

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      39
      40
      41
      42
      43
      44
      45
      46
      47
      48
      49
      50
      51
      52
      53
      54
      55
      56
      57
      58
      59
      60
      61
      62
      63
      
      from flask import Blueprint, jsonify, request, abort
      from ...models import Post, User
      from ... import db, auth
      from ...schemas import post_schema, posts_schema
      
      posts_bp = Blueprint('posts_v1', __name__)
      
      @posts_bp.route('/posts', methods=['GET'])
      def get_posts():
          user_id = request.args.get('user_id', type=int)
          query = Post.query
          if user_id:
              query = query.filter_by(user_id=user_id)
          posts = query.all()
          return jsonify({'posts': posts_schema.dump(posts)})
      
      @posts_bp.route('/posts/<int:post_id>', methods=['GET'])
      def get_post(post_id):
          post = Post.query.get_or_404(post_id, description='Post not found')
          return jsonify(post_schema.dump(post))
      
      @posts_bp.route('/posts', methods=['POST'])
      @auth.login_required  # 需要認證
      def create_post():
          if not request.is_json:
              abort(400, description='Request must be JSON')
          data = request.get_json()
          if 'title' not in data or 'content' not in data:
              abort(400, description='Missing title or content')
          post = Post(
              title=data['title'],
              content=data['content'],
              user_id=auth.current_user().id  # 使用當前認證用戶的 ID
          )
          db.session.add(post)
          db.session.commit()
          return jsonify(post_schema.dump(post)), 201
      
      @posts_bp.route('/posts/<int:post_id>', methods=['PUT'])
      @auth.login_required
      def update_post(post_id):
          post = Post.query.get_or_404(post_id, description='Post not found')
          if post.user_id != auth.current_user().id:
              abort(403, description='You can only edit your own posts')
          if not request.is_json:
              abort(400, description='Request must be JSON')
          data = request.get_json()
          if 'title' in data:
              post.title = data['title']
          if 'content' in data:
              post.content = data['content']
          db.session.commit()
          return jsonify(post_schema.dump(post)), 200
      
      @posts_bp.route('/posts/<int:post_id>', methods=['DELETE'])
      @auth.login_required
      def delete_post(post_id):
          post = Post.query.get_or_404(post_id, description='Post not found')
          if post.user_id != auth.current_user().id:
              abort(403, description='You can only delete your own posts')
          db.session.delete(post)
          db.session.commit()
          return jsonify({'message': 'Post deleted'}), 200
      
    • 修改 app/routes/v1/users.py,支持密碼:

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      
      from flask import Blueprint, jsonify, request, abort
      from ...models import User
      from ... import db
      from ...schemas import user_schema, users_schema
      
      users_bp = Blueprint('users_v1', __name__)
      
      @users_bp.route('/users', methods=['GET'])
      def get_users():
          users = User.query.all()
          return jsonify({'users': users_schema.dump(users)})
      
      @users_bp.route('/users', methods=['POST'])
      def create_user():
          if not request.is_json:
              abort(400, description='Request must be JSON')
          data = request.get_json()
          if 'username' not in data or 'password' not in data:
              abort(400, description='Missing username or password')
          if User.query.filter_by(username=data['username']).first():
              abort(400, description='Username already exists')
          user = User(username=data['username'], password=data['password'])
          db.session.add(user)
          db.session.commit()
          return jsonify(user_schema.dump(user)), 201
      
  5. 更新序列化器

    • 修改 app/schemas.py,隱藏密碼:

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      
      from . import ma
      from .models import User, Post
      
      class PostSchema(ma.SQLAlchemyAutoSchema):
          class Meta:
              model = Post
              include_fk = True
              fields = ('id', 'title', 'content', 'created_at', 'user_id', 'category')
      
      class UserSchema(ma.SQLAlchemyAutoSchema):
          class Meta:
              model = User
              fields = ('id', 'username', 'posts')  # 排除 password
          posts = ma.Nested('PostSchema', many=True)
      
      post_schema = PostSchema()
      posts_schema = PostSchema(many=True)
      user_schema = UserSchema()
      users_schema = UserSchema(many=True)
      
  6. 運行應用

    • 刪除舊的 blog.db(因表結構改變),運行:
      1
      
      python run.py
      
  7. 測試 API

    • 使用 Postman 測試:
      • POST /api/v1/users
        • Body:{"username": "alice", "password": "1234"}
        • 預期響應:{"id": 1, "username": "alice", "posts": []}
      • POST /api/v1/posts(未認證):
        • Body:{"title": "My Post", "content": "Hello"}
        • 預期響應:401 Unauthorized
      • POST /api/v1/posts(認證):
        • Headers:Authorization: Basic YWxpY2U6MTIzNA==(Base64 編碼 “alice:1234”)
        • Body:{"title": "My Post", "content": "Hello"}
        • 預期響應:{"id": 1, "title": "My Post", ...}
      • PUT /api/v1/posts/1(錯誤用戶):
        • 用另一用戶(例如 “bob:5678”)認證,應返回 403。
      • DELETE /api/v1/posts/1(正確用戶):
        • 用 “alice:1234” 認證,應返回 {"message": "Post deleted"}
  8. 作業

    • 在 v2 的 posts 路由中添加認證,要求與 v1 一致。
    • 添加一個端點 GET /api/v1/me,返回當前認證用戶的信息。

注意事項

  • 目前密碼是明文存儲,下一天會引入加密。
  • 在 Postman 中設置 Authorization 時,選擇 “Basic Auth” 並輸入用戶名和密碼即可自動生成標頭。
  • 表結構改變後需重建數據庫。

本文章以 CC BY 4.0 授權