feat: surface failed token sign-in alerts (Closes #543)

frontend/dashboard.css

@@ -30,6 +30,8 @@ button { background: linear-gradient(180deg,#1f3a5f,#15324d); border:1px solid #
 .security-activity-header h3, .security-activity-header p { margin:0 0 6px; }
 .security-activity-list { display:grid; gap:8px; margin:12px 0; }
 .security-event { padding:12px; border:1px solid #243d5d; border-radius:12px; background:#0d1c30; }
+.authentication-alert { border-color:#d69e2e; background:#241b09; }
+.authentication-alert strong { color:#f6c453; }
 .security-event strong, .security-event span { display:block; overflow-wrap:anywhere; }
 #load-more-security-activity { width:100%; min-height:44px; }
 button:hover { filter: brightness(1.15); }

frontend/session.js

@@ -142,6 +142,20 @@
         }
         try {
           const page = await boundary.listSecurityEvents(append ? activityCursor : null);
+          if (!append) {
+            page.authentication_alerts.forEach(alert => {
+              const row = root.document.createElement('article');
+              row.className = 'security-event authentication-alert';
+              const formatted = boundary.formatAuthenticationAlert(alert);
+              const title = root.document.createElement('strong');
+              title.textContent = formatted.title;
+              const details = root.document.createElement('span');
+              details.className = 'small muted';
+              details.textContent = formatted.detail;
+              row.append(title, details);
+              activityList.append(row);
+            });
+          }
           const labels = {
             sign_in: 'Signed in',
             sign_out: 'Signed out',
@@ -665,6 +679,21 @@
     return response.json();
   }
 
+  function formatAuthenticationAlert(alert) {
+    const failed = Math.max(0, Number(alert?.failed_count) || 0);
+    const blocked = Math.max(0, Number(alert?.blocked_count) || 0);
+    const method = alert?.method === 'passkey' ? 'passkey' : 'token';
+    const title = `${failed} failed ${method} sign-in${failed === 1 ? '' : 's'}`
+      + (blocked ? ` · ${blocked} blocked` : '');
+    const formatTime = value => new Date(Number(value) * 1000).toLocaleString(
+      'en-US', { timeZone: 'UTC' },
+    );
+    return {
+      title,
+      detail: `${formatTime(alert?.first_at)} – ${formatTime(alert?.last_at)}`,
+    };
+  }
+
   async function listSecurityEvents(cursor = null) {
     const query = new URLSearchParams({ limit: '25' });
     if (Number.isInteger(cursor) && cursor > 0) query.set('cursor', String(cursor));
@@ -673,6 +702,8 @@
     const payload = await response.json();
     return {
       events: Array.isArray(payload.events) ? payload.events : [],
+      authentication_alerts: Array.isArray(payload.authentication_alerts)
+        ? payload.authentication_alerts : [],
       next_cursor: Number.isInteger(payload.next_cursor) ? payload.next_cursor : null,
     };
   }
@@ -722,6 +753,7 @@
     listActiveDevices,
     listPasskeys,
     listSecurityEvents,
+    formatAuthenticationAlert,
     enrollPasskey,
     revokeActiveDevice,
     revokePasskey,

src/login_attempt_store.py

@@ -49,6 +49,9 @@ class LoginAttemptStore:
         window_seconds: int,
         max_entries: int = 10_000,
         lock_timeout_seconds: float = 0.1,
+        alert_bucket_seconds: int = 60 * 60,
+        alert_retention_seconds: int = 30 * 24 * 60 * 60,
+        max_alert_buckets: int = 720,
     ) -> None:
         self.path = Path(path)
         self.clock = clock
@@ -56,6 +59,9 @@ class LoginAttemptStore:
         self.window_seconds = max(1, window_seconds)
         self.max_entries = max(1, max_entries)
         self.lock_timeout_seconds = lock_timeout_seconds
+        self.alert_bucket_seconds = max(1, alert_bucket_seconds)
+        self.alert_retention_seconds = max(1, alert_retention_seconds)
+        self.max_alert_buckets = max(1, max_alert_buckets)
 
     @staticmethod
     def _digest(source: str) -> str:
@@ -85,6 +91,19 @@ class LoginAttemptStore:
                 )
                 """
             )
+            connection.execute(
+                """
+                CREATE TABLE IF NOT EXISTS login_alerts (
+                    method TEXT NOT NULL,
+                    bucket_started_at INTEGER NOT NULL,
+                    failed_count INTEGER NOT NULL DEFAULT 0,
+                    blocked_count INTEGER NOT NULL DEFAULT 0,
+                    first_at INTEGER NOT NULL,
+                    last_at INTEGER NOT NULL,
+                    PRIMARY KEY (method, bucket_started_at)
+                )
+                """
+            )
             return connection
         except (OSError, sqlite3.Error) as exc:
             raise LoginAttemptStoreError(
@@ -154,7 +173,35 @@ class LoginAttemptStore:
                 "Sign-in throttling is temporarily unavailable"
             ) from exc
 
-    def record_failure(self, source: str) -> None:
+    def _record_alert(
+        self, connection: sqlite3.Connection, method: str, *, blocked: bool, now: float
+    ) -> None:
+        safe_method = method if method in {"token", "passkey"} else "unknown"
+        occurred_at = int(now)
+        bucket = occurred_at - (occurred_at % self.alert_bucket_seconds)
+        failed = 0 if blocked else 1
+        blocked_count = 1 if blocked else 0
+        connection.execute(
+            """
+            INSERT INTO login_alerts(
+                method, bucket_started_at, failed_count, blocked_count, first_at, last_at
+            ) VALUES (?, ?, ?, ?, ?, ?)
+            ON CONFLICT(method, bucket_started_at) DO UPDATE SET
+                failed_count = failed_count + excluded.failed_count,
+                blocked_count = blocked_count + excluded.blocked_count,
+                first_at = MIN(first_at, excluded.first_at),
+                last_at = MAX(last_at, excluded.last_at)
+            """,
+            (safe_method, bucket, failed, blocked_count, occurred_at, occurred_at),
+        )
+        connection.execute(
+            "DELETE FROM login_alerts WHERE (method, bucket_started_at) NOT IN "
+            "(SELECT method, bucket_started_at FROM login_alerts "
+            "ORDER BY bucket_started_at DESC LIMIT ?)",
+            (self.max_alert_buckets,),
+        )
+
+    def record_failure(self, source: str, *, method: str = "token") -> None:
         now = self.clock()
         source_hash = self._digest(source)
         try:
@@ -188,11 +235,52 @@ class LoginAttemptStore:
                     """,
                     (self.max_entries,),
                 )
+                self._record_alert(connection, method, blocked=False, now=now)
         except (OSError, sqlite3.Error) as exc:
             raise LoginAttemptStoreError(
                 "Sign-in throttling is temporarily unavailable"
             ) from exc
 
+    def record_blocked(self, *, method: str = "token") -> None:
+        now = self.clock()
+        try:
+            with self._connect() as connection:
+                connection.execute("BEGIN IMMEDIATE")
+                self._record_alert(connection, method, blocked=True, now=now)
+        except (OSError, sqlite3.Error) as exc:
+            raise LoginAttemptStoreError(
+                "Sign-in throttling is temporarily unavailable"
+            ) from exc
+
+    def list_alerts(self, *, limit: int = 24) -> list[dict[str, int | str]]:
+        bounded_limit = min(100, max(1, limit))
+        now = int(self.clock())
+        try:
+            with self._connect() as connection:
+                connection.execute(
+                    "DELETE FROM login_alerts WHERE last_at < ?",
+                    (now - self.alert_retention_seconds,),
+                )
+                rows = connection.execute(
+                    "SELECT method, failed_count, blocked_count, first_at, last_at "
+                    "FROM login_alerts ORDER BY bucket_started_at DESC LIMIT ?",
+                    (bounded_limit,),
+                ).fetchall()
+        except (OSError, sqlite3.Error) as exc:
+            raise LoginAttemptStoreError(
+                "Sign-in throttling is temporarily unavailable"
+            ) from exc
+        return [
+            {
+                "method": row[0],
+                "failed_count": row[1],
+                "blocked_count": row[2],
+                "first_at": row[3],
+                "last_at": row[4],
+            }
+            for row in rows
+        ]
+
     def clear(self, source: str) -> None:
         try:
             with self._connect() as connection:

src/main.py

@@ -947,6 +947,14 @@ async def sign_in(payload: DashboardSignIn, request: Request, response: Response
             headers={"Cache-Control": "no-store"},
         )
     if retry_after:
+        try:
+            await asyncio.to_thread(attempts.record_blocked, method="token")
+        except LoginAttemptStoreError:
+            return JSONResponse(
+                {"detail": "Sign-in throttling is temporarily unavailable"},
+                status_code=503,
+                headers={"Cache-Control": "no-store"},
+            )
         return JSONResponse(
             {"detail": "Too many sign-in attempts"},
             status_code=429,
@@ -1034,7 +1042,10 @@ async def list_security_events(
         page = await asyncio.to_thread(
             _security_event_store().list, limit=limit, cursor=cursor
         )
-    except SecurityEventStoreError:
+        authentication_alerts = await asyncio.to_thread(
+            _login_attempt_store().list_alerts, limit=24
+        )
+    except (SecurityEventStoreError, LoginAttemptStoreError):
         raise HTTPException(
             status_code=503, detail="Security activity is temporarily unavailable"
         )
@@ -1052,6 +1063,7 @@ async def list_security_events(
                 }
                 for event in page.events
             ],
+            "authentication_alerts": authentication_alerts,
             "next_cursor": page.next_cursor,
         },
         headers={"Cache-Control": "no-store"},

tests/test_dashboard_auth.py

@@ -825,6 +825,44 @@ async def test_sign_in_throttles_repeated_failures_with_retry_guidance(access_co
     assert blocked.headers["cache-control"] == "no-store"
 
 
+@pytest.mark.anyio
+async def test_security_activity_surfaces_failed_and_blocked_token_sign_ins(access_control):
+    attacker_transport = httpx.ASGITransport(app=main.app, client=("203.0.113.7", 1234))
+    operator_transport = httpx.ASGITransport(app=main.app, client=("198.51.100.4", 1234))
+    async with httpx.AsyncClient(
+        transport=attacker_transport, base_url="https://test"
+    ) as attacker:
+        failures = [
+            await attacker.post(
+                "/api/v1/session",
+                json={"access_token": "wrong", "device_label": "Unknown"},
+            )
+            for _ in range(4)
+        ]
+    async with httpx.AsyncClient(
+        transport=operator_transport, base_url="https://test"
+    ) as operator:
+        signed_in = await operator.post(
+            "/api/v1/session",
+            json={
+                "access_token": "correct horse battery staple",
+                "device_label": "Phone",
+            },
+        )
+        activity = await operator.get("/api/v1/security-events")
+
+    assert [response.status_code for response in failures] == [401, 401, 401, 429]
+    assert signed_in.status_code == 200
+    assert activity.status_code == 200
+    assert activity.headers["cache-control"] == "no-store"
+    alerts = activity.json()["authentication_alerts"]
+    assert len(alerts) == 1
+    assert alerts[0]["method"] == "token"
+    assert alerts[0]["failed_count"] == 3
+    assert alerts[0]["blocked_count"] == 1
+    assert alerts[0]["first_at"] <= alerts[0]["last_at"]
+
+
 @pytest.mark.anyio
 async def test_successful_sign_in_clears_prior_failures(access_control):
     transport = httpx.ASGITransport(app=main.app, client=("203.0.113.8", 1234))
@@ -1555,6 +1593,9 @@ async def test_sign_in_throttle_lookup_does_not_block_the_event_loop(
             time.sleep(0.15)
             return 17
 
+        def record_blocked(self, *, method):
+            assert method == "token"
+
     monkeypatch.setattr(main, "_login_attempt_store", lambda: SlowAttempts())
     transport = httpx.ASGITransport(app=main.app, client=("203.0.113.10", 1234))
     async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:

tests/test_dashboard_session_frontend.py

@@ -736,6 +736,34 @@ process.stdout.write(JSON.stringify(state));
     ]
 
 
+def test_security_activity_formats_failed_sign_in_alerts_for_mobile_scanability():
+    result = run_session_scenario(
+        """
+state.formatted = boundary.formatAuthenticationAlert({
+  method:'token', failed_count:12, blocked_count:3, first_at:1000, last_at:1060
+});
+process.stdout.write(JSON.stringify(state));
+"""
+    )
+
+    assert result["formatted"] == {
+        "title": "12 failed token sign-ins · 3 blocked",
+        "detail": "1/1/1970, 12:16:40 AM – 1/1/1970, 12:17:40 AM",
+    }
+
+
+def test_security_activity_renders_authentication_alerts_before_journal_events():
+    source = SESSION_JS.read_text()
+
+    alerts = source.index("page.authentication_alerts.forEach")
+    events = source.index("page.events.forEach")
+    assert alerts < events
+    assert "boundary.formatAuthenticationAlert(alert)" in source
+    assert "row.className = 'security-event authentication-alert'" in source
+    css = (ROOT / "frontend" / "dashboard.css").read_text()
+    assert ".authentication-alert { border-color:#d69e2e;" in css
+
+
 def test_security_activity_explains_pending_outcome_confirmation():
     source = SESSION_JS.read_text()
 

tests/test_login_attempt_store.py

@@ -74,6 +74,100 @@ def test_failure_ledger_evicts_oldest_sources_at_its_size_limit(tmp_path):
         assert connection.execute("SELECT COUNT(*) FROM login_attempts").fetchone() == (2,)
 
 
+def test_failed_sign_ins_are_aggregated_across_workers_and_survive_throttle_clear(tmp_path):
+    now = [1_000.0]
+    database = tmp_path / "login-attempts.sqlite3"
+    first = LoginAttemptStore(
+        database,
+        clock=lambda: now[0],
+        max_failures=3,
+        window_seconds=60,
+        alert_bucket_seconds=300,
+    )
+    second = LoginAttemptStore(
+        database,
+        clock=lambda: now[0],
+        max_failures=3,
+        window_seconds=60,
+        alert_bucket_seconds=300,
+    )
+
+    first.record_failure("203.0.113.7", method="token")
+    now[0] = 1_010.0
+    second.record_failure("198.51.100.4", method="token")
+    first.clear("203.0.113.7")
+
+    assert second.list_alerts() == [
+        {
+            "method": "token",
+            "failed_count": 2,
+            "blocked_count": 0,
+            "first_at": 1_000,
+            "last_at": 1_010,
+        }
+    ]
+
+
+def test_rate_blocked_sign_ins_increment_a_distinct_privacy_safe_count(tmp_path):
+    store = LoginAttemptStore(
+        tmp_path / "login-attempts.sqlite3",
+        clock=lambda: 1_000.0,
+        max_failures=3,
+        window_seconds=60,
+    )
+
+    store.record_blocked(method="token")
+
+    assert store.list_alerts() == [
+        {
+            "method": "token",
+            "failed_count": 0,
+            "blocked_count": 1,
+            "first_at": 1_000,
+            "last_at": 1_000,
+        }
+    ]
+    with sqlite3.connect(store.path) as connection:
+        columns = [row[1] for row in connection.execute("PRAGMA table_info(login_alerts)")]
+    assert "source_hash" not in columns
+
+
+def test_expired_sign_in_alert_buckets_are_not_returned(tmp_path):
+    now = [1_000.0]
+    store = LoginAttemptStore(
+        tmp_path / "login-attempts.sqlite3",
+        clock=lambda: now[0],
+        max_failures=3,
+        window_seconds=60,
+        alert_retention_seconds=300,
+    )
+    store.record_failure("203.0.113.7")
+
+    now[0] = 1_301.0
+
+    assert store.list_alerts() == []
+
+
+def test_sign_in_alert_bucket_count_is_bounded_during_rotating_source_floods(tmp_path):
+    now = [1_000.0]
+    store = LoginAttemptStore(
+        tmp_path / "login-attempts.sqlite3",
+        clock=lambda: now[0],
+        max_failures=3,
+        window_seconds=60,
+        alert_bucket_seconds=60,
+        max_alert_buckets=2,
+    )
+    for index, timestamp in enumerate((1_000.0, 1_060.0, 1_120.0)):
+        now[0] = timestamp
+        store.record_failure(f"203.0.113.{index}")
+
+    with sqlite3.connect(store.path) as connection:
+        count = connection.execute("SELECT COUNT(*) FROM login_alerts").fetchone()[0]
+
+    assert count == 2
+
+
 def test_named_admission_budget_is_atomic_across_instances_and_source_scoped(tmp_path):
     now = [1_000.0]
     database = tmp_path / "login-attempts.sqlite3"

tests/test_security_activity.py

@@ -53,6 +53,7 @@ async def test_authenticated_security_activity_lists_private_sign_in_history(sec
                 "status": "completed",
             }
         ],
+        "authentication_alerts": [],
         "next_cursor": None,
     }
     persisted = (security_access / "security.sqlite3").read_bytes()