from __future__ import annotations import hmac import json import os from dataclasses import dataclass, field from hashlib import pbkdf2_hmac from pathlib import Path from secrets import token_bytes from uuid import uuid4 PBKDF2_ITERATIONS = 600_000 SALT_BYTES = 16 class LocalAccountStateError(RuntimeError): """Raised when persisted local account state is missing or invalid.""" @dataclass(frozen=True) class LocalAccountState: username: str salt: bytes = field(repr=False) iterations: int password_hash: bytes = field(repr=False) class LocalAccountStore: def __init__(self, path: str | Path) -> None: self.path = Path(path) def load(self) -> LocalAccountState | None: if not self.path.exists(): return None try: payload = json.loads(self.path.read_text(encoding="utf-8")) except (OSError, ValueError, json.JSONDecodeError) as exc: raise LocalAccountStateError("local account state is unreadable") from exc if not isinstance(payload, dict): raise LocalAccountStateError("local account state must be an object") username = payload.get("username") salt_hex = payload.get("salt") iterations = payload.get("iterations") password_hash_hex = payload.get("password_hash") if ( not isinstance(username, str) or not username or not isinstance(salt_hex, str) or not isinstance(iterations, int) or iterations <= 0 or not isinstance(password_hash_hex, str) ): raise LocalAccountStateError("local account state is invalid") try: salt = bytes.fromhex(salt_hex) password_hash = bytes.fromhex(password_hash_hex) except ValueError as exc: raise LocalAccountStateError("local account state is invalid") from exc return LocalAccountState( username=username, salt=salt, iterations=iterations, password_hash=password_hash, ) def create(self, username: str, password: str) -> LocalAccountState: if not username.strip(): raise ValueError("username must not be empty") if not password: raise ValueError("password must not be empty") salt = token_bytes(SALT_BYTES) state = LocalAccountState( username=username, salt=salt, iterations=PBKDF2_ITERATIONS, password_hash=_derive_hash(password, salt, PBKDF2_ITERATIONS), ) self._write(state) return state def verify(self, state: LocalAccountState, password: str) -> bool: candidate = _derive_hash(password, state.salt, state.iterations) return hmac.compare_digest(candidate, state.password_hash) def _write(self, state: LocalAccountState) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) temporary = self.path.with_name(f".{self.path.name}.{uuid4().hex}.tmp") payload = { "username": state.username, "salt": state.salt.hex(), "iterations": state.iterations, "password_hash": state.password_hash.hex(), } try: temporary.write_text( json.dumps(payload, ensure_ascii=True, indent=2) + "\n", encoding="utf-8", ) _restrict_permissions(temporary) os.replace(temporary, self.path) _restrict_permissions(self.path) finally: if temporary.exists(): temporary.unlink() def _derive_hash(password: str, salt: bytes, iterations: int) -> bytes: return pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations) def _restrict_permissions(path: Path) -> None: try: path.chmod(0o600) except OSError: return