Coverage for src/ecnl/security/token_verifier.py: 99%

1"""TokenVerifier implementation for the streamable-http transport. 

2 

3Implements ``mcp.server.auth.provider.TokenVerifier`` against the 

4auth-server's JWKS endpoint. Validates RS256-signed access tokens per 

5identity-platform-go ADR-0008 / ADR-0010 / ADR-0015 — the token's 

6``iss`` must match the configured issuer, the signing key must be the 

7one auth-server's JWKS publishes for the JOSE header's ``kid``, the 

8token must not be expired, and (when configured) the ``aud`` claim 

9must include the resource server's URL. 

10 

11The verifier surfaces ``actor_type`` and ``agent_id`` claims (ADR-0015) 

12through ``AccessToken.scopes`` and an extra dict so downstream 

13authorization decisions can read the principal kind without re-parsing 

14the JWT. 

15""" 

16 

17from __future__ import annotations 

18 

19import logging 

20import os 

21from dataclasses import dataclass 

22 

23import jwt 

24from mcp.server.auth.provider import AccessToken 

25 

26from .jwks import JWKSCache 

27 

28logger = logging.getLogger(__name__) 

29 

30 

31@dataclass 

32class JWKSTokenVerifier: 

33 """RS256 + JWKS bearer-token verifier. 

34 

35 Implements the duck-typed TokenVerifier protocol the mcp SDK 

36 expects. Returns ``None`` on every failure — the SDK turns ``None`` 

37 into a 401 response. Avoid leaking the rejection reason in the 

38 return value; per RFC 6749 §5.2 we never give a client more 

39 information than ``invalid_token``. 

40 """ 

41 

42 issuer: str 

43 """Expected ``iss`` claim. Tokens with any other issuer are rejected 

44 outright — this is the trust anchor for the entire validation 

45 chain.""" 

46 

47 audience: str | None 

48 """Expected ``aud`` claim. When set, the token must list this value 

49 in its audience array. RFC 8707 resource indicator semantics: this 

50 is the MCP server's public URL.""" 

51 

52 jwks: JWKSCache 

53 """Caches and refreshes auth-server's signing keys.""" 

54 

55 leeway_seconds: int = 30 

56 """Clock-skew tolerance applied to ``exp`` / ``iat`` / ``nbf``.""" 

57 

58 async def verify_token(self, token: str) -> AccessToken | None: 

59 """Validate the bearer token and project it into an AccessToken. 

60 

61 Returns ``None`` on every failure (bad signature, expired, 

62 wrong issuer, wrong audience, unknown ``kid``, missing 

63 required claim, malformed JWT). The mcp SDK's authentication 

64 middleware maps that to a 401 response with no body. 

65 """ 

66 try: 

67 header = jwt.get_unverified_header(token) 

68 except jwt.InvalidTokenError as exc: 

69 logger.debug("rejecting token: malformed JOSE header (%s)", exc) 

70 return None 

71 

72 kid = header.get("kid") 

73 if not kid: 

74 logger.debug("rejecting token: JOSE header missing kid") 

75 return None 

76 

77 try: 

78 key = await self.jwks.get(kid) 

79 except KeyError as exc: 

80 logger.debug("rejecting token: %s", exc) 

81 return None 

82 

83 try: 

84 claims = jwt.decode( 

85 token, 

86 key=key, 

87 algorithms=["RS256"], 

88 issuer=self.issuer, 

89 audience=self.audience, 

90 leeway=self.leeway_seconds, 

91 options={"require": ["exp", "iat", "iss", "sub"]}, 

92 ) 

93 except jwt.InvalidTokenError as exc: 

94 logger.debug("rejecting token: %s", exc) 

95 return None 

96 

97 return _to_access_token(token, claims) 

98 

99 

100def _to_access_token(raw: str, claims: dict) -> AccessToken: 

101 """Project verified claims into the mcp SDK's AccessToken shape. 

102 

103 ``client_id`` mirrors the JWT's ``client_id`` claim (set by 

104 auth-server on every issued token); ``scopes`` parses the 

105 space-delimited RFC 9068 ``scope`` claim. ``actor_type``, 

106 ``agent_id``, and ``sub`` are duplicated into the scopes list as 

107 ``actor_type:<value>`` etc. so downstream policy code can read them 

108 via the SDK without a separate claim accessor. The raw claims are 

109 not exposed by the AccessToken model — duplicating into scopes is 

110 the conventional bridge. 

111 """ 

112 scopes = _parse_scope(claims.get("scope")) 

113 if actor_type := claims.get("actor_type"): 

114 scopes.append(f"actor_type:{actor_type}") 

115 if agent_id := claims.get("agent_id"): 

116 scopes.append(f"agent_id:{agent_id}") 

117 if sub := claims.get("sub"): 

118 scopes.append(f"sub:{sub}") 

119 return AccessToken( 

120 token=raw, 

121 client_id=str(claims.get("client_id") or claims.get("sub") or ""), 

122 scopes=scopes, 

123 expires_at=int(claims["exp"]), 

124 ) 

125 

126 

127def _parse_scope(scope: str | None) -> list[str]: 

128 """Split a space-delimited RFC 9068 scope claim into a list.""" 

129 if not scope: 

130 return [] 

131 return [s for s in scope.split() if s] 

132 

133 

134def build_token_verifier() -> JWKSTokenVerifier | None: 

135 """Build the verifier from environment variables. 

136 

137 Returns ``None`` when ``MCP_AUTH_ENABLED`` is unset or false — the 

138 caller in :mod:`ecnl.server` interprets that as "skip the auth 

139 middleware entirely" and leaves the streamable-http transport 

140 open. Defaulting to off keeps the local-dev flow unchanged. 

141 

142 Required when enabled: 

143 * ``MCP_AUTH_ISSUER_URL`` — auth-server origin used to discover 

144 JWKS and validate the ``iss`` claim. 

145 

146 Optional: 

147 * ``MCP_AUTH_RESOURCE_URL`` — this server's public URL; when set 

148 every token must list it in its ``aud`` claim (RFC 8707). 

149 * ``MCP_AUTH_JWKS_TTL_SECONDS`` — JWKS cache lifetime (default 3600). 

150 * ``MCP_AUTH_LEEWAY_SECONDS`` — clock-skew tolerance (default 30). 

151 """ 

152 if not _env_flag("MCP_AUTH_ENABLED"): 

153 return None 

154 issuer = os.environ.get("MCP_AUTH_ISSUER_URL", "").strip() 

155 if not issuer: 

156 raise RuntimeError("MCP_AUTH_ENABLED is true but MCP_AUTH_ISSUER_URL is unset") 

157 resource = os.environ.get("MCP_AUTH_RESOURCE_URL", "").strip() or None 

158 ttl = _env_int("MCP_AUTH_JWKS_TTL_SECONDS", 3600) 

159 leeway = _env_int("MCP_AUTH_LEEWAY_SECONDS", 30) 

160 return JWKSTokenVerifier( 

161 issuer=issuer, 

162 audience=resource, 

163 jwks=JWKSCache(issuer_url=issuer, ttl_seconds=ttl), 

164 leeway_seconds=leeway, 

165 ) 

166 

167 

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

169 """Read an env var as a boolean. 

170 

171 Accepts ``1``, ``true``, ``yes``, ``on`` (case-insensitive) as true. 

172 """ 

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

174 if not value: 

175 return default 

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

177 

178 

179def _env_int(name: str, default: int) -> int: 

180 """Read an env var as an int, returning ``default`` on bad input.""" 

181 raw = os.environ.get(name, "").strip() 

182 if not raw: 

183 return default 

184 try: 

185 return int(raw) 

186 except ValueError: 

187 logger.warning("%s=%r is not an integer; using default %d", name, raw, default) 

188 return default