Stop Web Push when device sessions expire #554
|
|
@ -205,6 +205,15 @@ async def session_management_id(session: Session) -> str:
|
||||||
return await asyncio.to_thread(_session_store().management_id, session.session_id)
|
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:
|
async def touch_session(session: Session) -> bool:
|
||||||
return await asyncio.to_thread(
|
return await asyncio.to_thread(
|
||||||
_session_store().touch,
|
_session_store().touch,
|
||||||
|
|
|
||||||
|
|
@ -94,6 +94,7 @@ async def _push_poll_loop() -> None:
|
||||||
_push_subscription_store,
|
_push_subscription_store,
|
||||||
_push_configuration(),
|
_push_configuration(),
|
||||||
notifications,
|
notifications,
|
||||||
|
session_active=dashboard_auth.managed_session_active,
|
||||||
lease_seconds=lease_seconds,
|
lease_seconds=lease_seconds,
|
||||||
send_timeout_seconds=send_timeout,
|
send_timeout_seconds=send_timeout,
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -40,6 +40,7 @@ async def dispatch_unread_updates(
|
||||||
unread: Callable[[], Awaitable[dict]],
|
unread: Callable[[], Awaitable[dict]],
|
||||||
send: Callable[[dict, str], Awaitable[None]] | None = None,
|
send: Callable[[dict, str], Awaitable[None]] | None = None,
|
||||||
*,
|
*,
|
||||||
|
session_active: Callable[[str], Awaitable[bool]] | None = None,
|
||||||
lease_seconds: float = 60.0,
|
lease_seconds: float = 60.0,
|
||||||
send_timeout_seconds: float = 10.0,
|
send_timeout_seconds: float = 10.0,
|
||||||
) -> int:
|
) -> int:
|
||||||
|
|
@ -61,8 +62,24 @@ async def dispatch_unread_updates(
|
||||||
for item in page.get("items", [])
|
for item in page.get("items", [])
|
||||||
if isinstance(item, dict) and str(item.get("notification_id", "")).isdigit()
|
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
|
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:
|
for thread_id in delivery.thread_ids:
|
||||||
still_owner = await asyncio.to_thread(
|
still_owner = await asyncio.to_thread(
|
||||||
store.acquire_dispatch_lease,
|
store.acquire_dispatch_lease,
|
||||||
|
|
|
||||||
|
|
@ -271,6 +271,24 @@ class SessionStore:
|
||||||
raise SessionStoreError("Session is no longer active")
|
raise SessionStoreError("Session is no longer active")
|
||||||
return row[0]
|
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:
|
def revoke_managed(self, management_id: str) -> bool:
|
||||||
try:
|
try:
|
||||||
with self._connect() as connection:
|
with self._connect() as connection:
|
||||||
|
|
|
||||||
|
|
@ -5,7 +5,7 @@ from types import SimpleNamespace
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from src import main
|
from src import dashboard_auth, main
|
||||||
from src.push_notifications import PushConfiguration, dispatch_unread_updates
|
from src.push_notifications import PushConfiguration, dispatch_unread_updates
|
||||||
from src.push_subscription_store import PushSubscriptionStore
|
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 == []
|
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):
|
def test_replacing_subscription_resets_delivery_cursor_and_revocation_removes_device(tmp_path):
|
||||||
store = PushSubscriptionStore(tmp_path / "push.sqlite3")
|
store = PushSubscriptionStore(tmp_path / "push.sqlite3")
|
||||||
original = {
|
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"]
|
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
|
@pytest.mark.anyio
|
||||||
async def test_authenticated_device_can_subscribe_report_status_and_unsubscribe(tmp_path, monkeypatch):
|
async def test_authenticated_device_can_subscribe_report_status_and_unsubscribe(tmp_path, monkeypatch):
|
||||||
store = PushSubscriptionStore(tmp_path / "push.sqlite3")
|
store = PushSubscriptionStore(tmp_path / "push.sqlite3")
|
||||||
|
|
|
||||||
|
|
@ -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"
|
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):
|
def test_touch_extends_only_the_matching_live_session(tmp_path):
|
||||||
now = [1_000.0]
|
now = [1_000.0]
|
||||||
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue
Block a user