249 lines
9.9 KiB
Python
249 lines
9.9 KiB
Python
"""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]) -> 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, 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
|