eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c

import jwt
from datetime import datetime, timedelta
from typing import Dict, Optional
from flask import Flask, request, jsonify
from flask_jwt_extended import JWTManager, create_access_token, get_jwt_identity
from werkzeug.security import check_password_hash, generate_password_hash
import hashlib
class JWTTokenUtil:
    """
    JWT Token工具類
    """
    def __init__(self, secret_key: str, access_token_expire_minutes: int = 30):
        """
        初始化JWT工具類
        :param secret_key: 密鑰
        :param access_token_expire_minutes: 訪問令牌過期時(shí)間（分鐘）
        """
        self.secret_key = secret_key
        self.access_token_expire_minutes = access_token_expire_minutes
    def generate_token(self, user_info: Dict) -> str:
        """
        生成JWT Token
        :param user_info: 用戶信息
        :return: JWT Token字符串
        """
        # 設(shè)置Token過期時(shí)間
        expire = datetime.utcnow() + timedelta(minutes=self.access_token_expire_minutes)
        # 構(gòu)建Token載荷
        payload = {
            "sub": user_info.get("username"),
            "user_id": user_info.get("id"),
            "user_roles": user_info.get("roles", []),
            "exp": expire,
            "iat": datetime.utcnow(),
            "jti": hashlib.md5(f"{user_info.get('username')}_{expire}".encode()).hexdigest()
        }
        # 生成Token
        token = jwt.encode(payload, self.secret_key, algorithm="HS256")
        return token
    def verify_token(self, token: str) -> Optional[Dict]:
        """
        驗(yàn)證JWT Token
        :param token: JWT Token字符串
        :return: 用戶信息或None
        """
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=["HS256"])
            return payload
        except jwt.ExpiredSignatureError:
            # Token已過期
            return None
        except jwt.InvalidTokenError:
            # Token無(wú)效
            return None
    def get_user_info_from_token(self, token: str) -> Optional[Dict]:
        """
        從Token中提取用戶信息
        :param token: JWT Token字符串
        :return: 用戶信息
        """
        payload = self.verify_token(token)
        if payload:
            return {
                "username": payload.get("sub"),
                "user_id": payload.get("user_id"),
                "roles": payload.get("user_roles", [])
            }
        return None
class TokenAuthenticationFilter:
    """
    Token認(rèn)證過濾器
    """
    def __init__(self, jwt_util: JWTTokenUtil, excluded_urls: list = None):
        """
        初始化Token認(rèn)證過濾器
        :param jwt_util: JWT工具類實(shí)例
        :param excluded_urls: 不需要認(rèn)證的URL列表
        """
        self.jwt_util = jwt_util
        self.excluded_urls = excluded_urls or []
    def should_filter(self, request_url: str) -> bool:
        """
        判斷是否需要進(jìn)行Token認(rèn)證
        :param request_url: 請(qǐng)求URL
        :return: True需要認(rèn)證，F(xiàn)alse不需要認(rèn)證
        """
        for excluded_url in self.excluded_urls:
            if request_url.startswith(excluded_url):
                return False
        return True
    def do_filter(self, request):
        """
        執(zhí)行Token認(rèn)證過濾
        :param request: HTTP請(qǐng)求對(duì)象
        :return: 認(rèn)證結(jié)果
        """
        if not self.should_filter(request.path):
            return True  # 不需要認(rèn)證的URL直接通過
        # 從請(qǐng)求頭獲取Token
        auth_header = request.headers.get('Authorization')
        if not auth_header or not auth_header.startswith('Bearer '):
            return False  # 缺少認(rèn)證頭或格式不正確
        token = auth_header[7:]  # 移除"Bearer "前綴
        # 驗(yàn)證Token
        user_info = self.jwt_util.get_user_info_from_token(token)
        if not user_info:
            return False  # Token無(wú)效
        # 將用戶信息存儲(chǔ)到請(qǐng)求上下文
        request.user_info = user_info
        return True
class UserDetailsService:
    """
    用戶詳情服務(wù)
    """
    def __init__(self):
        self.password_encoder = generate_password_hash
    def load_user_by_username(self, username: str):
        """
        根據(jù)用戶名加載用戶信息
        :param username: 用戶名
        :return: 用戶對(duì)象
        """
        # 模擬從數(shù)據(jù)庫(kù)查詢用戶信息
        # 在實(shí)際應(yīng)用中，這里應(yīng)該是對(duì)數(shù)據(jù)庫(kù)的查詢操作
        user_data = {
            "id": "1",
            "username": username,
            "password": generate_password_hash("password123"),
            "roles": ["ROLE_USER", "ROLE_ADMIN"],
            "enabled": True,
            "account_non_expired": True,
            "credentials_non_expired": True,
            "account_non_locked": True
        }
        # 檢查用戶狀態(tài)
        if not user_data["enabled"]:
            raise Exception("用戶已被禁用")
        if not user_data["account_non_expired"]:
            raise Exception("賬戶已過期")
        if not user_data["credentials_non_expired"]:
            raise Exception("憑證已過期")
        if not user_data["account_non_locked"]:
            raise Exception("賬戶已被鎖定")
        return user_data
    def authenticate(self, username: str, password: str) -> Optional[Dict]:
        """
        用戶認(rèn)證
        :param username: 用戶名
        :param password: 密碼
        :return: 認(rèn)證成功返回用戶信息，失敗返回None
        """
        try:
            user = self.load_user_by_username(username)
            if check_password_hash(user["password"], password):
                return user
            return None
        except Exception:
            return None

class JwtAuthenticationTokenFilter:
    """
    JWT認(rèn)證Token過濾器
    """
    def __init__(self, jwt_util: JWTTokenUtil, user_details_service: UserDetailsService):
        """
        初始化JWT認(rèn)證過濾器
        :param jwt_util: JWT工具類
        :param user_details_service: 用戶詳情服務(wù)
        """
        self.jwt_util = jwt_util
        self.user_details_service = user_details_service
        self.token_header = "Authorization"
        self.token_prefix = "Bearer "
    def do_filter_internal(self, request):
        """
        內(nèi)部過濾邏輯
        :param request: HTTP請(qǐng)求
        :return: 過濾結(jié)果
        """
        auth_header = request.headers.get(self.token_header)
        if auth_header and auth_header.startswith(self.token_prefix):
            token = auth_header[len(self.token_prefix):]
            username = self.jwt_util.get_user_info_from_token(token).get("username") if token else None
            if username:
                # 檢查SecurityContext中是否已存在認(rèn)證信息
                if not self.security_context_exists():
                    # 從數(shù)據(jù)庫(kù)加載用戶詳情
                    user_details = self.user_details_service.load_user_by_username(username)
                    if self.jwt_util.verify_token(token):
                        # 設(shè)置認(rèn)證信息到SecurityContext
                        self.set_authentication(request, user_details)
    def security_context_exists(self) -> bool:
        """
        檢查SecurityContext是否存在
        :return: True存在，F(xiàn)alse不存在
        """
        # 在實(shí)際實(shí)現(xiàn)中，這里會(huì)檢查Spring Security的SecurityContext
        return False
    def set_authentication(self, request, user_details: Dict):
        """
        設(shè)置認(rèn)證信息
        :param request: HTTP請(qǐng)求
        :param user_details: 用戶詳情
        """
        # 創(chuàng)建認(rèn)證對(duì)象并設(shè)置到SecurityContext
        # 在實(shí)際實(shí)現(xiàn)中，這里會(huì)創(chuàng)建UsernamePasswordAuthenticationToken
        pass

class FeignTokenRelayInterceptor:
    """
    Feign客戶端Token傳遞攔截器
    """
    def __init__(self, jwt_util: JWTTokenUtil):
        """
        初始化Token傳遞攔截器
        :param jwt_util: JWT工具類
        """
        self.jwt_util = jwt_util
    def apply(self, request_template):
        """
        應(yīng)用攔截器邏輯
        :param request_template: 請(qǐng)求模板
        """
        # 從當(dāng)前請(qǐng)求獲取Token
        current_token = self.get_current_token()
        if current_token:
            # 將Token添加到請(qǐng)求頭
            request_template.header("Authorization", f"Bearer {current_token}")
    def get_current_token(self) -> Optional[str]:
        """
        獲取當(dāng)前請(qǐng)求的Token
        :return: Token字符串
        """
        # 從當(dāng)前SecurityContext獲取Token
        # 在實(shí)際實(shí)現(xiàn)中，這里會(huì)從Spring Security上下文獲取
        pass
class TokenRelayService:
    """
    Token傳遞服務(wù)
    """
    def __init__(self, jwt_util: JWTTokenUtil):
        self.jwt_util = jwt_util
    def relay_token(self, target_service_url: str, request_data: Dict):
        """
        傳遞Token到目標(biāo)服務(wù)
        :param target_service_url: 目標(biāo)服務(wù)URL
        :param request_data: 請(qǐng)求數(shù)據(jù)
        :return: 響應(yīng)結(jié)果
        """
        # 獲取當(dāng)前Token
        current_token = self.get_current_token()
        # 構(gòu)建帶認(rèn)證頭的請(qǐng)求
        headers = {
            "Authorization": f"Bearer {current_token}",
            "Content-Type": "application/json"
        }
        # 發(fā)送請(qǐng)求到目標(biāo)服務(wù)
        import requests
        response = requests.post(target_service_url, json=request_data, headers=headers)
        return response.json()

class TokenRefreshService:
    """
    Token刷新服務(wù)
    """
    def __init__(self, jwt_util: JWTTokenUtil, redis_client):
        """
        初始化Token刷新服務(wù)
        :param jwt_util: JWT工具類
        :param redis_client: Redis客戶端
        """
        self.jwt_util = jwt_util
        self.redis = redis_client
        self.refresh_token_expire_days = 30
    def generate_refresh_token(self, user_info: Dict) -> str:
        """
        生成刷新Token
        :param user_info: 用戶信息
        :return: 刷新Token
        """
        expire = datetime.utcnow() + timedelta(days=self.refresh_token_expire_days)
        payload = {
            "sub": user_info.get("username"),
            "user_id": user_info.get("user_id"),
            "exp": expire,
            "iat": datetime.utcnow(),
            "type": "refresh"
        }
        refresh_token = jwt.encode(payload, self.jwt_util.secret_key, algorithm="HS256")
        # 將刷新Token存儲(chǔ)到Redis，設(shè)置過期時(shí)間
        self.redis.setex(
            f"refresh_token:{refresh_token}", 
            self.refresh_token_expire_days * 24 * 60 * 60, 
            user_info.get("user_id")
        )
        return refresh_token
    def refresh_access_token(self, refresh_token: str) -> Optional[Dict]:
        """
        刷新訪問Token
        :param refresh_token: 刷新Token
        :return: 新的Token信息
        """
        try:
            # 驗(yàn)證刷新Token
            payload = jwt.decode(refresh_token, self.jwt_util.secret_key, algorithms=["HS256"])
            if payload.get("type") != "refresh":
                return None
            # 檢查刷新Token是否存在于Redis中
            user_id = self.redis.get(f"refresh_token:{refresh_token}")
            if not user_id:
                return None
            # 生成新的訪問Token
            user_info = {
                "username": payload.get("sub"),
                "id": payload.get("user_id"),
                "roles": self.get_user_roles(payload.get("user_id"))
            }
            new_access_token = self.jwt_util.generate_token(user_info)
            return {
                "access_token": new_access_token,
                "refresh_token": refresh_token,
                "token_type": "Bearer",
                "expires_in": self.jwt_util.access_token_expire_minutes * 60
            }
        except jwt.ExpiredSignatureError:
            # 刷新Token已過期，刪除Redis中的記錄
            self.redis.delete(f"refresh_token:{refresh_token}")
            return None
        except jwt.InvalidTokenError:
            return None
    def revoke_refresh_token(self, refresh_token: str) -> bool:
        """
        注銷刷新Token
        :param refresh_token: 刷新Token
        :return: 注銷結(jié)果
        """
        result = self.redis.delete(f"refresh_token:{refresh_token}")
        return result > 0
    def get_user_roles(self, user_id: str) -> list:
        """
        獲取用戶角色
        :param user_id: 用戶ID
        :return: 角色列表
        """
        # 在實(shí)際應(yīng)用中，這里會(huì)查詢數(shù)據(jù)庫(kù)獲取用戶角色
        return ["ROLE_USER"]

class RbacAuthorityService:
    """
    基于RBAC的權(quán)限服務(wù)
    """
    def __init__(self, redis_client):
        self.redis = redis_client
    def has_permission(self, user_info: Dict, resource: str, action: str) -> bool:
        """
        檢查用戶是否有權(quán)限執(zhí)行指定操作
        :param user_info: 用戶信息
        :param resource: 資源
        :param action: 操作
        :return: 有權(quán)限返回True，否則返回False
        """
        user_id = user_info.get("user_id")
        roles = user_info.get("roles", [])
        # 檢查用戶角色是否具有訪問資源的權(quán)限
        for role in roles:
            permission_key = f"permission:{role}:{resource}:{action}"
            has_perm = self.redis.get(permission_key)
            if has_perm:
                return True
        return False
    def get_user_permissions(self, user_id: str) -> Dict:
        """
        獲取用戶所有權(quán)限
        :param user_id: 用戶ID
        :return: 權(quán)限字典
        """
        # 獲取用戶角色
        user_roles = self.get_user_roles(user_id)
        permissions = {}
        for role in user_roles:
            role_perms = self.get_role_permissions(role)
            permissions.update(role_perms)
        return permissions
    def get_user_roles(self, user_id: str) -> list:
        """
        獲取用戶角色列表
        :param user_id: 用戶ID
        :return: 角色列表
        """
        # 在實(shí)際應(yīng)用中，這里會(huì)查詢數(shù)據(jù)庫(kù)或緩存
        return ["ROLE_USER", "ROLE_ADMIN"]
    def get_role_permissions(self, role: str) -> Dict:
        """
        獲取角色權(quán)限
        :param role: 角色
        :return: 權(quán)限字典
        """
        # 在實(shí)際應(yīng)用中，這里會(huì)查詢數(shù)據(jù)庫(kù)或緩存
        return {
            "user:read": True,
            "user:write": role == "ROLE_ADMIN",
            "admin:manage": role == "ROLE_ADMIN"
        }
class PermissionEvaluator:
    """
    權(quán)限評(píng)估器
    """
    def __init__(self, rbac_service: RbacAuthorityService):
        self.rbac_service = rbac_service
    def evaluate(self, user_info: Dict, request_resource: str, request_action: str) -> bool:
        """
        評(píng)估用戶權(quán)限
        :param user_info: 用戶信息
        :param request_resource: 請(qǐng)求資源
        :param request_action: 請(qǐng)求操作
        :return: 有權(quán)限返回True，否則返回False
        """
        return self.rbac_service.has_permission(user_info, request_resource, request_action)

from flask import Flask
from flask_security import Security, SQLAlchemyUserDatastore
from flask_login import LoginManager
class SecurityConfig:
    """
    安全配置類
    """
    def __init__(self):
        self.app = Flask(__name__)
        self.configure_security()
    def configure_security(self):
        """
        配置安全設(shè)置
        """
        # JWT配置
        self.app.config['JWT_SECRET_KEY'] = 'your-super-secret-jwt-key'
        self.app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(minutes=30)
        self.app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)
        # 密碼編碼器配置
        self.app.config['SECURITY_PASSWORD_HASH'] = 'bcrypt'
        self.app.config['SECURITY_PASSWORD_SALT'] = 'your-salt'
        # 安全頭配置
        self.app.config['SECURITY_CSRF_PROTECTION'] = True
    def configure_http_security(self):
        """
        配置HTTP安全
        """
        # 在實(shí)際Spring Boot應(yīng)用中，這對(duì)應(yīng)SecurityConfig類
        pass
    def get_security_filter_chain(self):
        """
        獲取安全過濾器鏈
        :return: 過濾器鏈配置
        """
        # 定義不需要認(rèn)證的URL
        permit_all_urls = [
            "/auth/login",
            "/auth/register", 
            "/auth/refresh",
            "/actuator/**",
            "/v2/api-docs",
            "/swagger-ui/**",
            "/webjars/**"
        ]
        # 定義安全過濾器鏈
        filter_chain = {
            "permit_all": permit_all_urls,
            "authenticated": ["/**"],  # 其他所有路徑需要認(rèn)證
            "filters": [
                "CorsFilter",
                "TokenAuthenticationFilter", 
                "RequestContextFilter",
                "LogoutFilter",
                "UsernamePasswordAuthenticationFilter",
                "DefaultLoginPageGeneratingFilter",
                "DefaultLogoutPageGeneratingFilter",
                "BasicAuthenticationFilter",
                "RequestCacheAwareFilter",
                "SecurityContextHolderAwareRequestFilter",
                "AnonymousAuthenticationFilter",
                "SessionManagementFilter",
                "ExceptionTranslationFilter",
                "FilterSecurityInterceptor"
            ]
        }
        return filter_chain
class WebSecurityConfig:
    """
    Web安全配置
    """
    def __init__(self):
        self.security_config = SecurityConfig()
    def configure(self, http):
        """
        配置HTTP安全
        :param http: HTTP安全配置對(duì)象
        """
        # 配置不需要認(rèn)證的路徑
        http.authorize_requests() \
            .ant_matchers(["/auth/login", "/auth/register", "/auth/refresh"]).permit_all() \
            .ant_matchers(["/actuator/**", "/v2/api-docs", "/swagger-ui/**"]).permit_all() \
            .any_request().authenticated()
        # 配置JWT認(rèn)證過濾器
        http.add_filter_before(
            JwtAuthenticationTokenFilter(), 
            UsernamePasswordAuthenticationFilter
        )
        # 禁用CSRF，因?yàn)槭褂肨oken認(rèn)證
        http.csrf().disable()
        # 禁用session，因?yàn)槭褂肨oken認(rèn)證
        http.session_management().session_creation_policy(SessionCreationPolicy.STATELESS)
        # 配置異常處理
        http.exception_handling() \
            .authentication_entry_point(CustomAuthenticationEntryPoint()) \
            .access_denied_handler(CustomAccessDeniedHandler())

from flask import Flask, request, jsonify
from flask_jwt_extended import create_access_token, create_refresh_token
class AuthController:
    """
    認(rèn)證控制器
    """
    def __init__(self, user_details_service: UserDetailsService, 
                 jwt_util: JWTTokenUtil, token_refresh_service: TokenRefreshService):
        self.user_details_service = user_details_service
        self.jwt_util = jwt_util
        self.token_refresh_service = token_refresh_service
    def login(self, login_form: Dict):
        """
        用戶登錄
        :param login_form: 登錄表單
        :return: 認(rèn)證結(jié)果
        """
        username = login_form.get("username")
        password = login_form.get("password")
        # 驗(yàn)證用戶憑據(jù)
        user = self.user_details_service.authenticate(username, password)
        if not user:
            return {
                "success": False,
                "message": "用戶名或密碼錯(cuò)誤"
            }
        # 生成訪問Token
        access_token = self.jwt_util.generate_token(user)
        # 生成刷新Token
        refresh_token = self.token_refresh_service.generate_refresh_token(user)
        return {
            "success": True,
            "data": {
                "access_token": access_token,
                "refresh_token": refresh_token,
                "token_type": "Bearer",
                "expires_in": self.jwt_util.access_token_expire_minutes * 60,
                "user_info": {
                    "id": user.get("id"),
                    "username": user.get("username"),
                    "roles": user.get("roles")
                }
            }
        }
    def refresh_token(self, refresh_token: str):
        """
        刷新Token
        :param refresh_token: 刷新Token
        :return: 新的Token信息
        """
        token_info = self.token_refresh_service.refresh_access_token(refresh_token)
        if token_info:
            return {
                "success": True,
                "data": token_info
            }
        else:
            return {
                "success": False,
                "message": "刷新Token無(wú)效或已過期"
            }
    def logout(self, request):
        """
        用戶登出
        :param request: HTTP請(qǐng)求
        :return: 登出結(jié)果
        """
        auth_header = request.headers.get('Authorization')
        if auth_header and auth_header.startswith('Bearer '):
            token = auth_header[7:]
            # 嘗試將Token加入黑名單（可選）
            self.blacklist_token(token)
        return {
            "success": True,
            "message": "登出成功"
        }
    def blacklist_token(self, token: str):
        """
        將Token加入黑名單
        :param token: Token字符串
        """
        # 在實(shí)際應(yīng)用中，可以將Token存儲(chǔ)到Redis中，設(shè)置過期時(shí)間
        # 這樣可以防止Token在過期前被使用
        pass

class ProtectedResourceController:
    """
    受保護(hù)資源控制器
    """
    def __init__(self, permission_evaluator: PermissionEvaluator):
        self.permission_evaluator = permission_evaluator
    def get_user_info(self, request):
        """
        獲取用戶信息
        :param request: HTTP請(qǐng)求
        :return: 用戶信息
        """
        user_info = request.user_info  # 從請(qǐng)求上下文獲取用戶信息
        return {
            "success": True,
            "data": user_info
        }
    def get_sensitive_data(self, request):
        """
        獲取敏感數(shù)據(jù)
        :param request: HTTP請(qǐng)求
        :return: 敏感數(shù)據(jù)
        """
        user_info = request.user_info
        # 檢查用戶權(quán)限
        if not self.permission_evaluator.evaluate(user_info, "sensitive_data", "read"):
            return {
                "success": False,
                "message": "沒有訪問權(quán)限"
            }
        # 返回敏感數(shù)據(jù)
        return {
            "success": True,
            "data": self.fetch_sensitive_data()
        }
    def fetch_sensitive_data(self):
        """
        獲取敏感數(shù)據(jù)
        :return: 敏感數(shù)據(jù)
        """
        # 在實(shí)際應(yīng)用中，這里會(huì)查詢數(shù)據(jù)庫(kù)獲取敏感數(shù)據(jù)
        return {
            "id": "12345",
            "name": "敏感數(shù)據(jù)",
            "content": "這是需要特殊權(quán)限才能訪問的敏感信息"
        }

import bcrypt
from passlib.hash import pbkdf2_sha256
class PasswordSecurityUtil:
    """
    密碼安全工具類
    """
    @staticmethod
    def encode_password(raw_password: str) -> str:
        """
        加密密碼
        :param raw_password: 原始密碼
        :return: 加密后的密碼
        """
        # 使用bcrypt進(jìn)行密碼加密
        salt = bcrypt.gensalt()
        hashed = bcrypt.hashpw(raw_password.encode('utf-8'), salt)
        return hashed.decode('utf-8')
    @staticmethod
    def matches(raw_password: str, encoded_password: str) -> bool:
        """
        驗(yàn)證密碼
        :param raw_password: 原始密碼
        :param encoded_password: 加密后的密碼
        :return: 驗(yàn)證結(jié)果
        """
        return bcrypt.checkpw(raw_password.encode('utf-8'), encoded_password.encode('utf-8'))
    @staticmethod
    def validate_password_strength(password: str) -> Dict[str, bool]:
        """
        驗(yàn)證密碼強(qiáng)度
        :param password: 密碼
        :return: 驗(yàn)證結(jié)果
        """
        import re
        validation_result = {
            "length_valid": len(password) >= 8,
            "has_uppercase": bool(re.search(r'[A-Z]', password)),
            "has_lowercase": bool(re.search(r'[a-z]', password)),
            "has_digit": bool(re.search(r'\d', password)),
            "has_special": bool(re.search(r'[!@#$%^&*(),.?":{}|<>]', password))
        }
        validation_result["is_strong"] = all(validation_result.values())
        return validation_result

class OptimizedTokenValidator:
    """
    優(yōu)化的Token驗(yàn)證器
    """
    def __init__(self, redis_client, jwt_util: JWTTokenUtil):
        self.redis = redis_client
        self.jwt_util = jwt_util
        self.token_cache_ttl = 300  # 5分鐘緩存
    def validate_token(self, token: str) -> Optional[Dict]:
        """
        驗(yàn)證Token（帶緩存）
        :param token: Token字符串
        :return: 用戶信息或None
        """
        # 首先檢查緩存
        cached_user_info = self.redis.get(f"token_cache:{token}")
        if cached_user_info:
            return eval(cached_user_info)  # 在實(shí)際應(yīng)用中應(yīng)使用安全的序列化方式
        # 緩存未命中，驗(yàn)證Token
        user_info = self.jwt_util.get_user_info_from_token(token)
        if user_info:
            # 緩存驗(yàn)證結(jié)果
            self.redis.setex(
                f"token_cache:{token}", 
                self.token_cache_ttl, 
                str(user_info)
            )
        return user_info

import logging
from datetime import datetime
class SecurityAuditLogger:
    """
    安全審計(jì)日志記錄器
    """
    def __init__(self):
        self.logger = logging.getLogger("security-audit")
        self.logger.setLevel(logging.INFO)
    def log_authentication_success(self, username: str, ip_address: str):
        """
        記錄認(rèn)證成功事件
        :param username: 用戶名
        :param ip_address: IP地址
        """
        self.logger.info(f"AUTH_SUCCESS - User: {username}, IP: {ip_address}, Time: {datetime.now()}")
    def log_authentication_failure(self, username: str, ip_address: str, reason: str):
        """
        記錄認(rèn)證失敗事件
        :param username: 用戶名
        :param ip_address: IP地址
        :param reason: 失敗原因
        """
        self.logger.warning(f"AUTH_FAILURE - User: {username}, IP: {ip_address}, Reason: {reason}, Time: {datetime.now()}")
    def log_authorization_failure(self, username: str, resource: str, ip_address: str):
        """
        記錄授權(quán)失敗事件
        :param username: 用戶名
        :param resource: 資源
        :param ip_address: IP地址
        """
        self.logger.warning(f"AUTHZ_FAILURE - User: {username}, Resource: {resource}, IP: {ip_address}, Time: {datetime.now()}")
    def log_token_refresh(self, username: str, ip_address: str):
        """
        記錄Token刷新事件
        :param username: 用戶名
        :param ip_address: IP地址
        """
        self.logger.info(f"TOKEN_REFRESH - User: {username}, IP: {ip_address}, Time: {datetime.now()}")