skripts/: Admin-Hilfsskripte für Bulk-Aktionen ins Repo aufgenommen

- assign_credential_to_all_devices.py: weist allen aktiven Geraeten auf
  einen Schlag dieselben Zugangsdaten zu (Name als optionales Argument,
  Standard 'HAPS'), inkl. Eintrag im Aenderungsverlauf.
- accept_all_hostkeys.py: akzeptiert die SSH-Host-Keys aller aktiven
  Geraete in einem Rutsch (paralleler Verbindungsaufbau + Schluesseltausch,
  kein Login) -- nutzt dieselbe known_hosts-Datei wie die App, ein
  geaenderter (statt unbekannter) Key wird bewusst NICHT automatisch
  uebernommen, sondern separat als Warnung ausgegeben.

Beide live gegen POETEST verifiziert.
This commit is contained in:
2026-08-14 00:13:11 +02:00
parent d72db26a86
commit d77327cee1
2 changed files with 145 additions and 0 deletions
+104
View File
@@ -0,0 +1,104 @@
#!/usr/bin/env python3
"""Akzeptiert die SSH-Host-Keys aller aktiven Geraete (Clients) in einem
Rutsch -- macht technisch dasselbe wie "Verbindung testen" im Browser-
Terminal (Verbindungsaufbau -> Host-Key abfragen -> "yes" -> trennen),
nur automatisiert fuer alle Geraete statt eines einzelnen.
Nutzt bewusst dieselbe known_hosts-Datei/dasselbe Format wie die App
(SSH_KNOWN_HOSTS_PATH in app.py) -- Geraete, deren Key hier akzeptiert
wird, muessen danach in der Weboberflaeche NICHT mehr einzeln bestaetigt
werden (SSH-Verbindungstest und Wartungs-Aktionen teilen sich dieselbe
Datei).
Aufruf auf dem Server: sudo /srv/tesm/venv/bin/python3 accept_all_hostkeys.py
"""
import socket
import sqlite3
from concurrent.futures import ThreadPoolExecutor, as_completed
import paramiko
DB_PATH = "/srv/tesm/sqlite.db"
KNOWN_HOSTS_PATH = "/srv/tesm/known_hosts"
DEFAULT_SSH_PORT = 22
CONNECT_TIMEOUT = 4
MAX_WORKERS = 32
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
devices = conn.execute(
"SELECT name, ip, ssh_port FROM devices WHERE deleted_at IS NULL ORDER BY name"
).fetchall()
conn.close()
def fetch_key(dev):
"""Nur der Handshake (Verbindungsaufbau + Schluesseltausch), KEIN Login
-- exakt der Teil, der im Browser-Terminal vor der Passwortabfrage
passiert. Parallel je Gerät, sonst dauert eine Liste mit vielen
offline Geraeten (je bis zu CONNECT_TIMEOUT Sekunden) unnoetig lange."""
host = dev["ip"]
port = dev["ssh_port"] or DEFAULT_SSH_PORT
try:
sock = socket.create_connection((host, port), timeout=CONNECT_TIMEOUT)
transport = paramiko.Transport(sock)
transport.start_client(timeout=CONNECT_TIMEOUT)
server_key = transport.get_remote_server_key()
transport.close()
return dev, server_key, None
except (OSError, paramiko.SSHException, EOFError) as e:
return dev, None, str(e)
hostkeys = paramiko.HostKeys()
try:
hostkeys.load(KNOWN_HOSTS_PATH)
except (OSError, paramiko.SSHException):
pass
accepted, already_known, changed, unreachable = [], [], [], []
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
futures = [pool.submit(fetch_key, dev) for dev in devices]
for future in as_completed(futures):
dev, server_key, err = future.result()
host = dev["ip"]
port = dev["ssh_port"] or DEFAULT_SSH_PORT
lookup_host = host if port == DEFAULT_SSH_PORT else f"[{host}]:{port}"
if err is not None:
unreachable.append((dev["name"], host, err))
continue
existing = hostkeys.lookup(lookup_host)
if existing and existing.get(server_key.get_name()):
if existing[server_key.get_name()] == server_key:
already_known.append((dev["name"], host))
continue
else:
# Nicht blind ueberschreiben -- das waere genau das MITM-
# Szenario, vor dem die App im Terminal auch schuetzt (harte
# Ablehnung statt automatischer Uebernahme eines GEAENDERTEN
# Keys).
changed.append((dev["name"], host))
continue
hostkeys.add(lookup_host, server_key.get_name(), server_key)
accepted.append((dev["name"], host))
hostkeys.save(KNOWN_HOSTS_PATH)
print(f"Neu akzeptiert: {len(accepted)}")
for name, host in accepted:
print(f" + {name} ({host})")
print(f"Bereits bekannt: {len(already_known)}")
if changed:
print(f"ACHTUNG -- Host-Key geaendert, NICHT automatisch uebernommen ({len(changed)}):")
for name, host in changed:
print(f" ! {name} ({host}) -- manuell in der App pruefen (moeglicher MITM oder Geraetetausch)")
print(f"Nicht erreichbar: {len(unreachable)}")
for name, host, err in unreachable:
print(f" - {name} ({host}): {err}")
@@ -0,0 +1,41 @@
#!/usr/bin/env python3
"""Weist allen aktiven Geraeten (Clients) auf einen Schlag dieselben
Zugangsdaten zu -- spart das manuelle Durchklicken jedes einzelnen
Geraets in der Weboberflaeche.
Aufruf auf dem Server:
sudo /srv/tesm/venv/bin/python3 assign_credential_to_all_devices.py [NAME]
NAME ist der Name des Zugangsdaten-Eintrags (Standard: "HAPS")."""
import sqlite3
import sys
from datetime import datetime
DB_PATH = "/srv/tesm/sqlite.db"
CRED_NAME = sys.argv[1] if len(sys.argv) > 1 else "HAPS"
conn = sqlite3.connect(DB_PATH)
cred = conn.execute(
"SELECT id FROM credentials WHERE name = ? AND deleted_at IS NULL", (CRED_NAME,)
).fetchone()
if not cred:
print(f"Fehler: Zugangsdaten '{CRED_NAME}' nicht gefunden (oder im Papierkorb).")
raise SystemExit(1)
credential_id = cred[0]
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
cur = conn.execute(
"UPDATE devices SET credential_id=?, last_modified_by=?, last_modified_at=? WHERE deleted_at IS NULL",
(credential_id, "Admin (Bulk)", now),
)
count = cur.rowcount
conn.execute(
"INSERT INTO audit_log (ts, username, action, target, details) VALUES (?, ?, ?, ?, ?)",
(now, "system", "devices.bulk_credential", CRED_NAME, f"{count} Geräte auf Zugangsdaten '{CRED_NAME}' gesetzt"),
)
conn.commit()
conn.close()
print(f"{count} Geräte auf Zugangsdaten '{CRED_NAME}' (id={credential_id}) gesetzt.")