from flask import Flask, session, request, jsonify, make_response
from flask_session import Session
import os
from datetime import timedelta
from werkzeug.security import check_password_hash, generate_password_hash
import redis
import uuid

class SessionAuth:
    """基于Session的身份认证系统"""
    
    def __init__(self, app=None):
        self.app = app
        if app is not None:
            self.init_app(app)
    
    def init_app(self, app):
        """初始化Session配置"""
        # 配置Session
        app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'dev-secret-key')
        app.config['SESSION_TYPE'] = 'redis'  # 使用Redis存储Session
        app.config['SESSION_PERMANENT'] = True
        app.config['SESSION_USE_SIGNER'] = True  # 对Session ID进行签名
        app.config['SESSION_KEY_PREFIX'] = 'session:'
        app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=24)
        
        # 初始化Flask-Session
        Session(app)
        
        # 模拟用户数据库
        self.users = {
            'alice': {
                'password': generate_password_hash('password123'),
                'user_id': 1,
                'email': 'alice@example.com',
                'role': 'user'
            },
            'bob': {
                'password': generate_password_hash('secret456'),
                'user_id': 2,
                'email': 'bob@example.com',
                'role': 'admin'
            }
        }
        
        # 注册路由
        self.register_routes(app)
    
    def register_routes(self, app):
        """注册认证路由"""
        
        @app.route('/api/session/login', methods=['POST'])
        def login():
            """用户登录"""
            data = request.get_json()
            username = data.get('username')
            password = data.get('password')
            
            # 验证用户凭证
            user = self.authenticate_user(username, password)
            if not user:
                return jsonify({
                    'error': '无效的用户名或密码'
                }), 401
            
            # 创建Session
            session.permanent = True
            session['user_id'] = user['user_id']
            session['username'] = username
            session['role'] = user['role']
            session['logged_in'] = True
            
            return jsonify({
                'message': '登录成功',
                'user': {
                    'user_id': user['user_id'],
                    'username': username,
                    'email': user['email'],
                    'role': user['role']
                }
            })
        
        @app.route('/api/session/logout', methods=['POST'])
        def logout():
            """用户登出"""
            session.clear()
            return jsonify({'message': '登出成功'})
        
        @app.route('/api/session/profile')
        def get_profile():
            """获取用户资料（需要认证）"""
            if not session.get('logged_in'):
                return jsonify({'error': '未认证'}), 401
            
            return jsonify({
                'user_id': session['user_id'],
                'username': session['username'],
                'role': session['role']
            })
        
        @app.route('/api/session/refresh', methods=['POST'])
        def refresh_session():
            """刷新Session有效期"""
            if not session.get('logged_in'):
                return jsonify({'error': '未认证'}), 401
            
            # 在Flask-Session中，访问session会自动刷新过期时间
            session.modified = True
            
            return jsonify({'message': 'Session已刷新'})
    
    def authenticate_user(self, username, password):
        """验证用户凭证"""
        user = self.users.get(username)
        if user and check_password_hash(user['password'], password):
            return user
        return None

# 初始化Flask应用
def create_session_app():
    app = Flask(__name__)
    
    # 初始化Session认证
    session_auth = SessionAuth(app)
    
    return app

# 运行应用
if __name__ == '__main__':
    app = create_session_app()
    app.run(debug=True, port=5000)

import secrets
from flask import Flask
from datetime import timedelta

class SecureSessionConfig:
    """安全的Session配置"""
    
    @staticmethod
    def get_secure_config():
        """返回安全配置"""
        return {
            'SECRET_KEY': secrets.token_urlsafe(32),  # 强密钥
            'SESSION_COOKIE_HTTPONLY': True,  # 防止XSS读取Cookie
            'SESSION_COOKIE_SECURE': True,    # 仅HTTPS传输
            'SESSION_COOKIE_SAMESITE': 'Lax', # CSRF保护
            'PERMANENT_SESSION_LIFETIME': timedelta(hours=1),  # 合理过期时间
            'SESSION_REFRESH_EACH_REQUEST': True,  # 每次请求刷新
        }

# Session安全中间件
def session_security_middleware(app):
    """Session安全中间件"""
    
    @app.before_request
    def validate_session():
        """验证Session安全性"""
        # 检查User-Agent一致性
        current_ua = request.headers.get('User-Agent', '')
        stored_ua = session.get('user_agent')
        
        if session.get('logged_in'):
            if not stored_ua:
                # 首次登录存储User-Agent
                session['user_agent'] = current_ua
            elif stored_ua != current_ua:
                # User-Agent变化，可能是会话劫持
                session.clear()
                return jsonify({'error': '会话异常，请重新登录'}), 401
    
    @app.after_request
    def set_security_headers(response):
        """设置安全头部"""
        response.headers['X-Content-Type-Options'] = 'nosniff'
        response.headers['X-Frame-Options'] = 'DENY'
        response.headers['X-XSS-Protection'] = '1; mode=block'
        return response

JWT（JSON Web Token）是一种基于令牌的无状态认证机制。它将用户信息编码到令牌中，客户端在每次请求时携带该令牌。

JWT认证流程：

JWT由三部分组成，用点号分隔：

Header：令牌类型和签名算法
Payload：包含声明（用户信息等）
Signature：验证令牌完整性的签名

Header.Payload.Signature

import jwt
import datetime
from functools import wraps
from flask import Flask, request, jsonify
from werkzeug.security import generate_password_hash, check_password_hash
from typing import Dict, Optional, Union
import secrets

class JWTAuth:
    """基于JWT的身份认证系统"""
    
    def __init__(self, app=None):
        self.app = app
        self.algorithm = 'HS256'
        self.secret_key = secrets.token_urlsafe(32)
        
        # 令牌黑名单（用于实现登出）
        self.token_blacklist = set()
        
        # 模拟用户数据库
        self.users = {
            'alice': {
                'password': generate_password_hash('password123'),
                'user_id': 1,
                'email': 'alice@example.com',
                'role': 'user'
            },
            'bob': {
                'password': generate_password_hash('secret456'),
                'user_id': 2,
                'email': 'bob@example.com',
                'role': 'admin'
            }
        }
        
        if app is not None:
            self.init_app(app)
    
    def init_app(self, app):
        """初始化JWT配置"""
        self.app = app
        self.register_routes(app)
    
    def create_access_token(self, user_data: Dict, expires_delta: Optional[datetime.timedelta] = None) -> str:
        """创建访问令牌"""
        to_encode = user_data.copy()
        
        if expires_delta:
            expire = datetime.datetime.utcnow() + expires_delta
        else:
            expire = datetime.datetime.utcnow() + datetime.timedelta(hours=1)
        
        to_encode.update({
            'exp': expire,
            'iat': datetime.datetime.utcnow(),
            'type': 'access'
        })
        
        encoded_jwt = jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)
        return encoded_jwt
    
    def create_refresh_token(self, user_id: int) -> str:
        """创建刷新令牌"""
        expire = datetime.datetime.utcnow() + datetime.timedelta(days=30)
        
        refresh_data = {
            'user_id': user_id,
            'exp': expire,
            'iat': datetime.datetime.utcnow(),
            'type': 'refresh'
        }
        
        encoded_jwt = jwt.encode(refresh_data, self.secret_key, algorithm=self.algorithm)
        return encoded_jwt
    
    def verify_token(self, token: str) -> Optional[Dict]:
        """验证JWT令牌"""
        try:
            # 检查令牌是否在黑名单中
            if token in self.token_blacklist:
                return None
            
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
            return payload
        except jwt.ExpiredSignatureError:
            return None  # 令牌过期
        except jwt.InvalidTokenError:
            return None  # 无效令牌
    
    def authenticate_user(self, username: str, password: str) -> Optional[Dict]:
        """验证用户凭证"""
        user = self.users.get(username)
        if user and check_password_hash(user['password'], password):
            return {
                'user_id': user['user_id'],
                'username': username,
                'email': user['email'],
                'role': user['role']
            }
        return None
    
    def register_routes(self, app):
        """注册认证路由"""
        
        @app.route('/api/jwt/login', methods=['POST'])
        def login():
            """用户登录并获取JWT令牌"""
            data = request.get_json()
            username = data.get('username')
            password = data.get('password')
            
            # 验证用户凭证
            user_data = self.authenticate_user(username, password)
            if not user_data:
                return jsonify({
                    'error': '无效的用户名或密码'
                }), 401
            
            # 创建访问令牌和刷新令牌
            access_token = self.create_access_token(user_data)
            refresh_token = self.create_refresh_token(user_data['user_id'])
            
            return jsonify({
                'message': '登录成功',
                'access_token': access_token,
                'refresh_token': refresh_token,
                'token_type': 'bearer',
                'expires_in': 3600,  # 1小时
                'user': user_data
            })
        
        @app.route('/api/jwt/refresh', methods=['POST'])
        def refresh():
            """使用刷新令牌获取新的访问令牌"""
            data = request.get_json()
            refresh_token = data.get('refresh_token')
            
            if not refresh_token:
                return jsonify({'error': '刷新令牌必填'}), 400
            
            # 验证刷新令牌
            payload = self.verify_token(refresh_token)
            if not payload or payload.get('type') != 'refresh':
                return jsonify({'error': '无效的刷新令牌'}), 401
            
            # 获取用户信息
            user_id = payload['user_id']
            user = next((u for u in self.users.values() if u['user_id'] == user_id), None)
            if not user:
                return jsonify({'error': '用户不存在'}), 404
            
            # 创建新的访问令牌
            user_data = {
                'user_id': user['user_id'],
                'username': next(username for username, u in self.users.items() if u['user_id'] == user_id),
                'email': user['email'],
                'role': user['role']
            }
            
            new_access_token = self.create_access_token(user_data)
            
            return jsonify({
                'access_token': new_access_token,
                'token_type': 'bearer',
                'expires_in': 3600
            })
        
        @app.route('/api/jwt/logout', methods=['POST'])
        def logout():
            """用户登出（将令牌加入黑名单）"""
            auth_header = request.headers.get('Authorization', '')
            if auth_header.startswith('Bearer '):
                token = auth_header[7:]  # 移除'Bearer '前缀
                self.token_blacklist.add(token)
            
            return jsonify({'message': '登出成功'})
        
        @app.route('/api/jwt/profile')
        def get_profile():
            """获取用户资料（需要JWT认证）"""
            # 使用装饰器进行认证
            user_data = self.get_current_user()
            if not user_data:
                return jsonify({'error': '未认证或令牌无效'}), 401
            
            return jsonify({
                'user_id': user_data['user_id'],
                'username': user_data['username'],
                'email': user_data['email'],
                'role': user_data['role']
            })
    
    def get_current_user(self) -> Optional[Dict]:
        """从请求中获取当前用户"""
        auth_header = request.headers.get('Authorization', '')
        
        if not auth_header.startswith('Bearer '):
            return None
        
        token = auth_header[7:]  # 移除'Bearer '前缀
        payload = self.verify_token(token)
        
        return payload
    
    def jwt_required(self, f):
        """JWT认证装饰器"""
        @wraps(f)
        def decorated_function(*args, **kwargs):
            user_data = self.get_current_user()
            if not user_data:
                return jsonify({'error': '认证令牌无效或已过期'}), 401
            
            # 将用户信息添加到请求上下文
            request.current_user = user_data
            return f(*args, **kwargs)
        return decorated_function
    
    def role_required(self, role: str):
        """角色权限装饰器"""
        def decorator(f):
            @wraps(f)
            @self.jwt_required
            def decorated_function(*args, **kwargs):
                if request.current_user.get('role') != role:
                    return jsonify({'error': '权限不足'}), 403
                return f(*args, **kwargs)
            return decorated_function
        return decorator

# 初始化Flask应用
def create_jwt_app():
    app = Flask(__name__)
    
    # 初始化JWT认证
    jwt_auth = JWTAuth(app)
    
    # 添加需要特定权限的路由
    @app.route('/api/jwt/admin')
    @jwt_auth.role_required('admin')
    def admin_dashboard():
        """管理员仪表板（需要admin角色）"""
        return jsonify({
            'message': '欢迎来到管理员面板',
            'user': request.current_user
        })
    
    return app

# 运行应用
if __name__ == '__main__':
    app = create_jwt_app()
    app.run(debug=True, port=5001)

import hmac
import hashlib
from datetime import datetime, timedelta

class SecureJWTAuth(JWTAuth):
    """增强安全性的JWT实现"""
    
    def __init__(self, app=None):
        super().__init__(app)
        # 记录令牌指纹，防止重放攻击
        self.token_fingerprints = {}
    
    def create_secure_access_token(self, user_data: Dict, client_fingerprint: str) -> str:
        """创建安全的访问令牌"""
        # 生成令牌指纹
        token_fingerprint = self._generate_token_fingerprint(user_data['user_id'], client_fingerprint)
        
        # 将指纹加入用户数据
        user_data_with_fp = user_data.copy()
        user_data_with_fp['fp'] = token_fingerprint
        
        access_token = self.create_access_token(user_data_with_fp)
        
        # 存储令牌指纹
        self.token_fingerprints[token_fingerprint] = {
            'user_id': user_data['user_id'],
            'created_at': datetime.utcnow(),
            'last_used': datetime.utcnow()
        }
        
        return access_token
    
    def verify_secure_token(self, token: str, client_fingerprint: str) -> Optional[Dict]:
        """验证安全令牌"""
        payload = self.verify_token(token)
        if not payload:
            return None
        
        # 验证令牌指纹
        token_fp = payload.get('fp')
        if not token_fp or not self._validate_token_fingerprint(token_fp, client_fingerprint, payload['user_id']):
            return None
        
        # 更新最后使用时间
        if token_fp in self.token_fingerprints:
            self.token_fingerprints[token_fp]['last_used'] = datetime.utcnow()
        
        return payload
    
    def _generate_token_fingerprint(self, user_id: int, client_fingerprint: str) -> str:
        """生成令牌指纹"""
        data = f"{user_id}:{client_fingerprint}:{datetime.utcnow().timestamp()}"
        return hmac.new(
            self.secret_key.encode(),
            data.encode(),
            hashlib.sha256
        ).hexdigest()
    
    def _validate_token_fingerprint(self, token_fp: str, client_fingerprint: str, user_id: int) -> bool:
        """验证令牌指纹"""
        if token_fp not in self.token_fingerprints:
            return False
        
        fp_data = self.token_fingerprints[token_fp]
        
        # 检查指纹是否过期（24小时）
        if datetime.utcnow() - fp_data['created_at'] > timedelta(hours=24):
            del self.token_fingerprints[token_fp]
            return False
        
        # 验证用户ID匹配
        return fp_data['user_id'] == user_id

# JWT令牌自动清理任务
def start_token_cleanup_task(jwt_auth: SecureJWTAuth):
    """启动令牌清理任务"""
    import threading
    import time
    
    def cleanup_task():
        while True:
            time.sleep(3600)  # 每小时清理一次
            current_time = datetime.utcnow()
            
            # 清理过期的令牌指纹
            expired_fps = [
                fp for fp, data in jwt_auth.token_fingerprints.items()
                if current_time - data['created_at'] > timedelta(hours=24)
            ]
            
            for fp in expired_fps:
                del jwt_auth.token_fingerprints[fp]
            
            # 清理黑名单中的旧令牌（超过7天）
            # 注意：实际项目中黑名单应该使用Redis等有TTL的数据存储
    
    thread = threading.Thread(target=cleanup_task, daemon=True)
    thread.start()

特性维度 Session JWT

状态管理有状态（服务器存储）无状态（客户端存储）

扩展性需要共享Session存储天然支持水平扩展

性能每次请求需要查询Session存储仅需验证签名，性能更好

存储开销服务器存储开销大服务器无存储，但令牌体积较大

安全性易受CSRF攻击，需要额外防护内置签名验证，但需要防范XSS

移动端支持 Cookie在移动端支持有限原生支持移动端

实时吊销立即生效需要黑名单机制，无法立即吊销

特性维度	Session	JWT
状态管理	有状态（服务器存储）	无状态（客户端存储）
扩展性	需要共享Session存储	天然支持水平扩展
性能	每次请求需要查询Session存储	仅需验证签名，性能更好
存储开销	服务器存储开销大	服务器无存储，但令牌体积较大
安全性	易受CSRF攻击，需要额外防护	内置签名验证，但需要防范XSS
移动端支持	Cookie在移动端支持有限	原生支持移动端
实时吊销	立即生效	需要黑名单机制，无法立即吊销

Session存储成本模型：

C_session=N_users×S_session×R_access

其中：

N_users = 活跃用户数

S_session = 单个Session平均大小

R_access = 访问频率因子

JWT传输成本模型：C_jwt=N_requests×S_jwt×O_verification

其中：

N_requests = 请求数量

S_jwt = JWT令牌大小

O_verification = 验证操作复杂度

class HybridAuth:
    """混合认证策略：结合Session和JWT的优点"""
    
    def __init__(self, app=None):
        self.app = app
        self.session_auth = SessionAuth()
        self.jwt_auth = JWTAuth()
        
        if app is not None:
            self.init_app(app)
    
    def init_app(self, app):
        """初始化混合认证"""
        self.session_auth.init_app(app)
        self.jwt_auth.init_app(app)
        
        self.register_hybrid_routes(app)
    
    def register_hybrid_routes(self, app):
        """注册混合认证路由"""
        
        @app.route('/api/hybrid/login', methods=['POST'])
        def hybrid_login():
            """混合登录：创建Session和JWT"""
            data = request.get_json()
            username = data.get('username')
            password = data.get('password')
            
            # 验证用户凭证
            user = self.session_auth.authenticate_user(username, password)
            if not user:
                return jsonify({'error': '无效的凭证'}), 401
            
            # 创建Session（用于Web端）
            session.permanent = True
            session['user_id'] = user['user_id']
            session['username'] = username
            session['role'] = user['role']
            session['logged_in'] = True
            
            # 创建JWT（用于API/移动端）
            user_data = {
                'user_id': user['user_id'],
                'username': username,
                'email': user['email'],
                'role': user['role']
            }
            access_token = self.jwt_auth.create_access_token(user_data)
            refresh_token = self.jwt_auth.create_refresh_token(user['user_id'])
            
            response_data = {
                'message': '登录成功',
                'user': user_data,
                'session_available': True,
                'jwt_available': True,
                'access_token': access_token,
                'refresh_token': refresh_token
            }
            
            # 根据客户端类型返回不同响应
            user_agent = request.headers.get('User-Agent', '').lower()
            if 'mobile' in user_agent or request.args.get('client') == 'mobile':
                # 移动端：主要使用JWT
                return jsonify({
                    **response_data,
                    'primary_auth': 'jwt'
                })
            else:
                # Web端：主要使用Session
                return jsonify({
                    **response_data,
                    'primary_auth': 'session'
                })
        
        @app.route('/api/hybrid/profile')
        def hybrid_profile():
            """混合认证获取用户资料"""
            # 尝试Session认证
            if session.get('logged_in'):
                return jsonify({
                    'user_id': session['user_id'],
                    'username': session['username'],
                    'role': session['role'],
                    'auth_method': 'session'
                })
            
            # 尝试JWT认证
            user_data = self.jwt_auth.get_current_user()
            if user_data:
                return jsonify({
                    **user_data,
                    'auth_method': 'jwt'
                })
            
            return jsonify({'error': '未认证'}), 401
        
        @app.route('/api/hybrid/logout', methods=['POST'])
        def hybrid_logout():
            """混合登出：清除Session和JWT"""
            # 清除Session
            session.clear()
            
            # 将JWT加入黑名单
            auth_header = request.headers.get('Authorization', '')
            if auth_header.startswith('Bearer '):
                token = auth_header[7:]
                self.jwt_auth.token_blacklist.add(token)
            
            return jsonify({'message': '登出成功'})

def create_hybrid_app():
    """创建混合认证应用"""
    app = Flask(__name__)
    hybrid_auth = HybridAuth(app)
    return app

class SmartAuthRouter:
    """智能认证路由：根据请求特征选择认证方式"""
    
    def __init__(self, session_auth, jwt_auth):
        self.session_auth = session_auth
        self.jwt_auth = jwt_auth
    
    def authenticate_request(self, request):
        """智能认证请求"""
        # 分析请求特征
        features = self._analyze_request_features(request)
        
        # 根据特征选择认证策略
        if features['is_web_browser'] and not features['is_api_client']:
            # Web浏览器：优先使用Session
            return self._authenticate_with_session(request)
        else:
            # API客户端/移动端：优先使用JWT
            return self._authenticate_with_jwt(request)
    
    def _analyze_request_features(self, request):
        """分析请求特征"""
        user_agent = request.headers.get('User-Agent', '').lower()
        
        return {
            'is_web_browser': any(browser in user_agent for browser in 
                                ['chrome', 'firefox', 'safari', 'edge']),
            'is_api_client': 'api-client' in user_agent or 
                           request.headers.get('X-API-Client') == 'true',
            'has_session_cookie': 'session' in request.cookies,
            'has_auth_header': request.headers.get('Authorization', '').startswith('Bearer ')
        }
    
    def _authenticate_with_session(self, request):
        """使用Session认证"""
        if session.get('logged_in'):
            return {
                'user_id': session['user_id'],
                'username': session['username'],
                'role': session['role'],
                'auth_method': 'session'
            }
        return None
    
    def _authenticate_with_jwt(self, request):
        """使用JWT认证"""
        return self.jwt_auth.get_current_user()

import re
from flask import request, abort

class SecurityEnhancer:
    """安全增强器"""
    
    @staticmethod
    def validate_password_strength(password: str) -> bool:
        """验证密码强度"""
        if len(password) < 8:
            return False
        
        # 检查包含大写、小写、数字、特殊字符
        has_upper = bool(re.search(r'[A-Z]', password))
        has_lower = bool(re.search(r'[a-z]', password))
        has_digit = bool(re.search(r'd', password))
        has_special = bool(re.search(r'[!@#$%^&*(),.?":{}|<>]', password))
        
        # 至少满足其中三项
        return sum([has_upper, has_lower, has_digit, has_special]) >= 3
    
    @staticmethod
    def detect_brute_force(ip_address: str, max_attempts: int = 5, window_minutes: int = 15):
        """检测暴力破解攻击"""
        # 实际项目中应该使用Redis等外部存储
        import time
        current_time = time.time()
        window_seconds = window_minutes * 60
        
        # 清理过期的尝试记录
        SecurityEnhancer.failed_attempts = {
            ip: attempts for ip, attempts in SecurityEnhancer.failed_attempts.items()
            if current_time - attempts['last_attempt'] < window_seconds
        }
        
        if ip_address in SecurityEnhancer.failed_attempts:
            attempts = SecurityEnhancer.failed_attempts[ip_address]
            if attempts['count'] >= max_attempts:
                return True  # 触发暴力破解防护
        
        return False
    
    @staticmethod
    def record_failed_attempt(ip_address: str):
        """记录失败的登录尝试"""
        import time
        current_time = time.time()
        
        if ip_address not in SecurityEnhancer.failed_attempts:
            SecurityEnhancer.failed_attempts[ip_address] = {
                'count': 0,
                'last_attempt': current_time
            }
        
        SecurityEnhancer.failed_attempts[ip_address]['count'] += 1
        SecurityEnhancer.failed_attempts[ip_address]['last_attempt'] = current_time
    
    # 静态变量存储失败尝试
    failed_attempts = {}

class RateLimiter:
    """API速率限制器"""
    
    def __init__(self):
        self.requests = {}
    
    def is_rate_limited(self, identifier: str, max_requests: int, window_seconds: int) -> bool:
        """检查是否达到速率限制"""
        import time
        current_time = time.time()
        window_start = current_time - window_seconds
        
        # 清理过期的请求记录
        if identifier in self.requests:
            self.requests[identifier] = [
                req_time for req_time in self.requests[identifier]
                if req_time > window_start
            ]
        
        # 检查请求次数
        if identifier not in self.requests:
            self.requests[identifier] = []
        
        if len(self.requests[identifier]) >= max_requests:
            return True
        
        # 记录当前请求
        self.requests[identifier].append(current_time)
        return False

# 使用示例
rate_limiter = RateLimiter()

def apply_rate_limit(f):
    """应用速率限制的装饰器"""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        client_ip = request.remote_addr
        if rate_limiter.is_rate_limited(client_ip, max_requests=100, window_seconds=3600):
            return jsonify({'error': '请求过于频繁，请稍后重试'}), 429
        return f(*args, **kwargs)
    return decorated_function

import os
from datetime import timedelta

class ProductionSecurityConfig:
    """生产环境安全配置"""
    
    @staticmethod
    def get_session_config():
        """生产环境Session配置"""
        return {
            'SESSION_TYPE': 'redis',
            'SESSION_REDIS': 'redis://localhost:6379/0',
            'SESSION_USE_SIGNER': True,
            'SESSION_PERMANENT': True,
            'SESSION_KEY_PREFIX': 'prod_session:',
            'PERMANENT_SESSION_LIFETIME': timedelta(hours=2),
            'SESSION_COOKIE_SECURE': True,
            'SESSION_COOKIE_HTTPONLY': True,
            'SESSION_COOKIE_SAMESITE': 'Lax'
        }
    
    @staticmethod
    def get_jwt_config():
        """生产环境JWT配置"""
        return {
            'JWT_SECRET_KEY': os.environ['JWT_SECRET_KEY'],  # 从环境变量获取
            'JWT_ALGORITHM': 'HS256',
            'JWT_ACCESS_TOKEN_EXPIRES': timedelta(hours=1),
            'JWT_REFRESH_TOKEN_EXPIRES': timedelta(days=30),
            'JWT_BLACKLIST_ENABLED': True,
            'JWT_BLACKLIST_TOKEN_CHECKS': ['access', 'refresh']
        }
    
    @staticmethod
    def get_redis_config():
        """Redis配置（用于Session和JWT黑名单）"""
        return {
            'host': os.environ.get('REDIS_HOST', 'localhost'),
            'port': int(os.environ.get('REDIS_PORT', 6379)),
            'db': 0,
            'password': os.environ.get('REDIS_PASSWORD'),
            'decode_responses': True
        }

性能测试框架

import time
import asyncio
import statistics
from concurrent.futures import ThreadPoolExecutor

class AuthPerformanceTester:
    """认证性能测试器"""
    
    def __init__(self, base_url, num_requests=1000, concurrency=10):
        self.base_url = base_url
        self.num_requests = num_requests
        self.concurrency = concurrency
        self.results = {
            'session': [],
            'jwt': []
        }
    
    def test_session_auth(self):
        """测试Session认证性能"""
        import requests
        
        # 先登录获取Session
        session = requests.Session()
        login_response = session.post(
            f'{self.base_url}/api/session/login',
            json={'username': 'alice', 'password': 'password123'}
        )
        
        if login_response.status_code != 200:
            print("Session登录失败")
            return
        
        # 测试认证请求性能
        def make_request():
            start_time = time.time()
            response = session.get(f'{self.base_url}/api/session/profile')
            end_time = time.time()
            return end_time - start_time
        
        times = self._run_concurrent_requests(make_request)
        self.results['session'] = times
    
    def test_jwt_auth(self):
        """测试JWT认证性能"""
        import requests
        
        # 先登录获取JWT
        login_response = requests.post(
            f'{self.base_url}/api/jwt/login',
            json={'username': 'alice', 'password': 'password123'}
        )
        
        if login_response.status_code != 200:
            print("JWT登录失败")
            return
        
        token_data = login_response.json()
        access_token = token_data['access_token']
        
        headers = {'Authorization': f'Bearer {access_token}'}
        
        # 测试认证请求性能
        def make_request():
            start_time = time.time()
            response = requests.get(
                f'{self.base_url}/api/jwt/profile',
                headers=headers
            )
            end_time = time.time()
            return end_time - start_time
        
        times = self._run_concurrent_requests(make_request)
        self.results['jwt'] = times
    
    def _run_concurrent_requests(self, request_func):
        """并发执行请求"""
        with ThreadPoolExecutor(max_workers=self.concurrency) as executor:
            futures = [
                executor.submit(request_func) 
                for _ in range(self.num_requests)
            ]
            times = [future.result() for future in futures]
        return times
    
    def generate_report(self):
        """生成性能测试报告"""
        print("=" * 50)
        print("认证机制性能测试报告")
        print("=" * 50)
        
        for auth_type in ['session', 'jwt']:
            times = self.results[auth_type]
            if not times:
                continue
                
            avg_time = statistics.mean(times) * 1000  # 转换为毫秒
            min_time = min(times) * 1000
            max_time = max(times) * 1000
            std_dev = statistics.stdev(times) * 1000 if len(times) > 1 else 0
            
            print(f"n{auth_type.upper()} 认证:")
            print(f"  平均响应时间: {avg_time:.2f}ms")
            print(f"  最小响应时间: {min_time:.2f}ms")
            print(f"  最大响应时间: {max_time:.2f}ms")
            print(f"  标准差: {std_dev:.2f}ms")
            print(f"  请求总数: {len(times)}")
        
        # 性能对比
        if self.results['session'] and self.results['jwt']:
            session_avg = statistics.mean(self.results['session']) * 1000
            jwt_avg = statistics.mean(self.results['jwt']) * 1000
            
            if session_avg > 0:
                improvement = ((session_avg - jwt_avg) / session_avg) * 100
                print(f"n性能对比:")
                print(f"  JWT相比Session性能提升: {improvement:.1f}%")

# 使用示例
if __name__ == '__main__':
    tester = AuthPerformanceTester('http://localhost:5000', num_requests=100)
    tester.test_session_auth()
    tester.test_jwt_auth()
    tester.generate_report()

通用安全措施：

✅ 强制使用HTTPS

✅ 实施强密码策略

✅ 登录失败次数限制

✅ 会话超时设置

✅ 安全头部配置

Session安全：

✅ HttpOnly和Secure Cookie标志

✅ SameSite Cookie策略

✅ Session固定攻击防护

✅ 定期轮换Session密钥

JWT安全：

✅ 使用强密钥和合适算法

✅ 设置合理的令牌过期时间

✅ 实现令牌吊销机制

✅ 防范重放攻击

Session优化：

使用Redis等内存存储替代数据库

实现Session数据压缩

设置合理的Session清理策略

JWT优化：

控制Payload大小，避免存储敏感信息

使用高效的签名算法

实现分布式黑名单管理

Session和JWT都是成熟的认证方案，没有绝对的优劣，只有适合特定场景的选择：

选择Session：当你需要严格的会话控制、实时吊销能力，且应用架构相对简单时

选择JWT：当你构建无状态API、需要水平扩展、或支持多种客户端时

考虑混合方案：当你的应用需要同时支持Web和移动端时

关键决策因素：

架构需求：单体应用vs微服务

安全要求：实时控制vs性能优先

客户端类型：纯Web vs 多客户端

团队经验：技术栈熟悉程度

无论选择哪种方案，安全永远是第一位的。建议在生产环境中进行充分的安全测试和性能基准测试，确保认证系统既安全又高效。

记住，技术选型不是一次性的决定。随着业务的发展和技术环境的变化，定期重新评估你的认证方案，确保它仍然是最适合当前需求的选择。

以上就是Python中两大Web身份认证主流方案的实现指南的详细内容，更多关于Python身份认证的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

Python Flask token身份认证的示例代码(附完整代码)

python调用腾讯云实名认证接口辨别身份证真假

python实现身份证实名认证的方法实例

华夏福临门盛世加强版怎么样？产品责任是重中之重

国寿福庆典版值得买吗？附投保建议

理财中低风险是几级？中低风险理财会亏吗

瑞泰泰安心防癌险优缺点是什么？这份分析不容错过

2020最新自助免费申请Office365教育版，免费5TOneDrive云盘详细图文教程

MacBook怎么绕过BootCamp安装Win10双系统？

国内知名的前端博客

域名为什么会被墙？如何检测域名是否被墙？域名被墙如何处理？

最新的windows xp sp3产品密匙 xp sp3序列号 xp sp3激活码

Python中两大Web身份认证主流方案的实现指南

目录

身份认证（Authentication）是确认用户身份的过程，即验证"你是谁"。与之相关的概念是授权（Authorization），它决定"你能做什么"。在Web应用中，认证机制确保只有合法用户能够访问受保护的资源。

Session是基于服务器端状态的认证机制。其核心思想是在服务器存储用户状态，通过Cookie将Session ID传递给客户端。 Session认证流程：

JWT（JSON Web Token）是一种基于令牌的无状态认证机制。它将用户信息编码到令牌中，客户端在每次请求时携带该令牌。 JWT认证流程：

JWT由三部分组成，用点号分隔： Header：令牌类型和签名算法 Payload：包含声明（用户信息等） Signature：验证令牌完整性的签名 Header.Payload.Signature

Session优化： 使用Redis等内存存储替代数据库 实现Session数据压缩 设置合理的Session清理策略 JWT优化： 控制Payload大小，避免存储敏感信息 使用高效的签名算法 实现分布式黑名单管理

More posts

站内搜索

标签云

热门文章

友情链接

Session是基于服务器端状态的认证机制。其核心思想是在服务器存储用户状态，通过Cookie将Session ID传递给客户端。

Session认证流程：

JWT（JSON Web Token）是一种基于令牌的无状态认证机制。它将用户信息编码到令牌中，客户端在每次请求时携带该令牌。

JWT认证流程：

JWT由三部分组成，用点号分隔：

Header：令牌类型和签名算法

Payload：包含声明（用户信息等）

Signature：验证令牌完整性的签名

Header.Payload.Signature

Session优化：

使用Redis等内存存储替代数据库

实现Session数据压缩

设置合理的Session清理策略

JWT优化：

控制Payload大小，避免存储敏感信息

使用高效的签名算法

实现分布式黑名单管理