""" Datenzugriffsschicht fuer die ITSM-Plattform (PostgreSQL, mandantenfaehig). Ersetzt die fruehere JSON-Datei-Ablage. Grund: Betrieb unter ISO 27001, DSGVO und NIS 2 verlangt belastbare Zugriffskontrolle, Nachvollziehbarkeit (Audit-Log) und geordnete Backup-/Wiederherstellungsfaehigkeit -- das leistet eine Datei- Ablage nicht. ENV: DATABASE_URL postgresql://user:pass@host:5432/dbname """ import os import json import datetime as _dt from contextlib import contextmanager import psycopg2 import psycopg2.extras import psycopg2.pool DATABASE_URL = os.getenv("DATABASE_URL", "") _pool = None def _get_pool(): global _pool if _pool is None: _pool = psycopg2.pool.SimpleConnectionPool(1, 10, dsn=DATABASE_URL) return _pool @contextmanager def get_conn(): pool = _get_pool() conn = pool.getconn() try: yield conn conn.commit() except Exception: conn.rollback() raise finally: pool.putconn(conn) def _dict_cursor(conn): return conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) def init_db(): """Legt das Schema an (idempotent). Wird beim App-Start aufgerufen.""" here = os.path.dirname(os.path.abspath(__file__)) schema_path = os.path.join(here, "schema.sql") with open(schema_path, "r", encoding="utf-8") as f: ddl = f.read() with get_conn() as conn: with conn.cursor() as cur: cur.execute(ddl) def any_tenant_exists(): with get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT 1 FROM tenants LIMIT 1") return cur.fetchone() is not None # ── Tenants / Ersteinrichtung ────────────────────────────────────────────────── def create_tenant(name, dsb_name, dsb_email, retention_tickets_days, retention_audit_days, sla_antwort_minuten, sla_loesung_minuten): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """INSERT INTO tenants (name, dsb_name, dsb_email, retention_tickets_days, retention_audit_days, sla_antwort_minuten, sla_loesung_minuten) VALUES (%s,%s,%s,%s,%s,%s,%s) RETURNING id""", (name, dsb_name, dsb_email, retention_tickets_days, retention_audit_days, sla_antwort_minuten, sla_loesung_minuten), ) return cur.fetchone()["id"] def get_tenant(tenant_id): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute("SELECT * FROM tenants WHERE id=%s", (tenant_id,)) return cur.fetchone() def update_tenant_settings(tenant_id, dsb_name, dsb_email, retention_tickets_days, retention_audit_days, sla_antwort_minuten, sla_loesung_minuten): with get_conn() as conn: with conn.cursor() as cur: cur.execute( """UPDATE tenants SET dsb_name=%s, dsb_email=%s, retention_tickets_days=%s, retention_audit_days=%s, sla_antwort_minuten=%s, sla_loesung_minuten=%s WHERE id=%s""", (dsb_name, dsb_email, retention_tickets_days, retention_audit_days, sla_antwort_minuten, sla_loesung_minuten, tenant_id), ) # ── Users ─────────────────────────────────────────────────────────────────────── def create_user(tenant_id, email, password_hash, role="admin", vorname=None, nachname=None, telefon=None, abteilung=None, adresse=None): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """INSERT INTO users (tenant_id, email, password_hash, role, vorname, nachname, telefon, abteilung, adresse) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s) RETURNING id""", (tenant_id, email, password_hash, role, vorname, nachname, telefon, abteilung, adresse), ) return cur.fetchone()["id"] def get_user_by_email(email): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute("SELECT * FROM users WHERE email=%s AND active=TRUE", (email,)) return cur.fetchone() def touch_last_login(user_id): with get_conn() as conn: with conn.cursor() as cur: cur.execute("UPDATE users SET last_login_at=now() WHERE id=%s", (user_id,)) def email_exists(email): with get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT 1 FROM users WHERE email=%s", (email,)) return cur.fetchone() is not None _USER_COLUMNS = ("id, email, role, active, auth_source, created_at, last_login_at, " "vorname, nachname, telefon, abteilung, adresse") def list_users(tenant_id): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( "SELECT %s FROM users WHERE tenant_id=%%s ORDER BY email" % _USER_COLUMNS, (tenant_id,)) return cur.fetchall() def get_user(tenant_id, user_id): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( "SELECT %s FROM users WHERE tenant_id=%%s AND id=%%s" % _USER_COLUMNS, (tenant_id, user_id)) return cur.fetchone() def update_user_profile(tenant_id, user_id, vorname, nachname, telefon, abteilung, adresse): with get_conn() as conn: with conn.cursor() as cur: cur.execute( """UPDATE users SET vorname=%s, nachname=%s, telefon=%s, abteilung=%s, adresse=%s WHERE tenant_id=%s AND id=%s""", (vorname, nachname, telefon, abteilung, adresse, tenant_id, user_id), ) def set_user_role(tenant_id, user_id, role): with get_conn() as conn: with conn.cursor() as cur: cur.execute("UPDATE users SET role=%s WHERE tenant_id=%s AND id=%s", (role, tenant_id, user_id)) def set_user_active(tenant_id, user_id, active): with get_conn() as conn: with conn.cursor() as cur: cur.execute("UPDATE users SET active=%s WHERE tenant_id=%s AND id=%s", (active, tenant_id, user_id)) # ── Audit-Log ─────────────────────────────────────────────────────────────────── def log_audit(tenant_id, user_id, aktion, entity_typ=None, entity_id=None, details=None, ip=None): with get_conn() as conn: with conn.cursor() as cur: cur.execute( """INSERT INTO audit_log (tenant_id, user_id, aktion, entity_typ, entity_id, details, ip) VALUES (%s,%s,%s,%s,%s,%s,%s)""", (tenant_id, user_id, aktion, entity_typ, entity_id, json.dumps(details) if details is not None else None, ip), ) def list_audit(tenant_id, limit=200): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """SELECT a.*, u.email AS user_email FROM audit_log a LEFT JOIN users u ON u.id = a.user_id WHERE a.tenant_id=%s ORDER BY a.zeit DESC LIMIT %s""", (tenant_id, limit), ) return cur.fetchall() # ── Services ──────────────────────────────────────────────────────────────────── def create_service(tenant_id, name, beschreibung, kategorie, gebucht, endpoint): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """INSERT INTO services (tenant_id, name, beschreibung, kategorie, gebucht, endpoint) VALUES (%s,%s,%s,%s,%s,%s) RETURNING id""", (tenant_id, name, beschreibung, kategorie, gebucht, endpoint), ) return cur.fetchone()["id"] def list_services(tenant_id): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute("SELECT * FROM services WHERE tenant_id=%s ORDER BY id", (tenant_id,)) return cur.fetchall() # ── Tickets ───────────────────────────────────────────────────────────────────── def _next_ticket_nr(tenant_id): year = _dt.datetime.utcnow().year prefix = "TKT-%d-" % year with get_conn() as conn: with conn.cursor() as cur: cur.execute( """SELECT ticket_nr FROM tickets WHERE tenant_id=%s AND ticket_nr LIKE %s ORDER BY ticket_nr DESC LIMIT 1""", (tenant_id, prefix + "%"), ) row = cur.fetchone() seq = 1 if row: try: seq = int(row[0].split("-")[-1]) + 1 except Exception: seq = 1 return "%s%06d" % (prefix, seq) def create_ticket(tenant_id, titel, beschreibung, service_id, prioritaet, kategorie, zugewiesen_an, ersteller_id, sla_antwort_minuten, sla_loesung_minuten, actor_label): ticket_nr = _next_ticket_nr(tenant_id) with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """INSERT INTO tickets (tenant_id, ticket_nr, titel, beschreibung, service_id, prioritaet, kategorie, zugewiesen_an, ersteller_id, sla_antwort_minuten, sla_loesung_minuten) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) RETURNING id""", (tenant_id, ticket_nr, titel, beschreibung, service_id, prioritaet, kategorie, zugewiesen_an, ersteller_id, sla_antwort_minuten, sla_loesung_minuten), ) tid = cur.fetchone()["id"] cur.execute( """INSERT INTO ticket_timeline (ticket_id, tenant_id, akteur, text) VALUES (%s,%s,%s,%s)""", (tid, tenant_id, actor_label, "Ticket angelegt."), ) return tid, ticket_nr def list_tickets(tenant_id): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """SELECT t.*, s.name AS service_name FROM tickets t LEFT JOIN services s ON s.id = t.service_id WHERE t.tenant_id=%s ORDER BY t.updated_at DESC""", (tenant_id,), ) return cur.fetchall() def get_ticket(tenant_id, ticket_id): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """SELECT t.*, s.name AS service_name FROM tickets t LEFT JOIN services s ON s.id = t.service_id WHERE t.tenant_id=%s AND t.id=%s""", (tenant_id, ticket_id), ) ticket = cur.fetchone() if not ticket: return None cur.execute( "SELECT * FROM ticket_timeline WHERE ticket_id=%s ORDER BY zeit DESC", (ticket_id,), ) ticket = dict(ticket) ticket["timeline"] = cur.fetchall() return ticket def update_ticket_status(tenant_id, ticket_id, status, actor_label): with get_conn() as conn: with conn.cursor() as cur: cur.execute( """UPDATE tickets SET status=%s, updated_at=now(), fortschritt = CASE WHEN %s IN ('Geloest','Geschlossen') THEN 100 ELSE fortschritt END WHERE tenant_id=%s AND id=%s""", (status, status, tenant_id, ticket_id), ) cur.execute( """INSERT INTO ticket_timeline (ticket_id, tenant_id, akteur, text) VALUES (%s,%s,%s,%s)""", (ticket_id, tenant_id, actor_label, "Status geaendert auf '%s'." % status), ) # ── Aufbewahrungsfrist-Bereinigung (DSGVO Speicherbegrenzung) ───────────────── _RETENTION_LOCK_KEY = 727271 # feste Postgres-Advisory-Lock-ID fuer diesen Job def run_retention_cleanup(): """Loescht abgeschlossene Tickets und Audit-Log-Eintraege, die aelter sind als die pro Mandant hinterlegte Aufbewahrungsfrist (retention_tickets_days / retention_audit_days). Offene/laufende Tickets werden NIE geloescht, nur solche im Status 'Geloest'/'Geschlossen'. Nutzt einen Postgres-Advisory-Lock: laeuft der Job bereits (z. B. durch einen anderen Worker-Prozess), wird dieser Aufruf uebersprungen statt parallel zu loeschen. """ with get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT pg_try_advisory_lock(%s)", (_RETENTION_LOCK_KEY,)) got_lock = cur.fetchone()[0] if not got_lock: return {"skipped": "lock_not_acquired"} try: with _dict_cursor(conn) as cur: cur.execute("SELECT id, name, retention_tickets_days, retention_audit_days FROM tenants") tenants = cur.fetchall() summary = [] for t in tenants: tid = t["id"] with conn.cursor() as cur: cur.execute( """DELETE FROM tickets WHERE tenant_id=%s AND status IN ('Geloest','Geschlossen') AND updated_at < now() - (%s || ' days')::interval""", (tid, t["retention_tickets_days"]), ) tickets_deleted = cur.rowcount cur.execute( """DELETE FROM audit_log WHERE tenant_id=%s AND aktion <> 'retention_cleanup' AND zeit < now() - (%s || ' days')::interval""", (tid, t["retention_audit_days"]), ) audit_deleted = cur.rowcount if tickets_deleted or audit_deleted: log_audit(tid, None, "retention_cleanup", "tenant", str(tid), {"tickets_deleted": tickets_deleted, "audit_log_deleted": audit_deleted}) summary.append({"tenant_id": tid, "tenant": t["name"], "tickets_deleted": tickets_deleted, "audit_log_deleted": audit_deleted}) return {"summary": summary} finally: with conn.cursor() as cur: cur.execute("SELECT pg_advisory_unlock(%s)", (_RETENTION_LOCK_KEY,)) # ── CMDB (Configuration Management Database) ───────────────────────────────── def create_ci(tenant_id, name, ci_typ, status, beschreibung, attribute): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """INSERT INTO configuration_items (tenant_id, name, ci_typ, status, beschreibung, attribute) VALUES (%s,%s,%s,%s,%s,%s) RETURNING id""", (tenant_id, name, ci_typ, status, beschreibung, json.dumps(attribute or {})), ) return cur.fetchone()["id"] def list_cis(tenant_id, ci_typ=None): with get_conn() as conn: with _dict_cursor(conn) as cur: if ci_typ: cur.execute( "SELECT * FROM configuration_items WHERE tenant_id=%s AND ci_typ=%s ORDER BY name", (tenant_id, ci_typ)) else: cur.execute( "SELECT * FROM configuration_items WHERE tenant_id=%s ORDER BY ci_typ, name", (tenant_id,)) return cur.fetchall() def get_ci(tenant_id, ci_id): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( "SELECT * FROM configuration_items WHERE tenant_id=%s AND id=%s", (tenant_id, ci_id)) ci = cur.fetchone() if not ci: return None ci = dict(ci) cur.execute( """SELECT r.*, c.name AS to_name FROM ci_relationships r JOIN configuration_items c ON c.id = r.to_ci_id WHERE r.tenant_id=%s AND r.from_ci_id=%s ORDER BY r.id""", (tenant_id, ci_id)) ci["rel_out"] = cur.fetchall() cur.execute( """SELECT r.*, c.name AS from_name FROM ci_relationships r JOIN configuration_items c ON c.id = r.from_ci_id WHERE r.tenant_id=%s AND r.to_ci_id=%s ORDER BY r.id""", (tenant_id, ci_id)) ci["rel_in"] = cur.fetchall() return ci def update_ci(tenant_id, ci_id, name, ci_typ, status, beschreibung, attribute): with get_conn() as conn: with conn.cursor() as cur: cur.execute( """UPDATE configuration_items SET name=%s, ci_typ=%s, status=%s, beschreibung=%s, attribute=%s, updated_at=now() WHERE tenant_id=%s AND id=%s""", (name, ci_typ, status, beschreibung, json.dumps(attribute or {}), tenant_id, ci_id), ) def delete_ci(tenant_id, ci_id): with get_conn() as conn: with conn.cursor() as cur: cur.execute("DELETE FROM configuration_items WHERE tenant_id=%s AND id=%s", (tenant_id, ci_id)) def create_ci_relationship(tenant_id, from_ci_id, to_ci_id, beziehungs_typ): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """INSERT INTO ci_relationships (tenant_id, from_ci_id, to_ci_id, beziehungs_typ) VALUES (%s,%s,%s,%s) RETURNING id""", (tenant_id, from_ci_id, to_ci_id, beziehungs_typ), ) return cur.fetchone()["id"] def delete_ci_relationship(tenant_id, rel_id): with get_conn() as conn: with conn.cursor() as cur: cur.execute("DELETE FROM ci_relationships WHERE tenant_id=%s AND id=%s", (tenant_id, rel_id)) # ── Wissensdatenbank ────────────────────────────────────────────────────────── def create_kb_article(tenant_id, titel, kategorie, inhalt, tags, autor_user_id): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """INSERT INTO kb_articles (tenant_id, titel, kategorie, inhalt, tags, autor_user_id) VALUES (%s,%s,%s,%s,%s,%s) RETURNING id""", (tenant_id, titel, kategorie, inhalt, tags, autor_user_id), ) return cur.fetchone()["id"] def list_kb_articles(tenant_id, kategorie=None, q=None): with get_conn() as conn: with _dict_cursor(conn) as cur: sql = """SELECT k.*, u.email AS autor_email FROM kb_articles k LEFT JOIN users u ON u.id = k.autor_user_id WHERE k.tenant_id=%s""" params = [tenant_id] if kategorie: sql += " AND k.kategorie=%s" params.append(kategorie) if q: sql += " AND (k.titel ILIKE %s OR k.inhalt ILIKE %s OR k.tags ILIKE %s)" like = "%%%s%%" % q params += [like, like, like] sql += " ORDER BY k.updated_at DESC" cur.execute(sql, params) return cur.fetchall() def get_kb_article(tenant_id, article_id): with get_conn() as conn: with _dict_cursor(conn) as cur: cur.execute( """SELECT k.*, u.email AS autor_email FROM kb_articles k LEFT JOIN users u ON u.id = k.autor_user_id WHERE k.tenant_id=%s AND k.id=%s""", (tenant_id, article_id)) return cur.fetchone() def update_kb_article(tenant_id, article_id, titel, kategorie, inhalt, tags): with get_conn() as conn: with conn.cursor() as cur: cur.execute( """UPDATE kb_articles SET titel=%s, kategorie=%s, inhalt=%s, tags=%s, updated_at=now() WHERE tenant_id=%s AND id=%s""", (titel, kategorie, inhalt, tags, tenant_id, article_id), ) def delete_kb_article(tenant_id, article_id): with get_conn() as conn: with conn.cursor() as cur: cur.execute("DELETE FROM kb_articles WHERE tenant_id=%s AND id=%s", (tenant_id, article_id))