《Python编程实战》9.1 认证与会话：JWT / OAuth2

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9   ← header：{"alg":"HS256","typ":"JWT"}
.
eyJzdWIiOiI0MiIsInJvbGUiOiJhZG1pbiJ9   ← payload：claims（用户、角色、exp…）
.
yRhCk-4O6YC-UKFxJVyiIMYQa0Kx__hUDW6c3vIx5Y4   ← signature：对前两段签名

import base64, hashlib, hmac, json, time

def b64url_encode(raw: bytes) -> str:
    # JWT 用 base64url 且去掉 = 填充
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")

def b64url_decode(segment: str) -> bytes:
    pad = "=" * (-len(segment) % 4)          # 还原被去掉的填充
    return base64.urlsafe_b64decode(segment + pad)

def sign(payload: dict, secret: str, *, expires_in: int = 900) -> str:
    header = {"alg": "HS256", "typ": "JWT"}
    now = int(time.time())
    body = {**payload, "iat": now, "exp": now + expires_in}
    seg = ".".join(
        b64url_encode(json.dumps(part, separators=(",", ":"), ensure_ascii=False).encode())
        for part in (header, body)
    )
    sig = hmac.new(secret.encode(), seg.encode("ascii"), hashlib.sha256).digest()
    return f"{seg}.{b64url_encode(sig)}"

token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiI0MiIsInJvbGUiOiJhZG1pbiIsImlhdCI6MTc5MTUxNjg5NywiZXhwIjoxNzkxNTE3Nzk3fQ.pG4G4OP0KMs4_cbzeJQLYoOCvcsNROjJmPiYKK0BVxA
段数: 3

class InvalidToken(Exception):
    pass

def verify(token: str, secret: str, *, now: int | None = None) -> dict:
    now = int(time.time()) if now is None else now
    try:
        head_b64, body_b64, sig_b64 = token.split(".")
    except ValueError:
        raise InvalidToken("token 结构不是三段") from None

    try:
        header = json.loads(b64url_decode(head_b64))
    except (ValueError, UnicodeDecodeError):
        raise InvalidToken("header 不是合法 base64url/JSON") from None
    if header.get("alg") != "HS256":                 # 白名单，挡住 alg=none
        raise InvalidToken(f"不接受的算法: {header.get('alg')!r}")

    signing_input = f"{head_b64}.{body_b64}".encode("ascii")
    expected = hmac.new(secret.encode(), signing_input, hashlib.sha256).digest()
    try:
        got = b64url_decode(sig_b64)
    except ValueError:
        raise InvalidToken("签名段不是合法 base64url") from None
    if not hmac.compare_digest(expected, got):        # 恒定时间比较，防时序攻击
        raise InvalidToken("签名不匹配")

    try:
        body = json.loads(b64url_decode(body_b64))
    except (ValueError, UnicodeDecodeError):
        raise InvalidToken("payload 不是合法 base64url/JSON") from None
    if "exp" in body and now >= int(body["exp"]):
        raise InvalidToken("已过期")
    if "nbf" in body and now < int(body["nbf"]):
        raise InvalidToken("尚未生效")
    return body

校验: {'sub': '42', 'role': 'admin', 'iat': 1791516897, 'exp': 1791517797}
篡改 payload 被拒: 签名不匹配
篡改签名被拒: 签名不匹配
alg=none 被拒: 不接受的算法: 'none'
错误密钥被拒: 签名不匹配

# 同一 token，用不同的 now 校验
verify(make(exp_offset=900), SECRET, now=base)   # exp 在 900 秒后

未过期: 42
exp=    0: 拒绝 -> 已过期
exp=   -1: 拒绝 -> 已过期
exp= -300: 拒绝 -> 已过期
nbf 未到: 拒绝 -> 尚未生效

# verify() 内两处时间判断改为带 leeway 的版本（leeway 默认 60 秒）
if "exp" in body and now >= int(body["exp"]) + leeway:   # 过期后 leeway 秒内仍接受
    raise InvalidToken("已过期")
if "nbf" in body and now < int(body["nbf"]) - leeway:     # 提前 leeway 秒内先接受
    raise InvalidToken("尚未生效")

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from typing import Annotated

app = FastAPI()
bearer = HTTPBearer(auto_error=False)   # 自己控错误信息，别让它默认抛

def current_user(creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)]) -> dict:
    if creds is None or creds.scheme.lower() != "bearer":
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "缺少 Bearer 令牌")
    try:
        return verify(creds.credentials, SECRET)
    except InvalidToken as e:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, f"令牌无效: {e}") from e

@app.get("/me")
def me(user: Annotated[dict, Depends(current_user)]) -> dict:
    return {"sub": user["sub"], "role": user["role"]}

登录: 200 bearer
带令牌: 200 {'sub': 'ada', 'role': 'user'}
无令牌: 401 缺少 Bearer 令牌
坏令牌: 401 令牌无效: header 不是合法 base64url/JSON
错密码: 401 用户名或密码错误

import base64, hashlib, secrets

def b64url(b: bytes) -> str:
    return base64.urlsafe_b64encode(b).rstrip(b"=").decode()

verifier = b64url(secrets.token_bytes(32))               # 43 字符，高熵随机
challenge = b64url(hashlib.sha256(verifier.encode()).digest())

code_verifier : WSF7vycD_gTbbQbtDTaUsJAhvjJDSjRkUbOLuXJdBus 43
code_challenge: DC7eY1tNmU_4NyDv1E3vMJQHIXS6UCwiMjDa_zAaToE 43
method: S256
校验通过: True