MCP-AUTHZ-02 — Per-Tool Scope Definition#

# Python example with scope-based authorization
from typing import Dict, List, Set, Optional
import re
import fnmatch

class ToolAuthorizationManager:
    def __init__(self):
        self.tool_scopes = {}
        self.scope_hierarchy = {
            "fs:admin": ["fs:write", "fs:read"],
            "fs:write": ["fs:read"],
            "db:admin": ["db:write", "db:read"],
            "db:write": ["db:read"]
        }

    def register_tool(self, tool_name: str, config: Dict):
        """Register tool with its required scopes and constraints."""
        self.tool_scopes[tool_name] = {
            "required_scopes": set(config.get("scopes", [])),
            "resource_patterns": config.get("resource_patterns", ["*"]),
            "parameter_constraints": config.get("parameter_constraints", {}),
            "rate_limit": config.get("rate_limit"),
            "denied_operations": set(config.get("denied_patterns", []))
        }

    def authorize_tool_execution(self, tool_name: str, user_scopes: Set[str],
                                parameters: Dict) -> tuple[bool, Optional[str]]:
        """Check if user can execute tool with given parameters."""

        if tool_name not in self.tool_scopes:
            return False, f"Unknown tool: {tool_name}"

        tool_config = self.tool_scopes[tool_name]

        # Check scope authorization
        effective_scopes = self._expand_scopes(user_scopes)
        if not tool_config["required_scopes"].issubset(effective_scopes):
            missing = tool_config["required_scopes"] - effective_scopes
            return False, f"Missing scopes: {missing}"

        # Check parameter constraints
        for param, constraint in tool_config["parameter_constraints"].items():
            if param in parameters:
                if isinstance(constraint, str):  # Regex pattern
                    if not re.match(constraint, str(parameters[param])):
                        return False, f"Parameter {param} violates constraint"
                elif isinstance(constraint, (int, float)):  # Max value
                    if parameters[param] > constraint:
                        return False, f"Parameter {param} exceeds maximum"

        # Check resource patterns for file operations
        if "path" in parameters:
            allowed = False
            for pattern in tool_config["resource_patterns"]:
                if fnmatch.fnmatch(parameters["path"], pattern):
                    allowed = True
                    break
            if not allowed:
                return False, f"Resource path not allowed: {parameters['path']}"

        # Check for denied operations in query tools
        if "query" in parameters:
            query_upper = parameters["query"].upper()
            for denied in tool_config["denied_operations"]:
                if denied in query_upper:
                    return False, f"Operation not allowed: {denied}"

        return True, None

    def _expand_scopes(self, user_scopes: Set[str]) -> Set[str]:
        """Expand scopes based on hierarchy."""
        expanded = set(user_scopes)
        for scope in user_scopes:
            if scope in self.scope_hierarchy:
                expanded.update(self.scope_hierarchy[scope])
        return expanded

    def get_available_tools(self, user_scopes: Set[str]) -> List[str]:
        """Return list of tools available to user based on scopes."""
        effective_scopes = self._expand_scopes(user_scopes)
        available = []
        for tool_name, config in self.tool_scopes.items():
            if config["required_scopes"].issubset(effective_scopes):
                available.append(tool_name)
        return available

// TypeScript example with decorators for scope enforcement
interface ToolScope {
    scopes: string[];
    resourcePatterns?: string[];
    parameterConstraints?: Record<string, any>;
    rateLimit?: string;
}

// Decorator for tool scope definition
function RequireScopes(scopeConfig: ToolScope) {
    return function(target: any, propertyName: string, descriptor: PropertyDescriptor) {
        const originalMethod = descriptor.value;

        descriptor.value = async function(...args: any[]) {
            const context = args[0]; // Assume first arg is context with user info

            // Validate scopes
            const userScopes = context.user?.scopes || [];
            const hasRequiredScopes = scopeConfig.scopes.every(
                scope => userScopes.includes(scope)
            );

            if (!hasRequiredScopes) {
                throw new AuthorizationError(
                    `Missing required scopes: ${scopeConfig.scopes.join(', ')}`
                );
            }

            // Validate parameters
            const params = args[1];
            if (scopeConfig.parameterConstraints) {
                for (const [param, constraint] of Object.entries(
                    scopeConfig.parameterConstraints
                )) {
                    if (!validateParameter(params[param], constraint)) {
                        throw new ValidationError(
                            `Parameter ${param} violates constraints`
                        );
                    }
                }
            }

            // Check rate limits
            if (scopeConfig.rateLimit) {
                const limited = await checkRateLimit(
                    context.user.id,
                    propertyName,
                    scopeConfig.rateLimit
                );
                if (limited) {
                    throw new RateLimitError('Rate limit exceeded');
                }
            }

            return originalMethod.apply(this, args);
        };

        return descriptor;
    };
}

// Usage example
class FileTools {
    @RequireScopes({
        scopes: ['fs:read'],
        resourcePatterns: ['/public/**', '/shared/**'],
        parameterConstraints: {
            path: /^\/public\/|^\/shared\//,
            maxSize: 10 * 1024 * 1024
        }
    })
    async readFile(context: Context, params: { path: string }): Promise<string> {
        // Implementation
    }

    @RequireScopes({
        scopes: ['fs:write', 'fs:admin'],
        resourcePatterns: ['/uploads/**'],
        rateLimit: '100/hour'
    })
    async writeFile(context: Context, params: { path: string, content: string }) {
        // Implementation
    }
}