185 lines
6.8 KiB
Python
185 lines
6.8 KiB
Python
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")
|