Spring Security与JWT实现微服务安全认证的方法

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: 访问令牌过期时间（分钟）
        """
        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字符串
        """
        # 设置Token过期时间
        expire = datetime.utcnow() + timedelta(minutes=self.access_token_expire_minutes)
        # 构建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]:
        """
        验证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无效
            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认证过滤器
    """
    def __init__(self, jwt_util: JWTTokenUtil, excluded_urls: list = None):
        """
        初始化Token认证过滤器
        :param jwt_util: JWT工具类实例
        :param excluded_urls: 不需要认证的URL列表
        """
        self.jwt_util = jwt_util
        self.excluded_urls = excluded_urls or []
    def should_filter(self, request_url: str) -> bool:
        """
        判断是否需要进行Token认证
        :param request_url: 请求URL
        :return: True需要认证，False不需要认证
        """
        for excluded_url in self.excluded_urls:
            if request_url.startswith(excluded_url):
                return False
        return True
    def do_filter(self, request):
        """
        执行Token认证过滤
        :param request: HTTP请求对象
        :return: 认证结果
        """
        if not self.should_filter(request.path):
            return True  # 不需要认证的URL直接通过
        # 从请求头获取Token
        auth_header = request.headers.get('Authorization')
        if not auth_header or not auth_header.startswith('Bearer '):
            return False  # 缺少认证头或格式不正确
        token = auth_header[7:]  # 移除"Bearer "前缀
        # 验证Token
        user_info = self.jwt_util.get_user_info_from_token(token)
        if not user_info:
            return False  # Token无效
        # 将用户信息存储到请求上下文
        request.user_info = user_info
        return True
class UserDetailsService:
    """
    用户详情服务
    """
    def __init__(self):
        self.password_encoder = generate_password_hash
    def load_user_by_username(self, username: str):
        """
        根据用户名加载用户信息
        :param username: 用户名
        :return: 用户对象
        """
        # 模拟从数据库查询用户信息
        # 在实际应用中，这里应该是对数据库的查询操作
        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
        }
        # 检查用户状态
        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]:
        """
        用户认证
        :param username: 用户名
        :param password: 密码
        :return: 认证成功返回用户信息，失败返回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认证Token过滤器
    """
    def __init__(self, jwt_util: JWTTokenUtil, user_details_service: UserDetailsService):
        """
        初始化JWT认证过滤器
        :param jwt_util: JWT工具类
        :param user_details_service: 用户详情服务
        """
        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):
        """
        内部过滤逻辑
        :param request: HTTP请求
        :return: 过滤结果
        """
        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中是否已存在认证信息
                if not self.security_context_exists():
                    # 从数据库加载用户详情
                    user_details = self.user_details_service.load_user_by_username(username)
                    if self.jwt_util.verify_token(token):
                        # 设置认证信息到SecurityContext
                        self.set_authentication(request, user_details)
    def security_context_exists(self) -> bool:
        """
        检查SecurityContext是否存在
        :return: True存在，False不存在
        """
        # 在实际实现中，这里会检查Spring Security的SecurityContext
        return False
    def set_authentication(self, request, user_details: Dict):
        """
        设置认证信息
        :param request: HTTP请求
        :param user_details: 用户详情
        """
        # 创建认证对象并设置到SecurityContext
        # 在实际实现中，这里会创建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):
        """
        应用拦截器逻辑
        :param request_template: 请求模板
        """
        # 从当前请求获取Token
        current_token = self.get_current_token()
        if current_token:
            # 将Token添加到请求头
            request_template.header("Authorization", f"Bearer {current_token}")
    def get_current_token(self) -> Optional[str]:
        """
        获取当前请求的Token
        :return: Token字符串
        """
        # 从当前SecurityContext获取Token
        # 在实际实现中，这里会从Spring Security上下文获取
        pass
class TokenRelayService:
    """
    Token传递服务
    """
    def __init__(self, jwt_util: JWTTokenUtil):
        self.jwt_util = jwt_util
    def relay_token(self, target_service_url: str, request_data: Dict):
        """
        传递Token到目标服务
        :param target_service_url: 目标服务URL
        :param request_data: 请求数据
        :return: 响应结果
        """
        # 获取当前Token
        current_token = self.get_current_token()
        # 构建带认证头的请求
        headers = {
            "Authorization": f"Bearer {current_token}",
            "Content-Type": "application/json"
        }
        # 发送请求到目标服务
        import requests
        response = requests.post(target_service_url, json=request_data, headers=headers)
        return response.json()

class TokenRefreshService:
    """
    Token刷新服务
    """
    def __init__(self, jwt_util: JWTTokenUtil, redis_client):
        """
        初始化Token刷新服务
        :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存储到Redis，设置过期时间
        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:
            # 验证刷新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: 注销结果
        """
        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: 角色列表
        """
        # 在实际应用中，这里会查询数据库获取用户角色
        return ["ROLE_USER"]

class RbacAuthorityService:
    """
    基于RBAC的权限服务
    """
    def __init__(self, redis_client):
        self.redis = redis_client
    def has_permission(self, user_info: Dict, resource: str, action: str) -> bool:
        """
        检查用户是否有权限执行指定操作
        :param user_info: 用户信息
        :param resource: 资源
        :param action: 操作
        :return: 有权限返回True，否则返回False
        """
        user_id = user_info.get("user_id")
        roles = user_info.get("roles", [])
        # 检查用户角色是否具有访问资源的权限
        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:
        """
        获取用户所有权限
        :param user_id: 用户ID
        :return: 权限字典
        """
        # 获取用户角色
        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: 角色列表
        """
        # 在实际应用中，这里会查询数据库或缓存
        return ["ROLE_USER", "ROLE_ADMIN"]
    def get_role_permissions(self, role: str) -> Dict:
        """
        获取角色权限
        :param role: 角色
        :return: 权限字典
        """
        # 在实际应用中，这里会查询数据库或缓存
        return {
            "user:read": True,
            "user:write": role == "ROLE_ADMIN",
            "admin:manage": role == "ROLE_ADMIN"
        }
class PermissionEvaluator:
    """
    权限评估器
    """
    def __init__(self, rbac_service: RbacAuthorityService):
        self.rbac_service = rbac_service
    def evaluate(self, user_info: Dict, request_resource: str, request_action: str) -> bool:
        """
        评估用户权限
        :param user_info: 用户信息
        :param request_resource: 请求资源
        :param request_action: 请求操作
        :return: 有权限返回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):
        """
        配置安全设置
        """
        # 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安全
        """
        # 在实际Spring Boot应用中，这对应SecurityConfig类
        pass
    def get_security_filter_chain(self):
        """
        获取安全过滤器链
        :return: 过滤器链配置
        """
        # 定义不需要认证的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": ["/**"],  # 其他所有路径需要认证
            "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安全配置对象
        """
        # 配置不需要认证的路径
        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认证过滤器
        http.add_filter_before(
            JwtAuthenticationTokenFilter(), 
            UsernamePasswordAuthenticationFilter
        )
        # 禁用CSRF，因为使用Token认证
        http.csrf().disable()
        # 禁用session，因为使用Token认证
        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:
    """
    认证控制器
    """
    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: 认证结果
        """
        username = login_form.get("username")
        password = login_form.get("password")
        # 验证用户凭据
        user = self.user_details_service.authenticate(username, password)
        if not user:
            return {
                "success": False,
                "message": "用户名或密码错误"
            }
        # 生成访问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无效或已过期"
            }
    def logout(self, request):
        """
        用户登出
        :param request: HTTP请求
        :return: 登出结果
        """
        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字符串
        """
        # 在实际应用中，可以将Token存储到Redis中，设置过期时间
        # 这样可以防止Token在过期前被使用
        pass

class ProtectedResourceController:
    """
    受保护资源控制器
    """
    def __init__(self, permission_evaluator: PermissionEvaluator):
        self.permission_evaluator = permission_evaluator
    def get_user_info(self, request):
        """
        获取用户信息
        :param request: HTTP请求
        :return: 用户信息
        """
        user_info = request.user_info  # 从请求上下文获取用户信息
        return {
            "success": True,
            "data": user_info
        }
    def get_sensitive_data(self, request):
        """
        获取敏感数据
        :param request: HTTP请求
        :return: 敏感数据
        """
        user_info = request.user_info
        # 检查用户权限
        if not self.permission_evaluator.evaluate(user_info, "sensitive_data", "read"):
            return {
                "success": False,
                "message": "没有访问权限"
            }
        # 返回敏感数据
        return {
            "success": True,
            "data": self.fetch_sensitive_data()
        }
    def fetch_sensitive_data(self):
        """
        获取敏感数据
        :return: 敏感数据
        """
        # 在实际应用中，这里会查询数据库获取敏感数据
        return {
            "id": "12345",
            "name": "敏感数据",
            "content": "这是需要特殊权限才能访问的敏感信息"
        }

import bcrypt
from passlib.hash import pbkdf2_sha256
class PasswordSecurityUtil:
    """
    密码安全工具类
    """
    @staticmethod
    def encode_password(raw_password: str) -> str:
        """
        加密密码
        :param raw_password: 原始密码
        :return: 加密后的密码
        """
        # 使用bcrypt进行密码加密
        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:
        """
        验证密码
        :param raw_password: 原始密码
        :param encoded_password: 加密后的密码
        :return: 验证结果
        """
        return bcrypt.checkpw(raw_password.encode('utf-8'), encoded_password.encode('utf-8'))
    @staticmethod
    def validate_password_strength(password: str) -> Dict[str, bool]:
        """
        验证密码强度
        :param password: 密码
        :return: 验证结果
        """
        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:
    """
    优化的Token验证器
    """
    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]:
        """
        验证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)  # 在实际应用中应使用安全的序列化方式
        # 缓存未命中，验证Token
        user_info = self.jwt_util.get_user_info_from_token(token)
        if user_info:
            # 缓存验证结果
            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:
    """
    安全审计日志记录器
    """
    def __init__(self):
        self.logger = logging.getLogger("security-audit")
        self.logger.setLevel(logging.INFO)
    def log_authentication_success(self, username: str, ip_address: str):
        """
        记录认证成功事件
        :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):
        """
        记录认证失败事件
        :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):
        """
        记录授权失败事件
        :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()}")