fix(security): healthz/readyz, request framing, trace authority on main

CHANGELOG.md

@@ -0,0 +1,10 @@
+# Changelog
+
+## Unreleased
+
+### Security
+- Minimal unauthenticated `/healthz` liveness; admin `/readyz` inventory (issue #118).
+- Fail-closed Content-Length framing for JSON bodies (issue #119).
+- Orchestration traces require authority beyond inference scope (issue #117).
+
+

README.md

@@ -192,7 +192,7 @@ is read from a **KV config store**, never `os.getenv`.
   backend (local in-process backend standalone), and records one usage-ledger row
   per original vector with the full attribution dimensions (service, team,
   group, company, provider) carried in `metadata`.
-- **Health.** `GET /healthz` is an unauthenticated liveness probe.
+- **Health.** `GET /healthz` is an unauthenticated minimal liveness probe (`status` + `service` only). `GET /readyz` is admin-authenticated readiness with agent/backend/usage inventory for operators.
 - **Standalone + optional pg-llm-batch integration.** The hub runs standalone
   with the in-memory config store and local batch backend; wiring a Postgres DSN
   and an installed/deployed `pg_llm_batch` client activates the KV/secret stores,

contextual_orchestrator/cost_ledger.py

@@ -583,12 +583,12 @@ def _seed_dimension_catalog(self) -> None:
         ph = self._placeholder()
         cur = self._conn.cursor()
         for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG):
-            cur.execute(
+            cur.execute(  # nosemgrep -- sqlalchemy-execute-raw-query FP: only the DB-API placeholder char is interpolated; the value is bound.
                 f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}",  # nosec B608 - ph is a DB-API placeholder.
                 (name,),
             )
             if cur.fetchone() is None:
-                cur.execute(
+                cur.execute(  # nosemgrep -- sqlalchemy-execute-raw-query FP: only DB-API placeholder chars are interpolated; values are bound.
                     "INSERT INTO cost_attribution_dimensions "
                     f"(dimension_name, dimension_label, dimension_order) VALUES ({ph}, {ph}, {ph})",  # nosec B608 - ph is a DB-API placeholder.
                     (name, label, order),
@@ -602,7 +602,7 @@ def append(self, record: UsageRecord) -> None:
         placeholders = ", ".join(ph for _ in _USAGE_COLUMNS)
         columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(
+        cur.execute(  # nosemgrep -- sqlalchemy-execute-raw-query FP: columns are the fixed _USAGE_COLUMNS constant; values are bound.
             f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})",  # nosec B608 - columns are fixed _USAGE_COLUMNS.
             tuple(row.get(column) for column in _USAGE_COLUMNS),
         )
@@ -622,7 +622,7 @@ def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[
         where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
         columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params))  # nosec B608 - columns and clauses are fixed.
+        cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params))  # nosec B608 - columns and clauses are fixed.  # nosemgrep -- sqlalchemy-execute-raw-query FP: fixed columns and clause templates; all values are bound.
         return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()]
 
 

contextual_orchestrator/orchestrator.py

@@ -230,7 +230,7 @@ def __init__(
     @staticmethod
     def _build_ssl_context(ca_bundle: str | None, verify_tls: bool) -> ssl.SSLContext:
         if not verify_tls:
-            return ssl._create_unverified_context()  # nosec B323 - explicit dev-only provider TLS opt-out.
+            return ssl._create_unverified_context()  # nosec B323 - explicit dev-only provider TLS opt-out.  # nosemgrep -- unverified-ssl-context: intentional, default-secure (verify_tls defaults True) dev-only opt-out for self-signed endpoints.
         if ca_bundle:
             if not os.path.isfile(ca_bundle):
                 raise ValueError(f"provider CA bundle does not exist: {ca_bundle}")
@@ -307,7 +307,7 @@ def _send(self, agent: ModelAgent, payload: dict[str, Any]) -> str:
 
     def _open_provider(self, request: urllib.request.Request) -> Any:
         """Open a provider request built from a validated provider URL."""
-        return urllib.request.urlopen(  # nosec B310 - request URL comes from _provider_url after provider validation.
+        return urllib.request.urlopen(  # nosec B310 - request URL comes from _provider_url after provider validation.  # nosemgrep -- dynamic-urllib-use: URL is built by _provider_url after scheme/host validation; egress to loopback/private/reserved is blocked.
             request,
             timeout=self.timeout,
             context=self._ssl_context,

contextual_orchestrator/server.py

@@ -116,6 +116,23 @@ def authorize(self, headers: Any, scope: str, client_address: str) -> None:
         if not expected or not secrets.compare_digest(token, expected):
             raise RequestError(401, "unauthorized", "bearer token is invalid for this scope")
 
+    def may_disclose_trace(self, headers: Any, scope: str) -> bool:
+        """Return True when the verified caller may receive orchestration traces.
+
+        Inference-only credentials never receive planner/workflow evidence.
+        Admin-scope callers may. Single-token deployments may when the host sets
+        ``expose_trace_by_default`` or the request is handled under admin scope.
+        """
+        if scope == "admin":
+            return True
+        if scope != "inference":
+            return False
+        # Split-token inference path: never disclose traces.
+        if self.admin_token and self.inference_token and not self.auth_token:
+            return False
+        # Single shared token: host policy gate only (still needs request bool).
+        return bool(self.expose_trace_by_default)
+
     def check_rate_limit(self, key: str) -> None:
         """Apply a simple per-client fixed-window request budget."""
         now = time.monotonic()
@@ -171,6 +188,69 @@ def _coerce_json(payload: bytes) -> dict[str, Any]:
     return value
 
 
+def _coerce_optional_bool(value: Any, field_name: str) -> bool | None:
+    """Parse an optional JSON bool fail-closed; reject truthy strings/numbers."""
+    if value is None:
+        return None
+    if isinstance(value, bool):
+        return value
+    raise RequestError(
+        400,
+        "invalid_boolean",
+        f"{field_name} must be a JSON boolean",
+        {"field": field_name},
+    )
+
+
+def _parse_content_length(headers: Any, max_body_bytes: int) -> int:
+    """Return a validated Content-Length for fixed-length JSON request bodies.
+
+    Rejects missing, negative, non-decimal, overflow, and transfer-coded framing
+    so ``rfile.read`` never receives a negative size or unbounded read.
+    """
+    # BaseHTTPRequestHandler collapses duplicate headers with commas; treat that
+    # as ambiguous framing and reject rather than guessing.
+    raw = headers.get("content-length")
+    if raw is None or raw == "":
+        raise RequestError(411, "length_required", "Content-Length is required for JSON request bodies")
+    if isinstance(raw, str) and "," in raw:
+        raise RequestError(400, "invalid_content_length", "duplicate or ambiguous Content-Length")
+    text = str(raw).strip()
+    if not text.isdigit():
+        # Reject signed forms ("-1", "+10"), hex, and non-decimal tokens.
+        raise RequestError(400, "invalid_content_length", "Content-Length must be an unsigned decimal integer")
+    # Leading zeros are allowed by isdigit; value itself must fit max.
+    body_size = int(text)
+    if body_size > max_body_bytes:
+        raise RequestError(413, "request_too_large", "request body exceeds configured limit")
+    transfer = (headers.get("transfer-encoding") or "").strip()
+    if transfer and transfer.lower() != "identity":
+        raise RequestError(
+            400,
+            "unsupported_transfer_encoding",
+            "chunked or non-identity Transfer-Encoding is not accepted for JSON bodies",
+        )
+    return body_size
+
+
+def _resolve_include_trace(
+    body: dict[str, Any],
+    security: SecurityConfig,
+    headers: Any,
+    scope: str,
+) -> bool:
+    """Fail-closed orchestration-trace disclosure decision.
+
+    Requires (1) verified trace authority for the caller scope and (2) an
+    explicit JSON boolean request flag when the host default is off.
+    """
+    if not security.may_disclose_trace(headers, scope):
+        return False
+    requested = _coerce_optional_bool(body.get("include_orchestration_trace"), "include_orchestration_trace")
+    if requested is None:
+        return bool(security.expose_trace_by_default)
+    return requested
+
 def _reject_unknown_keys(body: dict[str, Any], allowed: set[str]) -> None:
     unknown = sorted(set(body) - allowed)
     if unknown:
@@ -336,9 +416,14 @@ def do_GET(self) -> None:  # noqa: N802
                     self._send(OPENAPI_SPEC)
                     return
                 if path == "/healthz":
-                    # Unauthenticated liveness probe for containers/orchestrators.
+                    # Unauthenticated liveness only: process is up. No inventory.
+                    self._send({"status": "ok", "service": "contextual-orchestrator"})
+                    return
+                if path == "/readyz":
+                    # Authenticated readiness/diagnostics: operator inventory.
+                    self._authorize("admin")
                     self._send({
-                        "status": "ok",
+                        "status": "ready",
                         "service": "contextual-orchestrator",
                         "agent_count": len(orchestrator.agents),
                         "batch_backend": coordinator.batch_backend.name,
@@ -733,7 +818,7 @@ def do_POST(self) -> None:  # noqa: N802
                         return
                     messages = _validate_messages(body.get("messages"))
                     mode = _validate_mode(body.get("orchestration") or body.get("orchestration_mode") or body.get("mode") or "auto")
-                    include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default))
+                    include_trace = _resolve_include_trace(body, security, self.headers, scope)
                     stream = body.get("stream", False)
                     if not isinstance(stream, bool):
                         raise RequestError(400, "invalid_request", "stream must be a boolean")
@@ -856,7 +941,7 @@ def do_POST(self) -> None:  # noqa: N802
                     except KeyError:
                         self._send_error(404, "batch_job_not_found", f"batch job {job_id} not found")
                         return
-                    self._send(_response_payload(retrieved, include_trace=True))
+                    self._send(_response_payload(retrieved, include_trace=security.may_disclose_trace(self.headers, "inference")))
                     return
                 if path == "/v1/responses":
                     # The Responses API has no chat-completions verifier equivalent,
@@ -884,7 +969,7 @@ def do_POST(self) -> None:  # noqa: N802
                     if not isinstance(prompt, str):
                         raise RequestError(400, "invalid_request", "prompt must be a string")
                     mode = _validate_mode(body.get("mode", "auto"))
-                    include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default))
+                    include_trace = _resolve_include_trace(body, security, self.headers, scope)
                     result = self._run(lambda: orchestrator.run([{"role": "user", "content": prompt}], mode=mode))
                     self._send(_response_payload(result, include_trace))
                     return
@@ -894,7 +979,7 @@ def do_POST(self) -> None:  # noqa: N802
                     if not isinstance(prompt, str) or not prompt:
                         raise RequestError(400, "invalid_request", "prompt_text is required")
                     mode = _validate_mode(body.get("run_mode", "auto"))
-                    include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default))
+                    include_trace = _resolve_include_trace(body, security, self.headers, scope)
                     result = self._run(lambda: orchestrator.run([{"role": "user", "content": prompt}], mode=mode))
                     self._send(_response_payload(result, include_trace), 201)
                     return
@@ -906,7 +991,7 @@ def do_POST(self) -> None:  # noqa: N802
                     if not isinstance(prompts, list) or not prompts:
                         raise RequestError(400, "invalid_request", "prompts must be a non-empty array")
                     mode = _validate_mode(body.get("run_mode", "auto"))
-                    include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default))
+                    include_trace = _resolve_include_trace(body, security, self.headers, scope)
                     evaluation_run = self._run(lambda: orchestrator.run_evaluation([str(item) for item in prompts], mode=mode))
                     self._send(_response_payload(evaluation_run, include_trace), 201)
                     return
@@ -960,10 +1045,10 @@ def _parse_optional_int(self, query: dict[str, list[str]], field_name: str) -> i
         def _read_json(self) -> dict[str, Any]:
             if self.headers.get("content-type", "").split(";", 1)[0].strip().lower() != "application/json":
                 raise RequestError(415, "unsupported_media_type", "content-type must be application/json")
-            body_size = int(self.headers.get("content-length", "0"))
-            if body_size > security.max_body_bytes:
-                raise RequestError(413, "request_too_large", "request body exceeds configured limit")
+            body_size = _parse_content_length(self.headers, security.max_body_bytes)
             raw = self.rfile.read(body_size)
+            if len(raw) != body_size:
+                raise RequestError(400, "incomplete_body", "request body shorter than Content-Length")
             return _coerce_json(raw) if raw else {}
 
         def log_message(self, format: str, *args: object) -> None:

docs/architecture.md

@@ -53,3 +53,21 @@ The product is not a Fugu clone. It is a control-plane prototype for the same pu
 - replayable evaluation runs before any learned coordinator replaces the deterministic policy.
 
 See [product_planning.md](product_planning.md) for the product reboot.
+
+## Health and readiness
+
+Unauthenticated `GET /healthz` returns only process liveness (`status`, `service`).
+Operator inventory (agent counts, batch backends, usage counts) is on
+authenticated `GET /readyz` (admin scope). Inference callers cannot request
+orchestration traces without verified admin/trace authority.
+
+
+## Research citations (APA 7th)
+
+Chen, L., Zaharia, M., & Zou, J. (2023). *FrugalGPT: How to use large language models while reducing cost and improving performance* (arXiv:2305.05176). https://doi.org/10.48550/arXiv.2305.05176
+
+Ong, I., Almahairi, A., Wu, V., Chiang, W.-L., Wu, T., Gonzalez, J. E., Kadous, M. W., & Stoica, I. (2024). *RouteLLM: Learning to route LLMs with preference data* (arXiv:2406.18665). https://doi.org/10.48550/arXiv.2406.18665
+
+Ding, D., Mallick, A., Wang, C., Sim, R., Mukherjee, S., Rühle, V., Lakshmanan, L. V. S., & Awadallah, A. H. (2024). *Hybrid LLM: Cost-efficient and quality-aware query routing* (arXiv:2404.14618). https://doi.org/10.48550/arXiv.2404.14618
+
+National Institute of Standards and Technology. (2022). *Secure software development framework (SSDF) version 1.1* (NIST SP 800-218). https://doi.org/10.6028/NIST.SP.800-218

tests/test_cost_review_server.py

@@ -56,9 +56,7 @@ def test_healthz_is_unauthenticated_and_ok() -> None:
     finally:
         server.shutdown()
     assert status == 200
-    assert body["status"] == "ok"
-    assert body["service"] == "contextual-orchestrator"
-    assert "batch_backend" in body
+    assert body == {"status": "ok", "service": "contextual-orchestrator"}
 
 
 def test_chat_completion_reports_real_usage_and_records_cost() -> None:

tests/test_healthz.py

@@ -1,9 +1,10 @@
-"""Container liveness probe: /healthz must answer without any auth token."""
+"""Container liveness vs readiness probe contracts."""
 from __future__ import annotations
 
 import json
 import sys
 import threading
+import urllib.error
 import urllib.request
 from pathlib import Path
 
@@ -14,18 +15,24 @@
 from contextual_orchestrator import ModelAgent, TaskOrchestrator  # noqa: E402
 from contextual_orchestrator.server import SecurityConfig, build_server  # noqa: E402
 
+_TEST_ADMIN = "admin_secret"  # noqa: S105
+_TEST_INFERENCE = "inference_secret"  # noqa: S105
 
-def test_healthz_is_unauthenticated_liveness() -> None:
+
+def _start():
     orchestrator = TaskOrchestrator([ModelAgent("probe_agent", "mock-agent", tags=("reasoning",))])
     server = build_server(
         orchestrator,
         port=0,
-        security=SecurityConfig(admin_token="admin_secret", inference_token="inference_secret"),
+        security=SecurityConfig(admin_token=_TEST_ADMIN, inference_token=_TEST_INFERENCE),
     )
     thread = threading.Thread(target=server.serve_forever, daemon=True)
     thread.start()
-    port = server.server_address[1]
+    return server, thread, server.server_address[1]
+
 
+def test_healthz_is_unauthenticated_minimal_liveness() -> None:
+    server, thread, port = _start()
     try:
         with urllib.request.urlopen(f"http://127.0.0.1:{port}/healthz", timeout=5) as response:
             status = response.status
@@ -35,7 +42,52 @@ def test_healthz_is_unauthenticated_liveness() -> None:
         thread.join(timeout=5)
 
     assert status == 200
-    assert body["status"] == "ok"
+    assert body == {"status": "ok", "service": "contextual-orchestrator"}
+    # Operational inventory must not leak on unauthenticated liveness.
+    for forbidden in (
+        "agent_count",
+        "batch_backend",
+        "embedding_batch_backend",
+        "usage_record_count",
+        "agents",
+        "ready",
+    ):
+        assert forbidden not in body
+
+
+def test_readyz_requires_admin_and_exposes_inventory() -> None:
+    server, thread, port = _start()
+    try:
+        unauth = urllib.request.Request(f"http://127.0.0.1:{port}/readyz")
+        try:
+            urllib.request.urlopen(unauth, timeout=5)
+            raise AssertionError("readyz must require auth")
+        except urllib.error.HTTPError as exc:
+            assert exc.code == 401
+
+        inference = urllib.request.Request(
+            f"http://127.0.0.1:{port}/readyz",
+            headers={"authorization": f"Bearer {_TEST_INFERENCE}"},
+        )
+        try:
+            urllib.request.urlopen(inference, timeout=5)
+            raise AssertionError("readyz must not accept inference token")
+        except urllib.error.HTTPError as exc:
+            assert exc.code == 401
+
+        admin = urllib.request.Request(
+            f"http://127.0.0.1:{port}/readyz",
+            headers={"authorization": f"Bearer {_TEST_ADMIN}"},
+        )
+        with urllib.request.urlopen(admin, timeout=5) as response:
+            status = response.status
+            body = json.loads(response.read().decode("utf-8"))
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+    assert status == 200
+    assert body["status"] == "ready"
     assert body["service"] == "contextual-orchestrator"
     assert body["agent_count"] == 1
     assert body["batch_backend"]
@@ -44,5 +96,6 @@ def test_healthz_is_unauthenticated_liveness() -> None:
 
 
 if __name__ == "__main__":
-    test_healthz_is_unauthenticated_liveness()
+    test_healthz_is_unauthenticated_minimal_liveness()
+    test_readyz_requires_admin_and_exposes_inventory()
     print("ok")