diff --git a/solstone/apps/observer/tests/test_register.py b/solstone/apps/observer/tests/test_register.py index c0bdecd6e..4200141ed 100644 --- a/solstone/apps/observer/tests/test_register.py +++ b/solstone/apps/observer/tests/test_register.py @@ -91,6 +91,24 @@ def test_register_loopback_returns_pinned_response(observer_env): assert data["protocol_version"] == 2 +def test_register_extension_origin_returns_pinned_response(observer_env): + env = observer_env() + + resp = env.client.post( + "/app/observer/register", + json=VALID_REGISTER_PAYLOAD, + headers={ + "Origin": "chrome-extension://fgfnkcefedeheoeamppkiiloncfekakf", + "Sec-Fetch-Site": "cross-site", + }, + ) + + assert resp.status_code == 200 + data = resp.get_json() + assert set(data) == {"key", "prefix", "name", "ingest_url", "protocol_version"} + assert data["name"] == "fedora.tmux" + + def test_register_same_stream_twice_reuses_key(observer_env): env = observer_env() diff --git a/solstone/convey/root.py b/solstone/convey/root.py index 212a18239..98ddb7de0 100644 --- a/solstone/convey/root.py +++ b/solstone/convey/root.py @@ -141,6 +141,9 @@ def require_access() -> Any: _LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1"}) _SAME_SITE_FETCH = frozenset({"same-origin", "same-site", "none"}) _STATE_CHANGING_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"}) +_TRUSTED_OBSERVER_EXTENSION_ORIGIN = ( + "chrome-extension://fgfnkcefedeheoeamppkiiloncfekakf" +) def _hostname(value: str, *, has_scheme: bool) -> str | None: @@ -156,6 +159,22 @@ def _hostname(value: str, *, has_scheme: bool) -> str | None: return None +def _is_trusted_observer_extension_request() -> bool: + """The pinned solstone-browser extension calling the observer API. + + Chrome's extension service worker sends its ``chrome-extension://`` + 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. + """ + path = request.path + if not (path == "/app/observer" or path.startswith("/app/observer/")): + return False + return request.headers.get("Origin") == _TRUSTED_OBSERVER_EXTENSION_ORIGIN + + @bp.before_app_request def guard_loopback_origin() -> Any: """Reject cross-origin / rebound requests to the local (``dl``) surface. @@ -178,6 +197,11 @@ def guard_loopback_origin() -> Any: # Cross-site guard (state-changing methods) — closes same-host CSRF. if request.method in _STATE_CHANGING_METHODS: + # The pinned observer extension is a trusted local caller; its + # chrome-extension:// Origin / cross-site Sec-Fetch would otherwise trip + # the browser CSRF guard. The observer route still gates on loopback. + if _is_trusted_observer_extension_request(): + return None sec_fetch = request.headers.get("Sec-Fetch-Site") if sec_fetch is not None and sec_fetch not in _SAME_SITE_FETCH: return error_response( diff --git a/tests/test_loopback_guard.py b/tests/test_loopback_guard.py index b78fa96e0..17005458e 100644 --- a/tests/test_loopback_guard.py +++ b/tests/test_loopback_guard.py @@ -35,9 +35,9 @@ def app(tmp_path, monkeypatch): return create_app(str(journal)) -def _guard(app, *, headers, method="GET", mode="dl"): +def _guard(app, *, headers, method="GET", mode="dl", path="/api/whatever"): """Run the guard in a request context; return None (pass) or (resp, status).""" - with app.test_request_context("/api/whatever", method=method, headers=headers): + with app.test_request_context(path, method=method, headers=headers): g.identity = _identity(mode) return guard_loopback_origin() @@ -153,6 +153,83 @@ def test_pl_via_spl_unaffected(app): ) +# --- Trusted observer extension exception --- + + +TRUSTED_EXT_ORIGIN = "chrome-extension://fgfnkcefedeheoeamppkiiloncfekakf" + + +def test_dl_trusted_extension_origin_on_observer_path_passes(app): + result = _guard( + app, + method="POST", + path="/app/observer/register", + headers={ + "Host": "127.0.0.1:5015", + "Origin": TRUSTED_EXT_ORIGIN, + "Sec-Fetch-Site": "cross-site", + }, + ) + + assert result is None + + +def test_dl_trusted_extension_origin_on_nonobserver_path_rejected(app): + result = _guard( + app, + method="POST", + headers={ + "Host": "127.0.0.1:5015", + "Origin": TRUSTED_EXT_ORIGIN, + "Sec-Fetch-Site": "cross-site", + }, + ) + + assert _rejected(result) + + +def test_dl_other_extension_origin_on_observer_path_rejected(app): + result = _guard( + app, + method="POST", + path="/app/observer/register", + headers={ + "Host": "127.0.0.1:5015", + "Origin": "chrome-extension://aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + }, + ) + + assert _rejected(result) + + +def test_dl_trusted_extension_nonloopback_host_rejected(app): + result = _guard( + app, + method="POST", + path="/app/observer/register", + headers={ + "Host": "attacker.example", + "Origin": TRUSTED_EXT_ORIGIN, + }, + ) + + assert _rejected(result) + + +def test_dl_trusted_extension_path_boundary_rejected(app): + result = _guard( + app, + method="POST", + path="/app/observerX", + headers={ + "Host": "127.0.0.1:5015", + "Origin": TRUSTED_EXT_ORIGIN, + }, + ) + + assert _rejected(result) + + # --- Wiring: the guard fires through real request dispatch (before_app_request) ---