fix: discover models from an allowlisted configured gateway

README.md

@@ -129,7 +129,7 @@ Run an evaluation against that server with `--temperature 0` for repeatable judg
 Model-based conduct verification requires `fast-mlsirm` in the same runtime and fails closed when it is absent or broken; fast-mlsirm sends its judge completion through this contextual-orchestrator gateway, so no direct provider fallback is used. “Same runtime” means that the exact interpreter used for the live run can import both packages: install both checkouts into one environment (prefer editable installs), or expose both source roots with `PYTHONPATH` during a source run. Before a live judge benchmark, run `python -m contextual_orchestrator check-fast-mlsirm` with that exact interpreter. It prints the interpreter, package version, transitive-import status, and contextual contract check, and exits nonzero on a missing dependency or contract mismatch. Do not run the preflight in one virtual environment and the judge in another. See [ADR 0001](docs/planning/adrs/0001-fail-closed-model-judgment.md).
 
 The agent pool is manageable at runtime: `POST`/`PATCH`/`DELETE` on `/api/v1/agent_pools/default/worker_agents[/{id}]` add, govern, and remove model-group members. Pass `--agents-db PATH` (or `CONTEXTUAL_ORCHESTRATOR_AGENTS_DB`) to persist those changes to a stdlib sqlite file — stored changes overlay the seed agents file at startup, and removals write disabled tombstones so they survive restarts; without it the pool is in-memory as before.
-Beyond the local MLX/llama.cpp discovery above, `python -m contextual_orchestrator discover-models [--agents-db PATH]` discovers models from remote providers (OpenAI, OpenRouter, NVIDIA NIM ×2 keys, Bytez) for any subset of their KV-registered credentials, and can persist them into the same `--agents-db` sqlite file, added disabled by default. See [docs/kv-credentials.md](docs/kv-credentials.md#multi-provider-auto-discovery) for the credential-name table and cost-based auto-selection.
+Beyond the local MLX/llama.cpp discovery above, `python -m contextual_orchestrator discover-models [--agents-db PATH]` discovers models from remote providers (OpenAI, OpenRouter, NVIDIA NIM ×2 keys, Bytez, and an allowlisted OpenAI-compatible gateway) for any subset of their KV-registered credentials, and can persist them into the same `--agents-db` sqlite file, added disabled by default. See [docs/kv-credentials.md](docs/kv-credentials.md#multi-provider-auto-discovery) for the credential-name table and cost-based auto-selection.
 
 Seed the credential into the KV once at bootstrap:
 

contextual_orchestrator/__main__.py

@@ -13,8 +13,12 @@
 from .credentials import get_credential, register_credential
 from .kv_config import InMemoryConfigStore
 from .model_discovery import (
+    CONFIGURED_GATEWAY_CREDENTIAL_NAME,
+    PROVIDER_MODEL_SOURCES,
+    ProviderModelSource,
     agent_from_discovered,
     agent_id_for,
+    configured_gateway_source,
     discover_all_models,
     free_discovered_models,
     refresh_price_book,
@@ -207,6 +211,17 @@ def _register_credential_command(argv: list[str]) -> None:
     print(json.dumps({"registered": args.name, "backend": "kv"}, ensure_ascii=False))
 
 
+def _bootstrap_discovery_sources() -> tuple[ProviderModelSource, ...]:
+    """Promote a configured gateway secret to KV and return discovery sources."""
+    source = configured_gateway_source(os.environ)
+    if source is None:
+        return PROVIDER_MODEL_SOURCES
+    secret = os.environ.get(CONFIGURED_GATEWAY_CREDENTIAL_NAME, "")
+    if secret.strip():
+        register_credential(CONFIGURED_GATEWAY_CREDENTIAL_NAME, secret.rstrip("\r\n"))
+    return (*PROVIDER_MODEL_SOURCES, source)
+
+
 def _discover_models_command(argv: list[str]) -> None:
     """Query every provider with a KV-registered credential and report the models found.
 
@@ -242,7 +257,7 @@ def _discover_models_command(argv: list[str]) -> None:
     if args.enable_cheapest and not args.agents_db:
         parser.error("--enable-cheapest requires --agents-db")
 
-    discovered, errors = discover_all_models()
+    discovered, errors = discover_all_models(_bootstrap_discovery_sources())
     free_models = free_discovered_models(discovered)
     reported = free_models if args.free_only else discovered
     price_book = PriceBook(InMemoryConfigStore())
@@ -283,7 +298,12 @@ def _discover_models_command(argv: list[str]) -> None:
 
 def _auto_discover_runtime_agents(orchestrator: TaskOrchestrator) -> dict[str, list[str]]:
     """Discover and activate only models with explicit chat capability evidence."""
-    discovered, _errors = discover_all_models()
+    sources = _bootstrap_discovery_sources()
+    discovered, _errors = (
+        discover_all_models()
+        if sources is PROVIDER_MODEL_SOURCES
+        else discover_all_models(sources)
+    )
     chat_models = [model for model in discovered if "chat" in model.capabilities]
     existing_ids = {agent.id for agent in orchestrator.candidates}
     agents = [

contextual_orchestrator/model_discovery.py

@@ -17,10 +17,12 @@
 import json
 import math
 import re
+import ssl
 import urllib.error
 import urllib.request
 from dataclasses import dataclass, replace
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Mapping
+from urllib.parse import urlsplit, urlunsplit
 
 from .chat_capability import is_general_chat_agent_model_id
 from .credentials import get_credential
@@ -33,6 +35,7 @@
 _CAPABILITY_NAMES = {"embeddings": "embedding"}
 _MODELS_DEV_URL = "https://models.dev/api.json"
 _MODELS_DEV_OPENCODE_PROVIDER = "opencode"
+CONFIGURED_GATEWAY_CREDENTIAL_NAME = "LLM_GATEWAY_API_KEY"
 
 
 def _provider_discovery_error_code(exc: Exception) -> str:
@@ -70,6 +73,57 @@ class ProviderModelSource:
     capabilities: tuple[str, ...] = ()
 
 
+def configured_gateway_source(
+    environ: Mapping[str, str],
+) -> ProviderModelSource | None:
+    """Build one allowlisted OpenAI-compatible gateway source at bootstrap.
+
+    The URL is non-secret bootstrap transport. The API key is referenced only
+    by its KV credential name; callers may promote the environment value into
+    the credential registry before runtime discovery starts.
+    """
+    values = {
+        value.strip().rstrip("/")
+        for name in ("LLM_GATEWAY_API_URL", "LLM_GATEWAY_URL")
+        if isinstance((value := environ.get(name)), str) and value.strip()
+    }
+    if not values:
+        return None
+    if len(values) != 1:
+        raise ValueError("LLM gateway URL settings must identify the same endpoint")
+    raw_url = values.pop()
+    parsed = urlsplit(raw_url)
+    if (
+        parsed.scheme != "https"
+        or not parsed.hostname
+        or parsed.username
+        or parsed.password
+        or parsed.query
+        or parsed.fragment
+    ):
+        raise ValueError("LLM gateway URL must be a credential-free HTTPS base URL")
+    allowed_hosts = {
+        host.strip().casefold()
+        for host in environ.get(
+            "CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS", ""
+        ).split(",")
+        if host.strip()
+    }
+    if parsed.hostname.casefold() not in allowed_hosts:
+        raise ValueError("LLM gateway host must be present in the provider allowlist")
+    path = parsed.path.rstrip("/")
+    if not path.endswith("/v1"):
+        path += "/v1"
+    base_url = urlunsplit(("https", parsed.netloc, path, "", ""))
+    return ProviderModelSource(
+        provider_name="configured_gateway",
+        credential_name=CONFIGURED_GATEWAY_CREDENTIAL_NAME,
+        list_url=f"{base_url}/models",
+        chat_base_url=base_url,
+        capabilities=("chat",),
+    )
+
+
 # NVIDIA NIM is listed twice under two KV credential names (primary + sub) so both
 # keys participate in upstream load balancing without a second provider identity.
 PROVIDER_MODEL_SOURCES: tuple[ProviderModelSource, ...] = (
@@ -157,7 +211,18 @@ def _fetch_json(url: str, *, api_key: str = "", auth_scheme: str = "Bearer", tim
     headers = {"authorization": f"{auth_scheme} {api_key}"} if api_key else {}
     request = urllib.request.Request(url, headers=headers, method="GET")
     # Scheme is enforced to https:// immediately above; url is never attacker-controlled.
-    with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310 - fixed https provider hosts  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
+    try:
+        response = urllib.request.urlopen(request, timeout=timeout)  # noqa: S310 - validated https provider hosts  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
+    except urllib.error.URLError as exc:
+        if not isinstance(exc.reason, ssl.SSLCertVerificationError):
+            raise
+        import certifi
+
+        context = ssl.create_default_context(cafile=certifi.where())
+        response = urllib.request.urlopen(  # noqa: S310 - validated https provider hosts  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
+            request, timeout=timeout, context=context
+        )
+    with response:
         return json.loads(response.read().decode("utf-8"))
 
 

docs/kv-credentials.md

@@ -213,6 +213,14 @@ box, all resolved through `get_credential` (never fabricated, never read from
 | NVIDIA NIM (primary)| `NVIDIA_NIM_API_KEY`     | `Bearer <token>`   |
 | NVIDIA NIM (sub)    | `NVIDIA_NIM_API_KEY_SUB` | `Bearer <token>`   |
 | Bytez               | `BYTEZ_API_KEY`          | `Key <token>`      |
+| Configured OpenAI-compatible gateway | `LLM_GATEWAY_API_KEY` | `Bearer <token>` |
+
+For a configured gateway, the one-shot discovery/bootstrap boundary accepts
+`LLM_GATEWAY_API_URL` (or the equivalent `LLM_GATEWAY_URL`) only when its HTTPS
+host is listed in `CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS`. It promotes
+`LLM_GATEWAY_API_KEY` into the KV before discovery; provider requests and normal
+runtime routing then read the key only from KV. The full chat catalog remains
+visible, while `--free-only` still selects only structured zero-price evidence.
 
 Bytez's `Key <token>` scheme (rather than `Bearer`) is why `ModelAgent` has an
 `auth_scheme` field (default `"Bearer"`) — set it per agent when a provider

pyproject.toml

@@ -5,6 +5,7 @@ description = "Paper-grounded model orchestration lab with an enterprise admin c
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
+  "certifi>=2024.8.30",
   "cryptography>=43.0",
   "opentelemetry-api>=1.30.0",
   "opentelemetry-sdk>=1.30.0",

tests/test_discover_models_cli.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import os
 import sys
 import urllib.parse
 from io import StringIO
@@ -14,6 +15,7 @@
 from contextual_orchestrator.__main__ import main  # noqa: E402
 from contextual_orchestrator.credentials import (  # noqa: E402
     InMemoryCredentialBackend,
+    get_credential,
     register_credential,
     set_backend,
 )
@@ -97,6 +99,80 @@ def urlopen(request, timeout=None):
     ]
 
 
+def test_discover_models_bootstraps_allowlisted_openai_gateway_from_environment() -> None:
+    """A one-shot env secret enters KV before full paid/free chat discovery."""
+    set_backend(InMemoryCredentialBackend())
+    stdout = StringIO()
+    environment = {
+        "LLM_GATEWAY_API_URL": "https://llm-gateway-dev.hyosungitx.com/v1",
+        "LLM_GATEWAY_URL": "https://llm-gateway-dev.hyosungitx.com/v1/",
+        "LLM_GATEWAY_API_KEY": "gateway-secret\n",
+        "CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS": "llm-gateway-dev.hyosungitx.com",
+    }
+
+    def urlopen(request, timeout=None):
+        assert request.full_url == "https://llm-gateway-dev.hyosungitx.com/v1/models"
+        assert request.headers["Authorization"] == "Bearer gateway-secret"
+        return _Response({
+            "data": [
+                {"id": "gpt-5.6-sol"},
+                {
+                    "id": "community/free-chat",
+                    "pricing": {"prompt": "0", "completion": "0"},
+                },
+                {"id": "text-embedding-3-large"},
+            ]
+        })
+
+    try:
+        with (
+            patch.dict(os.environ, environment, clear=True),
+            patch.object(sys, "argv", ["contextual-orchestrator", "discover-models"]),
+            patch.object(sys, "stdout", stdout),
+            patch(
+                "contextual_orchestrator.model_discovery.urllib.request.urlopen",
+                side_effect=urlopen,
+            ),
+        ):
+            main()
+            assert get_credential("LLM_GATEWAY_API_KEY") == "gateway-secret"
+    finally:
+        set_backend(None)
+
+    report = json.loads(stdout.getvalue())
+    assert report["discovered_count"] == 2
+    assert report["free_tier_count"] == 1
+    assert [row["model"] for row in report["models"]] == [
+        "gpt-5.6-sol",
+        "community/free-chat",
+    ]
+
+
+def test_configured_gateway_requires_explicit_host_allowlist() -> None:
+    """Bootstrap URL configuration cannot create an unrestricted SSRF target."""
+    set_backend(InMemoryCredentialBackend())
+    try:
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "LLM_GATEWAY_URL": "https://gateway.example/v1",
+                    "LLM_GATEWAY_API_KEY": "secret",
+                },
+                clear=True,
+            ),
+            patch.object(sys, "argv", ["contextual-orchestrator", "discover-models"]),
+        ):
+            try:
+                main()
+            except ValueError as exc:
+                assert "allowlist" in str(exc)
+            else:  # pragma: no cover
+                raise AssertionError("unallowlisted gateway must fail")
+    finally:
+        set_backend(None)
+
+
 def test_discover_models_persists_to_agents_db(tmp_path) -> None:
     from contextual_orchestrator import TaskOrchestrator
     from contextual_orchestrator.orchestrator import ModelAgent

tests/test_model_discovery_boundaries.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import ssl
 import urllib.error
 from unittest.mock import patch
 
@@ -86,6 +87,30 @@ def urlopen(request, timeout=None):
     assert excinfo.value.error_code == "timeout"
 
 
+def test_system_ca_failure_retries_with_certifi_verification() -> None:
+    """A stale host trust store may fall back to certifi without disabling TLS."""
+    calls = []
+
+    def urlopen(request, **kwargs):
+        calls.append(kwargs)
+        if len(calls) == 1:
+            raise urllib.error.URLError(
+                ssl.SSLCertVerificationError(1, "unable to get local issuer")
+            )
+        return _Response({"data": []})
+
+    with patch(
+        "contextual_orchestrator.model_discovery.urllib.request.urlopen",
+        side_effect=urlopen,
+    ):
+        assert _fetch_json("https://provider.example/v1/models", timeout=1) == {
+            "data": []
+        }
+    assert "context" not in calls[0]
+    assert calls[1]["context"].verify_mode == ssl.CERT_REQUIRED
+    assert calls[1]["context"].check_hostname is True
+
+
 def test_malformed_json_maps_to_invalid_response_code() -> None:
     """A non-JSON provider body is invalid_response, not a crash."""
     register_credential("OPENAI_API_KEY", "sk-openai")