security: close three pre-pen-test findings (S-2, S-3, S-1 hardening)

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

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

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,

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]:

tests/unit/test_egress_guard.py

@@ -26,13 +26,32 @@
         "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:
     with pytest.raises(UnsafeEgressURLError):
         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",
     [

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'"