chore: integrate validated provider egress hardening into lock prerequisite

CHANGELOG.md

@@ -0,0 +1,17 @@
+# Changelog
+
+All notable changes to Contextual Orchestrator are documented in this file.
+
+The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and the project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [Unreleased]
+
+### Security
+
+- Pin each HTTPS provider connection to the exact public addresses approved during validation, preserve the original hostname for TLS verification, bypass environment proxy resolution, and reject redirects to close DNS-rebinding and credential-forwarding SSRF paths.
+- Reject provider hosts that resolve to any non-globally-routable address, including RFC 6598 shared address space, while retaining explicit multicast, private, loopback, link-local, and reserved-address protections.
+- Document narrowly scoped Semgrep suppressions for parameter-bound database queries, the explicit development-only TLS verification opt-out, and provider URLs that pass the egress guard.
+
+### Changed
+
+- Pin Atheris by Python interpreter so the Python 3.11 fuzz job and the newer central coverage-evidence image both install a published, hash-locked wheel.

contextual_orchestrator/__init__.py

@@ -37,9 +37,12 @@
 from .cost_router import CostRoutingCoordinator
 from .credentials import NotConfigured, get_credential, register_credential
 from .kv_config import InMemoryConfigStore, get_config_store
-from .orchestrator import ModelAgent, TaskOrchestrator, WorkflowStep, load_agents
+from .orchestrator import ModelAgent, ModelClient as _ModelClient, TaskOrchestrator, WorkflowStep, load_agents
+from .provider_transport import install_provider_transport as _install_provider_transport
 from .token_counting import HeuristicTokenCounter, build_token_counter
 
+_install_provider_transport(_ModelClient)
+
 __all__ = [
     "ModelAgent",
     "TaskOrchestrator",

contextual_orchestrator/cost_ledger.py

@@ -583,8 +583,8 @@ 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(
-                f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}",  # nosec B608 - ph is a DB-API placeholder.
+            cur.execute(  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
+                f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}",  # nosec B608 - ph is a DB-API placeholder; the value is bound, not interpolated.
                 (name,),
             )
             if cur.fetchone() is None:
@@ -602,8 +602,8 @@ def append(self, record: UsageRecord) -> None:
         placeholders = ", ".join(ph for _ in _USAGE_COLUMNS)
         columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(
-            f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})",  # nosec B608 - columns are fixed _USAGE_COLUMNS.
+        cur.execute(  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
+            f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})",  # nosec B608 - columns are the fixed _USAGE_COLUMNS constant; values are bound.
             tuple(row.get(column) for column in _USAGE_COLUMNS),
         )
         self._conn.commit()
@@ -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/clauses are fixed templates, values bound. nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
         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 opt-out; default verify_tls=True uses ssl.create_default_context(). nosemgrep: python.lang.security.unverified-ssl-context.unverified-ssl-context
         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 - URL from _provider_url after egress/SSRF validation. nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
             request,
             timeout=self.timeout,
             context=self._ssl_context,
@@ -473,8 +473,15 @@ def _validate_provider(self, agent: ModelAgent) -> None:
             raise RuntimeError(f"{agent.id} provider host is not allowlisted")
         for address in socket.getaddrinfo(hostname, parsed.port or 443, type=socket.SOCK_STREAM):
             ip_address = ipaddress.ip_address(address[4][0])
+            # ``not is_global`` rejects every non-globally-routable target, including
+            # ranges that carry none of the explicit flags below — notably RFC 6598
+            # shared address space (100.64.0.0/10, carrier-grade NAT / cloud-internal)
+            # and the unspecified address. The explicit flags are kept because some
+            # non-public multicast addresses report ``is_global`` True and must still
+            # be blocked.
             if (
-                ip_address.is_private
+                not ip_address.is_global
+                or ip_address.is_private
                 or ip_address.is_loopback
                 or ip_address.is_link_local
                 or ip_address.is_multicast

contextual_orchestrator/provider_transport.py

@@ -0,0 +1,208 @@
+"""DNS-pinned HTTPS transport for validated model-provider egress.
+
+The legacy orchestration module validates provider DNS answers before sending a
+request. A normal URL opener resolves the hostname again during connection,
+which leaves a time-of-check/time-of-use gap if DNS changes between validation
+and socket creation. This module installs a narrow transport extension on
+``ModelClient``: the second validation answer is retained, the socket connects
+only to one of those approved addresses, TLS still verifies the original
+hostname, environment proxies are bypassed, and redirects are rejected.
+"""
+
+from __future__ import annotations
+
+import http.client
+import ipaddress
+import socket
+import ssl
+from typing import Any, Iterator
+import urllib.error
+import urllib.request
+from urllib.parse import urlparse
+
+
+class _PinnedHTTPSConnection(http.client.HTTPSConnection):
+    """Connect to one validated IP while retaining the provider hostname for TLS."""
+
+    def __init__(
+        self,
+        server_hostname: str,
+        pinned_ip: str,
+        port: int,
+        timeout: float,
+        context: ssl.SSLContext,
+    ) -> None:
+        """Configure a direct TLS connection to a previously validated address."""
+        super().__init__(server_hostname, port=port, timeout=timeout, context=context)
+        self._pinned_ip = pinned_ip
+        self._server_hostname = server_hostname
+
+    def connect(self) -> None:
+        """Dial the pinned IP and verify the certificate against the original host."""
+        raw_socket = socket.create_connection(
+            (self._pinned_ip, self.port),
+            self.timeout,
+            self.source_address,
+        )
+        try:
+            self.sock = self._context.wrap_socket(
+                raw_socket,
+                server_hostname=self._server_hostname,
+            )
+        except Exception:  # noqa: BLE001 - close the raw socket, then preserve the TLS failure.
+            raw_socket.close()
+            raise
+
+
+class _ProviderHTTPResponse:
+    """Provider response wrapper that deterministically closes its connection."""
+
+    def __init__(self, response: Any, connection: Any) -> None:
+        """Retain the response and direct connection for context-managed cleanup."""
+        self._response = response
+        self._connection = connection
+
+    def __enter__(self) -> "_ProviderHTTPResponse":
+        """Return this response wrapper from a context manager."""
+        return self
+
+    def __exit__(self, _exc_type: Any, _exc: Any, _traceback: Any) -> None:
+        """Close the response and its connection when leaving the context."""
+        self.close()
+
+    def __iter__(self) -> Iterator[bytes]:
+        """Iterate raw response lines for server-sent-event streaming."""
+        return iter(self._response)
+
+    def __getattr__(self, name: str) -> Any:
+        """Delegate response metadata such as status and headers."""
+        return getattr(self._response, name)
+
+    def read(self, *args: Any, **kwargs: Any) -> bytes:
+        """Read bytes from the underlying provider response."""
+        return self._response.read(*args, **kwargs)
+
+    def close(self) -> None:
+        """Close both resources even when response cleanup raises."""
+        try:
+            self._response.close()
+        finally:
+            self._connection.close()
+
+
+def _validated_public_addresses(hostname: str, port: int, provider_label: str) -> tuple[str, ...]:
+    """Resolve, validate, and deduplicate addresses approved for one connection."""
+    validated_addresses: list[str] = []
+    for address in socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM):
+        resolved_address = ipaddress.ip_address(address[4][0])
+        if (
+            not resolved_address.is_global
+            or resolved_address.is_private
+            or resolved_address.is_loopback
+            or resolved_address.is_link_local
+            or resolved_address.is_multicast
+            or resolved_address.is_reserved
+        ):
+            raise RuntimeError(f"{provider_label} provider resolves to non-public address")
+        normalized_address = str(resolved_address)
+        if normalized_address not in validated_addresses:
+            validated_addresses.append(normalized_address)
+    if not validated_addresses:
+        raise RuntimeError(f"{provider_label} provider host did not resolve")
+    return tuple(validated_addresses)
+
+
+def install_provider_transport(model_client_type: type[Any]) -> None:
+    """Install DNS-pinned HTTPS validation and connection methods exactly once."""
+    if getattr(model_client_type, "_dns_pinned_transport_installed", False):
+        return
+
+    original_validate_provider = model_client_type._validate_provider
+
+    def validate_provider(self: Any, agent: Any) -> None:
+        """Validate provider policy, then retain the exact public DNS answer used."""
+        parsed = urlparse(agent.base_url)
+        hostname = parsed.hostname.lower() if parsed.hostname else ""
+        port = parsed.port or 443
+        pin_key = (hostname, port)
+        pins = getattr(self._local, "provider_address_pins", {})
+        pins.pop(pin_key, None)
+        self._local.provider_address_pins = pins
+
+        original_validate_provider(self, agent)
+        addresses = _validated_public_addresses(hostname, port, agent.id)
+        pins[pin_key] = addresses
+
+    def open_provider(self: Any, request: urllib.request.Request) -> Any:
+        """Open a request on a validation-time address without following redirects.
+
+        Public provider methods require HTTPS and invoke ``validate_provider``
+        before this transport. Plain HTTP remains delegated to urllib only for
+        the repository's private loopback integration helpers; the public policy
+        boundary rejects HTTP before provider egress.
+        """
+        parsed = urlparse(request.full_url)
+        if parsed.scheme == "http":
+            return urllib.request.urlopen(  # nosec B310 - public validation rejects HTTP; private loopback test seam only. nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
+                request,
+                timeout=self.timeout,
+            )
+        if parsed.scheme != "https" or not parsed.hostname:
+            raise RuntimeError("provider request URL must use http(s)")
+
+        port = parsed.port or 443
+        pin_key = (parsed.hostname.lower(), port)
+        pins = getattr(self._local, "provider_address_pins", {})
+        addresses = pins.get(pin_key)
+        if not addresses:
+            raise RuntimeError("provider request has no validated address pin")
+
+        target = parsed.path or "/"
+        if parsed.params:
+            target = f"{target};{parsed.params}"
+        if parsed.query:
+            target = f"{target}?{parsed.query}"
+        headers = dict(request.header_items())
+        headers["Connection"] = "close"
+
+        last_error: BaseException | None = None
+        connection_type = getattr(self, "_https_connection_class", _PinnedHTTPSConnection)
+        for pinned_ip in addresses:
+            connection = connection_type(
+                parsed.hostname,
+                pinned_ip,
+                port,
+                self.timeout,
+                self._ssl_context,
+            )
+            try:
+                connection.request(
+                    request.get_method(),
+                    target,
+                    body=request.data,
+                    headers=headers,
+                )
+                response = connection.getresponse()
+            except (OSError, http.client.HTTPException) as exc:
+                connection.close()
+                last_error = exc
+                continue
+            if response.status >= 300:
+                status = response.status
+                reason = response.reason
+                response_headers = response.headers
+                response.close()
+                connection.close()
+                raise urllib.error.HTTPError(
+                    request.full_url,
+                    status,
+                    reason,
+                    response_headers,
+                    None,
+                )
+            return _ProviderHTTPResponse(response, connection)
+        raise urllib.error.URLError(last_error or "provider connection failed")
+
+    model_client_type._validate_provider = validate_provider
+    model_client_type._open_provider = open_provider
+    model_client_type._dns_pinned_transport_installed = True

fuzz/requirements-atheris.in

@@ -1,3 +1,9 @@
-# Atheris coverage-guided job deps (Python 3.11). Compile: uv pip compile fuzz/requirements-atheris.in --generate-hashes --python-version 3.11 --universal -o fuzz/requirements-atheris.txt
+# Atheris coverage-guided job deps. atheris is published per-interpreter: the
+# repo fuzz job runs CPython 3.11, where the newest published wheel is 3.0.0,
+# while the central OpenCode coverage-evidence image runs a newer CPython
+# (3.13+) where only 3.1.0 is published. Pin per interpreter with environment
+# markers so a single hash lock satisfies both --require-hashes installs.
+# Compile: uv pip compile fuzz/requirements-atheris.in --generate-hashes --python-version 3.11 --universal -o fuzz/requirements-atheris.txt
 pip
-atheris==3.0.0
+atheris==3.0.0; python_version < "3.13"
+atheris==3.1.0; python_version >= "3.13"

fuzz/requirements-atheris.txt

@@ -1,11 +1,16 @@
 # This file was autogenerated by uv via the following command:
 #    uv pip compile fuzz/requirements-atheris.in --generate-hashes --python-version 3.11 --universal -o fuzz/requirements-atheris.txt
-atheris==3.0.0 \
+atheris==3.0.0 ; python_full_version < '3.13' \
     --hash=sha256:1f0929c7bc3040f3fe4102e557718734190cf2d7718bbb8e3ce6d3eb56ef5bb3 \
     --hash=sha256:510e502c57b6dc615fb174066407af620d4c7f73cf08a782c86e7761bf12c4eb \
     --hash=sha256:8a5c8a781467c187da40fd29139784193e2647058831f837f675d0bb8cbd8746 \
     --hash=sha256:a402cdca8a650d1371050b1f9552eb4cdc488d2db64950d603c4560318365eac
     # via -r fuzz/requirements-atheris.in
+atheris==3.1.0 ; python_full_version >= '3.13' \
+    --hash=sha256:315a0b5c819852b1ffe1ca72efc389c7724881f2c33e4aacb8c6bcec49bd5011 \
+    --hash=sha256:ec5e11f21a4c197fe91f7aea2b2de88e623c73a21fc07b105ac6329a1588457b \
+    --hash=sha256:f8a9f51ce8369026e8eb7b7174835e8c4c85a1a6db5d9add36c15100779d2a39
+    # via -r fuzz/requirements-atheris.in
 pip==26.1.2 \
     --hash=sha256:382ff9f685ee3bc25864f820aa50505825f10f5458ffff07e30a6d96e5715cab \
     --hash=sha256:f49cd134c61cf2fd75e0ce2676db03e4054504a5a4986d00f8299ae632dc4605

pyproject.toml

@@ -20,6 +20,7 @@ db = [
 ]
 fuzz = [
   "atheris==3.0.0; python_version < '3.13'",
+  "atheris==3.1.0; python_version >= '3.13'",
 ]
 
 [tool.contextual_orchestrator]