117 lines
4.7 KiB
Python
117 lines
4.7 KiB
Python
"""Admin-Funktionen: Passwortaenderung (GUI+Grafana gemeinsam) und
|
|
Regel-Neuanwendung (Ausbaustufe 4 Task 2).
|
|
|
|
KRITISCH (siehe Global Constraints des Plans): die .env wird im API-Container
|
|
als Einzeldatei-Bind-Mount eingehaengt. Ein solcher Mount folgt dem Inode -
|
|
die Datei MUSS in-place ueberschrieben werden (open r+/truncate unter flock),
|
|
NIEMALS ueber Temp-Datei+rename (das erzeugt einen neuen Inode und würde vom
|
|
Container nicht mehr gesehen).
|
|
"""
|
|
import fcntl
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
from base64 import b64encode
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import current_password_hash, hash_password, verify_password
|
|
from app.config import get_settings
|
|
from app.models.tables import Transaction
|
|
from app.services.categorize import apply_rules
|
|
|
|
|
|
def _sync_grafana_password(old: str, new: str, base_url: str) -> None:
|
|
"""PUT /api/admin/users/1/password gegen Grafana, Basic-Auth mit dem
|
|
ALTEN Passwort (admin:old) - Grafana authentifiziert den Request noch mit
|
|
dem bisherigen Passwort, aendert es aber auf `new`. Wird IMMER vor dem
|
|
Schreiben der .env aufgerufen: schlaegt Grafana fehl, bleibt die Datei
|
|
unangetastet (kein inkonsistenter Zwischenzustand GUI-Hash != Grafana)."""
|
|
url = f"{base_url.rstrip('/')}/api/admin/users/1/password"
|
|
auth_header = "Basic " + b64encode(f"admin:{old}".encode()).decode()
|
|
body = json.dumps({"password": new}).encode()
|
|
req = urllib.request.Request(
|
|
url, data=body, method="PUT",
|
|
headers={"Authorization": auth_header, "Content-Type": "application/json"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
if resp.status >= 400:
|
|
raise RuntimeError(
|
|
f"Grafana-Passwortänderung fehlgeschlagen (HTTP {resp.status})."
|
|
)
|
|
except urllib.error.HTTPError as exc:
|
|
raise RuntimeError(
|
|
f"Grafana-Passwortänderung fehlgeschlagen (HTTP {exc.code})."
|
|
) from exc
|
|
except urllib.error.URLError as exc:
|
|
raise RuntimeError(
|
|
f"Grafana-Passwortänderung fehlgeschlagen: {exc.reason}"
|
|
) from exc
|
|
except (TimeoutError, OSError) as exc:
|
|
# Ein Hang in der Lesephase (nach erfolgreichem Verbindungsaufbau)
|
|
# kann ein rohes TimeoutError/socket.timeout werfen, das urlopen NICHT
|
|
# in ein URLError verpackt (URLError selbst ist zwar ein OSError, aber
|
|
# dieser Pfad faengt Faelle ab, die es nicht bis dorthin schaffen).
|
|
# Ohne diesen Fang wuerde die Exception bis in den Router durchschlagen
|
|
# und dort als 500 statt als deutsche 400-Fehlermeldung enden.
|
|
raise RuntimeError(
|
|
f"Grafana-Passwortänderung fehlgeschlagen: {exc}"
|
|
) from exc
|
|
|
|
|
|
def _rewrite_env_file(env_file: Path, new_password: str, new_hash: str) -> None:
|
|
"""Ersetzt FB_PASSWORD und FB_GUI_PASSWORD_HASH IN-PLACE (open r+ unter
|
|
flock, truncate) - der Inode der Datei bleibt unveraendert, siehe
|
|
Modul-Docstring."""
|
|
with open(env_file, "r+", encoding="utf-8") as f:
|
|
fcntl.flock(f, fcntl.LOCK_EX)
|
|
try:
|
|
lines = f.read().splitlines()
|
|
out = []
|
|
for line in lines:
|
|
if line.startswith("FB_PASSWORD="):
|
|
out.append(f"FB_PASSWORD='{new_password}'")
|
|
elif line.startswith("FB_GUI_PASSWORD_HASH="):
|
|
out.append(f"FB_GUI_PASSWORD_HASH='{new_hash}'")
|
|
else:
|
|
out.append(line)
|
|
f.seek(0)
|
|
f.write("\n".join(out) + "\n")
|
|
f.truncate()
|
|
finally:
|
|
fcntl.flock(f, fcntl.LOCK_UN)
|
|
|
|
|
|
def change_password(old: str, new: str) -> None:
|
|
"""Aendert das gemeinsame GUI-/Grafana-Passwort. Reihenfolge ist bindend:
|
|
1) altes Passwort gegen den aktuellen Hash verifizieren,
|
|
2) Grafana AKTUALISIEREN (bricht bei Fehler ab, ohne die .env
|
|
anzufassen),
|
|
3) erst danach die .env in-place neu schreiben.
|
|
"""
|
|
if not verify_password(old, current_password_hash()):
|
|
raise ValueError("Das alte Passwort ist falsch.")
|
|
|
|
settings = get_settings()
|
|
_sync_grafana_password(old, new, settings.grafana_url)
|
|
|
|
new_hash = hash_password(new)
|
|
_rewrite_env_file(settings.env_file, new, new_hash)
|
|
|
|
|
|
def apply_rules_retroactively(session: Session) -> int:
|
|
"""Wendet die aktuellen Kategorie-Regeln rueckwirkend auf alle
|
|
bestaetigten, noch unkategorisierten Buchungen an und committet."""
|
|
txs = session.execute(
|
|
select(Transaction).where(
|
|
Transaction.status == "confirmed",
|
|
Transaction.category_id.is_(None),
|
|
)
|
|
).scalars().all()
|
|
hits = apply_rules(session, txs)
|
|
session.commit()
|
|
return hits
|