本文目录导读：
我来从多个维度详细介绍Python接口安全加固的实战方案。
身份认证层加固
JWT令牌认证
import jwt
from datetime import datetime, timedelta
from functools import wraps
from flask import request, jsonify
SECRET_KEY = "your-secret-key-here"
def generate_token(user_id, role):
payload = {
'user_id': user_id,
'role': role,
'exp': datetime.utcnow() + timedelta(hours=2), # 短有效期
'iat': datetime.utcnow(),
'jti': str(uuid.uuid4()) # 唯一标识，防止重放
}
return jwt.encode(payload, SECRET_KEY, algorithm='HS256')
def token_required(f):
@wraps(f)
def decorated(*args, **kwargs):
token = request.headers.get('Authorization')
if not token or not token.startswith('Bearer '):
return jsonify({'error': 'Missing or invalid token'}), 401
try:
token = token.split(' ')[1]
payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
# 检查token是否被撤销（需要在Redis中维护黑名单）
if check_token_revoked(payload['jti']):
return jsonify({'error': 'Token revoked'}), 401
request.current_user = payload
except jwt.ExpiredSignatureError:
return jsonify({'error': 'Token expired'}), 401
except jwt.InvalidTokenError:
return jsonify({'error': 'Invalid token'}), 401
return f(*args, **kwargs)
return decorated
@app.route('/api/secure/data', methods=['GET'])
@token_required
def get_secure_data():
return jsonify({'data': '敏感数据'})
基于角色的访问控制(RBAC)
def role_required(allowed_roles):
def decorator(f):
@wraps(f)
def decorated(*args, **kwargs):
if request.current_user.get('role') not in allowed_roles:
return jsonify({'error': 'Insufficient permissions'}), 403
return f(*args, **kwargs)
return decorated
return decorator
@app.route('/api/admin/users', methods=['GET'])
@token_required
@role_required(['admin', 'superadmin'])
def get_users():
return jsonify({'users': []})
请求验证与过滤
输入验证
from pydantic import BaseModel, Field, validator
from flask_pydantic import validate
class UserCreateRequest(BaseModel):
username: str = Field(..., min_length=3, max_length=50, regex='^[a-zA-Z0-9_]+$')
email: str = Field(..., regex='^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
age: int = Field(..., ge=0, le=150)
@validator('username')
def username_must_be_valid(cls, v):
if v.lower() in ['admin', 'root', 'system']:
raise ValueError('Username not allowed')
return v
@app.route('/api/users', methods=['POST'])
@validate(body=UserCreateRequest)
def create_user():
data = request.body_params
# 安全的创建用户逻辑
return jsonify({'message': 'User created'}), 201
SQL注入防护
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
# 错误示例 - 不要这样做
@app.route('/api/users/search')
def search_users_vulnerable():
keyword = request.args.get('q', '')
# 危险：直接拼接SQL
query = f"SELECT * FROM users WHERE name LIKE '%{keyword}%'"
result = db.session.execute(query)
return jsonify({'results': [r.to_dict() for r in result]})
# 正确示例 - 参数化查询
@app.route('/api/users/search')
def search_users_secure():
keyword = request.args.get('q', '')
# 安全：使用参数化查询
query = text("SELECT * FROM users WHERE name LIKE :keyword")
result = db.session.execute(query, {'keyword': f'%{keyword}%'})
return jsonify({'results': [r.to_dict() for r in result]})
通信安全
HTTPS强制和HSTS
from flask_talisman import Talisman
app = Flask(__name__)
# 配置安全头
csp = {
'default-src': "'self'",
'script-src': "'self' 'unsafe-inline'",
'style-src': "'self' 'unsafe-inline'",
}
talisman = Talisman(
app,
force_https=True, # 强制HTTPS
strict_transport_security=True, # HSTS
strict_transport_security_preload=True,
session_cookie_secure=True,
content_security_policy=csp,
feature_policy="camera 'none'; microphone 'none'"
)
API请求加密
from cryptography.fernet import Fernet
import base64
class RequestEncryption:
def __init__(self):
# 生产环境应从密钥管理系统获取
self.key = Fernet.generate_key()
self.cipher = Fernet(self.key)
def encrypt_payload(self, data: dict) -> str:
json_str = json.dumps(data)
encrypted = self.cipher.encrypt(json_str.encode())
return base64.urlsafe_b64encode(encrypted).decode()
def decrypt_payload(self, encrypted_data: str) -> dict:
try:
encrypted = base64.urlsafe_b64decode(encrypted_data)
decrypted = self.cipher.decrypt(encrypted)
return json.loads(decrypted.decode())
except Exception as e:
raise ValueError("解密失败")
@app.route('/api/secure/transfer', methods=['POST'])
def secure_transfer():
encrypted_data = request.json.get('encrypted_data')
try:
decrypted = RequestEncryption().decrypt_payload(encrypted_data)
# 处理解密后的数据
return jsonify({'status': 'success'})
except ValueError:
return jsonify({'error': 'Invalid encrypted data'}), 400
限流与防护
API速率限制
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
limiter = Limiter(
app,
key_func=get_remote_address,
default_limits=["200 per day", "50 per hour"]
)
# 针对不同端点设置不同限制
@app.route("/api/login")
@limiter.limit("5 per minute", override_defaults=False)
def login():
return "Login endpoint"
@app.route("/api/search")
@limiter.limit("30 per minute")
def search():
return "Search endpoint"
请求签名验证
import hmac
import hashlib
def verify_request_signature(request_data, signature, secret_key):
"""HMAC签名验证"""
# 将请求参数按字母序排列
sorted_params = sorted(request_data.items())
message = '&'.join([f"{k}={v}" for k, v in sorted_params])
# 计算签名
expected_signature = hmac.new(
secret_key.encode(),
message.encode(),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(signature, expected_signature)
@app.route('/api/webhook', methods=['POST'])
def webhook_handler():
signature = request.headers.get('X-Signature')
if not verify_request_signature(request.json, signature, WEBHOOK_SECRET):
return jsonify({'error': 'Invalid signature'}), 401
# 处理webhook
return jsonify({'status': 'ok'})
安全日志与监控
import logging
from flask import g, request
import json
# 配置安全审计日志
audit_logger = logging.getLogger('audit')
audit_logger.setLevel(logging.INFO)
handler = logging.FileHandler('audit.log')
handler.setFormatter(logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
))
audit_logger.addHandler(handler)
@app.before_request
def log_request():
g.start_time = time.time()
# 记录敏感操作
if request.method in ['POST', 'PUT', 'DELETE']:
audit_logger.info(json.dumps({
'ip': request.remote_addr,
'method': request.method,
'path': request.path,
'user_agent': request.user_agent.string,
'timestamp': datetime.utcnow().isoformat()
}))
@app.after_request
def log_response(response):
duration = time.time() - g.start_time
# 记录错误响应
if response.status_code >= 400:
audit_logger.warning(json.dumps({
'ip': request.remote_addr,
'path': request.path,
'status_code': response.status_code,
'duration': duration
}))
return response
敏感信息保护
import re
class SensitiveDataFilter:
@staticmethod
def mask_sensitive_data(data: dict) -> dict:
"""脱敏敏感数据"""
masked = data.copy()
if 'password' in masked:
masked['password'] = '********'
if 'credit_card' in masked:
# 保留后4位
card = masked['credit_card']
masked['credit_card'] = f"****-****-****-{card[-4:]}"
if 'email' in masked:
email = masked['email']
parts = email.split('@')
masked['email'] = f"{parts[0][:2]}***@{parts[1]}"
if 'phone' in masked:
phone = masked['phone']
masked['phone'] = f"{phone[:3]}****{phone[-4:]}"
return masked
# 响应数据脱敏
@app.after_request
def filter_sensitive_data(response):
if response.is_json and response.status_code == 200:
data = response.get_json()
if isinstance(data, dict):
filtered = SensitiveDataFilter.mask_sensitive_data(data)
response.set_data(json.dumps(filtered))
return response
依赖安全
# requirements.txt 安全最佳实践
"""
# 固定版本号
Flask==2.3.2
SQLAlchemy==2.0.19
PyJWT==2.8.0
# 安全扫描工具
safety==2.3.5
bandit==1.7.5
"""
# 安全扫描脚本
import subprocess
def security_check():
"""运行安全检查"""
print("Running safety check...")
subprocess.run(["safety", "check", "--full-report"])
print("Running bandit check...")
subprocess.run(["bandit", "-r", "."])
print("Checking for vulnerable dependencies...")
subprocess.run(["pip-audit"])
实施建议
- 分层防护：不要依赖单一的安全机制
- 最小权限：只授予必要的权限
- 持续监控：实时监控异常行为
- 定期审计：定期审查安全日志和代码
- 及时更新：保持依赖库和框架的最新版本
- 安全测试：定期进行渗透测试和安全评估
方案需要根据实际业务场景和合规要求进行调整。
