AI 에이전트 토큰 오남용 방어: OAuth 2.1 PKCE + RFC 8707 Resource Indicators 실전 가이드

GET /authorize?
  response_type=code
  &client_id=agent-orchestrator
  &resource=https://api.service-a.example.com
  &scope=read:data
  &code_challenge=BASE64URL(SHA256(verifier))
  &code_challenge_method=S256

{
  "iss": "https://auth.example.com",
  "sub": "user-123",
  "aud": "https://api.service-a.example.com",
  "scope": "read:data",
  "exp": 1713100000
}

┌──────────────────────────────────────────────────────────────────────┐
│  사용자 → 오케스트레이터 에이전트                                        │
│                                                                       │
│  Step 1: 이메일 서비스용 토큰 요청 (PKCE + resource=email-uri)           │
│  ┌──────────────────┐   인가 요청    ┌───────────────────┐            │
│  │  오케스트레이터   │ ────────────→ │  인가 서버 (AS)   │            │
│  │  (code_verifier) │ ←─────────── │  code_challenge   │            │
│  │                  │   인가 코드   │  검증 + aud 제한   │            │
│  │                  │ ────────────→ │  토큰 발급         │            │
│  │                  │ ←─────────── │                   │            │
│  └────────┬─────────┘ 이메일 전용 토큰 └───────────────────┘            │
│           │  {aud: "email-uri"}                                       │
│           ↓                                                           │
│  Step 2: 이메일 MCP 서버 호출                                           │
│  ┌──────────────────┐   Bearer 토큰  ┌───────────────────┐            │
│  │  오케스트레이터   │ ────────────→ │  이메일 MCP 서버  │            │
│  │                  │              │  aud 검증: ✓      │            │
│  └──────────────────┘              └───────────────────┘            │
│                                                                       │
│  Step 3: 캘린더 서비스용 별도 토큰 요청 (resource=calendar-uri)           │
│  ┌──────────────────┐               ┌───────────────────┐            │
│  │  오케스트레이터   │ ────────────→ │  인가 서버 (AS)   │            │
│  │ (새 code_verifier) │ ←─────────── │  캘린더 전용 토큰  │            │
│  └────────┬─────────┘               └───────────────────┘            │
│           │  {aud: "calendar-uri"}                                    │
│           ↓                                                           │
│  이메일 토큰으로 캘린더 API 호출 시도 → aud mismatch → 즉시 거부됨       │
└──────────────────────────────────────────────────────────────────────┘

import secrets
import hashlib
import base64
import httpx
 
def generate_pkce_pair() -> tuple[str, str]:
    """PKCE code_verifier와 code_challenge 쌍을 생성합니다."""
    verifier = secrets.token_urlsafe(64)  # 약 86자, S256 요건 충족
    digest = hashlib.sha256(verifier.encode()).digest()
    challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
    return verifier, challenge
 
async def get_resource_token(
    auth_server: str,
    client_id: str,
    resource_uri: str,  # RFC 8707 핵심: 각 서비스마다 별도 URI
    scope: str,
) -> str:
    verifier, challenge = generate_pkce_pair()
 
    # 1단계: resource 파라미터와 code_challenge를 포함한 인가 요청
    auth_params = {
        "response_type": "code",
        "client_id": client_id,
        "resource": resource_uri,
        "scope": scope,
        "code_challenge": challenge,
        "code_challenge_method": "S256",  # plain 메서드는 사용하지 않음
        "redirect_uri": "https://agent.example.com/callback",
    }
 
    # 에이전트 환경에서는 브라우저 리다이렉트 없이 콜백 처리가 필요합니다.
    # Device Authorization Flow(RFC 8628) 또는 로컬 콜백 서버(127.0.0.1:PORT)
    # 패턴을 사용하는 것을 권장합니다. 아래 함수는 해당 처리가 완료된 후
    # 인가 코드를 반환하는 플레이스홀더입니다.
    auth_code = await redirect_and_get_code(auth_server, auth_params)
 
    # 2단계: code_verifier 원문으로 토큰 교환
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{auth_server}/token",
            data={
                "grant_type": "authorization_code",
                "code": auth_code,
                "client_id": client_id,
                "resource": resource_uri,   # 토큰 요청에도 반드시 포함 (RFC 8707)
                "code_verifier": verifier,
                "redirect_uri": "https://agent.example.com/callback",
            },
        )
        response.raise_for_status()  # HTTP 오류 시 예외 발생 (4xx, 5xx)
        token_data = response.json()
 
    # verifier는 함수 스코프 종료 시 자동 소멸
    return token_data["access_token"]
 
# 오케스트레이터: 각 서비스마다 별도 토큰 요청
email_token = await get_resource_token(
    auth_server="https://auth.example.com",
    client_id="orchestrator-agent",
    resource_uri="https://email.mcp.example.com",
    scope="read:email",
)
 
calendar_token = await get_resource_token(
    auth_server="https://auth.example.com",
    client_id="orchestrator-agent",
    resource_uri="https://calendar.mcp.example.com",
    scope="write:events",
)
 
# email_token으로 calendar API 호출 시 → aud mismatch로 즉시 거부됩니다

from jose import jwt, JWTError
from fastapi import HTTPException, Request
 
EXPECTED_AUDIENCE = "https://email.mcp.example.com"
 
async def verify_token_audience(request: Request) -> dict:
    """aud 클레임 검증 — 리소스 서버 측 구현이 필수입니다."""
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing bearer token")
 
    token = auth_header[7:]
    try:
        payload = jwt.decode(
            token,
            key=get_public_key(),
            algorithms=["RS256"],
            audience=EXPECTED_AUDIENCE,  # aud 불일치 시 JWTError 발생
        )
    except JWTError as e:
        raise HTTPException(status_code=401, detail=f"Token validation failed: {e}")
 
    return payload

import { createHash, randomBytes } from "crypto";
 
interface AgentTokenRequest {
  tenantId: string;
  resourceType: "data" | "analytics" | "admin";
  scope: string;
}
 
function buildResourceUri(tenantId: string, resourceType: string): string {
  // RFC 8707 권장: 테넌트 식별자를 URI 경로에 포함
  return `https://api.example.com/tenant/${tenantId}/${resourceType}`;
}
 
function generatePkcePair(): { verifier: string; challenge: string } {
  const verifier = randomBytes(64).toString("base64url");
  const challenge = createHash("sha256")
    .update(verifier)
    .digest("base64url");
  return { verifier, challenge };
}
 
async function requestTenantScopedToken(
  req: AgentTokenRequest,
  authServerUrl: string,
  clientId: string
): Promise<string> {
  const resourceUri = buildResourceUri(req.tenantId, req.resourceType);
  const { verifier, challenge } = generatePkcePair();
 
  const authUrl = new URL(`${authServerUrl}/authorize`);
  authUrl.searchParams.set("response_type", "code");
  authUrl.searchParams.set("client_id", clientId);
  authUrl.searchParams.set("resource", resourceUri);
  authUrl.searchParams.set("scope", req.scope);
  authUrl.searchParams.set("code_challenge", challenge);
  authUrl.searchParams.set("code_challenge_method", "S256");
 
  // 에이전트 환경의 콜백 처리: Device Authorization Flow 또는
  // 로컬 콜백 서버(127.0.0.1:PORT) 패턴을 사용하는 것을 권장합니다.
  const code = await redirectAndGetCode(authUrl.toString());
 
  const tokenResponse = await fetch(`${authServerUrl}/token`, {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body: new URLSearchParams({
      grant_type: "authorization_code",
      code,
      client_id: clientId,
      resource: resourceUri,
      code_verifier: verifier,
    }),
  });
 
  if (!tokenResponse.ok) {
    const errorBody = await tokenResponse.text();
    throw new Error(`Token request failed (${tokenResponse.status}): ${errorBody}`);
  }
 
  const { access_token } = await tokenResponse.json();
  // verifier는 이 블록 종료 후 GC 대상
  return access_token;
}

import { expressjwt } from "express-jwt";
import jwksRsa from "jwks-rsa";
 
// JWKS 시크릿 헬퍼는 모듈 레벨에서 한 번만 생성해 공유합니다.
// cache: true 설정으로 매 요청마다 인가 서버에 JWKS 조회하는 것을 방지합니다.
const sharedJwksSecret = jwksRsa.expressJwtSecret({
  jwksUri: "https://auth.example.com/.well-known/jwks.json",
  cache: true,
  cacheMaxEntries: 5,
  cacheMaxAge: 600_000, // 10분
});
 
function createTenantAudienceValidator(tenantId: string, resourceType: string) {
  const expectedAudience = `https://api.example.com/tenant/${tenantId}/${resourceType}`;
 
  return expressjwt({
    secret: sharedJwksSecret,   // 공유 인스턴스 재사용 (성능)
    audience: expectedAudience, // tenant-scoped aud 검증
    algorithms: ["RS256"],
  });
}
 
// 각 테넌트 라우터에 독립적인 audience 검증기 적용
app.use(
  "/tenant/:tenantId/data",
  (req, res, next) =>
    createTenantAudienceValidator(req.params.tenantId, "data")(req, res, next)
);

{
  "name": "email-service-audience",
  "protocol": "openid-connect",
  "protocolMapper": "oidc-audience-mapper",
  "config": {
    "included.client.audience": "email-mcp-service",
    "access.token.claim": "true",
    "id.token.claim": "false"
  }
}