JWT 安全机制与最佳实践指南

2.2 JWT 声明类型

JWT 标准定义了多种声明类型：

class JWTClaims:
    """JWT 声明类型定义"""
    # 注册声明（Registered Claims）
    ISSUER = 'iss'      # 签发者
    SUBJECT = 'sub'     # 主题
    AUDIENCE = 'aud'    # 接收方
    EXPIRATION_TIME = 'exp'  # 过期时间
    NOT_BEFORE = 'nbf'   # 生效时间
    ISSUED_AT = 'iat'    # 签发时间
    JWT_ID = 'jti'       # JWT ID

    # 公共声明（Public Claims）
    # 可以自定义，但应避免冲突
    # 私有声明（Private Claims）
    # 双方约定的自定义声明

    def create_secure_jwt_payload(user_id: str, username: str, roles: list) -> Dict[str, Any]:
        """创建安全的 JWT 负载"""
        import time
        now = int(time.time())
        return {
            # 注册声明
            JWTClaims.ISSUER: 'auth-server',
            JWTClaims.SUBJECT: user_id,
            JWTClaims.EXPIRATION_TIME: now + 3600,  # 1 小时后过期
            JWTClaims.ISSUED_AT: now,
            JWTClaims.JWT_ID: base64.urlsafe_b64encode(
                hmac.new(str(now).encode(), user_id.encode(), hashlib.sha256).digest()
            ).decode().rstrip('='),
            # 私有声明
            'username': username,
            'roles': roles,
            'permissions': ['read', 'write']
        }

# 创建安全负载示例
secure_payload = create_secure_jwt_payload('user123', 'john_doe', ['USER', 'ADMIN'])
print(f"安全 JWT 负载：{secure_payload}")

3. JWT 加密算法

3.1 对称加密算法（HS256, HS384, HS512）

class SymmetricJWT:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
        self.algorithms = {
            'HS256': hashlib.sha256,
            'HS384': hashlib.sha384,
            'HS512': hashlib.sha512
        }

    def encode(self, payload: Dict[str, Any], algorithm: str = 'HS256') -> str:
        """使用对称算法编码 JWT"""
        if algorithm not in self.algorithms:
            raise ValueError(f"不支持的算法：{algorithm}")
        # 创建头部
        header = {'alg': algorithm, 'typ': 'JWT'}
        # 编码头部和负载
        header_encoded = self._base64_encode(json.dumps(header, separators=(',', ':')))
        payload_encoded = self._base64_encode(json.dumps(payload, separators=(',', ':')))
        # 创建签名
        signature_input = f"{header_encoded}.{payload_encoded}"
        signature = hmac.new(
            self.secret_key.encode(), signature_input.encode(), self.algorithms[algorithm]
        ).digest()
        signature_encoded = self._base64_encode(signature)
        return f"{header_encoded}.{payload_encoded}.{signature_encoded}"

    def decode(self, token: str, algorithm: str = 'HS256') -> Optional[Dict[str, Any]]:
        """使用对称算法解码 JWT"""
        if algorithm not in self.algorithms:
            raise ValueError(f"不支持的算法：{algorithm}")
        try:
            parts = token.split('.')
            if len(parts) != 3:
                return None
            header_encoded, payload_encoded, signature_encoded = parts
            # 验证签名
            signature_input = f"{header_encoded}.{payload_encoded}"
            expected_signature = hmac.new(
                self.secret_key.encode(), signature_input.encode(), self.algorithms[algorithm]
            ).digest()
            expected_signature_encoded = self._base64_encode(expected_signature)
            if not hmac.compare_digest(signature_encoded, expected_signature_encoded):
                return None
            # 解码负载
            payload_json = base64.urlsafe_b64decode(payload_encoded + '==')
            return json.loads(payload_json)
        except Exception:
            return None

    def _base64_encode(self, data) -> str:
        """Base64 URL 安全编码"""
        if isinstance(data, str):
            data = data.encode()
        encoded = base64.urlsafe_b64encode(data).decode()
        return encoded.rstrip('=')

# 对称加密示例
symmetric_jwt = SymmetricJWT('my_secret_key')
payload = {'user_id': '123', 'username': 'john', 'exp': int(time.time()) + 3600}
token = symmetric_jwt.encode(payload)
print(f"对称加密 JWT: {token}")
decoded_payload = symmetric_jwt.decode(token)
print(f"解码结果：{decoded_payload}")

3.2 非对称加密算法（RS256, RS384, RS512）

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.exceptions import InvalidSignature

class AsymmetricJWT:
    def __init__(self, private_key_pem: str = None, public_key_pem: str = None):
        if private_key_pem:
            self.private_key = serialization.load_pem_private_key(
                private_key_pem.encode(), password=None
            )
        else:
            self.private_key = None
        if public_key_pem:
            self.public_key = serialization.load_pem_public_key(public_key_pem.encode())
        else:
            self.public_key = None

    @classmethod
    def generate_key_pair(cls):
        """生成 RSA 密钥对"""
        private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        public_key = private_key.public_key()
        private_pem = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption()
        ).decode()
        public_pem = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        ).decode()
        return cls(private_pem, public_pem)

    def encode(self, payload: Dict[str, Any], algorithm: str = 'RS256') -> str:
        """使用非对称算法编码 JWT"""
        if not self.private_key:
            raise ValueError("没有可用的私钥")
        header = {'alg': algorithm, 'typ': 'JWT'}
        header_encoded = self._base64_encode(json.dumps(header, separators=(',', ':')))
        payload_encoded = self._base64_encode(json.dumps(payload, separators=(',', ':')))
        signature_input = f"{header_encoded}.{payload_encoded}"
        # 使用私钥签名
        signature = self.private_key.sign(
            signature_input.encode(), padding.PKCS1v15(), hashes.SHA256()
        )
        signature_encoded = self._base64_encode(signature)
        return f"{header_encoded}.{payload_encoded}.{signature_encoded}"

    def decode(self, token: str, algorithm: str = 'RS256') -> Optional[Dict[str, Any]]:
        """使用非对称算法解码 JWT"""
        if not self.public_key:
            raise ValueError("没有可用的公钥")
        try:
            parts = token.split('.')
            if len(parts) != 3:
                return None
            header_encoded, payload_encoded, signature_encoded = parts
            # 验证签名
            signature_input = f"{header_encoded}.{payload_encoded}"
            signature = base64.urlsafe_b64decode(signature_encoded + '==')
            try:
                self.public_key.verify(
                    signature, signature_input.encode(), padding.PKCS1v15(), hashes.SHA256()
                )
            except InvalidSignature:
                return None
            # 解码负载
            payload_json = base64.urlsafe_b64decode(payload_encoded + '==')
            return json.loads(payload_json)
        except Exception:
            return None

    def _base64_encode(self, data) -> str:
        """Base64 URL 安全编码"""
        if isinstance(data, str):
            data = data.encode()
        encoded = base64.urlsafe_b64encode(data).decode()
        return encoded.rstrip('=')

# 非对称加密示例
asymmetric_jwt = AsymmetricJWT.generate_key_pair()
payload = {'user_id': '456', 'username': 'jane', 'exp': int(time.time()) + 3600}
token = asymmetric_jwt.encode(payload)
print(f"非对称加密 JWT: {token[:50]}...")
decoded_payload = asymmetric_jwt.decode(token)
print(f"解码结果：{decoded_payload}")

4. JWT 在 Spring Security 中的应用

4.1 Spring Security JWT 配置

在 auth 项目中，JWT 被用作 OAuth2 的访问令牌格式：

class SpringSecurityJWTConfig:
    def __init__(self, signing_key: str):
        self.signing_key = signing_key
        self.jwt_util = SymmetricJWT(signing_key)

    def create_access_token(self, user_details: Dict[str, Any], client_id: str, scopes: list) -> str:
        """创建访问令牌"""
        import time
        now = int(time.time())
        payload = {
            # OAuth2 标准声明
            'user_name': user_details.get('username'),
            'client_id': client_id,
            'scope': scopes,
            'authorities': user_details.get('authorities', []),
            # JWT 标准声明
            'exp': now + 43200,  # 12 小时
            'iat': now,
            'jti': base64.urlsafe_b64encode(
                hmac.new(str(now).encode(), f"{user_details.get('username')}_{client_id}".encode(), hashlib.sha256).digest()
            ).decode().rstrip('='),
            # 自定义声明
            'token_type': 'access_token'
        }
        return self.jwt_util.encode(payload)

    def create_refresh_token(self, user_details: Dict[str, Any], client_id: str) -> str:
        """创建刷新令牌"""
        import time
        now = int(time.time())
        payload = {
            'user_name': user_details.get('username'),
            'client_id': client_id,
            'exp': now + 2592000,  # 30 天
            'iat': now,
            'jti': base64.urlsafe_b64encode(
                hmac.new(str(now).encode(), f"refresh_{user_details.get('username')}_{client_id}".encode(), hashlib.sha256).digest()
            ).decode().rstrip('='),
            'token_type': 'refresh_token'
        }
        return self.jwt_util.encode(payload)

    def validate_token(self, token: str) -> Optional[Dict[str, Any]]:
        """验证令牌"""
        payload = self.jwt_util.decode(token)
        if not payload:
            return None
        # 检查过期时间
        exp = payload.get('exp')
        if exp and int(time.time()) > exp:
            return None
        return payload

# 使用示例
jwt_config = SpringSecurityJWTConfig('123456')  # 从配置获取
user_details = {'username': 'admin', 'authorities': ['ADMIN', 'USER']}
access_token = jwt_config.create_access_token(
    user_details=user_details, client_id='test_client', scopes=['read', 'write']
)
refresh_token = jwt_config.create_refresh_token(
    user_details=user_details, client_id='test_client'
)
print(f"访问令牌：{access_token[:50]}...")
print(f"刷新令牌：{refresh_token[:50]}...")
# 验证令牌
validated_payload = jwt_config.validate_token(access_token)
print(f"令牌验证结果：{validated_payload is not None}")

4.2 JWT Token Store 实现

class JwtTokenStore:
    def __init__(self, jwt_converter):
        self.jwt_converter = jwt_converter
        self.token_store = {}  # 在实际应用中可能是 Redis 等外部存储

    def store_access_token(self, token, authentication):
        """存储访问令牌（JWT 模式下通常不存储，仅验证）"""
        # JWT 是自包含的，不需要存储在服务器端
        # 但在某些场景下可能需要黑名单或撤销列表
        token_value = token.get('value')
        self.token_store[token_value] = {
            'authentication': authentication,
            'expiration': token.get('expiration')
        }

    def read_access_token(self, token_value: str):
        """读取访问令牌"""
        # JWT 模式下，直接解析令牌内容
        return self.jwt_converter.decode(token_value)

    def remove_access_token(self, token):
        """移除访问令牌"""
        token_value = token.get('value')
        if token_value in self.token_store:
            del self.token_store[token_value]


class JwtAccessTokenConverter:
    def __init__(self, signing_key: str):
        self.signing_key = signing_key
        self.jwt_util = SymmetricJWT(signing_key)

    def encode(self, authentication: Dict[str, Any]) -> str:
        """编码认证信息为 JWT"""
        # 将 Spring Security 的 Authentication 对象转换为 JWT
        payload = self.authentication_to_payload(authentication)
        return self.jwt_util.encode(payload)

    def decode(self, token: str) -> Dict[str, Any]:
        """解码 JWT 为认证信息"""
        payload = self.jwt_util.decode(token)
        if not payload:
            raise ValueError("无效的 JWT 令牌")
        return self.payload_to_authentication(payload)

    def authentication_to_payload(self, authentication: Dict[str, Any]) -> Dict[str, Any]:
        """将认证信息转换为 JWT 负载"""
        import time
        now = int(time.time())
        return {
            'user_name': authentication.get('user_name'),
            'client_id': authentication.get('client_id'),
            'scope': authentication.get('scope', []),
            'authorities': authentication.get('authorities', []),
            'exp': now + 43200,  # 12 小时
            'iat': now,
            'jti': base64.urlsafe_b64encode(
                hmac.new(str(now).encode(), authentication.get('user_name', '').encode(), hashlib.sha256).digest()
            ).decode().rstrip('='),
            'token_type': 'access_token'
        }

    def payload_to_authentication(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        """将 JWT 负载转换为认证信息"""
        return {
            'user_name': payload.get('user_name'),
            'client_id': payload.get('client_id'),
            'scope': payload.get('scope', []),
            'authorities': payload.get('authorities', []),
            'exp': payload.get('exp'),
            'iat': payload.get('iat'),
            'jti': payload.get('jti'),
            'token_type': payload.get('token_type')
        }

5. JWT 安全漏洞与防护

5.1 "none"算法漏洞

class JWSNoneAttackProtection:
    """防护 JWS none 算法攻击"""
    ALLOWED_ALGORITHMS = {'HS256', 'HS384', 'HS512', 'RS256', 'RS384', 'RS512'}

    def decode_with_algorithm_validation(self, token: str, expected_algorithms: set) -> Optional[Dict[str, Any]]:
        """带算法验证的 JWT 解码"""
        try:
            parts = token.split('.')
            if len(parts) != 3:
                return None
            header_encoded = parts[0]
            payload_encoded = parts[1]
            # 解码头部以检查算法
            header_json = base64.urlsafe_b64decode(header_encoded + '==')
            header = json.loads(header_json)
            algorithm = header.get('alg')
            # 检查算法是否在允许列表中
            if algorithm not in self.ALLOWED_ALGORITHMS:
                print(f"检测到不安全的算法：{algorithm}")
                return None
            # 检查算法是否在预期列表中
            if algorithm not in expected_algorithms:
                print(f"算法不在预期列表中：{algorithm}")
                return None
            # 继续正常的解码过程
            # ... 实际解码逻辑 ...
            return self.decode_jwt_payload(payload_encoded)
        except Exception as e:
            print(f"JWT 解码失败：{e}")
            return None

    def decode_jwt_payload(self, payload_encoded: str) -> Dict[str, Any]:
        """解码 JWT 负载"""
        payload_json = base64.urlsafe_b64decode(payload_encoded + '==')
        return json.loads(payload_json)

# 使用示例
protection = JWSNoneAttackProtection()
# 模拟一个使用"none"算法的恶意 JWT（头部部分）
malicious_header = base64.urlsafe_b64encode(
    json.dumps({'alg': 'none', 'typ': 'JWT'}).encode()
).decode().rstrip('=')
# 防护机制会拒绝这种令牌
result = protection.decode_with_algorithm_validation(f"{malicious_header}.payload.signature", {'HS256'})
print(f"防护结果：{result}")

5.2 密钥混淆攻击防护

class KeyConfusionAttackProtection:
    """防护密钥混淆攻击（RS256 vs HS256）"""
    def __init__(self, rsa_public_key_pem: str, symmetric_secret: str):
        from cryptography.hazmat.primitives import serialization
        self.rsa_public_key = serialization.load_pem_public_key(rsa_public_key_pem.encode())
        self.symmetric_secret = symmetric_secret

    def safe_decode(self, token: str, expected_algorithm: str) -> Optional[Dict[str, Any]]:
        """安全解码 JWT，防止算法混淆"""
        try:
            parts = token.split('.')
            if len(parts) != 3:
                return None
            header_encoded, payload_encoded, signature_encoded = parts
            # 解码头部
            header_json = base64.urlsafe_b64decode(header_encoded + '==')
            header = json.loads(header_json)
            algorithm = header.get('alg')
            # 验证算法与预期是否一致
            if algorithm != expected_algorithm:
                print(f"算法不匹配：期望 {expected_algorithm}, 实际 {algorithm}")
                return None
            # 根据算法类型使用相应的验证方法
            if algorithm.startswith('RS'):
                # RSA 算法：使用公钥验证
                return self._verify_rsa_token(token, algorithm)
            elif algorithm.startswith('HS'):
                # HMAC 算法：使用密钥验证
                return self._verify_hmac_token(token, algorithm)
            else:
                return None
        except Exception as e:
            print(f"安全解码失败：{e}")
            return None

    def _verify_rsa_token(self, token: str, algorithm: str) -> Optional[Dict[str, Any]]:
        """验证 RSA 签名的 JWT"""
        from cryptography.hazmat.primitives.asymmetric import padding
        from cryptography.exceptions import InvalidSignature
        parts = token.split('.')
        header_encoded, payload_encoded, signature_encoded = parts
        signature_input = f"{header_encoded}.{payload_encoded}"
        signature = base64.urlsafe_b64decode(signature_encoded + '==')
        try:
            self.rsa_public_key.verify(
                signature, signature_input.encode(), padding.PKCS1v15(), self._get_hash_algorithm(algorithm)
            )
            # 验证成功，解码负载
            payload_json = base64.urlsafe_b64decode(payload_encoded + '==')
            return json.loads(payload_json)
        except InvalidSignature:
            return None

    def _verify_hmac_token(self, token: str, algorithm: str) -> Optional[Dict[str, Any]]:
        """验证 HMAC 签名的 JWT"""
        parts = token.split('.')
        header_encoded, payload_encoded, signature_encoded = parts
        signature_input = f"{header_encoded}.{payload_encoded}"
        expected_signature = hmac.new(
            self.symmetric_secret.encode(), signature_input.encode(), self._get_hash_algorithm(algorithm)
        ).digest()
        expected_signature_encoded = base64.urlsafe_b64encode(expected_signature).decode().rstrip('=')
        if not hmac.compare_digest(signature_encoded, expected_signature_encoded):
            return None
        # 验证成功，解码负载
        payload_json = base64.urlsafe_b64decode(payload_encoded + '==')
        return json.loads(payload_json)

    def _get_hash_algorithm(self, algorithm: str):
        """根据算法名称获取哈希算法"""
        if algorithm in ['RS256', 'HS256']:
            return hashes.SHA256()
        elif algorithm in ['RS384', 'HS384']:
            return hashes.SHA384()
        elif algorithm in ['RS512', 'HS512']:
            return hashes.SHA512()
        else:
            raise ValueError(f"不支持的算法：{algorithm}")

# 生成测试密钥对
test_asymmetric = AsymmetricJWT.generate_key_pair()
protection = KeyConfusionAttackProtection(
    test_asymmetric.public_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    ).decode(),
    'test_secret'
)
# 测试安全解码
payload = {'user': 'test', 'exp': int(time.time()) + 3600}
token = test_asymmetric.encode(payload)
safe_result = protection.safe_decode(token, 'RS256')
print(f"安全解码结果：{safe_result}")

6. JWT 最佳实践

6.1 令牌生命周期管理

import time
from datetime import datetime, timedelta
from typing import Dict, Any, Optional

class JWTLifecycleManager:
    def __init__(self, signing_key: str):
        self.signing_key = signing_key
        self.jwt_util = SymmetricJWT(signing_key)
        self.blacklist = {}  # 令牌黑名单

    def create_token_with_metadata(self, user_id: str, username: str, roles: list, expires_in: int = 3600) -> Dict[str, Any]:
        """创建带元数据的令牌"""
        now = int(time.time())
        payload = {
            'sub': user_id,
            'username': username,
            'roles': roles,
            'iat': now,
            'exp': now + expires_in,
            'nbf': now,  # 生效时间
            'jti': self._generate_token_id(user_id, now),
            'iss': 'auth-server',
            'aud': 'resource-server'
        }
        token = self.jwt_util.encode(payload)
        return {
            'access_token': token,
            'token_type': 'Bearer',
            'expires_in': expires_in,
            'refresh_token': self._create_refresh_token(user_id, username),
            'scope': 'read write'
        }

    def _generate_token_id(self, user_id: str, timestamp: int) -> str:
        """生成唯一令牌 ID"""
        import secrets
        random_part = secrets.token_urlsafe(8)
        return f"{user_id}_{timestamp}_{random_part}"

    def _create_refresh_token(self, user_id: str, username: str) -> str:
        """创建刷新令牌"""
        now = int(time.time())
        refresh_payload = {
            'sub': user_id,
            'username': username,
            'iat': now,
            'exp': now + 2592000,  # 30 天
            'jti': self._generate_token_id(f"refresh_{user_id}", now),
            'token_type': 'refresh'
        }
        return self.jwt_util.encode(refresh_payload)

    def validate_token(self, token: str, require_fresh: bool = True) -> Optional[Dict[str, Any]]:
        """验证令牌"""
        payload = self.jwt_util.decode(token)
        if not payload:
            return None
        now = int(time.time())
        # 检查过期时间
        exp = payload.get('exp')
        if exp and now > exp:
            return None
        # 检查生效时间
        nbf = payload.get('nbf')
        if nbf and now < nbf:
            return None
        # 检查令牌是否在黑名单中
        jti = payload.get('jti')
        if jti and jti in self.blacklist:
            if now < self.blacklist[jti]['expires_at']:
                return None
        # 检查是否需要新鲜令牌（防止重放攻击）
        if require_fresh:
            iat = payload.get('iat')
            if iat and (now - iat) > 3600:  # 1 小时内的令牌被认为是新鲜的
                pass
        # 可以选择拒绝过旧的令牌
        return payload

    def revoke_token(self, token: str, grace_period: int = 300) -> bool:
        """撤销令牌"""
        payload = self.jwt_util.decode(token)
        if not payload:
            return False
        jti = payload.get('jti')
        if not jti:
            return False
        # 将令牌加入黑名单，设置宽限期
        self.blacklist[jti] = {
            'expires_at': int(time.time()) + grace_period,
            'revoked_at': int(time.time())
        }
        return True

    def refresh_access_token(self, refresh_token: str, new_expires_in: int = 3600) -> Optional[Dict[str, Any]]:
        """刷新访问令牌"""
        refresh_payload = self.jwt_util.decode(refresh_token)
        if not refresh_payload:
            return None
        # 验证刷新令牌类型
        if refresh_payload.get('token_type') != 'refresh':
            return None
        # 检查刷新令牌是否过期
        now = int(time.time())
        exp = refresh_payload.get('exp')
        if exp and now > exp:
            return None
        # 创建新的访问令牌
        user_id = refresh_payload.get('sub')
        username = refresh_payload.get('username')
        roles = refresh_payload.get('roles', [])
        return self.create_token_with_metadata(user_id, username, roles, new_expires_in)

# 使用示例
lifecycle_manager = JWTLifecycleManager('secure_signing_key')
# 创建令牌
token_result = lifecycle_manager.create_token_with_metadata(
    user_id='user123', username='john_doe', roles=['USER', 'ADMIN'], expires_in=3600
)
print(f"访问令牌：{token_result['access_token'][:50]}...")
print(f"刷新令牌：{token_result['refresh_token'][:50]}...")
# 验证令牌
validation_result = lifecycle_manager.validate_token(token_result['access_token'])
print(f"令牌验证结果：{validation_result is not None}")
# 刷新令牌
refresh_result = lifecycle_manager.refresh_access_token(token_result['refresh_token'])
if refresh_result:
    print(f"刷新后的令牌：{refresh_result['access_token'][:50]}...")

6.2 令牌存储与缓存

import redis
from typing import Dict, Any, Optional

class JWTCacheManager:
    def __init__(self, redis_host: str = 'localhost', redis_port: int = 6379):
        try:
            self.redis_client = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
            self.redis_client.ping()  # 测试连接
        except:
            print("Redis 连接失败，使用内存缓存")
            self.redis_client = None
            self.memory_cache = {}

    def cache_token_validation(self, token_hash: str, is_valid: bool, ttl: int = 300):
        """缓存令牌验证结果"""
        if self.redis_client:
            self.redis_client.setex(f"token_valid:{token_hash}", ttl, str(is_valid))
        else:
            self.memory_cache[f"token_valid:{token_hash}"] = {
                'value': is_valid,
                'expires_at': time.time() + ttl
            }

    def get_cached_validation(self, token_hash: str) -> Optional[bool]:
        """获取缓存的验证结果"""
        if self.redis_client:
            cached = self.redis_client.get(f"token_valid:{token_hash}")
            return cached.lower() == 'true' if cached else None
        else:
            cache_key = f"token_valid:{token_hash}"
            if cache_key in self.memory_cache:
                cached = self.memory_cache[cache_key]
                if time.time() < cached['expires_at']:
                    return cached['value']
                else:
                    del self.memory_cache[cache_key]
            # 清除过期缓存
            return None

    def cache_user_permissions(self, user_id: str, permissions: list, ttl: int = 3600):
        """缓存用户权限"""
        if self.redis_client:
            self.redis_client.setex(f"user_perms:{user_id}", ttl, json.dumps(permissions))
        else:
            self.memory_cache[f"user_perms:{user_id}"] = {
                'value': permissions,
                'expires_at': time.time() + ttl
            }

    def get_cached_permissions(self, user_id: str) -> Optional[list]:
        """获取缓存的用户权限"""
        if self.redis_client:
            cached = self.redis_client.get(f"user_perms:{user_id}")
            return json.loads(cached) if cached else None
        else:
            cache_key = f"user_perms:{user_id}"
            if cache_key in self.memory_cache:
                cached = self.memory_cache[cache_key]
                if time.time() < cached['expires_at']:
                    return cached['value']
                else:
                    del self.memory_cache[cache_key]
            return None

# 使用示例
cache_manager = JWTCacheManager()
# 模拟缓存令牌验证结果
token_hash = hashlib.sha256(b"example_token").hexdigest()
cache_manager.cache_token_validation(token_hash, True)
cached_result = cache_manager.get_cached_validation(token_hash)
print(f"缓存验证结果：{cached_result}")
# 缓存用户权限
cache_manager.cache_user_permissions('user123', ['read', 'write', 'admin'])
cached_perms = cache_manager.get_cached_permissions('user123')
print(f"缓存的用户权限：{cached_perms}")

7. 实践案例：构建安全的 JWT 系统

7.1 完整的 JWT 认证服务

class SecureJWTAuthService:
    def __init__(self, signing_key: str):
        self.signing_key = signing_key
        self.jwt_util = SymmetricJWT(signing_key)
        self.lifecycle_manager = JWTLifecycleManager(signing_key)
        self.cache_manager = JWTCacheManager()
        self.user_store = {}  # 模拟用户存储

    def register_user(self, user_id: str, username: str, password: str, roles: list):
        """注册用户"""
        import hashlib
        hashed_password = hashlib.sha256(password.encode()).hexdigest()
        self.user_store[user_id] = {
            'username': username,
            'password': hashed_password,
            'roles': roles,
            'permissions': self._derive_permissions_from_roles(roles)
        }

    def _derive_permissions_from_roles(self, roles: list) -> list:
        """根据角色推导权限"""
        role_permissions = {
            'USER': ['read_profile', 'update_profile'],
            'ADMIN': ['read_profile', 'update_profile', 'manage_users', 'system_admin'],
            'MODERATOR': ['read_profile', 'moderate_content']
        }
        permissions = set()
        for role in roles:
            if role in role_permissions:
                permissions.update(role_permissions[role])
        return list(permissions)

    def authenticate_user(self, username: str, password: str) -> Optional[Dict[str, Any]]:
        """用户认证"""
        import hashlib
        hashed_password = hashlib.sha256(password.encode()).hexdigest()
        for user_id, user_info in self.user_store.items():
            if user_info['username'] == username and user_info['password'] == hashed_password:
                return {
                    'user_id': user_id,
                    'username': username,
                    'roles': user_info['roles'],
                    'permissions': user_info['permissions']
                }
        return None

    def login(self, username: str, password: str) -> Optional[Dict[str, Any]]:
        """用户登录，返回 JWT 令牌"""
        user_info = self.authenticate_user(username, password)
        if not user_info:
            return None
        # 创建令牌
        token_result = self.lifecycle_manager.create_token_with_metadata(
            user_id=user_info['user_id'],
            username=user_info['username'],
            roles=user_info['roles']
        )
        # 缓存用户权限
        self.cache_manager.cache_user_permissions(
            user_info['user_id'], user_info['permissions']
        )
        return token_result

    def verify_token(self, token: str) -> Optional[Dict[str, Any]]:
        """验证令牌"""
        # 首先检查缓存
        token_hash = hashlib.sha256(token.encode()).hexdigest()
        cached_result = self.cache_manager.get_cached_validation(token_hash)
        if cached_result is False:
            return None  # 明确的无效令牌
        if cached_result is True:
            # 从缓存中获取用户信息
            payload = self.jwt_util.decode(token)
            if payload:
                user_perms = self.cache_manager.get_cached_permissions(payload.get('sub'))
                if user_perms:
                    payload['permissions'] = user_perms
                return payload
        # 缓存中没有或已过期，进行完整验证
        payload = self.lifecycle_manager.validate_token(token)
        if payload:
            # 验证成功，缓存结果
            self.cache_manager.cache_token_validation(token_hash, True)
            # 获取并缓存用户权限
            user_perms = self.cache_manager.get_cached_permissions(payload.get('sub'))
            if not user_perms:
                user_info = self.user_store.get(payload.get('sub'))
                if user_info:
                    user_perms = user_info.get('permissions', [])
                    self.cache_manager.cache_user_permissions(payload.get('sub'), user_perms)
            if user_perms:
                payload['permissions'] = user_perms
            return payload

    def has_permission(self, token: str, required_permission: str) -> bool:
        """检查用户是否有特定权限"""
        payload = self.verify_token(token)
        if not payload:
            return False
        permissions = payload.get('permissions', [])
        return required_permission in permissions

# 使用示例
auth_service = SecureJWTAuthService('production_signing_key')
# 注册用户
auth_service.register_user(
    user_id='user001', username='admin_user', password='secure_password', roles=['ADMIN']
)
# 用户登录
login_result = auth_service.login('admin_user', 'secure_password')
if login_result:
    print(f"登录成功，访问令牌：{login_result['access_token'][:50]}...")
    # 验证令牌
    verification_result = auth_service.verify_token(login_result['access_token'])
    print(f"令牌验证结果：{verification_result is not None}")
    # 检查权限
    has_admin_perm = auth_service.has_permission(login_result['access_token'], 'system_admin')
    print(f"具有管理员权限：{has_admin_perm}")
else:
    print("登录失败")

8. 安全监控与日志

8.1 JWT 安全监控

import logging
from datetime import datetime
from typing import Dict, Any

class JWTSecurityMonitor:
    def __init__(self):
        self.logger = logging.getLogger('jwt_security')
        self.logger.setLevel(logging.INFO)
        # 创建文件处理器
        handler = logging.FileHandler('jwt_security.log')
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        # 统计信息
        self.stats = {
            'total_tokens_issued': 0,
            'total_tokens_verified': 0,
            'invalid_tokens': 0,
            'suspicious_activities': 0
        }

    def log_token_issuance(self, user_id: str, client_id: str, token_type: str):
        """记录令牌颁发"""
        self.stats['total_tokens_issued'] += 1
        self.logger.info(f"TOKEN_ISSUED - User: {user_id}, Client: {client_id}, Type: {token_type}")

    def log_token_verification(self, token_hash: str, is_valid: bool, user_id: str = None):
        """记录令牌验证"""
        self.stats['total_tokens_verified'] += 1
        if not is_valid:
            self.stats['invalid_tokens'] += 1
        status = "VALID" if is_valid else "INVALID"
        self.logger.info(f"TOKEN_VERIFIED - Hash: {token_hash[:16]}, Status: {status}, User: {user_id}")

    def log_suspicious_activity(self, activity_type: str, details: Dict[str, Any]):
        """记录可疑活动"""
        self.stats['suspicious_activities'] += 1
        self.logger.warning(f"SUSPICIOUS_ACTIVITY - Type: {activity_type}, Details: {details}")

    def log_permission_denied(self, user_id: str, required_permission: str, token_info: str):
        """记录权限拒绝"""
        self.logger.info(f"PERMISSION_DENIED - User: {user_id}, Permission: {required_permission}, Token: {token_info}")

    def get_security_stats(self) -> Dict[str, int]:
        """获取安全统计信息"""
        return self.stats.copy()

# 使用示例
monitor = JWTSecurityMonitor()
# 模拟安全事件
monitor.log_token_issuance('user123', 'web_client', 'access_token')
monitor.log_token_verification('hash123abc', True, 'user123')
monitor.log_suspicious_activity('multiple_failed_attempts', {'user_id': 'attacker', 'attempts': 10, 'timeframe': '5 minutes'})
print(f"安全统计：{monitor.get_security_stats()}")

9. 注意事项

1. 密钥安全：JWT 签名密钥必须安全存储，不应硬编码在代码中
2. 过期时间：合理设置令牌过期时间，平衡安全性和用户体验
3. 算法选择：在生产环境中优先使用非对称算法
4. 令牌撤销：实现令牌黑名单机制以应对安全事件
5. 传输安全：始终通过 HTTPS 传输 JWT 令牌

10. 常见问题解答

Q1: JWT 的安全性如何？
A1: JWT 的安全性取决于实现方式。正确实现的 JWT 是安全的，但需要注意算法选择、密钥管理、过期时间设置等安全要点。

Q2: 如何处理 JWT 的撤销？
A2: JWT 是无状态的，撤销需要额外机制，如黑名单、短期令牌、令牌存储等。

Q3: JWT vs Session 的优缺点？
A3: JWT 无状态、可扩展，但难以撤销；Session 有状态、易管理，但扩展性较差。

11. 总结

JWT 作为一种开放标准，为现代 Web 应用和微服务架构提供了强大的认证和授权机制。通过本文的详细分析和实践示例，开发者可以构建安全、可靠的 JWT 系统。

在实际项目中，应根据具体需求选择合适的算法、合理设置令牌策略，并持续关注安全最佳实践。

12. 扩展阅读

1. JWT RFC 7519
2. JWS RFC 7515
3. JWE RFC 7516
4. JWT 安全最佳实践

参考资料

1. JWT 规范文档
2. OAuth2 与 JWT 集成指南
3. Spring Security JWT 实现
4. JWT 安全漏洞与防护措施