From 91e5b1e0c9023b3092baf862f2ce1fd46e2315a5 Mon Sep 17 00:00:00 2001 From: timmy Date: Tue, 11 Aug 2026 07:54:56 +0000 Subject: [PATCH] fix: stop push for expired device sessions (Closes #553) --- src/dashboard_auth.py | 9 ++++ src/main.py | 1 + src/push_notifications.py | 19 ++++++- src/session_store.py | 18 +++++++ tests/test_push_notifications.py | 93 +++++++++++++++++++++++++++++++- tests/test_session_store.py | 14 +++++ 6 files changed, 152 insertions(+), 2 deletions(-) diff --git a/src/dashboard_auth.py b/src/dashboard_auth.py index c6cbd88..562826d 100644 --- a/src/dashboard_auth.py +++ b/src/dashboard_auth.py @@ -205,6 +205,15 @@ async def session_management_id(session: Session) -> str: return await asyncio.to_thread(_session_store().management_id, session.session_id) +async def managed_session_active(management_id: str) -> bool: + status = await asyncio.to_thread( + _session_store().managed_status, + management_id, + idle_timeout_seconds=idle_timeout_seconds(), + ) + return status == "active" + + async def touch_session(session: Session) -> bool: return await asyncio.to_thread( _session_store().touch, diff --git a/src/main.py b/src/main.py index f009f52..0a7e13d 100644 --- a/src/main.py +++ b/src/main.py @@ -94,6 +94,7 @@ async def _push_poll_loop() -> None: _push_subscription_store, _push_configuration(), notifications, + session_active=dashboard_auth.managed_session_active, lease_seconds=lease_seconds, send_timeout_seconds=send_timeout, ) diff --git a/src/push_notifications.py b/src/push_notifications.py index 17c8642..5656dd6 100644 --- a/src/push_notifications.py +++ b/src/push_notifications.py @@ -40,6 +40,7 @@ async def dispatch_unread_updates( unread: Callable[[], Awaitable[dict]], send: Callable[[dict, str], Awaitable[None]] | None = None, *, + session_active: Callable[[str], Awaitable[bool]] | None = None, lease_seconds: float = 60.0, send_timeout_seconds: float = 10.0, ) -> int: @@ -61,8 +62,24 @@ async def dispatch_unread_updates( for item in page.get("items", []) if isinstance(item, dict) and str(item.get("notification_id", "")).isdigit() } + deliveries = await asyncio.to_thread(store.claim_unseen, thread_ids) + if session_active is not None: + try: + authorized = [await session_active(item.session_id) for item in deliveries] + except Exception: + # Authorization state is mandatory for delivery. Preserve subscriptions + # so a temporary registry failure can be retried safely. + return 0 + for delivery, active in zip(deliveries, authorized): + if not active: + await asyncio.to_thread(store.delete_session, delivery.session_id) + deliveries = [ + delivery + for delivery, active in zip(deliveries, authorized) + if active + ] count = 0 - for delivery in await asyncio.to_thread(store.claim_unseen, thread_ids): + for delivery in deliveries: for thread_id in delivery.thread_ids: still_owner = await asyncio.to_thread( store.acquire_dispatch_lease, diff --git a/src/session_store.py b/src/session_store.py index 8a351d2..44c1c36 100644 --- a/src/session_store.py +++ b/src/session_store.py @@ -271,6 +271,24 @@ class SessionStore: raise SessionStoreError("Session is no longer active") return row[0] + def managed_status(self, management_id: str, *, idle_timeout_seconds: int) -> str: + """Return the authorization state for a durable device identifier.""" + now = int(self.clock()) + try: + with self._connect() as connection: + row = connection.execute( + "SELECT expires_at, last_active_at FROM active_sessions " + "WHERE management_id = ?", + (management_id,), + ).fetchone() + except (OSError, sqlite3.Error) as exc: + raise SessionStoreError("Session registry is temporarily unavailable") from exc + if row is None or row[0] <= now: + return "revoked" + if row[1] + max(1, idle_timeout_seconds) <= now: + return "idle" + return "active" + def revoke_managed(self, management_id: str) -> bool: try: with self._connect() as connection: diff --git a/tests/test_push_notifications.py b/tests/test_push_notifications.py index 5984fb6..4c76ee1 100644 --- a/tests/test_push_notifications.py +++ b/tests/test_push_notifications.py @@ -5,7 +5,7 @@ from types import SimpleNamespace import pytest -from src import main +from src import dashboard_auth, main from src.push_notifications import PushConfiguration, dispatch_unread_updates from src.push_subscription_store import PushSubscriptionStore @@ -55,6 +55,43 @@ def test_subscription_store_delivers_each_unread_thread_once_per_device(tmp_path assert second == [] +@pytest.mark.anyio +async def test_managed_session_active_applies_configured_idle_deadline(monkeypatch): + calls = [] + + class Store: + def managed_status(self, management_id, *, idle_timeout_seconds): + calls.append((management_id, idle_timeout_seconds)) + return "active" if management_id == "active-device" else "idle" + + monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "321") + monkeypatch.setattr(dashboard_auth, "_session_store", lambda: Store()) + + assert await dashboard_auth.managed_session_active("active-device") is True + assert await dashboard_auth.managed_session_active("idle-device") is False + assert calls == [("active-device", 321), ("idle-device", 321)] + + +@pytest.mark.anyio +async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch): + captured = {} + + async def no_wait(_seconds): + return None + + async def stop_after_capture(*_args, **kwargs): + captured.update(kwargs) + raise asyncio.CancelledError + + monkeypatch.setattr(main.asyncio, "sleep", no_wait) + monkeypatch.setattr(main, "dispatch_unread_updates", stop_after_capture) + + with pytest.raises(asyncio.CancelledError): + await main._push_poll_loop() + + assert captured["session_active"] is dashboard_auth.managed_session_active + + def test_replacing_subscription_resets_delivery_cursor_and_revocation_removes_device(tmp_path): store = PushSubscriptionStore(tmp_path / "push.sqlite3") original = { @@ -217,6 +254,60 @@ async def test_timed_out_endpoint_does_not_stall_push_fanout(tmp_path): assert sent == ["https://push.example/session-b"] +@pytest.mark.anyio +async def test_dispatch_removes_inactive_sessions_without_blocking_active_devices(tmp_path): + store = PushSubscriptionStore(tmp_path / "push.sqlite3") + for session_id in ("expired-device", "active-device"): + store.upsert(session_id, { + "endpoint": f"https://push.example/{session_id}", + "keys": {"p256dh": "public-key", "auth": "auth-secret"}, + }) + sent = [] + + async def unread(): + return {"items": [{"notification_id": 21}]} + + async def session_active(management_id): + return management_id == "active-device" + + async def send(subscription, _payload): + sent.append(subscription["endpoint"]) + + config = PushConfiguration("public", "private", "mailto:ops@example.com") + assert await dispatch_unread_updates( + store, config, unread, send, session_active=session_active + ) == 1 + assert store.is_subscribed("expired-device") is False + assert store.is_subscribed("active-device") is True + assert sent == ["https://push.example/active-device"] + + +@pytest.mark.anyio +async def test_dispatch_fails_closed_and_retains_subscriptions_when_session_registry_fails(tmp_path): + store = PushSubscriptionStore(tmp_path / "push.sqlite3") + store.upsert("active-device", { + "endpoint": "https://push.example/active-device", + "keys": {"p256dh": "public-key", "auth": "auth-secret"}, + }) + sent = [] + + async def unread(): + return {"items": [{"notification_id": 22}]} + + async def registry_unavailable(_management_id): + raise RuntimeError("session registry unavailable") + + async def send(subscription, _payload): + sent.append(subscription) + + config = PushConfiguration("public", "private", "mailto:ops@example.com") + assert await dispatch_unread_updates( + store, config, unread, send, session_active=registry_unavailable + ) == 0 + assert store.is_subscribed("active-device") is True + assert sent == [] + + @pytest.mark.anyio async def test_authenticated_device_can_subscribe_report_status_and_unsubscribe(tmp_path, monkeypatch): store = PushSubscriptionStore(tmp_path / "push.sqlite3") diff --git a/tests/test_session_store.py b/tests/test_session_store.py index 1a7e148..bea7ae4 100644 --- a/tests/test_session_store.py +++ b/tests/test_session_store.py @@ -88,6 +88,20 @@ def test_session_status_reports_idle_without_background_validation_extending_act assert store.status("phone-session", 3_000, idle_timeout_seconds=900) == "idle" +def test_managed_session_status_enforces_absolute_and_idle_expiry(tmp_path): + now = [1_000.0] + store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0]) + store.activate("phone-session", 3_000, management_id="phone-device") + store.activate("short-session", 1_200, management_id="short-device") + + assert store.managed_status("phone-device", idle_timeout_seconds=900) == "active" + now[0] = 1_200.0 + assert store.managed_status("short-device", idle_timeout_seconds=900) == "revoked" + now[0] = 1_900.0 + assert store.managed_status("phone-device", idle_timeout_seconds=900) == "idle" + assert store.managed_status("missing-device", idle_timeout_seconds=900) == "revoked" + + def test_touch_extends_only_the_matching_live_session(tmp_path): now = [1_000.0] store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])