From 93deb60c7d8ba2fa2f0269de6d841d9eddd15a56 Mon Sep 17 00:00:00 2001 From: patrick Date: Thu, 5 Nov 2026 23:51:40 +0100 Subject: [PATCH] feat: Phase 2 UI base (login, roles, settings, audit, CLI) Co-Authored-By: Claude Sonnet 5.5 --- CLAUDE.md | 11 +- app/cli.py | 98 +++++++++ app/constants.py | 3 + app/db.py | 12 ++ app/fileconfig.py | 14 ++ app/main.py | 68 ++++++- app/ratelimit.py | 34 ++++ app/totp.py | 52 +++++ app/users.py | 184 +++++++++++++++++ app/web/__init__.py | 0 app/web/account.py | 89 +++++++++ app/web/auth.py | 99 ++++++++++ app/web/deps.py | 89 +++++++++ app/web/pages.py | 319 ++++++++++++++++++++++++++++++ app/web/static/app.css | 178 +++++++++++++++++ app/web/static/app.js | 66 +++++++ app/web/templates/account.html | 47 +++++ app/web/templates/audit.html | 26 +++ app/web/templates/base.html | 41 ++++ app/web/templates/dashboard.html | 27 +++ app/web/templates/error.html | 9 + app/web/templates/login.html | 15 ++ app/web/templates/login_totp.html | 14 ++ app/web/templates/settings.html | 31 +++ app/web/templates/setup.html | 16 ++ app/web/templates/users.html | 68 +++++++ tests/test_ratelimit.py | 63 ++++++ tests/test_totp.py | 54 +++++ tests/test_users.py | 155 +++++++++++++++ tests/test_web.py | 257 ++++++++++++++++++++++++ 30 files changed, 2128 insertions(+), 11 deletions(-) create mode 100644 app/cli.py create mode 100644 app/ratelimit.py create mode 100644 app/totp.py create mode 100644 app/users.py create mode 100644 app/web/__init__.py create mode 100644 app/web/account.py create mode 100644 app/web/auth.py create mode 100644 app/web/deps.py create mode 100644 app/web/pages.py create mode 100644 app/web/static/app.css create mode 100644 app/web/static/app.js create mode 100644 app/web/templates/account.html create mode 100644 app/web/templates/audit.html create mode 100644 app/web/templates/base.html create mode 100644 app/web/templates/dashboard.html create mode 100644 app/web/templates/error.html create mode 100644 app/web/templates/login.html create mode 100644 app/web/templates/login_totp.html create mode 100644 app/web/templates/settings.html create mode 100644 app/web/templates/setup.html create mode 100644 app/web/templates/users.html create mode 100644 tests/test_ratelimit.py create mode 100644 tests/test_totp.py create mode 100644 tests/test_users.py create mode 100644 tests/test_web.py diff --git a/CLAUDE.md b/CLAUDE.md index 4419268..3b2ad23 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -18,7 +18,12 @@ Vollständiger Plan: `/home/sysops/.claude/plans/ja-alle-bernehmen-eager-lark.md - Alarm-Flow: Rohalarm sofort, KI-Analyse per Edit nachgereicht (Telegram editMessage, Matrix m.replace). Max 8 Zeilen DE. Timeout 60 s, 2 Retries, sonst "KI nicht verfügbar". Bündelung pro Host 60 s, max 10 Prompts/min, Recovery ohne KI. Downtime: kein Push, Verlauf "unterdrückt". UI-Stummschaltung mit Ablauf. - Befehle: gemeinsamer Handler, `/` in Telegram, `!` in Matrix, Allowlist pro Backend. - Agent-Loop nur für `/ask`, `/explain`, Tiefenanalyse. 5 feste Lese-Tools (`get_service_status`, `get_host_services`, `get_state_history`, `get_downtimes`, `get_problems`), max 6 Schritte, 90 s, Tool-Output untrusted, alle Aufrufe im Audit-Log. -- Bot schreibt nicht in Checkmk. +- Bot liest Checkmk standardmäßig nur (Automation-User mit Lese-Rolle). Einzige Schreibaktion: Downtime setzen, siehe nächster Punkt. +- Ruhigstellen per Chat (`/mute` und `/downtime` in Telegram, `!mute` und `!downtime` in Matrix): + - A, `mute`: Bot-lokale Stummschaltung mit Dauer, Checkmk bleibt unberührt. Standard. Alarme bieten Buttons "2 h stumm" (Telegram Inline-Buttons, Matrix: `!mute 2h` als Antwort auf die Alarmnachricht). + - B, `downtime`: echte Checkmk-Downtime per REST (`POST /domain-types/downtime/collections/service`, Endpunkt noch gegen echte Instanz prüfen). Eigener Checkmk-User nur mit Downtime-Recht, Secret getrennt vom Lese-User. In der UI abschaltbar, standardmäßig aus, maximale Dauer einstellbar (z.B. 24 h). + - B nur durch Allowlist-User nach Bestätigung (Host, Service, Dauer, Kommentar anzeigen, dann Button bzw. `ja`). Agent und Modell haben keine Schreib-Tools und lösen B nie aus. + - Jede Aktion im Audit-Log: User, Backend, Host, Service, Dauer. - UI: FastAPI, Jinja2, HTMX, deutsch, hell/dunkel. Lokale Nutzer (Argon2, optional TOTP), Rollen admin/viewer, `checkbot create-admin`, Setup-Assistent. - Secrets in SQLite mit AES-GCM. Nur `master.key`, Session-Secret, Bind-Adresse in `/etc/checkbot/` (chmod 600). UI zeigt Secrets nie an. - Audit-Log ohne Secret-Werte. Prompts versioniert mit Rollback, Testbereich. @@ -29,8 +34,8 @@ Vollständiger Plan: `/home/sysops/.claude/plans/ja-alle-bernehmen-eager-lark.md 1. Gerüst: Schema, `crypto.py`, Checkmk- und Ollama-Client, `/health` 2. UI-Basis: Login, Rollen, Setup-Assistent, Verbindungstests, Audit 3. Chat-Interface, Telegram, Poller, Alarm-Flow -4. Matrix-Adapter, Routing, Stummschaltung -5. Command-Handler, Agent-Loop +4. Matrix-Adapter, Routing, Stummschaltung (UI) +5. Command-Handler (inkl. `mute`, `downtime` mit Bestätigung), Agent-Loop 6. Prompt-Versionierung, Testbereich, Verlauf, Retention 7. Deployment (`setup-lxc.sh`, Caddy, systemd, Checkmk-Agent) 8. Tests diff --git a/app/cli.py b/app/cli.py new file mode 100644 index 0000000..393b54d --- /dev/null +++ b/app/cli.py @@ -0,0 +1,98 @@ +import argparse +import getpass +import os +import secrets +import sys +from pathlib import Path + +from .crypto import SecretBox, generate_master_key +from .db import Database +from .fileconfig import MASTER_KEY_FILE, SESSION_SECRET_FILE, load_file_config, load_master_key +from .users import ROLE_ADMIN, UserError, UserStore + +CLI_ACTOR = "cli" +SECRET_FILE_MODE = 0o600 +CONFIG_DIR_MODE = 0o700 +SESSION_SECRET_BYTES = 48 + + +def _write_secret_file(path: Path, content: str) -> bool: + if path.exists(): + return False + path.parent.mkdir(parents=True, exist_ok=True, mode=CONFIG_DIR_MODE) + descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, SECRET_FILE_MODE) + with os.fdopen(descriptor, "w") as handle: + handle.write(content) + return True + + +def _user_store() -> UserStore: + config = load_file_config() + box = SecretBox(load_master_key(config.config_dir / MASTER_KEY_FILE)) + return UserStore(Database(config.db_path), box) + + +def _ask_password() -> str: + password = getpass.getpass("Passwort: ") + if password != getpass.getpass("Passwort wiederholen: "): + raise UserError("Passwörter stimmen nicht überein") + return password + + +def cmd_gen_keys(args: argparse.Namespace) -> int: + config = load_file_config() + files = { + MASTER_KEY_FILE: generate_master_key(), + SESSION_SECRET_FILE: secrets.token_urlsafe(SESSION_SECRET_BYTES), + } + for name, content in files.items(): + created = _write_secret_file(config.config_dir / name, content) + print(f"{name}: {'angelegt' if created else 'existiert bereits, unverändert'}") + return 0 + + +def cmd_create_admin(args: argparse.Namespace) -> int: + username = args.username or input("Benutzername: ").strip() + try: + _user_store().create(username, _ask_password(), ROLE_ADMIN, actor=CLI_ACTOR) + except UserError as exc: + print(f"Fehler: {exc}", file=sys.stderr) + return 1 + print(f"Admin '{username}' angelegt") + return 0 + + +def cmd_set_password(args: argparse.Namespace) -> int: + users = _user_store() + user = users.get_by_name(args.username) + if user is None: + print("Fehler: Benutzer nicht gefunden", file=sys.stderr) + return 1 + try: + users.set_password(user.id, _ask_password(), actor=CLI_ACTOR) + except UserError as exc: + print(f"Fehler: {exc}", file=sys.stderr) + return 1 + print(f"Passwort für '{args.username}' gesetzt") + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="checkbot") + commands = parser.add_subparsers(dest="command", required=True) + commands.add_parser("gen-keys") + create = commands.add_parser("create-admin") + create.add_argument("username", nargs="?") + reset = commands.add_parser("set-password") + reset.add_argument("username") + args = parser.parse_args(argv) + handlers = { + "gen-keys": cmd_gen_keys, + "create-admin": cmd_create_admin, + "set-password": cmd_set_password, + } + return handlers[args.command](args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/app/constants.py b/app/constants.py index 5a28ae6..e0a9fe7 100644 --- a/app/constants.py +++ b/app/constants.py @@ -9,6 +9,8 @@ OLLAMA_URL = "ollama.url" OLLAMA_TOKEN = "ollama.token" OLLAMA_TIMEOUT = "ollama.timeout" OLLAMA_CA_FILE = "ollama.ca_file" +OLLAMA_MODEL_ALERT = "ollama.model_alert" +OLLAMA_MODEL_AGENT = "ollama.model_agent" HEALTH_PING = "health.ping" @@ -16,3 +18,4 @@ FLAG_ON = "1" FLAG_OFF = "0" DEFAULT_OLLAMA_TIMEOUT = "60" +DEFAULT_OLLAMA_MODEL = "qwen3" diff --git a/app/db.py b/app/db.py index 82e3b7f..12cc4d6 100644 --- a/app/db.py +++ b/app/db.py @@ -21,6 +21,18 @@ MIGRATIONS = [ ); CREATE INDEX audit_ts ON audit(ts); """, + """ + CREATE TABLE users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT NOT NULL UNIQUE COLLATE NOCASE, + password_hash TEXT NOT NULL, + role TEXT NOT NULL CHECK (role IN ('admin', 'viewer')), + totp_secret TEXT, + session_version INTEGER NOT NULL DEFAULT 0, + disabled INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL + ); + """, ] diff --git a/app/fileconfig.py b/app/fileconfig.py index 9d29a11..e09e70c 100644 --- a/app/fileconfig.py +++ b/app/fileconfig.py @@ -3,10 +3,13 @@ import os from dataclasses import dataclass from pathlib import Path +from .constants import FLAG_OFF, FLAG_ON + ENV_CONFIG_DIR = "CHECKBOT_CONFIG_DIR" ENV_DB_PATH = "CHECKBOT_DB" ENV_BIND_HOST = "CHECKBOT_BIND_HOST" ENV_BIND_PORT = "CHECKBOT_BIND_PORT" +ENV_SECURE_COOKIES = "CHECKBOT_SECURE_COOKIES" DEFAULT_CONFIG_DIR = "/etc/checkbot" DEFAULT_DB_PATH = "/var/lib/checkbot/checkbot.db" @@ -15,6 +18,8 @@ DEFAULT_BIND_PORT = "8080" MASTER_KEY_FILE = "master.key" MASTER_KEY_LEN = 32 +SESSION_SECRET_FILE = "session.secret" +SESSION_SECRET_MIN_LEN = 32 @dataclass(frozen=True) @@ -23,6 +28,7 @@ class FileConfig: db_path: Path bind_host: str bind_port: int + secure_cookies: bool = True def load_file_config() -> FileConfig: @@ -31,6 +37,7 @@ def load_file_config() -> FileConfig: db_path=Path(os.environ.get(ENV_DB_PATH, DEFAULT_DB_PATH)), bind_host=os.environ.get(ENV_BIND_HOST, DEFAULT_BIND_HOST), bind_port=int(os.environ.get(ENV_BIND_PORT, DEFAULT_BIND_PORT)), + secure_cookies=os.environ.get(ENV_SECURE_COOKIES, FLAG_ON) != FLAG_OFF, ) @@ -39,3 +46,10 @@ def load_master_key(path: Path) -> bytes: if len(key) != MASTER_KEY_LEN: raise ValueError(f"master key in {path} must be {MASTER_KEY_LEN} bytes") return key + + +def load_session_secret(path: Path) -> str: + secret = path.read_text().strip() + if len(secret) < SESSION_SECRET_MIN_LEN: + raise ValueError(f"session secret in {path} must have at least {SESSION_SECRET_MIN_LEN} characters") + return secret diff --git a/app/main.py b/app/main.py index 37c8e91..c8c4935 100644 --- a/app/main.py +++ b/app/main.py @@ -1,25 +1,77 @@ -from fastapi import FastAPI +from pathlib import Path + +from fastapi import Depends, FastAPI, Request from fastapi.responses import JSONResponse +from fastapi.staticfiles import StaticFiles +from starlette.exceptions import HTTPException as StarletteHTTPException +from starlette.middleware.sessions import SessionMiddleware from .crypto import SecretBox from .db import Database -from .fileconfig import MASTER_KEY_FILE, load_file_config, load_master_key +from .fileconfig import (MASTER_KEY_FILE, SESSION_SECRET_FILE, FileConfig, load_file_config, + load_master_key, load_session_secret) from .health import build_checks, run_checks +from .ratelimit import FailureLimiter from .settings import SettingsStore +from .users import UserStore +from .web import account, auth, pages +from .web.deps import LoginRequired, redirect, templates, verify_csrf + +STATIC_DIR = Path(__file__).parent / "web" / "static" +SESSION_COOKIE = "checkbot_session" +SESSION_MAX_AGE_SECONDS = 8 * 3600 + +SECURITY_HEADERS = { + "Content-Security-Policy": ("default-src 'self'; base-uri 'self'; form-action 'self'; " + "frame-ancestors 'none'"), + "X-Content-Type-Options": "nosniff", + "X-Frame-Options": "DENY", + "Referrer-Policy": "same-origin", +} -def create_app() -> FastAPI: - config = load_file_config() +def create_app(config: FileConfig | None = None) -> FastAPI: + config = config or load_file_config() box = SecretBox(load_master_key(config.config_dir / MASTER_KEY_FILE)) + session_secret = load_session_secret(config.config_dir / SESSION_SECRET_FILE) db = Database(config.db_path) - store = SettingsStore(db, box) - app = FastAPI(title="checkbot") - app.state.store = store + app = FastAPI(title="checkbot", dependencies=[Depends(verify_csrf)], + docs_url=None, redoc_url=None, openapi_url=None) + app.state.db = db + app.state.store = SettingsStore(db, box) + app.state.users = UserStore(db, box) + app.state.limiter = FailureLimiter() + app.state.totp_pending = {} + + app.add_middleware( + SessionMiddleware, secret_key=session_secret, session_cookie=SESSION_COOKIE, + max_age=SESSION_MAX_AGE_SECONDS, same_site="strict", https_only=config.secure_cookies, + ) + + @app.middleware("http") + async def security_headers(request: Request, call_next): + response = await call_next(request) + for name, value in SECURITY_HEADERS.items(): + response.headers.setdefault(name, value) + return response + + @app.exception_handler(LoginRequired) + async def login_required(request: Request, exc: LoginRequired): + return redirect("/login") + + @app.exception_handler(StarletteHTTPException) + async def http_error(request: Request, exc: StarletteHTTPException): + context = {"user": None, "csrf": "", "flashes": [], "status": exc.status_code, "detail": exc.detail} + return templates.TemplateResponse(request, "error.html", context, status_code=exc.status_code) @app.get("/health") async def health() -> JSONResponse: - result = await run_checks(build_checks(store)) + result = await run_checks(build_checks(app.state.store)) return JSONResponse(result, status_code=200 if result["ok"] else 503) + app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") + app.include_router(auth.router) + app.include_router(account.router) + app.include_router(pages.router) return app diff --git a/app/ratelimit.py b/app/ratelimit.py new file mode 100644 index 0000000..ef683e0 --- /dev/null +++ b/app/ratelimit.py @@ -0,0 +1,34 @@ +import time +from collections import defaultdict +from typing import Callable + +MAX_FAILURES = 5 +WINDOW_SECONDS = 300.0 + + +class FailureLimiter: + def __init__(self, max_failures: int = MAX_FAILURES, window: float = WINDOW_SECONDS, + clock: Callable[[], float] = time.monotonic): + self._max = max_failures + self._window = window + self._clock = clock + self._hits: dict[str, list[float]] = defaultdict(list) + + def _prune(self, key: str) -> list[float]: + cutoff = self._clock() - self._window + hits = [t for t in self._hits[key] if t > cutoff] + if hits: + self._hits[key] = hits + else: + self._hits.pop(key, None) + return hits + + def blocked(self, key: str) -> bool: + return len(self._prune(key)) >= self._max + + def record(self, key: str) -> None: + self._prune(key) + self._hits[key].append(self._clock()) + + def reset(self, key: str) -> None: + self._hits.pop(key, None) diff --git a/app/totp.py b/app/totp.py new file mode 100644 index 0000000..1beb5fa --- /dev/null +++ b/app/totp.py @@ -0,0 +1,52 @@ +import base64 +import hashlib +import hmac +import os +import struct +import time +from urllib.parse import quote + +STEP_SECONDS = 30 +DIGITS = 6 +WINDOW = 1 +SECRET_BYTES = 20 +ISSUER = "checkbot" + + +def generate_secret() -> str: + return base64.b32encode(os.urandom(SECRET_BYTES)).decode().rstrip("=") + + +def _decode(secret: str) -> bytes: + return base64.b32decode(secret + "=" * (-len(secret) % 8)) + + +def _code_at(secret: str, counter: int) -> str: + digest = hmac.new(_decode(secret), struct.pack(">Q", counter), hashlib.sha1).digest() + offset = digest[-1] & 0x0F + number = struct.unpack(">I", digest[offset:offset + 4])[0] & 0x7FFFFFFF + return str(number % 10 ** DIGITS).zfill(DIGITS) + + +def _counter(now: float | None) -> int: + return int((time.time() if now is None else now) // STEP_SECONDS) + + +def current_code(secret: str, now: float | None = None) -> str: + return _code_at(secret, _counter(now)) + + +def verify(secret: str, code: str, now: float | None = None) -> bool: + code = code.replace(" ", "") + if len(code) != DIGITS or not code.isdigit(): + return False + counter = _counter(now) + return any( + hmac.compare_digest(_code_at(secret, counter + offset), code) + for offset in range(-WINDOW, WINDOW + 1) + ) + + +def provisioning_uri(secret: str, username: str) -> str: + return (f"otpauth://totp/{quote(ISSUER)}:{quote(username)}" + f"?secret={secret}&issuer={quote(ISSUER)}&digits={DIGITS}&period={STEP_SECONDS}") diff --git a/app/users.py b/app/users.py new file mode 100644 index 0000000..b4cd890 --- /dev/null +++ b/app/users.py @@ -0,0 +1,184 @@ +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") diff --git a/app/web/__init__.py b/app/web/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/web/account.py b/app/web/account.py new file mode 100644 index 0000000..cb248ed --- /dev/null +++ b/app/web/account.py @@ -0,0 +1,89 @@ +import time + +from fastapi import APIRouter, Depends, Form, Request + +from .. import totp +from ..users import MIN_PASSWORD_LEN, User, UserError +from .deps import FLASH_ERROR, FLASH_OK, SESSION_VERSION, current_user, flash, redirect, render + +router = APIRouter() + +TOTP_PENDING_TTL_SECONDS = 600 + +MSG_PASSWORD_MISMATCH = "Die neuen Passwörter stimmen nicht überein." +MSG_PASSWORD_WRONG = "Aktuelles Passwort falsch." +MSG_PASSWORD_CHANGED = "Passwort geändert." +MSG_TOTP_ENABLED = "Zwei-Faktor-Anmeldung aktiviert." +MSG_TOTP_DISABLED = "Zwei-Faktor-Anmeldung deaktiviert." +MSG_TOTP_INVALID = "Code ungültig." +MSG_TOTP_EXPIRED = "Einrichtung abgelaufen, bitte neu starten." + + +def _pending_secret(request: Request, user: User) -> str | None: + entry = request.app.state.totp_pending.get(user.id) + if entry is None: + return None + secret, expires = entry + if time.time() > expires: + request.app.state.totp_pending.pop(user.id, None) + return None + return secret + + +@router.get("/account") +def account(request: Request, user: User = Depends(current_user)): + secret = None if user.has_totp else _pending_secret(request, user) + uri = totp.provisioning_uri(secret, user.username) if secret else None + return render(request, "account.html", user=user, min_len=MIN_PASSWORD_LEN, + totp_secret=secret, totp_uri=uri) + + +@router.post("/account/password") +def change_password(request: Request, current: str = Form(""), new: str = Form(""), + confirm: str = Form(""), user: User = Depends(current_user)): + users = request.app.state.users + if new != confirm: + flash(request, MSG_PASSWORD_MISMATCH, FLASH_ERROR) + elif users.authenticate(user.username, current) is None: + flash(request, MSG_PASSWORD_WRONG, FLASH_ERROR) + else: + try: + users.set_password(user.id, new, actor=user.username) + except UserError as exc: + flash(request, str(exc), FLASH_ERROR) + else: + request.session[SESSION_VERSION] = users.get(user.id).session_version + flash(request, MSG_PASSWORD_CHANGED, FLASH_OK) + return redirect("/account") + + +@router.post("/account/totp/start") +def totp_start(request: Request, user: User = Depends(current_user)): + request.app.state.totp_pending[user.id] = ( + totp.generate_secret(), time.time() + TOTP_PENDING_TTL_SECONDS) + return redirect("/account") + + +@router.post("/account/totp/confirm") +def totp_confirm(request: Request, code: str = Form(""), user: User = Depends(current_user)): + secret = _pending_secret(request, user) + if secret is None: + flash(request, MSG_TOTP_EXPIRED, FLASH_ERROR) + elif not totp.verify(secret, code): + flash(request, MSG_TOTP_INVALID, FLASH_ERROR) + else: + request.app.state.users.set_totp(user.id, secret, actor=user.username) + request.app.state.totp_pending.pop(user.id, None) + flash(request, MSG_TOTP_ENABLED, FLASH_OK) + return redirect("/account") + + +@router.post("/account/totp/disable") +def totp_disable(request: Request, password: str = Form(""), user: User = Depends(current_user)): + users = request.app.state.users + if users.authenticate(user.username, password) is None: + flash(request, MSG_PASSWORD_WRONG, FLASH_ERROR) + else: + users.set_totp(user.id, None, actor=user.username) + flash(request, MSG_TOTP_DISABLED, FLASH_OK) + return redirect("/account") diff --git a/app/web/auth.py b/app/web/auth.py new file mode 100644 index 0000000..98cedba --- /dev/null +++ b/app/web/auth.py @@ -0,0 +1,99 @@ +import time + +from fastapi import APIRouter, Depends, Form, Request + +from .. import audit, totp +from ..users import User +from .deps import (FLASH_ERROR, SESSION_UID, SESSION_VERSION, current_user, flash, redirect, + render) + +router = APIRouter() + +PENDING_UID = "pending_uid" +PENDING_TS = "pending_ts" +PENDING_TTL_SECONDS = 300 +STATUS_UNAUTHORIZED = 401 +STATUS_TOO_MANY = 429 + +MSG_LOGIN_FAILED = "Anmeldung fehlgeschlagen." +MSG_BLOCKED = "Zu viele Fehlversuche. Bitte später erneut versuchen." +MSG_TOTP_INVALID = "Code ungültig." +AUDIT_TARGET_MAX = 64 + + +def _complete_login(request: Request, user: User): + request.session.clear() + request.session[SESSION_UID] = user.id + request.session[SESSION_VERSION] = user.session_version + audit.record(request.app.state.db, user.username, "login", user.username) + return redirect("/") + + +def _pending_uid(request: Request) -> int | None: + uid = request.session.get(PENDING_UID) + started = request.session.get(PENDING_TS, 0) + if not uid or time.time() - started > PENDING_TTL_SECONDS: + return None + return uid + + +@router.get("/login") +def login_form(request: Request): + return render(request, "login.html") + + +@router.post("/login") +def login(request: Request, username: str = Form(""), password: str = Form("")): + state = request.app.state + key = username.lower() + if state.limiter.blocked(key): + flash(request, MSG_BLOCKED, FLASH_ERROR) + return render(request, "login.html", status_code=STATUS_TOO_MANY) + user = state.users.authenticate(username, password) + if user is None: + state.limiter.record(key) + audit.record(state.db, username[:AUDIT_TARGET_MAX], "login.failed", username[:AUDIT_TARGET_MAX]) + flash(request, MSG_LOGIN_FAILED, FLASH_ERROR) + return render(request, "login.html", status_code=STATUS_UNAUTHORIZED) + state.limiter.reset(key) + if user.has_totp: + request.session.clear() + request.session[PENDING_UID] = user.id + request.session[PENDING_TS] = time.time() + return redirect("/login/totp") + return _complete_login(request, user) + + +@router.get("/login/totp") +def totp_form(request: Request): + if _pending_uid(request) is None: + return redirect("/login") + return render(request, "login_totp.html") + + +@router.post("/login/totp") +def totp_submit(request: Request, code: str = Form("")): + uid = _pending_uid(request) + if uid is None: + return redirect("/login") + state = request.app.state + key = f"totp:{uid}" + if state.limiter.blocked(key): + request.session.clear() + flash(request, MSG_BLOCKED, FLASH_ERROR) + return render(request, "login.html", status_code=STATUS_TOO_MANY) + user = state.users.get(uid) + secret = state.users.totp_secret(uid) + if user is None or user.disabled or not secret or not totp.verify(secret, code): + state.limiter.record(key) + flash(request, MSG_TOTP_INVALID, FLASH_ERROR) + return render(request, "login_totp.html", status_code=STATUS_UNAUTHORIZED) + state.limiter.reset(key) + return _complete_login(request, user) + + +@router.post("/logout") +def logout(request: Request, user: User = Depends(current_user)): + audit.record(request.app.state.db, user.username, "logout", user.username) + request.session.clear() + return redirect("/login") diff --git a/app/web/deps.py b/app/web/deps.py new file mode 100644 index 0000000..817e579 --- /dev/null +++ b/app/web/deps.py @@ -0,0 +1,89 @@ +import secrets +from datetime import datetime +from pathlib import Path + +from fastapi import Depends, HTTPException, Request +from fastapi.responses import RedirectResponse +from fastapi.templating import Jinja2Templates +from markupsafe import Markup, escape + +from ..users import User + +BASE_DIR = Path(__file__).parent +SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"}) + +CSRF_FIELD = "csrf" +CSRF_HEADER = "x-csrf-token" +SESSION_UID = "uid" +SESSION_VERSION = "sv" +SESSION_CSRF = "csrf" +SESSION_FLASH = "flash" + +FLASH_INFO = "info" +FLASH_OK = "ok" +FLASH_ERROR = "error" + +TS_FORMAT = "%d.%m.%Y %H:%M:%S" +REDIRECT_STATUS = 303 + +templates = Jinja2Templates(directory=str(BASE_DIR / "templates")) +templates.env.filters["fmt_ts"] = lambda ts: datetime.fromtimestamp(ts).strftime(TS_FORMAT) + + +class LoginRequired(Exception): + pass + + +def redirect(url: str) -> RedirectResponse: + return RedirectResponse(url, status_code=REDIRECT_STATUS) + + +def csrf_token(request: Request) -> str: + token = request.session.get(SESSION_CSRF) + if not token: + token = secrets.token_urlsafe(32) + request.session[SESSION_CSRF] = token + return token + + +async def verify_csrf(request: Request) -> None: + if request.method in SAFE_METHODS: + return + expected = request.session.get(SESSION_CSRF) + supplied = request.headers.get(CSRF_HEADER) + if supplied is None: + supplied = (await request.form()).get(CSRF_FIELD) + if not expected or not supplied or not secrets.compare_digest(str(supplied), expected): + raise HTTPException(403, "CSRF-Prüfung fehlgeschlagen") + + +def flash(request: Request, text: str, kind: str = FLASH_INFO) -> None: + request.session.setdefault(SESSION_FLASH, []).append({"kind": kind, "text": text}) + + +def render(request: Request, name: str, user: User | None = None, status_code: int = 200, **context): + token = csrf_token(request) + page_context = { + "user": user, + "csrf": token, + "csrf_input": Markup('') % (CSRF_FIELD, escape(token)), + "flashes": request.session.pop(SESSION_FLASH, []), + **context, + } + return templates.TemplateResponse(request, name, page_context, status_code=status_code) + + +def current_user(request: Request) -> User: + uid = request.session.get(SESSION_UID) + user = request.app.state.users.get(uid) if uid else None + if user is None or user.disabled or user.session_version != request.session.get(SESSION_VERSION): + request.session.pop(SESSION_UID, None) + request.session.pop(SESSION_VERSION, None) + raise LoginRequired + return user + + +def require_admin(user: User = Depends(current_user)) -> User: + if not user.is_admin: + raise HTTPException(403, "Nur für Admins") + return user diff --git a/app/web/pages.py b/app/web/pages.py new file mode 100644 index 0000000..0667e88 --- /dev/null +++ b/app/web/pages.py @@ -0,0 +1,319 @@ +import re +from dataclasses import dataclass +from typing import Awaitable, Callable + +from fastapi import APIRouter, Depends, Form, HTTPException, Request +from fastapi.responses import JSONResponse + +from .. import audit +from ..checkmk import CheckmkClient +from ..constants import (CHECKMK_CA_FILE, CHECKMK_SECRET, CHECKMK_SITE, CHECKMK_URL, CHECKMK_USER, + CHECKMK_VERIFY_TLS, DEFAULT_OLLAMA_MODEL, DEFAULT_OLLAMA_TIMEOUT, FLAG_OFF, + FLAG_ON, OLLAMA_CA_FILE, OLLAMA_MODEL_AGENT, OLLAMA_MODEL_ALERT, + OLLAMA_TIMEOUT, OLLAMA_TOKEN, OLLAMA_URL) +from ..health import build_checks, run_checks +from ..ollama import OllamaClient +from ..users import MIN_PASSWORD_LEN, ROLES, User, UserError +from .deps import (FLASH_ERROR, FLASH_OK, current_user, flash, redirect, render, require_admin) + +router = APIRouter() + +SITE_RE = re.compile(r"^[A-Za-z0-9_-]+$") +URL_RE = re.compile(r"^https?://[^\s/]+") +AUDIT_LIMIT = 200 +STATUS_INVALID = 422 +ERROR_DETAIL_MAX = 300 +MODEL_LIST_PREVIEW = 10 +LATEST_TAG = ":latest" + +MSG_SELF_DISABLE = "Das eigene Konto kann nicht gesperrt werden." +MSG_SELF_DELETE = "Das eigene Konto kann nicht gelöscht werden." +MSG_SAVED = "Gespeichert." + + +@dataclass(frozen=True) +class Field: + key: str + label: str + kind: str = "text" + required: bool = False + secret: bool = False + placeholder: str = "" + default: str = "" + + @property + def name(self) -> str: + return self.key.replace(".", "_") + + +@dataclass(frozen=True) +class Section: + slug: str + title: str + fields: tuple[Field, ...] + validate: Callable[[dict], list[str]] + test: Callable[[dict], Awaitable[str]] + + +def _validate_checkmk(values: dict) -> list[str]: + errors = [] + if not URL_RE.match(values[CHECKMK_URL]): + errors.append("URL muss mit http:// oder https:// beginnen.") + if not SITE_RE.match(values[CHECKMK_SITE]): + errors.append("Site darf nur Buchstaben, Ziffern, _ und - enthalten.") + return errors + + +def _validate_ollama(values: dict) -> list[str]: + errors = [] + if not URL_RE.match(values[OLLAMA_URL]): + errors.append("URL muss mit http:// oder https:// beginnen.") + try: + if float(values.get(OLLAMA_TIMEOUT) or DEFAULT_OLLAMA_TIMEOUT) <= 0: + raise ValueError + except ValueError: + errors.append("Timeout muss eine positive Zahl sein.") + return errors + + +async def _test_checkmk(values: dict) -> str: + verify = False if values.get(CHECKMK_VERIFY_TLS) == FLAG_OFF else (values.get(CHECKMK_CA_FILE) or True) + client = CheckmkClient(values[CHECKMK_URL], values[CHECKMK_SITE], values[CHECKMK_USER], + values[CHECKMK_SECRET], verify=verify) + try: + info = await client.version() + problems = await client.problem_services() + finally: + await client.aclose() + return f"Checkmk {info.get('versions', {}).get('checkmk', '?')}, {len(problems)} Probleme gelesen" + + +def _has_model(name: str, available: list[str]) -> bool: + return name in available or f"{name}{LATEST_TAG}" in available + + +async def _test_ollama(values: dict) -> str: + client = OllamaClient( + values[OLLAMA_URL], values.get(OLLAMA_TOKEN) or None, + timeout=float(values.get(OLLAMA_TIMEOUT) or DEFAULT_OLLAMA_TIMEOUT), retries=0, + verify=values.get(OLLAMA_CA_FILE) or True, + ) + try: + version = await client.version() + models = await client.list_models() + finally: + await client.aclose() + missing = [m for m in (values.get(OLLAMA_MODEL_ALERT), values.get(OLLAMA_MODEL_AGENT)) + if m and not _has_model(m, models)] + if missing: + raise RuntimeError(f"Modell nicht vorhanden: {', '.join(missing)}") + return f"Ollama {version}, {len(models)} Modelle: {', '.join(models[:MODEL_LIST_PREVIEW])}" + + +SECTIONS = { + "checkmk": Section( + slug="checkmk", + title="Checkmk", + fields=( + Field(CHECKMK_URL, "URL", required=True, placeholder="https://checkmk.example.de"), + Field(CHECKMK_SITE, "Site", required=True, placeholder="prod"), + Field(CHECKMK_USER, "Automation-User", required=True), + Field(CHECKMK_SECRET, "Automation-Secret", kind="password", required=True, secret=True), + Field(CHECKMK_CA_FILE, "CA-Datei (optional)", placeholder="/etc/ssl/certs/intern-ca.pem"), + Field(CHECKMK_VERIFY_TLS, "TLS-Zertifikat prüfen", kind="checkbox"), + ), + validate=_validate_checkmk, + test=_test_checkmk, + ), + "ollama": Section( + slug="ollama", + title="Ollama", + fields=( + Field(OLLAMA_URL, "URL", required=True, placeholder="http://office.sysops.de:11434"), + Field(OLLAMA_TOKEN, "Bearer-Token (optional)", kind="password", secret=True), + Field(OLLAMA_TIMEOUT, "Timeout in Sekunden", default=DEFAULT_OLLAMA_TIMEOUT), + Field(OLLAMA_CA_FILE, "CA-Datei (optional)"), + Field(OLLAMA_MODEL_ALERT, "Modell für Alarmtexte", required=True, default=DEFAULT_OLLAMA_MODEL), + Field(OLLAMA_MODEL_AGENT, "Modell für den Agenten", required=True, default=DEFAULT_OLLAMA_MODEL), + ), + validate=_validate_ollama, + test=_test_ollama, + ), +} + + +def _section(slug: str) -> Section: + section = SECTIONS.get(slug) + if section is None: + raise HTTPException(404, "Unbekannter Bereich") + return section + + +def _section_done(store, section: Section) -> bool: + return all(store.get(f.key) for f in section.fields if f.required) + + +def _collect(form, section: Section) -> dict[str, str | None]: + submitted: dict[str, str | None] = {} + for field in section.fields: + if field.kind == "checkbox": + submitted[field.key] = FLAG_ON if form.get(field.name) else FLAG_OFF + continue + raw = str(form.get(field.name, "")).strip() + submitted[field.key] = (raw or None) if field.secret else raw + return submitted + + +def _effective(store, submitted: dict[str, str | None]) -> dict[str, str]: + return {key: (value if value is not None else store.get(key) or "") for key, value in submitted.items()} + + +def _check(section: Section, values: dict[str, str]) -> list[str]: + missing = [f"{f.label} fehlt." for f in section.fields if f.required and not values.get(f.key)] + return missing or section.validate(values) + + +def _form_values(section: Section, values: dict[str, str]) -> dict: + shown = {} + for field in section.fields: + if field.secret: + continue + if field.kind == "checkbox": + shown[field.name] = values.get(field.key, FLAG_ON) == FLAG_ON + else: + shown[field.name] = values.get(field.key, field.default) + return shown + + +def _stored_values(store, section: Section) -> dict[str, str]: + return {f.key: store.get(f.key, FLAG_ON if f.kind == "checkbox" else f.default) or "" + for f in section.fields if not f.secret} + + +@router.get("/") +async def dashboard(request: Request, user: User = Depends(current_user)): + store = request.app.state.store + result = await run_checks(build_checks(store)) + setup_open = user.is_admin and not all(_section_done(store, s) for s in SECTIONS.values()) + return render(request, "dashboard.html", user=user, result=result, setup_open=setup_open) + + +@router.get("/setup") +def setup(request: Request, user: User = Depends(require_admin)): + store = request.app.state.store + steps = [{"slug": s.slug, "title": s.title, "done": _section_done(store, s)} for s in SECTIONS.values()] + return render(request, "setup.html", user=user, steps=steps) + + +@router.get("/settings/{slug}") +def settings_form(request: Request, slug: str, user: User = Depends(require_admin)): + section = _section(slug) + store = request.app.state.store + return render(request, "settings.html", user=user, section=section, + values=_form_values(section, _stored_values(store, section)), + secret_set={f.name: store.is_set(f.key) for f in section.fields if f.secret}) + + +@router.post("/settings/{slug}") +async def settings_save(request: Request, slug: str, user: User = Depends(require_admin)): + section = _section(slug) + store = request.app.state.store + submitted = _collect(await request.form(), section) + effective = _effective(store, submitted) + errors = _check(section, effective) + if errors: + for message in errors: + flash(request, message, FLASH_ERROR) + return render(request, "settings.html", user=user, section=section, status_code=STATUS_INVALID, + values=_form_values(section, effective), + secret_set={f.name: store.is_set(f.key) for f in section.fields if f.secret}) + fields = {f.key: f for f in section.fields} + for key, value in submitted.items(): + if value is None: + continue + if fields[key].secret or store.get(key) != value: + store.set(key, value, user.username, secret=fields[key].secret) + flash(request, MSG_SAVED, FLASH_OK) + return redirect(f"/settings/{slug}") + + +@router.post("/settings/{slug}/test") +async def settings_test(request: Request, slug: str, user: User = Depends(require_admin)): + section = _section(slug) + store = request.app.state.store + effective = _effective(store, _collect(await request.form(), section)) + errors = _check(section, effective) + if errors: + return JSONResponse({"ok": False, "detail": " ".join(errors)}) + try: + detail = await section.test(effective) + except Exception as exc: + return JSONResponse({"ok": False, "detail": f"{type(exc).__name__}: {exc}"[:ERROR_DETAIL_MAX]}) + audit.record(request.app.state.db, user.username, "settings.test", slug) + return JSONResponse({"ok": True, "detail": detail}) + + +@router.get("/users") +def users_page(request: Request, user: User = Depends(require_admin)): + return render(request, "users.html", user=user, users=request.app.state.users.list(), + roles=ROLES, min_len=MIN_PASSWORD_LEN) + + +@router.post("/users") +def users_create(request: Request, username: str = Form(""), password: str = Form(""), + role: str = Form(ROLES[-1]), admin: User = Depends(require_admin)): + try: + request.app.state.users.create(username.strip(), password, role, actor=admin.username) + flash(request, f"Benutzer '{username.strip()}' angelegt.", FLASH_OK) + except UserError as exc: + flash(request, str(exc), FLASH_ERROR) + return redirect("/users") + + +def _user_action(request: Request, action: Callable[[], None], ok_message: str): + try: + action() + flash(request, ok_message, FLASH_OK) + except UserError as exc: + flash(request, str(exc), FLASH_ERROR) + return redirect("/users") + + +@router.post("/users/{user_id}/role") +def users_role(request: Request, user_id: int, role: str = Form(""), admin: User = Depends(require_admin)): + users = request.app.state.users + return _user_action(request, lambda: users.set_role(user_id, role, admin.username), "Rolle geändert.") + + +@router.post("/users/{user_id}/disable") +def users_disable(request: Request, user_id: int, disabled: str = Form(FLAG_ON), + admin: User = Depends(require_admin)): + if user_id == admin.id: + flash(request, MSG_SELF_DISABLE, FLASH_ERROR) + return redirect("/users") + users = request.app.state.users + turn_off = disabled == FLAG_ON + return _user_action(request, lambda: users.set_disabled(user_id, turn_off, admin.username), + "Benutzer gesperrt." if turn_off else "Benutzer entsperrt.") + + +@router.post("/users/{user_id}/password") +def users_password(request: Request, user_id: int, password: str = Form(""), + admin: User = Depends(require_admin)): + users = request.app.state.users + return _user_action(request, lambda: users.set_password(user_id, password, admin.username), + "Passwort gesetzt.") + + +@router.post("/users/{user_id}/delete") +def users_delete(request: Request, user_id: int, admin: User = Depends(require_admin)): + if user_id == admin.id: + flash(request, MSG_SELF_DELETE, FLASH_ERROR) + return redirect("/users") + users = request.app.state.users + return _user_action(request, lambda: users.delete(user_id, admin.username), "Benutzer gelöscht.") + + +@router.get("/audit") +def audit_page(request: Request, user: User = Depends(require_admin)): + return render(request, "audit.html", user=user, entries=audit.recent(request.app.state.db, AUDIT_LIMIT)) diff --git a/app/web/static/app.css b/app/web/static/app.css new file mode 100644 index 0000000..a0536ba --- /dev/null +++ b/app/web/static/app.css @@ -0,0 +1,178 @@ +:root { + --bg: #f4f6f8; + --surface: #ffffff; + --text: #1c2430; + --muted: #5b6675; + --border: #d5dbe3; + --accent: #1d5fd1; + --accent-text: #ffffff; + --ok: #1b7f3b; + --ok-bg: #e3f4e8; + --error: #b3261e; + --error-bg: #fbe6e4; + --info: #1d5fd1; + --info-bg: #e5eefb; + --warn: #8a5a00; + --warn-bg: #fdf0d5; + --gutter: 16px; +} + +@media (prefers-color-scheme: dark) { + :root { + --bg: #12161c; + --surface: #1b212a; + --text: #e4e8ee; + --muted: #9aa5b4; + --border: #2f3844; + --accent: #5b9bff; + --accent-text: #0b1220; + --ok: #62d083; + --ok-bg: #16311f; + --error: #ff8a80; + --error-bg: #3a1c1a; + --info: #8db8ff; + --info-bg: #182a47; + --warn: #f2c15f; + --warn-bg: #3a2c0e; + } +} + +* { box-sizing: border-box; } + +html { -webkit-text-size-adjust: 100%; } + +body { + margin: 0; + background: var(--bg); + color: var(--text); + font: 16px/1.5 system-ui, -apple-system, "Segoe UI", Roboto, sans-serif; + overflow-x: hidden; +} + +a { color: var(--accent); } + +h1 { font-size: 1.5rem; margin: 0 0 1rem; } +h2 { font-size: 1.15rem; margin: 0 0 .75rem; } + +.container { + max-width: 960px; + margin: 0 auto; + padding: var(--gutter); +} + +.topbar { + background: var(--surface); + border-bottom: 1px solid var(--border); +} + +.topbar-inner { + max-width: 960px; + margin: 0 auto; + padding: 10px var(--gutter); + display: flex; + flex-wrap: wrap; + align-items: center; + gap: 8px 20px; +} + +.brand { font-weight: 700; text-decoration: none; color: var(--text); } + +.nav { display: flex; flex-wrap: wrap; align-items: center; gap: 6px 16px; } +.nav a { text-decoration: none; } +.nav a:hover { text-decoration: underline; } + +.card { + background: var(--surface); + border: 1px solid var(--border); + border-radius: 8px; + padding: var(--gutter); + margin-bottom: var(--gutter); +} + +.card.narrow { max-width: 420px; margin-left: auto; margin-right: auto; } + +.muted { color: var(--muted); } + +form.inline { display: inline; margin: 0; } +.row { display: flex; flex-wrap: wrap; gap: 8px; align-items: center; margin-top: 6px; } +.row select, .row input { width: auto; flex: 1 1 140px; margin: 0; } + +label { display: block; margin: 12px 0 4px; font-weight: 600; } +label.check { display: flex; align-items: center; gap: 8px; font-weight: 400; } + +input[type="text"], input[type="password"], select { + width: 100%; + padding: 9px 10px; + font: inherit; + color: var(--text); + background: var(--bg); + border: 1px solid var(--border); + border-radius: 6px; +} + +input[type="checkbox"] { width: 1.1rem; height: 1.1rem; } + +input:focus, select:focus, button:focus-visible { + outline: 2px solid var(--accent); + outline-offset: 1px; +} + +.actions { display: flex; flex-wrap: wrap; gap: 8px; margin-top: 16px; } + +button { + font: inherit; + padding: 8px 14px; + border-radius: 6px; + border: 1px solid var(--accent); + background: var(--accent); + color: var(--accent-text); + cursor: pointer; +} + +button.secondary { background: transparent; color: var(--accent); } +button.danger { background: transparent; color: var(--error); border-color: var(--error); } +button.link { + background: none; + border: none; + padding: 0; + color: var(--accent); + text-decoration: underline; +} +button:disabled { opacity: .6; cursor: progress; } + +.flash { + padding: 10px 12px; + border-radius: 6px; + border: 1px solid; + margin-bottom: var(--gutter); +} +.flash.ok { color: var(--ok); background: var(--ok-bg); border-color: var(--ok); } +.flash.error { color: var(--error); background: var(--error-bg); border-color: var(--error); } +.flash.info { color: var(--info); background: var(--info-bg); border-color: var(--info); } + +.test-result { min-height: 1.5em; margin: 12px 0 0; overflow-wrap: anywhere; } +.test-result.ok { color: var(--ok); } +.test-result.error { color: var(--error); } + +.table-wrap { overflow-x: auto; } +table { width: 100%; border-collapse: collapse; } +th, td { text-align: left; padding: 8px; border-bottom: 1px solid var(--border); vertical-align: top; } +th { color: var(--muted); font-weight: 600; white-space: nowrap; } + +.badge { + display: inline-block; + padding: 1px 8px; + border-radius: 999px; + font-size: .85rem; + font-weight: 600; + vertical-align: middle; +} +.badge.ok { color: var(--ok); background: var(--ok-bg); } +.badge.error { color: var(--error); background: var(--error-bg); } +.badge.info { color: var(--info); background: var(--info-bg); } +.badge.warn { color: var(--warn); background: var(--warn-bg); } + +.steps { padding-left: 1.25rem; } +.steps li { margin-bottom: 8px; } + +code.break { overflow-wrap: anywhere; word-break: break-all; } diff --git a/app/web/static/app.js b/app/web/static/app.js new file mode 100644 index 0000000..377a58e --- /dev/null +++ b/app/web/static/app.js @@ -0,0 +1,66 @@ +"use strict"; + +const CSRF_HEADER = "X-CSRF-Token"; +const CLASS_OK = "ok"; +const CLASS_ERROR = "error"; +const MSG_TESTING = "Teste …"; +const MSG_NETWORK_ERROR = "Netzwerkfehler: "; +const MSG_BAD_RESPONSE = "Unerwartete Antwort (HTTP "; + +function csrfToken() { + const meta = document.querySelector('meta[name="csrf"]'); + return meta ? meta.content : ""; +} + +function showResult(target, kind, text) { + target.classList.remove(CLASS_OK, CLASS_ERROR); + if (kind) { + target.classList.add(kind); + } + target.textContent = text; +} + +async function runTest(button) { + const form = button.closest("form"); + const target = form.querySelector(".test-result"); + button.disabled = true; + showResult(target, null, MSG_TESTING); + try { + const response = await fetch(button.dataset.testUrl, { + method: "POST", + headers: { [CSRF_HEADER]: csrfToken() }, + body: new FormData(form), + credentials: "same-origin", + }); + let data = null; + try { + data = await response.json(); + } catch (_) { + data = null; + } + if (data && typeof data.ok === "boolean") { + showResult(target, data.ok ? CLASS_OK : CLASS_ERROR, String(data.detail ?? "")); + } else { + showResult(target, CLASS_ERROR, MSG_BAD_RESPONSE + response.status + ")"); + } + } catch (error) { + showResult(target, CLASS_ERROR, MSG_NETWORK_ERROR + error.message); + } finally { + button.disabled = false; + } +} + +document.addEventListener("click", (event) => { + const button = event.target.closest("button[data-test-url]"); + if (button) { + event.preventDefault(); + runTest(button); + } +}); + +document.addEventListener("submit", (event) => { + const form = event.target; + if (form.matches("form[data-confirm]") && !window.confirm(form.dataset.confirm)) { + event.preventDefault(); + } +}); diff --git a/app/web/templates/account.html b/app/web/templates/account.html new file mode 100644 index 0000000..ace2127 --- /dev/null +++ b/app/web/templates/account.html @@ -0,0 +1,47 @@ +{% extends "base.html" %} +{% block title %}Konto{% endblock %} +{% block content %} +

Konto: {{ user.username }}

+
+

Passwort ändern

+
+ {{ csrf_input }} + + + + + + +
+
+
+ +
+

Zwei-Faktor-Anmeldung (TOTP)

+ {% if user.has_totp %} +

Die Zwei-Faktor-Anmeldung ist aktiv.

+
+ {{ csrf_input }} + + +
+
+ {% elif totp_secret %} +

Secret in der Authenticator-App eintragen (oder die URI importieren), dann den angezeigten Code bestätigen.

+

Secret: {{ totp_secret }}

+

URI: {{ totp_uri }}

+
+ {{ csrf_input }} + + +
+
+ {% else %} +

Die Zwei-Faktor-Anmeldung ist nicht aktiv.

+
+ {{ csrf_input }} + +
+ {% endif %} +
+{% endblock %} diff --git a/app/web/templates/audit.html b/app/web/templates/audit.html new file mode 100644 index 0000000..7be9ee0 --- /dev/null +++ b/app/web/templates/audit.html @@ -0,0 +1,26 @@ +{% extends "base.html" %} +{% block title %}Audit{% endblock %} +{% block content %} +

Audit-Log

+
+
+ + + + {% for e in entries %} + + + + + + + + + {% else %} + + {% endfor %} + +
ZeitBenutzerAktionZielAltNeu
{{ e.ts | fmt_ts }}{{ e.user }}{{ e.action }}{{ e.target }}{{ e.old_value if e.old_value is not none else "–" }}{{ e.new_value if e.new_value is not none else "–" }}
Keine Einträge.
+
+
+{% endblock %} diff --git a/app/web/templates/base.html b/app/web/templates/base.html new file mode 100644 index 0000000..d958935 --- /dev/null +++ b/app/web/templates/base.html @@ -0,0 +1,41 @@ + + + + + + +{% block title %}checkbot{% endblock %} – checkbot + + + + +
+
+ checkbot + {% if user %} + + {% endif %} +
+
+
+ {% for f in flashes %} +
{{ f.text }}
+ {% endfor %} + {% block content %}{% endblock %} +
+ + diff --git a/app/web/templates/dashboard.html b/app/web/templates/dashboard.html new file mode 100644 index 0000000..bc70059 --- /dev/null +++ b/app/web/templates/dashboard.html @@ -0,0 +1,27 @@ +{% extends "base.html" %} +{% block title %}Dashboard{% endblock %} +{% block content %} +

Dashboard

+{% if setup_open %} +
Die Einrichtung ist noch nicht abgeschlossen. Zum Setup-Assistenten
+{% endif %} +
+

Systemstatus + {{ 'Alles in Ordnung' if result.ok else 'Probleme' }} +

+
+ + + + {% for name, check in result.checks.items() %} + + + + + + {% endfor %} + +
PrüfungStatusDetails
{{ name }}{{ 'OK' if check.ok else 'Fehler' }}{{ check.detail }}
+
+
+{% endblock %} diff --git a/app/web/templates/error.html b/app/web/templates/error.html new file mode 100644 index 0000000..c4c652f --- /dev/null +++ b/app/web/templates/error.html @@ -0,0 +1,9 @@ +{% extends "base.html" %} +{% block title %}Fehler {{ status }}{% endblock %} +{% block content %} +
+

Fehler {{ status }}

+

{{ detail }}

+

Zur Startseite

+
+{% endblock %} diff --git a/app/web/templates/login.html b/app/web/templates/login.html new file mode 100644 index 0000000..16aa20d --- /dev/null +++ b/app/web/templates/login.html @@ -0,0 +1,15 @@ +{% extends "base.html" %} +{% block title %}Anmelden{% endblock %} +{% block content %} +
+

Anmelden

+
+ {{ csrf_input }} + + + + +
+
+
+{% endblock %} diff --git a/app/web/templates/login_totp.html b/app/web/templates/login_totp.html new file mode 100644 index 0000000..372cead --- /dev/null +++ b/app/web/templates/login_totp.html @@ -0,0 +1,14 @@ +{% extends "base.html" %} +{% block title %}Zwei-Faktor-Code{% endblock %} +{% block content %} +
+

Zwei-Faktor-Code

+

Bitte den 6-stelligen Code aus der Authenticator-App eingeben.

+
+ {{ csrf_input }} + + +
+
+
+{% endblock %} diff --git a/app/web/templates/settings.html b/app/web/templates/settings.html new file mode 100644 index 0000000..ec28327 --- /dev/null +++ b/app/web/templates/settings.html @@ -0,0 +1,31 @@ +{% extends "base.html" %} +{% block title %}{{ section.title }}{% endblock %} +{% block content %} +

{{ section.title }}

+
+
+ {{ csrf_input }} + {% for field in section.fields %} + {% if field.kind == "checkbox" %} + + {% elif field.secret %} + + + {% else %} + + + {% endif %} + {% endfor %} +
+ + +
+

+
+
+{% endblock %} diff --git a/app/web/templates/setup.html b/app/web/templates/setup.html new file mode 100644 index 0000000..46bfafd --- /dev/null +++ b/app/web/templates/setup.html @@ -0,0 +1,16 @@ +{% extends "base.html" %} +{% block title %}Setup{% endblock %} +{% block content %} +

Setup-Assistent

+
+
    + {% for step in steps %} +
  1. + {{ step.title }} + {{ 'erledigt' if step.done else 'offen' }} +
    2. + {% endfor %} +
+

Die Anbindung der Chat-Backends (Telegram, Matrix) folgt in einer späteren Phase.

+
+{% endblock %} diff --git a/app/web/templates/users.html b/app/web/templates/users.html new file mode 100644 index 0000000..134b672 --- /dev/null +++ b/app/web/templates/users.html @@ -0,0 +1,68 @@ +{% extends "base.html" %} +{% block title %}Benutzer{% endblock %} +{% block content %} +

Benutzer

+
+

Benutzer anlegen

+
+ {{ csrf_input }} + + + + + + +
+
+
+ +
+

Bestehende Benutzer

+
+ + + + {% for u in users %} + + + + + + + + {% endfor %} + +
NameRolle2FAStatusAktionen
{{ u.username }}{% if u.id == user.id %} du{% endif %} +
+ {{ csrf_input }} + + +
+ 		{{ 'ja' if u.has_totp else 'nein' }}{{ 'gesperrt' if u.disabled else 'aktiv' }} +
+ {{ csrf_input }} + + +
+ {% if u.id != user.id %} +
+
+ {{ csrf_input }} + + +
+
+ {{ csrf_input }} + +
+
+ {% endif %} +
+
+
+{% endblock %} diff --git a/tests/test_ratelimit.py b/tests/test_ratelimit.py new file mode 100644 index 0000000..cd2c9d3 --- /dev/null +++ b/tests/test_ratelimit.py @@ -0,0 +1,63 @@ +from app.ratelimit import MAX_FAILURES, WINDOW_SECONDS, FailureLimiter + + +class Clock: + def __init__(self): + self.now = 1000.0 + + def __call__(self): + return self.now + + +def make(): + clock = Clock() + return FailureLimiter(clock=clock), clock + + +def test_blocks_after_max_failures(): + limiter, _ = make() + for _ in range(MAX_FAILURES - 1): + limiter.record("alice") + assert not limiter.blocked("alice") + limiter.record("alice") + assert limiter.blocked("alice") + + +def test_keys_are_independent(): + limiter, _ = make() + for _ in range(MAX_FAILURES): + limiter.record("alice") + assert not limiter.blocked("bob") + + +def test_block_expires_with_clock(): + limiter, clock = make() + for _ in range(MAX_FAILURES): + limiter.record("alice") + clock.now += WINDOW_SECONDS - 1 + assert limiter.blocked("alice") + clock.now += 2 + assert not limiter.blocked("alice") + + +def test_old_failures_do_not_accumulate(): + limiter, clock = make() + for _ in range(MAX_FAILURES - 1): + limiter.record("alice") + clock.now += WINDOW_SECONDS + 1 + limiter.record("alice") + assert not limiter.blocked("alice") + + +def test_reset_clears_block(): + limiter, _ = make() + for _ in range(MAX_FAILURES): + limiter.record("alice") + limiter.reset("alice") + assert not limiter.blocked("alice") + + +def test_reset_unknown_key_is_noop(): + limiter, _ = make() + limiter.reset("nobody") + assert not limiter.blocked("nobody") diff --git a/tests/test_totp.py b/tests/test_totp.py new file mode 100644 index 0000000..3623f68 --- /dev/null +++ b/tests/test_totp.py @@ -0,0 +1,54 @@ +from app import totp + +RFC_SECRET = "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ" +RFC_TIME = 59 +RFC_CODE = "287082" + + +def test_rfc6238_vector(): + assert totp.current_code(RFC_SECRET, now=RFC_TIME) == RFC_CODE + + +def test_verify_accepts_current_code(): + assert totp.verify(RFC_SECRET, RFC_CODE, now=RFC_TIME) + + +def test_verify_accepts_spaces(): + assert totp.verify(RFC_SECRET, "287 082", now=RFC_TIME) + + +def test_window_previous_and_next_step(): + assert totp.verify(RFC_SECRET, RFC_CODE, now=RFC_TIME + totp.STEP_SECONDS) + assert totp.verify(RFC_SECRET, RFC_CODE, now=RFC_TIME - totp.STEP_SECONDS) + + +def test_outside_window_rejected(): + assert not totp.verify(RFC_SECRET, RFC_CODE, now=RFC_TIME + 2 * totp.STEP_SECONDS) + assert not totp.verify(RFC_SECRET, RFC_CODE, now=RFC_TIME - 2 * totp.STEP_SECONDS) + + +def test_wrong_code_rejected(): + assert not totp.verify(RFC_SECRET, "000000", now=RFC_TIME) + + +def test_short_code_rejected(): + assert not totp.verify(RFC_SECRET, RFC_CODE[:5], now=RFC_TIME) + + +def test_non_numeric_code_rejected(): + assert not totp.verify(RFC_SECRET, "28708a", now=RFC_TIME) + assert not totp.verify(RFC_SECRET, "", now=RFC_TIME) + + +def test_generated_secret_roundtrip(): + secret = totp.generate_secret() + assert totp.verify(secret, totp.current_code(secret)) + + +def test_provisioning_uri(): + uri = totp.provisioning_uri(RFC_SECRET, "max mustermann") + assert uri.startswith("otpauth://totp/checkbot:max%20mustermann?") + assert f"secret={RFC_SECRET}" in uri + assert "issuer=checkbot" in uri + assert f"digits={totp.DIGITS}" in uri + assert f"period={totp.STEP_SECONDS}" in uri diff --git a/tests/test_users.py b/tests/test_users.py new file mode 100644 index 0000000..b303011 --- /dev/null +++ b/tests/test_users.py @@ -0,0 +1,155 @@ +import os + +import pytest +from argon2 import PasswordHasher + +from app import audit +from app.crypto import SecretBox +from app.db import Database +from app.users import MIN_PASSWORD_LEN, ROLE_ADMIN, ROLE_VIEWER, UserError, UserStore + +PASSWORD = "correct horse battery" +ACTOR = "tester" + + +@pytest.fixture +def db(): + return Database(":memory:") + + +@pytest.fixture +def store(db): + hasher = PasswordHasher(time_cost=1, memory_cost=8, parallelism=1) + return UserStore(db, SecretBox(os.urandom(32)), hasher) + + +def test_create_and_authenticate(store): + created = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + assert created.username == "alice" + assert created.is_admin + assert not created.has_totp + assert store.authenticate("alice", PASSWORD).id == created.id + assert store.count() == 1 + + +def test_wrong_password(store): + store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + assert store.authenticate("alice", "wrong password!!") is None + + +def test_unknown_user(store): + assert store.authenticate("ghost", PASSWORD) is None + + +def test_password_too_short(store): + with pytest.raises(UserError): + store.create("alice", "x" * (MIN_PASSWORD_LEN - 1), ROLE_ADMIN, ACTOR) + + +@pytest.mark.parametrize("name", ["ab", "a b c", "ümlaut", "x" * 65, "", "a/b"]) +def test_invalid_username(store, name): + with pytest.raises(UserError): + store.create(name, PASSWORD, ROLE_VIEWER, ACTOR) + + +def test_invalid_role(store): + with pytest.raises(UserError): + store.create("alice", PASSWORD, "root", ACTOR) + + +def test_duplicate_case_insensitive(store): + store.create("Alice", PASSWORD, ROLE_ADMIN, ACTOR) + with pytest.raises(UserError): + store.create("alice", PASSWORD, ROLE_VIEWER, ACTOR) + + +def test_last_admin_guard_set_role(store): + admin = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + with pytest.raises(UserError): + store.set_role(admin.id, ROLE_VIEWER, ACTOR) + + +def test_last_admin_guard_set_disabled(store): + admin = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + with pytest.raises(UserError): + store.set_disabled(admin.id, True, ACTOR) + + +def test_last_admin_guard_delete(store): + admin = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + with pytest.raises(UserError): + store.delete(admin.id, ACTOR) + + +def test_second_admin_allows_changes(store): + first = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + store.create("bob", PASSWORD, ROLE_ADMIN, ACTOR) + store.set_role(first.id, ROLE_VIEWER, ACTOR) + assert store.get(first.id).role == ROLE_VIEWER + store.set_disabled(first.id, True, ACTOR) + assert store.get(first.id).disabled + store.delete(first.id, ACTOR) + assert store.get(first.id) is None + + +def test_viewer_can_be_removed_freely(store): + store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + viewer = store.create("bob", PASSWORD, ROLE_VIEWER, ACTOR) + store.set_disabled(viewer.id, True, ACTOR) + store.delete(viewer.id, ACTOR) + assert store.count() == 1 + + +def test_set_password_bumps_session_version(store): + user = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + store.set_password(user.id, "another long password", ACTOR) + updated = store.get(user.id) + assert updated.session_version == user.session_version + 1 + assert store.authenticate("alice", PASSWORD) is None + assert store.authenticate("alice", "another long password") is not None + + +def test_set_password_validates_length(store): + user = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + with pytest.raises(UserError): + store.set_password(user.id, "short", ACTOR) + + +def test_disabled_user_cannot_login(store): + store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + bob = store.create("bob", PASSWORD, ROLE_VIEWER, ACTOR) + store.set_disabled(bob.id, True, ACTOR) + assert store.authenticate("bob", PASSWORD) is None + store.set_disabled(bob.id, False, ACTOR) + assert store.authenticate("bob", PASSWORD) is not None + + +def test_unknown_user_id(store): + with pytest.raises(UserError): + store.set_role(999, ROLE_VIEWER, ACTOR) + + +def test_totp_secret_encrypted_and_roundtrip(store, db): + user = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + secret = "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ" + store.set_totp(user.id, secret, ACTOR) + raw = db.query("SELECT totp_secret FROM users WHERE id=?", (user.id,))[0]["totp_secret"] + assert raw and secret not in raw + assert store.totp_secret(user.id) == secret + assert store.get(user.id).has_totp + store.set_totp(user.id, None, ACTOR) + assert store.totp_secret(user.id) is None + assert not store.get(user.id).has_totp + + +def test_audit_entries_without_password(store, db): + user = store.create("alice", PASSWORD, ROLE_ADMIN, ACTOR) + store.set_password(user.id, "another long password", ACTOR) + store.set_totp(user.id, "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ", ACTOR) + entries = audit.recent(db) + actions = {e["action"] for e in entries} + assert {"user.create", "user.password", "user.totp.enable"} <= actions + dump = repr(entries) + assert PASSWORD not in dump + assert "another long password" not in dump + assert "GEZDGNBVGY3TQOJQ" not in dump diff --git a/tests/test_web.py b/tests/test_web.py new file mode 100644 index 0000000..a96fd66 --- /dev/null +++ b/tests/test_web.py @@ -0,0 +1,257 @@ +import re +import secrets + +import pytest +from fastapi.testclient import TestClient + +from app import audit, totp +from app.checkmk import CheckmkClient +from app.crypto import generate_master_key +from app.fileconfig import MASTER_KEY_FILE, SESSION_SECRET_FILE, FileConfig +from app.main import create_app +from app.users import ROLE_ADMIN, ROLE_VIEWER + +CSRF_RE = re.compile(r'name="csrf" value="([^"]+)"') +PASSWORD = "correct horse battery" +ADMIN = "admin1" +VIEWER = "viewer1" +CHECKMK_SECRET_VALUE = "geheimes-automation-secret" +LOGIN_FAIL_LIMIT = 5 + + +@pytest.fixture +def app(tmp_path): + (tmp_path / MASTER_KEY_FILE).write_text(generate_master_key()) + (tmp_path / SESSION_SECRET_FILE).write_text(secrets.token_urlsafe(48)) + config = FileConfig(config_dir=tmp_path, db_path=tmp_path / "t.db", bind_host="127.0.0.1", + bind_port=8080, secure_cookies=False) + return create_app(config) + + +@pytest.fixture +def client(app): + return TestClient(app, follow_redirects=False) + + +@pytest.fixture +def users(app): + return app.state.users + + +def csrf_from(client, path): + response = client.get(path) + match = CSRF_RE.search(response.text) + assert match, f"kein CSRF-Token auf {path} (Status {response.status_code})" + return match.group(1) + + +def login(client, username, password): + token = csrf_from(client, "/login") + return client.post("/login", data={"csrf": token, "username": username, "password": password}) + + +def login_ok(client, username=ADMIN, password=PASSWORD): + response = login(client, username, password) + assert response.status_code == 303 + return response + + +def post(client, path, form_page, **data): + token = csrf_from(client, form_page) + return client.post(path, data={"csrf": token, **data}) + + +def checkmk_form(**overrides): + data = { + "checkmk_url": "https://checkmk.example.de", + "checkmk_site": "prod", + "checkmk_user": "automation", + "checkmk_secret": CHECKMK_SECRET_VALUE, + "checkmk_ca_file": "", + "checkmk_verify_tls": "1", + } + data.update(overrides) + return data + + +@pytest.fixture +def admin(users): + return users.create(ADMIN, PASSWORD, ROLE_ADMIN, actor="test") + + +@pytest.fixture +def viewer(users): + return users.create(VIEWER, PASSWORD, ROLE_VIEWER, actor="test") + + +def test_unauthenticated_redirects_to_login(client): + response = client.get("/") + assert response.status_code == 303 + assert response.headers["location"] == "/login" + + +def test_login_ok(client, admin): + login_ok(client) + response = client.get("/") + assert response.status_code == 200 + assert "Dashboard" in response.text + + +def test_wrong_password(client, app, admin): + response = login(client, ADMIN, "falsches passwort!!") + assert response.status_code == 401 + assert "login.failed" in [e["action"] for e in audit.recent(app.state.db)] + + +def test_lockout_after_failures(client, admin): + for _ in range(LOGIN_FAIL_LIMIT): + assert login(client, ADMIN, "falsches passwort!!").status_code == 401 + assert login(client, ADMIN, PASSWORD).status_code == 429 + + +def test_post_without_csrf_forbidden(client, admin): + response = client.post("/login", data={"username": ADMIN, "password": PASSWORD}) + assert response.status_code == 403 + + +def test_viewer_forbidden_on_admin_pages(client, viewer): + login_ok(client, VIEWER) + for path in ("/users", "/audit", "/settings/checkmk"): + assert client.get(path).status_code == 403 + assert client.get("/").status_code == 200 + + +def test_admin_saves_settings_secret_encrypted(client, app, admin): + login_ok(client) + response = post(client, "/settings/checkmk", "/settings/checkmk", **checkmk_form()) + assert response.status_code == 303 + store = app.state.store + assert store.get("checkmk.secret") == CHECKMK_SECRET_VALUE + raw = app.state.db.query("SELECT value, is_secret FROM settings WHERE key='checkmk.secret'")[0] + assert raw["is_secret"] == 1 + assert CHECKMK_SECRET_VALUE not in raw["value"] + page = client.get("/settings/checkmk") + assert CHECKMK_SECRET_VALUE not in page.text + assert "gesetzt" in page.text + assert CHECKMK_SECRET_VALUE not in repr(audit.recent(app.state.db)) + + +def test_empty_secret_keeps_stored_value(client, app, admin): + login_ok(client) + post(client, "/settings/checkmk", "/settings/checkmk", **checkmk_form()) + post(client, "/settings/checkmk", "/settings/checkmk", **checkmk_form(checkmk_secret="", checkmk_site="test")) + assert app.state.store.get("checkmk.secret") == CHECKMK_SECRET_VALUE + assert app.state.store.get("checkmk.site") == "test" + + +def test_settings_validation_error(client, app, admin): + login_ok(client) + response = post(client, "/settings/checkmk", "/settings/checkmk", + **checkmk_form(checkmk_url="checkmk.example.de")) + assert response.status_code == 422 + assert "http://" in response.text + assert app.state.store.get("checkmk.url") is None + + +def test_settings_unknown_section(client, admin): + login_ok(client) + assert client.get("/settings/gibtsnicht").status_code == 404 + + +def test_settings_test_endpoint(client, admin, monkeypatch): + async def fake_version(self): + return {"versions": {"checkmk": "2.3.0"}} + + async def fake_problems(self): + return [] + + monkeypatch.setattr(CheckmkClient, "version", fake_version) + monkeypatch.setattr(CheckmkClient, "problem_services", fake_problems) + login_ok(client) + token = csrf_from(client, "/settings/checkmk") + response = client.post("/settings/checkmk/test", data=checkmk_form(), headers={"X-CSRF-Token": token}) + assert response.status_code == 200 + body = response.json() + assert body["ok"] is True + assert "2.3.0" in body["detail"] + + +def test_settings_test_endpoint_reports_validation(client, admin): + login_ok(client) + token = csrf_from(client, "/settings/checkmk") + response = client.post("/settings/checkmk/test", data=checkmk_form(checkmk_url="nope"), + headers={"X-CSRF-Token": token}) + assert response.json()["ok"] is False + + +def test_session_invalid_after_password_change(client, users, admin): + login_ok(client) + assert client.get("/").status_code == 200 + users.set_password(admin.id, "another long password", actor="test") + response = client.get("/") + assert response.status_code == 303 + assert response.headers["location"] == "/login" + + +def test_totp_login_flow(client, users, admin): + secret = totp.generate_secret() + users.set_totp(admin.id, secret, actor="test") + response = login(client, ADMIN, PASSWORD) + assert response.status_code == 303 + assert response.headers["location"] == "/login/totp" + assert client.get("/").status_code == 303 + token = csrf_from(client, "/login/totp") + bad = client.post("/login/totp", data={"csrf": token, "code": "000000"}) + assert bad.status_code == 401 + response = client.post("/login/totp", data={"csrf": token, "code": totp.current_code(secret)}) + assert response.status_code == 303 + assert response.headers["location"] == "/" + assert client.get("/").status_code == 200 + + +def test_totp_page_requires_pending_login(client): + response = client.get("/login/totp") + assert response.status_code == 303 + assert response.headers["location"] == "/login" + + +def test_create_change_role_delete_user_via_ui(client, users, admin): + login_ok(client) + response = post(client, "/users", "/users", username="newbie", password=PASSWORD, role=ROLE_VIEWER) + assert response.status_code == 303 + created = users.get_by_name("newbie") + assert created and created.role == ROLE_VIEWER + + post(client, f"/users/{created.id}/role", "/users", role=ROLE_ADMIN) + assert users.get(created.id).role == ROLE_ADMIN + + post(client, f"/users/{created.id}/delete", "/users") + assert users.get(created.id) is None + + +def test_create_user_validation_message(client, admin): + login_ok(client) + post(client, "/users", "/users", username="newbie", password="kurz", role=ROLE_VIEWER) + assert "Passwort: mindestens" in client.get("/users").text + + +def test_last_admin_guard_shows_flash(client, users, admin): + login_ok(client) + post(client, f"/users/{admin.id}/role", "/users", role=ROLE_VIEWER) + assert "letzte aktive Admin" in client.get("/users").text + + +def test_cannot_delete_or_disable_self(client, users, admin): + login_ok(client) + post(client, f"/users/{admin.id}/delete", "/users") + assert users.get(admin.id) is not None + post(client, f"/users/{admin.id}/disable", "/users", disabled="1") + assert not users.get(admin.id).disabled + + +def test_health_json(client): + response = client.get("/health") + assert response.status_code == 503 + body = response.json() + assert body["ok"] is False + assert "checks" in body