当前位置：首页 > article >正文

学习Flask：[特殊字符] Day 5：认证与授权

article 2025/3/7 3:01:03

学习目标：实现用户系统

from flask_jwt_extended import create_access_token, jwt_required

@app.route('/login', methods=['POST'])
def login():
    # 验证用户逻辑
    access_token = create_access_token(identity=user.id)
    return {'access_token': access_token}

@app.route('/protected')
@jwt_required()
def protected():
    return {'message': '登录用户可见'}

✅ 实践任务：

安装JWT扩展：pip install flask-jwt-extended
实现用户登录/注册接口
添加路由保护装饰器
实现基于角色的访问控制（RBAC）

要在 Flask 应用中使用 JWT（JSON Web Tokens）进行身份验证，你可以使用 Flask-JWT-Extended 扩展。以下是安装和基本使用的步骤：

1. 安装 Flask-JWT-Extended

在终端中运行以下命令以安装 Flask-JWT-Extended：

pip install Flask-JWT-Extended

2. 更新 app.py 文件

接下来，更新 app.py 文件以集成 JWT 身份验证。以下是一个示例，展示如何使用 Flask-JWT-Extended 进行用户注册和登录：

# app.py
from flask import Flask, jsonify, render_template, request
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_restful import Api, Resource
from marshmallow import Schema, fields, ValidationError
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity

app = Flask(__name__)

# 配置数据库
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'  # 使用 SQLite 数据库
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JWT_SECRET_KEY'] = 'your_jwt_secret_key'  # 设置 JWT 密钥
db = SQLAlchemy(app)
migrate = Migrate(app, db)  # 初始化 Flask-Migrate
api = Api(app)  # 初始化 Flask-RESTful
jwt = JWTManager(app)  # 初始化 Flask-JWT-Extended

# 定义用户模型
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password = db.Column(db.String(120), nullable=False)

# 定义 Marshmallow 验证模式
class UserSchema(Schema):
    username = fields.String(required=True, validate=lambda s: len(s) > 0)
    password = fields.String(required=True, validate=lambda s: len(s) > 0)

user_schema = UserSchema()

# 定义用户资源
class UserResource(Resource):
    def post(self):
        try:
            data = user_schema.load(request.get_json())  # 验证请求数据
        except ValidationError as err:
            return {'errors': err.messages}, 400  # 返回验证错误

        new_user = User(username=data['username'], password=data['password'])
        db.session.add(new_user)
        db.session.commit()
        return {'message': 'User created', 'id': new_user.id}, 201

# 定义登录资源
class LoginResource(Resource):
    def post(self):
        data = request.get_json()
        username = data.get('username')
        password = data.get('password')

        user = User.query.filter_by(username=username, password=password).first()
        if user:
            access_token = create_access_token(identity=user.id)
            return {'access_token': access_token}, 200
        return {'message': 'Invalid credentials'}, 401

# 添加 API 路由
api.add_resource(UserResource, '/api/users', endpoint='user_list')
api.add_resource(LoginResource, '/api/login', endpoint='login')

@app.route('/')
def home():
    return render_template('home.html')

if __name__ == '__main__':
    app.run(debug=True)

说明：

安装 Flask-JWT-Extended：使用 pip install Flask-JWT-Extended 安装 JWT 扩展。

配置 JWT 密钥：在 app.config 中设置 JWT_SECRET_KEY，这是用于加密和解密 JWT 的密钥。

初始化 JWT：使用 jwt = JWTManager(app) 初始化 JWT 扩展。

4. 用户注册：在 UserResource 中，处理用户注册。

用户登录：创建 LoginResource 类，处理用户登录并生成 JWT 访问令牌。

添加 API 路由：将登录资源添加到 API 路由中。

3. 测试 JWT 身份验证

你可以使用 Postman 测试 JWT 身份验证：

1. 创建用户（POST）

请求类型：POST

URL：http://127.0.0.1:5000/api/users

请求体：

    {
        "username": "新用户",
        "password": "密码"
    }

2. 登录用户（POST）

请求类型：POST

URL：http://127.0.0.1:5000/api/login

请求体：

    {
        "username": "新用户",
        "password": "密码"
    }

成功登录后，你将收到一个包含 access_token 的响应。

4. 使用 JWT 保护其他 API 端点

你可以使用 @jwt_required() 装饰器来保护其他 API 端点，确保只有持有有效 JWT 的用户才能访问。例如：

class ProtectedResource(Resource):
    @jwt_required()
    def get(self):
        current_user_id = get_jwt_identity()  # 获取当前用户 ID
        return {'message': f'Hello, user {current_user_id}'}, 200

api.add_resource(ProtectedResource, '/api/protected', endpoint='protected')

5. 运行应用

确保你的项目结构正确，然后在终端中运行 python app.py。你现在可以使用 JWT 进行身份验证。

要实现用户登录和注册接口，我们将使用 Flask-JWT-Extended 来处理 JWT 身份验证。以下是如何实现用户注册和登录接口的步骤：

1. 更新 app.py 文件

以下是更新后的 app.py 文件，包含用户注册和登录的实现：

# app.py
from flask import Flask, jsonify, render_template, request, redirect, url_for
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_restful import Api, Resource
from marshmallow import Schema, fields, ValidationError
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity

app = Flask(__name__)

# 配置数据库
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'  # 使用 SQLite 数据库
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JWT_SECRET_KEY'] = 'your_jwt_secret_key'  # 设置 JWT 密钥
db = SQLAlchemy(app)
migrate = Migrate(app, db)  # 初始化 Flask-Migrate
api = Api(app)  # 初始化 Flask-RESTful
jwt = JWTManager(app)  # 初始化 Flask-JWT-Extended

# 定义用户模型
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password = db.Column(db.String(120), nullable=False)
    posts = db.relationship('Post', backref='author', lazy=True)  # 一对多关系

# 定义文章模型
class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(120), nullable=False)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)  # 外键

# 定义 Marshmallow 验证模式
class UserSchema(Schema):
    username = fields.String(required=True, validate=lambda s: len(s) > 0)
    password = fields.String(required=True, validate=lambda s: len(s) > 0)

user_schema = UserSchema()

# 定义用户注册资源
class UserRegister(Resource):
    def post(self):
        try:
            data = user_schema.load(request.get_json())  # 验证请求数据
        except ValidationError as err:
            return {'errors': err.messages}, 400  # 返回验证错误

        # 检查用户名是否已存在
        if User.query.filter_by(username=data['username']).first():
            return {'message': 'Username already exists'}, 400

        new_user = User(username=data['username'], password=data['password'])
        db.session.add(new_user)
        db.session.commit()
        return {'message': 'User created', 'id': new_user.id}, 201

# 定义用户登录资源
class UserLogin(Resource):
    def post(self):
        data = request.get_json()
        username = data.get('username')
        password = data.get('password')

        user = User.query.filter_by(username=username, password=password).first()
        if user:
            access_token = create_access_token(identity=user.id)
            return {'access_token': access_token}, 200
        return {'message': 'Invalid credentials'}, 401

# 添加 API 路由
api.add_resource(UserRegister, '/api/user/register', endpoint='user_register')
api.add_resource(UserLogin, '/api/login', endpoint='login')

@app.route('/')
def home():
    users = User.query.all()  # 从数据库中获取所有用户
    return render_template('home.html', users=users)

@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        
        # 检查用户名是否已存在
        existing_user = User.query.filter_by(username=username).first()
        if existing_user:
            return "用户名已存在，请选择其他用户名。", 400
        
        new_user = User(username=username, password=password)
        db.session.add(new_user)  # 添加新用户到数据库
        db.session.commit()  # 提交更改
        return redirect(url_for('home'))  # 注册成功后重定向到首页
    
    return render_template('register.html')

@app.route('/profile')
def profile():
    return render_template('profile.html')

@app.route('/api')
def api():
    return jsonify({"message": "这是API端点。"})

@app.route('/post', methods=['GET', 'POST'])
def create_post():
    if request.method == 'POST':
        title = request.form['title']
        content = request.form['content']
        user_id = request.form['user_id']  # 假设你从表单中获取用户ID
        
        new_post = Post(title=title, content=content, user_id=user_id)
        db.session.add(new_post)
        db.session.commit()
        return redirect(url_for('home'))
    
    users = User.query.all()  # 获取所有用户以供选择
    return render_template('create_post.html', users=users)

if __name__ == '__main__':
    app.run(debug=True)

说明：

1.用户模型：定义了 User 模型，包含 id、username 和 password 字段。

2. 用户注册资源：

UserRegister 类处理用户注册。

在 post 方法中，验证请求数据，检查用户名是否已存在。

如果用户名可用，创建新用户并保存到数据库。

3.用户登录资源：

UserLogin 类处理用户登录。
在 post 方法中，验证用户名和密码。

如果验证成功，生成 JWT 访问令牌并返回。

添加 API 路由：

将注册和登录资源添加到 API 路由中。

2. 测试用户注册和登录接口

你可以使用 Postman 测试这些接口：

1. 用户注册（POST）

请求类型：POST

URL：http://127.0.0.1:5000/api/user/register

请求体：

    {
        "username": "新用户",
        "password": "密码"
    }

2. 用户登录（POST）

请求类型：POST

URL：http://127.0.0.1:5000/api/login

请求体：

    {
        "username": "新用户",
        "password": "密码"
    }

成功登录后，你将收到一个包含 access_token 的响应。

3. 运行应用

确保你的项目结构正确，然后在终端中运行 python app.py。你现在可以使用用户注册和登录接口。

要为 Flask 应用中的特定路由添加保护装饰器，以确保只有经过身份验证的用户可以访问这些路由，你可以使用 Flask-JWT-Extended 提供的 @jwt_required() 装饰器。以下是如何实现路由保护的步骤：

1. 更新 app.py 文件

在你的 app.py 文件中，添加一个受保护的资源示例，使用 @jwt_required() 装饰器来保护该路由。以下是更新后的代码示例：

# app.py
from flask import Flask, jsonify, render_template, request, redirect, url_for
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_restful import Api, Resource
from marshmallow import Schema, fields, ValidationError
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity

app = Flask(__name__)

# 配置数据库
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'  # 使用 SQLite 数据库
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JWT_SECRET_KEY'] = 'your_jwt_secret_key'  # 设置 JWT 密钥
db = SQLAlchemy(app)
migrate = Migrate(app, db)  # 初始化 Flask-Migrate
api = Api(app)  # 初始化 Flask-RESTful
jwt = JWTManager(app)  # 初始化 Flask-JWT-Extended

# 定义用户模型
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password = db.Column(db.String(120), nullable=False)
    posts = db.relationship('Post', backref='author', lazy=True)  # 一对多关系

# 定义文章模型
class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(120), nullable=False)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)  # 外键

# 定义 Marshmallow 验证模式
class UserSchema(Schema):
    username = fields.String(required=True, validate=lambda s: len(s) > 0)
    password = fields.String(required=True, validate=lambda s: len(s) > 0)

user_schema = UserSchema()

# 定义用户注册资源
class UserRegister(Resource):
    def post(self):
        try:
            data = user_schema.load(request.get_json())  # 验证请求数据
        except ValidationError as err:
            return {'errors': err.messages}, 400  # 返回验证错误

        # 检查用户名是否已存在
        if User.query.filter_by(username=data['username']).first():
            return {'message': 'Username already exists'}, 400

        new_user = User(username=data['username'], password=data['password'])
        db.session.add(new_user)
        db.session.commit()
        return {'message': 'User created', 'id': new_user.id}, 201

# 定义用户登录资源
class UserLogin(Resource):
    def post(self):
        data = request.get_json()
        username = data.get('username')
        password = data.get('password')

        user = User.query.filter_by(username=username, password=password).first()
        if user:
            access_token = create_access_token(identity=user.id)
            return {'access_token': access_token}, 200
        return {'message': 'Invalid credentials'}, 401

# 定义受保护的资源
class ProtectedResource(Resource):
    @jwt_required()  # 保护此路由
    def get(self):
        current_user_id = get_jwt_identity()  # 获取当前用户 ID
        return {'message': f'Hello, user {current_user_id}'}, 200

# 添加 API 路由
api.add_resource(UserRegister, '/api/user/register', endpoint='user_register')
api.add_resource(UserLogin, '/api/login', endpoint='login')
api.add_resource(ProtectedResource, '/api/protected', endpoint='protected')  # 添加受保护的路由

@app.route('/')
def home():
    users = User.query.all()  # 从数据库中获取所有用户
    return render_template('home.html', users=users)

@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        
        # 检查用户名是否已存在
        existing_user = User.query.filter_by(username=username).first()
        if existing_user:
            return "用户名已存在，请选择其他用户名。", 400
        
        new_user = User(username=username, password=password)
        db.session.add(new_user)  # 添加新用户到数据库
        db.session.commit()  # 提交更改
        return redirect(url_for('home'))  # 注册成功后重定向到首页
    
    return render_template('register.html')

@app.route('/profile')
def profile():
    return render_template('profile.html')

@app.route('/api')
def api():
    return jsonify({"message": "这是API端点。"})

@app.route('/post', methods=['GET', 'POST'])
def create_post():
    if request.method == 'POST':
        title = request.form['title']
        content = request.form['content']
        user_id = request.form['user_id']  # 假设你从表单中获取用户ID
        
        new_post = Post(title=title, content=content, user_id=user_id)
        db.session.add(new_post)
        db.session.commit()
        return redirect(url_for('home'))
    
    users = User.query.all()  # 获取所有用户以供选择
    return render_template('create_post.html', users=users)

if __name__ == '__main__':
    app.run(debug=True)

说明：

1.定义受保护的资源：

创建 ProtectedResource 类，使用 @jwt_required() 装饰器保护 get 方法。

只有持有有效 JWT 的用户才能访问此路由。

2. 获取当前用户 ID：

使用 get_jwt_identity() 函数获取当前用户的 ID。

3. 添加 API 路由：

将受保护的资源添加到 API 路由中，使用 /api/protected 路由。

测试受保护的路由

1.你可以使用 Postman 测试受保护的路由：

登录用户（POST）以获取 JWT 访问令牌。

2. 访问受保护的资源（GET）：

在请求头中添加 Authorization，值为 Bearer your_jwt_access_token。

检查响应：如果访问成功，你将收到包含用户 ID 的消息。

通过以上步骤，你可以成功添加受保护的资源并使用 JWT 进行身份验证。

要实现基于角色的访问控制（RBAC）在 Flask 应用中，你可以使用 Flask-JWT-Extended 来管理用户的角色，并根据角色来控制对特定路由的访问。以下是实现 RBAC 的步骤：

1. 更新用户模型

首先，你需要在用户模型中添加一个角色字段。可以使用字符串来表示角色，例如 "admin"、"user" 等。

# 定义用户模型
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password = db.Column(db.String(120), nullable=False)
    role = db.Column(db.String(20), default='user')  # 添加角色字段
    posts = db.relationship('Post', backref='author', lazy=True)  # 一对多关系

2. 更新用户注册逻辑

在用户注册时，可以为用户分配一个角色。以下是更新后的用户注册资源：

class UserRegister(Resource):
    def post(self):
        try:
            data = user_schema.load(request.get_json())  # 验证请求数据
        except ValidationError as err:
            return {'errors': err.messages}, 400  # 返回验证错误

        # 检查用户名是否已存在
        if User.query.filter_by(username=data['username']).first():
            return {'message': 'Username already exists'}, 400

        # 默认角色为 'user'
        new_user = User(username=data['username'], password=data['password'], role='user')
        db.session.add(new_user)
        db.session.commit()
        return {'message': 'User created', 'id': new_user.id}, 201

3. 创建角色检查装饰器

创建一个装饰器，用于检查用户的角色。以下是一个示例：

from flask_jwt_extended import jwt_required, get_jwt_identity

def role_required(role):
    def wrapper(fn):
        @jwt_required()
        def decorated(*args, **kwargs):
            current_user_id = get_jwt_identity()
            user = User.query.get(current_user_id)
            if user.role != role:
                return {'msg': 'Access denied'}, 403  # 403 Forbidden
            return fn(*args, **kwargs)
        return decorated
    return wrapper

4. 使用角色检查装饰器保护路由

现在你可以使用 @role_required 装饰器来保护特定的路由。例如，创建一个仅限管理员访问的受保护资源：

class AdminResource(Resource):
    @role_required('admin')  # 仅限管理员访问
    def get(self):
        return {'msg': 'Welcome, admin!'}, 200

# 添加 API 路由
api.add_resource(AdminResource, '/api/admin', endpoint='admin')

5. 测试 RBAC

1. 创建用户

使用用户注册接口创建一个普通用户（角色为 "user"）。

使用用户注册接口创建一个管理员用户（角色为 "admin"），可以在注册时手动设置角色。

2. 登录用户

使用登录接口获取 JWT 令牌。

3. 测试访问控制

使用普通用户的 JWT 令牌访问 /api/admin 路由，应该返回 403 Forbidden。

使用管理员用户的 JWT 令牌访问 /api/admin 路由，应该返回 200 OK。

完整示例

以下是更新后的 app.py 文件的部分示例：

# app.py
from flask import Flask, jsonify, render_template, request, redirect, url_for
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_restful import Api, Resource
from marshmallow import Schema, fields, ValidationError
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity

app = Flask(__name__)

# 配置数据库
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JWT_SECRET_KEY'] = 'your_jwt_secret_key'
db = SQLAlchemy(app)
migrate = Migrate(app, db)
api = Api(app)
jwt = JWTManager(app)

# 定义用户模型
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password = db.Column(db.String(120), nullable=False)
    role = db.Column(db.String(20), default='user')  # 添加角色字段

# 定义用户注册资源
class UserRegister(Resource):
    def post(self):
        # ...（注册逻辑）
        new_user = User(username=data['username'], password=data['password'], role='user')
        # ...（保存用户）

# 创建角色检查装饰器
def role_required(role):
    def wrapper(fn):
        @jwt_required()
        def decorated(*args, **kwargs):
            current_user_id = get_jwt_identity()
            user = User.query.get(current_user_id)
            if user.role != role:
                return {'msg': 'Access denied'}, 403
            return fn(*args, **kwargs)
        return decorated
    return wrapper

# 定义管理员资源
class AdminResource(Resource):
    @role_required('admin')
    def get(self):
        return {'msg': 'Welcome, admin!'}, 200

# 添加 API 路由
api.add_resource(UserRegister, '/api/user/register', endpoint='user_register')
api.add_resource(AdminResource, '/api/admin', endpoint='admin')

if __name__ == '__main__':
    app.run(debug=True)