Pomerium Core 代理认证实战：Keycloak + JWT 验签保护内部应用 | IDaaS Book

# config.yaml
address: :443
http_redirect_addr: :80
authenticate_service_url: https://authenticate.example.com

# --- IdP ---
idp_provider: oidc
idp_client_id: pomerium-core
idp_client_secret: "${KEYCLOAK_CLIENT_SECRET}"
idp_provider_url: https://authenticate.example.com/realms/internal
idp_scopes: [openid, email, profile, groups]

# --- 密钥（生产用 *_file，把值放进 Secret）---
signing_key_file: /pomerium/ec_private.pem      # 给上游验签用的签名私钥
cookie_secret_file: /pomerium/cookie_secret     # 会话 Cookie 加解密
shared_secret_file: /pomerium/shared_secret     # 服务间互认证

# --- 路由 ---
routes:
  - from: https://grafana.internal.example.com
    to: http://grafana.monitoring.svc.cluster.local:3000
    pass_identity_headers: true                 # 注入签名 JWT
    policy:
      - allow:
          and:
            - domain:
                is: example.com
            - claim/department:
                is: platform

  - from: https://console.internal.example.com
    to: http://console.internal.svc.cluster.local:8080
    pass_identity_headers: true
    idle_timeout: 10m
    policy:
      - allow:
          and:
            - claim/groups:
                contains: iam-admins
      - deny:
          or:
            - email:
                is: [email protected]

# app.py —— Flask + PyJWT，进程内缓存 JWKS
from flask import Flask, request, jsonify
import jwt
from jwt import PyJWKClient

APP_HOST = "console.internal.example.com"
JWKS_URL = f"https://{APP_HOST}/.well-known/pomerium/jwks.json"
jwks = PyJWKClient(JWKS_URL, cache_keys=True)

app = Flask(__name__)

@app.get("/api/whoami")
def whoami():
    raw = request.headers.get("X-Pomerium-Jwt-Assertion")
    if not raw:
        return jsonify(error="missing identity header"), 401
    try:
        key = jwks.get_signing_key_from_jwt(raw)
        claims = jwt.decode(
            raw,
            key.key,
            algorithms=["ES256", "RS256"],
            audience=APP_HOST,     # 必须校验：多路由共用一套部署时的安全边界
            issuer=APP_HOST,
            leeway=10,             # exp 只有 5 分钟，时钟容差别开太大
        )
    except jwt.PyJWTError as exc:
        return jsonify(error=f"invalid identity: {exc}"), 401
    return jsonify(email=claims["email"], sub=claims["sub"], groups=claims.get("groups"))

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)