diff --git a/src/serving/api/egress_guard.py b/src/serving/api/egress_guard.py index 680c635..3aae0e4 100644 --- a/src/serving/api/egress_guard.py +++ b/src/serving/api/egress_guard.py @@ -45,6 +45,18 @@ class UnsafeEgressURLError(ValueError): def _ip_is_public(ip: str) -> bool: addr = ipaddress.ip_address(ip) + # An IPv4-mapped IPv6 address (``::ffff:169.254.169.254``) embeds an IPv4 + # target the connect will actually reach. CPython's ``is_private`` / + # ``is_link_local`` do reflect the embedded v4 on 3.11 for the ranges that + # matter here, but that is version-dependent (tightened in 3.13); unwrap to + # the embedded IPv4 and judge THAT, so the guard does not hinge on the stdlib + # version of the prod image. Same for 6to4 (``2002::/16``) which carries a v4 + # in the next 32 bits. (pre-pen-test audit, S-1 feeder) + if isinstance(addr, ipaddress.IPv6Address): + if addr.ipv4_mapped is not None: + return _ip_is_public(str(addr.ipv4_mapped)) + if addr.sixtofour is not None: + return _ip_is_public(str(addr.sixtofour)) return not ( addr.is_private or addr.is_loopback diff --git a/src/serving/api/main.py b/src/serving/api/main.py index e58597d..3a2e217 100644 --- a/src/serving/api/main.py +++ b/src/serving/api/main.py @@ -658,9 +658,15 @@ async def health_ready(request: Request) -> Response: """ checks = await run_in_threadpool(_readiness_checks, request.app.state) ready = all(check["status"] == "ok" for check in checks) + # This probe is auth-exempt (k8s/Compose carry no API key). The internal + # check dicts carry `backend` (engine identity) and `detail` (str(exc), which + # can embed internal hostnames/DSNs); expose only name+status so an + # unauthenticated caller cannot fingerprint the topology (pre-pen-test audit, + # S-3). The failing detail is still logged server-side by the check itself. + public_checks = [{"name": c["name"], "status": c["status"]} for c in checks] return JSONResponse( status_code=200 if ready else 503, - content={"status": "ready" if ready else "not_ready", "checks": checks}, + content={"status": "ready" if ready else "not_ready", "checks": public_checks}, ) @@ -709,4 +715,50 @@ async def health(request: Request) -> dict: request.app.state.health_cache_expires_at = ( now + request.app.state.health_cache_ttl_seconds ) - return cast(dict, health_payload) + return _public_health_view(cast(dict, health_payload)) + + +# Metric keys safe to expose on the auth-exempt /v1/health. Fail-closed +# allowlist: only this process's own resource gauges and data-freshness, never +# topology recon (backend identity, broker/topic counts, Flink job counts). A +# metric added to a component in the future does NOT leak until listed here +# (pre-pen-test audit, S-3). +_PUBLIC_HEALTH_METRICS = frozenset( + { + "pool_size", + "read_in_use", + "read_available", + "read_utilization", + "write_in_use", + "last_event_age_seconds", + } +) + + +def _public_health_view(payload: dict) -> dict: + """Strip a component-health payload to what is safe to serve unauthenticated. + + `/v1/health` is auth-exempt, yet each component's `message`/`metrics` carry + recon: `f"Kafka unavailable: {exc}"` (internal hostnames/URLs), broker/topic + counts, cluster sizes, and the serving backend's identity. Keep the overall + status, each component's name/status/source (the documented contract), and + only the allowlisted operational gauges (pool utilization, data freshness); + drop `message` and every non-allowlisted metric. Agents still get the + per-component status they use to caveat freshness, and pool-utilization + observability (its only exposure — not on /metrics) is preserved. + """ + components = [] + for c in payload.get("components", []): + metrics = {k: v for k, v in (c.get("metrics") or {}).items() if k in _PUBLIC_HEALTH_METRICS} + components.append( + { + "name": c.get("name"), + "status": c.get("status"), + "source": c.get("source"), + "metrics": metrics, + } + ) + view = {"status": payload.get("status"), "components": components} + if "checked_at" in payload: + view["checked_at"] = payload["checked_at"] + return view diff --git a/src/serving/api/routers/agent_query.py b/src/serving/api/routers/agent_query.py index 48fce5f..36c704d 100644 --- a/src/serving/api/routers/agent_query.py +++ b/src/serving/api/routers/agent_query.py @@ -23,6 +23,7 @@ get_version_registry, resolve_request_version, ) +from src.serving.backends import BackendExecutionError, BackendMissingTableError from src.serving.cache import ENTITY_TTL_SECONDS, QueryCache, cache_entity_key from src.serving.control_plane import get_control_plane_store from src.serving.semantic_layer.stage_clock import coerce_dt, resolve_breach, stage_budget @@ -32,6 +33,39 @@ router = APIRouter(tags=["agent"]) +def _client_safe_error(exc: ValueError, req: Request, status_code: int) -> HTTPException: + """Map a query-layer ``ValueError`` to an ``HTTPException`` that never leaks + engine internals. + + The query helpers catch a ``BackendExecutionError`` (raw ClickHouse/DuckDB + text — engine type, SQL fragments, table/column names) and re-raise it as + ``ValueError(f"... failed: {backend_error}")`` with ``from e``. Returning that + detail verbatim fingerprints the backend to any authenticated caller + (pre-pen-test audit, S-2). When the cause is a ``BackendExecutionError``, log + the real text server-side under the correlation id and return a generic + detail carrying only that id; a plain ``ValueError`` is request-level + validation (safe and useful to the caller) and is returned verbatim. The + caller's ``status_code`` is preserved in both branches. + """ + cause = exc.__cause__ + if isinstance(cause, BackendMissingTableError): + # A missing serving table is an actionable operator signal (run + # provisioning), not engine internals — but the upstream message embeds + # the scoped-SQL table reference, so return a clean fixed detail that + # keeps the "not materialized" signal without the SQL. `/health/ready` + # carries the full provisioning hint for operators. + return HTTPException( + status_code=status_code, + detail="serving table is not materialized yet — run provisioning", + ) + if isinstance(cause, BackendExecutionError): + correlation_id = getattr(req.state, "correlation_id", None) + logger.error("backend_execution_error", detail=str(exc), correlation_id=correlation_id) + ref = f" (ref {correlation_id})" if correlation_id else "" + return HTTPException(status_code=status_code, detail=f"backend query failed{ref}") + return HTTPException(status_code=status_code, detail=str(exc)) + + # Engine call-signature compatibility (F-4): older engine implementations and # many test fakes predate the ``tenant_id`` / ``allowed_tables`` kwargs. The # helpers below replace three hand-rolled nested try/except TypeError cascades @@ -516,7 +550,7 @@ async def natural_language_query(request: NLQueryRequest, req: Request) -> Query except HTTPException: raise except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) from None + raise _client_safe_error(e, req, status_code=400) from None if result.get("sql") is not None: span.set_attribute("query.sql", result["sql"]) span.set_attribute("query.rows", int(result.get("row_count", 0))) @@ -618,7 +652,7 @@ async def get_entity( optional_kwargs={"tenant_id": tenant_id}, ) except ValueError as e: - raise HTTPException(status_code=503, detail=str(e)) from None + raise _client_safe_error(e, req, status_code=503) from None if result is None: raise HTTPException(status_code=404, detail=f"{entity_type}/{entity_id} not found") @@ -673,7 +707,7 @@ async def get_order_timeline(order_id: str, req: Request) -> OrderTimelineRespon try: payload = await run_in_threadpool(_build_order_timeline, req, order_id) except ValueError as e: - raise HTTPException(status_code=503, detail=str(e)) from None + raise _client_safe_error(e, req, status_code=503) from None if payload is None: raise HTTPException(status_code=404, detail=f"order/{order_id} not found") @@ -766,7 +800,7 @@ async def get_metric( as_of=as_of, ) except ValueError as e: - raise HTTPException(status_code=503, detail=str(e)) from None + raise _client_safe_error(e, req, status_code=503) from None metric_payload = { "metric_name": metric_name, diff --git a/src/serving/api/routers/batch.py b/src/serving/api/routers/batch.py index 2a19c87..0e43e69 100644 --- a/src/serving/api/routers/batch.py +++ b/src/serving/api/routers/batch.py @@ -3,6 +3,7 @@ from copy import copy from typing import Any, Literal +import structlog from fastapi import APIRouter, Request from pydantic import BaseModel, Field @@ -11,10 +12,35 @@ _call_in_threadpool_with_kwarg_fallback, _ensure_metric_allowed, ) +from src.serving.backends import BackendExecutionError, BackendMissingTableError +logger = structlog.get_logger() router = APIRouter(tags=["agent"]) +def _safe_item_error(exc: Exception, req: Request) -> str: + """Return a client-safe per-item error string. + + A batch item runs the same entity/metric/NL path as the single-item routes, + so its failures carry the same raw engine text — directly as a + ``BackendExecutionError`` or wrapped as ``ValueError(f"... failed: {e}")``. + Genericise those (log the real text server-side under the correlation id) and + keep plain validation messages verbatim (pre-pen-test audit, S-2). + """ + if isinstance(exc, BackendMissingTableError) or isinstance( + exc.__cause__, BackendMissingTableError + ): + # Actionable provisioning signal, kept clean of the scoped-SQL reference + # (mirrors _client_safe_error). + return "serving table is not materialized yet — run provisioning" + if isinstance(exc, BackendExecutionError) or isinstance(exc.__cause__, BackendExecutionError): + correlation_id = getattr(req.state, "correlation_id", None) + logger.error("batch_backend_error", detail=str(exc), correlation_id=correlation_id) + ref = f" (ref {correlation_id})" if correlation_id else "" + return f"backend query failed{ref}" + return str(exc) + + class BatchItem(BaseModel): id: str type: Literal["entity", "metric", "query"] @@ -57,7 +83,7 @@ async def _execute_item(item: BatchItem, req: Request) -> BatchResult: data = await _execute_query_item(item, req) return BatchResult(id=item.id, status="ok", data=data) except Exception as exc: - return BatchResult(id=item.id, status="error", error=str(exc)) + return BatchResult(id=item.id, status="error", error=_safe_item_error(exc, req)) async def _execute_entity_item(item: BatchItem, req: Request) -> dict[str, Any]: diff --git a/tests/unit/test_egress_guard.py b/tests/unit/test_egress_guard.py index 9aa1b2c..3d397ef 100644 --- a/tests/unit/test_egress_guard.py +++ b/tests/unit/test_egress_guard.py @@ -26,6 +26,13 @@ "http://0.0.0.0/", # unspecified "https://[fd00::1]/", # IPv6 unique-local "http://[fe80::1]/", # IPv6 link-local + # IPv4-mapped IPv6 embedding an internal v4 target — the connect reaches + # the embedded IPv4, so the guard must judge that, not the wrapper + # (pre-pen-test audit, S-1 feeder; robust across CPython versions). + "http://[::ffff:169.254.169.254]/latest/meta-data/", # mapped metadata + "http://[::ffff:127.0.0.1]/", # mapped loopback + "http://[::ffff:10.0.0.1]/", # mapped private + "http://[2002:a9fe:a9fe::1]/", # 6to4 embedding 169.254.169.254 ], ) def test_validate_rejects_non_public_addresses(url: str) -> None: @@ -33,6 +40,18 @@ def test_validate_rejects_non_public_addresses(url: str) -> None: validate_public_url(url) +def test_mapped_public_ipv4_is_allowed(monkeypatch: pytest.MonkeyPatch) -> None: + # The unwrap must not over-reject: a mapped *public* IPv4 stays public. + monkeypatch.setattr( + socket, + "getaddrinfo", + lambda *a, **k: [ + (socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("::ffff:93.184.216.34", 80, 0, 0)) + ], + ) + validate_public_url("http://mapped-public.example.com/hook") # must not raise + + @pytest.mark.parametrize( "url", [ diff --git a/tests/unit/test_error_sanitization.py b/tests/unit/test_error_sanitization.py new file mode 100644 index 0000000..26bc029 --- /dev/null +++ b/tests/unit/test_error_sanitization.py @@ -0,0 +1,105 @@ +"""Unit coverage for the client-facing error sanitizers (pre-pen-test audit S-2). + +The query helpers wrap a ``BackendExecutionError`` (raw ClickHouse/DuckDB text — +engine type, SQL fragments, table/column names) as ``ValueError("... failed: +{backend_error}")`` and the batch path may surface it directly. The router +boundary must return a generic detail for those while keeping plain +request-validation messages verbatim. These pin that split without the HTTP +stack (the sanitizers are pure functions of the exception + request). +""" + +from __future__ import annotations + +from types import SimpleNamespace + +from src.serving.api.routers.agent_query import _client_safe_error +from src.serving.api.routers.batch import _safe_item_error +from src.serving.backends import BackendExecutionError, BackendMissingTableError + +_RAW = "Code: 60. DB::Exception: Table agentflow.orders_v2 doesn't exist on clickhouse-0" + + +def _req(correlation_id: str | None = "corr-123") -> SimpleNamespace: + return SimpleNamespace(state=SimpleNamespace(correlation_id=correlation_id)) + + +def _reraise_from(cause: BackendExecutionError) -> ValueError: + """A ValueError carrying a BackendExecutionError cause, as the query helpers + produce with ``raise ValueError(...) from e`` — the cause chain is what the + sanitizer keys on.""" + err = ValueError(f"Entity lookup failed: {cause}") + err.__cause__ = cause + return err + + +# ── _client_safe_error (entity/metric/NL routes) ───────────────── + + +def test_backend_wrapped_valueerror_is_genericised() -> None: + exc = _reraise_from(BackendExecutionError(_RAW)) + http = _client_safe_error(exc, _req(), status_code=503) + assert http.status_code == 503 + assert http.detail == "backend query failed (ref corr-123)" + assert "clickhouse" not in http.detail + assert "orders_v2" not in http.detail + + +def test_missing_table_gets_clean_actionable_detail_without_sql() -> None: + # BackendMissingTableError is an operator provisioning signal; keep the + # actionable "not materialized" phrasing but drop the scoped-SQL reference + # the upstream message embeds. + cause = BackendMissingTableError( + "Table '(SELECT * EXCLUDE (tenant_id) FROM users_enriched) AS \"users_enriched\"' " + "for entity 'user' is not materialized yet" + ) + err = ValueError(f"Entity lookup failed: {cause}") + err.__cause__ = cause + http = _client_safe_error(err, _req(), status_code=503) + assert http.status_code == 503 + assert "is not materialized yet" in http.detail + assert "SELECT" not in http.detail + assert "users_enriched" not in http.detail + + +def test_plain_validation_valueerror_is_verbatim() -> None: + # A request-level validation error carries no backend cause — keep it, it + # helps the caller fix the request. + http = _client_safe_error(ValueError("window must be one of 1h, 24h"), _req(), status_code=400) + assert http.status_code == 400 + assert http.detail == "window must be one of 1h, 24h" + + +def test_generic_detail_without_correlation_id_has_no_ref() -> None: + http = _client_safe_error( + _reraise_from(BackendExecutionError(_RAW)), _req(None), status_code=503 + ) + assert http.detail == "backend query failed" + + +# ── _safe_item_error (batch route) ─────────────────────────────── + + +def test_batch_direct_backend_error_is_genericised() -> None: + msg = _safe_item_error(BackendExecutionError(_RAW), _req()) + assert msg == "backend query failed (ref corr-123)" + assert "clickhouse" not in msg + + +def test_batch_wrapped_backend_error_is_genericised() -> None: + msg = _safe_item_error(_reraise_from(BackendExecutionError(_RAW)), _req()) + assert msg == "backend query failed (ref corr-123)" + + +def test_batch_missing_table_gets_clean_actionable_detail() -> None: + msg = _safe_item_error( + BackendMissingTableError("... FROM orders_v2 ... not materialized"), _req() + ) + assert "is not materialized yet" in msg + assert "orders_v2" not in msg + + +def test_batch_plain_validation_error_is_verbatim() -> None: + # Matches the existing batch contract test: "Unknown metric" must reach the + # caller so they can correct the item. + msg = _safe_item_error(ValueError("Unknown metric 'ghost'"), _req()) + assert msg == "Unknown metric 'ghost'" diff --git a/tests/unit/test_health_public_view.py b/tests/unit/test_health_public_view.py new file mode 100644 index 0000000..40c4612 --- /dev/null +++ b/tests/unit/test_health_public_view.py @@ -0,0 +1,99 @@ +"""Unit coverage for the unauthenticated health-view sanitizer (audit S-3). + +`/v1/health` is auth-exempt, so its payload must not carry recon: component +`message` text (``f"Kafka unavailable: {exc}"`` with internal hostnames), +`metrics` (broker/topic counts, cluster sizes), or backend identity. The view +keeps overall status and each component's name/status/source (the documented +contract) and nothing else. +""" + +from __future__ import annotations + +from src.serving.api.main import _public_health_view + +_RAW_PAYLOAD = { + "status": "unhealthy", + "checked_at": "2026-07-19T07:00:00+00:00", + "components": [ + { + "name": "kafka", + "status": "unhealthy", + "message": "Kafka unavailable: [Errno -2] Name or service not known: kafka-broker-0.internal:9092", + "metrics": {"brokers": 0, "topics": 0}, + "source": "placeholder", + }, + { + "name": "serving", + "status": "healthy", + "message": "clickhouse reachable", + "metrics": {"backend": "clickhouse"}, + "source": "live", + }, + { + "name": "freshness", + "status": "healthy", + "message": "last event 4s ago", + "metrics": {"last_event_age_seconds": 4}, + "source": "live", + }, + ], +} + + +def test_view_keeps_only_name_status_source_per_component() -> None: + view = _public_health_view(_RAW_PAYLOAD) + for component in view["components"]: + assert set(component) == {"name", "status", "source", "metrics"} + + +def test_view_drops_error_strings_counts_and_backend_identity() -> None: + # message (internal hostnames), broker/topic counts and backend identity are + # all recon and must be gone. + flat = repr(_public_health_view(_RAW_PAYLOAD)) + for leak in ("kafka-broker-0.internal", "Errno", "clickhouse", "brokers", "topics"): + assert leak not in flat + + +def test_view_keeps_only_allowlisted_operational_metrics() -> None: + view = _public_health_view(_RAW_PAYLOAD) + by_name = {c["name"]: c for c in view["components"]} + # topology metrics dropped ... + assert by_name["kafka"]["metrics"] == {} + assert by_name["serving"]["metrics"] == {} # backend identity dropped + # ... freshness gauge (agent-facing, benign) preserved. + assert by_name["freshness"]["metrics"] == {"last_event_age_seconds": 4} + + +def test_view_preserves_allowlisted_pool_gauges() -> None: + payload = { + "status": "healthy", + "components": [ + { + "name": "duckdb_pool", + "status": "healthy", + "source": "live", + "metrics": {"pool_size": 4, "read_utilization": 0.25, "secret_dsn": "host:5432"}, + } + ], + } + metrics = _public_health_view(payload)["components"][0]["metrics"] + assert metrics == {"pool_size": 4, "read_utilization": 0.25} # non-allowlisted key dropped + + +def test_view_preserves_overall_status_and_component_statuses() -> None: + view = _public_health_view(_RAW_PAYLOAD) + assert view["status"] == "unhealthy" + assert [c["status"] for c in view["components"]] == ["unhealthy", "healthy", "healthy"] + assert [c["name"] for c in view["components"]] == ["kafka", "serving", "freshness"] + + +def test_view_preserves_checked_at_when_present() -> None: + assert _public_health_view(_RAW_PAYLOAD)["checked_at"] == "2026-07-19T07:00:00+00:00" + + +def test_view_tolerates_missing_optional_keys() -> None: + view = _public_health_view({"status": "healthy", "components": [{"name": "x", "status": "ok"}]}) + assert view == { + "status": "healthy", + "components": [{"name": "x", "status": "ok", "source": None, "metrics": {}}], + }