From 399d750c539bd3e41bfaae021652f2314db0ac54 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 01:23:49 +0000 Subject: [PATCH 01/10] fix(security): close SSRF egress gap for non-globally-routable addresses MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ModelClient._validate_provider is the SSRF/egress guard: it resolves a provider host and must reject any address that is not a public, globally-routable target ("provider resolves to non-public address"). It only checked is_private/is_loopback/is_link_local/is_multicast/ is_reserved, but that flag set does not cover every non-public range. RFC 6598 shared address space (100.64.0.0/10 — carrier-grade NAT, and commonly used for cloud-internal services/proxies) reports False for all five flags while ipaddress.is_global is also False, so a provider whose host resolved into 100.64.0.0/10 (or its IPv4-mapped ::ffff:100.64.x form, or the unspecified address on interpreter versions where is_private is False for it) passed validation and became a reachable internal SSRF target. Fix: also reject `not ip_address.is_global`. The explicit flags are kept because some non-public multicast addresses report is_global True and must still be blocked, so the OR-combination is strictly wider than before with no regression: every previously blocked address stays blocked, genuinely public unicast addresses stay allowed, and the shared-address-space gap is closed. Regression tests (getaddrinfo stubbed for deterministic offline checks): - a host resolving to 100.64.0.1 must be rejected (fails before this fix) - a host resolving to 8.8.8.8 must still be accepted (guards over-blocking) Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01REEc4WtvMHbGD23XK6xbLK --- contextual_orchestrator/orchestrator.py | 9 +++- tests/test_security_hardening.py | 58 +++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 1 deletion(-) diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 0097b722e..d6a952c5d 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -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 diff --git a/tests/test_security_hardening.py b/tests/test_security_hardening.py index 67134ea6b..715995216 100644 --- a/tests/test_security_hardening.py +++ b/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") From 035a7cb7d9394247112d405e1759b6b68d322598 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 3 Aug 2026 09:39:16 +0000 Subject: [PATCH 02/10] fix(security): suppress five verified false-positive Semgrep findings The required Semgrep (multi-language SAST) gate failed on five findings, blocking OpenCode approval on the SSRF-egress fix. All five are verified false positives that already carry `# nosec` justifications; each now also gets the matching scoped `# nosemgrep` so the gate reflects real risk: - cost_ledger.py x3 sqlalchemy-execute-raw-query (ERROR): parameterized DB-API queries -- the f-strings interpolate only the placeholder symbol (?/%s) and the fixed _USAGE_COLUMNS constant / fixed clause templates; every value is bound as a driver parameter, so no untrusted value reaches raw SQL. - orchestrator.py unverified-ssl-context (ERROR): secure by default (verify_tls=True -> ssl.create_default_context()); ssl._create_unverified_context() is only reached on the explicit, documented dev-only verify_tls=False opt-out. - orchestrator.py dynamic-urllib-use-detected (WARN): the urlopen target is _provider_url(agent) after provider egress/SSRF validation (loopback/private/ reserved blocked), not user-controlled. Comments only (no behavior change); the gate is not weakened -- only these exact rule+line pairs are suppressed, with justification. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01REEc4WtvMHbGD23XK6xbLK --- contextual_orchestrator/cost_ledger.py | 10 +++++----- contextual_orchestrator/orchestrator.py | 4 ++-- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index d3943c5be..3ede24e81 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/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()] diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index d6a952c5d..7c1b0d115 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/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, From e36ecda7a1159b839c92b10f2ab03ba82782c9c0 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 3 Aug 2026 12:22:11 +0000 Subject: [PATCH 03/10] fix(fuzz): pin atheris per-interpreter (3.0.0 for <3.13, 3.1.0 for >=3.13) atheris publishes different newest versions per CPython: 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. A single unconditional pin cannot satisfy both --require-hashes installs of this one lock: - pinning 3.0.0 fails the central coverage image build on 3.13+ ("No matching distribution found for atheris==3.0.0" -> "Trusted coverage tool image build failed before PR execution"), blocking OpenCode approval for every PR against this base; - pinning 3.1.0 fails the repo's own "Atheris coverage-guided" job on 3.11 ("No matching distribution found for atheris==3.1.0"). Split the pin with environment markers (atheris==3.0.0 for python_version < 3.13, atheris==3.1.0 for >= 3.13) and regenerate the hash lock with the recorded `uv pip compile ... --python-version 3.11 --universal` command, so both interpreters resolve a published, hashed wheel. Verified: pip on 3.11 selects 3.0.0 (cp311 wheel), pip on 3.13+ selects 3.1.0. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01REEc4WtvMHbGD23XK6xbLK --- fuzz/requirements-atheris.in | 10 ++++++++-- fuzz/requirements-atheris.txt | 7 ++++++- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/fuzz/requirements-atheris.in b/fuzz/requirements-atheris.in index b930f7524..f90eaaec8 100644 --- a/fuzz/requirements-atheris.in +++ b/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" diff --git a/fuzz/requirements-atheris.txt b/fuzz/requirements-atheris.txt index b3e913ba6..49919d9c3 100644 --- a/fuzz/requirements-atheris.txt +++ b/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 From 0703a6b086568f3a2de7aadfa098a4a337c9c036 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 4 Aug 2026 08:41:58 +0900 Subject: [PATCH 04/10] docs: start changelog for security and coverage fixes --- CHANGELOG.md | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) create mode 100644 CHANGELOG.md diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 000000000..dca687bc6 --- /dev/null +++ b/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. From f969c6b93fea56c378587dea2b180cb66d3acd6f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 00:34:40 +0000 Subject: [PATCH 05/10] fix(fuzz): mirror atheris per-interpreter split into pyproject [fuzz] extra MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The hash lock (fuzz/requirements-atheris.txt) selects atheris 3.0.0 for CPython <3.13 and 3.1.0 for >=3.13, but the pyproject [fuzz] extra only carried the <3.13 pin, so `pip install .[fuzz]` on CPython 3.13+ installed no atheris at all — the extra-install and lockfile-install paths diverged. Add the matching `atheris==3.1.0; python_version >= "3.13"` branch so both paths resolve the same dependency on every interpreter. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01REEc4WtvMHbGD23XK6xbLK --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 65bd69eac..a476d2790 100644 --- a/pyproject.toml +++ b/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] From 9a08d65b1ae26e3b28fb58ad4fb9afd5d0cfb5c9 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 4 Aug 2026 12:18:00 +0900 Subject: [PATCH 06/10] fix(security): add DNS-pinned provider transport --- contextual_orchestrator/provider_transport.py | 208 ++++++++++++++++++ 1 file changed, 208 insertions(+) create mode 100644 contextual_orchestrator/provider_transport.py diff --git a/contextual_orchestrator/provider_transport.py b/contextual_orchestrator/provider_transport.py new file mode 100644 index 000000000..fa87232d3 --- /dev/null +++ b/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 From 179204f95a428a3787ce967fffd9e1ea6c2682c5 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 4 Aug 2026 12:18:18 +0900 Subject: [PATCH 07/10] fix(security): install DNS-pinned provider transport --- contextual_orchestrator/__init__.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/contextual_orchestrator/__init__.py b/contextual_orchestrator/__init__.py index 70dbd71c6..79f7e3cba 100644 --- a/contextual_orchestrator/__init__.py +++ b/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", From 4a95e5ef1cd3310955464eb1a5c1b2586475fc9f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 4 Aug 2026 12:19:01 +0900 Subject: [PATCH 08/10] test(security): cover DNS-pinned provider transport --- tests/test_provider_address_pinning.py | 372 +++++++++++++++++++++++++ 1 file changed, 372 insertions(+) create mode 100644 tests/test_provider_address_pinning.py diff --git a/tests/test_provider_address_pinning.py b/tests/test_provider_address_pinning.py new file mode 100644 index 000000000..ee426c287 --- /dev/null +++ b/tests/test_provider_address_pinning.py @@ -0,0 +1,372 @@ +"""Regression tests for DNS-pinned provider connections.""" + +from __future__ import annotations + +import ssl +import urllib.error +import urllib.request +from unittest import mock + +import pytest + +from contextual_orchestrator import ModelAgent +from contextual_orchestrator.credentials import InMemoryCredentialBackend, set_backend +from contextual_orchestrator.orchestrator import ModelClient +from contextual_orchestrator.provider_transport import ( + _PinnedHTTPSConnection, + _ProviderHTTPResponse, + _validated_public_addresses, + install_provider_transport, +) + + +class _FakeResponse: + """Observable provider response double.""" + + def __init__(self, status: int = 200, body: bytes = b"ok") -> None: + """Initialize status, content, headers, and cleanup state.""" + self.status = status + self.reason = "provider status" + self.headers = {"location": "https://attacker.example/v1"} + self.body = body + self.closed = False + + def read(self, *_args: object, **_kwargs: object) -> bytes: + """Return configured response bytes.""" + return self.body + + def close(self) -> None: + """Record response cleanup.""" + self.closed = True + + def __iter__(self): + """Iterate one response line.""" + return iter([self.body]) + + +class _FakeConnection: + """Observable pinned TLS connection double.""" + + created: list["_FakeConnection"] = [] + responses: dict[str, _FakeResponse] = {} + failing_ips: set[str] = set() + + def __init__( + self, + hostname: str, + pinned_ip: str, + port: int, + timeout: float, + context: ssl.SSLContext, + ) -> None: + """Capture construction inputs for transport assertions.""" + self.hostname = hostname + self.pinned_ip = pinned_ip + self.port = port + self.timeout = timeout + self.context = context + self.request_call: tuple[object, ...] | None = None + self.closed = False + self.created.append(self) + + def request( + self, + method: str, + target: str, + body: bytes | None = None, + headers: dict[str, str] | None = None, + ) -> None: + """Capture a request or simulate one address-level failure.""" + self.request_call = (method, target, body, headers) + if self.pinned_ip in self.failing_ips: + raise OSError("address unavailable") + + def getresponse(self) -> _FakeResponse: + """Return the response configured for this address.""" + return self.responses[self.pinned_ip] + + def close(self) -> None: + """Record connection cleanup.""" + self.closed = True + + +@pytest.fixture(autouse=True) +def _reset_credential_backend(): + """Restore the process-global credential backend after every test.""" + try: + yield + finally: + set_backend(None) + + +def _configured_client() -> tuple[ModelClient, ModelAgent]: + """Build a client and HTTPS agent with a resolvable KV credential.""" + backend = InMemoryCredentialBackend() + backend.set("MODEL_KEY", "test-provider-secret") + set_backend(backend) + client = ModelClient(timeout=11) + client._https_connection_class = _FakeConnection + agent = ModelAgent( + "provider_agent", + "provider-model", + "https://api.example.com:8443/v1", + "MODEL_KEY", + ) + _FakeConnection.created = [] + _FakeConnection.responses = {} + _FakeConnection.failing_ips = set() + return client, agent + + +def _public_dns_answers() -> list[tuple[int, int, int, str, tuple[str, int]]]: + """Return duplicate and distinct globally routable IPv4 answers.""" + return [ + (2, 1, 6, "", ("93.184.216.34", 8443)), + (2, 1, 6, "", ("93.184.216.34", 8443)), + (2, 1, 6, "", ("93.184.216.35", 8443)), + ] + + +def test_transport_installer_is_idempotent() -> None: + """Repeated package initialization cannot wrap validation more than once.""" + validate_method = ModelClient._validate_provider + open_method = ModelClient._open_provider + install_provider_transport(ModelClient) + assert ModelClient._validate_provider is validate_method + assert ModelClient._open_provider is open_method + + +def test_validated_public_addresses_supports_ipv6_and_deduplicates() -> None: + """Address validation returns unique normalized public IPv4 and IPv6 pins.""" + answers = [ + (2, 1, 6, "", ("93.184.216.34", 443)), + (2, 1, 6, "", ("93.184.216.34", 443)), + (10, 1, 6, "", ("2606:2800:220:1:248:1893:25c8:1946", 443, 0, 0)), + ] + with mock.patch( + "contextual_orchestrator.provider_transport.socket.getaddrinfo", + return_value=answers, + ): + assert _validated_public_addresses("api.example.com", 443, "provider_agent") == ( + "93.184.216.34", + "2606:2800:220:1:248:1893:25c8:1946", + ) + + +@pytest.mark.parametrize("unsafe_address", ["127.0.0.1", "100.64.0.1", "224.0.0.1"]) +def test_validated_public_addresses_rejects_unsafe_answer(unsafe_address: str) -> None: + """Any unsafe member of a DNS answer causes validation to fail closed.""" + answers = [ + (2, 1, 6, "", ("93.184.216.34", 443)), + (2, 1, 6, "", (unsafe_address, 443)), + ] + with mock.patch( + "contextual_orchestrator.provider_transport.socket.getaddrinfo", + return_value=answers, + ): + with pytest.raises(RuntimeError, match="non-public address"): + _validated_public_addresses("api.example.com", 443, "provider_agent") + + +def test_validated_public_addresses_rejects_empty_answer() -> None: + """An empty resolver answer cannot silently create an unpinned request.""" + with mock.patch( + "contextual_orchestrator.provider_transport.socket.getaddrinfo", + return_value=[], + ): + with pytest.raises(RuntimeError, match="did not resolve"): + _validated_public_addresses("api.example.com", 443, "provider_agent") + + +def test_validate_then_open_uses_same_dns_answer_without_reresolution() -> None: + """The connected addresses come only from the validation-time DNS answer.""" + client, agent = _configured_client() + with mock.patch( + "contextual_orchestrator.orchestrator.socket.getaddrinfo", + return_value=_public_dns_answers(), + ) as policy_resolver, mock.patch( + "contextual_orchestrator.provider_transport.socket.getaddrinfo", + return_value=_public_dns_answers(), + ) as pin_resolver: + client._validate_provider(agent) + assert policy_resolver.call_count == 1 + assert pin_resolver.call_count == 1 + + _FakeConnection.failing_ips = {"93.184.216.34"} + _FakeConnection.responses = {"93.184.216.35": _FakeResponse(body=b"success")} + request = urllib.request.Request( + "https://api.example.com:8443/v1/chat;mode=fast?trace=yes", + data=b"{}", + headers={"authorization": "Bearer secret"}, + method="POST", + ) + with mock.patch( + "contextual_orchestrator.orchestrator.socket.getaddrinfo", + side_effect=AssertionError("transport must not resolve policy DNS again"), + ), mock.patch( + "contextual_orchestrator.provider_transport.socket.getaddrinfo", + side_effect=AssertionError("transport must not resolve pin DNS again"), + ): + with client._open_provider(request) as response: + assert response.read() == b"success" + + first, second = _FakeConnection.created + assert first.closed is True + assert second.hostname == "api.example.com" + assert second.port == 8443 + assert second.timeout == 11 + assert second.request_call == ( + "POST", + "/v1/chat;mode=fast?trace=yes", + b"{}", + {"Authorization": "Bearer secret", "Connection": "close"}, + ) + assert second.closed is True + + +def test_failed_revalidation_clears_existing_pin() -> None: + """A later unsafe DNS answer cannot reuse a formerly valid cached address.""" + client, agent = _configured_client() + with mock.patch( + "contextual_orchestrator.orchestrator.socket.getaddrinfo", + return_value=_public_dns_answers(), + ), mock.patch( + "contextual_orchestrator.provider_transport.socket.getaddrinfo", + return_value=_public_dns_answers(), + ): + client._validate_provider(agent) + + unsafe = [(2, 1, 6, "", ("127.0.0.1", 8443))] + with mock.patch( + "contextual_orchestrator.orchestrator.socket.getaddrinfo", + return_value=unsafe, + ): + with pytest.raises(RuntimeError, match="non-public address"): + client._validate_provider(agent) + + request = urllib.request.Request("https://api.example.com:8443/v1/chat") + with pytest.raises(RuntimeError, match="no validated address pin"): + client._open_provider(request) + + +def test_redirect_response_is_rejected_without_following_location() -> None: + """A redirect cannot forward provider credentials to another destination.""" + client, agent = _configured_client() + answers = [(2, 1, 6, "", ("93.184.216.34", 8443))] + with mock.patch( + "contextual_orchestrator.orchestrator.socket.getaddrinfo", + return_value=answers, + ), mock.patch( + "contextual_orchestrator.provider_transport.socket.getaddrinfo", + return_value=answers, + ): + client._validate_provider(agent) + + response = _FakeResponse(status=302) + _FakeConnection.responses = {"93.184.216.34": response} + with pytest.raises(urllib.error.HTTPError) as exc_info: + client._open_provider( + urllib.request.Request("https://api.example.com:8443/v1/chat", method="POST") + ) + assert exc_info.value.code == 302 + assert response.closed is True + assert _FakeConnection.created[0].closed is True + assert len(_FakeConnection.created) == 1 + + +def test_all_pinned_addresses_failing_returns_network_error() -> None: + """Exhausting all approved addresses yields one urllib-compatible error.""" + client, agent = _configured_client() + with mock.patch( + "contextual_orchestrator.orchestrator.socket.getaddrinfo", + return_value=_public_dns_answers(), + ), mock.patch( + "contextual_orchestrator.provider_transport.socket.getaddrinfo", + return_value=_public_dns_answers(), + ): + client._validate_provider(agent) + + _FakeConnection.failing_ips = {"93.184.216.34", "93.184.216.35"} + with pytest.raises(urllib.error.URLError, match="address unavailable"): + client._open_provider(urllib.request.Request("https://api.example.com:8443", method="GET")) + assert all(connection.closed for connection in _FakeConnection.created) + + +def test_open_provider_rejects_unsupported_scheme() -> None: + """The low-level transport independently rejects non-HTTP provider schemes.""" + client = ModelClient() + with pytest.raises(RuntimeError, match=r"http\(s\)"): + client._open_provider(urllib.request.Request("file:///etc/passwd")) + + +def test_provider_response_delegates_metadata_iteration_read_and_cleanup() -> None: + """The wrapper preserves response behavior and always closes its connection.""" + response = mock.Mock() + response.status = 200 + response.read.return_value = b"payload" + response.__iter__ = mock.Mock(return_value=iter([b"one", b"two"])) + connection = mock.Mock() + wrapper = _ProviderHTTPResponse(response, connection) + with wrapper as entered: + assert entered is wrapper + assert wrapper.status == 200 + assert wrapper.read(4) == b"payload" + assert list(wrapper) == [b"one", b"two"] + response.read.assert_called_once_with(4) + response.close.assert_called_once_with() + connection.close.assert_called_once_with() + + +def test_provider_response_closes_connection_when_response_close_fails() -> None: + """Connection cleanup survives an exception from response cleanup.""" + response = mock.Mock() + response.close.side_effect = RuntimeError("close failed") + connection = mock.Mock() + wrapper = _ProviderHTTPResponse(response, connection) + with pytest.raises(RuntimeError, match="close failed"): + wrapper.close() + connection.close.assert_called_once_with() + + +def test_pinned_https_connection_dials_ip_and_preserves_sni() -> None: + """The direct socket uses the pin while TLS verifies the original hostname.""" + raw_socket = mock.Mock() + wrapped_socket = object() + context = mock.Mock() + context.wrap_socket.return_value = wrapped_socket + with mock.patch( + "contextual_orchestrator.provider_transport.socket.create_connection", + return_value=raw_socket, + ) as create_connection: + connection = _PinnedHTTPSConnection( + "api.example.com", + "93.184.216.34", + 443, + 7.0, + context, + ) + connection.connect() + create_connection.assert_called_once_with(("93.184.216.34", 443), 7.0, None) + context.wrap_socket.assert_called_once_with(raw_socket, server_hostname="api.example.com") + assert connection.sock is wrapped_socket + + +def test_pinned_https_connection_closes_socket_when_tls_setup_fails() -> None: + """A TLS setup failure cannot leak the already-connected raw socket.""" + raw_socket = mock.Mock() + context = mock.Mock() + context.wrap_socket.side_effect = ssl.SSLError("handshake failed") + with mock.patch( + "contextual_orchestrator.provider_transport.socket.create_connection", + return_value=raw_socket, + ): + connection = _PinnedHTTPSConnection( + "api.example.com", + "93.184.216.34", + 443, + 7.0, + context, + ) + with pytest.raises(ssl.SSLError, match="handshake failed"): + connection.connect() + raw_socket.close.assert_called_once_with() From 0d183be217c70da0507ee1345b9b08a6d650321f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 4 Aug 2026 12:19:09 +0900 Subject: [PATCH 09/10] docs: record DNS-pinned provider transport --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index dca687bc6..3c986a728 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and ### 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. From b9163f4e088318b3a9d4498868639993845567f5 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 4 Aug 2026 12:20:52 +0900 Subject: [PATCH 10/10] test(security): use one shared DNS seam --- tests/test_provider_address_pinning.py | 24 ++++-------------------- 1 file changed, 4 insertions(+), 20 deletions(-) diff --git a/tests/test_provider_address_pinning.py b/tests/test_provider_address_pinning.py index ee426c287..ac0af01cb 100644 --- a/tests/test_provider_address_pinning.py +++ b/tests/test_provider_address_pinning.py @@ -182,15 +182,11 @@ def test_validate_then_open_uses_same_dns_answer_without_reresolution() -> None: """The connected addresses come only from the validation-time DNS answer.""" client, agent = _configured_client() with mock.patch( - "contextual_orchestrator.orchestrator.socket.getaddrinfo", - return_value=_public_dns_answers(), - ) as policy_resolver, mock.patch( "contextual_orchestrator.provider_transport.socket.getaddrinfo", return_value=_public_dns_answers(), - ) as pin_resolver: + ) as resolver: client._validate_provider(agent) - assert policy_resolver.call_count == 1 - assert pin_resolver.call_count == 1 + assert resolver.call_count == 2 _FakeConnection.failing_ips = {"93.184.216.34"} _FakeConnection.responses = {"93.184.216.35": _FakeResponse(body=b"success")} @@ -201,11 +197,8 @@ def test_validate_then_open_uses_same_dns_answer_without_reresolution() -> None: method="POST", ) with mock.patch( - "contextual_orchestrator.orchestrator.socket.getaddrinfo", - side_effect=AssertionError("transport must not resolve policy DNS again"), - ), mock.patch( "contextual_orchestrator.provider_transport.socket.getaddrinfo", - side_effect=AssertionError("transport must not resolve pin DNS again"), + side_effect=AssertionError("transport must not resolve DNS again"), ): with client._open_provider(request) as response: assert response.read() == b"success" @@ -228,9 +221,6 @@ def test_failed_revalidation_clears_existing_pin() -> None: """A later unsafe DNS answer cannot reuse a formerly valid cached address.""" client, agent = _configured_client() with mock.patch( - "contextual_orchestrator.orchestrator.socket.getaddrinfo", - return_value=_public_dns_answers(), - ), mock.patch( "contextual_orchestrator.provider_transport.socket.getaddrinfo", return_value=_public_dns_answers(), ): @@ -238,7 +228,7 @@ def test_failed_revalidation_clears_existing_pin() -> None: unsafe = [(2, 1, 6, "", ("127.0.0.1", 8443))] with mock.patch( - "contextual_orchestrator.orchestrator.socket.getaddrinfo", + "contextual_orchestrator.provider_transport.socket.getaddrinfo", return_value=unsafe, ): with pytest.raises(RuntimeError, match="non-public address"): @@ -254,9 +244,6 @@ def test_redirect_response_is_rejected_without_following_location() -> None: client, agent = _configured_client() answers = [(2, 1, 6, "", ("93.184.216.34", 8443))] with mock.patch( - "contextual_orchestrator.orchestrator.socket.getaddrinfo", - return_value=answers, - ), mock.patch( "contextual_orchestrator.provider_transport.socket.getaddrinfo", return_value=answers, ): @@ -278,9 +265,6 @@ def test_all_pinned_addresses_failing_returns_network_error() -> None: """Exhausting all approved addresses yields one urllib-compatible error.""" client, agent = _configured_client() with mock.patch( - "contextual_orchestrator.orchestrator.socket.getaddrinfo", - return_value=_public_dns_answers(), - ), mock.patch( "contextual_orchestrator.provider_transport.socket.getaddrinfo", return_value=_public_dns_answers(), ):