stackchain-dashboard/src/passkey_store.py

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


class PasskeyStore:
    def __init__(self, path: str | Path, *, clock: Callable[[], float]) -> None:
        self.path = Path(path)
        self.clock = clock

    @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,
                    purpose TEXT NOT NULL,
                    action TEXT NOT NULL,
                    target TEXT NOT NULL,
                    expires_at INTEGER NOT NULL
                )
                """
            )
            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,
        ttl_seconds: int = 120,
    ) -> None:
        now = int(self.clock())
        try:
            with self._connect() as connection:
                connection.execute("DELETE FROM passkey_challenges WHERE expires_at <= ?", (now,))
                connection.execute(
                    "INSERT INTO passkey_challenges("
                    "challenge_hash, session_hash, purpose, action, target, expires_at"
                    ") VALUES (?, ?, ?, ?, ?, ?)",
                    (
                        self._digest(challenge),
                        self._digest(session_id) if session_id else None,
                        purpose,
                        action,
                        target,
                        now + max(1, ttl_seconds),
                    ),
                )
        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 "
                    "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 "
                    "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 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) -> None:
        try:
            with self._connect() as connection:
                connection.execute(
                    "DELETE FROM passkey_credentials WHERE management_id = ?", (management_id,)
                )
        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