Coverage for mcpgateway / middleware / path_filter.py: 100%

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

2"""Centralized path filtering for middleware chain optimization. 

3 

4Copyright 2025 

5SPDX-License-Identifier: Apache-2.0 

6 

7This module provides cached path exclusion checks for middleware to reduce 

8per-request overhead. Each middleware has specific skip semantics that are 

9preserved (exact vs prefix matching). 

10 

11Important: preserve existing skip semantics (exact vs prefix). 

12- ObservabilityMiddleware: exact matches + "/static/" prefix + configured include/exclude patterns 

13- AuthContextMiddleware: exact matches + "/static/" prefix 

14- RequestLoggingMiddleware: prefix matches 

15- DBQueryLoggingMiddleware: exact matches + "/static" prefix (no trailing slash) 

16 

17Note on /healthz: This endpoint is used by translate.py for standalone MCP server 

18wrapping, while the main gateway uses /health and /ready. Both are included 

19for compatibility across deployment modes. 

20""" 

21 

22# Standard 

23from functools import lru_cache 

24import logging 

25import re 

26from typing import FrozenSet, Pattern, Tuple 

27 

28# First-Party 

29from mcpgateway.config import settings 

30 

31logger = logging.getLogger(__name__) 

32 

33# Observability: exact matches + "/static/" prefix + allowlist 

34# NOTE: /healthz is included for translate.py compatibility (gateway uses /health, /ready) 

35# See: mcpgateway/translate.py, mcpgateway/config.py:observability_include_paths/observability_exclude_paths 

36OBSERVABILITY_SKIP_EXACT: FrozenSet[str] = frozenset( 

37 [ 

38 "/health", 

39 "/healthz", # translate.py only, kept for compatibility 

40 "/ready", 

41 "/metrics", 

42 "/admin/events", 

43 ] 

44) 

45OBSERVABILITY_SKIP_PREFIXES: Tuple[str, ...] = ("/static/", "/admin/observability/") 

46 

47# AuthContext: exact matches + "/static/" prefix (preserves pre-allowlist behavior) 

48AUTH_CONTEXT_SKIP_EXACT: FrozenSet[str] = frozenset( 

49 [ 

50 "/health", 

51 "/healthz", # translate.py only, kept for compatibility 

52 "/ready", 

53 "/metrics", 

54 ] 

55) 

56AUTH_CONTEXT_SKIP_PREFIXES: Tuple[str, ...] = ("/static/",) 

57 

58# Request logging: prefix matches (current behavior skips "/health/security") 

59REQUEST_LOG_SKIP_PREFIXES: Tuple[str, ...] = ( 

60 "/health", 

61 "/healthz", 

62 "/static", 

63 "/favicon.ico", 

64) 

65 

66# DB query logging: exact matches + "/static" prefix (no trailing slash) 

67# NOTE: /healthz included for translate.py compatibility 

68DB_QUERY_LOG_SKIP_EXACT: FrozenSet[str] = frozenset( 

69 [ 

70 "/health", 

71 "/healthz", # translate.py only, kept for compatibility 

72 "/ready", 

73 ] 

74) 

75DB_QUERY_LOG_SKIP_PREFIXES: Tuple[str, ...] = ("/static",) 

76 

77 

78def _matches_prefix(path: str, prefixes: Tuple[str, ...]) -> bool: 

79 """Return True if path starts with any prefix in prefixes. 

80 

81 Args: 

82 path: The URL path to check 

83 prefixes: Tuple of prefix strings to match against 

84 

85 Returns: 

86 True if path starts with any of the prefixes 

87 

88 Examples: 

89 >>> _matches_prefix("/static/css/app.css", ("/static/", "/assets/")) 

90 True 

91 >>> _matches_prefix("/api/users", ("/static/", "/assets/")) 

92 False 

93 >>> _matches_prefix("/health", ("/health",)) 

94 True 

95 >>> _matches_prefix("/healthy", ("/health/",)) 

96 False 

97 """ 

98 return any(path.startswith(prefix) for prefix in prefixes) 

99 

100 

101def _matches_any_regex(path: str, patterns: Tuple[Pattern[str], ...]) -> bool: 

102 """Return True if path matches any regex in patterns. 

103 

104 Args: 

105 path: The URL path to check. 

106 patterns: Tuple of compiled regex patterns to evaluate. 

107 

108 Returns: 

109 True if any pattern matches the path. 

110 

111 Examples: 

112 >>> import re 

113 >>> patterns = (re.compile(r"^/api/v[0-9]+/"), re.compile(r"^/rpc/?$")) 

114 >>> _matches_any_regex("/api/v1/users", patterns) 

115 True 

116 >>> _matches_any_regex("/rpc", patterns) 

117 True 

118 >>> _matches_any_regex("/health", patterns) 

119 False 

120 >>> _matches_any_regex("/api/users", patterns) 

121 False 

122 """ 

123 return any(pattern.search(path) for pattern in patterns) 

124 

125 

126@lru_cache(maxsize=1) 

127def _get_observability_include_regex() -> Tuple[Pattern[str], ...]: 

128 """Compile include regex patterns from settings for observability filtering. 

129 

130 Returns: 

131 Tuple of compiled regex patterns; invalid patterns are skipped. 

132 """ 

133 compiled: list[Pattern[str]] = [] 

134 for pattern in settings.observability_include_paths: 

135 try: 

136 compiled.append(re.compile(pattern)) 

137 except re.error as exc: 

138 logger.warning("Invalid observability_include_paths regex '%s': %s", pattern, exc) 

139 return tuple(compiled) 

140 

141 

142@lru_cache(maxsize=1) 

143def _get_observability_exclude_regex() -> Tuple[Pattern[str], ...]: 

144 """Compile exclude regex patterns from settings for observability filtering. 

145 

146 Returns: 

147 Tuple of compiled regex patterns; invalid patterns are skipped. 

148 """ 

149 compiled: list[Pattern[str]] = [] 

150 for pattern in settings.observability_exclude_paths: 

151 try: 

152 compiled.append(re.compile(pattern)) 

153 except re.error as exc: 

154 logger.warning("Invalid observability_exclude_paths regex '%s': %s", pattern, exc) 

155 return tuple(compiled) 

156 

157 

158@lru_cache(maxsize=256) 

159def should_skip_observability(path: str) -> bool: 

160 """Skip logic for ObservabilityMiddleware. 

161 

162 Skips health endpoints (exact match), static files (prefix match), configured 

163 include/exclude patterns (include first, then exclude). 

164 

165 Args: 

166 path: The URL path from request.url.path 

167 

168 Returns: 

169 True if observability should be skipped for this path 

170 

171 Examples: 

172 >>> should_skip_observability("/health") 

173 True 

174 >>> should_skip_observability("/metrics") 

175 True 

176 >>> should_skip_observability("/static/css/app.css") 

177 True 

178 >>> should_skip_observability("/health/security") 

179 True 

180 >>> should_skip_observability("/tools") 

181 True 

182 >>> should_skip_observability("/rpc") 

183 False 

184 """ 

185 if path in OBSERVABILITY_SKIP_EXACT or _matches_prefix(path, OBSERVABILITY_SKIP_PREFIXES): 

186 return True 

187 

188 if _matches_any_regex(path, _get_observability_exclude_regex()): 

189 return True 

190 

191 include_patterns = _get_observability_include_regex() 

192 if include_patterns and not _matches_any_regex(path, include_patterns): 

193 return True 

194 

195 return False 

196 

197 

198@lru_cache(maxsize=256) 

199def should_skip_auth_context(path: str) -> bool: 

200 """Skip logic for AuthContextMiddleware. 

201 

202 Args: 

203 path: The URL path from request.url.path 

204 

205 Returns: 

206 True if auth context extraction should be skipped for this path 

207 

208 Examples: 

209 >>> should_skip_auth_context("/health") 

210 True 

211 >>> should_skip_auth_context("/static/js/app.js") 

212 True 

213 >>> should_skip_auth_context("/tools") 

214 False 

215 """ 

216 return path in AUTH_CONTEXT_SKIP_EXACT or _matches_prefix(path, AUTH_CONTEXT_SKIP_PREFIXES) 

217 

218 

219@lru_cache(maxsize=256) 

220def should_skip_request_logging(path: str) -> bool: 

221 """Skip logic for RequestLoggingMiddleware. 

222 

223 Uses prefix matching - this means /health/security will also be skipped 

224 (intentional: preserves existing behavior). 

225 

226 Args: 

227 path: The URL path from request.url.path 

228 

229 Returns: 

230 True if request logging should be skipped for this path 

231 

232 Examples: 

233 >>> should_skip_request_logging("/health") 

234 True 

235 >>> should_skip_request_logging("/health/security") 

236 True 

237 >>> should_skip_request_logging("/static/css/app.css") 

238 True 

239 >>> should_skip_request_logging("/favicon.ico") 

240 True 

241 >>> should_skip_request_logging("/ready") 

242 False 

243 >>> should_skip_request_logging("/metrics") 

244 False 

245 """ 

246 return _matches_prefix(path, REQUEST_LOG_SKIP_PREFIXES) 

247 

248 

249@lru_cache(maxsize=256) 

250def should_skip_db_query_logging(path: str) -> bool: 

251 """Skip logic for DBQueryLoggingMiddleware. 

252 

253 Skips health endpoints (exact match) and static files (prefix match). 

254 Note: /metrics is NOT skipped for DB query logging (may query DB for metrics). 

255 

256 Args: 

257 path: The URL path from request.url.path 

258 

259 Returns: 

260 True if DB query logging should be skipped for this path 

261 

262 Examples: 

263 >>> should_skip_db_query_logging("/health") 

264 True 

265 >>> should_skip_db_query_logging("/ready") 

266 True 

267 >>> should_skip_db_query_logging("/static/css/app.css") 

268 True 

269 >>> should_skip_db_query_logging("/health/security") 

270 False 

271 >>> should_skip_db_query_logging("/metrics") 

272 False 

273 """ 

274 return path in DB_QUERY_LOG_SKIP_EXACT or _matches_prefix(path, DB_QUERY_LOG_SKIP_PREFIXES) 

275 

276 

277def clear_all_caches() -> None: 

278 """Clear all path filter caches. 

279 

280 Useful for testing to ensure cache state doesn't affect test isolation. 

281 

282 Examples: 

283 >>> clear_all_caches() 

284 >>> should_skip_request_logging.cache_info().currsize >= 0 

285 True 

286 """ 

287 should_skip_observability.cache_clear() 

288 should_skip_auth_context.cache_clear() 

289 should_skip_request_logging.cache_clear() 

290 should_skip_db_query_logging.cache_clear() 

291 _get_observability_include_regex.cache_clear() 

292 _get_observability_exclude_regex.cache_clear()