import re import sqlite3 import time from dataclasses import dataclass from argon2 import PasswordHasher from argon2.exceptions import InvalidHashError, VerifyMismatchError from . import audit from .crypto import SecretBox from .db import Database ROLE_ADMIN = "admin" ROLE_VIEWER = "viewer" ROLES = (ROLE_ADMIN, ROLE_VIEWER) MIN_PASSWORD_LEN = 12 MAX_PASSWORD_LEN = 256 USERNAME_RE = re.compile(r"^[A-Za-z0-9._-]{3,64}$") DUMMY_PASSWORD = "dummy-password-for-constant-time" USER_COLUMNS = "id, username, role, totp_secret IS NOT NULL AS has_totp, session_version, disabled" class UserError(Exception): pass @dataclass(frozen=True) class User: id: int username: str role: str has_totp: bool session_version: int disabled: bool @property def is_admin(self) -> bool: return self.role == ROLE_ADMIN def _to_user(row: sqlite3.Row) -> User: return User( id=row["id"], username=row["username"], role=row["role"], has_totp=bool(row["has_totp"]), session_version=row["session_version"], disabled=bool(row["disabled"]), ) class UserStore: def __init__(self, db: Database, box: SecretBox, hasher: PasswordHasher | None = None): self._db = db self._box = box self._hasher = hasher or PasswordHasher() self._dummy_hash = self._hasher.hash(DUMMY_PASSWORD) def count(self) -> int: return self._db.query("SELECT COUNT(*) AS n FROM users")[0]["n"] def get(self, user_id: int) -> User | None: rows = self._db.query(f"SELECT {USER_COLUMNS} FROM users WHERE id=?", (user_id,)) return _to_user(rows[0]) if rows else None def get_by_name(self, username: str) -> User | None: rows = self._db.query(f"SELECT {USER_COLUMNS} FROM users WHERE username=?", (username,)) return _to_user(rows[0]) if rows else None def list(self) -> list[User]: rows = self._db.query(f"SELECT {USER_COLUMNS} FROM users ORDER BY username") return [_to_user(row) for row in rows] def create(self, username: str, password: str, role: str, actor: str) -> User: if not USERNAME_RE.fullmatch(username): raise UserError("Benutzername: 3-64 Zeichen aus Buchstaben, Ziffern, . _ -") self._check_password(password) self._check_role(role) try: self._db.execute( "INSERT INTO users(username, password_hash, role, created_at) VALUES (?,?,?,?)", (username, self._hasher.hash(password), role, int(time.time())), ) except sqlite3.IntegrityError as exc: raise UserError("Benutzername existiert bereits") from exc audit.record(self._db, actor, "user.create", username, None, role) return self.get_by_name(username) def authenticate(self, username: str, password: str) -> User | None: rows = self._db.query( "SELECT id, password_hash FROM users WHERE username=? AND disabled=0", (username,) ) if not rows: self._verify(self._dummy_hash, password) return None stored = rows[0]["password_hash"] if not self._verify(stored, password): return None if self._hasher.check_needs_rehash(stored): self._db.execute( "UPDATE users SET password_hash=? WHERE id=?", (self._hasher.hash(password), rows[0]["id"]), ) return self.get(rows[0]["id"]) def set_password(self, user_id: int, password: str, actor: str) -> None: user = self._require(user_id) self._check_password(password) self._db.execute( "UPDATE users SET password_hash=?, session_version=session_version+1 WHERE id=?", (self._hasher.hash(password), user_id), ) audit.record(self._db, actor, "user.password", user.username) def set_role(self, user_id: int, role: str, actor: str) -> None: user = self._require(user_id) self._check_role(role) if role != ROLE_ADMIN: self._guard_last_admin(user) self._db.execute("UPDATE users SET role=? WHERE id=?", (role, user_id)) audit.record(self._db, actor, "user.role", user.username, user.role, role) def set_disabled(self, user_id: int, disabled: bool, actor: str) -> None: user = self._require(user_id) if disabled: self._guard_last_admin(user) self._db.execute("UPDATE users SET disabled=? WHERE id=?", (int(disabled), user_id)) audit.record(self._db, actor, "user.disable" if disabled else "user.enable", user.username) def delete(self, user_id: int, actor: str) -> None: user = self._require(user_id) self._guard_last_admin(user) self._db.execute("DELETE FROM users WHERE id=?", (user_id,)) audit.record(self._db, actor, "user.delete", user.username) def totp_secret(self, user_id: int) -> str | None: rows = self._db.query("SELECT totp_secret FROM users WHERE id=?", (user_id,)) if not rows or rows[0]["totp_secret"] is None: return None return self._box.decrypt(rows[0]["totp_secret"], self._totp_context(user_id)) def set_totp(self, user_id: int, secret: str | None, actor: str) -> None: user = self._require(user_id) stored = self._box.encrypt(secret, self._totp_context(user_id)) if secret else None self._db.execute("UPDATE users SET totp_secret=? WHERE id=?", (stored, user_id)) audit.record(self._db, actor, "user.totp.enable" if secret else "user.totp.disable", user.username) @staticmethod def _totp_context(user_id: int) -> str: return f"user.totp.{user_id}" def _verify(self, password_hash: str, password: str) -> bool: try: return self._hasher.verify(password_hash, password) except (VerifyMismatchError, InvalidHashError): return False def _require(self, user_id: int) -> User: user = self.get(user_id) if user is None: raise UserError("Benutzer nicht gefunden") return user def _guard_last_admin(self, user: User) -> None: if not user.is_admin or user.disabled: return admins = self._db.query( "SELECT COUNT(*) AS n FROM users WHERE role=? AND disabled=0", (ROLE_ADMIN,) )[0]["n"] if admins <= 1: raise UserError("Der letzte aktive Admin kann nicht geändert werden") @staticmethod def _check_password(password: str) -> None: if len(password) < MIN_PASSWORD_LEN: raise UserError(f"Passwort: mindestens {MIN_PASSWORD_LEN} Zeichen") if len(password) > MAX_PASSWORD_LEN: raise UserError(f"Passwort: höchstens {MAX_PASSWORD_LEN} Zeichen") @staticmethod def _check_role(role: str) -> None: if role not in ROLES: raise UserError("Ungültige Rolle")