201 lines
7.2 KiB
Python
201 lines
7.2 KiB
Python
"""
|
|
Authorization policy engine for MCP tools
|
|
"""
|
|
|
|
import re
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PolicyAction(Enum):
|
|
ALLOW = "allow"
|
|
DENY = "deny"
|
|
|
|
|
|
@dataclass
|
|
class ToolPolicy:
|
|
"""Policy rule for MCP tool access"""
|
|
|
|
tool_pattern: str # regex pattern for tool names
|
|
required_scopes: list[str]
|
|
action: PolicyAction = PolicyAction.ALLOW
|
|
conditions: dict[str, Any] | None = None
|
|
|
|
def matches_tool(self, tool_name: str) -> bool:
|
|
"""Check if the policy applies to given tool"""
|
|
return bool(re.match(self.tool_pattern, tool_name))
|
|
|
|
def evaluate_scopes(self, user_scopes: list[str]) -> bool:
|
|
"""Check if user has required scopes"""
|
|
return all(scope in user_scopes for scope in self.required_scopes)
|
|
|
|
|
|
class PolicyEngine:
|
|
"""Authorization policy engine for Turkish legal database tools"""
|
|
|
|
def __init__(self):
|
|
self.policies: list[ToolPolicy] = []
|
|
self.default_action = PolicyAction.DENY
|
|
|
|
def add_policy(self, policy: ToolPolicy):
|
|
"""Add a policy rule"""
|
|
self.policies.append(policy)
|
|
logger.debug(f"Added policy: {policy.tool_pattern} -> {policy.required_scopes}")
|
|
|
|
def add_tool_scope_policy(
|
|
self,
|
|
tool_pattern: str,
|
|
required_scopes: str | list[str],
|
|
action: PolicyAction = PolicyAction.ALLOW,
|
|
):
|
|
"""Convenience method to add tool-scope policy"""
|
|
if isinstance(required_scopes, str):
|
|
required_scopes = [required_scopes]
|
|
|
|
policy = ToolPolicy(
|
|
tool_pattern=tool_pattern, required_scopes=required_scopes, action=action
|
|
)
|
|
self.add_policy(policy)
|
|
|
|
def authorize_tool_call(
|
|
self,
|
|
tool_name: str,
|
|
user_scopes: list[str],
|
|
user_claims: dict[str, Any] | None = None,
|
|
) -> tuple[bool, str | None]:
|
|
"""
|
|
Authorize a tool call
|
|
|
|
Returns:
|
|
(authorized: bool, reason: Optional[str])
|
|
"""
|
|
|
|
logger.debug(f"Authorizing tool '{tool_name}' for user with scopes: {user_scopes}")
|
|
|
|
matching_policies = [
|
|
policy for policy in self.policies if policy.matches_tool(tool_name)
|
|
]
|
|
|
|
if not matching_policies:
|
|
if self.default_action == PolicyAction.ALLOW:
|
|
logger.debug(f"No policies found for '{tool_name}', allowing by default")
|
|
return True, None
|
|
else:
|
|
logger.warning(f"No policies found for '{tool_name}', denying by default")
|
|
return False, f"No policy found for tool '{tool_name}', default deny"
|
|
|
|
# Check for explicit deny policies first
|
|
for policy in matching_policies:
|
|
if policy.action == PolicyAction.DENY:
|
|
if policy.evaluate_scopes(user_scopes):
|
|
logger.warning(f"Explicit deny policy matched for '{tool_name}'")
|
|
return False, f"Explicit deny policy for tool '{tool_name}'"
|
|
|
|
# Check allow policies
|
|
allow_policies = [
|
|
p for p in matching_policies if p.action == PolicyAction.ALLOW
|
|
]
|
|
|
|
if not allow_policies:
|
|
logger.warning(f"No allow policies found for '{tool_name}'")
|
|
return False, f"No allow policies found for tool '{tool_name}'"
|
|
|
|
for policy in allow_policies:
|
|
if policy.evaluate_scopes(user_scopes):
|
|
if self._evaluate_conditions(policy.conditions, user_claims):
|
|
logger.debug(f"Authorization granted for '{tool_name}'")
|
|
return True, None
|
|
|
|
logger.warning(f"Insufficient scopes for '{tool_name}'. Required: {[p.required_scopes for p in allow_policies]}, User has: {user_scopes}")
|
|
return False, f"Insufficient scopes for tool '{tool_name}'"
|
|
|
|
def _evaluate_conditions(
|
|
self,
|
|
conditions: dict[str, Any] | None,
|
|
user_claims: dict[str, Any] | None,
|
|
) -> bool:
|
|
"""Evaluate additional policy conditions"""
|
|
|
|
if not conditions:
|
|
return True
|
|
|
|
if not user_claims:
|
|
logger.debug("No user claims provided, conditions evaluation failed")
|
|
return False
|
|
|
|
for key, expected_value in conditions.items():
|
|
user_value = user_claims.get(key)
|
|
|
|
if isinstance(expected_value, list):
|
|
if user_value not in expected_value:
|
|
logger.debug(f"Condition failed: {key} = {user_value} not in {expected_value}")
|
|
return False
|
|
elif user_value != expected_value:
|
|
logger.debug(f"Condition failed: {key} = {user_value} != {expected_value}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def get_allowed_tools(self, user_scopes: list[str]) -> list[str]:
|
|
"""Get list of tool patterns user is allowed to call"""
|
|
|
|
allowed_tools = []
|
|
|
|
for policy in self.policies:
|
|
if policy.action == PolicyAction.ALLOW and policy.evaluate_scopes(
|
|
user_scopes
|
|
):
|
|
allowed_tools.append(policy.tool_pattern)
|
|
|
|
return allowed_tools
|
|
|
|
|
|
def create_turkish_legal_policies() -> PolicyEngine:
|
|
"""Create policy set for Turkish legal database MCP server"""
|
|
|
|
engine = PolicyEngine()
|
|
|
|
# Administrative tools (full access)
|
|
engine.add_tool_scope_policy(".*", ["mcp:tools:admin"])
|
|
|
|
# Search tools - require read access
|
|
engine.add_tool_scope_policy("search.*", ["mcp:tools:read"])
|
|
|
|
# Fetch/get document tools - require read access
|
|
engine.add_tool_scope_policy("get_.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("fetch.*", ["mcp:tools:read"])
|
|
|
|
# Specific Turkish legal database tools
|
|
engine.add_tool_scope_policy("search_yargitay.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_danistay.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_anayasa.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_rekabet.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_kik.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_emsal.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_uyusmazlik.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_sayistay.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_.*_bedesten", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_yerel_hukuk.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_istinaf_hukuk.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("search_kyb.*", ["mcp:tools:read"])
|
|
|
|
# Document retrieval tools
|
|
engine.add_tool_scope_policy("get_.*_document.*", ["mcp:tools:read"])
|
|
engine.add_tool_scope_policy("get_.*_markdown", ["mcp:tools:read"])
|
|
|
|
# Write operations (if any future tools need them)
|
|
engine.add_tool_scope_policy("create_.*", ["mcp:tools:write"])
|
|
engine.add_tool_scope_policy("update_.*", ["mcp:tools:write"])
|
|
engine.add_tool_scope_policy("delete_.*", ["mcp:tools:write"])
|
|
|
|
logger.info("Created Turkish legal database policy engine")
|
|
return engine
|
|
|
|
|
|
def create_default_policies() -> PolicyEngine:
|
|
"""Create a default policy set for MCP servers (backwards compatibility)"""
|
|
return create_turkish_legal_policies() |