soc_ai/bootstrap_credential.py

Rama main · réplica de lectura del repositorio original.

"""Location and lifecycle of the one-shot bootstrap-admin password sidecar.

Two places need the SAME path: startup writes it when it invents the initial
admin password (:func:`soc_ai.main._persist_bootstrap_credential`) and the
self-service password change retires it (``POST /api/v1/me/password``). A route
module cannot import ``soc_ai.main`` — ``main`` imports the routers — and a
second copy of the filename in the route would drift silently the day the
sidecar moves, so the path logic lives here and both sides call it.
"""

from __future__ import annotations

import hashlib
import logging
from pathlib import Path
from typing import Any

_LOGGER = logging.getLogger(__name__)

#: The account :func:`soc_ai.store.auth.bootstrap_admin` creates on a fresh DB.
BOOTSTRAP_ADMIN_USERNAME = "admin"

_FILENAME = "bootstrap-admin-password.txt"


def bootstrap_credential_path(settings: Any) -> Path:
    """Where the generated bootstrap password is parked (mode 0600)."""
    return Path(settings.soc_ai_data_dir) / _FILENAME


def clear_bootstrap_credential(settings: Any, username: str) -> bool:
    """Retire the sidecar once the bootstrap admin has set its own password.

    Returns True only when a file was actually removed. Callers use this
    strictly as a post-success cleanup, so every failure mode is non-fatal:

    - a different account changed its password → not ours to delete;
    - the file was never written (the operator supplied
      ``BOOTSTRAP_ADMIN_PASSWORD``) or was already deleted by hand → nothing
      to do, and emphatically not an error the analyst should see;
    - the data dir is read-only → log it, but the password change itself
      already succeeded and must still report success.
    """
    if username != BOOTSTRAP_ADMIN_USERNAME:
        return False
    path = bootstrap_credential_path(settings)
    try:
        path.unlink()
    except FileNotFoundError:
        return False
    except OSError:
        _LOGGER.warning(
            "could not remove the bootstrap credential sidecar at %s after the "
            "admin password change — delete it by hand",
            path,
        )
        return False
    _LOGGER.info("removed the bootstrap credential sidecar at %s (password changed)", path)
    return True


# bcrypt reads at most 72 bytes; a longer value can never be the stored password.
_BCRYPT_MAX_BYTES = 72

# (stored hash, digest of the candidate) -> did the candidate verify. A bcrypt
# check costs a quarter second, and ``GET /me`` asks on every page load. The
# stored hash changes with the password, so a cached answer can never outlive
# the password it describes.
_GENERATED_CACHE: dict[tuple[str, str], bool] = {}
_GENERATED_CACHE_MAX = 64


#: Where a generated password can sit. The values are the doctor's words.
SOURCE_SIDECAR = "sidecar"
SOURCE_ENV = "env"


def _generated_candidates(settings: Any) -> list[tuple[str, str]]:
    """The passwords soc-ai generated for the bootstrap admin, if any remain.

    Two sources: the sidecar the first start writes, and the
    ``BOOTSTRAP_ADMIN_PASSWORD`` that ``setup.sh`` generates into ``.env``. Both
    sit in a plaintext file, so an admin still on either one has a password a
    reader of that file knows.
    """
    out: list[tuple[str, str]] = []
    try:
        text = bootstrap_credential_path(settings).read_text(encoding="utf-8").strip()
    except (OSError, UnicodeDecodeError):
        text = ""
    if text:
        out.append((SOURCE_SIDECAR, text))
    fixed = getattr(settings, "bootstrap_admin_password", None)
    if fixed is not None:
        value = str(fixed.get_secret_value())
        if value and value != text:
            out.append((SOURCE_ENV, value))
    return [(src, c) for src, c in out if len(c.encode()) <= _BCRYPT_MAX_BYTES]


async def generated_password_source(settings: Any, username: str, password_hash: str) -> str | None:
    """Where the bootstrap admin's live generated password sits, or None.

    Only the bootstrap account can hold one. Another account whose password
    happens to equal it is not flagged: nothing told that user the value.
    """
    if username != BOOTSTRAP_ADMIN_USERNAME or not password_hash:
        return None
    from soc_ai.store.auth import verify_password  # noqa: PLC0415 - keep this module light

    for source, candidate in _generated_candidates(settings):
        key = (password_hash, hashlib.sha256(candidate.encode()).hexdigest())
        hit = _GENERATED_CACHE.get(key)
        if hit is None:
            try:
                hit = await verify_password(candidate, password_hash)
            except ValueError:
                hit = False
            if len(_GENERATED_CACHE) >= _GENERATED_CACHE_MAX:
                _GENERATED_CACHE.clear()
            _GENERATED_CACHE[key] = hit
        if hit:
            return source
    return None


async def uses_generated_password(settings: Any, username: str, password_hash: str) -> bool:
    """True when the bootstrap admin still signs in with a generated password."""
    return await generated_password_source(settings, username, password_hash) is not None