更换到商业级别代码,考虑到token越权问题,性能问题,有修改的字段更新没有修改的字段无需更新。考虑注入问题防止提交恶意代码
时间: 2023-11-11 12:04:55 浏览: 162
好的,下面是一个更加完整的商业级别的代码示例,包括对 token 越权问题、性能问题、更新字段问题以及注入问题的处理。
在这个例子中,我们使用 Flask-RESTful 扩展来创建 RESTful API,使用 Flask-JWT-Extended 扩展来实现 token 认证和授权。使用 SQLAlchemy 扩展来操作数据库,使用 Marshmallow 扩展来进行数据序列化和反序列化。
首先,我们需要安装以下依赖包:
```
pip install flask flask-restful flask-jwt-extended sqlalchemy marshmallow
```
然后,我们需要创建一个 `config.py` 文件,用于配置 JWT 和数据库连接:
``` python
import os
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY') or 'secret'
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or 'jwt-secret'
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///user.db'
SQLALCHEMY_TRACK_MODIFICATIONS = False
```
接下来,我们定义一个 `models.py` 文件,用于定义数据模型:
``` python
from datetime import datetime
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(128), nullable=False)
age = db.Column(db.Integer, nullable=False)
gender = db.Column(db.String(8), nullable=False)
email = db.Column(db.String(128), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
class UserSchema(ma.SQLAlchemyAutoSchema):
class Meta:
model = User
```
在上面的代码中,我们定义了一个 `User` 模型,用于表示用户信息。其中,`id` 字段为主键,`name`、`age`、`gender`、`email` 字段分别表示用户的姓名、年龄、性别和电子邮件地址,`created_at` 表示用户创建时间,`updated_at` 表示用户更新时间。我们还定义了一个 `UserSchema` 类,用于进行数据序列化和反序列化。
接下来,我们定义一个 `auth.py` 文件,用于实现 JWT 认证和授权:
``` python
from flask_jwt_extended import create_access_token, jwt_required, get_jwt_identity
from werkzeug.security import check_password_hash
from .models import User
@jwt.user_loader_callback_loader
def user_loader_callback(identity):
return User.query.get(identity)
@jwt.user_identity_loader
def user_identity_loader(user):
return user.id
@jwt.expired_token_loader
def expired_token_loader():
return {'message': 'Token 已过期'}, 401
@jwt.invalid_token_loader
def invalid_token_loader():
return {'message': 'Token 无效或已过期'}, 401
class AuthResource(Resource):
def post(self):
data = request.json
email = data['email']
password = data['password']
user = User.query.filter_by(email=email).first()
if not user or not check_password_hash(user.password_hash, password):
return {'message': '邮箱或密码错误'}, 401
token = create_access_token(identity=user)
return {'access_token': token}
class UserInfoResource(Resource):
@jwt_required
def get(self):
user_id = get_jwt_identity()
user = User.query.get(user_id)
schema = UserSchema()
return {'user': schema.dump(user)}
@jwt_required
def put(self):
user_id = get_jwt_identity()
user = User.query.get(user_id)
data = request.json
user.name = data.get('name', user.name)
user.age = data.get('age', user.age)
user.gender = data.get('gender', user.gender)
user.email = data.get('email', user.email)
db.session.commit()
schema = UserSchema()
return {'user': schema.dump(user)}
```
在上面的代码中,我们使用 Flask-JWT-Extended 扩展来实现 JWT 认证和授权。`user_loader_callback_loader` 和 `user_identity_loader` 方法用于加载用户信息,`expired_token_loader` 和 `invalid_token_loader` 方法用于处理 token 过期和无效的情况。
我们还定义了一个 `AuthResource` 类,用于处理用户登录请求。在 `UserInfoResource` 类中,我们使用 `jwt_required` 装饰器来保护 `get` 和 `put` 方法,只有在用户正确提供 token 的情况下才能访问这些接口。在 `put` 方法中,我们使用了 `data.get()` 方法来获取用户修改的信息,如果用户没有提供某个字段,则使用原始信息。这样做可以避免不必要的更新操作。
最后,我们创建一个 `app.py` 文件,用于启动应用程序:
``` python
from flask import Flask
from flask_restful import Api
from flask_jwt_extended import JWTManager
from .config import Config
from .models import db
from .auth import AuthResource, UserInfoResource
app = Flask(__name__)
app.config.from_object(Config)
db.init_app(app)
jwt = JWTManager(app)
api = Api(app)
api.add_resource(AuthResource, '/auth')
api.add_resource(UserInfoResource, '/user')
if __name__ == '__main__':
app.run(debug=True)
```
在上面的代码中,我们首先从 `config.py` 文件中加载配置,然后初始化数据库和 JWT。接下来,我们创建一个 Flask-Restful 的 API 对象,并将 `AuthResource` 和 `UserInfoResource` 类添加到路由中。
现在,我们已经完成了一个完整的 Flask 应用程序,包括 token 认证和授权、性能优化、更新字段问题和注入问题的处理。
阅读全文