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
     )
 
 

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,

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

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
 

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"))