Coverage for src/ecnl/adapters/inbound/authorization.py: 100%

1"""Inbound authorization adapters. 

2 

3Two implementations: 

4 

5* :class:`PassThroughAuthorizer` — the default; allows every call. 

6 This matches the local-dev / stdio posture where the trust boundary 

7 is the process boundary itself. 

8* :class:`PolicyServiceAuthorizer` — the production adapter; calls 

9 ``authorization-policy-service`` per the architecture roadmap. 

10 

11The factory :func:`build_authorizer` reads the environment and chooses 

12between them so the composition root (:mod:`nwsl.server`) doesn't have 

13to branch on config. 

14""" 

15 

16from __future__ import annotations 

17 

18import logging 

19import os 

20from dataclasses import dataclass 

21 

22import httpx 

23 

24from ...ports.inbound import AuthorizationRequest, Authorizer, Decision 

25 

26logger = logging.getLogger(__name__) 

27 

28 

29class PassThroughAuthorizer: 

30 """Authorizer that allows every call. 

31 

32 Default wiring for local development and the stdio transport. 

33 Logs each decision at debug so an operator who flips the log 

34 level can confirm the pass-through is in fact intentional. 

35 """ 

36 

37 async def authorize(self, req: AuthorizationRequest) -> Decision: 

38 logger.debug("pass-through authorize: tool=%s actor=%s", req.tool_name, req.actor_type) 

39 return Decision.allow() 

40 

41 

42@dataclass 

43class PolicyServiceAuthorizer: 

44 """Authorizer backed by ``authorization-policy-service``. 

45 

46 Maps the MCP tool call onto the policy service's ``/evaluate`` 

47 request shape (``subject_id``, ``resource``, ``action``). The 

48 resource is ``mcp:<server>:<tool_name>`` so policies can target 

49 individual tools; the action is ``invoke``. 

50 """ 

51 

52 base_url: str 

53 """Public URL of authorization-policy-service. Trailing slashes 

54 are trimmed at construction time.""" 

55 

56 server_name: str 

57 """The MCP server's identifier, used as the middle segment of the 

58 resource path. ``jk-mcp-ecnl`` here; ``jk-mcp-nwsl`` in that repo.""" 

59 

60 http_client: httpx.AsyncClient | None = None 

61 """Optional pre-built async client. When ``None`` the adapter 

62 constructs a short-lived client per request. Production wiring 

63 should pass a shared client so connection pooling kicks in.""" 

64 

65 timeout_seconds: float = 1.5 

66 """Hard ceiling on policy-evaluation latency. The policy service 

67 is an in-network call; anything slower than this is a sign that 

68 the service is unavailable, and per ADR-0019 we fail closed 

69 rather than hold up the tool dispatch.""" 

70 

71 fail_closed: bool = True 

72 """When ``True`` (default) any error from the policy call results 

73 in :meth:`Decision.deny`. Set to ``False`` only in environments 

74 where availability outranks correctness (typically internal 

75 staging where a missing policy is a bug, not a security event).""" 

76 

77 def __post_init__(self) -> None: 

78 self.base_url = self.base_url.rstrip("/") 

79 

80 async def authorize(self, req: AuthorizationRequest) -> Decision: 

81 body, err = await self._fetch_decision(req) 

82 if err is not None: 

83 return self._error_decision(err) 

84 if bool(body.get("allowed")): 

85 return Decision.allow() 

86 # The policy service's reason is acceptable to surface — it 

87 # already obeys the no-enumeration rule per ADR-0006. 

88 return Decision.deny(str(body.get("reason") or "tool call not permitted")) 

89 

90 async def _fetch_decision(self, req: AuthorizationRequest) -> tuple[dict, httpx.HTTPError | None]: 

91 """POST the authorization request to the policy service. 

92 

93 Returns ``(body, None)`` on success or ``({}, error)`` when the 

94 call fails. Split from :meth:`authorize` so each function 

95 stays under the project's cyclomatic-complexity cap. 

96 """ 

97 client = self.http_client or httpx.AsyncClient(timeout=self.timeout_seconds) 

98 owns_client = self.http_client is None 

99 payload = { 

100 "subject_id": req.subject or req.client_id or req.agent_id, 

101 "resource": f"mcp:{self.server_name}:{req.tool_name}", 

102 "action": "invoke", 

103 } 

104 try: 

105 resp = await client.post(f"{self.base_url}/evaluate", json=payload) 

106 resp.raise_for_status() 

107 return resp.json(), None 

108 except httpx.HTTPError as exc: 

109 logger.warning("policy service call failed: %s", exc) 

110 return {}, exc 

111 finally: 

112 if owns_client: 

113 await client.aclose() 

114 

115 def _error_decision(self, _err: httpx.HTTPError) -> Decision: 

116 """Decide between deny / allow on a policy-service error. 

117 

118 ``fail_closed`` is the production default; the staging-only 

119 flag flips it to allow so a degraded policy service does not 

120 take down a non-billable environment. 

121 """ 

122 if self.fail_closed: 

123 return Decision.deny("authorization unavailable") 

124 return Decision.allow() 

125 

126 

127def build_authorizer() -> Authorizer: 

128 """Construct the authorizer from environment variables. 

129 

130 Returns :class:`PassThroughAuthorizer` unless 

131 ``MCP_AUTHZ_URL`` is set to a non-empty value; the presence of the 

132 URL is what signals that policy enforcement is desired. Other 

133 knobs: 

134 

135 * ``MCP_AUTHZ_SERVER_NAME`` — overrides the default server name 

136 used as the middle segment of the resource path. Defaults to 

137 ``jk-mcp-ecnl``. 

138 * ``MCP_AUTHZ_TIMEOUT_SECONDS`` — policy-call ceiling 

139 (default 1.5). 

140 * ``MCP_AUTHZ_FAIL_OPEN`` — when truthy, the adapter degrades to 

141 allow on policy-service errors. Default fail-closed. 

142 """ 

143 url = os.environ.get("MCP_AUTHZ_URL", "").strip() 

144 if not url: 

145 return PassThroughAuthorizer() 

146 server_name = os.environ.get("MCP_AUTHZ_SERVER_NAME", "jk-mcp-ecnl").strip() or "jk-mcp-ecnl" 

147 timeout_raw = os.environ.get("MCP_AUTHZ_TIMEOUT_SECONDS", "1.5").strip() 

148 try: 

149 timeout = float(timeout_raw) 

150 except ValueError: 

151 logger.warning("MCP_AUTHZ_TIMEOUT_SECONDS=%r is not a float; using 1.5", timeout_raw) 

152 timeout = 1.5 

153 fail_open = _env_flag("MCP_AUTHZ_FAIL_OPEN") 

154 return PolicyServiceAuthorizer( 

155 base_url=url, 

156 server_name=server_name, 

157 timeout_seconds=timeout, 

158 fail_closed=not fail_open, 

159 ) 

160 

161 

162def _env_flag(name: str, default: bool = False) -> bool: 

163 """Boolean env-var parser shared with :mod:`nwsl.security`.""" 

164 value = os.environ.get(name, "").strip().lower() 

165 if not value: 

166 return default 

167 return value in {"1", "true", "yes", "on"}