alientim/tesm
0
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 41 Wiki Activity

Im-/Export: fehlende Felder ergaenzt, NGINX neu, known_hosts (v1.1.11)

Browse Source 
Vollstaendiger Audit aller Export-/Import-Kategorien gegen das tatsaechliche
DB-Schema (7 parallele Pruef-Agents je Kategorie) ergab mehrere Luecken --
alle behoben:

- Auditlog-Export (v1.1.10-Nacharbeit): kein Deckel mehr bei 500 Zeilen,
  exportiert wird das komplette aktuell in der DB stehende Auditlog
  (bereits archivierte Tage liegen ohnehin als eigene Dateien vor).
- Clients: ssh_port und Zugangsdaten-Zuordnung (als portabler
  credential_name statt roher ID) fehlten komplett im Export.
- Switche: ssh_port fehlte komplett im Export.
- Zugangsdaten: category (z.B. "Linux-Client" fuer Wartung) fehlte.
- Custom-Gruppen: is_default fehlte (war beim Import zusaetzlich hart auf 0
  gesetzt statt aus der Importdatei uebernommen).
- LDAP-Gruppenzuordnungen: app_group_id wurde als ROHE lokale Datenbank-ID
  exportiert -- auf einem frisch installierten Zielsystem (der eigentliche
  Zweck dieser Funktion, siehe Seiten-Untertitel "Umzug auf eine neue
  Umgebung") haben Gruppen dort andere IDs, die Zuordnung waere kaputt oder
  auf die falsche Gruppe gezeigt. Jetzt wird der Gruppenname exportiert
  (bzw. "Admin" als Sonderfall) und beim Import auf dem Zielsystem wieder zur
  dortigen ID aufgeloest -- unbekannte Zielgruppen werden übersprungen statt
  eine kaputte Referenz anzulegen.
- ldap_fileshare_mappings (AD-Gruppe -> Freigabe) fehlte im Export/Import
  komplett -- alle Fileshare-Zuordnungen gingen bei jedem Umzug verloren.
- ldap_required_login_group (Login-Beschraenkung auf eine AD-Gruppe) fehlte
  in LDAP_SETTING_KEYS und wurde nie exportiert.
- NGINX-Einstellungen (Domain, Ports, SSL/HSTS-Schalter) hatten bisher gar
  keine Export-/Import-Kategorie. Bewusst AUSGESCHLOSSEN bleiben ssl_source,
  das TLS-Zertifikat und der private Schluessel sowie jeglicher
  Let's-Encrypt-/certbot-Zustand -- das sind host-gebundene reale Dateien
  bzw. eine ACME-Registrierung fuer genau diesen Host, kein portabler
  Einstellungswert.
- known_hosts: sobald Zugangsdaten exportiert werden, wird zusaetzlich der
  Inhalt von known_hosts mitexportiert (jeder per SSH bereits bestaetigte
  Host-Key gehoert inhaltlich zu den Zugangsdaten) und beim Import ergaenzend
  (nicht ueberschreibend) mit Zeilen-Deduplizierung in die eigene
  known_hosts-Datei eingespielt.

avatar_filename (Benutzer) bleibt bewusst vom Export ausgeschlossen -- reiner
Dateiname ohne die zugehoerige Bilddatei waere auf dem Zielsystem nur ein
kaputter Verweis, analog zur bewussten Nichtaufnahme von TLS-Zertifikat/
Schluessel bei NGINX.

Verifiziert per isoliertem Roundtrip-Test gegen eine Kopie der echten
Datenbank (nie die Produktivdaten selbst): Export -> Loeschen -> Re-Import
fuer jede geaenderte Kategorie plus known_hosts-Merge (inkl.
Deduplizierungs-Probe durch zweifachen Import derselben Datei), sowie ein
echter Browser-Exportlauf gegen POETEST zur Bestaetigung der UI/Label-Texte.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
alientim
2026-08-22 22:27:31 +02:00
co-authored by Claude Sonnet 5
parent 0f0fa99e53
commit 2b05d012ee
2 changed files with 172 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
srv/tesm/VERSION
+1 -1
View File
@@ -1 +1 @@
1.1.10
1.1.11
srv/tesm/app.py
+171 -32
View File
@@ -5392,47 +5392,99 @@ def save_nav_order():
def _read_known_hosts_content():
"""Für den Export: der Inhalt von known_hosts gehört inhaltlich zu den
Zugangsdaten (jeder per SSH bereits bestätigte Host-Key eines Geräts/
Switches, das über gespeicherte Zugangsdaten erreicht wurde) -- landet
daher NICHT als eigene EXPORT_BUILDERS-Kategorie, sondern als
zusätzlicher Payload-Schlüssel "known_hosts", sobald "credentials"
mitexportiert wird (siehe export_data())."""
try:
with open(SSH_KNOWN_HOSTS_PATH, "r", encoding="utf-8") as f:
content = f.read()
return content if content.strip() else None
except OSError:
return None
def _merge_known_hosts_content(new_content):
"""Für den Import: ergänzt die eigene known_hosts-Datei um die im
Export enthaltenen Zeilen -- bewusst ergänzend statt überschreibend
(die Zielumgebung kann bereits eigene, unabhängig bestätigte Host-Keys
haben) und mit Deduplizierung nach vollständiger Zeile, damit ein
wiederholter Import derselben Exportdatei nicht bei jedem Mal weitere
Duplikate anhäuft. Gibt die Anzahl tatsächlich neu hinzugefügter
Zeilen zurück."""
try:
try:
with open(SSH_KNOWN_HOSTS_PATH, "r", encoding="utf-8") as f:
existing_lines = set(f.read().splitlines())
except OSError:
existing_lines = set()
new_lines = [ln for ln in new_content.splitlines() if ln.strip() and ln not in existing_lines]
if not new_lines:
return 0
os.makedirs(os.path.dirname(SSH_KNOWN_HOSTS_PATH) or ".", exist_ok=True)
with open(SSH_KNOWN_HOSTS_PATH, "a", encoding="utf-8") as f:
f.write("\n".join(new_lines) + "\n")
return len(new_lines)
except OSError:
return 0
EXPORT_SECTIONS = [
("devices", "Clients"),
("switches", "Switche"),
("credentials", "Zugangsdaten"),
("credentials", "Zugangsdaten (inkl. bekannter SSH-Host-Keys aus known_hosts)"),
("users", "Benutzer (lokal)"),
("groups", "Custom-Gruppen"),
("ldap", "LDAP/AD-Einstellungen"),
("dhcp", "DHCP-Einstellungen"),
("logs", f"Logs (letzte {500} Einträge)"),
("nginx", "NGINX-Einstellungen (Domain, Ports, SSL/HSTS — ohne Zertifikat/Schlüssel)"),
("logs", "Auditlog (alle aktuell in der Datenbank vorhandenen Einträge)"),
]
EXPORT_SECTION_LABELS = dict(EXPORT_SECTIONS)
IMPORT_EXPORT_ADMIN_ONLY_SECTIONS = {"users", "groups", "ldap"}
LOG_EXPORT_LIMIT = 500
LDAP_SETTING_KEYS = [
"ldap_enabled", "ldap_server", "ldap_port", "ldap_use_ssl", "ldap_tls_skip_verify",
"ldap_base_dn", "ldap_user_filter_attr", "ldap_default_group",
"ldap_base_dn", "ldap_user_filter_attr", "ldap_default_group", "ldap_required_login_group",
]
def _export_devices(conn):
return [dict(r) for r in conn.execute(
"SELECT mac, ip, port, name, switch_hostname, is_active FROM devices WHERE deleted_at IS NULL"
).fetchall()]
return [dict(r) for r in conn.execute("""
SELECT devices.mac, devices.ip, devices.port, devices.name, devices.switch_hostname,
devices.is_active, devices.ssh_port, credentials.name AS credential_name
FROM devices LEFT JOIN credentials ON credentials.id = devices.credential_id
WHERE devices.deleted_at IS NULL
""").fetchall()]
def _export_switches(conn):
return [dict(r) for r in conn.execute("""
SELECT switches.hostname, switches.ip, credentials.name AS credential_name
SELECT switches.hostname, switches.ip, switches.ssh_port, credentials.name AS credential_name
FROM switches LEFT JOIN credentials ON credentials.id = switches.credential_id
WHERE switches.deleted_at IS NULL
""").fetchall()]
def _export_credentials(conn):
rows = conn.execute("SELECT name, username, password FROM credentials WHERE deleted_at IS NULL").fetchall()
return [{"name": r["name"], "username": r["username"], "password": decrypt_password(r["password"])} for r in rows]
rows = conn.execute("SELECT name, username, password, category FROM credentials WHERE deleted_at IS NULL").fetchall()
return [
{"name": r["name"], "username": r["username"], "password": decrypt_password(r["password"]), "category": r["category"]}
for r in rows
]
def _export_users(conn):
"""avatar_filename wird bewusst NICHT exportiert -- das ist nur ein
Dateiname, die eigentliche Bilddatei liegt unter static/uploads/avatars
auf DIESEM Host und würde auf dem Zielsystem nicht mitkommen; ein
exportierter Verweis auf eine dort nicht existierende Datei wäre
schlimmer als gar keiner (kaputtes <img>, siehe auch die analoge
Begründung bei _export_nginx() für TLS-Zertifikat/Schlüssel)."""
rows = conn.execute(
"SELECT id, username, password, first_name, last_name, email, is_admin, is_locked "
"FROM users WHERE auth_source='local' AND deleted_at IS NULL"
@@ -5453,7 +5505,7 @@ def _export_users(conn):
def _export_groups(conn):
rows = conn.execute("SELECT id, name FROM groups WHERE is_system=0 AND deleted_at IS NULL").fetchall()
rows = conn.execute("SELECT id, name, is_default FROM groups WHERE is_system=0 AND deleted_at IS NULL").fetchall()
result = []
for g in rows:
perms = [p["permission"] for p in conn.execute(
@@ -5464,23 +5516,34 @@ def _export_groups(conn):
"WHERE ug.group_id=? AND u.deleted_at IS NULL",
(g["id"],),
).fetchall()]
result.append({"name": g["name"], "permissions": perms, "members": members})
result.append({"name": g["name"], "is_default": g["is_default"], "permissions": perms, "members": members})
return result
def _export_ldap(conn):
"""app_group_id ist eine rohe, lokale groups.id (oder der Sonderwert
"admin") -- über eine Umgebungsgrenze hinweg garantiert NICHT stabil
(frische Installation vergibt Gruppen-IDs neu). Exportiert wird daher
der Gruppenname (analog _export_switches()' credential_name), _import_ldap
löst ihn auf dem Zielsystem wieder zur dortigen ID auf -- gleiches
"admin"/g.name-Mapping wie in _ldap_group_mappings() weiter oben."""
settings_data = {k: get_setting(k) for k in LDAP_SETTING_KEYS if get_setting(k) not in (None, "")}
data = {"settings": settings_data, "group_mappings": []}
data = {"settings": settings_data, "group_mappings": [], "fileshare_mappings": []}
bind = conn.execute(
"SELECT username, password FROM service_accounts WHERE purpose=?", (LDAP_BIND_SERVICE_ACCOUNT_PURPOSE,)
).fetchone()
if bind:
data["bind_username"] = bind["username"]
data["bind_password"] = decrypt_password(bind["password"])
data["group_mappings"] = [dict(m) for m in conn.execute(
"SELECT ad_group_dn, ad_group_name, app_group_id FROM ldap_group_mappings"
data["group_mappings"] = [dict(m) for m in conn.execute("""
SELECT m.ad_group_dn, m.ad_group_name,
CASE WHEN m.app_group_id = 'admin' THEN 'Admin' ELSE g.name END AS app_group_name
FROM ldap_group_mappings m LEFT JOIN groups g ON g.id = m.app_group_id AND g.deleted_at IS NULL
""").fetchall()]
data["fileshare_mappings"] = [dict(m) for m in conn.execute(
"SELECT ad_group_dn, ad_group_name, share_label, share_unc FROM ldap_fileshare_mappings"
).fetchall()]
if not (settings_data or "bind_username" in data or data["group_mappings"]):
if not (settings_data or "bind_username" in data or data["group_mappings"] or data["fileshare_mappings"]):
return None
return data
@@ -5513,9 +5576,31 @@ def _export_dhcp(conn):
return {"settings": settings_data, "subnets": subnets, "reservations": reservations, "options": options}
NGINX_SETTING_KEYS = ["nginx_server_name", "nginx_http_port", "nginx_https_port", "ssl_enabled", "hsts_enabled"]
def _export_nginx(conn):
"""Exportiert nur portable Konfiguration (Domain, Ports, SSL/HSTS-
Schalter) -- bewusst OHNE ssl_source, TLS-Zertifikat/Schlüssel oder
Let's-Encrypt-/certbot-Zustand: das sind entweder host-gebundene reale
Dateien (TESM_SSL_CERT_PATH/TESM_SSL_KEY_PATH) oder eine ACME-
Registrierung für genau diesen Host/diese Domain, kein portabler
Einstellungswert (gleiche Begründung wie bei avatar_filename in
_export_users())."""
settings_data = {k: get_setting(k) for k in NGINX_SETTING_KEYS if get_setting(k) not in (None, "")}
if not settings_data:
return None
return {"settings": settings_data}
def _export_logs(conn):
"""Exportiert das komplette, aktuell in der DB stehende Auditlog --
bereits archivierte Tage (siehe _archive_old_audit_log_rows) liegen
als eigene Dateien unter AUDIT_ARCHIVE_DIR und sind bewusst NICHT Teil
dieses Exports, das Auditlog-Export-Formular exportiert nur die
Datenbank-Kategorien."""
rows = conn.execute(
"SELECT ts, username, action, target, details FROM audit_log ORDER BY id DESC LIMIT ?", (LOG_EXPORT_LIMIT,)
"SELECT ts, username, action, target, details FROM audit_log ORDER BY id DESC"
).fetchall()
return [dict(r) for r in rows] if rows else None
@@ -5523,19 +5608,28 @@ def _export_logs(conn):
EXPORT_BUILDERS = {
"devices": _export_devices, "switches": _export_switches, "credentials": _export_credentials,
"users": _export_users, "groups": _export_groups, "ldap": _export_ldap, "dhcp": _export_dhcp,
"logs": _export_logs,
"nginx": _export_nginx, "logs": _export_logs,
}
def _import_devices(conn, items):
n = 0
for d in items:
cred_id = None
if d.get("credential_name"):
row = conn.execute(
"SELECT id FROM credentials WHERE name=? AND deleted_at IS NULL", (d["credential_name"],)
).fetchone()
cred_id = row["id"] if row else None
conn.execute(
"""INSERT INTO devices (mac, ip, port, name, switch_hostname, is_active) VALUES (?, ?, ?, ?, ?, ?)
"""INSERT INTO devices (mac, ip, port, name, switch_hostname, is_active, ssh_port, credential_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(mac) DO UPDATE SET ip=excluded.ip, port=excluded.port, name=excluded.name,
switch_hostname=excluded.switch_hostname, is_active=excluded.is_active
switch_hostname=excluded.switch_hostname, is_active=excluded.is_active,
ssh_port=excluded.ssh_port, credential_id=excluded.credential_id
WHERE devices.deleted_at IS NULL""",
(d["mac"], d["ip"], d.get("port"), d["name"], d.get("switch_hostname"), d.get("is_active", 1)),
(d["mac"], d["ip"], d.get("port"), d["name"], d.get("switch_hostname"), d.get("is_active", 1),
d.get("ssh_port"), cred_id),
)
n += 1
return n
@@ -5551,10 +5645,11 @@ def _import_switches(conn, items):
).fetchone()
cred_id = row["id"] if row else None
conn.execute(
"""INSERT INTO switches (hostname, ip, credential_id) VALUES (?, ?, ?)
ON CONFLICT(hostname) DO UPDATE SET ip=excluded.ip, credential_id=excluded.credential_id
"""INSERT INTO switches (hostname, ip, ssh_port, credential_id) VALUES (?, ?, ?, ?)
ON CONFLICT(hostname) DO UPDATE SET ip=excluded.ip, ssh_port=excluded.ssh_port,
credential_id=excluded.credential_id
WHERE switches.deleted_at IS NULL""",
(s["hostname"], s["ip"], cred_id),
(s["hostname"], s["ip"], s.get("ssh_port"), cred_id),
)
n += 1
return n
@@ -5564,10 +5659,11 @@ def _import_credentials(conn, items):
n = 0
for c in items:
conn.execute(
"""INSERT INTO credentials (name, username, password) VALUES (?, ?, ?)
ON CONFLICT(name) DO UPDATE SET username=excluded.username, password=excluded.password
"""INSERT INTO credentials (name, username, password, category) VALUES (?, ?, ?, ?)
ON CONFLICT(name) DO UPDATE SET username=excluded.username, password=excluded.password,
category=excluded.category
WHERE credentials.deleted_at IS NULL""",
(c["name"], c["username"], encrypt_password(c["password"])),
(c["name"], c["username"], encrypt_password(c["password"]), c.get("category", "switch")),
)
n += 1
return n
@@ -5620,7 +5716,10 @@ def _import_groups(conn, items):
if existing:
group_id = existing["id"]
else:
cur = conn.execute("INSERT INTO groups (name, is_default, is_system) VALUES (?, 0, 0)", (g["name"],))
cur = conn.execute(
"INSERT INTO groups (name, is_default, is_system) VALUES (?, ?, 0)",
(g["name"], g.get("is_default", 0)),
)
group_id = cur.lastrowid
conn.execute("DELETE FROM group_permissions WHERE group_id=?", (group_id,))
valid_perms = [p for p in g.get("permissions", []) if p in ALL_PERMISSION_KEYS]
@@ -5653,10 +5752,27 @@ def _import_ldap(conn, data):
)
n = 0
for m in data.get("group_mappings", []):
app_group_name = m.get("app_group_name")
if app_group_name == "Admin":
app_group_id = "admin"
else:
grow = conn.execute(
"SELECT id FROM groups WHERE name=? AND deleted_at IS NULL", (app_group_name,)
).fetchone()
if not grow:
continue # Rechtegruppe existiert auf dem Zielsystem (noch) nicht -- Zuordnung überspringen statt kaputt anzulegen
app_group_id = grow["id"]
conn.execute(
"INSERT INTO ldap_group_mappings (ad_group_dn, ad_group_name, app_group_id) VALUES (?, ?, ?) "
"ON CONFLICT(ad_group_dn) DO UPDATE SET ad_group_name=excluded.ad_group_name, app_group_id=excluded.app_group_id",
(m["ad_group_dn"], m["ad_group_name"], m["app_group_id"]),
(m["ad_group_dn"], m["ad_group_name"], app_group_id),
)
n += 1
for m in data.get("fileshare_mappings", []):
conn.execute(
"INSERT INTO ldap_fileshare_mappings (ad_group_dn, ad_group_name, share_label, share_unc) VALUES (?, ?, ?, ?) "
"ON CONFLICT(ad_group_dn, share_unc) DO UPDATE SET ad_group_name=excluded.ad_group_name, share_label=excluded.share_label",
(m["ad_group_dn"], m["ad_group_name"], m["share_label"], m["share_unc"]),
)
n += 1
return n
@@ -5709,6 +5825,19 @@ def _import_dhcp(conn, data):
return n
def _import_nginx(conn, data):
n = 0
for key, value in data.get("settings", {}).items():
if key in NGINX_SETTING_KEYS:
conn.execute(
"INSERT INTO settings (key, value) VALUES (?, ?) "
"ON CONFLICT(key) DO UPDATE SET value=excluded.value",
(key, value),
)
n += 1
return n
def _import_logs(conn, items):
conn.executemany(
"INSERT INTO audit_log (ts, username, action, target, details) VALUES (?, ?, ?, ?, ?)",
@@ -5720,7 +5849,7 @@ def _import_logs(conn, items):
IMPORT_APPLIERS = {
"devices": _import_devices, "switches": _import_switches, "credentials": _import_credentials,
"users": _import_users, "groups": _import_groups, "ldap": _import_ldap, "dhcp": _import_dhcp,
"logs": _import_logs,
"nginx": _import_nginx, "logs": _import_logs,
}
_pending_imports = {}
@@ -5766,6 +5895,11 @@ def export_data():
flash("Keine Daten in den ausgewählten Kategorien vorhanden — nichts exportiert.", "danger")
return redirect(url_for("settings_import_export"))
if "credentials" in payload:
known_hosts_content = _read_known_hosts_content()
if known_hosts_content:
payload["known_hosts"] = known_hosts_content
salt = secrets.token_bytes(16)
export_fernet = _derive_export_fernet(passphrase, salt)
encrypted_payload = export_fernet.encrypt(json.dumps(payload).encode("utf-8"))
@@ -5778,8 +5912,10 @@ def export_data():
"payload": encrypted_payload.decode("utf-8"),
}
summary = ", ".join(f"{EXPORT_SECTION_LABELS[k]}: {counts[k]}" for k in payload)
log_action("data.export", ", ".join(EXPORT_SECTION_LABELS[k] for k in payload), summary)
summary = ", ".join(f"{EXPORT_SECTION_LABELS[k]}: {counts[k]}" for k in payload if k in EXPORT_SECTION_LABELS)
if "known_hosts" in payload:
summary += " (inkl. known_hosts)"
log_action("data.export", ", ".join(EXPORT_SECTION_LABELS[k] for k in payload if k in EXPORT_SECTION_LABELS), summary)
filename = f"tesm_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
response = jsonify(envelope)
@@ -5878,6 +6014,9 @@ def import_apply():
conn.close()
summary = ", ".join(f"{EXPORT_SECTION_LABELS.get(k, k)}: {results[k]}" for k in selected)
if "credentials" in selected and payload.get("known_hosts"):
added = _merge_known_hosts_content(payload["known_hosts"])
summary += f", known_hosts: {added} neue Zeile(n)"
log_action("data.import", ", ".join(EXPORT_SECTION_LABELS.get(k, k) for k in selected), summary)
flash(f"Import abgeschlossen: {summary}.", "success")
return redirect(url_for("settings_import_export"))