"""Magazyn kont zarządzanych z aplikacji (PRE-27). KONTO ADMINISTRACYJNE NIE LEŻY TUTAJ. Pochodzi z konfiguracji środowiska (APP_USER/APP_PASSWORD albo APP_USERS) i dlatego NIE DA SIĘ go skasować ani ograniczyć z poziomu ekranu — nawet przez pomyłkę, nawet spreparowanym żądaniem. Ekran zarządzania kontami operuje wyłącznie na tym pliku. Format: JSON `{"users": {"login": {"secret": "scrypt$...", "features": [...], "note": "...", "created": "ISO"}}}`. Hasła WYŁĄCZNIE jako hash scrypt — ten sam mechanizm, co dla kont z APP_USERS (security.hash_password), więc nie ma tu drugiej implementacji do rozjechania. Zapis jest ATOMOWY (plik tymczasowy + os.replace) i pod blokadą procesu. Plik leży na NFS, dzieli go kilka replik, a zwykły zapis „w miejscu" po przerwaniu zostawiłby plik obcięty — czyli utratę WSZYSTKICH kont naraz. os.replace jest w obrębie jednego katalogu operacją niepodzielną. """ from __future__ import annotations import json import os import tempfile import threading from datetime import datetime, timezone from app import features _lock = threading.Lock() class AccountsUnavailable(RuntimeError): """Pliku kont nie da się odczytać albo zapisać — problem z magazynem. Osobny wyjątek, bo to NIE JEST błąd danych: konta mogą być w porządku, tylko katalog jest niedostępny (złe prawa na udziale, nieprzemontowany wolumen, brak miejsca). Ekran ma wtedy POWIEDZIEĆ, co i gdzie jest nie tak — gołe 500 zostawia administratora z niczym, a to jedyny ekran, z którego może to naprawić.""" def store_path() -> str: """Ścieżka pliku kont. Domyślnie WŁASNY podkatalog prezentacji na NFS, nie katalog z bazami: prezentacja nie ma powodu widzieć baz interpretacyjnych, a zamontowanie ich tutaj obeszłoby bokiem zamknięcie dostępu z DAN-25.""" return os.getenv("ACCOUNTS_FILE", "/app/state/accounts.json") def _read() -> dict: path = store_path() try: with open(path, encoding="utf-8") as fh: data = json.load(fh) except FileNotFoundError: # Pierwsze uruchomienie: pliku jeszcze nie ma i to jest normalne. return {"users": {}} except json.JSONDecodeError: # Plik jest, ale nieczytelny. NIE nadpisujemy go pustym zbiorem — # to skasowałoby wszystkie konta przy pierwszym zapisie. raise AccountsUnavailable( f"Plik kont {path} jest uszkodzony (niepoprawny JSON). " f"Nic nie zostało zmienione — napraw albo usuń plik ręcznie.") from None except OSError as e: raise AccountsUnavailable( f"Nie mogę odczytać pliku kont {path}: {e.strerror or e}. " f"Najczęstsza przyczyna: prawa do katalogu na udziale albo " f"niezamontowany wolumen.") from e users = data.get("users") return {"users": users if isinstance(users, dict) else {}} def _write(data: dict) -> None: path = store_path() try: os.makedirs(os.path.dirname(path) or ".", exist_ok=True) except OSError as e: raise AccountsUnavailable( f"Nie mogę utworzyć katalogu na plik kont ({os.path.dirname(path)}): " f"{e.strerror or e}.") from e # Plik tymczasowy MUSI powstać w tym samym katalogu — os.replace jest # niepodzielne tylko w obrębie jednego systemu plików. # mkstemp też musi być w klamrze: przy katalogu tylko do odczytu wywala się # ONO pierwsze, jeszcze zanim dojdzie do zapisu i podmiany. try: fd, tmp = tempfile.mkstemp(dir=os.path.dirname(path) or ".", suffix=".tmp") except OSError as e: raise AccountsUnavailable( f"Nie mogę zapisać pliku kont {path}: {e.strerror or e}. " f"Najczęstsza przyczyna: udział zamontowany tylko do odczytu albo " f"prawa katalogu. Nic nie zostało zmienione.") from e try: with os.fdopen(fd, "w", encoding="utf-8") as fh: json.dump(data, fh, ensure_ascii=False, indent=1, sort_keys=True) fh.flush() os.fsync(fh.fileno()) os.replace(tmp, path) except OSError as e: try: os.unlink(tmp) except OSError: pass raise AccountsUnavailable( f"Nie mogę zapisać pliku kont {path}: {e.strerror or e}. " f"Najczęstsza przyczyna: udział zamontowany tylko do odczytu albo " f"prawa katalogu. Nic nie zostało zmienione.") from e except BaseException: try: os.unlink(tmp) except OSError: pass raise def all_users() -> dict[str, dict]: """Konta zarządzane: login → {features, note, created}. BEZ sekretów. Sekret nie wychodzi z tego modułu — ekran nigdy go nie potrzebuje, a to, czego nie ma w kontekście szablonu, nie wycieknie do HTML-a przez pomyłkę.""" out = {} for login, row in _read()["users"].items(): out[login] = { "features": sorted(features.normalise(row.get("features"))), "note": str(row.get("note") or ""), "created": str(row.get("created") or ""), } return out def secret_of(login: str) -> str | None: return (_read()["users"].get(login) or {}).get("secret") def permissions_of(login: str) -> frozenset[str]: return features.normalise((_read()["users"].get(login) or {}).get("features")) def exists(login: str) -> bool: return login in _read()["users"] def create(login: str, password: str, granted, note: str = "") -> None: """Zakłada konto. Podnosi ValueError, gdy login zajęty albo dane puste.""" from app.security import hash_password login = (login or "").strip() if not login: raise ValueError("Login nie może być pusty.") if not (password or "").strip(): raise ValueError("Hasło nie może być puste.") with _lock: data = _read() if login in data["users"]: raise ValueError(f"Konto „{login}” już istnieje.") data["users"][login] = { "secret": hash_password(password), "features": sorted(features.normalise(granted)), "note": (note or "").strip(), "created": datetime.now(timezone.utc).isoformat(timespec="seconds"), } _write(data) def update(login: str, granted=None, password: str = "", note: str | None = None) -> None: """Zmiana uprawnień, hasła albo opisu. Puste hasło = bez zmiany hasła.""" from app.security import hash_password with _lock: data = _read() row = data["users"].get(login) if row is None: raise ValueError(f"Nie ma konta „{login}”.") if granted is not None: row["features"] = sorted(features.normalise(granted)) if (password or "").strip(): row["secret"] = hash_password(password) if note is not None: row["note"] = note.strip() _write(data) def delete(login: str) -> None: with _lock: data = _read() if data["users"].pop(login, None) is None: raise ValueError(f"Nie ma konta „{login}”.") _write(data)