fix: bind background push to operator identity (Closes #1374)

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