MCP 멀티에이전트 위임 패턴: RFC 8693 Token Exchange와 감사 로그로 에이전트 체인 보안 설계하기

{
  "sub": "user-alice",
  "act": {
    "sub": "sub-agent",
    "act": {
      "sub": "orchestrator-agent"
    }
  }
}

POST /token HTTP/1.1
Content-Type: application/x-www-form-urlencoded
 
grant_type=urn:ietf:params:oauth:grant-type:token-exchange
&subject_token=<user_access_token>
&subject_token_type=urn:ietf:params:oauth:token-type:access_token
&actor_token=<orchestrator_agent_token>
&actor_token_type=urn:ietf:params:oauth:token-type:access_token
&scope=read:orders
&resource=https://api.example.com/orders

{
  "sub": "agent-instance-xyz",
  "agent_type": "orchestrator",
  "agent_model": "claude-sonnet-4-6",
  "agent_instance_id": "sess-abc123",
  "delegator_sub": "user-alice",
  "delegation_chain": [
    "user-alice",
    "orchestrator-agent",
    "sub-agent-b"
  ]
}

interface AuditLogEntry {
  timestamp: string;            // ISO 8601
  trace_id: string;             // 전체 요청 체인을 관통하는 ID
  action_type:
    | "tool_call"
    | "tool_response"
    | "decision"
    | "delegation"     // 에이전트 → 에이전트
    | "escalation"     // 에이전트 → 인간
    | "error";
  actor: {
    sub: string;                // 현재 행위 에이전트
    delegation_chain: string[]; // 위임 계보 (시간순)
  };
  resource: string;             // 접근 대상 리소스
  policy_evaluation: {
    decision: "allow" | "deny";
    matched_policy: string;
  };
  payload_hash: string;         // 불변성 검증용 해시
}

# mcp_gateway_middleware.py
import jwt
import os
from typing import TypedDict
from dataclasses import dataclass
from cryptography.hazmat.primitives.serialization import load_pem_public_key
 
# 공개키는 환경 변수에서 로드합니다. 하드코딩은 지양해야 합니다.
PUBLIC_KEY = load_pem_public_key(os.environ["JWT_PUBLIC_KEY"].encode())
 
# 위임 깊이 제한 (일반적으로 3~5 수준을 권장합니다)
# 너무 낮으면 복잡한 위임 시나리오를 처리할 수 없고,
# 너무 높으면 체인 전체의 보안 검토가 어려워집니다.
MAX_DELEGATION_DEPTH = 5
 
 
@dataclass
class PolicyResult:
    allowed: bool
    matched_policy: str
    reason: str | None = None
 
 
async def evaluate_policy(
    principal: str,
    resource: str,
    action: str,
    context: dict,
) -> PolicyResult:
    """
    외부 정책 엔진(Cerbos, OPA 등)과 연동해 RBAC 결정을 수행합니다.
 
    Cerbos 연동 예시:
        from cerbos.sdk.client import CerbosClient
        client = CerbosClient(host=os.environ["CERBOS_HOST"])
        resp = client.check_resource(
            principal=Principal(id=principal),
            resource=Resource(id=resource, kind="mcp_tool"),
            actions=[action],
            context=context,
        )
        return PolicyResult(
            allowed=resp.is_allowed(action),
            matched_policy=resp.outputs[0].policy_version if resp.outputs else "",
        )
    """
    raise NotImplementedError("정책 엔진 연동 구현이 필요합니다")
 
 
class TripleIdentityContext(TypedDict):
    user_sub: str
    agent_sub: str
    tool_id: str
    delegation_chain: list[str]
    scope: list[str]
 
 
async def verify_triple_identity(
    user_token: str,
    agent_token: str,
    tool_id: str,
    resource: str,
) -> TripleIdentityContext:
    """
    사용자 → 에이전트 → 도구 세 신원을 동시에 검증합니다.
    RFC 8693 act 클레임 체인을 통해 위임 맥락을 보존합니다.
    """
    # 1. 사용자 토큰 검증 (공개키 필수 — key 없이는 DecodeError 발생)
    user_claims = jwt.decode(
        user_token,
        key=PUBLIC_KEY,
        algorithms=["RS256"],
    )
 
    # 2. 에이전트 토큰 검증 및 act 클레임에서 위임 체인 추출
    agent_claims = jwt.decode(
        agent_token,
        key=PUBLIC_KEY,
        algorithms=["RS256"],
    )
    delegation_chain = extract_delegation_chain(agent_claims)
 
    # 3. 위임 깊이 제한 검증
    if len(delegation_chain) > MAX_DELEGATION_DEPTH:
        raise ValueError(
            f"Delegation depth {len(delegation_chain)} exceeds limit {MAX_DELEGATION_DEPTH}"
        )
 
    # 4. RFC 8707 리소스 인디케이터로 토큰 오용 방지
    if agent_claims.get("aud") != resource:
        raise ValueError(
            f"Token audience mismatch: expected {resource}, "
            f"got {agent_claims.get('aud')}"
        )
 
    # 5. 정책 엔진(Cerbos 등)으로 RBAC 평가
    policy_result = await evaluate_policy(
        principal=user_claims["sub"],
        resource=resource,
        action="mcp_tool_call",
        context={"tool_id": tool_id, "agent": agent_claims["sub"]},
    )
 
    if not policy_result.allowed:
        raise PermissionError(
            f"Policy denied: {policy_result.matched_policy}"
        )
 
    return TripleIdentityContext(
        user_sub=user_claims["sub"],
        agent_sub=agent_claims["sub"],
        tool_id=tool_id,
        delegation_chain=delegation_chain,
        scope=agent_claims.get("scope", "").split(),
    )
 
 
def extract_delegation_chain(claims: dict) -> list[str]:
    """
    act 클레임을 순회해 위임 계보를 시간순 리스트로 변환합니다.
 
    RFC 8693에서 바깥쪽 act.sub이 가장 직접적인(최종) 행위자이므로,
    순회 후 역순 정렬해 [원래 주체, 첫 위임자, ..., 최종 행위자] 형태로 반환합니다.
    """
    act_chain = []
    act = claims.get("act")
    while act:
        act_chain.append(act["sub"])  # 바깥쪽(최근) → 안쪽(이전) 순으로 추가
        act = act.get("act")
    act_chain.reverse()  # 이전 → 최근 순으로 정렬
    return [claims["sub"]] + act_chain  # [원래 주체, ..., 최종 행위자] 순

// delegation-negotiation.ts
 
// 위임 깊이 제한 (일반적으로 3~5를 권장합니다)
const MAX_DELEGATION_DEPTH = 5;
 
interface DelegationRequest {
  task_id: string;
  scope: string[];
  deadline: string;            // ISO 8601
  expected_artifact: string;
  sensitivity_level: "public" | "internal" | "confidential" | "restricted";
  delegation_chain: string[];  // 현재까지의 위임 경로
}
 
interface DelegationResponse {
  status: "accepted" | "rejected" | "counter_proposed";
  accepted_scope?: string[];   // counter_proposed 시 수락 가능한 축소 범위
  reason?: string;
}
 
async function requestDelegation(
  fromAgent: string,
  toAgent: string,
  request: DelegationRequest,
): Promise<DelegationResponse> {
  // 무한 위임 방지: 체인 깊이가 한계에 도달하면 거절
  if (request.delegation_chain.length >= MAX_DELEGATION_DEPTH) {
    return { status: "rejected", reason: "delegation_depth_exceeded" };
  }
 
  // 수신 에이전트의 권한 범위 내에서 처리 가능한지 확인
  const agentCapabilities = await getAgentCapabilities(toAgent);
  const feasibleScope = request.scope.filter(
    (s) => agentCapabilities.includes(s),
  );
 
  if (feasibleScope.length === 0) {
    return { status: "rejected", reason: "no_matching_capabilities" };
  }
 
  if (feasibleScope.length < request.scope.length) {
    // 일부 범위만 처리 가능한 경우 역제안
    return {
      status: "counter_proposed",
      accepted_scope: feasibleScope,
      reason: "partial_capability_match",
    };
  }
 
  // 위임 이벤트를 감사 로그에 기록
  await auditLog({
    action_type: "delegation",
    actor: { sub: fromAgent, delegation_chain: request.delegation_chain },
    resource: request.task_id,
    timestamp: new Date().toISOString(),
  });
 
  return { status: "accepted", accepted_scope: request.scope };
}

# revocation_propagation.py
import asyncio
from dataclasses import dataclass
 
# find_delegation_descendants 구현을 위해 필요한 데이터 구조:
# 각 Token Exchange 시점마다 위임 관계를 별도 저장소에 기록해야 합니다.
#
# Redis 예시:
#   KEY: delegation_tree:{parent_agent_id}
#   VALUE: SET of "{child_agent_id}:{issued_at}:{token_id}"
#
# PostgreSQL 예시:
#   CREATE TABLE delegation_tree (
#     parent_agent_id  TEXT,
#     child_agent_id   TEXT,
#     issued_at        TIMESTAMPTZ,
#     token_id         TEXT PRIMARY KEY
#   );
#
# 토큰 발급 시마다 이 테이블에 레코드를 삽입해야 취소 전파가 가능합니다.
 
 
@dataclass
class RevocationEvent:
    subject: str          # 취소된 주체 (사용자 또는 에이전트)
    reason: str
    trace_id: str
    issued_before: str    # 이 시각 이전 발급 토큰 모두 무효화 (ISO 8601)
 
 
async def propagate_revocation(event: RevocationEvent) -> None:
    """
    OpenID Shared Signals Framework를 활용해
    취소 이벤트를 체인 하위로 전파합니다.
    """
    # 취소된 주체의 위임 체인에 포함된 모든 에이전트 조회
    # (delegation_tree 저장소에서 BFS/DFS로 하위 노드를 순회합니다)
    affected_agents = await find_delegation_descendants(
        event.subject, event.issued_before
    )
 
    # 병렬로 모든 하위 에이전트 토큰 무효화
    revocation_tasks = [
        invalidate_agent_token(agent_id, event.trace_id)
        for agent_id in affected_agents
    ]
    results = await asyncio.gather(*revocation_tasks, return_exceptions=True)
 
    # 실패한 취소 작업은 별도 큐에 저장해 재시도
    failed = [
        affected_agents[i]
        for i, r in enumerate(results)
        if isinstance(r, Exception)
    ]
    if failed:
        await queue_revocation_retry(failed, event)