fix(security): read provider host allowlist from KV at request time

AGENTS.md

@@ -56,21 +56,24 @@ push or open a PR.
 - The reference implementation is xtrmLLMBatchPython's pgcrypto-encrypted
   Postgres credential registry (`get_credential(name)`); reuse that pattern (a
   DB-backed KV is fine) unless a dedicated KV is adopted.
-- **Known deviation to migrate:** this repo currently resolves provider API
-  keys from env — `ModelClient` reads `os.environ.get(agent.api_key_env)` in
-  `contextual_orchestrator/orchestrator.py` (and `CONTEXTUAL_ORCHESTRATOR_*`
-  tokens in `__main__.py`). Move these to KV-backed reads; keep env only as the
-  bootstrap path that seeds the KV.
+- **Known deviation to migrate:** `__main__.py` still accepts
+  `CONTEXTUAL_ORCHESTRATOR_*` tokens and bind/TLS flags from env as
+  **process bootstrap**. Provider API keys and the provider-host allowlist
+  are KV-backed at request time (`get_credential`,
+  `allowed_provider_hosts`). Do not reintroduce `os.getenv` inside
+  `ModelClient._validate_provider` or `ModelClient.chat`.
 
 ### This repo: the org LLM gateway
 
 - `contextual-orchestrator` is the org's **LLM-communication hub** — the
   OpenAI-compatible front door consumed by **gyeot** and **scopeweave**.
 - **Direction:** grow it toward a **LiteLLM-class multi-provider gateway**. The
   org is open to a **Rust/Python hybrid** to cut overhead.
-- Its `ModelClient` currently reads `os.environ.get(agent.api_key_env)` — this
-  is the KV-principle deviation above. Resolve the API key (including the org
-  `OPENAI_API_KEY`) from the **KV / credential registry**, not env.
+- Provider API keys resolve through `get_credential`. The provider-host
+  allowlist resolves through `allowed_provider_hosts()` on the **process-wide
+  runtime ConfigStore** (KV category `provider_egress`). That store is
+  `InMemoryConfigStore` unless bootstrap installed another backend with
+  `set_runtime_config_store()`. Env remains bootstrap transport only.
 - The **OpenCode review pipeline is separate** and stays on **GitHub Models** —
   do not change it.
 

CHANGELOG.md

@@ -0,0 +1,50 @@
+# Changelog
+
+All notable changes to this project are documented in this file.
+
+The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
+and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [Unreleased]
+
+### Fixed
+
+- Provider host allowlisting (`provider_egress.allowed_provider_hosts`) is
+  read from the **process-wide runtime ConfigStore** at request time, not from
+  `os.getenv` and not from a separately constructed Postgres `com_config`
+  unless that store was installed with `set_runtime_config_store()` at
+  bootstrap. `CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS` is copied into
+  that KV key once at process start (`seed_provider_egress_from_environ`).
+  Changing the env var on a running process no longer changes egress policy.
+  Buyer next action: call `set_runtime_config("provider_egress",
+  "allowed_provider_hosts", "api.example.com")` (or start the process with
+  the env var set so bootstrap can copy it). Do not write the key only into
+  a new `get_config_store(postgres_dsn=...)` instance and expect egress to
+  honor it.
+- Treat official-SDK JSON `null` on optional `tools[].function.description`,
+  `parameters`, and `strict` as omit-real: the keys are popped before
+  `proxy_completion` so upstream providers see an omitted field, not a null
+  schema. Non-null wrong types still fail closed with named `invalid_tools`.
+  Next action: send those fields only when you have a real string, JSON Schema
+  object, or boolean; SDK defaults of `null` are safe.
+- Fail closed on tools passthrough for `seed`, `stop`, `n>1`, `logprobs`,
+  `logit_bias`, and out-of-range penalties — the same named errors as the
+  orchestration path. Next action: omit those knobs on tool-calling requests.
+- Apply the request `temperature` on streamed route completions instead of
+  silently using `0.2`. Next action: send the temperature you want; streaming
+  no longer changes the sampling policy.
+
+### References
+
+- Joint Task Force. (2020). *Security and privacy controls for information
+  systems and organizations* (NIST Special Publication 800-53 Rev. 5).
+  National Institute of Standards and Technology.
+  https://doi.org/10.6028/NIST.SP.800-53r5
+- International Organization for Standardization. (2022). *Information
+  security, cybersecurity and privacy protection — Information security
+  controls* (ISO/IEC 27001:2022). https://www.iso.org/standard/27001
+- OpenAI. (2024). *Create chat completion*. OpenAI API reference.
+  https://platform.openai.com/docs/api-reference/chat/create
+- Bray, T. (Ed.). (2017). *The JavaScript Object Notation (JSON) data
+  interchange format* (RFC 8259). Internet Engineering Task Force.
+  https://doi.org/10.17487/RFC8259

CLAUDE.md

@@ -88,7 +88,7 @@ A stdlib-Python lab implementing a single OpenAI-compatible API that routes, del
 - `orchestrator.py` — the domain heart: `ModelAgent`, `WorkflowStep`, `OrchestrationPolicy`, `ModelClient`, `TaskOrchestrator`, secret/PII redaction, budget enforcement, spend analytics, and the commercial-readiness report generators behind `/api/v1/*`. Domain code stays here until a second implementation forces extraction (see `docs/code_conventions.md`).
 - `server.py` — HTTP delivery adapter and `SecurityConfig`; all request validation lives here.
 - `admin.py` — static HTML/CSS/JS for the `/admin` operator console (stays inline while the product is dependency-free).
-- `credentials.py` / `kv_config.py` — the KV seam: `get_credential`/`register_credential` over pluggable backends (`InMemoryCredentialBackend` default; pgcrypto-encrypted `PostgresCredentialBackend`, selected via `CONTEXTUAL_ORCHESTRATOR_KV_BACKEND`).
+- `credentials.py` / `kv_config.py` — the KV seam: `get_credential`/`register_credential` over pluggable backends (`InMemoryCredentialBackend` default; pgcrypto-encrypted `PostgresCredentialBackend`, selected via `CONTEXTUAL_ORCHESTRATOR_KV_BACKEND`). Request-time provider host allowlisting uses `allowed_provider_hosts()` (`provider_egress.allowed_provider_hosts`); env is bootstrap only.
 - `cost_ledger.py` / `cost_router.py` / `batch_routing.py` / `token_counting.py` — the cost-review + routing hub: prompt-safe usage ledger with seven attribution dimensions, `RoutingPolicy` (sync vs batch from request hints + KV thresholds), and the [pg-llm-batch](https://github.com/ContextualWisdomLab/pg-llm-batch) batch/embeddings backends (a local in-process backend keeps the standalone path working with no external service).
 - `api_contract.py` / `conventions.py` — API-shape and naming-rule enforcement helpers.
 - `__main__.py` — the single entry point: CLI completion, `--serve`, `--eval`, and the `register-credential` bootstrap subcommand.

README.md

@@ -68,15 +68,15 @@ Seed the credential into the KV once at bootstrap:
 echo "$OPENAI_API_KEY" | python -m contextual_orchestrator register-credential --name OPENAI_API_KEY --value-stdin
 ```
 
-Non-mock providers must use `https://` URLs and a **resolvable KV credential** — a non-mock agent whose credential is missing raises `NotConfigured` rather than falling back to an environment variable. The runtime blocks loopback, private, link-local, multicast, and reserved provider addresses before sending a key. Set `CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS` to a comma-separated host allowlist when only approved model gateways should be reachable. External calls use a timeout and default output token cap.
+Non-mock providers must use `https://` URLs and a **resolvable KV credential** — a non-mock agent whose credential is missing raises `NotConfigured` rather than falling back to an environment variable. The runtime blocks loopback, private, link-local, multicast, and reserved provider addresses before sending a key. Seed `provider_egress.allowed_provider_hosts` on the **process-wide runtime ConfigStore** with `set_runtime_config` (or set `CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS` at process start so bootstrap can copy it once) when only approved model gateways should be reachable. A write to a separately constructed Postgres `get_config_store()` is ignored unless that store was installed with `set_runtime_config_store()` at bootstrap. External calls use a timeout and default output token cap.
 
 > The legacy `api_key_env` field is still accepted for back-compat, but its value is now treated as the **credential name** in the KV, not as an environment variable to read. This supersedes the old `api_key_env` env pattern.
 
 ## Architecture
 
 One public interface:
 
-- `/v1/chat/completions` accepts normal chat messages, and `"stream": true` returns an OpenAI-compatible `text/event-stream` of `chat.completion.chunk` deltas terminated by `data: [DONE]`. In **route** mode the worker's tokens are streamed live as they arrive from the provider (real token streaming); in **conduct** mode the multi-step answer is produced then framed as deltas (a workflow can't honestly token-stream a synthesizer that hasn't run yet).
+- `/v1/chat/completions` accepts normal chat messages, and `"stream": true` returns an OpenAI-compatible `text/event-stream` of `chat.completion.chunk` deltas terminated by `data: [DONE]`. In **route** mode the worker's tokens are streamed live as they arrive from the provider (real token streaming); in **conduct** mode the multi-step answer is produced then framed as deltas (a workflow can't honestly token-stream a synthesizer that hasn't run yet). `/v1/completions`, `/v1/responses`, and `/v1/embeddings` are the matching compatibility surfaces. Send `stream_options.include_usage` / `include_obfuscation` as omitted, `null`, or `false`; `true` and unknown keys fail closed. Tool-calling bodies must send a non-empty `messages` array of objects — empty, omitted, or non-object entries fail closed before passthrough. Omit `routing.channel=batch` and `latency_tolerant=true` on tool-calling requests; this gateway has no batch job plane on passthrough. Omit `seed`, `stop`, `n>1`, and `logprobs` on tool-calling requests — those knobs are not applied on passthrough and fail closed instead of billing a silent drop. Streamed route requests apply the `temperature` you send; they do not silently fall back to `0.2`. Optional `tools[].function.description`, `parameters`, and `strict` sent as JSON `null` are popped before the provider hop so SDK defaults stay omit-equivalent.
 - `TaskOrchestrator.complete()` decides whether to route to one worker or run a short workflow.
 - `TaskOrchestrator.compare_to_baseline(prompts, mode)` (CLI `--eval PROMPT...`) measures the orchestration engine against a single-worker baseline — per-prompt and aggregate latency plus a structural coverage delta (contributing steps + verifier-pass presence). It is a measured tradeoff report, not a human-quality claim.
 - Responses include orchestration mode metadata, and trusted callers can request the full trace for audit.
@@ -256,6 +256,7 @@ python tests/test_admin_contract.py
 python tests/test_conventions.py
 python tests/test_api_contract.py
 python tests/test_security_hardening.py
+python tests/test_provider_host_allowlist_kv.py
 python tests/test_repository_security_metadata.py
 python tests/test_product_planning_contract.py
 python tests/test_plugin_driven_artifacts.py

contextual_orchestrator/__main__.py

@@ -8,6 +8,7 @@
 import sys
 
 from .credentials import register_credential
+from .kv_config import seed_provider_egress_from_environ
 from .orchestrator import ModelClient, TaskOrchestrator, load_agents
 from .server import SecurityConfig, serve
 
@@ -94,6 +95,7 @@ def main() -> None:
                         help="Measure orchestration vs a single-worker baseline on these prompts and print the report.")
     args = parser.parse_args()
 
+    seed_provider_egress_from_environ()
     client = ModelClient(ca_bundle=args.provider_ca_bundle, verify_tls=not args.insecure_skip_tls_verify)
     orchestrator = TaskOrchestrator(
         load_agents(args.agents),

contextual_orchestrator/cost_ledger.py

@@ -222,6 +222,8 @@ class UsageRecord:
 
     def as_dict(self) -> Dict[str, Any]:
         """Flatten the record (attribution inlined) for JSON + SQL storage."""
+        # Execution identity is evidence of what ran — never a client-chosen tag.
+        # Account/service/team/group/company remain descriptive attribution.
         row = {
             "usage_record_id": self.usage_record_id,
             "created_at": self.created_at,
@@ -583,12 +585,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 +604,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 +624,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()]
 
 
@@ -681,14 +683,31 @@ def record_usage(
     ) -> UsageRecord:
         """Compute cost, build a :class:`UsageRecord`, persist it, and return it."""
         if isinstance(attribution, dict) or attribution is None:
-            dims = AttributionDimensions.from_mapping(attribution)
+            # Strip caller-controlled execution identity before mapping so a
+            # client cannot spoof model/provider rollups (buyer-bill honesty).
+            if isinstance(attribution, dict):
+                cleaned = {
+                    key: value
+                    for key, value in attribution.items()
+                    if key not in {"model_name", "provider", "upstream_api"}
+                }
+            else:
+                cleaned = None
+            dims = AttributionDimensions.from_mapping(cleaned)
         else:
-            dims = attribution
-        # Keep the model_name dimension aligned with the served model unless the
-        # caller pinned it explicitly, and default the provider dimension too.
-        if dims.model_name == UNATTRIBUTED and model:
+            dims = AttributionDimensions(
+                account=attribution.account,
+                service=attribution.service,
+                upstream_api=UNATTRIBUTED,
+                model_name=UNATTRIBUTED,
+                team=attribution.team,
+                group=attribution.group,
+                company=attribution.company,
+            )
+        # Execution identity always wins — descriptive dimensions stay as-is.
+        if model:
             dims.model_name = model
-        if dims.upstream_api == UNATTRIBUTED and provider:
+        if provider:
             dims.upstream_api = provider
 
         cost_amount, currency = self.price_book.compute_cost(

contextual_orchestrator/kv_config.py

@@ -19,8 +19,17 @@
 
 from __future__ import annotations
 
+import os
+import threading
 from typing import Any, Dict, Iterable, Optional, Protocol, Tuple
 
+PROVIDER_EGRESS_CATEGORY = "provider_egress"
+ALLOWED_PROVIDER_HOSTS_KEY = "allowed_provider_hosts"
+_ALLOWED_HOSTS_ENV = "CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS"
+
+_runtime_store: ConfigStore | None = None
+_runtime_lock = threading.Lock()
+
 
 class ConfigStore(Protocol):
     """Minimal KV config contract shared by every backend."""
@@ -147,3 +156,74 @@ def get_config_store(
         return adapter
     except Exception:  # pragma: no cover - fall back when deps/DB unavailable
         return InMemoryConfigStore(seed=seed)
+
+
+def get_runtime_config_store() -> ConfigStore:
+    """Return the process-wide KV used at request time (created on first use)."""
+    global _runtime_store
+    if _runtime_store is None:
+        with _runtime_lock:
+            if _runtime_store is None:
+                _runtime_store = InMemoryConfigStore()
+    return _runtime_store
+
+
+def set_runtime_config_store(store: ConfigStore | None) -> None:
+    """Install (or, with ``None``, reset) the process-wide request-time KV."""
+    global _runtime_store
+    with _runtime_lock:
+        _runtime_store = store
+
+
+def reset_runtime_config_store() -> None:
+    """Drop the process-wide KV so the next read starts empty (tests)."""
+    set_runtime_config_store(None)
+
+
+def get_runtime_config(category: str, key: str, default: Any = None) -> Any:
+    """Read one request-time config value from the process KV. Never os.getenv."""
+    return get_runtime_config_store().get(category, key, default)
+
+
+def set_runtime_config(category: str, key: str, value: Any) -> None:
+    """Write one request-time config value into the process KV."""
+    get_runtime_config_store().set(category, key, value)
+
+
+def _parse_host_allowlist(raw: Any) -> frozenset[str]:
+    """Split a CSV or sequence of hosts into a lower-cased frozenset."""
+    if raw is None:
+        return frozenset()
+    if isinstance(raw, (list, tuple, set, frozenset)):
+        tokens = [str(item) for item in raw]
+    else:
+        tokens = str(raw).split(",")
+    return frozenset(token.strip().lower() for token in tokens if str(token).strip())
+
+
+def allowed_provider_hosts() -> frozenset[str]:
+    """Return the request-time provider host allowlist from the KV only.
+
+    An empty set means "no extra hostname filter" — public HTTPS hosts still
+    pass the private/loopback/reserved address checks in ``ModelClient``.
+    """
+    raw = get_runtime_config(PROVIDER_EGRESS_CATEGORY, ALLOWED_PROVIDER_HOSTS_KEY, "")
+    return _parse_host_allowlist(raw)
+
+
+def seed_provider_egress_from_environ() -> None:
+    """Bootstrap: copy the env allowlist into KV when that KV key is still empty.
+
+    This is the only allowed ``os.environ`` read for the host allowlist.
+    Request-time validation must call :func:`allowed_provider_hosts`.
+    The empty-check and write share ``_runtime_lock`` so concurrent
+    ``main()`` + ``serve()`` seeds cannot both observe an empty key.
+    """
+    store = get_runtime_config_store()
+    with _runtime_lock:
+        existing = store.get(PROVIDER_EGRESS_CATEGORY, ALLOWED_PROVIDER_HOSTS_KEY, None)
+        if existing not in (None, ""):
+            return
+        raw = os.environ.get(_ALLOWED_HOSTS_ENV, "")
+        if raw.strip():
+            store.set(PROVIDER_EGRESS_CATEGORY, ALLOWED_PROVIDER_HOSTS_KEY, raw)