"""
ITSM-Plattform -- eigenstaendiges Produkt (nicht Teil von AES), mandantenfaehig.
Betrieb gemaess ISO 27001 / DSGVO / NIS 2 (Nutzer-Vorgabe, 2026-07-13):
- Persistenz in PostgreSQL statt Dateien (Zugriffskontrolle, Backup/Recovery,
Verschluesselung im Ruhezustand ueber die DB moeglich).
- Audit-Log fuer alle sicherheitsrelevanten/aendernden Aktionen (ISO 27001
A.12.4, DSGVO Art. 30 Rechenschaftspflicht).
- Ersteinrichtungs-Assistent (/setup/new): legt pro Kunde einen eigenen
Mandanten (Tenant) mit eigenem Admin-Konto, SLA-Standardwerten und
DSGVO-Angaben (Datenschutzbeauftragter, Aufbewahrungsfristen) an.
Mandantenfaehig: mehrere Organisationen auf einer Instanz, Daten strikt
per tenant_id getrennt (kein Datenzugriff ueber Mandantengrenzen).
AES bleibt ein buchbarer Service im Katalog, integriert per HTTP (kein Code-
Import) -- siehe _aes_live_status().
ENV:
DATABASE_URL postgresql://user:pass@host:5432/dbname
ITSM_SECRET_KEY Flask-Session-Secret
AES_DASHBOARD_URL Basis-URL des AES-Dashboards fuer die Service-Integration
"""
import os
import html
import json
import secrets
import datetime as _dt
import urllib.request
import urllib.error
from flask import Flask, request, redirect, url_for, session, jsonify, abort
from werkzeug.security import generate_password_hash, check_password_hash
import db
AES_DASHBOARD_URL = os.getenv("AES_DASHBOARD_URL", "").rstrip("/")
app = Flask(__name__)
app.secret_key = os.getenv("ITSM_SECRET_KEY") or secrets.token_hex(32)
CATEGORIES = ["Incident", "Problem", "Task", "Change", "Release"]
PRIORITIES = ["Niedrig", "Mittel", "Hoch", "Kritisch"]
def _client_ip():
return request.headers.get("X-Forwarded-For", request.remote_addr or "")
def _now():
return _dt.datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
def _jsonable(v):
if isinstance(v, (_dt.datetime, _dt.date)):
return v.isoformat()
if isinstance(v, dict):
return {k: _jsonable(x) for k, x in v.items()}
if isinstance(v, list):
return [_jsonable(x) for x in v]
return v
# ── Auth / Session-Helfer ──────────────────────────────────────────────────────
def _current_user():
if not session.get("user_id"):
return None
return {
"id": session["user_id"],
"email": session.get("email"),
"tenant_id": session.get("tenant_id"),
"tenant_name": session.get("tenant_name"),
"role": session.get("role"),
}
def _require_login():
if not session.get("user_id"):
return redirect(url_for("login", next=request.path))
return None
def _require_admin():
r = _require_login()
if r:
return r
if session.get("role") != "admin":
return ("
' % (html.escape(label), value)
def _tickets_body(tenant_id, filter_status=None, filter_kategorie=None):
tickets = db.list_tickets(tenant_id)
kpis = _kpis(tickets)
kpi_html = "".join(_kpi_tile(k, v) for k, v in kpis.items())
shown = tickets
if filter_kategorie:
shown = [t for t in shown if t["kategorie"] == filter_kategorie]
tabs = [("", "Alle"), ("Offen", "Offen"), ("In Bearbeitung", "In Bearbeitung"),
("Warten", "Warten"), ("Geloest", "Geloest"), ("Ueberfaellig", "Ueberfaellig")]
base = "/probleme" if filter_kategorie == "Problem" else (
"/aenderungen" if filter_kategorie == "Change" else (
"/releases" if filter_kategorie == "Release" else "/tickets"))
tabs_html = "".join(
'%s' % (
base, ("?status=" + s) if s else "", "active" if (s or "") == (filter_status or "") else "", label)
for s, label in tabs)
if filter_status == "Ueberfaellig":
shown = [t for t in shown if _is_overdue(t)]
elif filter_status:
shown = [t for t in shown if t["status"] == filter_status]
rows = "".join(_ticket_row(t) for t in shown) or '
Keine Tickets.
'
services = db.list_services(tenant_id)
svc_opts = "".join('' % (s["id"], html.escape(s["name"])) for s in services)
new_form = """
Neues Ticket
""" % (
"".join('' % (c, c) for c in CATEGORIES),
"".join('' % (p, p) for p in PRIORITIES), svc_opts)
return """
Tickets
%s
%s
%s
Ticket-Nr
Titel (Service)
Status
Prioritaet
Kategorie
Zugewiesen an
Aktualisiert
Fortschritt
%s
""" % (kpi_html, new_form, tabs_html, rows)
@app.route("/tickets")
def tickets_route():
r = _require_login()
if r:
return r
return _shell("/tickets", "Tickets", _tickets_body(session["tenant_id"], request.args.get("status")))
@app.route("/probleme")
def probleme_route():
r = _require_login()
if r:
return r
return _shell("/probleme", "Probleme", _tickets_body(session["tenant_id"], request.args.get("status"), "Problem"))
@app.route("/aenderungen")
def aenderungen_route():
r = _require_login()
if r:
return r
return _shell("/aenderungen", "Aenderungen", _tickets_body(session["tenant_id"], request.args.get("status"), "Change"))
@app.route("/releases")
def releases_route():
r = _require_login()
if r:
return r
return _shell("/releases", "Releases", _tickets_body(session["tenant_id"], request.args.get("status"), "Release"))
@app.route("/tickets/new", methods=["POST"])
def tickets_new_route():
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
service_id = request.form.get("service_id") or None
tid, ticket_nr = db.create_ticket(
tenant_id,
request.form.get("titel", "").strip() or "(ohne Titel)",
request.form.get("beschreibung", "").strip(),
int(service_id) if service_id else None,
request.form.get("prioritaet", "Mittel"),
request.form.get("kategorie", "Task"),
session.get("email"),
session["user_id"],
480, 2880,
session.get("email"),
)
db.log_audit(tenant_id, session["user_id"], "ticket_created", "ticket", ticket_nr, None, _client_ip())
return redirect(url_for("tickets_route"))
@app.route("/tickets//status", methods=["POST"])
def ticket_status_route(ticket_id):
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
status = request.form.get("status", "Offen")
db.update_ticket_status(tenant_id, ticket_id, status, session.get("email"))
db.log_audit(tenant_id, session["user_id"], "ticket_status_changed", "ticket", str(ticket_id),
{"status": status}, _client_ip())
return jsonify({"ok": True})
@app.route("/api/tickets/")
def api_ticket(ticket_id):
r = _require_login()
if r:
return r
ticket = db.get_ticket(session["tenant_id"], ticket_id)
if not ticket:
abort(404)
return jsonify(_jsonable(ticket))
# ── Service-Katalog ─────────────────────────────────────────────────────────────
def _aes_live_status():
if not AES_DASHBOARD_URL:
return None
try:
req = urllib.request.Request(AES_DASHBOARD_URL + "/", method="GET")
with urllib.request.urlopen(req, timeout=4) as resp:
return 200 <= resp.status < 400
except Exception:
return None
def _services_body(tenant_id):
services = db.list_services(tenant_id)
cards = []
for s in services:
badge = 'gebucht' if s["gebucht"] else 'nicht gebucht'
link = ('
' + "".join(cards) + add_form
@app.route("/services")
def services_route():
r = _require_login()
if r:
return r
return _shell("/services", "Service-Katalog", _services_body(session["tenant_id"]))
@app.route("/services/new", methods=["POST"])
def services_new_route():
r = _require_login()
if r:
return r
tenant_id = session["tenant_id"]
sid = db.create_service(
tenant_id, request.form.get("name", "").strip(), request.form.get("beschreibung", "").strip(),
"Sonstiges", True, request.form.get("endpoint", "").strip(),
)
db.log_audit(tenant_id, session["user_id"], "service_created", "service", str(sid), None, _client_ip())
return redirect(url_for("services_route"))
# ── Administration: Einstellungen + Audit-Log ──────────────────────────────────
@app.route("/admin", methods=["GET", "POST"])
def admin_route():
r = _require_admin()
if r:
return r
tenant_id = session["tenant_id"]
notice = ""
if request.method == "POST":
db.update_tenant_settings(
tenant_id,
request.form.get("dsb_name", "").strip() or None,
request.form.get("dsb_email", "").strip() or None,
int(request.form.get("retention_tickets_days") or 1095),
int(request.form.get("retention_audit_days") or 1825),
int(request.form.get("sla_antwort_minuten") or 480),
int(request.form.get("sla_loesung_minuten") or 2880),
)
db.log_audit(tenant_id, session["user_id"], "tenant_settings_updated", "tenant", str(tenant_id), None, _client_ip())
notice = "
Gespeichert.
"
t = db.get_tenant(tenant_id)
body = """
Einstellungen
%s
%s
""" % (
html.escape(t["name"]), notice, t["sla_antwort_minuten"], t["sla_loesung_minuten"],
html.escape(t["dsb_name"] or ""), html.escape(t["dsb_email"] or ""),
t["retention_tickets_days"], t["retention_audit_days"])
return _shell("/admin", "Einstellungen", body)
@app.route("/admin/audit")
def admin_audit_route():
r = _require_admin()
if r:
return r
entries = db.list_audit(session["tenant_id"], limit=300)
rows = "".join(
"
%s
%s
%s
%s %s
" % (
e["zeit"].strftime("%Y-%m-%d %H:%M:%S"), html.escape(e["user_email"] or "(unbekannt)"),
html.escape(e["aktion"]), html.escape(e["entity_typ"] or ""), html.escape(e["entity_id"] or ""))
for e in entries
) or "
Keine Eintraege.
"
body = """
Audit-Log
Alle sicherheitsrelevanten Aktionen dieses Mandanten (Anmeldungen, Aenderungen). Aufbewahrung gemaess Einstellungen.
' % (
html.escape(title), html.escape(note))
@app.route("/wissen")
def wissen_route():
r = _require_login()
if r:
return r
return _shell("/wissen", "Wissensdatenbank",
_placeholder("Wissensdatenbank", "Noch nicht aufgebaut -- folgt in einer naechsten Ausbaustufe."))
@app.route("/assets")
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."))
@app.route("/")
def root_route():
if not db.any_tenant_exists():
return redirect(url_for("setup_new"))
if session.get("user_id"):
return redirect(url_for("tickets_route"))
return redirect(url_for("login"))
@app.route("/health")
def health_route():
try:
with db.get_conn() as conn:
with conn.cursor() as cur:
cur.execute("SELECT 1")
return jsonify({"status": "ok", "db": "ok"})
except Exception as e:
return jsonify({"status": "error", "db": str(e)}), 500
db.init_db()
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8090)