fix(security): healthz/readyz split, inbound framing, trace authority

README.md

@@ -39,7 +39,7 @@ HTTP serving is hardened for local lab use:
 - `/admin`, `/admin/state`, `/api/v1/*`, and `/v1/chat/completions` require a Bearer token. Use `--admin-token` and `--inference-token` to separate operator and runtime access, or `--auth-token` / `CONTEXTUAL_ORCHESTRATOR_TOKEN` for one local-development token.
 - Binding to `0.0.0.0` or `::` requires `--allow-public-bind`.
 - JSON request bodies, chat message roles, orchestration modes, body sizes, request rate, and concurrent run counts are validated before orchestration runs.
-- Full orchestration traces are not returned by default. Set `include_orchestration_trace: true` per chat request or start with `--expose-trace-by-default` when the caller is trusted.
+- Full orchestration traces are not returned by default. Set `include_orchestration_trace: true` and present a separately minted HMAC trace credential bound to the request tenant and exact resource; `--expose-trace-by-default` alone never grants trace authority.
 - State is in-memory by default. Pass `--state-db PATH` (or `CONTEXTUAL_ORCHESTRATOR_STATE_DB`) to persist workflow runs, evaluation runs, audit, and analytics to a stdlib sqlite file so they survive a restart; without it, behavior is unchanged.
 - Response caching is off by default. Pass `--cache-ttl SECONDS` to serve identical requests (same messages + mode) from an in-memory TTL+LRU cache and skip the provider calls; `0` disables it.
 - `ModelClient.batch_chat(agent, {custom_id: messages})` runs many requests through the provider's Batch API (async, 24h completion window, typically ~50% cheaper) — suited to evaluation/benchmark workloads, not latency-sensitive chat. The mock path answers synchronously.
@@ -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: it probes both batch backends and the usage ledger within a bounded deadline, returns `503` with `status=degraded` when a required dependency fails, and includes only secret-free inventory. Trace responses require a separate HMAC trace credential bound to tenant, exact resource, purpose `orchestration_trace`, expiry, and revocation; admin or inference bearer access alone is insufficient.
 - **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/__main__.py

@@ -82,8 +82,9 @@ def main() -> None:
                         help="Optional sqlite path so runtime agent-pool changes (add/patch/remove) survive restarts.")
     parser.add_argument("--provider-ca-bundle", default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_PROVIDER_CA_BUNDLE") or None,
                         help="Path to a CA bundle used to verify provider TLS (e.g. a corporate gateway root).")
-    parser.add_argument("--insecure-skip-tls-verify", action="store_true",
-                        help="Dev only: do not verify provider TLS certificates (insecure).")
+    parser.add_argument("--trace-authority-secret",
+                        default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_TRACE_AUTHORITY_SECRET", ""),
+                        help="HMAC secret for separately authorized trace credentials.")
     parser.add_argument("--budget-max-output-tokens", type=int, default=None,
                         help="Refuse new runs once estimated/reported output tokens reach this cap (default: no cap).")
     parser.add_argument("--budget-max-cost-usd", type=float, default=None,
@@ -94,7 +95,7 @@ def main() -> None:
                         help="Measure orchestration vs a single-worker baseline on these prompts and print the report.")
     args = parser.parse_args()
 
-    client = ModelClient(ca_bundle=args.provider_ca_bundle, verify_tls=not args.insecure_skip_tls_verify)
+    client = ModelClient(ca_bundle=args.provider_ca_bundle)
     orchestrator = TaskOrchestrator(
         load_agents(args.agents),
         client=client,
@@ -129,6 +130,7 @@ def main() -> None:
                 inference_token=args.inference_token,
                 allow_public_bind=args.allow_public_bind,
                 expose_trace_by_default=args.expose_trace_by_default,
+                trace_authority_secret=args.trace_authority_secret,
             ),
             clearfolio_url=args.clearfolio_url,
         )

contextual_orchestrator/batch_routing.py

@@ -200,6 +200,10 @@ class BatchBackend(Protocol):
 
     name: str
 
+    def readiness_check(self) -> Dict[str, Any]:
+        """Return a bounded, secret-free readiness result for the backend."""
+        ...
+
     def submit(self, requests: List[BatchRequest], metadata: Optional[Dict[str, Any]] = None) -> BatchJob:
         """Submit a batch of requests and return a job handle."""
         ...
@@ -229,6 +233,10 @@ def __init__(self, runner: Callable[[List[Dict[str, str]], str], Dict[str, Any]]
         self._runner = runner
         self._results: Dict[str, List[BatchResultItem]] = {}
 
+    def readiness_check(self) -> Dict[str, Any]:
+        """Report that the in-process batch backend is available."""
+        return {"ready": callable(self._runner), "backend": self.name}
+
     def submit(self, requests: List[BatchRequest], metadata: Optional[Dict[str, Any]] = None) -> BatchJob:
         """Run every request in-process and stash the results under a job id."""
         job_id = f"localbatch_{uuid.uuid4().hex}"
@@ -285,6 +293,10 @@ def __init__(
         self._assembler = payload_assembler
         self._jobs: Dict[str, Dict[str, Any]] = {}
 
+    def readiness_check(self) -> Dict[str, Any]:
+        """Report client configuration without performing external mutation."""
+        return {"ready": self._client is not None, "backend": self.name}
+
     def _assemble_payload(self, requests: List[BatchRequest]) -> str:
         if self._assembler is not None:
             return self._assembler.assemble(
@@ -434,6 +446,10 @@ class EmbeddingBatchBackend(Protocol):
 
     name: str
 
+    def readiness_check(self) -> Dict[str, Any]:
+        """Return a bounded, secret-free readiness result for the backend."""
+        ...
+
     def submit(
         self, requests: List[EmbeddingBatchRequest], metadata: Optional[Dict[str, Any]] = None
     ) -> BatchJob:
@@ -490,6 +506,10 @@ def __init__(
         self._token_counter = token_counter
         self._results: Dict[str, List[EmbeddingBatchResultItem]] = {}
 
+    def readiness_check(self) -> Dict[str, Any]:
+        """Report that the in-process embeddings backend is available."""
+        return {"ready": callable(self._embedder), "backend": self.name}
+
     def _count_tokens(self, text: str, model: str) -> int:
         if self._token_counter is not None:
             return int(self._token_counter.count_text(text, model))
@@ -549,6 +569,10 @@ def __init__(
         self._assembler = payload_assembler
         self._jobs: Dict[str, Dict[str, Any]] = {}
 
+    def readiness_check(self) -> Dict[str, Any]:
+        """Report client configuration without performing external mutation."""
+        return {"ready": self._client is not None, "backend": self.name}
+
     def _assemble_payload(self, requests: List[EmbeddingBatchRequest]) -> str:
         if self._assembler is not None:
             return self._assembler.assemble(

contextual_orchestrator/cost_ledger.py

@@ -583,7 +583,7 @@ 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,),
             )
@@ -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()]
 
 
@@ -737,6 +737,12 @@ def flush(self, timeout: Optional[float] = None) -> bool:
             return bool(flush(timeout=timeout))
         return True
 
+    def readiness_check(self) -> Dict[str, Any]:
+        """Return prompt-safe ledger readiness and storage failure evidence."""
+        health = self.telemetry_health()
+        failures = int(health.get("store_failures", 0))
+        return {"ready": failures == 0, "store_failures": failures}
+
     def telemetry_health(self) -> Dict[str, Any]:
         """Return prompt-safe ledger export health counters."""
         health = self._inline_health.as_dict()

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.
+            raise ValueError("provider TLS verification cannot be disabled")
         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,