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)