From cce8614875ef6db204d6403cc2ab3fa75e210f47 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Sun, 12 Jul 2026 02:54:12 -0600 Subject: [PATCH] feat(spp): route confidential egress over the attested channel Wire the confidential verifier hook to the RA-TLS transport, add the attested-channel pool and loopback forwarder, and keep readiness probes state-only. Route local/OpenHands egress through the forwarder, align confidential lane gating, and update fail-closed tests around the new transport seam. --- solstone/apps/thinking/routes.py | 10 +- solstone/apps/thinking/static/thinking.js | 1 + .../tests/test_providers_payload_extended.py | 29 ++ solstone/think/models.py | 42 ++- solstone/think/providers/local.py | 12 +- solstone/think/providers/local_endpoint.py | 9 +- solstone/think/providers/openhands.py | 4 +- solstone/think/services/spp_transport.py | 322 ++++++++++++++++++ tests/services/test_spp_transport.py | 190 +++++++++++ tests/test_local.py | 161 +++++++++ tests/test_local_endpoint.py | 24 ++ tests/test_no_implicit_cloud.py | 76 +++-- 12 files changed, 833 insertions(+), 47 deletions(-) create mode 100644 solstone/think/services/spp_transport.py create mode 100644 tests/services/test_spp_transport.py diff --git a/solstone/apps/thinking/routes.py b/solstone/apps/thinking/routes.py index 5ca3e1e63..2e2919373 100644 --- a/solstone/apps/thinking/routes.py +++ b/solstone/apps/thinking/routes.py @@ -1020,10 +1020,12 @@ def _lane_provider(request_data: dict[str, Any]) -> str | Any: endpoint = resolve_local_endpoint() local_endpoint_configured = not endpoint.is_bundled if lane == "confidential": - return error_response( - INVALID_OPERATION_FOR_STATE, - detail="confidential lane activation must use the confidential enable flow.", - ) + if spp.confidential_provenance() is None: + return error_response( + INVALID_OPERATION_FOR_STATE, + detail="confidential lane activation must use the confidential enable flow.", + ) + return "local" if lane == "local": if local_endpoint_configured: return error_response( diff --git a/solstone/apps/thinking/static/thinking.js b/solstone/apps/thinking/static/thinking.js index 64c1a28ee..f5012c258 100644 --- a/solstone/apps/thinking/static/thinking.js +++ b/solstone/apps/thinking/static/thinking.js @@ -418,6 +418,7 @@ function laneIsUsable(lane) { if (lane === 'byo') return byoIsUsable(); + if (lane === 'confidential') return confidentialProvenancePresent(); if (lane === 'local') return localIsReady() && !localEndpointConfigured(); return false; } diff --git a/solstone/apps/thinking/tests/test_providers_payload_extended.py b/solstone/apps/thinking/tests/test_providers_payload_extended.py index 3e7f598ff..5a9d8ce65 100644 --- a/solstone/apps/thinking/tests/test_providers_payload_extended.py +++ b/solstone/apps/thinking/tests/test_providers_payload_extended.py @@ -589,6 +589,7 @@ def test_lane_switch_to_confidential_rejects_without_config_write( ): client, journal_path = settings_client_with_journal config_path = journal_path / "config" / "journal.json" + config = json.loads(config_path.read_text()) before = config_path.read_bytes() response = client.put( @@ -605,6 +606,34 @@ def test_lane_switch_to_confidential_rejects_without_config_write( ) assert config_path.read_bytes() == before + config["providers"]["local"] = { + "endpoint_url": "http://host.test:8080/v1", + "served_model_id": "served-model", + } + config.setdefault("services", {})["confidential"] = { + "enabled_at": "2026-05-24T00:00:00Z", + "account_id": "acct-confidential", + "endpoint_url": "http://host.test:8080", + "served_model_id": "served-model", + "credential_created_at": "2026-05-24T00:00:00Z", + "credential_fingerprint_sha256": "fingerprint", + "prior_generate_provider": "google", + "prior_cogitate_provider": "openai", + "prior_local_endpoint": None, + } + _write_config(journal_path, config) + + response = client.put( + "/app/thinking/api/providers", + json={"lane": "confidential"}, + ) + + assert response.status_code == 200 + payload = response.get_json() + assert payload["active_lane"]["lane"] == "confidential" + assert payload["generate"]["provider"] == "local" + assert payload["cogitate"]["provider"] == "local" + def test_lane_for_provider_derives_confidential_from_local_endpoint_provenance(): assert ( diff --git a/solstone/think/models.py b/solstone/think/models.py index 9e68b381a..abf93a1a6 100644 --- a/solstone/think/models.py +++ b/solstone/think/models.py @@ -318,6 +318,18 @@ class AttestationStaleError(AttestationNotVerifiedError): _CONFIDENTIAL_ATTESTATION_VERIFIER: Callable[[dict[str, Any]], None] | None = None +def _confidential_attestation_verifier() -> Callable[[dict[str, Any]], None]: + global _CONFIDENTIAL_ATTESTATION_VERIFIER + + if _CONFIDENTIAL_ATTESTATION_VERIFIER is None: + from solstone.think.services.spp_transport import ( + verify_confidential_attestation, + ) + + _CONFIDENTIAL_ATTESTATION_VERIFIER = verify_confidential_attestation + return _CONFIDENTIAL_ATTESTATION_VERIFIER + + # --------------------------------------------------------------------------- # Prompt context discovery # @@ -1033,27 +1045,25 @@ def _raise_if_confidential_unverified() -> None: block = get_config().get("services", {}).get("confidential") if not isinstance(block, dict): return - verifier = _CONFIDENTIAL_ATTESTATION_VERIFIER - if verifier is not None: - verifier(block) - return - raise AttestationNotVerifiedError() + _confidential_attestation_verifier()(block) def confidential_egress_blocked() -> bool: - """Return whether confidential egress is currently gated closed. - - Boolean twin of _raise_if_confidential_unverified(): True when a - services.confidential block is present and attestation has not verified - (this lode: verifier unwired -> always blocked). Callers that must not - reach the confidential endpoint (readiness probes) consult this. - """ + """Return whether confidential egress is currently gated closed.""" - try: - _raise_if_confidential_unverified() + block = get_config().get("services", {}).get("confidential") + if not isinstance(block, dict): return False - except AttestationNotVerifiedError: - return True + + from datetime import datetime, timezone + + from solstone.think.services import spp + + state = spp.get_attestation_state() + return ( + state.session is None + or state.session.status(datetime.now(timezone.utc)) != "verified" + ) def get_model_provider(model: str) -> str: diff --git a/solstone/think/providers/local.py b/solstone/think/providers/local.py index ac0dab8e9..8708bcdef 100644 --- a/solstone/think/providers/local.py +++ b/solstone/think/providers/local.py @@ -632,6 +632,8 @@ def run_generate( import httpx + from solstone.think.services.spp_transport import confidential_egress_base_url + post_kwargs: dict[str, Any] = { "json": body, "timeout": timeout_s or _DEFAULT_TIMEOUT, @@ -639,8 +641,9 @@ def run_generate( if endpoint.credential: post_kwargs["headers"] = {"Authorization": f"Bearer {endpoint.credential}"} try: + base_url = confidential_egress_base_url(endpoint.base_url) response = httpx.post( - f"{endpoint.base_url}/v1/chat/completions", + f"{base_url}/v1/chat/completions", **post_kwargs, ) response.raise_for_status() @@ -688,9 +691,14 @@ async def run_agenerate( if endpoint.credential: post_kwargs["headers"] = {"Authorization": f"Bearer {endpoint.credential}"} try: + from solstone.think.services.spp_transport import ( + confidential_egress_base_url, + ) + + base_url = confidential_egress_base_url(endpoint.base_url) async with httpx.AsyncClient() as client: response = await client.post( - f"{endpoint.base_url}/v1/chat/completions", **post_kwargs + f"{base_url}/v1/chat/completions", **post_kwargs ) response.raise_for_status() return _parse_response(response.json()) diff --git a/solstone/think/providers/local_endpoint.py b/solstone/think/providers/local_endpoint.py index 4b3255cdd..3767ee61c 100644 --- a/solstone/think/providers/local_endpoint.py +++ b/solstone/think/providers/local_endpoint.py @@ -72,12 +72,11 @@ def probe_local_endpoint( ) -> tuple[bool, str | None]: """Run a shallow reachability probe against the configured endpoint root.""" - from solstone.think.models import confidential_egress_blocked + from solstone.think.services.spp_transport import confidential_probe_status - if confidential_egress_blocked(): - # Fail closed: never touch the confidential endpoint before attestation - # verifies. No request reaches unattested hardware. - return False, "attestation_not_yet_verified" + confidential_status = confidential_probe_status() + if confidential_status is not None: + return confidential_status import httpx diff --git a/solstone/think/providers/openhands.py b/solstone/think/providers/openhands.py index 113d4f5ca..f4228d202 100644 --- a/solstone/think/providers/openhands.py +++ b/solstone/think/providers/openhands.py @@ -133,12 +133,14 @@ def _build_llm(provider: str, model: str) -> Any: if provider == "local": from solstone.think.providers.local_endpoint import resolve_local_endpoint + from solstone.think.services.spp_transport import confidential_egress_base_url endpoint = resolve_local_endpoint() if not endpoint.is_bundled: + base_url = confidential_egress_base_url(endpoint.base_url) return LLM( model=f"openai/{endpoint.served_model_id}", - base_url=f"{endpoint.base_url}/v1", + base_url=f"{base_url}/v1", api_key=endpoint.credential or "EMPTY", native_tool_calling=False, timeout=LLM_TIMEOUT_S, diff --git a/solstone/think/services/spp_transport.py b/solstone/think/services/spp_transport.py new file mode 100644 index 000000000..719db19e9 --- /dev/null +++ b/solstone/think/services/spp_transport.py @@ -0,0 +1,322 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""SPP confidential RA-TLS transport and verifier hook.""" + +from __future__ import annotations + +import logging +import re +import secrets +import selectors +import socket +import threading +import time +from datetime import datetime, timezone +from typing import Any +from urllib.parse import urlsplit + +from OpenSSL import SSL + +from solstone.think.models import AttestationFailedError, AttestationStaleError +from solstone.think.providers.nvattest_install import resolve_nvattest_dir +from solstone.think.services import spp +from solstone.think.services.spp_attest.cadence import AttestationSession +from solstone.think.services.spp_attest.composite import verify_composite +from solstone.think.services.spp_attest.ratls.channel import ( + AttestedChannel, + RatlsChannelError, + RatlsEndpoint, + establish_attested_channel, +) +from solstone.think.services.spp_attest.ratls.contract import OWNER_NONCE_BYTES +from solstone.think.services.spp_attest.ratls.verify import RatlsVerificationError + +log = logging.getLogger(__name__) + +POOLED_CHANNEL_MAX_IDLE_S = 120.0 +FORWARDER_SELECT_TIMEOUT_S = 180.0 + +_LOCK = threading.RLock() +_POOL: list[AttestedChannel] = [] +_LISTENER: socket.socket | None = None +_LISTENER_THREAD: threading.Thread | None = None +_FORWARDER_BASE_URL: str | None = None +_CONFIDENTIAL_BLOCK: dict[str, Any] | None = None +_REASON_CODE_RE = re.compile(r"\(([a-z0-9_]+)\)$") + + +def _attestation_failed(reason_code: str, exc: BaseException | None = None) -> None: + log.warning("event=confidential_attestation_rejected reason=%s", reason_code) + spp.record_attestation_failed(reason_code) + error = AttestationFailedError( + f"the confidential attestation transport failed closed ({reason_code})" + ) + if exc is None: + raise error + raise error from exc + + +def _reason_from_attestation_failed(exc: AttestationFailedError) -> str: + match = _REASON_CODE_RE.search(exc.detail.strip()) + if match is not None: + return match.group(1) + return exc.reason_code + + +def _endpoint_from_block(block: dict[str, Any]) -> RatlsEndpoint: + endpoint_url = str(block.get("endpoint_url") or "") + parsed = urlsplit(endpoint_url) + if not parsed.hostname: + raise AttestationFailedError( + "the confidential endpoint configuration is invalid (endpoint_invalid)" + ) + try: + port = parsed.port + except ValueError as exc: + raise AttestationFailedError( + "the confidential endpoint configuration is invalid (endpoint_invalid)" + ) from exc + if port is None: + port = 443 if parsed.scheme == "https" else 80 + return RatlsEndpoint(host=parsed.hostname, port=port) + + +def _transport_live_locked() -> bool: + return ( + _LISTENER is not None + and _LISTENER_THREAD is not None + and _LISTENER_THREAD.is_alive() + and _FORWARDER_BASE_URL is not None + ) + + +def _close_channel(channel: AttestedChannel) -> None: + channel.close() + + +def _teardown_locked() -> None: + global _FORWARDER_BASE_URL, _LISTENER, _LISTENER_THREAD + + listener = _LISTENER + _LISTENER = None + _LISTENER_THREAD = None + _FORWARDER_BASE_URL = None + if listener is not None: + try: + listener.close() + except OSError: + pass + while _POOL: + _close_channel(_POOL.pop()) + + +def teardown_confidential_transport() -> None: + with _LOCK: + _teardown_locked() + + +def _discard_idle_locked(now_monotonic: float) -> None: + kept: list[AttestedChannel] = [] + for channel in _POOL: + if now_monotonic - channel.last_used_monotonic > POOLED_CHANNEL_MAX_IDLE_S: + _close_channel(channel) + else: + kept.append(channel) + _POOL[:] = kept + + +def _start_listener_locked() -> None: + global _FORWARDER_BASE_URL, _LISTENER, _LISTENER_THREAD + + if _transport_live_locked(): + return + listener = socket.socket() + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind(("127.0.0.1", 0)) + listener.listen() + listener.settimeout(1.0) + _LISTENER = listener + port = listener.getsockname()[1] + _FORWARDER_BASE_URL = f"http://127.0.0.1:{port}" + thread = threading.Thread( + target=_accept_loop, + args=(listener,), + name="spp-confidential-forwarder", + daemon=True, + ) + _LISTENER_THREAD = thread + thread.start() + + +def _establish_channel_locked(block: dict[str, Any], now: datetime) -> AttestedChannel: + try: + endpoint = _endpoint_from_block(block) + return establish_attested_channel( + endpoint, + owner_nonce=secrets.token_bytes(OWNER_NONCE_BYTES), + nvattest_dir=resolve_nvattest_dir(block.get("nvattest_dir")), + now=now, + composite_verifier=verify_composite, + monotonic_now=time.monotonic, + ) + except (RatlsChannelError, RatlsVerificationError) as exc: + _teardown_locked() + _attestation_failed(exc.reason_code, exc) + except AttestationFailedError as exc: + _teardown_locked() + _attestation_failed(_reason_from_attestation_failed(exc), exc) + except Exception as exc: + _teardown_locked() + _attestation_failed("unexpected_error", exc) + + +def verify_confidential_attestation(block: dict[str, Any]) -> None: + global _CONFIDENTIAL_BLOCK + + now = datetime.now(timezone.utc) + with _LOCK: + _CONFIDENTIAL_BLOCK = dict(block) + state = spp.get_attestation_state() + if ( + state.session is not None + and state.session.status(now) == "verified" + and _transport_live_locked() + ): + return + if ( + state.session is not None + and state.session.status(now) != "verified" + and _transport_live_locked() + ): + _teardown_locked() + raise AttestationStaleError( + "the confidential attestation cadence lapsed (attestation_stale)" + ) + + channel = _establish_channel_locked(block, now) + _start_listener_locked() + _POOL.append(channel) + spp.record_attestation_verified( + AttestationSession( + verdict=channel.verdict, + started_at=now, + tpm_heartbeat_at=now, + gpu_reattest_at=now, + ) + ) + + +def confidential_egress_base_url(endpoint_base_url: str) -> str: + block = spp.confidential_provenance() + if block is None: + return endpoint_base_url + verify_confidential_attestation(block) + with _LOCK: + if _FORWARDER_BASE_URL is None: + raise AttestationFailedError( + "the confidential forwarder is unavailable (forwarder_unavailable)" + ) + return _FORWARDER_BASE_URL + + +def confidential_probe_status( + now: datetime | None = None, +) -> tuple[bool, str | None] | None: + if spp.confidential_provenance() is None: + return None + now = now or datetime.now(timezone.utc) + state = spp.get_attestation_state() + if state.failure is not None: + return False, "attestation_failed" + if state.session is None: + return False, "attestation_not_yet_verified" + if state.session.status(now) == "stale": + return False, "attestation_stale" + return True, None + + +def _borrow_channel_locked(now_monotonic: float) -> AttestedChannel | None: + _discard_idle_locked(now_monotonic) + if not _POOL: + return None + return _POOL.pop() + + +def _borrow_or_establish_channel_locked(now: datetime) -> AttestedChannel | None: + state = spp.get_attestation_state() + if state.session is None or state.session.status(now) != "verified": + return None + channel = _borrow_channel_locked(time.monotonic()) + if channel is not None: + return channel + if _CONFIDENTIAL_BLOCK is None: + return None + return _establish_channel_locked(_CONFIDENTIAL_BLOCK, now) + + +def _accept_loop(listener: socket.socket) -> None: + while True: + try: + local, _addr = listener.accept() + except TimeoutError: + with _LOCK: + if listener is not _LISTENER: + return + continue + except OSError: + return + thread = threading.Thread( + target=_handle_loopback_connection, + args=(local,), + name="spp-confidential-forwarder-connection", + daemon=True, + ) + thread.start() + + +def _handle_loopback_connection(local: socket.socket) -> None: + channel: AttestedChannel | None = None + try: + with _LOCK: + channel = _borrow_or_establish_channel_locked(datetime.now(timezone.utc)) + if channel is None: + local.close() + return + outcome = _pump(local, channel.tls) + if outcome == "local_closed": + channel.last_used_monotonic = time.monotonic() + with _LOCK: + _POOL.append(channel) + channel = None + finally: + try: + local.close() + except OSError: + pass + if channel is not None: + _close_channel(channel) + + +def _pump(local: socket.socket, tls: SSL.Connection) -> str: + selector = selectors.DefaultSelector() + selector.register(local, selectors.EVENT_READ, (local, tls, "local")) + selector.register(tls, selectors.EVENT_READ, (tls, local, "tls")) + try: + while True: + events = selector.select(timeout=FORWARDER_SELECT_TIMEOUT_S) + if not events: + return "timeout" + for key, _mask in events: + source, destination, label = key.data + try: + chunk = source.recv(65536) + except (SSL.WantReadError, SSL.WantWriteError): + continue + except (OSError, SSL.Error): + return f"{label}_closed" + if not chunk: + return f"{label}_closed" + destination.sendall(chunk) + finally: + selector.close() diff --git a/tests/services/test_spp_transport.py b/tests/services/test_spp_transport.py new file mode 100644 index 000000000..0d706613b --- /dev/null +++ b/tests/services/test_spp_transport.py @@ -0,0 +1,190 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import json +import time +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import Mock + +import pytest + +from solstone.think.models import AttestationStaleError +from solstone.think.services import spp, spp_transport +from solstone.think.services.spp_attest.cadence import AttestationSession + + +class _FakeChannel: + def __init__( + self, verdict: object, last_used_monotonic: float | None = None + ) -> None: + self.verdict = verdict + self.last_used_monotonic = ( + time.monotonic() if last_used_monotonic is None else last_used_monotonic + ) + self.closed = False + + def close(self) -> None: + self.closed = True + + +class _FakeListener: + def __init__(self) -> None: + self.closed = False + + def close(self) -> None: + self.closed = True + + +class _AliveThread: + def is_alive(self) -> bool: + return True + + +@pytest.fixture(autouse=True) +def _clear_transport_state(): + spp.clear_attestation_state() + spp_transport.teardown_confidential_transport() + yield + spp.clear_attestation_state() + spp_transport.teardown_confidential_transport() + + +def _write_confidential_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + config = { + "services": { + "confidential": { + "enabled_at": "2026-05-24T00:00:00Z", + "account_id": "acct-test", + "endpoint_url": "https://spp.example.test:9443", + "served_model_id": "confidential-model", + "credential_created_at": "2026-05-24T00:00:00Z", + "credential_fingerprint_sha256": "fingerprint", + "prior_generate_provider": "google", + "prior_cogitate_provider": "openai", + "prior_local_endpoint": None, + } + } + } + config_dir = tmp_path / "config" + config_dir.mkdir(parents=True) + (config_dir / "journal.json").write_text(json.dumps(config), encoding="utf-8") + return config["services"]["confidential"] + + +def _patch_listener(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_start_listener() -> None: + spp_transport._LISTENER = _FakeListener() + spp_transport._LISTENER_THREAD = _AliveThread() + spp_transport._FORWARDER_BASE_URL = "http://127.0.0.1:4567" + + monkeypatch.setattr(spp_transport, "_start_listener_locked", fake_start_listener) + + +def _stale_session(verdict: object) -> AttestationSession: + old = datetime.now(timezone.utc) - timedelta(hours=2) + return AttestationSession( + verdict=verdict, + started_at=old, + tpm_heartbeat_at=old, + gpu_reattest_at=old, + ) + + +def test_verify_confidential_attestation_reuses_then_raises_stale_once_then_reestablishes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + block = _write_confidential_config(tmp_path, monkeypatch) + _patch_listener(monkeypatch) + verdict = object() + establish = Mock(side_effect=[_FakeChannel(verdict), _FakeChannel(verdict)]) + monkeypatch.setattr(spp_transport, "establish_attested_channel", establish) + + spp_transport.verify_confidential_attestation(block) + spp_transport.verify_confidential_attestation(block) + + assert establish.call_count == 1 + assert spp.get_attestation_state().session is not None + + spp.record_attestation_verified(_stale_session(verdict)) + with pytest.raises(AttestationStaleError): + spp_transport.verify_confidential_attestation(block) + + state = spp.get_attestation_state() + assert state.session is not None + assert state.session.status(datetime.now(timezone.utc)) == "stale" + assert establish.call_count == 1 + + spp_transport.verify_confidential_attestation(block) + + assert establish.call_count == 2 + assert spp.get_attestation_state().session is not None + assert ( + spp.get_attestation_state().session.status(datetime.now(timezone.utc)) + == "verified" + ) + + +def test_confidential_egress_base_url_returns_forwarder_not_configured_endpoint( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + block = _write_confidential_config(tmp_path, monkeypatch) + _patch_listener(monkeypatch) + establish = Mock(return_value=_FakeChannel(object())) + monkeypatch.setattr(spp_transport, "establish_attested_channel", establish) + + base_url = spp_transport.confidential_egress_base_url(block["endpoint_url"]) + + assert base_url == "http://127.0.0.1:4567" + assert base_url != block["endpoint_url"] + assert establish.call_args.args[0].host == "spp.example.test" + assert establish.call_args.args[0].port == 9443 + + +def test_confidential_probe_status_reads_state_without_attestation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _write_confidential_config(tmp_path, monkeypatch) + establish = Mock(side_effect=AssertionError("attestation ceremony attempted")) + monkeypatch.setattr(spp_transport, "establish_attested_channel", establish) + + assert spp_transport.confidential_probe_status() == ( + False, + "attestation_not_yet_verified", + ) + + now = datetime.now(timezone.utc) + spp.record_attestation_verified( + AttestationSession( + verdict=object(), + started_at=now, + tpm_heartbeat_at=now, + gpu_reattest_at=now, + ) + ) + assert spp_transport.confidential_probe_status(now) == (True, None) + + spp.record_attestation_verified(_stale_session(object())) + assert spp_transport.confidential_probe_status() == (False, "attestation_stale") + + spp.record_attestation_failed("gateway_unreachable") + assert spp_transport.confidential_probe_status() == (False, "attestation_failed") + establish.assert_not_called() + + +def test_pooled_channels_older_than_idle_limit_are_discarded() -> None: + stale = _FakeChannel(object(), last_used_monotonic=0.0) + fresh = _FakeChannel(object(), last_used_monotonic=119.0) + spp_transport._POOL[:] = [stale, fresh] + + spp_transport._discard_idle_locked(121.0) + + assert stale.closed is True + assert fresh.closed is False + assert spp_transport._POOL == [fresh] diff --git a/tests/test_local.py b/tests/test_local.py index 89eaedfda..c9434e601 100644 --- a/tests/test_local.py +++ b/tests/test_local.py @@ -813,6 +813,127 @@ def test_run_generate_byo_posts_to_normalized_endpoint_and_skips_connect(monkeyp assert result["text"] == "hello" +def test_run_generate_confidential_posts_to_forwarder_not_configured_endpoint( + monkeypatch, +): + provider = _provider() + + from solstone.think.providers import local_budget + from solstone.think.providers.local_endpoint import LocalEndpoint + + configured_endpoint = "https://spp.example.test" + forwarder = "http://127.0.0.1:4567" + monkeypatch.setattr( + provider, + "resolve_local_endpoint", + lambda: LocalEndpoint( + base_url=configured_endpoint, + served_model_id="confidential-model", + credential="confidential-token", + is_bundled=False, + ), + ) + monkeypatch.setattr( + "solstone.think.services.spp_transport.confidential_egress_base_url", + lambda base_url: forwarder if base_url == configured_endpoint else base_url, + ) + monkeypatch.setattr(local_budget, "count_tokens", lambda *_args, **_kwargs: 1) + monkeypatch.setattr( + "solstone.think.providers.local_server.connect", + lambda: (_ for _ in ()).throw(AssertionError("connect not expected")), + ) + captured = {} + + class Response: + def raise_for_status(self): + return None + + def json(self): + return { + "choices": [ + { + "message": {"content": "hello"}, + "finish_reason": "stop", + } + ], + } + + def fake_post(url, **kwargs): + captured.update({"url": url, **kwargs}) + return Response() + + import httpx + + monkeypatch.setattr(httpx, "post", fake_post) + + result = provider.run_generate("hello", model=LOCAL_MODEL) + + assert result["text"] == "hello" + assert captured["url"] == f"{forwarder}/v1/chat/completions" + assert configured_endpoint not in captured["url"] + + +def test_run_agenerate_confidential_posts_to_forwarder_not_configured_endpoint( + monkeypatch, +): + provider = _provider() + + from solstone.think.providers.local_endpoint import LocalEndpoint + + configured_endpoint = "https://spp.example.test" + forwarder = "http://127.0.0.1:4567" + monkeypatch.setattr( + provider, + "resolve_local_endpoint", + lambda: LocalEndpoint( + base_url=configured_endpoint, + served_model_id="confidential-model", + credential="confidential-token", + is_bundled=False, + ), + ) + monkeypatch.setattr( + "solstone.think.services.spp_transport.confidential_egress_base_url", + lambda base_url: forwarder if base_url == configured_endpoint else base_url, + ) + captured = {} + + class Response: + def raise_for_status(self): + return None + + def json(self): + return { + "choices": [ + { + "message": {"content": "hello"}, + "finish_reason": "stop", + } + ], + } + + class AsyncClient: + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + async def post(self, url, **kwargs): + captured.update({"url": url, **kwargs}) + return Response() + + import httpx + + monkeypatch.setattr(httpx, "AsyncClient", AsyncClient) + + result = asyncio.run(provider.run_agenerate("hello", model=LOCAL_MODEL)) + + assert result["text"] == "hello" + assert captured["url"] == f"{forwarder}/v1/chat/completions" + assert configured_endpoint not in captured["url"] + + def test_run_generate_byo_body_omits_bundled_qwen_sampling(monkeypatch): provider = _provider() monkeypatch.setattr(provider, "resolve_local_endpoint", _byo_endpoint) @@ -1314,6 +1435,46 @@ def test_openhands_local_byo_llm_kwargs(monkeypatch, credential, expected_key): assert "max_input_tokens" not in captured +def test_openhands_local_confidential_llm_uses_forwarder(monkeypatch): + from solstone.think.providers import local_endpoint, openhands + + configured_endpoint = "https://spp.example.test" + forwarder = "http://127.0.0.1:4567" + captured = {} + + class FakeLLM: + def __init__(self, **kwargs): + captured.update(kwargs) + + sdk_module = types.ModuleType("openhands.sdk") + sdk_module.LLM = FakeLLM + monkeypatch.setitem(sys.modules, "openhands.sdk", sdk_module) + monkeypatch.setattr( + local_endpoint, + "resolve_local_endpoint", + lambda: local_endpoint.LocalEndpoint( + base_url=configured_endpoint, + served_model_id="confidential-model", + credential="confidential-token", + is_bundled=False, + ), + ) + monkeypatch.setattr( + "solstone.think.services.spp_transport.confidential_egress_base_url", + lambda base_url: forwarder if base_url == configured_endpoint else base_url, + ) + monkeypatch.setattr( + "solstone.think.providers.local_server.connect", + lambda: (_ for _ in ()).throw(AssertionError("connect not expected")), + ) + + llm = openhands._build_llm("local", LOCAL_MODEL) + + assert isinstance(llm, FakeLLM) + assert captured["base_url"] == f"{forwarder}/v1" + assert configured_endpoint not in captured["base_url"] + + def test_local_context_window_split_floor_vs_tier(): import inspect diff --git a/tests/test_local_endpoint.py b/tests/test_local_endpoint.py index 16ea3cd5a..bcf74fcd5 100644 --- a/tests/test_local_endpoint.py +++ b/tests/test_local_endpoint.py @@ -97,6 +97,30 @@ def test_probe_local_endpoint_treats_any_response_as_reachable(monkeypatch): assert calls == [("http://h:8080", 0.2)] +def test_probe_local_endpoint_skips_get_for_confidential_status(monkeypatch): + import httpx + + def fake_get(url, timeout): + raise AssertionError("endpoint probe attempted") + + monkeypatch.setattr(httpx, "get", fake_get) + monkeypatch.setattr( + "solstone.think.services.spp_transport.confidential_probe_status", + lambda: (False, "attestation_not_yet_verified"), + ) + endpoint = local_endpoint.LocalEndpoint( + base_url="https://spp.example.test", + served_model_id="model", + credential=None, + is_bundled=False, + ) + + assert local_endpoint.probe_local_endpoint(endpoint) == ( + False, + "attestation_not_yet_verified", + ) + + @pytest.mark.parametrize("exc", ["connect", "timeout"]) def test_probe_local_endpoint_reports_transport_failures(monkeypatch, exc): import httpx diff --git a/tests/test_no_implicit_cloud.py b/tests/test_no_implicit_cloud.py index 6e4541857..ee4b77f76 100644 --- a/tests/test_no_implicit_cloud.py +++ b/tests/test_no_implicit_cloud.py @@ -29,6 +29,17 @@ from solstone.think.models import ( from solstone.think.providers import get_provider_module +@pytest.fixture(autouse=True) +def _clear_confidential_transport_state(): + from solstone.think.services import spp, spp_transport + + spp.clear_attestation_state() + spp_transport.teardown_confidential_transport() + yield + spp.clear_attestation_state() + spp_transport.teardown_confidential_transport() + + def _empty_journal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) for key in ("GOOGLE_API_KEY", "ANTHROPIC_API_KEY", "OPENAI_API_KEY"): @@ -93,8 +104,27 @@ def _confidential_config(*, provider_pins: bool = True) -> dict: return config -def _assert_attestation_not_verified(exc: AttestationNotVerifiedError) -> None: - assert exc.reason_code == "attestation_not_yet_verified" +def _install_failing_confidential_transport( + monkeypatch: pytest.MonkeyPatch, + reason_code: str = "gateway_unreachable", +) -> Mock: + from solstone.think.services import spp, spp_transport + from solstone.think.services.spp_attest.ratls.channel import RatlsChannelError + + spp.clear_attestation_state() + spp_transport.teardown_confidential_transport() + monkeypatch.setattr(models, "_CONFIDENTIAL_ATTESTATION_VERIFIER", None) + establish = Mock(side_effect=RatlsChannelError(reason_code)) + monkeypatch.setattr(spp_transport, "establish_attested_channel", establish) + return establish + + +def _assert_attestation_failed( + exc: AttestationFailedError, + reason_code: str = "gateway_unreachable", +) -> None: + assert exc.reason_code == "attestation_failed" + assert f"({reason_code})" in exc.detail def test_unconfigured_journal_resolves_to_no_brain(tmp_path, monkeypatch): @@ -128,29 +158,30 @@ def test_confidential_generate_stops_before_any_provider_dispatch( ): _empty_journal(tmp_path, monkeypatch) _write_journal_config(tmp_path, _confidential_config()) - monkeypatch.setattr(models, "_CONFIDENTIAL_ATTESTATION_VERIFIER", None) + establish = _install_failing_confidential_transport(monkeypatch) mocks = _cloud_call_mocks(monkeypatch) httpx_post = Mock(side_effect=AssertionError("local endpoint call attempted")) httpx_get = Mock(side_effect=AssertionError("endpoint probe attempted")) monkeypatch.setattr("httpx.post", httpx_post) monkeypatch.setattr("httpx.get", httpx_get) - with pytest.raises(AttestationNotVerifiedError) as generate_exc: + with pytest.raises(AttestationFailedError) as generate_exc: models.generate("hello", "any.context") - _assert_attestation_not_verified(generate_exc.value) + _assert_attestation_failed(generate_exc.value) - with pytest.raises(AttestationNotVerifiedError) as result_exc: + with pytest.raises(AttestationFailedError) as result_exc: models.generate_with_result("hello", "any.context") - _assert_attestation_not_verified(result_exc.value) + _assert_attestation_failed(result_exc.value) - with pytest.raises(AttestationNotVerifiedError) as async_exc: + with pytest.raises(AttestationFailedError) as async_exc: asyncio.run(models.agenerate("hello", "any.context")) - _assert_attestation_not_verified(async_exc.value) + _assert_attestation_failed(async_exc.value) for mock in mocks: mock.assert_not_called() httpx_post.assert_not_called() httpx_get.assert_not_called() + assert establish.call_count == 3 def test_confidential_cogitate_stops_before_any_provider_dispatch( @@ -159,7 +190,7 @@ def test_confidential_cogitate_stops_before_any_provider_dispatch( ): _empty_journal(tmp_path, monkeypatch) _write_journal_config(tmp_path, _confidential_config()) - monkeypatch.setattr(models, "_CONFIDENTIAL_ATTESTATION_VERIFIER", None) + establish = _install_failing_confidential_transport(monkeypatch) mocks = _cloud_call_mocks(monkeypatch) build_llm = Mock(side_effect=AssertionError("llm build attempted")) monkeypatch.setattr("solstone.think.providers.openhands._build_llm", build_llm) @@ -168,7 +199,7 @@ def test_confidential_cogitate_stops_before_any_provider_dispatch( monkeypatch.setattr("httpx.post", httpx_post) monkeypatch.setattr("httpx.get", httpx_get) - with pytest.raises(AttestationNotVerifiedError) as exc_info: + with pytest.raises(AttestationFailedError) as exc_info: asyncio.run( talents._execute_with_tools( {"provider": "google"}, @@ -176,12 +207,13 @@ def test_confidential_cogitate_stops_before_any_provider_dispatch( ) ) - _assert_attestation_not_verified(exc_info.value) + _assert_attestation_failed(exc_info.value) build_llm.assert_not_called() for mock in mocks: mock.assert_not_called() httpx_post.assert_not_called() httpx_get.assert_not_called() + establish.assert_called_once() def test_confidential_readiness_probe_fails_closed_before_endpoint_get( @@ -198,7 +230,6 @@ def test_confidential_readiness_probe_fails_closed_before_endpoint_get( "credential": "confidential-credential", } _write_journal_config(tmp_path, config) - monkeypatch.setattr(models, "_CONFIDENTIAL_ATTESTATION_VERIFIER", None) httpx_get = Mock(side_effect=AssertionError("endpoint probe attempted")) monkeypatch.setattr("httpx.get", httpx_get) @@ -219,25 +250,31 @@ def test_confidential_readiness_probe_fails_closed_before_endpoint_get( def test_confidential_attestation_error_is_non_retryable(tmp_path, monkeypatch): _empty_journal(tmp_path, monkeypatch) _write_journal_config(tmp_path, _confidential_config()) - monkeypatch.setattr(models, "_CONFIDENTIAL_ATTESTATION_VERIFIER", None) + establish = _install_failing_confidential_transport(monkeypatch) mocks = _cloud_call_mocks(monkeypatch) assert talents._is_retryable_error(AttestationNotVerifiedError()) is False for exc in (AttestationFailedError("x"), AttestationStaleError("x")): assert talents._is_retryable_error(exc) is False assert talents._should_fallback(exc) is False - assert models._CONFIDENTIAL_ATTESTATION_VERIFIER is None + from solstone.think.services.spp_transport import verify_confidential_attestation - with pytest.raises(AttestationNotVerifiedError): + assert ( + models._confidential_attestation_verifier() is verify_confidential_attestation + ) + + with pytest.raises(AttestationFailedError) as exc_info: asyncio.run( talents._execute_with_tools( {"provider": "google", "backup": "anthropic"}, lambda _event: None, ) ) + _assert_attestation_failed(exc_info.value) for mock in mocks: mock.assert_not_called() + establish.assert_called_once() def test_confidential_gate_keys_on_provenance_not_provider_resolution( @@ -248,18 +285,19 @@ def test_confidential_gate_keys_on_provenance_not_provider_resolution( config = _confidential_config(provider_pins=False) config["env"] = {"GOOGLE_API_KEY": "stray-google-key"} _write_journal_config(tmp_path, config) - monkeypatch.setattr(models, "_CONFIDENTIAL_ATTESTATION_VERIFIER", None) + establish = _install_failing_confidential_transport(monkeypatch) mocks = _cloud_call_mocks(monkeypatch) httpx_post = Mock(side_effect=AssertionError("local endpoint call attempted")) monkeypatch.setattr("httpx.post", httpx_post) - with pytest.raises(AttestationNotVerifiedError) as exc_info: + with pytest.raises(AttestationFailedError) as exc_info: models.generate("hello", "any.context") - _assert_attestation_not_verified(exc_info.value) + _assert_attestation_failed(exc_info.value) for mock in mocks: mock.assert_not_called() httpx_post.assert_not_called() + establish.assert_called_once() def test_confidential_stt_chokepoint_blocks_cloud_audio_egress( -- 2.54.0