Coverage for src / mcp_server_langgraph / core / checkpoint_validator.py: 83%
93 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-03 00:43 +0000
1"""Checkpoint configuration validation for fail-fast startup checks.
3This module provides validation for Redis checkpoint configuration to prevent
4runtime errors from malformed connection URLs. Implements the "fail fast"
5principle: detect configuration errors at application startup, not during
6first use.
8Background:
9-----------
10Production incident staging-758b8f744 where pods crashed at runtime due to
11unencoded special characters in Redis password. This validator would have
12caught the error during startup with a clear, actionable error message.
14Usage:
15------
16```python
17from mcp_server_langgraph.core.checkpoint_validator import validate_checkpoint_config
18from mcp_server_langgraph.core.config import settings
20# At application startup:
21validate_checkpoint_config(settings)
22```
23"""
25from typing import Any
26from urllib.parse import unquote
28from mcp_server_langgraph.observability.telemetry import logger
31class CheckpointValidationError(Exception):
32 """Raised when checkpoint configuration validation fails.
34 This exception indicates a configuration error that must be fixed
35 before the application can start safely.
36 """
39class CheckpointConfigValidator:
40 """Validator for checkpoint configuration with fail-fast semantics."""
42 # RFC 3986 reserved characters that MUST be percent-encoded in passwords
43 RESERVED_CHARS = set(":/?#[]@!$&'()*+,;=")
45 def __init__(self) -> None:
46 """Initialize checkpoint configuration validator."""
48 def validate_redis_url(
49 self,
50 url: str,
51 test_connection: bool = False,
52 ) -> None:
53 """Validate Redis connection URL format and encoding.
55 Args:
56 url: Redis connection URL to validate
57 test_connection: If True, attempt to connect to Redis (optional)
59 Raises:
60 CheckpointValidationError: If URL is invalid or improperly encoded
62 Examples:
63 >>> validator = CheckpointConfigValidator()
64 >>> validator.validate_redis_url("redis://:pass%2Fword@localhost:6379/1") # OK
65 >>> validator.validate_redis_url("redis://:pass/word@localhost:6379/1") # Raises error
66 """
67 if not url:
68 msg = (
69 "Redis URL cannot be empty when checkpoint_backend='redis'. "
70 "Please set CHECKPOINT_REDIS_URL environment variable."
71 )
72 raise CheckpointValidationError(msg)
74 # Validate basic URL format
75 if not url.startswith("redis://"):
76 msg = (
77 f"Invalid Redis URL scheme.\n"
78 f"Expected: redis://...\n"
79 f"Got: {url[:20] if len(url) > 20 else url}...\n"
80 f"Valid format: redis://[user]:[password]@[host]:[port]/[database]"
81 )
82 raise CheckpointValidationError(msg)
84 # Check for obviously invalid formats
85 if url == "redis://":
86 msg = (
87 "Incomplete Redis URL: 'redis://'\n"
88 "Valid format: redis://[user]:[password]@[host]:[port]/[database]\n"
89 "Example: redis://localhost:6379/1"
90 )
91 raise CheckpointValidationError(msg)
93 # Extract and validate password encoding
94 self._validate_password_encoding(url)
96 # Optionally test actual connection
97 if test_connection:
98 self._test_redis_connection(url)
100 def _validate_password_encoding(self, url: str) -> None:
101 """Validate that password (if present) is properly percent-encoded.
103 Args:
104 url: Redis URL to validate
106 Raises:
107 CheckpointValidationError: If password contains unencoded special characters
108 """
109 # Check if URL has authentication (contains @ after scheme)
110 if "@" not in url:
111 # No password, validation passes
112 return
114 try:
115 # Extract credentials portion (between :// and @)
116 after_scheme = url[8:] # Remove 'redis://'
117 at_pos = after_scheme.rfind("@")
119 if at_pos == -1: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true
120 return # No @ found (edge case)
122 credentials = after_scheme[:at_pos]
124 # Check if there's a password (contains :)
125 if ":" not in credentials: 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 return # No password
128 # Extract password (everything after first :)
129 colon_pos = credentials.find(":")
130 password = credentials[colon_pos + 1 :]
132 if not password: 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true
133 return # Empty password
135 # Check for unencoded special characters
136 unencoded_chars = self._find_unencoded_chars(password)
138 if unencoded_chars:
139 self._raise_encoding_error(url, password, unencoded_chars)
141 except CheckpointValidationError:
142 # Re-raise our validation errors
143 raise
144 except Exception as e:
145 # Catch any parsing errors
146 msg = (
147 f"Failed to parse Redis URL: {e!s}\n"
148 f"URL format should be: redis://[user]:[password]@[host]:[port]/[database]\n"
149 f"Ensure password is percent-encoded per RFC 3986."
150 )
151 raise CheckpointValidationError(msg)
153 def _find_unencoded_chars(self, password: str) -> set[str]:
154 """Find unencoded special characters in password.
156 Args:
157 password: Password string to check
159 Returns:
160 Set of unencoded special characters found
161 """
162 unencoded: set[str] = set()
164 for char in self.RESERVED_CHARS:
165 # Skip : as it's used as delimiter
166 if char == ":":
167 continue
169 if char in password:
170 # Check if it's part of a percent-encoded sequence
171 # If we find the literal character, it's unencoded
172 if not self._is_part_of_encoding(password, char): 172 ↛ 164line 172 didn't jump to line 164 because the condition on line 172 was always true
173 unencoded.add(char)
175 return unencoded
177 def _is_part_of_encoding(self, password: str, char: str) -> bool:
178 """Check if character is part of a %XX encoding sequence.
180 Args:
181 password: Password string
182 char: Character to check
184 Returns:
185 True if character is already properly encoded
186 """
187 # Simple heuristic: if we can decode the password without errors
188 # and it's different from the original, it was encoded
189 try:
190 decoded = unquote(password)
191 # If password contains %, assume it might be encoded
192 if "%" in password: 192 ↛ 194line 192 didn't jump to line 194 because the condition on line 192 was never true
193 # If decoding changed it, it was encoded
194 return decoded != password
195 except Exception:
196 pass
198 return False
200 def _raise_encoding_error(
201 self,
202 url: str,
203 password: str,
204 unencoded_chars: set[str],
205 ) -> None:
206 """Raise detailed error about unencoded characters.
208 Args:
209 url: Full Redis URL
210 password: Unencoded password
211 unencoded_chars: Set of problematic characters
213 Raises:
214 CheckpointValidationError: With detailed fix instructions
215 """
216 # Sanitize URL for error message (hide password)
217 sanitized_url = url.replace(password, "****")
219 # Create encoded example
220 from urllib.parse import quote
222 encoded_password = quote(password, safe="")
223 encoded_url = url.replace(password, encoded_password)
225 # Format error message with fix instructions
226 chars_list = ", ".join(f"'{c}'" for c in sorted(unencoded_chars))
228 error_msg = (
229 f"Redis URL contains unencoded special characters in password: {chars_list}\n\n"
230 f"**PRODUCTION INCIDENT REFERENCE**: This is the same error that caused\n"
231 f"staging-758b8f744 to crash with 'ValueError: Port could not be cast to integer'.\n\n"
232 f"Current URL (sanitized): {sanitized_url}\n\n"
233 f"**FIX**: Percent-encode the password per RFC 3986:\n\n"
234 f"Encoded URL example:\n"
235 f" {encoded_url}\n\n"
236 f"Character encoding reference:\n"
237 f" / → %2F\n"
238 f" + → %2B\n"
239 f" = → %3D\n"
240 f" @ → %40\n"
241 f" (space) → %20\n\n"
242 f"For Kubernetes External Secrets, use:\n"
243 f' redis-url: "redis://:{{{{ .redisPassword | urlquery }}}}@host:port/db"\n\n'
244 f"For local development, use:\n"
245 f" CHECKPOINT_REDIS_URL={encoded_url}\n"
246 )
248 raise CheckpointValidationError(error_msg)
250 def _test_redis_connection(self, url: str) -> None:
251 """Optionally test actual Redis connection.
253 Args:
254 url: Redis URL to test
256 Raises:
257 CheckpointValidationError: If connection fails
258 """
259 try:
260 import redis.asyncio as redis_async
262 logger.info(
263 "Testing Redis connection for checkpoint configuration",
264 extra={"redis_url": url},
265 )
267 # Attempt to create client (will fail if URL is malformed)
268 # Note: This doesn't actually connect, just validates URL parsing
269 _ = redis_async.from_url(url) # type: ignore[no-untyped-call]
271 # Close the client (cleanup)
272 # In production, actual connection test would be async
274 except ImportError as e:
275 msg = f"Redis client library not available: {e}\nInstall: pip install redis"
276 raise CheckpointValidationError(msg)
277 except ConnectionError as e:
278 msg = f"Failed to connect to Redis: {e}\nCheck that Redis is running and accessible at the specified URL."
279 raise CheckpointValidationError(msg)
280 except Exception as e:
281 msg = f"Redis connection test failed: {e}\nVerify the Redis URL is correct and Redis is accessible."
282 raise CheckpointValidationError(msg)
285def validate_checkpoint_config(settings: Any) -> None:
286 """Validate checkpoint configuration at application startup.
288 This function should be called during application initialization to
289 catch configuration errors early with clear error messages.
291 Args:
292 settings: Application settings object with checkpoint configuration
294 Raises:
295 CheckpointValidationError: If configuration is invalid
297 Example:
298 >>> from mcp_server_langgraph.core.config import settings
299 >>> validate_checkpoint_config(settings) # At startup
300 """
301 backend = settings.checkpoint_backend.lower()
303 if backend == "memory":
304 # Memory backend doesn't require validation
305 logger.info("Checkpoint backend: memory (no validation required)")
306 return
308 if backend == "redis": 308 ↛ 327line 308 didn't jump to line 327 because the condition on line 308 was always true
309 logger.info("Validating Redis checkpoint configuration...")
311 validator = CheckpointConfigValidator()
313 # Validate Redis URL
314 validator.validate_redis_url(
315 url=settings.checkpoint_redis_url,
316 test_connection=False, # Don't test connection at startup
317 )
319 logger.info(
320 "Redis checkpoint configuration validated successfully",
321 extra={
322 "redis_url": settings.checkpoint_redis_url,
323 "ttl_seconds": settings.checkpoint_redis_ttl,
324 },
325 )
326 else:
327 logger.warning(
328 f"Unknown checkpoint backend: {backend}. Validation skipped.",
329 extra={"backend": backend},
330 )