| 1 | """Fail-closed MCP risk and authorization policy.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | from app.contracts import ( |
| 5 | McpAuthContext, |
| 6 | McpErrorCode, |
| 7 | McpGatewayError, |
| 8 | McpRiskClass, |
| 9 | McpToolDefinition, |
| 10 | ) |
| 11 | |
| 12 | |
| 13 | class McpToolPolicy: |
| 14 | """Iteration 5A permits authenticated SAFE_READ definitions only.""" |
| 15 | |
| 16 | def authorize(self, definition: McpToolDefinition, auth: McpAuthContext) -> None: |
| 17 | if definition.risk_class is not McpRiskClass.SAFE_READ: |
| 18 | raise McpGatewayError(McpErrorCode.MCP_TOOL_DENIED) |
| 19 | if not auth.service_identity.strip(): |
| 20 | raise McpGatewayError(McpErrorCode.UNAUTHORIZED) |
| 21 | if definition.requires_user and auth.user_id is None: |
| 22 | raise McpGatewayError(McpErrorCode.UNAUTHORIZED) |
| 23 | missing_scopes = set(definition.required_scopes).difference(auth.scopes) |
| 24 | if missing_scopes: |
| 25 | raise McpGatewayError(McpErrorCode.FORBIDDEN) |