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的身份認(rèn)證系統(tǒng)"""
    
    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存儲(chǔ)Session
        app.config['SESSION_PERMANENT'] = True
        app.config['SESSION_USE_SIGNER'] = True  # 對(duì)Session ID進(jìn)行簽名
        app.config['SESSION_KEY_PREFIX'] = 'session:'
        app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(hours=24)
        
        # 初始化Flask-Session
        Session(app)
        
        # 模擬用戶數(shù)據(jù)庫(kù)
        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'
            }
        }
        
        # 注冊(cè)路由
        self.register_routes(app)
    
    def register_routes(self, app):
        """注冊(cè)認(rèn)證路由"""
        
        @app.route('/api/session/login', methods=['POST'])
        def login():
            """用戶登錄"""
            data = request.get_json()
            username = data.get('username')
            password = data.get('password')
            
            # 驗(yàn)證用戶憑證
            user = self.authenticate_user(username, password)
            if not user:
                return jsonify({
                    'error': '無(wú)效的用戶名或密碼'
                }), 401
            
            # 創(chuàng)建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():
            """獲取用戶資料（需要認(rèn)證）"""
            if not session.get('logged_in'):
                return jsonify({'error': '未認(rèn)證'}), 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': '未認(rèn)證'}), 401
            
            # 在Flask-Session中，訪問(wèn)session會(huì)自動(dòng)刷新過(guò)期時(shí)間
            session.modified = True
            
            return jsonify({'message': 'Session已刷新'})
    
    def authenticate_user(self, username, password):
        """驗(yàn)證用戶憑證"""
        user = self.users.get(username)
        if user and check_password_hash(user['password'], password):
            return user
        return None

# 初始化Flask應(yīng)用
def create_session_app():
    app = Flask(__name__)
    
    # 初始化Session認(rèn)證
    session_auth = SessionAuth(app)
    
    return app

# 運(yùn)行應(yīng)用
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),  # 強(qiáng)密鑰
            'SESSION_COOKIE_HTTPONLY': True,  # 防止XSS讀取Cookie
            'SESSION_COOKIE_SECURE': True,    # 僅HTTPS傳輸
            'SESSION_COOKIE_SAMESITE': 'Lax', # CSRF保護(hù)
            'PERMANENT_SESSION_LIFETIME': timedelta(hours=1),  # 合理過(guò)期時(shí)間
            'SESSION_REFRESH_EACH_REQUEST': True,  # 每次請(qǐng)求刷新
        }

# Session安全中間件
def session_security_middleware(app):
    """Session安全中間件"""
    
    @app.before_request
    def validate_session():
        """驗(yàn)證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:
                # 首次登錄存儲(chǔ)User-Agent
                session['user_agent'] = current_ua
            elif stored_ua != current_ua:
                # User-Agent變化，可能是會(huì)話劫持
                session.clear()
                return jsonify({'error': '會(huì)話異常，請(qǐng)重新登錄'}), 401
    
    @app.after_request
    def set_security_headers(response):
        """設(shè)置安全頭部"""
        response.headers['X-Content-Type-Options'] = 'nosniff'
        response.headers['X-Frame-Options'] = 'DENY'
        response.headers['X-XSS-Protection'] = '1; mode=block'
        return response

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的身份認(rèn)證系統(tǒng)"""
    
    def __init__(self, app=None):
        self.app = app
        self.algorithm = 'HS256'
        self.secret_key = secrets.token_urlsafe(32)
        
        # 令牌黑名單（用于實(shí)現(xiàn)登出）
        self.token_blacklist = set()
        
        # 模擬用戶數(shù)據(jù)庫(kù)
        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:
        """創(chuàng)建訪問(wèn)令牌"""
        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:
        """創(chuàng)建刷新令牌"""
        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]:
        """驗(yàn)證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  # 令牌過(guò)期
        except jwt.InvalidTokenError:
            return None  # 無(wú)效令牌
    
    def authenticate_user(self, username: str, password: str) -> Optional[Dict]:
        """驗(yàn)證用戶憑證"""
        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):
        """注冊(cè)認(rèn)證路由"""
        
        @app.route('/api/jwt/login', methods=['POST'])
        def login():
            """用戶登錄并獲取JWT令牌"""
            data = request.get_json()
            username = data.get('username')
            password = data.get('password')
            
            # 驗(yàn)證用戶憑證
            user_data = self.authenticate_user(username, password)
            if not user_data:
                return jsonify({
                    'error': '無(wú)效的用戶名或密碼'
                }), 401
            
            # 創(chuàng)建訪問(wèn)令牌和刷新令牌
            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小時(shí)
                'user': user_data
            })
        
        @app.route('/api/jwt/refresh', methods=['POST'])
        def refresh():
            """使用刷新令牌獲取新的訪問(wèn)令牌"""
            data = request.get_json()
            refresh_token = data.get('refresh_token')
            
            if not refresh_token:
                return jsonify({'error': '刷新令牌必填'}), 400
            
            # 驗(yàn)證刷新令牌
            payload = self.verify_token(refresh_token)
            if not payload or payload.get('type') != 'refresh':
                return jsonify({'error': '無(wú)效的刷新令牌'}), 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
            
            # 創(chuàng)建新的訪問(wèn)令牌
            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認(rèn)證）"""
            # 使用裝飾器進(jìn)行認(rèn)證
            user_data = self.get_current_user()
            if not user_data:
                return jsonify({'error': '未認(rèn)證或令牌無(wú)效'}), 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]:
        """從請(qǐng)求中獲取當(dāng)前用戶"""
        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認(rèn)證裝飾器"""
        @wraps(f)
        def decorated_function(*args, **kwargs):
            user_data = self.get_current_user()
            if not user_data:
                return jsonify({'error': '認(rèn)證令牌無(wú)效或已過(guò)期'}), 401
            
            # 將用戶信息添加到請(qǐng)求上下文
            request.current_user = user_data
            return f(*args, **kwargs)
        return decorated_function
    
    def role_required(self, role: str):
        """角色權(quán)限裝飾器"""
        def decorator(f):
            @wraps(f)
            @self.jwt_required
            def decorated_function(*args, **kwargs):
                if request.current_user.get('role') != role:
                    return jsonify({'error': '權(quán)限不足'}), 403
                return f(*args, **kwargs)
            return decorated_function
        return decorator

# 初始化Flask應(yīng)用
def create_jwt_app():
    app = Flask(__name__)
    
    # 初始化JWT認(rèn)證
    jwt_auth = JWTAuth(app)
    
    # 添加需要特定權(quán)限的路由
    @app.route('/api/jwt/admin')
    @jwt_auth.role_required('admin')
    def admin_dashboard():
        """管理員儀表板（需要admin角色）"""
        return jsonify({
            'message': '歡迎來(lái)到管理員面板',
            'user': request.current_user
        })
    
    return app

# 運(yùn)行應(yīng)用
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):
    """增強(qiáng)安全性的JWT實(shí)現(xiàn)"""
    
    def __init__(self, app=None):
        super().__init__(app)
        # 記錄令牌指紋，防止重放攻擊
        self.token_fingerprints = {}
    
    def create_secure_access_token(self, user_data: Dict, client_fingerprint: str) -> str:
        """創(chuàng)建安全的訪問(wèn)令牌"""
        # 生成令牌指紋
        token_fingerprint = self._generate_token_fingerprint(user_data['user_id'], client_fingerprint)
        
        # 將指紋加入用戶數(shù)據(jù)
        user_data_with_fp = user_data.copy()
        user_data_with_fp['fp'] = token_fingerprint
        
        access_token = self.create_access_token(user_data_with_fp)
        
        # 存儲(chǔ)令牌指紋
        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]:
        """驗(yàn)證安全令牌"""
        payload = self.verify_token(token)
        if not payload:
            return None
        
        # 驗(yàn)證令牌指紋
        token_fp = payload.get('fp')
        if not token_fp or not self._validate_token_fingerprint(token_fp, client_fingerprint, payload['user_id']):
            return None
        
        # 更新最后使用時(shí)間
        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:
        """驗(yàn)證令牌指紋"""
        if token_fp not in self.token_fingerprints:
            return False
        
        fp_data = self.token_fingerprints[token_fp]
        
        # 檢查指紋是否過(guò)期（24小時(shí)）
        if datetime.utcnow() - fp_data['created_at'] > timedelta(hours=24):
            del self.token_fingerprints[token_fp]
            return False
        
        # 驗(yàn)證用戶ID匹配
        return fp_data['user_id'] == user_id

# JWT令牌自動(dòng)清理任務(wù)
def start_token_cleanup_task(jwt_auth: SecureJWTAuth):
    """啟動(dòng)令牌清理任務(wù)"""
    import threading
    import time
    
    def cleanup_task():
        while True:
            time.sleep(3600)  # 每小時(shí)清理一次
            current_time = datetime.utcnow()
            
            # 清理過(guò)期的令牌指紋
            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]
            
            # 清理黑名單中的舊令牌（超過(guò)7天）
            # 注意：實(shí)際項(xiàng)目中黑名單應(yīng)該使用Redis等有TTL的數(shù)據(jù)存儲(chǔ)
    
    thread = threading.Thread(target=cleanup_task, daemon=True)
    thread.start()

class HybridAuth:
    """混合認(rèn)證策略：結(jié)合Session和JWT的優(yōu)點(diǎn)"""
    
    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):
        """初始化混合認(rèn)證"""
        self.session_auth.init_app(app)
        self.jwt_auth.init_app(app)
        
        self.register_hybrid_routes(app)
    
    def register_hybrid_routes(self, app):
        """注冊(cè)混合認(rèn)證路由"""
        
        @app.route('/api/hybrid/login', methods=['POST'])
        def hybrid_login():
            """混合登錄：創(chuàng)建Session和JWT"""
            data = request.get_json()
            username = data.get('username')
            password = data.get('password')
            
            # 驗(yàn)證用戶憑證
            user = self.session_auth.authenticate_user(username, password)
            if not user:
                return jsonify({'error': '無(wú)效的憑證'}), 401
            
            # 創(chuàng)建Session（用于Web端）
            session.permanent = True
            session['user_id'] = user['user_id']
            session['username'] = username
            session['role'] = user['role']
            session['logged_in'] = True
            
            # 創(chuàng)建JWT（用于API/移動(dòng)端）
            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
            }
            
            # 根據(jù)客戶端類型返回不同響應(yīng)
            user_agent = request.headers.get('User-Agent', '').lower()
            if 'mobile' in user_agent or request.args.get('client') == 'mobile':
                # 移動(dòng)端：主要使用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():
            """混合認(rèn)證獲取用戶資料"""
            # 嘗試Session認(rèn)證
            if session.get('logged_in'):
                return jsonify({
                    'user_id': session['user_id'],
                    'username': session['username'],
                    'role': session['role'],
                    'auth_method': 'session'
                })
            
            # 嘗試JWT認(rèn)證
            user_data = self.jwt_auth.get_current_user()
            if user_data:
                return jsonify({
                    **user_data,
                    'auth_method': 'jwt'
                })
            
            return jsonify({'error': '未認(rèn)證'}), 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():
    """創(chuàng)建混合認(rèn)證應(yīng)用"""
    app = Flask(__name__)
    hybrid_auth = HybridAuth(app)
    return app

class SmartAuthRouter:
    """智能認(rèn)證路由：根據(jù)請(qǐng)求特征選擇認(rèn)證方式"""
    
    def __init__(self, session_auth, jwt_auth):
        self.session_auth = session_auth
        self.jwt_auth = jwt_auth
    
    def authenticate_request(self, request):
        """智能認(rèn)證請(qǐng)求"""
        # 分析請(qǐng)求特征
        features = self._analyze_request_features(request)
        
        # 根據(jù)特征選擇認(rèn)證策略
        if features['is_web_browser'] and not features['is_api_client']:
            # Web瀏覽器：優(yōu)先使用Session
            return self._authenticate_with_session(request)
        else:
            # API客戶端/移動(dòng)端：優(yōu)先使用JWT
            return self._authenticate_with_jwt(request)
    
    def _analyze_request_features(self, request):
        """分析請(qǐng)求特征"""
        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認(rèn)證"""
        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認(rèn)證"""
        return self.jwt_auth.get_current_user()

import re
from flask import request, abort

class SecurityEnhancer:
    """安全增強(qiáng)器"""
    
    @staticmethod
    def validate_password_strength(password: str) -> bool:
        """驗(yàn)證密碼強(qiáng)度"""
        if len(password) < 8:
            return False
        
        # 檢查包含大寫(xiě)、小寫(xiě)、數(shù)字、特殊字符
        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))
        
        # 至少滿足其中三項(xiàng)
        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):
        """檢測(cè)暴力破解攻擊"""
        # 實(shí)際項(xiàng)目中應(yīng)該使用Redis等外部存儲(chǔ)
        import time
        current_time = time.time()
        window_seconds = window_minutes * 60
        
        # 清理過(guò)期的嘗試記錄
        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  # 觸發(fā)暴力破解防護(hù)
        
        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
    
    # 靜態(tài)變量存儲(chǔ)失敗嘗試
    failed_attempts = {}

class RateLimiter:
    """API速率限制器"""
    
    def __init__(self):
        self.requests = {}
    
    def is_rate_limited(self, identifier: str, max_requests: int, window_seconds: int) -> bool:
        """檢查是否達(dá)到速率限制"""
        import time
        current_time = time.time()
        window_start = current_time - window_seconds
        
        # 清理過(guò)期的請(qǐng)求記錄
        if identifier in self.requests:
            self.requests[identifier] = [
                req_time for req_time in self.requests[identifier]
                if req_time > window_start
            ]
        
        # 檢查請(qǐng)求次數(shù)
        if identifier not in self.requests:
            self.requests[identifier] = []
        
        if len(self.requests[identifier]) >= max_requests:
            return True
        
        # 記錄當(dāng)前請(qǐng)求
        self.requests[identifier].append(current_time)
        return False

# 使用示例
rate_limiter = RateLimiter()

def apply_rate_limit(f):
    """應(yīng)用速率限制的裝飾器"""
    @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': '請(qǐng)求過(guò)于頻繁，請(qǐng)稍后重試'}), 429
        return f(*args, **kwargs)
    return decorated_function

import os
from datetime import timedelta

class ProductionSecurityConfig:
    """生產(chǎn)環(huán)境安全配置"""
    
    @staticmethod
    def get_session_config():
        """生產(chǎn)環(huán)境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():
        """生產(chǎn)環(huán)境JWT配置"""
        return {
            'JWT_SECRET_KEY': os.environ['JWT_SECRET_KEY'],  # 從環(huán)境變量獲取
            '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:
    """認(rèn)證性能測(cè)試器"""
    
    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):
        """測(cè)試Session認(rèn)證性能"""
        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
        
        # 測(cè)試認(rèn)證請(qǐng)求性能
        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):
        """測(cè)試JWT認(rèn)證性能"""
        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}'}
        
        # 測(cè)試認(rèn)證請(qǐng)求性能
        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):
        """并發(fā)執(zhí)行請(qǐng)求"""
        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):
        """生成性能測(cè)試報(bào)告"""
        print("=" * 50)
        print("認(rèn)證機(jī)制性能測(cè)試報(bào)告")
        print("=" * 50)
        
        for auth_type in ['session', 'jwt']:
            times = self.results[auth_type]
            if not times:
                continue
                
            avg_time = statistics.mean(times) * 1000  # 轉(zhuǎn)換為毫秒
            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()} 認(rèn)證:")
            print(f"  平均響應(yīng)時(shí)間: {avg_time:.2f}ms")
            print(f"  最小響應(yīng)時(shí)間: {min_time:.2f}ms")
            print(f"  最大響應(yīng)時(shí)間: {max_time:.2f}ms")
            print(f"  標(biāo)準(zhǔn)差: {std_dev:.2f}ms")
            print(f"  請(qǐng)求總數(shù): {len(times)}")
        
        # 性能對(duì)比
        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性能對(duì)比:")
                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()