Yargı MCP

"""
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()