From a2d623e98051be8f08b483ee1bb09a8328d567bc Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Thu, 30 Jul 2026 16:21:15 -0600 Subject: [PATCH] feat(observer): bind records to verified devices Observer handles were portable bearer credentials. Store the owning device class and identifier on each observer record, require the bound device in the shared identity resolver, and make register mint records only from verified device proof so the handle becomes only a per-stream selector. The SSE stream now rechecks the bound device on a bounded interval so revocation closes busy streams without waiting for an idle heartbeat. --- solstone/apps/observer/routes.py | 186 ++++++++++++---- solstone/apps/observer/tests/conftest.py | 83 ++++++- .../apps/observer/tests/test_callosum_sse.py | 112 +++++++++- solstone/apps/observer/tests/test_prune.py | 10 + solstone/apps/observer/tests/test_register.py | 55 ++++- .../observer/tests/test_resolve_identity.py | 209 ++++++++++++++++-- solstone/apps/observer/utils.py | 94 +++++++- tests/link/test_wsgi_input_body.py | 5 +- tests/test_bad_media_corpus.py | 38 +++- 9 files changed, 698 insertions(+), 94 deletions(-) diff --git a/solstone/apps/observer/routes.py b/solstone/apps/observer/routes.py index 78de8ce07..d308d2f1a 100644 --- a/solstone/apps/observer/routes.py +++ b/solstone/apps/observer/routes.py @@ -20,6 +20,7 @@ import platform import queue import re import secrets +import time from pathlib import Path from typing import Any @@ -76,6 +77,9 @@ from solstone.think.utils import day_path, iter_segments, now_ms, segment_path from .processing_proof import has_terminal_processing_proof from .share_delete import DELETABLE_SOURCE_STREAMS, delete_source_stream from .utils import ( + DEVICE_BINDING_FIELD, + DEVICE_BINDING_KIND_BROWSER, + DEVICE_BINDING_KIND_CERT, DISPOSITION_RECEIVED_NOT_WRITTEN, MAX_INGEST_SEGMENT_ATTEMPTS, IngestFile, @@ -88,6 +92,7 @@ from .utils import ( get_observers_dir, list_observers, load_history, + observer_device_binding, observer_filename_prefix, pruned_segments, record_ingest_rejection, @@ -117,6 +122,7 @@ KEY_BYTES = 32 ACTIVE_THRESHOLD_MS = 30_000 STALE_THRESHOLD_MS = 120_000 FUTURE_CLOCK_DRIFT_TOLERANCE_MS = 5 * 60 * 1000 +_SSE_DEVICE_RECHECK_SECONDS = 5.0 OBSERVER_STATE_LABELS = { "connected": "connected", @@ -356,45 +362,89 @@ def api_list() -> Any: @observer_bp.route(_OBSERVER_CALLOSUM_SSE_RULE, methods=["GET"]) def callosum_sse() -> Any: """Stream Callosum events to an authenticated observer process.""" - _observer, key_prefix, error = resolve_observer_identity() + observer, key_prefix, error = resolve_observer_identity() if error is not None: return error + binding = observer_device_binding(observer) + bound_device = binding["device"] if binding is not None else "" + bound_kind = binding["kind"] if binding is not None else "" + observer_handle = observer["key"] handle = convey_bridge.register_sse_subscriber(key_prefix) def current_observer() -> dict | None: return ObserverRegistry.singleton().by_prefix(key_prefix) + def current_observer_rejection() -> tuple[Reason, str] | None: + observer_now = current_observer() + if not observer_now: + return AUTH_REQUIRED, "Authorization required" + if observer_now.get("revoked", False): + return PL_REVOKED, "Observer revoked" + if not observer_now.get("enabled", True): + return FEATURE_UNAVAILABLE, "Observer disabled" + return None + + def current_device_rejection() -> tuple[Reason, str] | None: + entry = AuthorizedClients(authorized_clients_path()).get(bound_device) + if entry is None or entry.kind != bound_kind: + return PL_REVOKED, "Paired device revoked" + if ( + bound_kind == DEVICE_BINDING_KIND_BROWSER + and entry.observer_handle != observer_handle + ): + return PL_REVOKED, "Paired device revoked" + return None + + def current_rejection() -> tuple[Reason, str] | None: + return current_observer_rejection() or current_device_rejection() + def generate(): try: + next_heartbeat_at = time.monotonic() + _SSE_HEARTBEAT_SECONDS + next_device_check_at = time.monotonic() + _SSE_DEVICE_RECHECK_SECONDS yield ": heartbeat\n\n" while True: if handle.dropped.is_set(): return + now = time.monotonic() + if now >= next_device_check_at: + rejection = current_rejection() + if rejection is not None: + reason, detail = rejection + yield _sse_error_event(reason, detail=detail) + return + next_device_check_at = now + _SSE_DEVICE_RECHECK_SECONDS + timeout = max( + 0.0, + min(next_heartbeat_at, next_device_check_at) - time.monotonic(), + ) try: - serialized_message = handle.queue.get( - timeout=_SSE_HEARTBEAT_SECONDS - ) + serialized_message = handle.queue.get(timeout=timeout) except queue.Empty: - observer_now = current_observer() - if not observer_now: - yield _sse_error_event( - AUTH_REQUIRED, detail="Authorization required" - ) - return - if observer_now.get("revoked", False): - yield _sse_error_event(PL_REVOKED, detail="Observer revoked") - return - if not observer_now.get("enabled", True): - yield _sse_error_event( - FEATURE_UNAVAILABLE, detail="Observer disabled" - ) - return - yield ": heartbeat\n\n" + now = time.monotonic() + if now >= next_device_check_at: + rejection = current_rejection() + if rejection is not None: + reason, detail = rejection + yield _sse_error_event(reason, detail=detail) + return + next_device_check_at = now + _SSE_DEVICE_RECHECK_SECONDS + if now >= next_heartbeat_at: + yield ": heartbeat\n\n" + next_heartbeat_at = now + _SSE_HEARTBEAT_SECONDS continue if handle.dropped.is_set(): return + now = time.monotonic() + if now >= next_device_check_at: + rejection = current_rejection() + if rejection is not None: + reason, detail = rejection + yield _sse_error_event(reason, detail=detail) + return + next_device_check_at = now + _SSE_DEVICE_RECHECK_SECONDS yield f"data: {serialized_message}\n\n" finally: convey_bridge.unregister_sse_subscriber(handle) @@ -462,30 +512,64 @@ def api_create() -> Any: _REGISTER_REQUIRED_FIELDS = ("platform", "hostname", "stream_type", "version") -def _is_trusted_localhost() -> bool: - """Direct-loopback check for observer-local endpoints.""" - is_localhost = request.remote_addr in ("127.0.0.1", "::1", "localhost") - proxy_headers = ( - request.headers.get("X-Forwarded-For") - or request.headers.get("X-Real-IP") - or request.headers.get("X-Forwarded-Host") - ) - return is_localhost and not proxy_headers - - -def _is_authorized_pl_identity() -> bool: - """Return True when the request arrived through a currently authorized PL.""" +def _authorized_pl_entry(): + """Return the authorized cert-class PL entry for this request, if any.""" identity = getattr(g, "identity", None) if getattr(identity, "mode", None) not in {"pl-direct", "pl-via-spl"}: - return False + return None fingerprint = getattr(identity, "fingerprint", None) if not isinstance(fingerprint, str) or not fingerprint: - return False - return AuthorizedClients(authorized_clients_path()).is_authorized(fingerprint) + return None + entry = AuthorizedClients(authorized_clients_path()).get(fingerprint) + if entry is None or entry.kind != DEVICE_BINDING_KIND_CERT: + return None + return entry + + +def _browser_register_token(data: dict[str, Any]) -> str | None: + if data.get("platform") != "browser" or data.get("stream_type") != "browser": + return None + hostname = data.get("hostname") + if not isinstance(hostname, str): + return None + token = hostname.strip().rsplit("-", 1)[-1] + if not re.fullmatch(r"[0-9a-f]{12}", token): + return None + return token + + +def _pending_browser_entry(data: dict[str, Any]): + token = _browser_register_token(data) + if token is None: + return None + matches = [ + entry + for entry in AuthorizedClients(authorized_clients_path()).snapshot() + if entry.kind == DEVICE_BINDING_KIND_BROWSER + and entry.observer_handle is None + and entry.fingerprint.removeprefix("sha256:").startswith(token) + ] + if len(matches) != 1: + return None + return matches[0] -def _is_trusted_register_caller() -> bool: - return _is_trusted_localhost() or _is_authorized_pl_identity() +def _device_binding_for_entry(entry) -> dict[str, str]: + return {"device": entry.fingerprint, "kind": entry.kind} + + +def _resolve_register_device_binding(data: dict[str, Any]) -> dict[str, str] | None: + entry = _authorized_pl_entry() + if entry is not None: + return _device_binding_for_entry(entry) + entry = _pending_browser_entry(data) + if entry is not None: + return _device_binding_for_entry(entry) + return None + + +def _is_trusted_register_caller(data: dict[str, Any]) -> bool: + return _resolve_register_device_binding(data) is not None def _register_descriptor(record: dict) -> dict: @@ -504,21 +588,23 @@ def _register_descriptor(record: dict) -> dict: def register() -> Any: """Self-register a local or through-link observer and lock stream identity. - The in-handler guard is the sole gate: direct loopback, or a request that - arrived through an authorized PL identity. The route is require_access-exempt - so an observer can register before setup completes. Mints the DL handle, - locks a stream onto the record, and returns the pinned descriptor response. + The in-handler guard is the sole gate: a request must carry an authorized + cert-class PL identity or match one pending browser-class ledger entry. The + route is require_access-exempt so an observer can register before setup + completes. Mints the DL handle, locks a stream onto the record, and returns + the pinned descriptor response. """ - # Register guard FIRST — untrusted callers mint nothing. - if not _is_trusted_register_caller(): + parsed = request.get_json(force=True, silent=True) + data = parsed if isinstance(parsed, dict) else {} + device_binding = _resolve_register_device_binding(data) + + # Register guard runs before field-specific validation; untrusted callers mint nothing. + if device_binding is None: return error_response( LOCAL_REQUEST_ONLY, - detail="Observer registration requires a direct localhost or paired-link request.", + detail="Observer registration requires a verified device pairing.", ) - parsed = request.get_json(force=True, silent=True) - data = parsed if isinstance(parsed, dict) else {} - for field in _REGISTER_REQUIRED_FIELDS: value = data.get(field) if not isinstance(value, str) or not value.strip(): @@ -536,6 +622,11 @@ def register() -> Any: existing = find_oldest_unrevoked_by_name(stream) if existing is not None: + if observer_device_binding(existing) != device_binding: + return error_response( + LOCAL_REQUEST_ONLY, + detail="Observer stream is bound to a different device.", + ) # Idempotent re-register: reuse the prior key/record, refresh the # mutable descriptor fields, preserve key/created_at/stats/last_seen. existing["platform"] = data["platform"].strip() @@ -571,6 +662,7 @@ def register() -> Any: "last_segment_received_at": None, "last_segment_day": None, "enabled": True, + DEVICE_BINDING_FIELD: device_binding, "stats": { "segments_received": 0, "bytes_received": 0, diff --git a/solstone/apps/observer/tests/conftest.py b/solstone/apps/observer/tests/conftest.py index eb6c9bb90..9164abec1 100644 --- a/solstone/apps/observer/tests/conftest.py +++ b/solstone/apps/observer/tests/conftest.py @@ -13,6 +13,7 @@ import json import os import sys from pathlib import Path +from typing import Any import pytest @@ -20,6 +21,8 @@ ROOT = Path(__file__).resolve().parents[4] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) +TEST_PL_FINGERPRINT = "sha256:" + ("c" * 64) + @pytest.fixture(scope="module") def observer_app(): @@ -49,10 +52,80 @@ def observer_env(tmp_path, monkeypatch, observer_app): the Flask test client along with the journal path. """ + from solstone.convey import root as convey_root from solstone.convey import state + from solstone.convey.secure_listener import ConveyIdentity + from solstone.observe.protocol import OBSERVER_HANDLE_HEADER + from solstone.think.link.auth import AuthorizedClients + from solstone.think.link.paths import authorized_clients_path original_journal_root = state.journal_root + def _pl_identity() -> ConveyIdentity: + return ConveyIdentity( + mode="pl-via-spl", + fingerprint=TEST_PL_FINGERPRINT, + device_label="pl-observer", + paired_at="2026-05-20T00:00:00Z", + session_id="session-1", + ) + + class BoundObserverClient: + def __init__(self, client): + self._client = client + + def __getattr__(self, name: str) -> Any: + return getattr(self._client, name) + + def _request_kwargs(self, path: str, kwargs: dict[str, Any]) -> dict[str, Any]: + adjusted = dict(kwargs) + overrides = dict(adjusted.pop("environ_overrides", {}) or {}) + headers = adjusted.get("headers") + if ( + "pl.identity" not in overrides + and path.startswith("/app/observer/") + and path != "/app/observer/register" + and isinstance(headers, dict) + and ("Authorization" in headers or OBSERVER_HANDLE_HEADER in headers) + ): + overrides["pl.identity"] = _pl_identity() + adjusted["environ_overrides"] = overrides + return adjusted + + def _bind_created_observer(self, response) -> None: + if response.status_code != 200: + return + data = response.get_json(silent=True) + key = data.get("key") if isinstance(data, dict) else None + if not isinstance(key, str) or not key: + return + from solstone.apps.observer.utils import load_observer, save_observer + + observer = load_observer(key) + if observer is None: + return + observer["device_binding"] = { + "device": TEST_PL_FINGERPRINT, + "kind": "cert", + } + assert save_observer(observer) + + def post(self, path: str, *args: Any, **kwargs: Any): + response = self._client.post( + path, *args, **self._request_kwargs(path, kwargs) + ) + if path == "/app/observer/api/create": + self._bind_created_observer(response) + return response + + def get(self, path: str, *args: Any, **kwargs: Any): + return self._client.get(path, *args, **self._request_kwargs(path, kwargs)) + + def delete(self, path: str, *args: Any, **kwargs: Any): + return self._client.delete( + path, *args, **self._request_kwargs(path, kwargs) + ) + def _create(): journal = tmp_path / "journal" journal.mkdir() @@ -72,7 +145,15 @@ def observer_env(tmp_path, monkeypatch, observer_app): # Set environment monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) state.journal_root = str(journal) - client = observer_app.test_client() + authorized = AuthorizedClients(authorized_clients_path()) + authorized.add( + TEST_PL_FINGERPRINT, + "pl-observer", + "instance-1", + paired_at="2026-05-20T00:00:00Z", + ) + monkeypatch.setattr(convey_root, "get_authorized_clients", lambda: authorized) + client = BoundObserverClient(observer_app.test_client()) class Env: def __init__(self): diff --git a/solstone/apps/observer/tests/test_callosum_sse.py b/solstone/apps/observer/tests/test_callosum_sse.py index 3862b8f25..5a1425f96 100644 --- a/solstone/apps/observer/tests/test_callosum_sse.py +++ b/solstone/apps/observer/tests/test_callosum_sse.py @@ -13,13 +13,20 @@ import pytest import solstone.apps.observer.routes as routes_module import solstone.convey.bridge as convey_bridge +import solstone.convey.root as root_module from solstone.apps.observer.routes import OBSERVER_CALLOSUM_SSE_ROUTE from solstone.apps.observer.utils import ( load_observer, save_observer, ) +from solstone.convey.secure_listener import ConveyIdentity from solstone.convey.sol_initiated.copy import KIND_SOL_CHAT_REQUEST from solstone.observe.protocol import OBSERVER_HANDLE_HEADER +from solstone.think.link.auth import AuthorizedClients +from solstone.think.link.paths import authorized_clients_path + +DEVICE_FINGERPRINT = "sha256:" + ("c" * 64) +BROWSER_FINGERPRINT = "sha256:" + ("b" * 64) @pytest.fixture(autouse=True) @@ -34,14 +41,60 @@ def clear_sse_subscribers() -> Iterator[None]: def _create_observer(env, name: str = "sse-test") -> tuple[str, str]: - resp = env.client.post( - "/app/observer/api/create", - json={"name": name}, - content_type="application/json", + key = f"{name}-key-123456789" + assert save_observer( + { + "key": key, + "name": name, + "created_at": 1, + "enabled": True, + "revoked": False, + "device_binding": {"device": BROWSER_FINGERPRINT, "kind": "browser"}, + "stats": {"segments_received": 0, "bytes_received": 0}, + } + ) + AuthorizedClients(authorized_clients_path()).add_browser( + fingerprint=BROWSER_FINGERPRINT, + device_label="sse-browser", + instance_id="instance-1", + pubkey_spki="30aa", + observer_handle=key, + ) + return key, key[:8] + + +def _pl_identity(fingerprint: str = DEVICE_FINGERPRINT) -> ConveyIdentity: + return ConveyIdentity( + mode="pl-direct", + fingerprint=fingerprint, + device_label="sse-device", + paired_at="2026-07-01T00:00:00Z", + session_id=None, ) - assert resp.status_code == 200 - data = resp.get_json() - return data["key"], data["prefix"] + + +def _authorize_device(fingerprint: str = DEVICE_FINGERPRINT) -> None: + AuthorizedClients(authorized_clients_path()).add( + fingerprint, + "sse-device", + "instance-1", + ) + + +def _create_bound_observer(env, name: str = "sse-bound") -> tuple[str, str]: + key = f"{name}-key-123456789" + assert save_observer( + { + "key": key, + "name": name, + "created_at": 1, + "enabled": True, + "revoked": False, + "device_binding": {"device": DEVICE_FINGERPRINT, "kind": "cert"}, + "stats": {"segments_received": 0, "bytes_received": 0}, + } + ) + return key, key[:8] def _route() -> str: @@ -218,6 +271,9 @@ def test_callosum_sse_handle_revocation_midstream_emits_error( env = observer_env() key, _key_prefix = _create_observer(env, "handle-sse") monkeypatch.setattr(routes_module, "_SSE_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr( + routes_module, "_SSE_DEVICE_RECHECK_SECONDS", 0.01, raising=False + ) resp = env.client.get( _route(), @@ -241,6 +297,48 @@ def test_callosum_sse_handle_revocation_midstream_emits_error( resp.close() +def test_busy_sse_closes_when_bound_device_removed(observer_env, monkeypatch): + env = observer_env() + _authorize_device() + monkeypatch.setattr( + root_module, + "get_authorized_clients", + lambda: AuthorizedClients(authorized_clients_path()), + ) + key, _key_prefix = _create_bound_observer(env, "busy-sse") + monkeypatch.setattr(routes_module, "_SSE_HEARTBEAT_SECONDS", 60) + monkeypatch.setattr( + routes_module, "_SSE_DEVICE_RECHECK_SECONDS", 0.01, raising=False + ) + + resp = env.client.get( + _route(), + headers={OBSERVER_HANDLE_HEADER: key}, + environ_overrides={"pl.identity": _pl_identity()}, + buffered=False, + ) + try: + assert resp.status_code == 200 + assert _next_chunk(resp) == ": heartbeat\n\n" + AuthorizedClients(authorized_clients_path()).remove(DEVICE_FINGERPRINT) + time.sleep(0.02) + convey_bridge._broadcast_callosum_event( + {"tract": "test", "event": "busy", "ts": 1} + ) + + chunk = _next_chunk(resp) + assert chunk.startswith("event: error\n") + data = _parse_sse_data(chunk) + assert data["reason_code"] == "pl_revoked" + + observer = load_observer(key) + assert observer is not None + assert observer.get("revoked") is False + assert observer.get("enabled") is True + finally: + resp.close() + + def test_callosum_sse_heartbeat(observer_env, monkeypatch): env = observer_env() key, _ = _create_observer(env) diff --git a/solstone/apps/observer/tests/test_prune.py b/solstone/apps/observer/tests/test_prune.py index 6607da234..302acd1d2 100644 --- a/solstone/apps/observer/tests/test_prune.py +++ b/solstone/apps/observer/tests/test_prune.py @@ -14,12 +14,15 @@ from solstone.apps.observer.utils import ( load_history, save_observer, ) +from solstone.think.link.auth import AuthorizedClients +from solstone.think.link.paths import authorized_clients_path from solstone.think.streams import read_segment_stream, write_segment_stream DAY = "20250103" STREAM = "field" AUDIO = b"observer prune upload bytes" KEY = "field-prune-key" +FINGERPRINT = "sha256:" + ("c" * 64) def _sha(data: bytes) -> str: @@ -29,10 +32,17 @@ def _sha(data: bytes) -> str: def _observer() -> dict: + AuthorizedClients(authorized_clients_path()).add( + FINGERPRINT, + "prune-device", + "instance-1", + paired_at="2026-05-20T00:00:00Z", + ) return { "key": KEY, "name": STREAM, "stream": STREAM, + "device_binding": {"device": FINGERPRINT, "kind": "cert"}, "created_at": 1, "last_seen": None, "enabled": True, diff --git a/solstone/apps/observer/tests/test_register.py b/solstone/apps/observer/tests/test_register.py index 3d4e06c27..6679097ed 100644 --- a/solstone/apps/observer/tests/test_register.py +++ b/solstone/apps/observer/tests/test_register.py @@ -51,7 +51,13 @@ def _observer_records(journal): def _register(client, **overrides): payload = {**VALID_REGISTER_PAYLOAD, **overrides} - return client.post("/app/observer/register", json=payload) + _authorize_pl() + return client.post( + "/app/observer/register", + json=payload, + environ_base={"REMOTE_ADDR": "192.168.1.5"}, + environ_overrides={"pl.identity": _pl_identity()}, + ) def _pl_identity(fingerprint: str = PL_FINGERPRINT) -> ConveyIdentity: @@ -76,7 +82,7 @@ def _assert_no_observer_records(journal) -> None: assert _observer_paths(journal) == [] -def test_register_loopback_returns_pinned_response(observer_env): +def test_register_authorized_pl_returns_pinned_response(observer_env): env = observer_env() resp = _register(env.client) @@ -91,7 +97,7 @@ def test_register_loopback_returns_pinned_response(observer_env): assert data["protocol_version"] == 2 -def test_register_extension_origin_returns_pinned_response(observer_env): +def test_register_extension_origin_without_device_proof_mints_nothing(observer_env): env = observer_env() resp = env.client.post( @@ -103,10 +109,9 @@ def test_register_extension_origin_returns_pinned_response(observer_env): }, ) - assert resp.status_code == 200 - data = resp.get_json() - assert set(data) == {"key", "prefix", "name", "ingest_url", "protocol_version"} - assert data["name"] == "fedora.tmux" + assert resp.status_code == 403 + assert resp.get_json()["reason_code"] == "local_request_only" + _assert_no_observer_records(env.journal) def test_register_same_stream_twice_reuses_key(observer_env): @@ -234,8 +239,14 @@ def test_register_skips_revoked_record_and_mints_fresh(observer_env): def test_register_requires_descriptor_fields(observer_env, field): env = observer_env() payload = {**VALID_REGISTER_PAYLOAD, field: " "} + _authorize_pl() - resp = env.client.post("/app/observer/register", json=payload) + resp = env.client.post( + "/app/observer/register", + json=payload, + environ_base={"REMOTE_ADDR": "192.168.1.5"}, + environ_overrides={"pl.identity": _pl_identity()}, + ) assert resp.status_code == 400 body = resp.get_json() @@ -313,6 +324,26 @@ def test_register_authorized_pl_identity_from_non_loopback(observer_env): assert records[0]["stream"] == "fedora.tmux" +def test_register_authorized_pl_mints_bound_record(observer_env): + env = observer_env() + _authorize_pl() + + resp = env.client.post( + "/app/observer/register", + json=VALID_REGISTER_PAYLOAD, + environ_base={"REMOTE_ADDR": "192.168.1.5"}, + environ_overrides={"pl.identity": _pl_identity()}, + ) + + assert resp.status_code == 200 + records = _observer_records(env.journal) + assert len(records) == 1 + assert records[0]["device_binding"] == { + "device": PL_FINGERPRINT, + "kind": "cert", + } + + def test_register_reuse_over_authorized_pl_path(observer_env): env = observer_env() _authorize_pl() @@ -435,7 +466,7 @@ def test_registered_observer_segments_legacy_record_uses_locked_stream(observer_ # fields are ignored. -def test_legacy_key_only_observer_still_honors_meta_stream(observer_env): +def test_unbound_key_only_observer_refuses_meta_stream(observer_env): env = observer_env() key = "legacy" + ("a" * 58) assert save_observer( @@ -466,10 +497,10 @@ def test_legacy_key_only_observer_still_honors_meta_stream(observer_env): }, ) - assert resp.status_code == 200 + assert resp.status_code == 401 + assert resp.get_json()["reason_code"] == "auth_required" expected_file = _day_dir(env) / "foo" / "120000_300" / "audio.flac" - assert expected_file.exists() - assert expected_file.read_bytes() == test_data + assert not expected_file.exists() def test_keyless_manifest_routes_accept_bearer_key(observer_env): diff --git a/solstone/apps/observer/tests/test_resolve_identity.py b/solstone/apps/observer/tests/test_resolve_identity.py index 49baeb856..02cf2dd2d 100644 --- a/solstone/apps/observer/tests/test_resolve_identity.py +++ b/solstone/apps/observer/tests/test_resolve_identity.py @@ -6,18 +6,35 @@ from __future__ import annotations import pytest from flask import Flask, g +import solstone.convey.root as root_module +from solstone.apps.observer.routes import OBSERVER_CALLOSUM_SSE_ROUTE from solstone.apps.observer.utils import ( resolve_observer_identity, save_observer, ) from solstone.convey.secure_listener import ConveyIdentity from solstone.observe.protocol import OBSERVER_HANDLE_HEADER +from solstone.think.link.auth import AuthorizedClients +from solstone.think.link.paths import authorized_clients_path from solstone.think.utils import now_ms DL_KEY = "dlkey123456789" HEADER_HANDLE = "headerhandle123456789" FINGERPRINT = "sha256:" + ("c" * 64) OTHER_FINGERPRINT = "sha256:" + ("d" * 64) +BROWSER_FINGERPRINT = "sha256:" + ("e" * 64) + + +ROUTE_CASES = ( + ("callosum_sse", "GET", OBSERVER_CALLOSUM_SSE_ROUTE), + ("delete_source", "DELETE", "/app/observer/source/screen"), + ("ingest_upload", "POST", "/app/observer/ingest"), + ("ingest_manifest", "GET", "/app/observer/ingest/manifest"), + ("ingest_manifest_day", "GET", "/app/observer/ingest/manifest/20250103"), + ("ingest_event", "POST", "/app/observer/ingest/event"), + ("ingest_health", "POST", "/app/observer/health"), + ("ingest_segments", "GET", "/app/observer/ingest/segments/20250103"), +) @pytest.fixture @@ -47,25 +64,97 @@ def _pl_identity(fingerprint: str) -> ConveyIdentity: ) -def _save_observer(handle: str, name: str) -> None: - assert save_observer( - { - "key": handle, - "name": name, - "created_at": now_ms(), - "enabled": True, - "stats": { - "segments_received": 0, - "bytes_received": 0, - }, - } +def _save_observer( + handle: str, + name: str, + *, + device_binding: dict[str, str] | None = None, +) -> None: + record = { + "key": handle, + "name": name, + "created_at": now_ms(), + "enabled": True, + "stats": { + "segments_received": 0, + "bytes_received": 0, + }, + } + if device_binding is not None: + record["device_binding"] = device_binding + assert save_observer(record) + + +def _authorize_cert(fingerprint: str = FINGERPRINT) -> None: + AuthorizedClients(authorized_clients_path()).add( + fingerprint, + "observer", + "instance-1", + ) + + +def _save_cert_observer( + handle: str, + name: str, + *, + fingerprint: str = FINGERPRINT, +) -> None: + _authorize_cert(fingerprint) + _save_observer( + handle, + name, + device_binding={"device": fingerprint, "kind": "cert"}, ) +def _authorize_browser( + *, + fingerprint: str = BROWSER_FINGERPRINT, + observer_handle: str = HEADER_HANDLE, +) -> None: + AuthorizedClients(authorized_clients_path()).add_browser( + fingerprint=fingerprint, + device_label="browser", + instance_id="instance-1", + pubkey_spki="30aa", + observer_handle=observer_handle, + ) + + +def _route_response(client, route_case, handle: str, *, identity=None): + _route_name, method, path = route_case + kwargs = {"headers": {OBSERVER_HANDLE_HEADER: handle}} + if identity is not None: + kwargs["environ_overrides"] = {"pl.identity": identity} + if method == "GET": + if path == OBSERVER_CALLOSUM_SSE_ROUTE: + kwargs["buffered"] = False + return client.get(path, **kwargs) + if method == "DELETE": + return client.delete(path, **kwargs) + if path == "/app/observer/ingest/event": + kwargs["json"] = {"tract": "observe", "event": "status"} + elif path == "/app/observer/health": + kwargs["json"] = {"status": "ok"} + else: + kwargs["data"] = {} + return client.post(path, **kwargs) + + +def _assert_route_reason(response, *, status: int, reason_code: str) -> None: + try: + assert response.status_code == status + body = response.get_json() + assert body["reason_code"] == reason_code + finally: + response.close() + + def test_resolve_dl_success_from_bearer(app_env): - _save_observer(DL_KEY, "dl") + _save_cert_observer(DL_KEY, "dl") with app_env.test_request_context(headers={"Authorization": f"Bearer {DL_KEY}"}): + g.identity = _pl_identity(FINGERPRINT) observer, prefix, error = resolve_observer_identity() assert error is None @@ -75,11 +164,12 @@ def test_resolve_dl_success_from_bearer(app_env): def test_resolve_dl_uses_bearer_key(app_env): header_key = "headerkey123456789" - _save_observer(header_key, "header") + _save_cert_observer(header_key, "header") with app_env.test_request_context( headers={"Authorization": f"Bearer {header_key}"} ): + g.identity = _pl_identity(FINGERPRINT) observer, prefix, error = resolve_observer_identity() assert error is None @@ -135,9 +225,10 @@ def test_resolve_dl_disabled(app_env): def test_resolve_handle_success_from_header(app_env): - _save_observer(HEADER_HANDLE, "header") + _save_cert_observer(HEADER_HANDLE, "header") with app_env.test_request_context(headers={OBSERVER_HANDLE_HEADER: HEADER_HANDLE}): + g.identity = _pl_identity(FINGERPRINT) observer, prefix, error = resolve_observer_identity() assert error is None @@ -148,11 +239,12 @@ def test_resolve_handle_success_from_header(app_env): def test_resolve_handle_success_from_bearer(app_env): bearer_handle = "bearerhandle123456789" - _save_observer(bearer_handle, "bearer") + _save_cert_observer(bearer_handle, "bearer") with app_env.test_request_context( headers={"Authorization": f"Bearer {bearer_handle}"} ): + g.identity = _pl_identity(FINGERPRINT) observer, prefix, error = resolve_observer_identity() assert error is None @@ -162,7 +254,7 @@ def test_resolve_handle_success_from_bearer(app_env): def test_resolve_header_takes_precedence_over_bearer(app_env): - _save_observer("headerfirst123456789", "header-first") + _save_cert_observer("headerfirst123456789", "header-first") _save_observer("bearersecond123456789", "bearer-second") with app_env.test_request_context( @@ -171,6 +263,7 @@ def test_resolve_header_takes_precedence_over_bearer(app_env): "Authorization": "Bearer bearersecond123456789", } ): + g.identity = _pl_identity(FINGERPRINT) observer, prefix, error = resolve_observer_identity() assert error is None @@ -191,7 +284,7 @@ def test_resolve_pl_phone_without_handle_is_auth_required(app_env): def test_resolve_pl_identity_with_header_uses_named_observer(app_env): - _save_observer(HEADER_HANDLE, "named-observer") + _save_cert_observer(HEADER_HANDLE, "named-observer") with app_env.test_request_context(headers={OBSERVER_HANDLE_HEADER: HEADER_HANDLE}): g.identity = _pl_identity(FINGERPRINT) @@ -203,8 +296,8 @@ def test_resolve_pl_identity_with_header_uses_named_observer(app_env): assert prefix == HEADER_HANDLE[:8] -def test_resolve_handle_is_independent_of_pl_fingerprint(app_env): - _save_observer(HEADER_HANDLE, "stable-observer") +def test_resolve_bound_cert_requires_matching_pl_fingerprint(app_env): + _save_cert_observer(HEADER_HANDLE, "stable-observer") with app_env.test_request_context(headers={OBSERVER_HANDLE_HEADER: HEADER_HANDLE}): g.identity = _pl_identity(FINGERPRINT) @@ -214,8 +307,78 @@ def test_resolve_handle_is_independent_of_pl_fingerprint(app_env): observer_again, prefix_again, error_again = resolve_observer_identity() assert error is None - assert error_again is None + payload, status = _error_payload(error_again) + assert status == 403 + assert payload["reason_code"] == "pl_revoked" assert observer["name"] == "stable-observer" - assert observer_again["name"] == "stable-observer" + assert observer_again is None assert prefix == HEADER_HANDLE[:8] - assert prefix_again == HEADER_HANDLE[:8] + assert prefix_again is None + + +@pytest.mark.parametrize( + "route_case", ROUTE_CASES, ids=[case[0] for case in ROUTE_CASES] +) +def test_all_device_routes_refuse_unbound_record(observer_env, route_case): + env = observer_env() + _save_observer(HEADER_HANDLE, "unbound-observer") + + response = _route_response(env.client, route_case, HEADER_HANDLE) + + _assert_route_reason(response, status=401, reason_code="auth_required") + + +@pytest.mark.parametrize( + "route_case", ROUTE_CASES, ids=[case[0] for case in ROUTE_CASES] +) +def test_all_device_routes_refuse_missing_matching_device_identity( + observer_env, + route_case, + monkeypatch, +): + env = observer_env() + _authorize_cert(FINGERPRINT) + monkeypatch.setattr( + root_module, + "get_authorized_clients", + lambda: AuthorizedClients(authorized_clients_path()), + ) + _save_observer( + HEADER_HANDLE, + "cert-observer", + device_binding={"device": FINGERPRINT, "kind": "cert"}, + ) + + response = _route_response( + env.client, + route_case, + HEADER_HANDLE, + identity=_pl_identity(OTHER_FINGERPRINT), + ) + + _assert_route_reason(response, status=403, reason_code="pl_revoked") + + +@pytest.mark.parametrize( + "route_case", ROUTE_CASES, ids=[case[0] for case in ROUTE_CASES] +) +@pytest.mark.parametrize("mismatch", ("wrong_kind", "wrong_handle")) +def test_all_device_routes_refuse_browser_kind_or_handle_mismatch( + observer_env, + route_case, + mismatch, +): + env = observer_env() + _save_observer( + HEADER_HANDLE, + "browser-observer", + device_binding={"device": BROWSER_FINGERPRINT, "kind": "browser"}, + ) + if mismatch == "wrong_kind": + _authorize_cert(BROWSER_FINGERPRINT) + else: + _authorize_browser(observer_handle="other-browser-handle") + + response = _route_response(env.client, route_case, HEADER_HANDLE) + + _assert_route_reason(response, status=403, reason_code="pl_revoked") diff --git a/solstone/apps/observer/utils.py b/solstone/apps/observer/utils.py index 14ee7234c..45a6c0876 100644 --- a/solstone/apps/observer/utils.py +++ b/solstone/apps/observer/utils.py @@ -50,6 +50,10 @@ DISPOSITION_WRITTEN = "written" DISPOSITION_ALREADY_HELD = "already_held" DISPOSITION_RECEIVED_NOT_WRITTEN = "received_not_written" _MEDIA_CONTENT_EXTENSIONS = AUDIO_EXTENSIONS | VIDEO_EXTENSIONS +DEVICE_BINDING_FIELD = "device_binding" +DEVICE_BINDING_KIND_CERT = "cert" +DEVICE_BINDING_KIND_BROWSER = "browser" +DEVICE_BINDING_KINDS = {DEVICE_BINDING_KIND_CERT, DEVICE_BINDING_KIND_BROWSER} @dataclass(frozen=True) @@ -59,6 +63,35 @@ class ObserverIdentityRejection: attempted_prefix: str | None +def _is_sha256_device(value: str) -> bool: + if not value.startswith("sha256:"): + return False + hex_part = value.removeprefix("sha256:") + return len(hex_part) == 64 and all(char in "0123456789abcdef" for char in hex_part) + + +def _normalize_device_binding(raw: object) -> dict[str, str] | None: + if raw is None: + return None + if not isinstance(raw, dict): + raise ValueError("device_binding must be an object") + device = raw.get("device") + kind = raw.get("kind") + if not isinstance(device, str) or not _is_sha256_device(device): + raise ValueError("device_binding.device must be sha256:<64 lowercase hex>") + if kind not in DEVICE_BINDING_KINDS: + raise ValueError("device_binding.kind must be cert or browser") + return {"device": device, "kind": kind} + + +def observer_device_binding(record: dict[str, Any]) -> dict[str, str] | None: + """Return a valid device binding from a loaded observer record, if present.""" + try: + return _normalize_device_binding(record.get(DEVICE_BINDING_FIELD)) + except ValueError: + return None + + def get_observers_dir(*, ensure_exists: bool = True) -> Path: """Get the observers storage directory.""" return get_app_storage_path("observer", "observers", ensure_exists=ensure_exists) @@ -293,10 +326,14 @@ def _validate_observer_record(record: dict[str, Any], path: Path) -> dict | None return None try: prefix = observer_filename_prefix(record) + binding = _normalize_device_binding(record.get(DEVICE_BINDING_FIELD)) except ValueError as exc: logger.warning("Skipping invalid observer record %s: %s", path, exc) return None - return _augment_record(record, prefix) + clean = dict(record) + if binding is not None: + clean[DEVICE_BINDING_FIELD] = binding + return _augment_record(clean, prefix) class ObserverRegistry: @@ -553,6 +590,11 @@ def _rejection_response(rejection: ObserverIdentityRejection) -> tuple[Any, int] def _resolve_identity() -> tuple[ dict | None, str | None, ObserverIdentityRejection | None ]: + from flask import g + + from solstone.think.link.auth import AuthorizedClients + from solstone.think.link.paths import authorized_clients_path + handle = _get_auth_key() if not handle: return ( @@ -582,6 +624,56 @@ def _resolve_identity() -> tuple[ if rejection is not None: return None, None, rejection + binding = observer_device_binding(observer) + if binding is None: + return ( + None, + None, + ObserverIdentityRejection( + AUTH_REQUIRED, + "Observer device binding required", + observer["filename_prefix"], + ), + ) + + entry = AuthorizedClients(authorized_clients_path()).get(binding["device"]) + if entry is None or entry.kind != binding["kind"]: + return ( + None, + None, + ObserverIdentityRejection( + PL_REVOKED, + "Paired device revoked", + observer["filename_prefix"], + ), + ) + + if binding["kind"] == DEVICE_BINDING_KIND_CERT: + identity = getattr(g, "identity", None) + if ( + getattr(identity, "mode", None) not in {"pl-direct", "pl-via-spl"} + or getattr(identity, "fingerprint", None) != binding["device"] + ): + return ( + None, + None, + ObserverIdentityRejection( + PL_REVOKED, + "Paired device revoked", + observer["filename_prefix"], + ), + ) + elif entry.observer_handle != handle: + return ( + None, + None, + ObserverIdentityRejection( + PL_REVOKED, + "Paired device revoked", + observer["filename_prefix"], + ), + ) + return observer, observer["filename_prefix"], None diff --git a/tests/link/test_wsgi_input_body.py b/tests/link/test_wsgi_input_body.py index b43f351f4..33591501f 100644 --- a/tests/link/test_wsgi_input_body.py +++ b/tests/link/test_wsgi_input_body.py @@ -218,7 +218,7 @@ def _observer_key(name: str, *, include_marker: bool = False) -> str: return f"{marker}-{digest}" -def _save_observer(key: str, stream: str) -> None: +def _save_observer(key: str, stream: str, fingerprint: str) -> None: assert save_observer( { "key": key, @@ -229,6 +229,7 @@ def _save_observer(key: str, stream: str) -> None: "label": None, "version": "test", "stream": stream, + "device_binding": {"device": fingerprint, "kind": "cert"}, "created_at": 1_700_000_000_000, "last_seen": None, "last_segment": None, @@ -268,7 +269,7 @@ async def _run_mux_ingest( fingerprint = "sha256:" + hashlib.sha256(name.encode("utf-8")).hexdigest() _authorize_fingerprint(monkeypatch, fingerprint) resolved_key = key or _observer_key(name, include_marker=include_credential_marker) - _save_observer(resolved_key, fixture.stream) + _save_observer(resolved_key, fixture.stream, fingerprint) emitted: list[dict[str, Any]] = [] monkeypatch.setattr( diff --git a/tests/test_bad_media_corpus.py b/tests/test_bad_media_corpus.py index 45f8b26ef..e673e28c7 100644 --- a/tests/test_bad_media_corpus.py +++ b/tests/test_bad_media_corpus.py @@ -56,6 +56,7 @@ DAY = "20990501" STREAM = "default" SEGMENT = "120000_300" FIXED_NOW = "2026-06-30T12:00:00Z" +TEST_PL_FINGERPRINT = "sha256:" + ("d" * 64) def _generate_result(text: str, finish_reason: str = "stop") -> dict[str, Any]: @@ -87,8 +88,19 @@ def observer_env(tmp_path, monkeypatch): monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) from solstone.convey import create_app + from solstone.convey import root as convey_root + from solstone.think.link.auth import AuthorizedClients + from solstone.think.link.paths import authorized_clients_path app = create_app(journal=str(journal)) + authorized = AuthorizedClients(authorized_clients_path()) + authorized.add( + TEST_PL_FINGERPRINT, + "bad-media-observer", + "instance-1", + paired_at="2026-05-20T00:00:00Z", + ) + monkeypatch.setattr(convey_root, "get_authorized_clients", lambda: authorized) class Env: def __init__(self): @@ -474,7 +486,30 @@ def _create_observer(env, name: str) -> str: content_type="application/json", ) assert response.status_code == 200 - return response.get_json()["key"] + key = response.get_json()["key"] + + from solstone.apps.observer.utils import load_observer, save_observer + + observer = load_observer(key) + assert observer is not None + observer["device_binding"] = { + "device": TEST_PL_FINGERPRINT, + "kind": "cert", + } + assert save_observer(observer) + return key + + +def _pl_identity(): + from solstone.convey.secure_listener import ConveyIdentity + + return ConveyIdentity( + mode="pl-via-spl", + fingerprint=TEST_PL_FINGERPRINT, + device_label="bad-media-observer", + paired_at="2026-05-20T00:00:00Z", + session_id="session-1", + ) def test_ac1_ingest_drops_zero_byte_keeps_valid_media(observer_env): @@ -496,6 +531,7 @@ def test_ac1_ingest_drops_zero_byte_keeps_valid_media(observer_env): (io.BytesIO(valid_data), "audio.flac"), ], }, + environ_overrides={"pl.identity": _pl_identity()}, ) assert response.status_code == 200 -- 2.51.2