Access Control for LLM Agent Tools Built with OPA — Designing a Runtime Context-Based Dynamic Policy Engine

Input(JSON) + Policy(Rego) + Data(JSON) → Decision(allow/deny)

# policy/mcp_tool_access.rego
package mcp.tool.access
 
import future.keywords.if
import future.keywords.in
 
default allow := false
 
# 기본 허용 조건: 아래 규칙을 모두 충족해야 함 (AND)
allow if {
    valid_role
    within_working_hours
    data_sensitivity_ok
    trusted_agent
}
 
# 역할 검증
valid_role if {
    input.principal.role in {"analyst", "engineer", "admin"}
}
 
# 근무 시간 검증 (UTC 기준 00:00~17:00)
# 주의: env.time은 서버가 UTC로 직접 채워야 합니다.
# 서버 로컬 타임존이 UTC가 아닌 경우 time.now_ns()를 서버에서 직접 주입하는 방식을 권장합니다.
within_working_hours if {
    hour := time.clock(time.parse_rfc3339_ns(input.env.time))[0]
    hour >= 0
    hour < 17
}
 
# 데이터 민감도 검증 (analyst는 민감도 3 미만만 허용)
data_sensitivity_ok if {
    input.principal.role == "analyst"
    input.tool.data_sensitivity < 3
}
 
data_sensitivity_ok if {
    input.principal.role in {"engineer", "admin"}
}
 
# 신뢰할 수 있는 에이전트 검증 (data.trusted_agents는 외부 JSON으로 관리)
trusted_agent if {
    input.client.agent_id in data.trusted_agents
}

# ETDI 도구 정의 — JWT 페이로드 (decoded 상태)
{
  "tool_id": "query_database_v2",
  "provider": "internal-data-team",
  "version": "2.1.0",
  "permissions": ["db:read", "schema:read"],
  "issued_at": "2026-04-14T00:00:00Z"
}
# 실제 전송 형태 (Header.Payload.Signature 포함):
# eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJ0b29sX2lkIjoicXVlcnlfZGF0YWJhc2VfdjIiLC4uLn0.<서명>

[LLM Agent]
    ↓ tool_call 요청
[MCP Host/Broker]
    ↓ ① ETDI 서명 검증 (정적 레이어)
      ├─ JWT 서명 유효성 확인
      ├─ 버전 불변성 확인 (Rug Pull 탐지)
      └─ 제공자 신원 확인 (Tool Squatting 탐지)
          └─ 검증 실패 → deny + 보안 로그
    ↓ ② OPA PDP 쿼리 (동적 레이어)
      ├─ 사용자 역할·속성 평가
      ├─ 시간·환경 컨텍스트 평가
      └─ 에이전트 식별자 평가
          └─ deny → 차단 + 감사 로그
    ↓ allow
[MCP Server] → 도구 실행

# mcp_server.py
import httpx
from datetime import datetime, timezone
from fastmcp import FastMCP, Context
from fastmcp.exceptions import ToolError
 
mcp = FastMCP("secure-data-server")
 
OPA_URL = "http://localhost:8181/v1/data/mcp/tool/access/allow"
 
async def check_policy(
    principal: dict,
    client: dict,
    tool_name: str,
    tool_metadata: dict,
    args: dict,
) -> bool:
    """OPA PDP에 정책 평가 요청을 보냅니다."""
    input_payload = {
        "input": {
            "principal": principal,
            "client": client,
            "tool": {
                "name": tool_name,
                **tool_metadata,
            },
            "args": args,
            "env": {
                # 환경 정보는 반드시 서버 측에서 직접 채워 넣어야 합니다.
                # 클라이언트/에이전트가 제공한 값을 그대로 신뢰하면 정책 우회가 가능합니다.
                "time": datetime.now(timezone.utc).isoformat(),
                "network": "internal",
            },
        }
    }
 
    async with httpx.AsyncClient(timeout=0.5) as http:  # 500ms 타임아웃
        try:
            resp = await http.post(OPA_URL, json=input_payload)
            resp.raise_for_status()
            return resp.json().get("result", False)
        except (httpx.TimeoutException, httpx.HTTPError):
            # OPA 장애 시 허용이 아닌 거부로 폴백 — deny-safe 원칙
            return False
 
 
@mcp.tool()
async def query_financial_records(
    table: str,
    filter_expr: str,
    ctx: Context,  # FastMCP가 Context 객체를 자동 주입합니다
) -> dict:
    """금융 레코드를 조회합니다."""
    # Context에서 호출자 정보를 추출하는 방법은 MCP 서버 구현에 따라 다릅니다.
    # 아래는 인증 미들웨어에서 ctx에 주입된 값을 사용하는 예시입니다.
    principal = getattr(ctx, "principal", {"role": "unknown"})
    client_info = getattr(ctx, "client", {"agent_id": "unknown"})
 
    allowed = await check_policy(
        principal=principal,
        client=client_info,
        tool_name="query_financial_records",
        tool_metadata={"server": "postgres-mcp", "data_sensitivity": 3},
        args={"table": table, "filter": filter_expr},
    )
 
    if not allowed:
        raise ToolError(
            "접근이 거부되었습니다. 현재 컨텍스트에서 이 도구를 사용할 권한이 없습니다.",
            code="POLICY_DENIED",
        )
 
    # 실제 DB 조회 로직
    return {"records": [], "total": 0}

# pyproject.toml — 최소 의존성
[project]
dependencies = [
    "fastmcp>=2.0",
    "httpx>=0.27",
    "pyjwt[crypto]>=2.8",
    "cryptography>=41.0",
]

# etdi_verifier.py
import jwt
from cryptography.hazmat.primitives import serialization
from dataclasses import dataclass
 
 
@dataclass
class ETDIToolDefinition:
    tool_id: str
    provider: str
    version: str
    permissions: list[str]
    raw_token: str  # 원본 JWT 문자열 (Header.Payload.Signature)
 
 
class ETDIVerificationError(Exception):
    pass
 
 
class ETDIVerifier:
    def __init__(self, trusted_providers: dict[str, str]):
        """
        trusted_providers: {provider_id: PEM 형식 공개 키 문자열}
        """
        # cryptography 41+ 에서는 backend 인자 없이 직접 호출합니다.
        self._keys = {
            pid: serialization.load_pem_public_key(pem.encode())
            for pid, pem in trusted_providers.items()
        }
        # 버전 레지스트리: {tool_id: 승인된 버전}
        # 프로덕션에서는 DB나 Redis에 영구 저장하고 재시작 시 복원해야 합니다.
        self._approved_versions: dict[str, str] = {}
 
    def verify(self, token: str) -> ETDIToolDefinition:
        """JWT 서명을 검증하고 ETDIToolDefinition을 반환합니다."""
        # 1단계: 헤더에서 제공자 ID 추출 (서명 검증 전 미리 확인)
        unverified = jwt.decode(token, options={"verify_signature": False})
        provider_id = unverified.get("provider")
 
        if provider_id not in self._keys:
            raise ETDIVerificationError(
                f"알 수 없는 제공자: {provider_id}. Tool Squatting 가능성이 있습니다."
            )
 
        # 2단계: 공개 키로 서명 검증 (JWT 위조·변조 탐지)
        try:
            payload = jwt.decode(
                token,
                key=self._keys[provider_id],
                algorithms=["RS256"],
            )
        except jwt.InvalidSignatureError as e:
            raise ETDIVerificationError("도구 서명 검증 실패: Rug Pull 공격 의심") from e
 
        tool_id = payload["tool_id"]
        current_version = payload["version"]
 
        # 3단계: 버전 불변성 확인 (Rug Pull 탐지)
        if tool_id in self._approved_versions:
            approved = self._approved_versions[tool_id]
            if approved != current_version:
                raise ETDIVerificationError(
                    f"도구 버전 불일치 — 승인된 버전: {approved}, "
                    f"현재 버전: {current_version}. 재승인이 필요합니다."
                )
 
        return ETDIToolDefinition(
            tool_id=tool_id,
            provider=provider_id,
            version=current_version,
            permissions=payload.get("permissions", []),
            raw_token=token,
        )
 
    def approve_version(self, tool_id: str, version: str) -> None:
        """사용자가 특정 버전을 명시적으로 승인합니다."""
        self._approved_versions[tool_id] = version

# policy_cache.py
import asyncio
import hashlib
import json
import time
from typing import Any
 
 
class PolicyDecisionCache:
    """OPA 정책 결정 결과를 TTL 기반으로 캐싱합니다."""
 
    def __init__(self, default_ttl: float = 30.0):
        self._cache: dict[str, tuple[bool, float]] = {}
        self._default_ttl = default_ttl
        self._lock = asyncio.Lock()
 
    def _make_key(self, input_data: dict[str, Any]) -> str:
        serialized = json.dumps(input_data, sort_keys=True)
        return hashlib.sha256(serialized.encode()).hexdigest()
 
    async def get(self, input_data: dict[str, Any]) -> bool | None:
        key = self._make_key(input_data)
        async with self._lock:
            if key in self._cache:
                result, expires_at = self._cache[key]
                if time.monotonic() < expires_at:
                    return result
                del self._cache[key]
        return None  # 캐시 미스
 
    async def set(
        self,
        input_data: dict[str, Any],
        decision: bool,
        ttl: float | None = None,
    ) -> None:
        key = self._make_key(input_data)
        expires_at = time.monotonic() + (ttl or self._default_ttl)
        async with self._lock:
            self._cache[key] = (decision, expires_at)
 
    async def invalidate_all(self) -> None:
        """정책 번들 갱신 시 전체 캐시를 무효화합니다."""
        async with self._lock:
            self._cache.clear()

# opa_client.py — 캐시를 활용한 OPA 클라이언트
import httpx
from policy_cache import PolicyDecisionCache
 
 
class OPAClient:
    def __init__(
        self,
        opa_url: str = "http://localhost:8181",
        cache_ttl: float = 30.0,
    ):
        self._base_url = opa_url
        self._cache = PolicyDecisionCache(default_ttl=cache_ttl)
 
    async def decide(
        self,
        policy_path: str,  # 예: "mcp.tool.access" → /v1/data/mcp/tool/access
        input_data: dict,
        ttl_override: float | None = None,
    ) -> bool:
        cached = await self._cache.get(input_data)
        if cached is not None:
            return cached
 
        # OPA Rego 패키지명(점 구분)을 URL 경로(슬래시 구분)로 변환합니다.
        # "mcp.tool.access" → "/v1/data/mcp/tool/access"
        path_segment = policy_path.replace(".", "/")
        url = f"{self._base_url}/v1/data/{path_segment}"
 
        async with httpx.AsyncClient(timeout=0.5) as client:
            resp = await client.post(url, json={"input": input_data})
            resp.raise_for_status()
            decision: bool = resp.json().get("result", False)
 
        # deny 결과는 짧은 TTL로 정책 변경이 빠르게 반영되도록 합니다.
        effective_ttl = ttl_override or (5.0 if not decision else 30.0)
        await self._cache.set(input_data, decision, ttl=effective_ttl)
        return decision

{
  "principal": {
    "id": "user-123",
    "role": "analyst",
    "department": "finance",
    "clearance_level": 2
  },
  "client": {
    "agent_id": "claude-agent-v3.1",
    "session_id": "sess_abc123",
    "session_age_minutes": 15,
    "ip_address": "10.0.1.45"
  },
  "tool": {
    "name": "query_financial_records",
    "server": "postgres-mcp",
    "version": "2.1.0",
    "data_sensitivity": 3,
    "provider": "internal-data-team"
  },
  "args": {
    "table": "revenue_q1_2026",
    "filter": "region = 'APAC'"
  },
  "env": {
    "time": "2026-04-14T10:30:00Z",
    "network_zone": "internal",
    "request_id": "req_xyz789"
  }
}

# policy/financial_tools.rego
package mcp.tool.financial
 
import future.keywords.if
import future.keywords.in
 
default allow := false
 
# 금융 도구 접근 허용 조건 (모두 충족 필수 — AND)
allow if {
    authorized_role
    sensitivity_within_clearance
    within_business_hours
    trusted_agent_version
    not suspicious_session
}
 
authorized_role if {
    input.principal.role in {"analyst", "finance_manager", "auditor"}
}
 
# 사용자 클리어런스 레벨이 도구 민감도 이상이어야 함
sensitivity_within_clearance if {
    input.principal.clearance_level >= input.tool.data_sensitivity
}
 
# 주의: env.time은 서버가 UTC로 직접 채워야 합니다. 클라이언트 제공값 사용 금지.
within_business_hours if {
    hour := time.clock(time.parse_rfc3339_ns(input.env.time))[0]
    hour >= 9
    hour < 18
}
 
# 신뢰할 수 있는 에이전트 버전 목록 (data.json으로 외부 관리)
trusted_agent_version if {
    input.client.agent_id in data.trusted_agent_ids
}
 
# 30분 이상 된 세션에서의 고민감도 접근은 재인증 필요
suspicious_session if {
    input.tool.data_sensitivity >= 4
    input.client.session_age_minutes > 30
}