08e · JWT：三段结构、四种绕过，和吊销这件难事

# 手工拆开一个 JWT —— 不需要密钥，也不需要任何第三方库
import base64, json

TOKEN = ("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"
         ".eyJzdWIiOiJ1XzQyIiwidGlkIjoiYWNtZSIsInJvbGUiOiJ1c2VyIiwiZXhwIjoxNzY3MjI1NjAwfQ"
         ".XHv6jpLDce_o34SAysGAdayLTPbNi1Laap5agXqsffE")


def b64d(seg):
    # ⭐ 是 base64url，而且结尾的 = 填充被去掉了，解码前要补回来
    return base64.urlsafe_b64decode(seg + "=" * (-len(seg) % 4))


h, p, s = TOKEN.split(".")
print("头部：", json.loads(b64d(h)))
print("载荷：", json.loads(b64d(p)))       # ⭐ 全程没有密钥，谁捡到 token 谁就能读
print("签名：", len(b64d(s)), "字节 —— 它证明「没被改」，不证明「没被看见」")

头部： {'alg': 'HS256', 'typ': 'JWT'}
载荷： {'sub': 'u_42', 'tid': 'acme', 'role': 'user', 'exp': 1767225600}
签名： 32 字节 —— 它证明「没被改」，不证明「没被看见」

# 💀 两种经典绕过：alg:none 与 算法混淆（RS256 当 HS256 用）
import base64, hashlib, hmac, json

# ⚠️ 教科书 RSA，只为让这段能用纯标准库跑起来；真实项目绝不自己实现签名
P, Q = 954780363788335820345537266331143731120811, 1192651600822816258890983000462819179490453
N, E = P * Q, 65537
D = pow(E, -1, (P - 1) * (Q - 1))                 # 私钥：只有签发方有
PUB = ("%d:%d" % (N, E)).encode()                 # ⭐ 公钥：JWKS 端点上谁都下得到


def b64e(raw): return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
def b64d(seg): return base64.urlsafe_b64decode(seg + "=" * (-len(seg) % 4))
def head(h, p): return b64e(json.dumps(h).encode()) + "." + b64e(json.dumps(p).encode())
def dg(si): return int.from_bytes(hashlib.sha256(si.encode()).digest(), "big")


def verify_naive(tok):
    """💀 反面教材：用哪个算法，由 token 自己的 header 说了算。"""
    h, p, s = tok.split(".")
    alg, si = json.loads(b64d(h))["alg"], h + "." + p   # ⚠️ 攻击者写什么这里就信什么
    if alg == "none":                                   # 💀 绕过一
        return json.loads(b64d(p))
    if alg == "RS256":
        ok = pow(int.from_bytes(b64d(s), "big"), E, N) == dg(si)
    else:                                               # 💀 绕过二：公钥被当成 HMAC 密钥
        ok = hmac.compare_digest(b64d(s), hmac.new(PUB, si.encode(), hashlib.sha256).digest())
    if not ok:
        raise ValueError("签名不对")
    return json.loads(b64d(p))


si = head({"alg": "RS256"}, {"sub": "u_42", "role": "user"})          # 服务端正常签发
print("① 正常签发   →", verify_naive(si + "." + b64e(pow(dg(si), D, N).to_bytes(35, "big"))))

print("② alg:none  →", verify_naive(head({"alg": "none"}, {"sub": "u_42", "role": "admin"}) + "."))

si = head({"alg": "HS256"}, {"sub": "u_42", "role": "admin"})         # 拿公钥当 HMAC 密钥重签
print("③ 算法混淆   →", verify_naive(si + "." + b64e(hmac.new(PUB, si.encode(), hashlib.sha256).digest())))

① 正常签发   → {'sub': 'u_42', 'role': 'user'}
② alg:none  → {'sub': 'u_42', 'role': 'admin'}
③ 算法混淆   → {'sub': 'u_42', 'role': 'admin'}

# ✅ 验签之外该查的三件事；顺带看看「不查 aud」会漏成什么样
import base64, hashlib, hmac, json, time

SECRET = b"idp-signing-key"          # 同一个签发方，一把密钥，签给好几个服务
MY_ISS, MY_AUD = "https://idp.example", "api://chat"


def b64e(raw): return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
def b64d(seg): return base64.urlsafe_b64decode(seg + "=" * (-len(seg) % 4))


def issue(payload):
    si = b64e(json.dumps({"alg": "HS256"}).encode()) + "." + b64e(json.dumps(payload).encode())
    return si + "." + b64e(hmac.new(SECRET, si.encode(), hashlib.sha256).digest())


def verify(tok, check_aud=True):
    h, p, s = tok.split(".")
    if json.loads(b64d(h)).get("alg") != "HS256":       # ⭐ 算法白名单：服务端定，不看 token
        raise ValueError("算法不在白名单")
    want = hmac.new(SECRET, (h + "." + p).encode(), hashlib.sha256).digest()
    if not hmac.compare_digest(b64d(s), want):          # ⭐ 等时比较，别用 ==
        raise ValueError("签名不对")
    body = json.loads(b64d(p))                          # ⭐ 验签通过之后才读载荷
    if body.get("exp", 0) <= int(time.time()):
        raise ValueError("已过期")
    if body.get("iss") != MY_ISS:
        raise ValueError("签发方不对")
    if check_aud and body.get("aud") != MY_AUD:         # ⭐ 这张票是不是发给【我】的
        raise ValueError("受众不对：%r" % body.get("aud"))
    return body


base = {"sub": "u_42", "iss": MY_ISS, "exp": int(time.time()) + 600}
other = issue(dict(base, aud="api://analytics", role="admin"))   # 同一家签给【别的服务】的票

print("① 发给我的票 →", verify(issue(dict(base, aud=MY_AUD)))["aud"])
print("② 不查 aud  →", verify(other, check_aud=False))            # 💀 签名合法，于是放行了
for n, (bad, why) in zip("③④", [(other, "查了 aud"), (issue(dict(base, aud=MY_AUD, exp=1)), "过期的票")]):
    try:
        verify(bad)
    except ValueError as e:
        print("%s %s  → 拒绝：%s" % (n, why, e))

① 发给我的票 → api://chat
② 不查 aud  → {'sub': 'u_42', 'iss': 'https://idp.example', 'exp': 1786360633, 'aud': 'api://analytics', 'role': 'admin'}
③ 查了 aud  → 拒绝：受众不对：'api://analytics'
④ 过期的票  → 拒绝：已过期