astrololo/services/presentation/app/accounts.py

"""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()


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:
    try:
        with open(store_path(), encoding="utf-8") as fh:
            data = json.load(fh)
    except (FileNotFoundError, json.JSONDecodeError):
        return {"users": {}}
    users = data.get("users")
    return {"users": users if isinstance(users, dict) else {}}


def _write(data: dict) -> None:
    path = store_path()
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    # Plik tymczasowy MUSI powstać w tym samym katalogu — os.replace jest
    # niepodzielne tylko w obrębie jednego systemu plików.
    fd, tmp = tempfile.mkstemp(dir=os.path.dirname(path) or ".", suffix=".tmp")
    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 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)