From a07cbb77ab06fd5169c7c60f2fb541854c110082 Mon Sep 17 00:00:00 2001 From: mscadm Date: Mon, 13 Jul 2026 15:28:55 +0000 Subject: [PATCH] feat: Nutzerverwaltung + CMDB + Wissensdatenbank (Grundgeruest) --- app.py | 467 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 459 insertions(+), 8 deletions(-) diff --git a/app.py b/app.py index 2fde133..e1af4ef 100644 --- a/app.py +++ b/app.py @@ -197,10 +197,37 @@ _NAV = [ ("Assets", [("/assets", "Asset-Liste")]), ("Administration", [ ("/admin", "Einstellungen"), + ("/admin/users", "Benutzer"), ("/admin/audit", "Audit-Log"), ]), ] +CI_TYPES = ["Server", "Software", "Lizenz", "Vertrag", "Netzwerkgeraet", "Sonstiges"] +CI_STATUS = ["Aktiv", "Inaktiv", "Wartung", "Ausgemustert"] +REL_TYPES = ["haengt ab von", "beinhaltet", "verbunden mit", "ersetzt"] +KB_KATEGORIEN = ["Allgemein", "Anleitung", "Stoerung", "Konfiguration", "FAQ"] + + +def _attrs_to_text(attrs): + """dict -> 'Schluessel: Wert' je Zeile, fuer die Textarea im Formular.""" + if not attrs: + return "" + return "\n".join("%s: %s" % (k, v) for k, v in attrs.items()) + + +def _text_to_attrs(text): + """'Schluessel: Wert' je Zeile -> dict. Leere/fehlerhafte Zeilen werden ignoriert.""" + out = {} + for line in (text or "").splitlines(): + line = line.strip() + if not line or ":" not in line: + continue + k, v = line.split(":", 1) + k = k.strip() + if k: + out[k] = v.strip() + return out + def _shell(active_path, title, body_html, extra_head=""): u = _current_user() @@ -758,10 +785,159 @@ def admin_audit_route(): return _shell("/admin/audit", "Audit-Log", body) -# ── Platzhalterseiten ─────────────────────────────────────────────────────────── -def _placeholder(title, note): - return '

%s

%s

' % ( - html.escape(title), html.escape(note)) +# ── Administration: Benutzerverwaltung ───────────────────────────────────────── +def _role_pill(role): + return '%s' % (html.escape(role), html.escape(role)) + + +def _users_body(tenant_id, notice=""): + users = db.list_users(tenant_id) + rows = [] + for u in users: + status_pill = ('aktiv' + if u["active"] else + 'gesperrt') + toggle_label = "Sperren" if u["active"] else "Entsperren" + role_opts = "".join( + '' % (r, "selected" if r == u["role"] else "", r) + for r in ("admin", "agent") + ) + rows.append(""" +%s +%s +%s +%s +%s + +
+ +
+
+ +
+ +""" % ( + html.escape(u["email"]), _role_pill(u["role"]), status_pill, + html.escape(u["auth_source"]), + u["last_login_at"].strftime("%Y-%m-%d %H:%M") if u["last_login_at"] else "nie", + u["id"], role_opts, u["id"], toggle_label, + )) + table = """ + +%s
E-MailRolleStatusQuelleLetzte AnmeldungAktionen
""" % ("".join(rows) or "Keine Benutzer.") + + add_form = """

Neuen Benutzer anlegen

+
Manuelle Verwaltung. Eine AD/LDAP-Anbindung ist als naechste Ausbaustufe vorgesehen (Spalte "Quelle" zeigt dann "ldap" statt "local").
+
+
+
+
+
+
+
+
+
+
Mindestens 10 Zeichen. Der Benutzer sollte das Passwort nach der ersten Anmeldung selbst aendern.
+ +
""" + + return '

Benutzerverwaltung

' + notice + table + add_form + + +@app.route("/admin/users", methods=["GET", "POST"]) +def admin_users_route(): + r = _require_admin() + if r: + return r + tenant_id = session["tenant_id"] + notice = "" + if request.method == "POST": + email = request.form.get("email", "").strip().lower() + role = request.form.get("role", "agent").strip() + if role not in ("admin", "agent"): + role = "agent" + pw = request.form.get("password", "") + pw2 = request.form.get("password2", "") + if not email or "@" not in email: + notice = "

Bitte eine gueltige E-Mail-Adresse angeben.

" + elif len(pw) < 10: + notice = "

Das Passwort muss mindestens 10 Zeichen lang sein.

" + elif pw != pw2: + notice = "

Die Passwoerter stimmen nicht ueberein.

" + elif db.email_exists(email): + notice = "

Diese E-Mail-Adresse ist bereits registriert.

" + else: + uid = db.create_user(tenant_id, email, generate_password_hash(pw), role=role) + db.log_audit(tenant_id, session["user_id"], "user_created", "user", str(uid), + {"email": email, "role": role, "auth_source": "local"}, _client_ip()) + notice = "

Benutzer angelegt.

" + return _shell("/admin/users", "Benutzerverwaltung", _users_body(tenant_id, notice)) + + +@app.route("/admin/users//role", methods=["POST"]) +def admin_users_role_route(user_id): + r = _require_admin() + if r: + return r + tenant_id = session["tenant_id"] + role = request.form.get("role", "agent").strip() + if role not in ("admin", "agent"): + return redirect(url_for("admin_users_route")) + db.set_user_role(tenant_id, user_id, role) + db.log_audit(tenant_id, session["user_id"], "user_role_changed", "user", str(user_id), + {"role": role}, _client_ip()) + return redirect(url_for("admin_users_route")) + + +@app.route("/admin/users//active", methods=["POST"]) +def admin_users_active_route(user_id): + r = _require_admin() + if r: + return r + tenant_id = session["tenant_id"] + current = db.get_user(tenant_id, user_id) + if not current: + return redirect(url_for("admin_users_route")) + new_active = not current["active"] + if user_id == session["user_id"] and not new_active: + return redirect(url_for("admin_users_route")) # eigenes Konto nicht selbst sperren + db.set_user_active(tenant_id, user_id, new_active) + db.log_audit(tenant_id, session["user_id"], + "user_activated" if new_active else "user_deactivated", + "user", str(user_id), None, _client_ip()) + return redirect(url_for("admin_users_route")) + + +# ── Wissensdatenbank ──────────────────────────────────────────────────────────── +def _kb_list_body(tenant_id): + kategorie = request.args.get("kategorie") or None + q = request.args.get("q") or None + articles = db.list_kb_articles(tenant_id, kategorie=kategorie, q=q) + tabs = "".join( + '%s' % ( + html.escape(k), "active" if kategorie == k else "", html.escape(k)) + for k in KB_KATEGORIEN + ) + tabs = 'Alle' % ("active" if not kategorie else "") + tabs + rows = [] + for a in articles: + tag_pills = "".join('%s' % html.escape(t.strip()) + for t in (a["tags"] or "").split(",") if t.strip()) + rows.append(""" +%s%s%s +%s%s""" % ( + a["id"], html.escape(a["titel"]), html.escape(a["kategorie"]), tag_pills, + html.escape(a["autor_email"] or ""), a["updated_at"].strftime("%Y-%m-%d %H:%M"))) + table = """ + +%s
TitelKategorieTagsAutorAktualisiert
""" % ("".join(rows) or "Keine Artikel.") + search_form = """
+ + +
""" % html.escape(q or "") + return ('

Wissensdatenbank

' + '
%s
%s%s' + '
+ Neuer Artikel
' % (tabs, search_form, table)) @app.route("/wissen") @@ -769,8 +945,141 @@ def wissen_route(): r = _require_login() if r: return r - return _shell("/wissen", "Wissensdatenbank", - _placeholder("Wissensdatenbank", "Noch nicht aufgebaut -- folgt in einer naechsten Ausbaustufe.")) + return _shell("/wissen", "Wissensdatenbank", _kb_list_body(session["tenant_id"])) + + +@app.route("/wissen/new", methods=["GET", "POST"]) +def wissen_new_route(): + r = _require_login() + if r: + return r + tenant_id = session["tenant_id"] + error = None + titel, kategorie, inhalt, tags = "", "Allgemein", "", "" + if request.method == "POST": + titel = request.form.get("titel", "").strip() + kategorie = request.form.get("kategorie", "Allgemein").strip() + inhalt = request.form.get("inhalt", "").strip() + tags = request.form.get("tags", "").strip() + if not titel: + error = "Titel ist erforderlich." + elif not inhalt: + error = "Inhalt darf nicht leer sein." + else: + aid = db.create_kb_article(tenant_id, titel, kategorie, inhalt, tags, session["user_id"]) + db.log_audit(tenant_id, session["user_id"], "kb_article_created", "kb_article", str(aid), + {"titel": titel}, _client_ip()) + return redirect(url_for("wissen_detail_route", article_id=aid)) + kat_opts = "".join('' % (k, "selected" if k == kategorie else "", k) + for k in KB_KATEGORIEN) + err_html = "

%s

" % html.escape(error) if error else "" + body = """

Neuer Wissensartikel

+
%s
+
+
+
+
+ +
""" % (err_html, html.escape(titel), kat_opts, html.escape(inhalt), html.escape(tags)) + return _shell("/wissen", "Neuer Wissensartikel", body) + + +@app.route("/wissen/") +def wissen_detail_route(article_id): + r = _require_login() + if r: + return r + a = db.get_kb_article(session["tenant_id"], article_id) + if not a: + abort(404) + tag_pills = "".join('%s' % html.escape(t.strip()) + for t in (a["tags"] or "").split(",") if t.strip()) + body = """

%s

+
%s · %s · aktualisiert %s
+
%s
+
%s
+Bearbeiten +Zurueck zur Uebersicht""" % ( + html.escape(a["titel"]), html.escape(a["kategorie"]), html.escape(a["autor_email"] or ""), + a["updated_at"].strftime("%Y-%m-%d %H:%M"), html.escape(a["inhalt"]), tag_pills, a["id"]) + return _shell("/wissen", a["titel"], body) + + +@app.route("/wissen//edit", methods=["GET", "POST"]) +def wissen_edit_route(article_id): + r = _require_login() + if r: + return r + tenant_id = session["tenant_id"] + a = db.get_kb_article(tenant_id, article_id) + if not a: + abort(404) + error = None + if request.method == "POST": + titel = request.form.get("titel", "").strip() + kategorie = request.form.get("kategorie", "Allgemein").strip() + inhalt = request.form.get("inhalt", "").strip() + tags = request.form.get("tags", "").strip() + if not titel or not inhalt: + error = "Titel und Inhalt duerfen nicht leer sein." + else: + db.update_kb_article(tenant_id, article_id, titel, kategorie, inhalt, tags) + db.log_audit(tenant_id, session["user_id"], "kb_article_updated", "kb_article", str(article_id), + {"titel": titel}, _client_ip()) + return redirect(url_for("wissen_detail_route", article_id=article_id)) + a = dict(a) + a.update({"titel": titel, "kategorie": kategorie, "inhalt": inhalt, "tags": tags}) + kat_opts = "".join('' % (k, "selected" if k == a["kategorie"] else "", k) + for k in KB_KATEGORIEN) + err_html = "

%s

" % html.escape(error) if error else "" + body = """

Artikel bearbeiten

+
%s
+
+
+
+
+ + + +
+
""" % (err_html, html.escape(a["titel"]), kat_opts, html.escape(a["inhalt"]), html.escape(a["tags"] or ""), article_id) + return _shell("/wissen", "Artikel bearbeiten", body) + + +@app.route("/wissen//delete", methods=["POST"]) +def wissen_delete_route(article_id): + r = _require_login() + if r: + return r + tenant_id = session["tenant_id"] + db.delete_kb_article(tenant_id, article_id) + db.log_audit(tenant_id, session["user_id"], "kb_article_deleted", "kb_article", str(article_id), None, _client_ip()) + return redirect(url_for("wissen_route")) + + +# ── CMDB (Configuration Management Database) ─────────────────────────────────── +def _ci_status_pill(status): + cls = {"Aktiv": "geloest", "Inaktiv": "geschlossen", "Wartung": "bearbeitung", "Ausgemustert": "ueberfaellig"}.get(status, "offen") + return '%s' % (cls, html.escape(status)) + + +def _cmdb_list_body(tenant_id): + ci_typ = request.args.get("typ") or None + cis = db.list_cis(tenant_id, ci_typ=ci_typ) + tabs = 'Alle' % ("active" if not ci_typ else "") + tabs += "".join('%s' % (t, "active" if ci_typ == t else "", t) for t in CI_TYPES) + rows = [] + for c in cis: + rows.append(""" +%s%s%s%s""" % ( + c["id"], html.escape(c["name"]), html.escape(c["ci_typ"]), _ci_status_pill(c["status"]), + html.escape(c["beschreibung"] or ""))) + table = """ + +%s
NameTypStatusBeschreibung
""" % ("".join(rows) or "Keine Configuration Items.") + return ('

CMDB -- Configuration Items

' + '
%s
%s' + '' % (tabs, table)) @app.route("/assets") @@ -778,8 +1087,150 @@ def assets_route(): r = _require_login() if r: return r - return _shell("/assets", "Asset-Liste", - _placeholder("Asset-Liste", "Noch nicht aufgebaut -- folgt in einer naechsten Ausbaustufe.")) + return _shell("/assets", "CMDB", _cmdb_list_body(session["tenant_id"])) + + +@app.route("/assets/new", methods=["GET", "POST"]) +def assets_new_route(): + r = _require_login() + if r: + return r + tenant_id = session["tenant_id"] + error = None + name, ci_typ, status, beschreibung, attr_text = "", "Server", "Aktiv", "", "" + if request.method == "POST": + name = request.form.get("name", "").strip() + ci_typ = request.form.get("ci_typ", "Sonstiges").strip() + status = request.form.get("status", "Aktiv").strip() + beschreibung = request.form.get("beschreibung", "").strip() + attr_text = request.form.get("attribute", "") + if not name: + error = "Name ist erforderlich." + else: + cid = db.create_ci(tenant_id, name, ci_typ, status, beschreibung, _text_to_attrs(attr_text)) + db.log_audit(tenant_id, session["user_id"], "ci_created", "ci", str(cid), + {"name": name, "ci_typ": ci_typ}, _client_ip()) + return redirect(url_for("assets_detail_route", ci_id=cid)) + typ_opts = "".join('' % (t, "selected" if t == ci_typ else "", t) for t in CI_TYPES) + status_opts = "".join('' % (s, "selected" if s == status else "", s) for s in CI_STATUS) + err_html = "

%s

" % html.escape(error) if error else "" + body = """

Neues Configuration Item

+
%s
+
+
+
+
+
+
+
+
+
+
+
+ +
""" % (err_html, html.escape(name), typ_opts, status_opts, html.escape(beschreibung), html.escape(attr_text)) + return _shell("/assets", "Neues Configuration Item", body) + + +@app.route("/assets/", methods=["GET", "POST"]) +def assets_detail_route(ci_id): + r = _require_login() + if r: + return r + tenant_id = session["tenant_id"] + error = None + if request.method == "POST": + name = request.form.get("name", "").strip() + ci_typ = request.form.get("ci_typ", "Sonstiges").strip() + status = request.form.get("status", "Aktiv").strip() + beschreibung = request.form.get("beschreibung", "").strip() + attr_text = request.form.get("attribute", "") + if not name: + error = "Name ist erforderlich." + else: + db.update_ci(tenant_id, ci_id, name, ci_typ, status, beschreibung, _text_to_attrs(attr_text)) + db.log_audit(tenant_id, session["user_id"], "ci_updated", "ci", str(ci_id), {"name": name}, _client_ip()) + c = db.get_ci(tenant_id, ci_id) + if not c: + abort(404) + typ_opts = "".join('' % (t, "selected" if t == c["ci_typ"] else "", t) for t in CI_TYPES) + status_opts = "".join('' % (s, "selected" if s == c["status"] else "", s) for s in CI_STATUS) + err_html = "

%s

" % html.escape(error) if error else "" + + other_cis = [x for x in db.list_cis(tenant_id) if x["id"] != ci_id] + ci_opts = "".join('' % (x["id"], html.escape(x["name"]), x["ci_typ"]) for x in other_cis) + rel_typ_opts = "".join('' % (t, t) for t in REL_TYPES) + rel_out_rows = "".join( + "
%s %s
" + "
" % ( + html.escape(r["beziehungs_typ"]), html.escape(r["to_name"]), ci_id, r["id"]) + for r in c["rel_out"] + ) or "

Keine ausgehenden Beziehungen.

" + rel_in_rows = "".join( + "
%s %s (eingehend)
" % ( + html.escape(r["beziehungs_typ"]), html.escape(r["from_name"])) + for r in c["rel_in"] + ) or "

Keine eingehenden Beziehungen.

" + + body = """

%s

+
%s
+
+
+
+
+
+
+
+
+
+
+
+ +
+

Beziehungen

+
+

Ausgehend

%s +

Eingehend

%s +

Neue Beziehung

+
+
+
+
+
+ +
+
""" % ( + html.escape(c["name"]), err_html, html.escape(c["name"]), typ_opts, status_opts, + html.escape(c["beschreibung"] or ""), _attrs_to_text(c["attribute"]), + rel_out_rows, rel_in_rows, ci_id, rel_typ_opts, ci_opts, + ) + return _shell("/assets", c["name"], body) + + +@app.route("/assets//relationships", methods=["POST"]) +def assets_relationship_add_route(ci_id): + r = _require_login() + if r: + return r + tenant_id = session["tenant_id"] + to_ci_id = request.form.get("to_ci_id", type=int) + beziehungs_typ = request.form.get("beziehungs_typ", "haengt ab von").strip() + if to_ci_id and to_ci_id != ci_id: + rid = db.create_ci_relationship(tenant_id, ci_id, to_ci_id, beziehungs_typ) + db.log_audit(tenant_id, session["user_id"], "ci_relationship_created", "ci_relationship", str(rid), + {"from": ci_id, "to": to_ci_id, "typ": beziehungs_typ}, _client_ip()) + return redirect(url_for("assets_detail_route", ci_id=ci_id)) + + +@app.route("/assets//relationships//delete", methods=["POST"]) +def assets_relationship_delete_route(ci_id, rel_id): + r = _require_login() + if r: + return r + tenant_id = session["tenant_id"] + db.delete_ci_relationship(tenant_id, rel_id) + db.log_audit(tenant_id, session["user_id"], "ci_relationship_deleted", "ci_relationship", str(rel_id), None, _client_ip()) + return redirect(url_for("assets_detail_route", ci_id=ci_id)) @app.route("/")