PrefectHQ/fastmcp · error · CIMDValidationError
{str(e) from SSRFError/SSRFFetchError}
Error message
{str(e) from SSRFError/SSRFFetchError} What it means
When fetching a CIMD document, the SSRF-safe HTTP layer (ssrf_safe_fetch_response) can raise SSRFError for requests that violate SSRF protections (private/internal IPs, disallowed schemes, DNS pinning failures, oversized/timeout responses). CIMDFetcher.fetch() converts those into CIMDValidationError so callers get a single CIMD-specific error type.
Source
Thrown at fastmcp_slim/fastmcp/server/auth/cimd.py:363
if cached.etag:
request_headers["If-None-Match"] = cached.etag
if cached.last_modified:
request_headers["If-Modified-Since"] = cached.last_modified
if request_headers:
allowed_status_codes = {200, 304}
try:
response = await ssrf_safe_fetch_response(
client_id_url,
require_path=True,
max_size=self.MAX_RESPONSE_SIZE,
timeout=self.timeout,
overall_timeout=30.0,
request_headers=request_headers,
allowed_status_codes=allowed_status_codes,
)
except SSRFError as e:
raise CIMDValidationError(str(e)) from e
except SSRFFetchError as e:
raise CIMDFetchError(str(e)) from e
if response.status_code == 304:
if cached is None:
raise CIMDFetchError(
"CIMD server returned 304 Not Modified without cached document"
)
now = time.time()
if self._has_freshness_headers(response.headers):
policy = self._parse_cache_policy(response.headers, now)
else:
# RFC allows 304 to omit unchanged headers. Preserve existing
# cache policy rather than resetting to fallback defaults.
policy = _CIMDCachePolicy(
etag=None,
last_modified=None,View on GitHub (pinned to 1f02114297)
Solutions
- Ensure the client_id is a public HTTPS URL reachable from the server without crossing SSRF protections
- If testing locally, run behind an allowed host or adjust SSRF allowlist configuration in ssrf.py
- Catch CIMDValidationError in your client-management flow and return a 400-style invalid_client response
- Inspect the wrapped SSRFError message for the specific violated rule (blocked IP, scheme, size, timeout)
Example fix
try:
doc = await fetcher.get_client(client_id)
except CIMDValidationError as e:
return JSONResponse(status_code=400, content={"error": str(e)}) Defensive patterns
Strategy: try-catch
Validate before calling
from urllib.parse import urlparse
def is_public_https(url: str) -> bool:
p = urlparse(url)
return p.scheme in ("http", "https") and not (
p.hostname in ("localhost",) or p.hostname and (
p.hostname.startswith("10.") or p.hostname.startswith("192.168.")
)
) Type guard
def is_http_url(u: object) -> bool:
if not isinstance(u, str):
return False
p = urlparse(u)
return p.scheme in ("http", "https") and bool(p.netloc) Try / catch
try:
doc = await fetcher.get_client(client_id)
except CIMDValidationError as e:
return JSONResponse(400, {"error": "invalid_client", "detail": str(e)}) Prevention
- Use public HTTPS URLs for client_id metadata
- Don't point metadata at localhost/private IPs in production
- Log the wrapped SSRF reason to distinguish policy blocks from bugs
When it happens
Trigger: CIMDFetcher.fetch(client_id_url) is called (directly, via get_client, or by token flows using the client_id as a URL) and validate_url/ssrf_safe_fetch_response rejects the URL — e.g. client_id points at localhost, a private RFC1918 address, a non-HTTP scheme, or DNS resolves to a blocked IP.
Common situations: A client_id that is not a public HTTPS URL; testing against local CIMD servers; DNS rebinding/internal-network hostnames in multi-tenant deployments; misconfigured metadata pointing at internal services.
Related errors
- CIMD jwks_uri failed SSRF validation: {e}
- str(e)
- Unexpected authorization response: {response.status_code}
- Invalid client assertion: {e}
- CIMD redirect_uri must have a host: {uri!r}
AI-assisted analysis of PrefectHQ/fastmcp@1f02114297 (2026-08-29).
Data as JSON: /api/errors/61b5d3b722f2046b.
Report an issue: GitHub.