security: bind passkeys to upstream identity (Closes #1448)

tests/test_access_revocation.py

@@ -6,6 +6,39 @@ from src.passkey_store import PasskeyStore
 from src.session_store import SessionStore, SessionStoreError
 
 
+def test_global_access_revocation_preserves_other_principal_credentials_and_sessions(
+    tmp_path,
+):
+    database = tmp_path / "sessions.sqlite3"
+    sessions = SessionStore(database, clock=lambda: 1_000.0)
+    passkeys = PasskeyStore(database, clock=lambda: 1_000.0)
+    for principal_id in (42, 84):
+        sessions.activate(
+            f"session-{principal_id}",
+            2_000,
+            management_id=f"management-{principal_id}",
+            principal_id=principal_id,
+            principal_login=f"operator-{principal_id}",
+        )
+        passkeys.register(
+            credential_id=f"credential-{principal_id}".encode(),
+            public_key=f"public-key-{principal_id}".encode(),
+            sign_count=0,
+            device_label=f"Device {principal_id}",
+            management_id=f"management-{principal_id}",
+            principal_id=principal_id,
+        )
+
+    passkeys.revoke_all_access(principal_id=42)
+
+    assert passkeys.all(principal_id=42) == []
+    assert [item.principal_id for item in passkeys.all(principal_id=84)] == [84]
+    with sqlite3.connect(database) as connection:
+        assert connection.execute(
+            "SELECT principal_id FROM active_sessions ORDER BY principal_id"
+        ).fetchall() == [(84,)]
+
+
 def test_global_access_revocation_rolls_back_every_credential_and_session_change(tmp_path):
     database = tmp_path / "sessions.sqlite3"
     sessions = SessionStore(database, clock=lambda: 1_000.0)
@@ -14,6 +47,8 @@ def test_global_access_revocation_rolls_back_every_credential_and_session_change
         2_000,
         management_id="phone-management-id",
         device_label="Phone",
+        principal_id=42,
+        principal_login="timmy",
     )
     sessions.mint_step_up(
         "phone-session",
@@ -28,6 +63,7 @@ def test_global_access_revocation_rolls_back_every_credential_and_session_change
         sign_count=0,
         device_label="Phone",
         management_id="phone-management-id",
+        principal_id=42,
     )
     passkeys.issue_challenge(
         b"pending-challenge",
@@ -43,7 +79,7 @@ def test_global_access_revocation_rolls_back_every_credential_and_session_change
         )
 
     with pytest.raises(SessionStoreError):
-        passkeys.revoke_all_access()
+        passkeys.revoke_all_access(principal_id=42)
 
     with sqlite3.connect(database) as connection:
         assert connection.execute("SELECT COUNT(*) FROM passkey_credentials").fetchone() == (1,)
@@ -76,6 +112,7 @@ def test_device_access_revocation_rolls_back_target_and_preserves_other_devices(
             sign_count=0,
             device_label=label,
             management_id=f"{slug}-management-id",
+            principal_id=42,
         )
     with sqlite3.connect(database) as connection:
         connection.execute(

tests/test_dashboard_auth.py

@@ -330,6 +330,56 @@ async def test_enrolled_passkey_can_sign_in_without_the_operator_token(
     assert "correct horse battery staple" not in signed_in.text
 
 
+@pytest.mark.anyio
+async def test_passkey_enrolled_for_another_upstream_principal_is_not_advertised(
+    access_control, monkeypatch
+):
+    identity = {"id": 42, "login": "timmy"}
+
+    async def upstream_user():
+        return dict(identity)
+
+    class VerifiedRegistration:
+        credential_id = b"phone-credential"
+        credential_public_key = b"credential-public-key"
+        sign_count = 0
+
+    monkeypatch.setattr(main, "current_user", upstream_user)
+    monkeypatch.setattr(
+        main.passkeys,
+        "verify_registration",
+        lambda **_kwargs: VerifiedRegistration(),
+    )
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as owner:
+        await owner.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
+        )
+        headers = {
+            "Origin": "https://test",
+            "X-CSRF-Token": owner.cookies["stackchain_csrf"],
+        }
+        grant = await fresh_grant(owner, "enroll_passkey", "current_device")
+        options = await owner.post(
+            "/api/v1/passkeys/registration/options",
+            headers={**headers, "X-Step-Up-Grant": grant},
+        )
+        enrolled = await owner.post(
+            "/api/v1/passkeys/registration/verify",
+            json={"challenge": options.json()["challenge"], "credential": {"id": "fake"}},
+            headers=headers,
+        )
+
+    identity.update(id=84, login="other-operator")
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as other:
+        denied = await other.post("/api/v1/passkeys/authentication/options")
+
+    assert enrolled.status_code == 201
+    assert denied.status_code == 404
+    assert denied.json() == {"detail": "No passkeys enrolled"}
+
+
 @pytest.mark.anyio
 async def test_stale_nonzero_passkey_counter_denies_sign_in_and_records_anomaly(
     access_control, monkeypatch
@@ -349,6 +399,7 @@ async def test_stale_nonzero_passkey_counter_denies_sign_in_and_records_anomaly(
         sign_count=4,
         device_label="Phone",
         management_id="phone-management-id",
+        principal_id=42,
     )
     transport = httpx.ASGITransport(app=main.app)
     async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
@@ -391,6 +442,7 @@ async def test_stale_nonzero_passkey_counter_denies_fresh_authorization(
         sign_count=4,
         device_label="Phone",
         management_id="phone-management-id",
+        principal_id=42,
     )
     transport = httpx.ASGITransport(app=main.app)
     async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
@@ -438,6 +490,7 @@ async def test_passkey_options_are_source_limited_before_challenge_generation(
         sign_count=0,
         device_label="Phone",
         management_id="phone-management-id",
+        principal_id=42,
     )
     first_source = httpx.ASGITransport(app=main.app, client=("203.0.113.7", 1234))
     other_source = httpx.ASGITransport(app=main.app, client=("203.0.113.8", 1234))
@@ -470,6 +523,7 @@ async def test_failed_passkey_verification_consumes_the_shared_sign_in_budget(
         sign_count=0,
         device_label="Phone",
         management_id="phone-management-id",
+        principal_id=42,
     )
     transport = httpx.ASGITransport(app=main.app, client=("203.0.113.9", 1234))
 
@@ -520,6 +574,7 @@ async def test_successful_passkey_sign_in_clears_prior_source_failures(
         sign_count=0,
         device_label="Phone",
         management_id="phone-management-id",
+        principal_id=42,
     )
     source = "203.0.113.10"
     attempts = main._login_attempt_store()
@@ -713,6 +768,7 @@ async def test_device_revocation_failure_preserves_its_passkey_and_session(
             sign_count=0,
             device_label="Phone",
             management_id=phone_device["management_id"],
+            principal_id=42,
         )
         grant = await fresh_grant(
             laptop, "revoke_device", phone_device["management_id"]
@@ -783,6 +839,7 @@ async def test_orphaned_passkey_is_listed_safely_and_can_be_selectively_revoked(
         sign_count=0,
         device_label="Backup key",
         management_id="backup-management-id",
+        principal_id=42,
     )
 
     async with httpx.AsyncClient(transport=transport, base_url="https://test") as laptop:
@@ -863,6 +920,7 @@ async def test_removing_a_remote_passkey_also_revokes_its_active_session(access_
             sign_count=0,
             device_label="Phone",
             management_id=phone_device["management_id"],
+            principal_id=42,
         )
         grant = await fresh_grant(
             laptop, "revoke_passkey", phone_device["management_id"]
@@ -907,6 +965,7 @@ async def test_removing_the_current_passkey_keeps_the_current_session_active(acc
             sign_count=0,
             device_label="Laptop",
             management_id=current["management_id"],
+            principal_id=42,
         )
         grant = await fresh_grant(client, "revoke_passkey", current["management_id"])
         removed = await client.delete(
@@ -1878,7 +1937,7 @@ async def test_sign_out_all_devices_registry_failure_sets_no_cookies(access_cont
         grant = await fresh_grant(client, "revoke_all_sessions", "all")
 
         class BrokenStore:
-            def revoke_all_access(self):
+            def revoke_all_access(self, *, principal_id):
                 raise SessionStoreError("database path and secret details")
 
         monkeypatch.setattr(main, "_passkey_store", lambda: BrokenStore())
@@ -1915,6 +1974,7 @@ async def test_sign_out_all_devices_failure_preserves_passkeys_and_sessions(acce
             sign_count=0,
             device_label="Phone",
             management_id=device["management_id"],
+            principal_id=42,
         )
         grant = await fresh_grant(client, "revoke_all_sessions", "all")
         with sqlite3.connect(store.path) as connection:

tests/test_passkey_store.py

@@ -3,6 +3,50 @@ import sqlite3
 from src.passkey_store import PasskeyStore
 
 
+def test_credentials_are_bound_to_one_upstream_principal_and_legacy_rows_fail_closed(
+    tmp_path,
+):
+    database = tmp_path / "passkeys.sqlite3"
+    with sqlite3.connect(database) as connection:
+        connection.execute(
+            """
+            CREATE TABLE passkey_credentials (
+                credential_id BLOB PRIMARY KEY,
+                public_key BLOB NOT NULL,
+                sign_count INTEGER NOT NULL,
+                device_label TEXT NOT NULL,
+                management_id TEXT NOT NULL UNIQUE,
+                created_at INTEGER NOT NULL
+            )
+            """
+        )
+        connection.execute(
+            "INSERT INTO passkey_credentials VALUES (?, ?, ?, ?, ?, ?)",
+            (b"legacy", b"legacy-key", 0, "Old phone", "legacy-management", 900),
+        )
+
+    store = PasskeyStore(database, clock=lambda: 1_000.0)
+    store.register(
+        credential_id=b"principal-101",
+        public_key=b"public-key",
+        sign_count=0,
+        device_label="Phone",
+        management_id="phone-management",
+        principal_id=101,
+    )
+
+    assert [item.credential_id for item in store.all(principal_id=101)] == [b"principal-101"]
+    assert store.all(principal_id=202) == []
+    assert store.get(b"principal-101", principal_id=101).principal_id == 101
+    assert store.get(b"principal-101", principal_id=202) is None
+    assert store.get(b"legacy", principal_id=101) is None
+    with sqlite3.connect(database) as connection:
+        assert connection.execute(
+            "SELECT principal_id FROM passkey_credentials WHERE credential_id = ?",
+            (b"legacy",),
+        ).fetchone() == (None,)
+
+
 def test_active_challenges_are_bounded_per_source_and_globally_across_instances(tmp_path):
     now = [1_000.0]
     database = tmp_path / "passkeys.sqlite3"
@@ -116,11 +160,12 @@ def test_passkey_counter_advancement_is_atomic_across_store_instances(tmp_path):
         sign_count=4,
         device_label="Phone",
         management_id="phone-management-id",
+        principal_id=42,
     )
 
     assert first.advance_counter(b"phone-credential", expected=4, new=5) is True
     assert second.advance_counter(b"phone-credential", expected=4, new=5) is False
-    assert first.get(b"phone-credential").sign_count == 5
+    assert first.get(b"phone-credential", principal_id=42).sign_count == 5
 
 
 def test_passkey_counter_rejects_non_advancing_values_but_supports_counterless_devices(
@@ -133,6 +178,7 @@ def test_passkey_counter_rejects_non_advancing_values_but_supports_counterless_d
         sign_count=4,
         device_label="Phone",
         management_id="phone-management-id",
+        principal_id=42,
     )
     store.register(
         credential_id=b"counterless-credential",
@@ -140,6 +186,7 @@ def test_passkey_counter_rejects_non_advancing_values_but_supports_counterless_d
         sign_count=0,
         device_label="Security key",
         management_id="counterless-management-id",
+        principal_id=42,
     )
 
     assert store.advance_counter(b"phone-credential", expected=4, new=4) is False

tests/test_security_activity.py

@@ -158,7 +158,7 @@ async def test_passkey_enrollment_reservation_failure_preserves_registry(
 
     assert enrolled.status_code == 503
     assert enrolled.json() == {"detail": "Security activity is temporarily unavailable"}
-    assert main._passkey_store().all() == []
+    assert main._passkey_store().all(principal_id=42) == []
 
 
 @pytest.mark.anyio