patrickandClaude Sonnet 5.5
453c077386
Lesender Selbsttest (Version, Services, Hosts, Downtimes) beim Start und täglich 04:00, Seite /selftest, Teilcheck checkmk_api in /health, Chat-Meldung bei Wechsel. Toleranter Antwort-Parser (value/members, state_type int/String), Fallbacks für Host-Abfrage und Downtime-Liste, Fehlerhinweise je HTTP-Status. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
268 lines
11 KiB
Python
268 lines
11 KiB
Python
import re
|
|
import time
|
|
from typing import 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 . import (alerting_section, downtime_section, matrix_section, retention_section, selftest_section,
|
|
telegram_section)
|
|
from .deps import (FLASH_ERROR, FLASH_OK, current_user, flash, redirect, render, require_admin)
|
|
from .problem_status import problems_with_status
|
|
from .sections import (URL_RE, Field, Section, check, collect, effective, find_section, form_values,
|
|
secret_flags, section_done, stored_values)
|
|
|
|
router = APIRouter()
|
|
|
|
SITE_RE = re.compile(r"^[A-Za-z0-9_-]+$")
|
|
AUDIT_LIMIT = 200
|
|
DAY_SECONDS = 86400
|
|
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."
|
|
|
|
|
|
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,
|
|
),
|
|
telegram_section.SLUG: telegram_section.SECTION,
|
|
matrix_section.SLUG: matrix_section.SECTION,
|
|
alerting_section.SLUG: alerting_section.SECTION,
|
|
downtime_section.SLUG: downtime_section.SECTION,
|
|
retention_section.SLUG: retention_section.SECTION,
|
|
selftest_section.SLUG: selftest_section.SECTION,
|
|
}
|
|
|
|
|
|
@router.get("/")
|
|
async def dashboard(request: Request, user: User = Depends(current_user)):
|
|
store = request.app.state.store
|
|
runtime = request.app.state.runtime
|
|
result = await run_checks(build_checks(store, runtime.status, runtime.selftest_repo))
|
|
setup_open = user.is_admin and not all(
|
|
section_done(store, s) for s in SECTIONS.values() if not s.optional)
|
|
return render(request, "dashboard.html", user=user, result=result, setup_open=setup_open,
|
|
alerts_24h=runtime.alerts.count_since(int(time.time()) - DAY_SECONDS),
|
|
open_problems=problems_with_status(request))
|
|
|
|
|
|
@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), "optional": s.optional}
|
|
for s in SECTIONS.values()]
|
|
return render(request, "setup.html", user=user, steps=steps)
|
|
|
|
|
|
def _settings_page(request: Request, user: User, section: Section, values: dict, status_code: int = 200):
|
|
extra = section.extra_context(request.app.state) if section.extra_context else {}
|
|
return render(request, "settings.html", user=user, section=section, values=form_values(section, values),
|
|
secret_set=secret_flags(request.app.state.store, section), status_code=status_code, **extra)
|
|
|
|
|
|
@router.get("/settings/{slug}")
|
|
def settings_form(request: Request, slug: str, user: User = Depends(require_admin)):
|
|
section = find_section(SECTIONS, slug)
|
|
return _settings_page(request, user, section, stored_values(request.app.state.store, section))
|
|
|
|
|
|
@router.post("/settings/{slug}")
|
|
async def settings_save(request: Request, slug: str, user: User = Depends(require_admin)):
|
|
section = find_section(SECTIONS, slug)
|
|
store = request.app.state.store
|
|
submitted = collect(await request.form(), section)
|
|
values = effective(store, submitted)
|
|
errors = check(section, values)
|
|
if errors:
|
|
for message in errors:
|
|
flash(request, message, FLASH_ERROR)
|
|
return _settings_page(request, user, section, values, STATUS_INVALID)
|
|
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 = find_section(SECTIONS, slug)
|
|
if section.test is None:
|
|
raise HTTPException(404, "Kein Verbindungstest für diesen Bereich")
|
|
store = request.app.state.store
|
|
values = effective(store, collect(await request.form(), section))
|
|
values = {**{key: store.get(key) or "" for key in section.context_keys}, **values}
|
|
errors = check(section, values)
|
|
if errors:
|
|
return JSONResponse({"ok": False, "detail": " ".join(errors)})
|
|
try:
|
|
detail = await section.test(values)
|
|
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))
|