Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/_experimental/mcp_server/client_allowlist.py: 47%

1""" 

2Gateway-level allowlist of MCP client applications (``general_settings.mcp_allowed_clients``). 

3 

4Each entry pairs an admin-chosen ``alias`` (shown in the dashboard and logs) with the ``value`` that 

5identifies the client. Only the value is compared, exactly and case-sensitively. 

6A caller that authenticated with a JWT is identified by the claim named in 

7``litellm_jwtauth.mcp_client_id_jwt_field``, a value asserted by the identity provider. 

8Every other caller is identified by the header named in ``general_settings.mcp_client_id_header``, 

9which the client picks itself, so that source is a policy control rather than a security boundary. 

10While the allowlist is set, a caller with no usable identity source is rejected. 

11""" 

12 

13from collections.abc import Mapping 

14from dataclasses import dataclass 

15from types import MappingProxyType 

16from typing import Final, Literal 

17 

18from pydantic import TypeAdapter, ValidationError 

19from typing_extensions import ReadOnly, TypedDict 

20 

21from litellm._logging import verbose_logger 

22from litellm.litellm_core_utils.dot_notation_indexing import get_nested_value 

23from litellm.types.mcp import MCPAllowedClient 

24 

25MCP_ALLOWED_CLIENTS_SETTING: Final = "mcp_allowed_clients" 

26MCP_CLIENT_ID_HEADER_SETTING: Final = "mcp_client_id_header" 

27MCP_CLIENT_ID_JWT_FIELD_SETTING: Final = "mcp_client_id_jwt_field" 

28_JWT_AUTH_SETTING: Final = "litellm_jwtauth" 

29 

30_ALLOWED_CLIENTS_ADAPTER: Final[TypeAdapter[list[MCPAllowedClient]]] = TypeAdapter(list[MCPAllowedClient]) 

31_OPTIONAL_NAME_ADAPTER: Final[TypeAdapter[str | None]] = TypeAdapter(str | None) 

32_OPTIONAL_MAPPING_ADAPTER: Final[TypeAdapter[dict[str, object] | None]] = TypeAdapter(dict[str, object] | None) 

33_NOBODY: Final[Mapping[str, str]] = MappingProxyType({}) 

34 

35 

36class MCPClientForbiddenBody(TypedDict): 

37 error: ReadOnly[Literal["Forbidden"]] 

38 details: ReadOnly[str] 

39 

40 

41@dataclass(frozen=True, slots=True) 

42class MCPClientAllowlist: 

43 """``aliases_by_value`` maps each admitted identity value to the alias the admin gave it.""" 

44 

45 aliases_by_value: Mapping[str, str] 

46 jwt_field: str | None 

47 header: str | None 

48 

49 

50@dataclass(frozen=True, slots=True) 

51class MCPClientIdentity: 

52 client_id: str 

53 source: Literal["jwt", "header"] 

54 source_name: str 

55 

56 @property 

57 def description(self) -> str: 

58 return f"'{self.client_id}' (from {'JWT claim' if self.source == 'jwt' else 'header'} '{self.source_name}')" 

59 

60 

61@dataclass(frozen=True, slots=True) 

62class MCPClientRejection: 

63 details: str 

64 

65 @property 

66 def response_body(self) -> MCPClientForbiddenBody: 

67 body: Final[MCPClientForbiddenBody] = {"error": "Forbidden", "details": self.details} 

68 return body 

69 

70 

71def _unidentified_rejection(reason: str) -> MCPClientRejection: 

72 return MCPClientRejection( 

73 details=f"{reason} This gateway only admits client applications listed in {MCP_ALLOWED_CLIENTS_SETTING}." 

74 ) 

75 

76 

77def parse_allowed_mcp_clients(raw_setting: object) -> Mapping[str, str] | None: 

78 """Value-to-alias mapping; None when the setting is absent (not enforced). A malformed setting admits nobody.""" 

79 if raw_setting is None: 79 ↛ 81line 79 didn't jump to line 81 because the condition on line 79 was always true

80 return None 

81 try: 

82 clients: Final = _ALLOWED_CLIENTS_ADAPTER.validate_python(raw_setting) 

83 except ValidationError: 

84 verbose_logger.warning( 

85 "%s is not a list of {alias, value} entries (%r); rejecting every MCP client until it is fixed", 

86 MCP_ALLOWED_CLIENTS_SETTING, 

87 raw_setting, 

88 ) 

89 return _NOBODY 

90 return MappingProxyType({client.value: client.alias for client in clients}) 

91 

92 

93def _parse_optional_name(setting_name: str, raw_setting: object) -> str | None: 

94 try: 

95 name: Final = _OPTIONAL_NAME_ADAPTER.validate_python(raw_setting) 

96 except ValidationError: 

97 verbose_logger.warning("%s is not a string (%r); ignoring it", setting_name, raw_setting) 

98 return None 

99 return name or None 

100 

101 

102def _jwt_field_from_general_settings(general_settings: Mapping[str, object]) -> str | None: 

103 try: 

104 jwt_auth: Final = _OPTIONAL_MAPPING_ADAPTER.validate_python(general_settings.get(_JWT_AUTH_SETTING)) 

105 except ValidationError: 

106 return None 

107 if jwt_auth is None: 

108 return None 

109 return _parse_optional_name( 

110 f"{_JWT_AUTH_SETTING}.{MCP_CLIENT_ID_JWT_FIELD_SETTING}", jwt_auth.get(MCP_CLIENT_ID_JWT_FIELD_SETTING) 

111 ) 

112 

113 

114def load_mcp_client_allowlist(general_settings: Mapping[str, object]) -> MCPClientAllowlist | None: 

115 """None when ``mcp_allowed_clients`` is unset, which admits every client.""" 

116 allowed_clients: Final = parse_allowed_mcp_clients(general_settings.get(MCP_ALLOWED_CLIENTS_SETTING)) 

117 if allowed_clients is None: 117 ↛ 119line 117 didn't jump to line 119 because the condition on line 117 was always true

118 return None 

119 header: Final = _parse_optional_name( 

120 MCP_CLIENT_ID_HEADER_SETTING, general_settings.get(MCP_CLIENT_ID_HEADER_SETTING) 

121 ) 

122 return MCPClientAllowlist( 

123 aliases_by_value=allowed_clients, 

124 jwt_field=_jwt_field_from_general_settings(general_settings), 

125 header=header.lower() if header is not None else None, 

126 ) 

127 

128 

129def resolve_mcp_client_identity( 

130 allowlist: MCPClientAllowlist, 

131 jwt_claims: Mapping[str, object] | None, 

132 headers: Mapping[str, str], 

133) -> MCPClientIdentity | MCPClientRejection: 

134 """A JWT caller is identified by its configured claim alone, so a header can never override the IdP.""" 

135 if jwt_claims is not None and allowlist.jwt_field is not None: 

136 claim: Final[object] = get_nested_value(data=jwt_claims, key_path=allowlist.jwt_field) 

137 if isinstance(claim, str) and claim: 

138 return MCPClientIdentity(client_id=claim, source="jwt", source_name=allowlist.jwt_field) 

139 return _unidentified_rejection( 

140 f"The JWT presented has no '{allowlist.jwt_field}' claim naming the client application." 

141 ) 

142 if allowlist.header is None: 

143 configured: Final = ( 

144 f"litellm_jwtauth.{MCP_CLIENT_ID_JWT_FIELD_SETTING} for JWT callers or {MCP_CLIENT_ID_HEADER_SETTING}" 

145 ) 

146 return _unidentified_rejection( 

147 f"No client identity source is configured for this request; set {configured} in general_settings." 

148 ) 

149 header_value: Final = headers.get(allowlist.header) 

150 if header_value: 

151 return MCPClientIdentity(client_id=header_value, source="header", source_name=allowlist.header) 

152 return _unidentified_rejection(f"The request has no '{allowlist.header}' header naming the client application.") 

153 

154 

155def check_mcp_client_allowed( 

156 allowlist: MCPClientAllowlist | None, 

157 jwt_claims: Mapping[str, object] | None, 

158 headers: Mapping[str, str], 

159) -> MCPClientRejection | None: 

160 if allowlist is None: 160 ↛ 162line 160 didn't jump to line 162 because the condition on line 160 was always true

161 return None 

162 identity: Final = resolve_mcp_client_identity(allowlist, jwt_claims, headers) 

163 if isinstance(identity, MCPClientRejection): 

164 return identity 

165 alias: Final = allowlist.aliases_by_value.get(identity.client_id) 

166 if alias is None: 

167 return MCPClientRejection( 

168 details=f"MCP client {identity.description} is not listed in this gateway's {MCP_ALLOWED_CLIENTS_SETTING}." 

169 ) 

170 verbose_logger.debug("Admitted MCP client '%s' identified as %s", alias, identity.description) 

171 return None