feat: add durable security activity center (Closes #493)

tests/test_dashboard_auth.py

@@ -17,6 +17,7 @@ def access_control(monkeypatch, tmp_path):
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy")
     monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
+    monkeypatch.setenv("STACKCHAIN_SECURITY_EVENT_DB", str(tmp_path / "security-events.sqlite3"))
     monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login-attempts.sqlite3"))
     monkeypatch.setenv("STACKCHAIN_LOGIN_MAX_FAILURES", "3")
     monkeypatch.setenv("STACKCHAIN_LOGIN_WINDOW_SECONDS", "60")
@@ -136,6 +137,7 @@ async def test_enrolled_passkey_can_sign_in_without_the_operator_token(
                 "target": "dashboard",
             },
         )
+        activity = await returning.get("/api/v1/security-events")
 
     assert enrolled.status_code == 201
     assert enrolled.json() == {"enrolled": True}
@@ -143,6 +145,14 @@ async def test_enrolled_passkey_can_sign_in_without_the_operator_token(
     assert sign_in_options.json()["allowCredentials"][0]["id"] == "cGhvbmUtY3JlZGVudGlhbA"
     assert signed_in.status_code == 200
     assert signed_in.json() == {"authenticated": True, "method": "passkey"}
+    assert activity.json()["events"][0] == {
+        "id": activity.json()["events"][0]["id"],
+        "kind": "sign_in",
+        "method": "passkey",
+        "device_label": "Phone",
+        "target": "dashboard",
+        "created_at": activity.json()["events"][0]["created_at"],
+    }
     assert "stackchain_session=" in signed_in.headers["set-cookie"]
     assert "correct horse battery staple" not in signed_in.text
 

tests/test_dashboard_session_frontend.py

@@ -596,6 +596,24 @@ process.stdout.write(JSON.stringify(state));
     assert result["confirmations"] == ["Sign out Pixel <script>?"]
 
 
+def test_security_activity_pages_are_loaded_with_stable_cursors():
+    result = run_session_scenario(
+        """
+state.responsePayload = {events:[{id:7,kind:'sign_in',method:'passkey',device_label:'Phone',target:'dashboard',created_at:1000}],next_cursor:7};
+const first = await boundary.listSecurityEvents();
+const second = await boundary.listSecurityEvents(first.next_cursor);
+state.first = first;
+process.stdout.write(JSON.stringify(state));
+"""
+    )
+
+    assert result["first"]["events"][0]["method"] == "passkey"
+    assert [request["url"] for request in result["requests"]] == [
+        "/dashboard/api/v1/security-events?limit=25",
+        "/dashboard/api/v1/security-events?limit=25&cursor=7",
+    ]
+
+
 def test_authenticated_dashboard_load_requests_queued_delivery_resume():
     result = run_session_scenario(
         """
@@ -618,4 +636,7 @@ async def test_dashboard_loads_session_boundary_first_and_offers_sign_out():
     assert '<button id="active-devices" type="button">Active devices</button>' in html
     assert 'id="active-devices-sheet"' in html
     assert 'aria-label="Active devices"' in html
+    assert 'aria-labelledby="security-activity-title"' in html
+    assert 'id="security-activity-list"' in html
+    assert '>Load older activity</button>' in html
     assert '#sign-out-all { min-height:44px; }' in html

tests/test_security_activity.py

@@ -0,0 +1,245 @@
+import sqlite3
+
+import httpx
+import pytest
+
+from src import main
+from src.security_event_store import SecurityEventStore
+
+
+@pytest.fixture
+def security_access(monkeypatch, tmp_path):
+    monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
+    monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
+    monkeypatch.setenv(
+        "STACKCHAIN_DASHBOARD_SESSION_SECRET",
+        "a-separate-session-signing-secret-with-enough-entropy",
+    )
+    monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
+    monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "attempts.sqlite3"))
+    monkeypatch.setenv("STACKCHAIN_SECURITY_EVENT_DB", str(tmp_path / "security.sqlite3"))
+    return tmp_path
+
+
+@pytest.mark.anyio
+async def test_authenticated_security_activity_lists_private_sign_in_history(security_access):
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as anonymous:
+        rejected = await anonymous.get("/api/v1/security-events")
+
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        signed_in = await client.post(
+            "/api/v1/session",
+            json={
+                "access_token": "correct horse battery staple",
+                "device_label": "Timmy's phone",
+            },
+        )
+        activity = await client.get("/api/v1/security-events", params={"limit": 25})
+
+    assert rejected.status_code == 401
+    assert signed_in.status_code == 200
+    assert activity.status_code == 200
+    assert activity.headers["cache-control"] == "no-store"
+    assert activity.json() == {
+        "events": [
+            {
+                "id": 1,
+                "kind": "sign_in",
+                "method": "token",
+                "device_label": "Timmy's phone",
+                "target": "dashboard",
+                "created_at": activity.json()["events"][0]["created_at"],
+            }
+        ],
+        "next_cursor": None,
+    }
+    persisted = (security_access / "security.sqlite3").read_bytes()
+    assert b"correct horse battery staple" not in persisted
+    assert b"stackchain_session" not in persisted
+
+
+@pytest.mark.anyio
+async def test_security_activity_limit_is_validated(security_access):
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
+        )
+        too_large = await client.get("/api/v1/security-events", params={"limit": 101})
+        invalid_cursor = await client.get("/api/v1/security-events", params={"cursor": 0})
+
+    assert too_large.status_code == 422
+    assert invalid_cursor.status_code == 422
+
+
+@pytest.mark.anyio
+async def test_revoked_device_activity_survives_live_session_removal(security_access):
+    transport = httpx.ASGITransport(app=main.app)
+    async with (
+        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
+        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
+    ):
+        await phone.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
+        )
+        await laptop.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
+        )
+        devices = (await laptop.get("/api/v1/sessions")).json()["devices"]
+        phone_device = next(device for device in devices if device["device_label"] == "Phone")
+        csrf = laptop.cookies["stackchain_csrf"]
+        grant = await laptop.post(
+            "/api/v1/fresh-authorization",
+            json={
+                "access_token": "correct horse battery staple",
+                "action": "revoke_device",
+                "target": phone_device["management_id"],
+            },
+            headers={"Origin": "https://test", "X-CSRF-Token": csrf},
+        )
+        revoked = await laptop.delete(
+            f"/api/v1/sessions/{phone_device['management_id']}",
+            headers={
+                "Origin": "https://test",
+                "X-CSRF-Token": csrf,
+                "X-Step-Up-Grant": grant.json()["grant"],
+            },
+        )
+        active = (await laptop.get("/api/v1/sessions")).json()["devices"]
+        history = (await laptop.get("/api/v1/security-events")).json()["events"]
+
+    assert revoked.status_code == 200
+    assert [device["device_label"] for device in active] == ["Laptop"]
+    assert [(event["kind"], event["device_label"]) for event in history] == [
+        ("device_revoked", "Phone"),
+        ("sign_in", "Laptop"),
+        ("sign_in", "Phone"),
+    ]
+
+
+@pytest.mark.anyio
+async def test_successful_protected_actions_record_only_bounded_targets(
+    security_access, monkeypatch
+):
+    async def yes(*_args):
+        return True
+
+    async def close(*_args):
+        return {"number": 7, "state": "closed"}
+
+    async def merge(*_args):
+        return {"number": 9, "merged": True, "state": "closed"}
+
+    monkeypatch.setattr(main.gitea_proxy, "is_assigned_issue", yes)
+    monkeypatch.setattr(main.gitea_proxy, "close_issue", close)
+    monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", yes)
+    monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge)
+
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
+        )
+        csrf_headers = {
+            "Origin": "https://test",
+            "X-CSRF-Token": client.cookies["stackchain_csrf"],
+        }
+
+        async def grant(action, target):
+            response = await client.post(
+                "/api/v1/fresh-authorization",
+                json={
+                    "access_token": "correct horse battery staple",
+                    "action": action,
+                    "target": target,
+                },
+                headers=csrf_headers,
+            )
+            return response.json()["grant"]
+
+        issue_grant = await grant("close_issue", "stackchain/api#7")
+        closed = await client.patch(
+            "/api/v1/repos/stackchain/api/issues/7/close",
+            headers={**csrf_headers, "X-Step-Up-Grant": issue_grant},
+        )
+        pull_grant = await grant("merge_pull", "stackchain/app#9")
+        merged = await client.post(
+            "/api/v1/repos/stackchain/app/pulls/9/merge",
+            json={"expected_head_sha": "abc123"},
+            headers={**csrf_headers, "X-Step-Up-Grant": pull_grant},
+        )
+        events = (await client.get("/api/v1/security-events")).json()["events"]
+
+    assert closed.status_code == 200
+    assert merged.status_code == 200
+    assert [(event["kind"], event["target"]) for event in events[:2]] == [
+        ("pull_merged", "stackchain/app#9"),
+        ("issue_closed", "stackchain/api#7"),
+    ]
+    persisted = (security_access / "security.sqlite3").read_bytes()
+    assert b"abc123" not in persisted
+
+
+@pytest.mark.anyio
+async def test_sign_out_is_journaled_before_the_session_is_removed(security_access):
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
+        )
+        signed_out = await client.delete(
+            "/api/v1/session",
+            headers={
+                "Origin": "https://test",
+                "X-CSRF-Token": client.cookies["stackchain_csrf"],
+            },
+        )
+
+    events = SecurityEventStore(
+        security_access / "security.sqlite3", clock=lambda: 0
+    ).list(limit=10).events
+    assert signed_out.status_code == 200
+    assert [(event.kind, event.device_label) for event in events[:2]] == [
+        ("sign_out", None),
+        ("sign_in", "Phone"),
+    ]
+
+
+@pytest.mark.anyio
+async def test_sign_out_all_is_journaled_after_sessions_are_removed(security_access):
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
+        )
+        csrf_headers = {
+            "Origin": "https://test",
+            "X-CSRF-Token": client.cookies["stackchain_csrf"],
+        }
+        grant = await client.post(
+            "/api/v1/fresh-authorization",
+            json={
+                "access_token": "correct horse battery staple",
+                "action": "revoke_all_sessions",
+                "target": "all",
+            },
+            headers=csrf_headers,
+        )
+        response = await client.delete(
+            "/api/v1/sessions",
+            headers={**csrf_headers, "X-Step-Up-Grant": grant.json()["grant"]},
+        )
+
+    events = SecurityEventStore(
+        security_access / "security.sqlite3", clock=lambda: 0
+    ).list(limit=10).events
+    assert response.status_code == 200
+    assert events[0].kind == "all_sessions_revoked"
+    assert events[0].target == "all_devices"

tests/test_security_event_store.py

@@ -0,0 +1,54 @@
+import sqlite3
+
+from src.security_event_store import SecurityEventStore
+
+
+def test_security_events_are_private_bounded_and_reverse_chronological(tmp_path):
+    now = [1_000]
+    store = SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: now[0])
+
+    store.record(
+        "sign_in",
+        method="token",
+        device_label="  Timmy   Phone  " + "x" * 80,
+        target="dashboard",
+    )
+    now[0] += 1
+    store.record("device_revoked", device_label="Old phone", target="device")
+
+    page = store.list(limit=1)
+    assert [(event.kind, event.device_label, event.target) for event in page.events] == [
+        ("device_revoked", "Old phone", "device")
+    ]
+    assert page.next_cursor is not None
+
+    older = store.list(limit=10, cursor=page.next_cursor)
+    assert older.events[0].kind == "sign_in"
+    assert older.events[0].method == "token"
+    assert older.events[0].device_label == ("Timmy Phone " + "x" * 52)
+
+    columns = {
+        row[1] for row in sqlite3.connect(store.path).execute("PRAGMA table_info(security_events)")
+    }
+    assert columns == {"id", "kind", "method", "device_label", "target", "created_at"}
+
+
+def test_security_event_retention_prunes_age_and_count(tmp_path):
+    now = [0]
+    store = SecurityEventStore(
+        tmp_path / "security.sqlite3",
+        clock=lambda: now[0],
+        max_events=3,
+        retention_seconds=10,
+    )
+    for index in range(4):
+        now[0] = index
+        store.record("sign_in", method="token", device_label=f"Device {index}")
+
+    assert [event.device_label for event in store.list(limit=10).events] == [
+        "Device 3", "Device 2", "Device 1"
+    ]
+
+    now[0] = 20
+    store.record("sign_out", device_label="Current")
+    assert [event.kind for event in store.list(limit=10).events] == ["sign_out"]