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