7.1.2 身份认证 · OAuth 2.0 / OIDC / SAML 选型

# 范式 1:基于 Session(Cookie + 服务端存储)
# 浏览器自动带 cookie,服务端从 session_id 查到用户
# 优点:可立即吊销; 缺点:有状态、跨域难、移动端不友好
session_id = "sess_abc123"
user = redis.get(f"session:{session_id}")

# 范式 2:基于 Token(JWT 自包含)
# Token 本身带用户信息,服务端只需验签
# 优点:无状态、跨域、移动端友好; 缺点:吊销难
jwt_token = "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyIjoiYWxpY2UifQ.signature"
payload = jwt.decode(jwt_token, SECRET, algorithms=["HS256"])

# Python:PKCE 生成示例
import hashlib, base64, secrets

def generate_pkce_pair():
    """生成 code_verifier 和 code_challenge"""
    # 1. 生成 43~128 字符的随机 verifier
    verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b'=').decode()
    # 2. 算 SHA256 挑战值
    challenge = base64.urlsafe_b64encode(
        hashlib.sha256(verifier.encode()).digest()
    ).rstrip(b'=').decode()
    return verifier, challenge

verifier, challenge = generate_pkce_pair()
print(f"code_verifier: {verifier}")
print(f"code_challenge: {challenge}")
print(f"code_challenge_method: S256")

# 模式 1:授权码模式(Web 应用后端,推荐)
# 步骤 A:拼授权 URL
from urllib.parse import urlencode

auth_url = "https://github.com/login/oauth/authorize?" + urlencode({
    "client_id": "your_client_id",
    "redirect_uri": "https://yourapp.com/callback",
    "response_type": "code",
    "scope": "user:email",
    "state": secrets.token_urlsafe(16),  # 防 CSRF
    "code_challenge": challenge,         # PKCE
    "code_challenge_method": "S256",
})

# 步骤 B:用 code 换 token
import requests
token_resp = requests.post("https://github.com/login/oauth/access_token", data={
    "client_id": "your_client_id",
    "client_secret": "your_client_secret",
    "code": "received_auth_code",
    "redirect_uri": "https://yourapp.com/callback",
    "code_verifier": verifier,  # PKCE 关键
}, headers={"Accept": "application/json"})
access_token = token_resp.json()["access_token"]

# 模式 2:客户端凭证模式(服务间调用,无用户)
# 适合:Microservice 内部通信、定时任务
token_resp = requests.post("https://idp.example.com/oauth/token", data={
    "grant_type": "client_credentials",
    "client_id": "service_a",
    "client_secret": "service_a_secret",
    "scope": "internal.api",
})
machine_token = token_resp.json()["access_token"]
# 有效期通常 1 小时,过期再换

# 模式 3:密码模式(已不推荐,仅遗留系统)
# ⚠️ 警告:用户必须把密码交给客户端,违反 OAuth 初衷
# 历史上用于:第一代移动 App、企业内部可信环境
token_resp = requests.post("https://idp.example.com/oauth/token", data={
    "grant_type": "password",
    "username": "alice",
    "password": "alice_password",
    "client_id": "legacy_mobile_app",
    "client_secret": "...",
}, headers={"Accept": "application/json"})
# RFC 6749 明确建议:新项目不要用此模式

# Refresh Token 换新 access_token
def refresh_access_token(refresh_token: str) -> dict:
    resp = requests.post("https://idp.example.com/oauth/token", data={
        "grant_type": "refresh_token",
        "refresh_token": refresh_token,
        "client_id": "your_client_id",
        "client_secret": "your_client_secret",
    })
    resp.raise_for_status()
    new_tokens = resp.json()
    # 新 access_token;refresh_token 可能轮换
    return new_tokens

# 服务端:用 Authlib 处理 OIDC
from authlib.integrations.flask_client import OAuth
from flask import Flask, redirect, url_for, session

app = Flask(__name__)
app.secret_key = "your-flask-secret"
oauth = OAuth(app)

# 注册 OIDC Provider(以 Google 为例)
oauth.register(
    name="google",
    server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
    client_id="your-google-client-id",
    client_secret="your-google-client-secret",
    client_kwargs={"scope": "openid email profile"},
)

@app.route("/login")
def login():
    nonce = secrets.token_urlsafe(16)  # 防 ID Token 重放
    session["nonce"] = nonce
    redirect_uri = url_for("callback", _external=True)
    return oauth.google.authorize_redirect(redirect_uri, nonce=nonce)

@app.route("/callback")
def callback():
    token = oauth.google.authorize_access_token()
    # ↓↓↓ 关键:解析并验证 ID Token
    user_info = oauth.google.parse_id_token(token, nonce=session["nonce"])
    # user_info = {"sub": "123", "email": "alice@...", "name": "Alice"}
    session["user"] = user_info
    return "Logged in as " + user_info["email"]

# 纯 Python 验证 ID Token(不依赖框架,适合微服务)
import jwt
from jwt import PyJWKClient

def verify_id_token(id_token: str, client_id: str, issuer: str):
    # 1. 从 IDP 拿 JWKS 公钥
    jwks_url = f"{issuer}/.well-known/jwks.json"
    jwks_client = PyJWKClient(jwks_url)
    signing_key = jwks_client.get_signing_key_from_jwt(id_token)

    # 2. 验签 + 校验 claims
    payload = jwt.decode(
        id_token,
        signing_key.key,
        algorithms=["RS256"],   # 强制算法,防「算法替换攻击」
        audience=client_id,     # aud 必须等于 client_id
        issuer=issuer,          # iss 必须等于 IDP URL
        options={"require": ["exp", "iat", "iss", "aud", "sub"]},
    )
    return payload  # 拿到用户身份

import jwt
import time
import uuid

# ===== 1. 签发 JWT(对称 HS256) =====
def issue_jwt_hs256(user_id: str, secret: str) -> str:
    payload = {
        "sub": user_id,                    # 用户 ID
        "iat": int(time.time()),           # 签发时间
        "exp": int(time.time()) + 3600,    # 1 小时后过期
        "jti": str(uuid.uuid4()),          # 唯一 ID,防重放
        "scope": "read:profile write:post",
        "iss": "auth.example.com",
        "aud": "api.example.com",
    }
    token = jwt.encode(payload, secret, algorithm="HS256")
    return token

# ===== 2. 验签 JWT =====
def verify_jwt(token: str, secret: str) -> dict:
    try:
        payload = jwt.decode(
            token, secret,
            algorithms=["HS256"],          # ← 强制白名单,防算法替换
            audience="api.example.com",
            issuer="auth.example.com",
            options={"require": ["exp", "iat", "sub"]},
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise ValueError("token expired")
    except jwt.InvalidTokenError as e:
        raise ValueError(f"invalid token: {e}")

# RS256:私钥签名,公钥验签(适合多服务)
from cryptography.hazmat.primitives import serialization

# 生成密钥对(生产环境用 OpenSSL 生成,不要在代码里生成)
# openssl genrsa -out private.pem 2048
# openssl rsa -in private.pem -pubout -out public.pem

with open("private.pem", "rb") as f:
    private_key = serialization.load_pem_private_key(f.read(), password=None)

with open("public.pem", "rb") as f:
    public_key = serialization.load_pem_public_key(f.read())

def issue_jwt_rs256(user_id: str) -> str:
    payload = {
        "sub": user_id,
        "iat": int(time.time()),
        "exp": int(time.time()) + 3600,
        "jti": str(uuid.uuid4()),
    }
    return jwt.encode(payload, private_key, algorithm="RS256",
                      headers={"kid": "key-2026-01"})

def verify_jwt_rs256(token: str) -> dict:
    return jwt.decode(token, public_key, algorithms=["RS256"],
                      audience="api.example.com",
                      issuer="auth.example.com")

# 从 JWKS 自动取公钥(OIDC 场景)
import requests
from jwt import PyJWKClient

jwks_url = "https://idp.example.com/.well-known/jwks.json"
jwks_client = PyJWKClient(jwks_url)

def verify_oidc_id_token(id_token: str, client_id: str, issuer: str) -> dict:
    signing_key = jwks_client.get_signing_key_from_jwt(id_token).key
    return jwt.decode(
        id_token, signing_key,
        algorithms=["RS256"],    # 严格白名单
        audience=client_id,
        issuer=issuer,
        options={"require": ["exp", "iat", "iss", "aud", "sub"]},
    )

# 陷阱 1:算法替换攻击(Algorithm Confusion)
# 攻击者把 alg=RS256 改成 alg=HS256,把公钥当 HMAC 密钥,就能伪造 token
# 防御:永远显式指定 algorithms=["RS256"],禁止接受 "none"

# 陷阱 2:不校验 exp/aud/iss
# 防御:options={"require": ["exp", "iat", "aud", "iss"]}

# 陷阱 3:JWT 存入 LocalStorage,被 XSS 偷走
# 防御:用 HttpOnly + Secure + SameSite=Strict Cookie

# 陷阱 4:签名密钥硬编码到前端代码
# 防御:前端永远不要持有签名密钥,只存 access_token

# 陷阱 5:Token 永不过期
# 防御:access_token ≤ 2h,refresh_token ≤ 30 天,绝对不能 > 90 天

# 攻击代码:把 RS256 替换成 HS256,用公钥当 HMAC 密钥
import jwt, requests

# 1. 从 IDP 的 JWKS 拿公钥(PEM 格式)
jwks = requests.get("https://idp.example.com/.well-known/jwks.json").json()
public_key_pem = convert_jwk_to_pem(jwks["keys"][0])

# 2. 用公钥作 HMAC 密钥,签 HS256 token
fake_token = jwt.encode(
    {"sub": "admin", "exp": 9999999999},
    public_key_pem,        # 公钥当密钥,关键!
    algorithm="HS256"      # 替换算法
)

# 3. 如果服务端没限制 algorithms,验签会通过 → 接管成功
# 防御:jwt.decode(token, key, algorithms=["RS256"])  # 白名单