From b40a44fc452d49bcb143d3e908757d22caa05b07 Mon Sep 17 00:00:00 2001 From: timmy Date: Tue, 25 Aug 2026 02:06:25 +0000 Subject: [PATCH] fix: bind background push to operator identity (Closes #1374) --- src/dashboard_auth.py | 11 ++++--- src/main.py | 20 +++++++++--- src/session_store.py | 10 ++++-- tests/test_push_notifications.py | 55 +++++++++++++++++++++++++++++++- tests/test_session_store.py | 43 +++++++++++++++++++++++++ 5 files changed, 128 insertions(+), 11 deletions(-) diff --git a/src/dashboard_auth.py b/src/dashboard_auth.py index 83188fa..66ed097 100644 --- a/src/dashboard_auth.py +++ b/src/dashboard_auth.py @@ -252,11 +252,14 @@ async def managed_session_active(management_id: str) -> bool: return status == "active" -async def managed_session_statuses(management_ids) -> dict[str, str]: +async def managed_session_statuses( + management_ids, *, expected_principal_id: int | None = None +) -> dict[str, str]: + options = {"idle_timeout_seconds": idle_timeout_seconds()} + if expected_principal_id is not None: + options["expected_principal_id"] = expected_principal_id return await asyncio.to_thread( - _session_store().managed_statuses, - management_ids, - idle_timeout_seconds=idle_timeout_seconds(), + _session_store().managed_statuses, management_ids, **options ) diff --git a/src/main.py b/src/main.py index 8338a52..7afcc59 100644 --- a/src/main.py +++ b/src/main.py @@ -136,6 +136,18 @@ async def _following_push_snapshot() -> dict: return await get_following(Response()) +async def _identity_bound_push_session_statuses( + management_ids: list[str], +) -> dict[str, str]: + user = await gitea_proxy.current_user() + principal_id = user.get("id") if isinstance(user, dict) else None + if isinstance(principal_id, bool) or not isinstance(principal_id, int) or principal_id <= 0: + raise ValueError("Authenticated Gitea identity is unavailable") + return await dashboard_auth.managed_session_statuses( + management_ids, expected_principal_id=principal_id + ) + + async def _push_poll_loop() -> None: interval = max(5.0, float(os.getenv("STACKCHAIN_PUSH_POLL_SECONDS", "30"))) deadline_interval = max( @@ -160,7 +172,7 @@ async def _push_poll_loop() -> None: _push_subscription_store, _push_configuration(), gitea_proxy.unread_notification_snapshot, - session_statuses=dashboard_auth.managed_session_statuses, + session_statuses=_identity_bound_push_session_statuses, lease_seconds=lease_seconds, send_timeout_seconds=send_timeout, max_concurrency=max_concurrency, @@ -172,7 +184,7 @@ async def _push_poll_loop() -> None: _push_subscription_store, _push_configuration(), gitea_proxy.assigned_issue_snapshot, - session_statuses=dashboard_auth.managed_session_statuses, + session_statuses=_identity_bound_push_session_statuses, send_timeout_seconds=send_timeout, lease_seconds=lease_seconds, max_concurrency=max_concurrency, @@ -183,7 +195,7 @@ async def _push_poll_loop() -> None: _push_subscription_store, _push_configuration(), _following_push_snapshot, - session_statuses=dashboard_auth.managed_session_statuses, + session_statuses=_identity_bound_push_session_statuses, send_timeout_seconds=send_timeout, lease_seconds=lease_seconds, max_concurrency=max_concurrency, @@ -194,7 +206,7 @@ async def _push_poll_loop() -> None: _push_subscription_store, _push_configuration(), _start_day_plan_snapshot, - session_statuses=dashboard_auth.managed_session_statuses, + session_statuses=_identity_bound_push_session_statuses, send_timeout_seconds=send_timeout, lease_seconds=lease_seconds, max_concurrency=max_concurrency, diff --git a/src/session_store.py b/src/session_store.py index e894ed3..7123728 100644 --- a/src/session_store.py +++ b/src/session_store.py @@ -303,7 +303,11 @@ class SessionStore: )[management_id] def managed_statuses( - self, management_ids, *, idle_timeout_seconds: int + self, + management_ids, + *, + idle_timeout_seconds: int, + expected_principal_id: int | None = None, ) -> dict[str, str]: """Return authorization states for durable device identifiers in one read.""" requested = set(management_ids) @@ -314,7 +318,7 @@ class SessionStore: with self._connect() as connection: placeholders = ",".join("?" for _ in requested) rows = connection.execute( - "SELECT management_id, expires_at, last_active_at " + "SELECT management_id, expires_at, last_active_at, principal_id " f"FROM active_sessions WHERE management_id IN ({placeholders})", tuple(requested), ).fetchall() @@ -329,6 +333,8 @@ class SessionStore: statuses[management_id] = "revoked" elif row[1] + idle_timeout <= now: statuses[management_id] = "idle" + elif expected_principal_id is not None and row[2] != expected_principal_id: + statuses[management_id] = "principal_mismatch" else: statuses[management_id] = "active" return statuses diff --git a/tests/test_push_notifications.py b/tests/test_push_notifications.py index e17d784..6330562 100644 --- a/tests/test_push_notifications.py +++ b/tests/test_push_notifications.py @@ -488,6 +488,59 @@ async def test_managed_session_statuses_apply_configured_idle_deadline_in_one_ca assert calls == [(["device-b", "device-a", "device-b"], 321)] +@pytest.mark.anyio +async def test_managed_session_statuses_bind_batch_to_expected_principal(monkeypatch): + calls = [] + + class Store: + def managed_statuses( + self, management_ids, *, idle_timeout_seconds, expected_principal_id + ): + calls.append( + (list(management_ids), idle_timeout_seconds, expected_principal_id) + ) + return {management_id: "active" for management_id in set(management_ids)} + + monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "321") + monkeypatch.setattr(dashboard_auth, "_session_store", lambda: Store()) + + assert await dashboard_auth.managed_session_statuses( + ["device-b", "device-a"], expected_principal_id=42 + ) == {"device-a": "active", "device-b": "active"} + assert calls == [(["device-b", "device-a"], 321, 42)] + + +@pytest.mark.anyio +async def test_push_session_statuses_use_current_upstream_identity(monkeypatch): + calls = [] + + async def current_user(): + return {"id": 42, "login": "timmy"} + + async def statuses(management_ids, *, expected_principal_id): + calls.append((list(management_ids), expected_principal_id)) + return {management_id: "active" for management_id in management_ids} + + monkeypatch.setattr(main.gitea_proxy, "current_user", current_user) + monkeypatch.setattr(main.dashboard_auth, "managed_session_statuses", statuses) + + assert await main._identity_bound_push_session_statuses(["phone-device"]) == { + "phone-device": "active" + } + assert calls == [(["phone-device"], 42)] + + +@pytest.mark.anyio +async def test_push_session_statuses_fail_closed_when_identity_is_invalid(monkeypatch): + async def current_user(): + return {"id": "not-an-integer", "login": "timmy"} + + monkeypatch.setattr(main.gitea_proxy, "current_user", current_user) + + with pytest.raises(ValueError, match="identity"): + await main._identity_bound_push_session_statuses(["phone-device"]) + + @pytest.mark.anyio async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch): captured = {} @@ -516,7 +569,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch await main._push_poll_loop() assert captured["unread"] is main.gitea_proxy.unread_notification_snapshot - assert captured["session_statuses"] is dashboard_auth.managed_session_statuses + assert captured["session_statuses"] is main._identity_bound_push_session_statuses assert captured["max_concurrency"] == 3 assert captured["max_individual_notifications"] == 4 diff --git a/tests/test_session_store.py b/tests/test_session_store.py index 23d4d09..7e6ec9d 100644 --- a/tests/test_session_store.py +++ b/tests/test_session_store.py @@ -151,6 +151,49 @@ def test_managed_session_statuses_classify_a_batch_with_one_read(tmp_path, monke ] +def test_managed_session_statuses_reject_other_and_legacy_principals_in_one_read( + tmp_path, monkeypatch +): + store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0) + store.activate( + "matching-session", + 3_000, + management_id="matching-device", + principal_id=42, + principal_login="timmy", + ) + store.activate( + "other-session", + 3_000, + management_id="other-device", + principal_id=84, + principal_login="other", + ) + store.activate("legacy-session", 3_000, management_id="legacy-device") + statements = [] + connect = sqlite3.connect + + def traced_connect(*args, **kwargs): + connection = connect(*args, **kwargs) + connection.set_trace_callback(statements.append) + return connection + + monkeypatch.setattr(session_store.sqlite3, "connect", traced_connect) + + assert store.managed_statuses( + ["matching-device", "other-device", "legacy-device"], + idle_timeout_seconds=900, + expected_principal_id=42, + ) == { + "matching-device": "active", + "other-device": "principal_mismatch", + "legacy-device": "principal_mismatch", + } + assert [statement for statement in statements if statement.lstrip().upper().startswith("SELECT")] == [ + next(statement for statement in statements if "FROM active_sessions" in statement) + ] + + def test_managed_session_statuses_skips_database_for_an_empty_batch(tmp_path, monkeypatch): store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0) monkeypatch.setattr(store, "_connect", lambda *args, **kwargs: pytest.fail("database opened")) -- 2.43.0