📚 课程目录第4章 OpenID Connect身份认证集成

4. OpenID Connect身份认证集成

任务说明

本案例为数字银行开放银行API网关集成OpenID Connect（OIDC）身份认证。需要完成：银行开发者门户作为OIDC Relying Party（RP），对接银行统一身份提供方（IdP），实现第三方应用用户的身份认证与ID 签发。核心交付物为一个可运行的OIDC认证流程模拟器，覆盖Authorization Code Flow的核心交互，并输出可供API网关校验的ID 。

OIDC认证流程设计

本案例聚焦银行开放平台最常用的Authorization Code Flow，参与方包括：第三方应用（客户端）、银行API网关（RP）、银行IdP。流程如下：

第三方应用引导用户访问银行IdP的授权端点，携带client_id、redirect_uri、response_type=code、scope=openid；
用户完成身份认证后，IdP返回授权码（code）至回调地址；
第三方应用使用授权码向IdP的令牌端点换取ID 和Access ；
API网关校验ID 的签名、issuer、audience和过期时间，校验通过后允许访问受保护的银行API。

核心代码实现

以下Python实现模拟银行IdP的授权端点与令牌端点，并包含API网关的ID 校验逻辑。使用cryptography库生成RSA密钥对并签名JWT。


# oidc_provider.py
import json
import time
import uuid
from urllib.parse import urlencode
from http.server import HTTPServer, BaseHTTPRequestHandler
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature

# ---------- 密钥生成（实际部署中由IdP保管私钥） ----------
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
public_key = private_key.public_key()

# 模拟用户数据库
USERS = {"alice": {"password": "pass123", "sub": "user-001", "name": "Alice"}}
# 模拟授权码存储
AUTH_CODES = {}

def base64url_encode(data: bytes) -> str:
    import base64
    return base64.urlsafe_b64encode(data).rstrip(b'=').decode()

def base64url_decode(data: str) -> bytes:
    import base64
    padding = '=' * (4 - len(data) % 4)
    return base64.urlsafe_b64decode(data + padding)

def create_jwt_header() -> str:
    header = {"alg": "RS256", "typ": "JWT", "kid": "bank-idp-key-1"}
    return base64url_encode(json.dumps(header).encode())

def create_id_(sub: str, aud: str, nonce: str = "") -> str:
    """生成符合OIDC规范的ID """
    now = int(time.time())
    payload = {
        "iss": "https://idp.bank.example.com",
        "sub": sub,
        "aud": aud,
        "exp": now + 3600,
        "iat": now,
        "nonce": nonce,
        "auth_time": now,
        "amr": ["pwd"]
    }
    header = create_jwt_header()
    payload_b64 = base64url_encode(json.dumps(payload).encode())
    signing_input = f"{header}.{payload_b64}".encode()
    signature = private_key.sign(signing_input, padding.PKCS1v15(), hashes.SHA256())
    # 将DER签名转为JWT所需的r||s格式
    r, s = private_key.sign(signing_input, padding.PKCS1v15(), hashes.SHA256()).hex()[:64], \
           private_key.sign(signing_input, padding.PKCS1v15(), hashes.SHA256()).hex()[64:]
    # 简化处理：直接使用原始签名（实际应使用encode_dss_signature）
    sig_b64 = base64url_encode(signature)
    return f"{header}.{payload_b64}.{sig_b64}"

class OIDCHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # 授权端点 /authorize
        if self.path.startswith("/authorize"):
            query = dict(urllib.parse.parse_qsl(urllib.parse.urlparse(self.path).query))
            client_id = query.get("client_id")
            redirect_uri = query.get("redirect_uri")
            scope = query.get("scope")
            state = query.get("state")
            nonce = query.get("nonce")
            
            if client_id != "bank-api-client" or "openid" not in scope:
                self.send_response(400)
                self.end_headers()
                self.wfile.write(b"Invalid client or scope")
                return
            
            # 模拟用户登录（实际应跳转登录页）
            # 此处直接生成授权码
            code = str(uuid.uuid4())
            AUTH_CODES[code] = {"client_id": client_id, "redirect_uri": redirect_uri, "sub": "user-001"}
            
            # 重定向回第三方应用
            redirect_url = f"{redirect_uri}?code={code}&state={state}"
            self.send_response(302)
            self.send_header("Location", redirect_url)
            self.end_headers()
    
    def do_POST(self):
        # 令牌端点 /
        if self.path.startswith("/"):
            content_length = int(self.headers["Content-Length"])
            body = self.rfile.read(content_length).decode()
            params = dict(urllib.parse.parse_qsl(body))
            
            code = params.get("code")
            client_id = params.get("client_id")
            client_secret = params.get("client_secret")
            redirect_uri = params.get("redirect_uri")
            
            if client_secret != "bank-secret-2024":
                self.send_response(401)
                self.end_headers()
                self.wfile.write(b"Invalid client secret")
                return
            
            stored = AUTH_CODES.get(code)
            if not stored or stored["client_id"] != client_id or stored["redirect_uri"] != redirect_uri:
                self.send_response(400)
                self.end_headers()
                self.wfile.write(b"Invalid authorization code")
                return
            
            # 生成ID 
            id_ = create_id_(sub=stored["sub"], aud=client_id)
            del AUTH_CODES[code]  # 一次性使用
            
            response = {
                "access_": "access--" + uuid.uuid4().hex,
                "_type": "Bearer",
                "expires_in": 3600,
                "id_": id_
            }
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps(response).encode())
    
    def log_message(self, format, *args):
        pass  # 静默日志

def run_server():
    server = HTTPServer(("localhost", 8080), OIDCHandler)
    print("OIDC Provider running on http://localhost:8080")
    server.serve_forever()

if __name__ == "__main__":
    run_server()


# api_gateway_validator.py
import json
import time
import base64
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.serialization import load_pem_public_key

# 从IdP获取的公钥（实际通过JWKS端点动态获取）
PUBLIC_KEY_PEM = b"""-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
-----END PUBLIC KEY-----"""

def base64url_decode(data: str) -> bytes:
    padding = '=' * (4 - len(data) % 4)
    return base64.urlsafe_b64decode(data + padding)

def validate_id_(id_: str, expected_aud: str) -> dict:
    """验证ID 签名、iss、aud、exp"""
    parts = id_.split(".")
    if len(parts) != 3:
        raise ValueError("Invalid ID  format")
    
    header_b64, payload_b64, signature_b64 = parts
    signing_input = f"{header_b64}.{payload_b64}".encode()
    signature = base64url_decode(signature_b64)
    
    # 加载公钥并验证签名
    public_key = load_pem_public_key(PUBLIC_KEY_PEM)
    try:
        public_key.verify(signature, signing_input, padding.PKCS1v15(), hashes.SHA256())
    except Exception as e:
        raise ValueError(f"Signature verification failed: {e}")
    
    # 解析payload
    payload = json.loads(base64url_decode(payload_b64))
    
    # 校验必要声明
    if payload.get("iss") != "https://idp.bank.example.com":
        raise ValueError("Invalid issuer")
    if payload.get("aud") != expected_aud:
        raise ValueError("Invalid audience")
    if payload.get("exp", 0) < int(time.time()):
        raise ValueError(" expired")
    
    return payload

# 使用示例（在API网关中调用）
if __name__ == "__main__":
    # 此处的id_由OIDC Provider实际签发
    # 实际运行时从Authorization头获取
    sample_ = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJodHRwczovL2lkcC5iYW5rLmV4YW1wbGUuY29tIiwic3ViIjoidXNlci0wMDEiLCJhdWQiOiJiYW5rLWFwaS1jbGllbnQiLCJleHAiOjE3MTAwMDAwMDAsImlhdCI6MTcwOTk5NjQwMH0.signature"
    # 实际验证需使用真实签发的，此处仅演示调用方式
    # payload = validate_id_(sample_, "bank-api-client")
    # print("验证通过:", payload["sub"])
    print("API网关校验模块就绪。实际运行时请传入IdP签发的ID 。")

运行与验证

依赖安装：

pip install cryptography

启动OIDC Provider：

python oidc_provider.py

验证步骤：

浏览器访问 http://localhost:8080/authorize?client_id=bank-api-client&redirect_uri=http://localhost:3000/callback&scope=openid&state=xyz&nonce=abc，观察是否302重定向至回调地址并携带code参数；
使用curl模拟令牌请求：

curl -X POST http://localhost:8080/ \
  -d "code=上一步获取的code" \
  -d "client_id=bank-api-client" \
  -d "client_secret=bank-secret-2024" \
  -d "redirect_uri=http://localhost:3000/callback"

预期返回JSON包含id_字段；

将返回的id_粘贴至api_gateway_validator.py的sample_变量，运行验证脚本，观察是否输出"验证通过"及sub声明。

判定标准：授权码流程完整走通，ID 包含正确的iss、aud、exp声明，签名验证通过且无异常抛出，即视为本案例实现成功。