fix(kv): provider host allowlist from runtime config store

contextual_orchestrator/__init__.py

@@ -36,7 +36,14 @@
 )
 from .cost_router import CostRoutingCoordinator
 from .credentials import NotConfigured, get_credential, register_credential
-from .kv_config import InMemoryConfigStore, get_config_store
+from .kv_config import (
+    InMemoryConfigStore,
+    allowed_provider_hosts,
+    get_config_store,
+    get_config_value,
+    set_config_value,
+    set_runtime_config_store,
+)
 from .orchestrator import ModelAgent, TaskOrchestrator, WorkflowStep, load_agents
 from .token_counting import HeuristicTokenCounter, build_token_counter
 
@@ -67,6 +74,10 @@
     # config / tokens
     "InMemoryConfigStore",
     "get_config_store",
+    "get_config_value",
+    "set_config_value",
+    "set_runtime_config_store",
+    "allowed_provider_hosts",
     "HeuristicTokenCounter",
     "build_token_counter",
     # routing / batch

contextual_orchestrator/__main__.py

@@ -8,6 +8,11 @@
 import sys
 
 from .credentials import register_credential
+from .kv_config import (
+    allowed_provider_hosts,
+    get_config_store,
+    set_runtime_config_store,
+)
 from .orchestrator import ModelClient, TaskOrchestrator, load_agents
 from .server import SecurityConfig, serve
 
@@ -119,6 +124,15 @@ def main() -> None:
             args.admin_token and args.inference_token
         ):
             parser.error("split token mode requires both --admin-token and --inference-token")
+        set_runtime_config_store(
+            get_config_store(
+                os.environ.get("CONTEXTUAL_ORCHESTRATOR_KV_DSN") or None,
+                fernet_key=(
+                    os.environ.get("CONTEXTUAL_ORCHESTRATOR_KV_PASSPHRASE") or None
+                ),
+            )
+        )
+        allowed_provider_hosts()
         serve(
             orchestrator,
             host=args.host,

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/kv_config.py

@@ -147,3 +147,68 @@ def get_config_store(
         return adapter
     except Exception:  # pragma: no cover - fall back when deps/DB unavailable
         return InMemoryConfigStore(seed=seed)
+
+
+# Process-wide runtime config (mirrors credentials.set_backend for non-secret KV).
+_runtime_config_store: Optional[ConfigStore] = None
+
+
+def get_runtime_config_store() -> ConfigStore:
+    """Return the process runtime config store (never secrets).
+
+    Defaults to an empty in-memory store. Tests inject via
+    :func:`set_runtime_config_store`. Request-time code must read tunables from
+    here (or an injected store), not ``os.getenv``.
+    """
+    global _runtime_config_store
+    if _runtime_config_store is None:
+        _runtime_config_store = InMemoryConfigStore()
+    return _runtime_config_store
+
+
+def set_runtime_config_store(store: Optional[ConfigStore]) -> None:
+    """Install or clear the process runtime config store (tests / bootstrap)."""
+    global _runtime_config_store
+    _runtime_config_store = store
+
+
+def get_config_value(category: str, key: str, default: Any = None) -> Any:
+    """Read one runtime config value from the process KV store."""
+    return get_runtime_config_store().get(category, key, default)
+
+
+def set_config_value(category: str, key: str, value: Any) -> None:
+    """Write one runtime config value into the process KV store."""
+    get_runtime_config_store().set(category, key, value)
+
+
+PROVIDER_ALLOWED_HOSTS_KEY = "allowed_hosts"
+PROVIDER_CONFIG_CATEGORY = "provider"
+# Bootstrap-only env name: may seed the KV once, never used as the runtime source.
+ALLOWED_PROVIDER_HOSTS_ENV = "CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS"
+
+
+def allowed_provider_hosts() -> set[str]:
+    """Return the operator host allowlist from the runtime KV config store.
+
+    Empty set means no extra host filter (public-IP checks still apply). When the
+    KV has no value yet, a one-shot bootstrap may seed from
+    ``CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS`` into the store and then
+    only the store is authoritative.
+    """
+    import os
+
+    store = get_runtime_config_store()
+    raw = store.get(PROVIDER_CONFIG_CATEGORY, PROVIDER_ALLOWED_HOSTS_KEY, None)
+    if raw is None:
+        # One-shot bootstrap: seed even when the env is empty so a later env
+        # mutation cannot re-seed the store after first read (KV purity).
+        env_val = os.environ.get(ALLOWED_PROVIDER_HOSTS_ENV, "")
+        seeded = env_val.strip() if isinstance(env_val, str) else ""
+        store.set(PROVIDER_CONFIG_CATEGORY, PROVIDER_ALLOWED_HOSTS_KEY, seeded)
+        raw = seeded
+    if isinstance(raw, (list, tuple, set)):
+        parts = [str(item) for item in raw]
+    else:
+        parts = str(raw).split(",")
+    return {part.strip().lower() for part in parts if part and str(part).strip()}

contextual_orchestrator/orchestrator.py

@@ -27,6 +27,7 @@
 
 from .conventions import require_object_name
 from .credentials import NotConfigured, get_credential
+from .kv_config import allowed_provider_hosts
 
 
 ChatMessage = dict[str, str]
@@ -230,7 +231,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 +308,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,
@@ -463,11 +464,8 @@ def _validate_provider(self, agent: ModelAgent) -> None:
         parsed = urlparse(agent.base_url)
         if parsed.scheme != "https" or not parsed.hostname:
             raise RuntimeError(f"{agent.id} base_url must use https")
-        allowed_hosts = {
-            host.strip().lower()
-            for host in os.environ.get("CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS", "").split(",")
-            if host.strip()
-        }
+        # Host allowlist is KV-backed (provider/allowed_hosts); env is bootstrap-only.
+        allowed_hosts = allowed_provider_hosts()
         hostname = parsed.hostname.lower()
         if allowed_hosts and hostname not in allowed_hosts:
             raise RuntimeError(f"{agent.id} provider host is not allowlisted")

docs/architecture.md

@@ -53,3 +53,7 @@ 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.
+
+## Provider host allowlist (KV)
+
+`ModelClient._validate_provider` reads `provider/allowed_hosts` from the runtime config KV via `allowed_provider_hosts()`. Env `CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS` is bootstrap-only.

docs/kv-credentials.md

@@ -74,19 +74,25 @@ CREATE TABLE IF NOT EXISTS provider_credentials (
 Secrets are encrypted at rest with `pgp_sym_encrypt(value, passphrase)` and read
 back with `pgp_sym_decrypt(encrypted_value, passphrase)`.
 
-## The single allowed env use: bootstrap transport
-
-Environment variables are permitted in **exactly one place** — as bootstrap
-transport to *connect to and unlock the KV itself*, never as the runtime source
-of a provider key:
-
-| Variable                                   | Role                                   |
-| ------------------------------------------ | -------------------------------------- |
-| `CONTEXTUAL_ORCHESTRATOR_KV_BACKEND`       | backend selector (`memory`/`postgres`) |
-| `CONTEXTUAL_ORCHESTRATOR_KV_DSN`           | Postgres DSN to reach the registry     |
-| `CONTEXTUAL_ORCHESTRATOR_KV_PASSPHRASE`    | pgcrypto passphrase to unlock secrets  |
-
-These open the KV. They are not provider API keys.
+## Allowed environment use: bootstrap transport and one-shot config seeding
+
+Environment variables are permitted only during process bootstrap. They may
+select, connect to, and unlock the KV, or seed the non-secret provider host
+allowlist exactly once. They are never the request-time source of a provider
+key or mutable request-time policy:
+
+| Variable | Bootstrap role |
+| -------- | -------------- |
+| `CONTEXTUAL_ORCHESTRATOR_KV_BACKEND` | Select the `memory` or `postgres` credential backend. |
+| `CONTEXTUAL_ORCHESTRATOR_KV_DSN` | Connect to the Postgres registry. |
+| `CONTEXTUAL_ORCHESTRATOR_KV_PASSPHRASE` | Unlock encrypted provider credentials. |
+| `CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS` | Seed `provider/allowed_hosts` once when that KV key is initially absent. |
+
+The first three values select or open the KV. The provider-host value is a
+one-shot bootstrap input copied into the runtime config store. After store
+installation and first seeding, the KV value is authoritative and later
+environment mutations have no effect. None of these values is a provider API
+key.
 
 ## Bootstrapping a credential
 
@@ -155,3 +161,28 @@ This credential seam is the durable first step of growing
 per-tenant scoping can grow behind without touching the routing engine. The
 Rust/Python hybrid gateway is a later, separately-approved effort and is **not**
 started here.
+
+## Non-secret runtime config (provider host allowlist)
+
+Operator tunables that are not secrets also resolve from the KV config store
+(`get_runtime_config_store` / `get_config_value`), never from `os.getenv` at
+request time.
+
+| Category | Key | Purpose |
+| -------- | --- | ------- |
+| `provider` | `allowed_hosts` | Comma-separated or list of HTTPS provider hostnames allowed for egress. Empty = no extra filter (public-IP checks still apply). |
+
+Bootstrap: if `provider/allowed_hosts` is unset, the first read may seed the store
+from `CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS`. After seeding, only the
+store is authoritative; live environment mutations do not change request-time
+policy.
+
+```python
+from contextual_orchestrator.kv_config import set_config_value
+
+set_config_value(
+    "provider",
+    "allowed_hosts",
+    "api.openai.com,integrate.api.nvidia.com",
+)
+```

pyproject.toml

@@ -44,3 +44,9 @@ ignore-nested-functions = true
 ignore-private = true
 ignore-property-decorators = true
 ignore-semiprivate = true
+
+[tool.ruff.lint.per-file-ignores]
+# The ledger interpolates only fixed column catalogs and DB-API placeholder
+# tokens; all request values remain bound parameters. Keep the suppression
+# scoped to the single audited module and the single false-positive rule.
+"contextual_orchestrator/cost_ledger.py" = ["S608"]