Implementing Per-Tool RBAC and Real-Time PII Blocking with Cloudflare DLP + OPA

┌──────────────────────────────────────────────────────┐
│  AI 에이전트 / MCP 클라이언트                        │
└────────────────────┬─────────────────────────────────┘
                     │ HTTP/SSE (Bearer JWT 포함)
┌────────────────────▼─────────────────────────────────┐
│  레이어 1: Cloudflare AI Gateway (전송 레이어)        │
│  ├─ DLP Firewall: 프롬프트·응답 민감 데이터 탐지     │
│  ├─ Guardrails: 유해 콘텐츠 분류                     │
│  └─ 레이트 리미팅, 감사 로그                         │
└────────────────────┬─────────────────────────────────┘
                     │
┌────────────────────▼─────────────────────────────────┐
│  레이어 2: OPA (정책 레이어)                         │
│  ├─ JWT 클레임 파싱                                  │
│  ├─ 도구 이름·사용자 역할·리소스 범위 평가           │
│  └─ allow / deny 결정 반환                           │
└────────────────────┬─────────────────────────────────┘
                     │
┌────────────────────▼─────────────────────────────────┐
│  [선행] OAuth 2.0 신원 레이어 (연결 전 토큰 발급)    │
│  ├─ Keycloak / Entra ID / Auth0                      │
│  └─ 서명된 JWT 발급, 도구별 스코프 포함              │
└──────────────────────────────────────────────────────┘

# policies/mcp/authz.rego
package mcp.authz
 
default allow = false
 
# JWT resource_access 클레임에 요청 도구가 포함된 경우 허용
allow {
  tool := input.tool_name
  server_id := input.server_id
  roles := input.token.resource_access[server_id].roles
  roles[_] == tool
}

# policies/mcp/authz_extended.rego
package mcp.authz
 
import future.keywords.in
 
default allow = false
 
# 관리자는 모든 도구 허용
allow {
  "admin" in input.token.realm_access.roles
}
 
# analyst 역할은 민감도 레벨 2 이하 도구만 허용
allow {
  "analyst" in input.token.realm_access.roles
  tool_sensitivity[input.tool_name] <= 2
}
 
# power_user는 민감도 3 이하 도구를 업무 시간(09~18시)에만 허용
# 참고: 테스트 가능성을 위해 현재 시각을 input.current_hour로 주입합니다.
allow {
  "power_user" in input.token.realm_access.roles
  tool_sensitivity[input.tool_name] <= 3
  input.current_hour >= 9
  input.current_hour < 18
}
 
# 도구별 민감도 레벨 매핑
tool_sensitivity := {
  "file_read":    1,
  "db_read":      2,
  "db_write":     3,
  "secret_fetch": 4,
  "admin_exec":   5,
}

# policies/mcp/authz_test.rego
package mcp.authz_test
 
import data.mcp.authz
 
# analyst가 file_read(민감도 1) 호출 → 허용
test_analyst_can_read_files {
  authz.allow with input as {
    "tool_name": "file_read",
    "server_id": "mcp-server-files",
    "current_hour": 10,
    "token": {
      "realm_access": {"roles": ["analyst"]},
      "resource_access": {}
    }
  }
}
 
# analyst가 secret_fetch(민감도 4) 호출 → 거부
test_analyst_cannot_fetch_secrets {
  not authz.allow with input as {
    "tool_name": "secret_fetch",
    "server_id": "mcp-server-secrets",
    "current_hour": 10,
    "token": {
      "realm_access": {"roles": ["analyst"]},
      "resource_access": {}
    }
  }
}
 
# power_user가 업무 시간(10시)에 db_write(민감도 3) 호출 → 허용
test_power_user_allowed_in_business_hours {
  authz.allow with input as {
    "tool_name": "db_write",
    "server_id": "mcp-server-db",
    "current_hour": 10,
    "token": {
      "realm_access": {"roles": ["power_user"]},
      "resource_access": {}
    }
  }
}
 
# power_user가 업무 시간 외(22시)에 db_write 호출 → 거부
test_power_user_blocked_outside_hours {
  not authz.allow with input as {
    "tool_name": "db_write",
    "server_id": "mcp-server-db",
    "current_hour": 22,
    "token": {
      "realm_access": {"roles": ["power_user"]},
      "resource_access": {}
    }
  }
}

# 정책 테스트 실행
opa test policies/ -v

# mcp_server/auth_interceptor.py
import httpx
from typing import Any
 
OPA_URL = "http://opa:8181/v1/data/mcp/authz"
 
async def check_opa_policy(
    tool_name: str,
    token_claims: dict,
    server_id: str,
    current_hour: int,
) -> bool:
    """
    OPA REST API(/v1/data/mcp/authz)에 정책 평가를 요청합니다.
    - 로컬 사이드카: 네트워크 홉 없이 sub-ms 수준
    - 원격 HTTP API: 약 5~15ms (timeout=0.5는 원격 호출 기준의 여유값)
    """
    payload = {
        "input": {
            "tool_name": tool_name,
            "server_id": server_id,
            "current_hour": current_hour,
            "token": token_claims,
        }
    }
    async with httpx.AsyncClient() as client:
        response = await client.post(OPA_URL, json=payload, timeout=0.5)
        result = response.json()
        # OPA 응답 구조: {"result": {"allow": true/false}}
        return result.get("result", {}).get("allow", False)
 
 
async def authorized_tool_call(
    tool_name: str,
    arguments: dict,
    token_claims: dict,
    server_id: str,
    current_hour: int,
) -> Any:
    """
    OPA 정책 평가 후 허용된 경우에만 도구를 실행합니다.
    MCP SDK 핸들러에서 직접 호출하는 방식으로 시그니처 충돌을 방지합니다.
    """
    allowed = await check_opa_policy(tool_name, token_claims, server_id, current_hour)
    if not allowed:
        raise PermissionError(
            f"도구 '{tool_name}' 호출이 정책에 의해 거부되었습니다."
        )
    return await dispatch_tool(tool_name, arguments)
 
 
async def dispatch_tool(tool_name: str, arguments: dict) -> Any:
    """실제 도구 구현을 디스패치합니다."""
    if tool_name == "file_read":
        return read_file(arguments["path"])
    elif tool_name == "db_write":
        return write_to_db(arguments["query"], arguments["data"])
    else:
        raise ValueError(f"알 수 없는 도구: {tool_name}")

# mcp_server/server.py — FastMCP 또는 공식 Python SDK와 연동 예시
from datetime import datetime, timezone
from mcp.server import Server
from mcp.types import Tool, TextContent
from auth_interceptor import authorized_tool_call
 
server = Server("mcp-server-files")
 
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(name="file_read", description="파일을 읽습니다", inputSchema={...}),
        Tool(name="db_write", description="DB에 씁니다", inputSchema={...}),
    ]
 
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    # JWT 파싱은 별도 미들웨어(Bearer 토큰 검증)에서 처리 후 context로 전달
    token_claims = server.get_context().token_claims
    current_hour = datetime.now(timezone.utc).hour
 
    result = await authorized_tool_call(
        tool_name=name,
        arguments=arguments,
        token_claims=token_claims,
        server_id="mcp-server-files",
        current_hour=current_hour,
    )
    return [TextContent(type="text", text=str(result))]

# cloudflare_dlp.tf
terraform {
  required_providers {
    cloudflare = {
      source  = "cloudflare/cloudflare"
      version = "~> 4.0"
    }
  }
}
 
# MCP 트래픽용 커스텀 DLP 프로파일
resource "cloudflare_dlp_profile" "mcp_sensitive_data" {
  account_id  = var.cloudflare_account_id
  name        = "MCP Sensitive Data Profile"
  type        = "custom"
  description = "MCP AI 트래픽에서 PII 및 자격증명 탐지"
 
  # 신용카드 번호 탐지 (사전 정의 패턴)
  entry {
    name    = "Credit Card Numbers"
    enabled = true
    type    = "predefined"
  }
 
  # API 키 커스텀 패턴 (sk-로 시작하는 OpenAI 키)
  entry {
    name    = "API Keys and Secrets"
    enabled = true
    type    = "custom"
    pattern {
      regex    = "sk-[a-zA-Z0-9]{48}"
      # context 필드로 주변 키워드를 조건에 추가하면 오탐을 줄일 수 있습니다
      context  = "(?i)(api[_-]?key|secret|token)"
      operator = "matchCount"
      count    = 1
    }
  }
 
  # 소스 코드 탐지
  entry {
    name    = "Source Code"
    enabled = true
    type    = "predefined"
  }
}
 
# AI Gateway에 DLP 정책 연결
resource "cloudflare_gateway_policy" "mcp_dlp_policy" {
  account_id  = var.cloudflare_account_id
  name        = "Block MCP Sensitive Data"
  description = "AI 프롬프트/응답의 민감 데이터 차단"
  enabled     = true
  precedence  = 10
 
  filters = ["http"]
  action  = "block"
 
  rule_settings {
    block_page_enabled = true
    block_page_reason  = "민감한 데이터가 감지되어 요청이 차단되었습니다."
 
    dlp_profile {
      id = cloudflare_dlp_profile.mcp_sensitive_data.id
    }
  }
 
  # AI Gateway 엔드포인트만 대상으로 지정
  traffic = "http.request.uri matches \".*gateway.ai.cloudflare.com.*\""
}

[사용자 → SSN 포함 프롬프트 전송]
          │
          ▼
  Cloudflare AI Gateway
  ┌──────────────────────────────────┐
  │ DLP Firewall                     │
  │  → SSN 패턴 감지 (###-##-####)   │
  │  → Action: BLOCK                 │
  │  → Logpush: S3/Splunk로 이벤트   │
  └──────────┬───────────────────────┘
             │ (차단 — LLM에 도달하지 않음)
             ▼
  [사용자에게 차단 응답 반환]
 
[정상 프롬프트]
  AI Gateway → LLM → 응답 반환
                        │
              [응답에 소스 코드 포함 시]
              DLP Firewall → BLOCK
              [클라이언트 전달 차단]