From fdb5bd99fe6268deda0a5afb2bd4665ca3276f0f Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Thu, 30 Jul 2026 16:23:58 -0600 Subject: [PATCH] test(observer): bind fixtures for contract recording Shared fixtures and contract recording now need the same verified device proof as real observer registration. Seed bound observer records with matching ledger entries, stamp contract recorder requests with request-scoped identity, and regenerate the observer client bundle manifest without changing reason-code sets. --- .../observer-client-contract/manifest.json | 12 ++-- solstone/convey/contract/observer_bundle.py | 2 +- .../contract/observer_bundle_recording.py | 71 ++++++++++++++++--- solstone/convey/root.py | 4 +- tests/fixtures/seed_observers.py | 19 +++++ tests/test_openapi_contract.py | 38 +++++++++- tests/test_openapi_schemathesis.py | 41 ++++++++++- 7 files changed, 164 insertions(+), 23 deletions(-) diff --git a/docs/openapi/observer-client-contract/manifest.json b/docs/openapi/observer-client-contract/manifest.json index 04e1df5e2..b284cbc49 100644 --- a/docs/openapi/observer-client-contract/manifest.json +++ b/docs/openapi/observer-client-contract/manifest.json @@ -14,7 +14,7 @@ } ], "bundle_schema_identity": "solstone.observer-client-contract-bundle.schema.v1", - "bundle_semver": "5.0.0", + "bundle_semver": "5.0.1", "component_closure": [ "CallosumEvent", "Error", @@ -55,13 +55,13 @@ "id": "bundle.projection_builder", "path": "solstone/convey/contract/observer_bundle.py", "role": "projection_builder", - "sha256": "456350b59b2e8115658403a06d780f78aad1eea46aa817693723179d3c5a129a" + "sha256": "b23ab1c682b251c48c288f25619185c814bff8adf8a3a558fa9727be53e64e97" }, { "id": "bundle.recording", "path": "solstone/convey/contract/observer_bundle_recording.py", "role": "fixture_vector_builder", - "sha256": "a41f0209cfd7a2d8ce5b1f6a8230a8d71d0813082894cd5c079c2b47599ef9f7" + "sha256": "52116eef254774af3144541f0b19ebfca24a62c6c2498ff350535f1afe0f7ba4" }, { "id": "extension.observer_sse_error_frame", @@ -133,13 +133,13 @@ "id": "producer.observer_routes", "path": "solstone/apps/observer/routes.py", "role": "producer", - "sha256": "1343be151811874e0e62760ea97d19d44be9b8eede3b2ed04ce52b86c94bf2f1" + "sha256": "f494608a544a51b7f4161b7fa2cad87fdf44b3bea99ae08bce583661a83233cc" }, { "id": "producer.observer_utils", "path": "solstone/apps/observer/utils.py", "role": "producer", - "sha256": "6399f8f04a83b24eef964e431c87fb99b5f75507675b4faa41478e466f53b8dc" + "sha256": "30b2ac11dbdf070434de2cfa68d7c461ae9d6b79d94715580d74c1169f9b14c9" }, { "id": "producer.protocol", @@ -151,7 +151,7 @@ "id": "producer.root_sse", "path": "solstone/convey/root.py", "role": "producer", - "sha256": "5d3f18c7fd43354412c3402cf19d348beb1437cfea5e00f3bc53f28a908246cd" + "sha256": "66aa7dda0ab358e67a2a36650639ffcb61dbafabee196d617cd9422287e2021a" }, { "id": "reason_codes", diff --git a/solstone/convey/contract/observer_bundle.py b/solstone/convey/contract/observer_bundle.py index 1496259bd..5c3d26ae4 100644 --- a/solstone/convey/contract/observer_bundle.py +++ b/solstone/convey/contract/observer_bundle.py @@ -20,7 +20,7 @@ from solstone.convey.contract.assemble import CALLOSUM_REGISTRY, build_document from solstone.observe import protocol INITIAL_BUNDLE_SEMVER = "1.0.0" -BUNDLE_SEMVER = "5.0.0" +BUNDLE_SEMVER = "5.0.1" GENERATOR_IDENTITY = "solstone.convey.contract.observer_bundle.v1" BUNDLE_SCHEMA_IDENTITY = "solstone.observer-client-contract-bundle.schema.v1" SCHEMA_DIALECT_URI = "https://json-schema.org/draft/2020-12/schema" diff --git a/solstone/convey/contract/observer_bundle_recording.py b/solstone/convey/contract/observer_bundle_recording.py index b74b7be4b..92e6b36a6 100644 --- a/solstone/convey/contract/observer_bundle_recording.py +++ b/solstone/convey/contract/observer_bundle_recording.py @@ -25,9 +25,15 @@ from solstone.convey.contract.observer_bundle import ( _sha256_text, render_json, ) +from solstone.convey.secure_listener import ConveyIdentity from solstone.observe import protocol from solstone.observe.processing_record import HANDLER_TRANSCRIBE, STATE_EMPTY from solstone.observe.processing_record import SCHEMA as PROCESSING_SCHEMA +from solstone.think.link.auth import AuthorizedClients +from solstone.think.link.paths import authorized_clients_path + +_RECORDING_FINGERPRINT = "sha256:" + ("1" * 64) +_RECORDING_LABEL = "Observer Contract Recorder" def build_fixture_and_vector_payloads( @@ -151,8 +157,15 @@ def _record_behavior_vectors() -> tuple[list[dict[str, Any]], list[dict[str, Any from solstone.apps.observer import routes as observer_routes from solstone.convey import bridge as convey_bridge from solstone.convey import create_app - - with _temporary_attr(observer_routes, "now_ms", lambda: 1700000000000): + from solstone.convey import root as convey_root + + authorized = AuthorizedClients(authorized_clients_path()) + with ( + _temporary_attr(observer_routes, "now_ms", lambda: 1700000000000), + _temporary_attr( + convey_root, "get_authorized_clients", lambda: authorized + ), + ): app = create_app(journal=str(journal.resolve())) app.config["TESTING"] = True client = app.test_client() @@ -164,7 +177,8 @@ def _record_behavior_vectors() -> tuple[list[dict[str, Any]], list[dict[str, Any fixture_id="recorded.auth.bearer.segments", vector_id="observer.auth.bearer", operation_id="observer.ingestSegments", - response=client.get( + response=_observer_get( + client, "/app/observer/ingest/segments/20250103", headers={ "Authorization": f"Bearer {auth_key}", @@ -180,7 +194,8 @@ def _record_behavior_vectors() -> tuple[list[dict[str, Any]], list[dict[str, Any fixture_id="recorded.auth.handle.segments", vector_id="observer.auth.handle", operation_id="observer.ingestSegments", - response=client.get( + response=_observer_get( + client, "/app/observer/ingest/segments/20250103", headers={ protocol.OBSERVER_HANDLE_HEADER: auth_key, @@ -198,7 +213,8 @@ def _record_behavior_vectors() -> tuple[list[dict[str, Any]], list[dict[str, Any fixture_id="recorded.segments.legacy.absent_header", vector_id="observer.ingestSegments.legacy_array.absent_header", operation_id="observer.ingestSegments", - response=client.get( + response=_observer_get( + client, "/app/observer/ingest/segments/20250103", headers={"Authorization": f"Bearer {segments_key}"}, ), @@ -213,7 +229,8 @@ def _record_behavior_vectors() -> tuple[list[dict[str, Any]], list[dict[str, Any "observer.ingestSegments.legacy_array.unparseable_header" ), operation_id="observer.ingestSegments", - response=client.get( + response=_observer_get( + client, "/app/observer/ingest/segments/20250103", headers={ "Authorization": f"Bearer {segments_key}", @@ -229,7 +246,8 @@ def _record_behavior_vectors() -> tuple[list[dict[str, Any]], list[dict[str, Any fixture_id="recorded.segments.v2.envelope", vector_id="observer.ingestSegments.v2_envelope", operation_id="observer.ingestSegments", - response=client.get( + response=_observer_get( + client, "/app/observer/ingest/segments/20250103", headers={ "Authorization": f"Bearer {segments_key}", @@ -377,7 +395,8 @@ def _record_behavior_vectors() -> tuple[list[dict[str, Any]], list[dict[str, Any fixture_id="recorded.segments.submitted_name_omitted", vector_id="observer.ingestSegments.submitted_name_fallback", operation_id="observer.ingestSegments", - response=client.get( + response=_observer_get( + client, "/app/observer/ingest/segments/20250108", headers={ "Authorization": f"Bearer {fallback_key}", @@ -412,7 +431,8 @@ def _record_behavior_vectors() -> tuple[list[dict[str, Any]], list[dict[str, Any fixture_id="recorded.segments.custody_statuses", vector_id="observer.ingestSegments.custody_statuses", operation_id="observer.ingestSegments", - response=client.get( + response=_observer_get( + client, "/app/observer/ingest/segments/20250109", headers={ "Authorization": f"Bearer {custody_key}", @@ -505,7 +525,8 @@ def _record_sse_vectors( root_response.close() sse_key = _register_observer(client, "vector-sse") - observer_response = client.get( + observer_response = _observer_get( + client, "/app/observer/callosum", headers={"Authorization": f"Bearer {sse_key}"}, buffered=False, @@ -542,7 +563,8 @@ def _record_sse_vectors( observer_response.close() error_key = _register_observer(client, "vector-sse-error") - error_response = client.get( + error_response = _observer_get( + client, "/app/observer/callosum", headers={protocol.OBSERVER_HANDLE_HEADER: error_key}, buffered=False, @@ -1053,8 +1075,16 @@ def _fixture_id( def _register_observer(client: Any, name: str) -> str: + AuthorizedClients(authorized_clients_path()).add( + _RECORDING_FINGERPRINT, + _RECORDING_LABEL, + "observer-contract-recorder", + paired_at="2026-01-01T00:00:00Z", + client_label=name, + ) response = client.post( "/app/observer/register", + environ_base={"pl.identity": _recording_pl_identity()}, json={ "hostname": name, "platform": "linux", @@ -1072,6 +1102,24 @@ def _register_observer(client: Any, name: str) -> str: return str(body["key"]) +def _recording_pl_identity() -> ConveyIdentity: + return ConveyIdentity( + mode="pl-via-spl", + fingerprint=_RECORDING_FINGERPRINT, + device_label=_RECORDING_LABEL, + paired_at="2026-01-01T00:00:00Z", + session_id="observer-contract-recorder", + ) + + +def _observer_get(client: Any, path: str, **kwargs: Any) -> Any: + return client.get( + path, + environ_overrides={"pl.identity": _recording_pl_identity()}, + **kwargs, + ) + + def _upload( client: Any, key: str, @@ -1081,6 +1129,7 @@ def _upload( ) -> Any: return client.post( "/app/observer/ingest", + environ_overrides={"pl.identity": _recording_pl_identity()}, headers={"Authorization": f"Bearer {key}"}, data={ "day": day, diff --git a/solstone/convey/root.py b/solstone/convey/root.py index 74c7adb53..9a679a4f1 100644 --- a/solstone/convey/root.py +++ b/solstone/convey/root.py @@ -162,8 +162,8 @@ def _is_trusted_observer_extension_request() -> bool: Origin (and may send ``Sec-Fetch-Site: cross-site``) on its loopback observer POSTs. Grant exactly the pinned extension id, and only on the ``/app/observer`` path namespace, an exemption from the browser - CSRF/Origin checks. The observer route's own direct-loopback gate - (``_is_trusted_register_caller``) remains the real authorization. + CSRF/Origin checks. The observer routes' own device identity checks remain + the real authorization. """ path = request.path if not (path == "/app/observer" or path.startswith("/app/observer/")): diff --git a/tests/fixtures/seed_observers.py b/tests/fixtures/seed_observers.py index 4ab187386..f52bdcc0d 100644 --- a/tests/fixtures/seed_observers.py +++ b/tests/fixtures/seed_observers.py @@ -16,6 +16,8 @@ from pathlib import Path from solstone.apps.observer.utils import list_observers, save_observer from solstone.convey import state +from solstone.think.link.auth import AuthorizedClients +from solstone.think.link.paths import authorized_clients_path from solstone.think.utils import get_journal, now_ms SEEDS: list[tuple[str, int | None]] = [ @@ -30,6 +32,10 @@ def _seed_key(name: str) -> str: return hashlib.sha256(name.encode("utf-8")).hexdigest() +def _seed_fingerprint(name: str) -> str: + return "sha256:" + hashlib.sha256(f"device:{name}".encode("utf-8")).hexdigest() + + def seed_observers() -> int: journal = Path(get_journal()).expanduser().resolve() if not journal.exists() or not journal.is_dir(): @@ -58,12 +64,25 @@ def seed_observers() -> int: current_now = now_ms() written = 0 + authorized = AuthorizedClients(authorized_clients_path()) for name, offset_ms in SEEDS: existing_record = existing_by_name.get(name, {}) last_seen = None if offset_ms is None else current_now - offset_ms + fingerprint = _seed_fingerprint(name) + authorized.add( + fingerprint, + name, + "sandbox-observer-fixture", + paired_at="2026-01-01T00:00:00Z", + client_label=name, + ) record = { "key": _seed_key(name), "name": name, + "device_binding": { + "device": fingerprint, + "kind": "cert", + }, "created_at": existing_record.get("created_at", current_now), "last_seen": last_seen, "last_segment": None, diff --git a/tests/test_openapi_contract.py b/tests/test_openapi_contract.py index 532cf7021..4a896a196 100644 --- a/tests/test_openapi_contract.py +++ b/tests/test_openapi_contract.py @@ -108,15 +108,28 @@ REGISTER_OBSERVER_PAYLOAD = { } PUSH_FINGERPRINT = "sha256:" + ("a" * 64) +OBSERVER_FINGERPRINT = "sha256:" + ("b" * 64) @pytest.fixture -def contract_app(tmp_path: Path): +def contract_app(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): journal = prepare_isolated_journal(tmp_path / "journal") mark_setup_complete(journal) with isolated_app_env(journal): app = create_app(journal=str(journal.resolve())) app.config["TESTING"] = True + from solstone.convey import root as convey_root + from solstone.think.link.auth import AuthorizedClients + from solstone.think.link.paths import authorized_clients_path + + authorized = AuthorizedClients(authorized_clients_path()) + authorized.add( + OBSERVER_FINGERPRINT, + "contract-observer", + "instance-1", + paired_at="2026-05-20T00:00:00Z", + ) + monkeypatch.setattr(convey_root, "get_authorized_clients", lambda: authorized) yield app, app.test_client(), journal @@ -163,13 +176,27 @@ def _assert_structured_error(body: dict[str, Any], document: dict[str, Any]) -> def _register_observer(client) -> str: - response = client.post("/app/observer/register", json=REGISTER_OBSERVER_PAYLOAD) + response = client.post( + "/app/observer/register", + json=REGISTER_OBSERVER_PAYLOAD, + environ_overrides={"pl.identity": _observer_identity()}, + ) assert response.status_code == 200, response.get_data(as_text=True) body = response.get_json() assert isinstance(body, dict) return str(body["key"]) +def _observer_identity() -> ConveyIdentity: + return ConveyIdentity( + mode="pl-via-spl", + fingerprint=OBSERVER_FINGERPRINT, + device_label="contract-observer", + paired_at="2026-05-20T00:00:00Z", + session_id="observer-contract-test", + ) + + def _push_identity() -> ConveyIdentity: return ConveyIdentity( mode="dl", @@ -230,10 +257,12 @@ def test_observer_auth_both_header_forms(contract_app): client.get( "/app/observer/ingest/manifest", headers={"Authorization": f"Bearer {key}"}, + environ_overrides={"pl.identity": _observer_identity()}, ), client.get( "/app/observer/ingest/manifest", headers={"X-Solstone-Observer": key}, + environ_overrides={"pl.identity": _observer_identity()}, ), ] @@ -251,6 +280,7 @@ def test_segments_protocol_version_shape(contract_app): legacy = client.get( "/app/observer/ingest/segments/20250103", headers={"Authorization": f"Bearer {key}"}, + environ_overrides={"pl.identity": _observer_identity()}, ) assert legacy.status_code == 200 assert isinstance(legacy.get_json(), list) @@ -261,6 +291,7 @@ def test_segments_protocol_version_shape(contract_app): "Authorization": f"Bearer {key}", "X-Solstone-Protocol-Version": "2", }, + environ_overrides={"pl.identity": _observer_identity()}, ) assert current.status_code == 200 body = current.get_json() @@ -286,6 +317,7 @@ def test_segment_file_status_enum_matches_live_day_listing(contract_app): "files": (io.BytesIO(b"contract upload"), "audio.flac"), }, content_type="multipart/form-data", + environ_overrides={"pl.identity": _observer_identity()}, ) assert upload.status_code == 200 @@ -295,6 +327,7 @@ def test_segment_file_status_enum_matches_live_day_listing(contract_app): "Authorization": f"Bearer {key}", "X-Solstone-Protocol-Version": "2", }, + environ_overrides={"pl.identity": _observer_identity()}, ) items = listing.get_json()["items"] statuses = [file_info["status"] for item in items for file_info in item["files"]] @@ -315,6 +348,7 @@ def test_multipart_and_json_parsing(contract_app): "files": (io.BytesIO(b"contract upload"), "audio.flac"), }, content_type="multipart/form-data", + environ_overrides={"pl.identity": _observer_identity()}, ) assert upload.status_code != 415 upload_body = upload.get_json() diff --git a/tests/test_openapi_schemathesis.py b/tests/test_openapi_schemathesis.py index 4a1d8fa4e..386d5f437 100644 --- a/tests/test_openapi_schemathesis.py +++ b/tests/test_openapi_schemathesis.py @@ -16,6 +16,7 @@ from schemathesis import GenerationMode import solstone.think.utils as think_utils from solstone.convey import create_app +from solstone.convey.secure_listener.identity import ConveyIdentity from solstone.think.convey_client import resolve_base_url from solstone.think.utils import get_journal from tests._baseline_harness import ( @@ -92,6 +93,7 @@ REGISTER_OBSERVER_PAYLOAD = { "stream_type": "desktop", "version": "1", } +OBSERVER_FINGERPRINT = "sha256:" + ("c" * 64) def _load_schema() -> schemathesis.schemas.BaseSchema: @@ -112,6 +114,16 @@ class ContractTarget: observer_key: str +class _ObserverIdentityMiddleware: + def __init__(self, app: Any) -> None: + self._app = app + + def __call__(self, environ: dict[str, Any], start_response: Any) -> Any: + if environ.get("HTTP_X_SOLSTONE_OBSERVER"): + environ["pl.identity"] = _observer_identity() + return self._app(environ, start_response) + + def _resolved_operation_ids(schema: schemathesis.schemas.BaseSchema) -> set[str]: operation_ids: set[str] = set() for result in schema.get_all_operations(): @@ -120,8 +132,22 @@ def _resolved_operation_ids(schema: schemathesis.schemas.BaseSchema) -> set[str] return operation_ids +def _observer_identity() -> ConveyIdentity: + return ConveyIdentity( + mode="pl-via-spl", + fingerprint=OBSERVER_FINGERPRINT, + device_label="schemathesis-observer", + paired_at="2026-05-20T00:00:00Z", + session_id="observer-schemathesis-test", + ) + + def _register_observer_wsgi(client: Any) -> str: - response = client.post("/app/observer/register", json=REGISTER_OBSERVER_PAYLOAD) + response = client.post( + "/app/observer/register", + json=REGISTER_OBSERVER_PAYLOAD, + environ_overrides={"pl.identity": _observer_identity()}, + ) assert response.status_code == 200, response.get_data(as_text=True) body = response.get_json() assert isinstance(body, dict) @@ -184,6 +210,19 @@ def contract_target(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): with isolated_app_env(journal): app = create_app(journal=str(journal.resolve())) app.config["TESTING"] = True + from solstone.convey import root as convey_root + from solstone.think.link.auth import AuthorizedClients + from solstone.think.link.paths import authorized_clients_path + + authorized = AuthorizedClients(authorized_clients_path()) + authorized.add( + OBSERVER_FINGERPRINT, + "schemathesis-observer", + "instance-1", + paired_at="2026-05-20T00:00:00Z", + ) + monkeypatch.setattr(convey_root, "get_authorized_clients", lambda: authorized) + app.wsgi_app = _ObserverIdentityMiddleware(app.wsgi_app) observer_key = _register_observer_wsgi(app.test_client()) yield ContractTarget(app=app, base_url=None, observer_key=observer_key) -- 2.51.2