MCP Tool Squatting · Rug Pull 차단: ETDI와 OAuth 서명 기반 무결성 검증 직접 구현하기

{
  "name": "send_email",
  "version": "1.2.0",
  "description": "지정된 수신자에게 이메일을 발송합니다.",
  "schema": {
    "input": {
      "type": "object",
      "properties": {
        "to": { "type": "string" },
        "subject": { "type": "string" },
        "body": { "type": "string" }
      },
      "required": ["to", "subject", "body"]
    },
    "output": { "type": "object" }
  },
  "permissions": ["email:send"],
  "provider": "trusted-email-service",
  "signature": "eyJhbGciOiJFUzI1NiJ9.eyJuYW1lIjoic2VuZF9lbWFpbCIsInZlcnNpb24iOiIxLjIuMCJ9.MEUCIQD...",
  "public_key_ref": "https://keys.trusted-email-service.com/.well-known/jwks.json"
}

sequenceDiagram
    participant S as MCP 서버
    participant JW as JWKS 엔드포인트
    participant C as MCP 클라이언트
 
    S->>S: 도구 정의 생성 + ES256 JWT 서명
    S->>JW: 공개 키 게시(/.well-known/jwks.json)
    C->>S: 도구 목록 조회
    S->>C: 서명된 도구 정의 반환
    C->>JW: 공개 키 조회 (TTL 캐싱)
    JW->>C: JWKS 응답
    C->>C: JWT 서명 검증
    alt 검증 성공
        C->>C: 버전·스코프 변경 확인
        C->>S: 도구 호출
    else 검증 실패
        C->>C: 도구 호출 거부
    end

# 의존성 설치
uv add cryptography PyJWT

# etdi_server.py
import json
import base64
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.backends import default_backend
import jwt  # PyJWT
 
class ETDIToolSigner:
    """도구 정의에 ECDSA P-256 JWT 서명을 생성하는 클래스"""
 
    def __init__(self):
        # 프로덕션에서는 HSM 또는 Secret Manager에서 키를 로드합니다
        self._private_key = ec.generate_private_key(
            ec.SECP256R1(), default_backend()
            # SECP256R1 = P-256: NIST 표준 곡선으로 성능·보안·호환성의
            # 균형이 잡혀 있으며 JOSE 생태계 전반에서 지원됩니다
        )
        self._public_key = self._private_key.public_key()
 
    def sign_tool_definition(self, tool_def: dict) -> dict:
        """
        도구 정의의 핵심 필드를 JWT 페이로드로 서명합니다.
        서명 대상: name + version + schema + permissions + provider
        """
        payload = {
            "name": tool_def["name"],
            "version": tool_def["version"],
            "schema": tool_def["schema"],
            # sorted()로 권한 목록 정렬 — 결정론적 직렬화가 핵심입니다
            "permissions": sorted(tool_def.get("permissions", [])),
            "provider": tool_def["provider"],
        }
        # PyJWT가 ES256(ECDSA + SHA-256) JWT를 생성합니다
        # jose(TypeScript) 등 표준 JOSE 라이브러리로 교차 검증이 가능합니다
        tool_def["signature"] = jwt.encode(
            payload, self._private_key, algorithm="ES256"
        )
        return tool_def
 
    def export_jwks(self) -> dict:
        """클라이언트가 서명 검증에 사용할 JWKS를 반환합니다."""
        # _private_key.public_key()로 분리한 공개 키 객체에서 좌표를 추출합니다
        pub_numbers = self._public_key.public_numbers()
 
        x = base64.urlsafe_b64encode(
            pub_numbers.x.to_bytes(32, 'big')
        ).rstrip(b'=').decode()
        y = base64.urlsafe_b64encode(
            pub_numbers.y.to_bytes(32, 'big')
        ).rstrip(b'=').decode()
 
        return {
            "keys": [{
                "kty": "EC",
                "crv": "P-256",
                "use": "sig",
                "alg": "ES256",
                "kid": "etdi-key-v1",
                "x": x,
                "y": y,
            }]
        }
 
 
# 도구 정의 예시
signer = ETDIToolSigner()
 
send_email_tool = {
    "name": "send_email",
    "version": "1.2.0",
    "description": "지정된 수신자에게 이메일을 발송합니다.",
    "schema": {
        "input": {
            "type": "object",
            "properties": {
                "to": {"type": "string", "format": "email"},
                "subject": {"type": "string"},
                "body": {"type": "string"}
            },
            "required": ["to", "subject", "body"]
        }
    },
    "permissions": ["email:send"],
    "provider": "trusted-email-service",
    "public_key_ref": "https://keys.trusted-email-service.com/.well-known/jwks.json"
}
 
signed_tool = signer.sign_tool_definition(send_email_tool)
print(json.dumps(signed_tool, indent=2, ensure_ascii=False))
 
jwks = signer.export_jwks()
print(json.dumps(jwks, indent=2))

# 의존성 설치
pnpm add jose

// etdi-verifier.ts
import { createRemoteJWKSet, jwtVerify } from 'jose';
 
interface ETDIToolDefinition {
  name: string;
  version: string;
  schema: Record<string, unknown>;
  permissions: string[];
  provider: string;
  signature: string;       // ES256 JWT
  public_key_ref: string;  // JWKS URL
}
 
interface VerificationResult {
  valid: boolean;
  reason?: string;
}
 
// createRemoteJWKSet은 내부적으로 키를 캐싱합니다
// 추가 TTL 제어가 필요하면 cooldownThreshold 옵션을 사용합니다
const jwksCache = new Map<string, ReturnType<typeof createRemoteJWKSet>>();
 
function getJWKS(jwksUrl: string) {
  if (!jwksCache.has(jwksUrl)) {
    jwksCache.set(jwksUrl, createRemoteJWKSet(new URL(jwksUrl)));
  }
  return jwksCache.get(jwksUrl)!;
}
 
/**
 * ETDI 도구 정의의 JWT 서명을 검증합니다.
 * 검증 실패 시 도구 호출을 거부해야 합니다.
 */
async function verifyETDITool(tool: ETDIToolDefinition): Promise<VerificationResult> {
  try {
    const JWKS = getJWKS(tool.public_key_ref);
 
    // jose가 JWKS 조회 + ES256 서명 검증을 한 번에 처리합니다
    const { payload } = await jwtVerify(tool.signature, JWKS, {
      algorithms: ['ES256'],
    });
 
    // JWT가 유효해도 서명된 내용과 실제 정의가 다를 수 있으므로
    // 핵심 필드를 직접 비교합니다
    const permissionsMatch =
      JSON.stringify([...tool.permissions].sort()) ===
      JSON.stringify([...((payload.permissions as string[]) ?? [])].sort());
 
    if (
      payload.name !== tool.name ||
      payload.version !== tool.version ||
      payload.provider !== tool.provider ||
      !permissionsMatch
    ) {
      return {
        valid: false,
        reason: '서명 페이로드와 도구 정의가 불일치합니다 — 변조 의심.',
      };
    }
 
    return { valid: true };
  } catch (error) {
    return {
      valid: false,
      reason: `검증 오류: ${error instanceof Error ? error.message : '알 수 없는 오류'}`,
    };
  }
}
 
/**
 * 버전 변경을 감지하여 재승인 여부를 판단합니다.
 * 승인된 버전은 로컬 SQLite나 JSON 파일에 저장합니다.
 */
function hasVersionChanged(approvedVersion: string, currentVersion: string): boolean {
  return approvedVersion !== currentVersion;
}
 
// 도구 발견 시 호출되는 핸들러
async function onToolDiscovered(tool: ETDIToolDefinition) {
  const result = await verifyETDITool(tool);
 
  if (!result.valid) {
    console.error(`[ETDI] 도구 "${tool.name}" 차단: ${result.reason}`);
    return; // 도구 등록 거부 — 에이전트에 노출되지 않습니다
  }
 
  const approvedVersion = getApprovedVersion(tool.name, tool.provider);
  if (approvedVersion && hasVersionChanged(approvedVersion, tool.version)) {
    console.warn(
      `[ETDI] 도구 "${tool.name}" 버전 변경 감지: ${approvedVersion} → ${tool.version}`
    );
    await promptUserReapproval(tool);
    return;
  }
 
  console.log(`[ETDI] 도구 "${tool.name}" v${tool.version} 서명 검증 완료`);
  registerTool(tool);
}
 
// 실제 구현에서는 별도 모듈에 위치합니다
declare function getApprovedVersion(name: string, provider: string): string | null;
declare function promptUserReapproval(tool: ETDIToolDefinition): Promise<void>;
declare function registerTool(tool: ETDIToolDefinition): void;

# 의존성 설치
uv add cryptography PyJWT requests

# oauth_scope_guard.py
from functools import wraps
from typing import Callable, Any
import jwt
from jwt.algorithms import ECAlgorithm
import requests
 
 
class OAuthScopeGuard:
    """
    도구 호출 전 OAuth 토큰의 서명을 JWKS 공개 키로 검증하고,
    ETDI 도구 정의의 permissions를 충족하는지 확인합니다.
    """
 
    def __init__(self, jwks_url: str):
        self.jwks_url = jwks_url
        self._cached_public_key = None
 
    def _fetch_public_key(self) -> Any:
        """JWKS 엔드포인트에서 EC 공개 키를 가져옵니다 (인메모리 캐싱)."""
        if self._cached_public_key:
            return self._cached_public_key
 
        resp = requests.get(self.jwks_url, timeout=5)
        resp.raise_for_status()
        keys = resp.json().get("keys", [])
 
        ec_key = next((k for k in keys if k.get("kty") == "EC"), None)
        if not ec_key:
            raise ValueError("JWKS에 EC 키가 없습니다.")
 
        # PyJWT 내장 변환 함수로 JWKS EC 키를 cryptography 공개 키 객체로 변환합니다
        self._cached_public_key = ECAlgorithm.from_jwk(ec_key)
        return self._cached_public_key
 
    def require_scopes(self, required_scopes: list[str]):
        """도구 핸들러에 적용하는 데코레이터"""
        def decorator(func: Callable) -> Callable:
            @wraps(func)
            async def wrapper(*args, request_context: dict, **kwargs) -> Any:
                token = request_context.get("oauth_token")
                if not token:
                    raise PermissionError("OAuth 토큰이 없습니다. 인증이 필요합니다.")
 
                # 1단계: JWKS 공개 키로 토큰 서명을 검증합니다
                # verify_signature: False는 절대 사용하지 않습니다
                try:
                    public_key = self._fetch_public_key()
                    claims = jwt.decode(
                        token,
                        public_key,
                        algorithms=["ES256"],
                    )
                except jwt.ExpiredSignatureError:
                    raise PermissionError("토큰이 만료됐습니다.")
                except jwt.InvalidTokenError as e:
                    raise PermissionError(f"유효하지 않은 토큰: {e}")
 
                # 2단계: 토큰 스코프와 도구 정의 permissions를 비교합니다
                token_scopes = set(claims.get("scope", "").split())
                missing_scopes = set(required_scopes) - token_scopes
 
                if missing_scopes:
                    raise PermissionError(
                        f"권한 부족. 필요한 스코프: {missing_scopes}. "
                        f"보유한 스코프: {token_scopes}"
                    )
 
                return await func(*args, request_context=request_context, **kwargs)
            return wrapper
        return decorator
 
 
# MCP 도구 핸들러에 적용 예시
guard = OAuthScopeGuard(jwks_url="https://auth.example.com/.well-known/jwks.json")
 
@guard.require_scopes(["email:send"])
async def handle_send_email(
    to: str,
    subject: str,
    body: str,
    *,
    request_context: dict,   # 키워드 전용 인자로 위치 인자와의 혼동을 방지합니다
) -> dict:
    """
    ETDI 정의의 permissions: ["email:send"]와
    런타임 OAuth 토큰의 scope가 일치해야만 실행됩니다.
    """
    return {"status": "sent", "to": to}