fix(security): pin provider egress to validated public addresses

CHANGELOG.md

@@ -0,0 +1,16 @@
+# 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
+
+- 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/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

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

tests/test_security_hardening.py

@@ -2,11 +2,13 @@
 
 import json
 import os
+import socket
 import threading
 import urllib.error
 import urllib.request
 from pathlib import Path
 import sys
+from unittest import mock
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
@@ -311,6 +313,60 @@ def test_provider_transport_rejects_protocol_relative_batch_paths() -> None:
         raise AssertionError("protocol-relative provider path should fail before urllib opens it")
 
 
+def test_external_provider_rejects_non_global_resolved_addresses() -> None:
+    # The egress guard must reject ANY non-globally-routable resolved address, not
+    # only the RFC1918/loopback/link-local/multicast/reserved flag set. RFC 6598
+    # shared address space (100.64.0.0/10, carrier-grade NAT and commonly used for
+    # cloud-internal services) is non-public yet carries NONE of those flags, so a
+    # host that resolves into it must still be blocked or it becomes an SSRF hole to
+    # internal targets. getaddrinfo is stubbed so the check is deterministic offline.
+    client = ModelClient()
+    backend = InMemoryCredentialBackend()
+    backend.set("MODEL_KEY", "sk-shared-space")
+    set_backend(backend)
+    shared_space_agent = ModelAgent(
+        "shared_space_agent", "gpt-example", "https://provider.example/v1", "MODEL_KEY"
+    )
+    resolved = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("100.64.0.1", 443))]
+    try:
+        with mock.patch(
+            "contextual_orchestrator.orchestrator.socket.getaddrinfo",
+            return_value=resolved,
+        ):
+            try:
+                client._validate_provider(shared_space_agent)
+            except RuntimeError as exc:
+                assert "non-public address" in str(exc)
+            else:
+                raise AssertionError(
+                    "provider resolving to RFC 6598 shared address space "
+                    "(100.64.0.0/10) must be rejected by the egress guard"
+                )
+    finally:
+        set_backend(None)
+
+
+def test_external_provider_allows_public_resolved_address() -> None:
+    # The tightened guard must not over-block: a genuinely public, globally
+    # routable resolved address still passes validation.
+    client = ModelClient()
+    backend = InMemoryCredentialBackend()
+    backend.set("MODEL_KEY", "sk-public")
+    set_backend(backend)
+    public_agent = ModelAgent(
+        "public_agent", "gpt-example", "https://provider.example/v1", "MODEL_KEY"
+    )
+    resolved = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 443))]
+    try:
+        with mock.patch(
+            "contextual_orchestrator.orchestrator.socket.getaddrinfo",
+            return_value=resolved,
+        ):
+            client._validate_provider(public_agent)  # must not raise
+    finally:
+        set_backend(None)
+
+
 def test_redact_value_preserves_non_string_scalars() -> None:
     assert redact_value(7) == 7
 
@@ -330,5 +386,7 @@ def test_redact_value_preserves_non_string_scalars() -> None:
     test_external_provider_rejects_insecure_or_unlisted_hosts()
     test_provider_transport_rejects_local_url_schemes_before_urllib()
     test_provider_transport_rejects_protocol_relative_batch_paths()
+    test_external_provider_rejects_non_global_resolved_addresses()
+    test_external_provider_allows_public_resolved_address()
     test_redact_value_preserves_non_string_scalars()
     print("ok")