MCP 게이트웨이를 Zero Trust PEP로 설계하는 법 — OAuth 2.1·OPA·에피머럴 토큰으로 최소 권한(Least Privilege) 구현하기

[에이전트] → [MCP 게이트웨이 (PEP)]
                  ↓ JWT 검증 → OPA 정책 평가 → 에피머럴 토큰 발급
              → [MCP Server A: DB]
              → [MCP Server B: File]
              → [MCP Server C: API]

[요청 진입]
    ↓
[예시 1] JWT 검증 + OPA 정책 평가 + 에피머럴 토큰 발급   ← 예시 2의 Rego 파일이 평가됨
    ↓
[예시 3] RFC 8693 토큰 교환 → 서버별 좁은 스코프 토큰 획득
    ↓
[예시 4] 사용자 역할 기반 도구 목록 필터링 (마이크로 세그멘테이션)
    ↓
[MCP Server로 프록시]

// gateway/middleware/auth.ts
import jwt from 'jsonwebtoken';
 
interface AgentIdentity {
  agentId: string;
  role: string;
  userId: string;
}
 
async function verifyAgentJWT(
  authHeader: string | undefined,
): Promise<AgentIdentity | null> {
  if (!authHeader?.startsWith('Bearer ')) {
    return null;
  }
 
  const token = authHeader.slice(7);
  const publicKey = process.env.AGENT_JWT_PUBLIC_KEY;
  if (!publicKey) {
    // 환경 변수 누락 시에도 Fail-Closed
    console.error('AGENT_JWT_PUBLIC_KEY not configured');
    return null;
  }
 
  try {
    const decoded = jwt.verify(token, publicKey, {
      algorithms: ['RS256'],
      issuer: process.env.JWT_ISSUER,
    }) as AgentIdentity & jwt.JwtPayload;
 
    return {
      agentId: decoded.agentId,
      role: decoded.role,
      userId: decoded.userId,
    };
  } catch {
    // 토큰 위조·만료·서명 불일치 모두 null 반환 → 401
    return null;
  }
}

// gateway/middleware/policy-enforcement.ts
import { Request, Response, NextFunction } from 'express';
 
interface MCPToolCallRequest {
  agentId: string;
  userId: string;
  toolName: string;
  params: Record<string, unknown>;
}
 
interface PolicyDecision {
  allowed: boolean;
  reason?: string;
}
 
async function queryOPA(input: {
  agent: AgentIdentity;
  user: string;
  tool: string;
  params: Record<string, unknown>;
  timestamp: number;
}): Promise<PolicyDecision> {
  try {
    const response = await fetch(
      `${process.env.OPA_URL}/v1/data/mcp/tool_access`,
      {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ input }),
      },
    );
 
    if (!response.ok) {
      // OPA 서버 오류 시 Fail-Closed: 정책 엔진 불가 → 기본 거부
      console.error(`OPA returned ${response.status}`);
      return { allowed: false, reason: 'policy engine unavailable' };
    }
 
    const data = await response.json();
    return {
      allowed: data.result?.allow ?? false,
      reason: data.result?.deny_reason,
    };
  } catch (err) {
    // 네트워크 오류도 Fail-Closed
    console.error('OPA connection error:', err);
    return { allowed: false, reason: 'policy engine unreachable' };
  }
}
 
async function enforceMCPPolicy(
  req: Request,
  res: Response,
  next: NextFunction,
): Promise<void> {
  const toolCall = req.body as MCPToolCallRequest;
 
  // 1단계: 에이전트 신원 검증
  const agentIdentity = await verifyAgentJWT(req.headers.authorization);
  if (!agentIdentity) {
    res.status(401).json({ error: 'Agent identity verification failed' });
    return;
  }
 
  // 2단계: OPA 정책 질의 (Fail-Closed 포함)
  const decision = await queryOPA({
    agent: agentIdentity,
    user: toolCall.userId,
    tool: toolCall.toolName,
    params: toolCall.params,
    timestamp: Date.now(),
  });
 
  if (!decision.allowed) {
    await auditLog({
      event: 'tool_call_denied',
      agentId: toolCall.agentId,
      tool: toolCall.toolName,
      reason: decision.reason,
    });
    res.status(403).json({ error: `Access denied: ${decision.reason}` });
    return;
  }
 
  // 3단계: 에피머럴 토큰 발급 (단일 작업용, 60초 만료)
  req.ephemeralToken = await mintEphemeralToken({
    scope: `tool:${toolCall.toolName}`,
    subject: toolCall.agentId,
    expiresIn: '60s',
  });
 
  next();
}

# policies/mcp/tool_access.rego
package mcp.tool_access
 
import future.keywords.if
import future.keywords.in
 
default allow = false
 
# 읽기 전용 역할은 읽기 도구만 허용
allow if {
  input.agent.role == "readonly"
  input.tool in readonly_tools
}
 
# 쓰기 역할은 읽기 + 쓰기 도구 허용 (관리 도구는 제외)
allow if {
  input.agent.role == "readwrite"
  input.tool in allowed_write_tools
  not is_admin_tool(input.tool)
}
 
# 시간 기반 제한: 업무 시간(KST 09:00-18:00)에만 쓰기 허용
allow if {
  input.agent.role == "readwrite"
  business_hours_kst
}
 
readonly_tools := {
  "database_query",
  "file_read",
  "metrics_fetch",
}
 
allowed_write_tools := readonly_tools | {
  "file_write",
  "database_insert",
}
 
is_admin_tool(tool) if {
  tool in {"database_drop", "user_delete", "config_override"}
}
 
business_hours_kst if {
  # time.now_ns()는 UTC 기준 나노초를 반환합니다
  # KST(UTC+9) 변환: UTC 시(hour)에 9를 더해 24로 나눕니다
  utc_hour := time.clock(time.now_ns())[0]
  kst_hour := (utc_hour + 9) % 24
  kst_hour >= 9
  kst_hour < 18
}
 
# 거부 사유를 함께 반환
deny_reason := "readonly role cannot use write tools" if {
  input.agent.role == "readonly"
  input.tool in allowed_write_tools
}
 
deny_reason := "admin tools require explicit approval" if {
  is_admin_tool(input.tool)
}
 
deny_reason := "write tools only available during business hours (KST 09-18)" if {
  input.agent.role == "readwrite"
  not business_hours_kst
}

# gateway/auth/token_exchange.py
import os
import httpx
from dataclasses import dataclass
 
@dataclass
class NarrowedToken:
    access_token: str
    scope: str
    expires_in: int
    target_server: str
 
async def exchange_for_narrow_token(
    broad_token: str,
    target_mcp_server: str,
    required_tool: str,
) -> NarrowedToken:
    """
    RFC 8693 Token Exchange: 광범위한 토큰 → 특정 서버·도구 전용 토큰
    RFC 8707: resource 파라미터에 동적 서버 URL을 담아 전송하므로,
    이 토큰은 target_mcp_server 외 다른 서버에서 사용될 수 없습니다
    resource 값을 하드코딩하면 이 바인딩 보장이 깨지므로 주의하세요
    """
    auth_server_url = os.environ["AUTH_SERVER_URL"]
 
    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(
                url=f"{auth_server_url}/oauth/token",
                data={
                    "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
                    "subject_token": broad_token,
                    "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
                    # RFC 8707: 동적 서버 URL 바인딩 — 절대 하드코딩 금지
                    "resource": f"https://mcp.example.com/servers/{target_mcp_server}",
                    "scope": f"tool:{required_tool}",
                    "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
                },
                timeout=5.0,
            )
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            # 토큰 교환 실패도 Fail-Closed
            raise RuntimeError(
                f"Token exchange failed: {e.response.status_code}"
            ) from e
        except httpx.RequestError as e:
            raise RuntimeError(f"Auth server unreachable: {e}") from e
 
        data = response.json()
 
    return NarrowedToken(
        access_token=data["access_token"],
        scope=data["scope"],
        expires_in=data["expires_in"],  # 보통 60~300초
        target_server=target_mcp_server,
    )
 
 
async def get_secret_for_legacy_server(
    server_name: str,
    vault_token: str,  # 전역 변수 대신 함수 파라미터로 전달 — 보안 필수
) -> str:
    """
    OAuth2를 지원하지 않는 레거시 MCP 서버용:
    HashiCorp Vault에서 동적으로 단기 시크릿을 검색합니다
    """
    vault_url = os.environ["VAULT_URL"]
 
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                url=f"{vault_url}/v1/secret/mcp/{server_name}",
                headers={"X-Vault-Token": vault_token},
                timeout=3.0,
            )
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise RuntimeError(
                f"Vault lookup failed: {e.response.status_code}"
            ) from e
 
        return response.json()["data"]["api_key"]

// gateway/tools/dynamic-filter.ts
interface UserContext {
  userId: string;
  roles: string[];
  department: string;
}
 
interface MCPTool {
  name: string;
  description: string;
  requiredRoles: string[];
  sensitivityLevel: 'low' | 'medium' | 'high';
}
 
const ALL_MCP_TOOLS: MCPTool[] = [
  {
    name: 'database_query',
    description: 'Read-only SQL query',
    requiredRoles: ['analyst', 'engineer'],
    sensitivityLevel: 'low',
  },
  {
    name: 'database_insert',
    description: 'Insert rows into database',
    requiredRoles: ['engineer'],
    sensitivityLevel: 'medium',
  },
  {
    name: 'payment_process',
    description: 'Process payment transaction',
    requiredRoles: ['payment_admin'],
    sensitivityLevel: 'high',
  },
  {
    name: 'user_delete',
    description: 'Delete user account',
    requiredRoles: ['super_admin'],
    sensitivityLevel: 'high',
  },
];
 
async function checkHighSensitivityApproval(
  toolName: string,
  userContext: UserContext,
): Promise<boolean> {
  // 고위험 도구별 허용 부서 목록 (마이크로 세그멘테이션)
  const sensitiveToolDepartments: Record<string, string[]> = {
    payment_process: ['finance', 'treasury'],
    user_delete: ['platform-ops'],
  };
 
  const allowedDepts = sensitiveToolDepartments[toolName];
  if (allowedDepts && !allowedDepts.includes(userContext.department)) {
    return false;
  }
 
  // 고위험 도구는 KST 업무 시간(09:00-18:00) 외 차단
  const kstHour = (new Date().getUTCHours() + 9) % 24;
  if (kstHour < 9 || kstHour >= 18) {
    return false;
  }
 
  return true;
}
 
async function getPermittedTools(userContext: UserContext): Promise<MCPTool[]> {
  // 세션 시작 시 기본 활성 도구 = 없음 (Zero Trust 원칙)
  const permittedTools: MCPTool[] = [];
 
  for (const tool of ALL_MCP_TOOLS) {
    const hasRequiredRole = tool.requiredRoles.some((role) =>
      userContext.roles.includes(role),
    );
 
    if (!hasRequiredRole) continue;
 
    if (tool.sensitivityLevel === 'high') {
      const approved = await checkHighSensitivityApproval(tool.name, userContext);
      if (!approved) continue;
    }
 
    permittedTools.push(tool);
  }
 
  // 허가된 도구만 반환 — 존재하지 않는 도구는 에이전트에 보이지 않음
  return permittedTools;
}

// gateway/tools/metadata-validation.ts
function validateToolMetadata(tool: MCPTool): boolean {
  const suspiciousPatterns = [
    /ignore previous instructions/i,
    /system:/i,
    /<\|.*\|>/,      // 특수 토큰 패턴
    /\[INST\]/i,     // 인젝션 마커
  ];
 
  return !suspiciousPatterns.some((pattern) =>
    pattern.test(tool.description),
  );
}