Coverage for mcpgateway / plugins / framework / hooks / http.py: 100%

1# -*- coding: utf-8 -*- 

2"""Location: ./mcpgateway/plugins/framework/models/http.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Teryl Taylor 

6 

7Pydantic models for http hooks and payloads. 

8""" 

9 

10# Standard 

11from enum import Enum 

12 

13# Third-Party 

14from pydantic import RootModel 

15 

16# First-Party 

17from mcpgateway.plugins.framework.models import PluginPayload, PluginResult 

18 

19 

20class HttpHeaderPayload(RootModel[dict[str, str]], PluginPayload): 

21 """An HTTP dictionary of headers used in the pre/post HTTP forwarding hooks.""" 

22 

23 def __iter__(self): # type: ignore[no-untyped-def] 

24 """Custom iterator function to override root attribute. 

25 

26 Returns: 

27 A custom iterator for header dictionary. 

28 """ 

29 return iter(self.root) 

30 

31 def __getitem__(self, item: str) -> str: 

32 """Custom getitem function to override root attribute. 

33 

34 Args: 

35 item: The http header key. 

36 

37 Returns: 

38 A custom accesser for the header dictionary. 

39 """ 

40 return self.root[item] 

41 

42 def __setitem__(self, key: str, value: str) -> None: 

43 """Mutation helper — operates on the underlying dict. 

44 

45 .. warning:: 

46 Because ``PluginPayload`` is frozen, this should only be used 

47 on freshly-created copies (e.g. inside ``model_copy``). The 

48 executor deep-copies payloads before handing them to plugins, 

49 so in-place writes on the copy are safe. 

50 

51 Args: 

52 key: The http header key. 

53 value: The http header value to be set. 

54 """ 

55 self.root[key] = value 

56 

57 def __len__(self) -> int: 

58 """Custom len function to override root attribute. 

59 

60 Returns: 

61 The len of the header dictionary. 

62 """ 

63 return len(self.root) 

64 

65 

66HttpHeaderPayloadResult = PluginResult[HttpHeaderPayload] 

67 

68 

69class HttpHookType(str, Enum): 

70 """Hook types for HTTP request processing and authentication. 

71 

72 These hooks allow plugins to: 

73 1. Transform request headers before processing (middleware layer) 

74 2. Implement custom user authentication systems (auth layer) 

75 3. Check and grant permissions (RBAC layer) 

76 4. Process responses after request completion (middleware layer) 

77 """ 

78 

79 HTTP_PRE_REQUEST = "http_pre_request" 

80 HTTP_POST_REQUEST = "http_post_request" 

81 HTTP_AUTH_RESOLVE_USER = "http_auth_resolve_user" 

82 HTTP_AUTH_CHECK_PERMISSION = "http_auth_check_permission" 

83 

84 

85class HttpPreRequestPayload(PluginPayload): 

86 """Payload for HTTP pre-request hook (middleware layer). 

87 

88 This payload contains immutable request metadata and a copy of headers 

89 that plugins can inspect. Invoked before any authentication processing. 

90 Plugins return only modified headers via PluginResult[HttpHeaderPayload]. 

91 

92 Attributes: 

93 path: HTTP path being requested. 

94 method: HTTP method (GET, POST, etc.). 

95 client_host: Client IP address (if available). 

96 client_port: Client port (if available). 

97 headers: Copy of HTTP headers that plugins can inspect and modify. 

98 """ 

99 

100 path: str 

101 method: str 

102 client_host: str | None = None 

103 client_port: int | None = None 

104 headers: HttpHeaderPayload 

105 

106 

107class HttpPostRequestPayload(HttpPreRequestPayload): 

108 """Payload for HTTP post-request hook (middleware layer). 

109 

110 Extends HttpPreRequestPayload with response information. 

111 Invoked after request processing is complete. 

112 Plugins can inspect response headers and status codes. 

113 

114 Attributes: 

115 response_headers: Response headers from the request (if available). 

116 status_code: HTTP status code from the response (if available). 

117 """ 

118 

119 response_headers: HttpHeaderPayload | None = None 

120 status_code: int | None = None 

121 

122 

123class HttpAuthResolveUserPayload(PluginPayload): 

124 """Payload for custom user authentication hook (auth layer). 

125 

126 Invoked inside get_current_user() to allow plugins to provide 

127 custom authentication mechanisms (LDAP, mTLS, external auth, etc.). 

128 Plugins return an authenticated user via PluginResult[dict]. 

129 

130 Attributes: 

131 credentials: The HTTP authorization credentials from bearer_scheme (if present). 

132 headers: Full request headers for custom auth extraction. 

133 client_host: Client IP address (if available). 

134 client_port: Client port (if available). 

135 """ 

136 

137 credentials: dict | None = None # HTTPAuthorizationCredentials serialized 

138 headers: HttpHeaderPayload 

139 client_host: str | None = None 

140 client_port: int | None = None 

141 

142 

143class HttpAuthCheckPermissionPayload(PluginPayload): 

144 """Payload for permission checking hook (RBAC layer). 

145 

146 Invoked before RBAC permission checks to allow plugins to: 

147 - Grant/deny permissions based on custom logic (e.g., token-based auth) 

148 - Bypass RBAC for certain authentication methods 

149 - Add additional permission checks (e.g., time-based, IP-based) 

150 - Implement custom authorization logic 

151 

152 Attributes: 

153 user_email: Email of the authenticated user 

154 permission: Required permission being checked (e.g., "tools.read", "servers.write") 

155 resource_type: Type of resource being accessed (e.g., "tool", "server", "prompt") 

156 team_id: Team context for the permission check (if applicable) 

157 is_admin: Whether the user has admin privileges 

158 auth_method: Authentication method used (e.g., "simple_token", "jwt", "oauth") 

159 client_host: Client IP address for IP-based permission checks 

160 user_agent: User agent string for device-based permission checks 

161 """ 

162 

163 user_email: str 

164 permission: str 

165 resource_type: str | None = None 

166 team_id: str | None = None 

167 is_admin: bool = False 

168 auth_method: str | None = None 

169 client_host: str | None = None 

170 user_agent: str | None = None 

171 

172 

173class HttpAuthCheckPermissionResultPayload(PluginPayload): 

174 """Result payload for permission checking hook. 

175 

176 Plugins return this to indicate whether permission should be granted. 

177 

178 Attributes: 

179 granted: Whether permission is granted (True) or denied (False) 

180 reason: Optional reason for the decision (for logging/auditing) 

181 """ 

182 

183 granted: bool 

184 reason: str | None = None 

185 

186 

187# Type aliases for hook results 

188HttpPreRequestResult = PluginResult[HttpHeaderPayload] 

189HttpPostRequestResult = PluginResult[HttpHeaderPayload] 

190HttpAuthResolveUserResult = PluginResult[dict] # Returns user dict (EmailUser serialized) 

191HttpAuthCheckPermissionResult = PluginResult[HttpAuthCheckPermissionResultPayload] 

192 

193 

194def _register_http_auth_hooks() -> None: 

195 """Register HTTP authentication and request hooks in the global registry. 

196 

197 This is called lazily to avoid circular import issues. 

198 Registers four hook types: 

199 - HTTP_PRE_REQUEST: Transform headers before authentication (middleware) 

200 - HTTP_POST_REQUEST: Inspect response after request completion (middleware) 

201 - HTTP_AUTH_RESOLVE_USER: Custom user authentication (auth layer) 

202 - HTTP_AUTH_CHECK_PERMISSION: Custom permission checking (RBAC layer) 

203 """ 

204 # Import here to avoid circular dependency at module load time 

205 # First-Party 

206 from mcpgateway.plugins.framework.hooks.registry import get_hook_registry # pylint: disable=import-outside-toplevel 

207 

208 registry = get_hook_registry() 

209 

210 # Only register if not already registered (idempotent) 

211 if not registry.is_registered(HttpHookType.HTTP_PRE_REQUEST): 

212 registry.register_hook(HttpHookType.HTTP_PRE_REQUEST, HttpPreRequestPayload, HttpPreRequestResult) 

213 registry.register_hook(HttpHookType.HTTP_POST_REQUEST, HttpPostRequestPayload, HttpPostRequestResult) 

214 registry.register_hook(HttpHookType.HTTP_AUTH_RESOLVE_USER, HttpAuthResolveUserPayload, HttpAuthResolveUserResult) 

215 registry.register_hook(HttpHookType.HTTP_AUTH_CHECK_PERMISSION, HttpAuthCheckPermissionPayload, HttpAuthCheckPermissionResult) 

216 

217 

218_register_http_auth_hooks()