Coverage for src/ecnl/ports/inbound.py: 100%
40 statements
« prev ^ index » next coverage.py v7.14.2, created at 2026-06-28 08:19 +0000
1"""Inbound ports the application service expects.
3The MCP server's inbound surface is the tool dispatch path: a tool
4handler reads the caller's identity from the bearer token, asks the
5:class:`Authorizer` whether the call may proceed, and only then invokes
6the application service. The port lives here so the adapter side can
7provide either a pass-through (local dev) or a policy-service-backed
8implementation without the application layer having to know which is
9wired.
10"""
12from __future__ import annotations
14from dataclasses import dataclass
15from enum import StrEnum
16from typing import Protocol
19class DecisionKind(StrEnum):
20 """Result of an authorization check.
22 A string-backed Enum keeps log messages and audit envelopes
23 readable without an extra translation step.
24 """
26 ALLOW = "allow"
27 DENY = "deny"
30@dataclass(frozen=True)
31class Decision:
32 """The authorizer's verdict on a single tool call.
34 ``reason`` is surfaced verbatim to the caller when the decision is
35 :class:`DecisionKind.DENY`. Per RFC 6749 §5.2 we don't want to leak
36 operational detail in error messages — implementations should
37 return a short, stable string the caller can show a user without
38 triggering an enumeration probe (e.g. "tool not allowed for
39 actor"), not the raw policy lookup result.
40 """
42 kind: DecisionKind
43 reason: str = ""
45 @classmethod
46 def allow(cls) -> Decision:
47 return cls(DecisionKind.ALLOW)
49 @classmethod
50 def deny(cls, reason: str = "tool call not permitted") -> Decision:
51 return cls(DecisionKind.DENY, reason)
53 @property
54 def allowed(self) -> bool:
55 return self.kind is DecisionKind.ALLOW
58@dataclass(frozen=True)
59class AuthorizationRequest:
60 """The full input to an authorization check.
62 Bundling the fields into a value object keeps the protocol stable
63 when new attributes ship (per-tool sensitivity, cost class, RAR
64 `authorization_details`) — adapter implementations can pick the
65 fields they care about without breaking signature changes
66 cascading through every caller.
67 """
69 actor_type: str
70 """One of "user", "service", "agent" per ADR-0015. Empty when
71 ``MCP_AUTH_ENABLED=false`` and no token was presented."""
73 agent_id: str
74 """Stable agent identifier per ADR-0015. Empty for non-agent
75 actors and for unauthenticated calls."""
77 subject: str
78 """The ``sub`` claim from the bearer token — typically the
79 human-readable user id, or the agent's own id when the agent
80 acts on its own behalf."""
82 client_id: str
83 """The ``client_id`` claim, which doubles as the OAuth client
84 identifier for audit / metering attribution."""
86 tool_name: str
87 """The name of the MCP tool being invoked, e.g. ``get_standings``."""
89 sensitivity: str = "public"
90 """Architecture-roadmap classification for the tool's data
91 sensitivity. Forwarded so policy engines can branch on it without
92 a separate tool-catalog lookup. Empty / unrecognised values
93 default to ``public`` at the adapter boundary."""
95 cost_class: str = "metered"
96 """Architecture-roadmap classification for the cost profile of
97 the call. ``metered`` means the operator pays per upstream
98 request; ``billable`` means the call emits a Lago billing event;
99 ``free`` means a no-op cache hit."""
101 rate_limit_class: str = "standard"
102 """Architecture-roadmap classification for the rate-limit bucket.
103 The actual ceiling lives in deployment config — this field is
104 the bucket key only."""
107class Authorizer(Protocol):
108 """Inbound authorization port.
110 Implementations are async because the production adapter calls
111 :mod:`authorization-policy-service` over HTTP; the pass-through is
112 still async to keep call sites identical regardless of which
113 implementation is wired.
114 """
116 async def authorize(self, req: AuthorizationRequest) -> Decision:
117 """Decide whether the supplied tool call may proceed."""
118 ...