85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
import secrets
|
|
import subprocess
|
|
|
|
from db import get_database_connection
|
|
from encryption import decrypt_psk, encrypt_psk
|
|
|
|
|
|
# Diagnosezugang (02.10.2026, siehe migrations/0006_diagnosezugang.sql):
|
|
# root-Zugang auf neu installierten Geraeten einer Organisation, nur vom
|
|
# Root-Admin schaltbar und einsehbar. encrypt_psk/decrypt_psk sind
|
|
# generische Fernet-Helfer (Name historisch vom WLAN-PSK).
|
|
|
|
|
|
def fetch_diagnosezugang(organization_id: str) -> dict:
|
|
with get_database_connection() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT aktiv, passwort_encrypted, updated_at FROM diagnosezugang WHERE organization_id = %s",
|
|
(organization_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
|
|
if row is None:
|
|
return {"aktiv": False, "passwort": None, "updated_at": None}
|
|
|
|
aktiv, passwort_encrypted, updated_at = row
|
|
return {"aktiv": aktiv, "passwort": decrypt_psk(passwort_encrypted), "updated_at": updated_at}
|
|
|
|
|
|
def aktiviere_diagnosezugang(organization_id: str, neues_passwort: bool = False) -> None:
|
|
"""
|
|
Schaltet den Zugang ein. Ein vorhandenes Passwort bleibt erhalten (sonst
|
|
passten frisch installierte und schon vorhandene Geraete nicht mehr
|
|
zusammen), ausser neues_passwort=True.
|
|
"""
|
|
|
|
vorhanden = fetch_diagnosezugang(organization_id)
|
|
passwort = vorhanden["passwort"]
|
|
if neues_passwort or not passwort:
|
|
passwort = secrets.token_urlsafe(18)
|
|
|
|
with get_database_connection() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO diagnosezugang (organization_id, aktiv, passwort_encrypted, updated_at)
|
|
VALUES (%s, TRUE, %s, CURRENT_TIMESTAMP)
|
|
ON CONFLICT (organization_id) DO UPDATE
|
|
SET aktiv = TRUE,
|
|
passwort_encrypted = EXCLUDED.passwort_encrypted,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
""",
|
|
(organization_id, encrypt_psk(passwort)),
|
|
)
|
|
|
|
|
|
def deaktiviere_diagnosezugang(organization_id: str) -> None:
|
|
with get_database_connection() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"UPDATE diagnosezugang SET aktiv = FALSE, updated_at = CURRENT_TIMESTAMP WHERE organization_id = %s",
|
|
(organization_id,),
|
|
)
|
|
|
|
|
|
def root_password_hash_fuer_build(organization_id: str) -> str | None:
|
|
"""
|
|
SHA-512-crypt-Hash fuer build_customer_iso.sh, oder None wenn der
|
|
Zugang aus ist. openssl statt Pythons crypt-Modul (seit 3.13 entfernt);
|
|
$6$ versteht jede glibc/libxcrypt der drei Zielsysteme.
|
|
"""
|
|
|
|
zugang = fetch_diagnosezugang(organization_id)
|
|
if not zugang["aktiv"] or not zugang["passwort"]:
|
|
return None
|
|
|
|
result = subprocess.run(
|
|
["openssl", "passwd", "-6", "-stdin"],
|
|
input=zugang["passwort"] + "\n",
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
return result.stdout.strip()
|