"""Durable, one-time WebAuthn challenges and device-bound passkey credentials.""" import hashlib import sqlite3 from dataclasses import dataclass from pathlib import Path from typing import Callable from src.session_store import SessionStoreError @dataclass(frozen=True) class StoredPasskey: credential_id: bytes public_key: bytes sign_count: int device_label: str management_id: str created_at: int class PasskeyStore: def __init__( self, path: str | Path, *, clock: Callable[[], float], max_challenges: int = 10_000, max_challenges_per_source: int = 10, ) -> None: self.path = Path(path) self.clock = clock self.max_challenges = max(1, max_challenges) self.max_challenges_per_source = max(1, max_challenges_per_source) @staticmethod def _digest(value: bytes | str) -> str: raw = value if isinstance(value, bytes) else value.encode() return hashlib.sha256(raw).hexdigest() def _connect(self) -> sqlite3.Connection: try: self.path.parent.mkdir(parents=True, exist_ok=True) connection = sqlite3.connect(self.path, timeout=0.1) connection.execute( """ CREATE TABLE IF NOT EXISTS 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( """ CREATE TABLE IF NOT EXISTS passkey_challenges ( challenge_hash TEXT PRIMARY KEY, session_hash TEXT, source_hash TEXT, purpose TEXT NOT NULL, action TEXT NOT NULL, target TEXT NOT NULL, expires_at INTEGER NOT NULL ) """ ) columns = { row[1] for row in connection.execute("PRAGMA table_info(passkey_challenges)") } if "source_hash" not in columns: connection.execute( "ALTER TABLE passkey_challenges ADD COLUMN source_hash TEXT" ) return connection except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc def issue_challenge( self, challenge: bytes, *, session_id: str | None, purpose: str, action: str, target: str, source: str | None = None, ttl_seconds: int = 120, ) -> None: now = int(self.clock()) source_hash = self._digest(source or session_id or f"{purpose}:{action}:{target}") try: with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") connection.execute("DELETE FROM passkey_challenges WHERE expires_at <= ?", (now,)) connection.execute( "INSERT INTO passkey_challenges(" "challenge_hash, session_hash, source_hash, purpose, action, target, expires_at" ") VALUES (?, ?, ?, ?, ?, ?, ?)", ( self._digest(challenge), self._digest(session_id) if session_id else None, source_hash, purpose, action, target, now + max(1, ttl_seconds), ), ) connection.execute( """ DELETE FROM passkey_challenges WHERE source_hash = ? AND challenge_hash NOT IN ( SELECT challenge_hash FROM passkey_challenges WHERE source_hash = ? ORDER BY expires_at DESC, rowid DESC LIMIT ? ) """, (source_hash, source_hash, self.max_challenges_per_source), ) connection.execute( """ DELETE FROM passkey_challenges WHERE challenge_hash NOT IN ( SELECT challenge_hash FROM passkey_challenges ORDER BY expires_at DESC, rowid DESC LIMIT ? ) """, (self.max_challenges,), ) except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc def consume_challenge( self, challenge: bytes, *, session_id: str | None, purpose: str, action: str, target: str, ) -> bool: now = int(self.clock()) try: with self._connect() as connection: connection.execute("DELETE FROM passkey_challenges WHERE expires_at <= ?", (now,)) cursor = connection.execute( "DELETE FROM passkey_challenges WHERE challenge_hash = ? " "AND session_hash IS ? AND purpose = ? AND action = ? AND target = ? " "AND expires_at > ?", ( self._digest(challenge), self._digest(session_id) if session_id else None, purpose, action, target, now, ), ) return cursor.rowcount == 1 except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc def register( self, *, credential_id: bytes, public_key: bytes, sign_count: int, device_label: str, management_id: str, ) -> None: try: with self._connect() as connection: connection.execute( "INSERT INTO passkey_credentials(credential_id, public_key, sign_count, " "device_label, management_id, created_at) VALUES (?, ?, ?, ?, ?, ?)", ( credential_id, public_key, sign_count, device_label, management_id, int(self.clock()), ), ) except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc def all(self) -> list[StoredPasskey]: try: with self._connect() as connection: rows = connection.execute( "SELECT credential_id, public_key, sign_count, device_label, management_id, created_at " "FROM passkey_credentials ORDER BY created_at DESC" ).fetchall() except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc return [StoredPasskey(*row) for row in rows] def get(self, credential_id: bytes) -> StoredPasskey | None: try: with self._connect() as connection: row = connection.execute( "SELECT credential_id, public_key, sign_count, device_label, management_id, created_at " "FROM passkey_credentials WHERE credential_id = ?", (credential_id,), ).fetchone() except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc return StoredPasskey(*row) if row else None def get_management_id(self, management_id: str) -> StoredPasskey | None: try: with self._connect() as connection: row = connection.execute( "SELECT credential_id, public_key, sign_count, device_label, management_id, created_at " "FROM passkey_credentials WHERE management_id = ?", (management_id,), ).fetchone() except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc return StoredPasskey(*row) if row else None def update_counter(self, credential_id: bytes, new_sign_count: int) -> bool: try: with self._connect() as connection: cursor = connection.execute( "UPDATE passkey_credentials SET sign_count = ? " "WHERE credential_id = ? AND sign_count <= ?", (new_sign_count, credential_id, new_sign_count), ) return cursor.rowcount == 1 except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc def revoke_management_id(self, management_id: str) -> bool: try: with self._connect() as connection: cursor = connection.execute( "DELETE FROM passkey_credentials WHERE management_id = ?", (management_id,) ) return cursor.rowcount == 1 except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc def revoke_access( self, management_id: str, *, preserve_session: bool ) -> tuple[bool, bool]: """Atomically remove one passkey and, when requested, its active session.""" try: with self._connect() as connection: credential = connection.execute( "SELECT 1 FROM passkey_credentials WHERE management_id = ?", (management_id,), ).fetchone() if credential is None: return False, False session_revoked = False if not preserve_session: session = connection.execute( "SELECT session_hash FROM active_sessions WHERE management_id = ?", (management_id,), ).fetchone() if session is not None: connection.execute( "DELETE FROM step_up_grants WHERE session_hash = ?", (session[0],) ) cursor = connection.execute( "DELETE FROM active_sessions WHERE management_id = ?", (management_id,), ) session_revoked = cursor.rowcount == 1 cursor = connection.execute( "DELETE FROM passkey_credentials WHERE management_id = ?", (management_id,), ) return cursor.rowcount == 1, session_revoked except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc def revoke_all(self) -> None: try: with self._connect() as connection: connection.execute("DELETE FROM passkey_credentials") connection.execute("DELETE FROM passkey_challenges") except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Passkey registry is temporarily unavailable") from exc def revoke_device_access(self, management_id: str) -> bool: """Atomically remove one active session, its grants, and its linked passkey.""" try: with self._connect() as connection: session = connection.execute( "SELECT session_hash FROM active_sessions WHERE management_id = ?", (management_id,), ).fetchone() if session is None: return False connection.execute( "DELETE FROM step_up_grants WHERE session_hash = ?", (session[0],) ) connection.execute( "DELETE FROM passkey_credentials WHERE management_id = ?", (management_id,) ) cursor = connection.execute( "DELETE FROM active_sessions WHERE management_id = ?", (management_id,) ) return cursor.rowcount == 1 except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Session registry is temporarily unavailable") from exc def revoke_all_access(self) -> None: """Atomically remove every passkey, challenge, grant, and active session.""" try: with self._connect() as connection: connection.execute("DELETE FROM passkey_credentials") connection.execute("DELETE FROM passkey_challenges") connection.execute("DELETE FROM step_up_grants") connection.execute("DELETE FROM active_sessions") except (OSError, sqlite3.Error) as exc: raise SessionStoreError("Session registry is temporarily unavailable") from exc