src/lib/auth/user_store.py

"""DB-backed user store for the low-privilege ``reader`` role.

This is the store for the preview / Reading-Room accounts an admin creates and
manages from the UI. It is **deliberately separate** from the env/TOML
credentials in :mod:`src.lib.auth.repo` (``get_credentials``):

  - Admins (and any bootstrap ``viewer``) come from ``AUTH_CREDENTIALS_TOML`` /
    Streamlit secrets / ``auth/credentials.toml`` — the bootstrap layer. A bad
    or empty ``app_users`` table can therefore never lock the operator out.
  - The create path forces the role to ``reader`` (see :func:`create_app_user`)
    — no caller can mint an admin through creation. Promotion to ``admin``/
    ``viewer`` happens ONLY through :func:`set_app_user_role`, which is gated
    by the admin router's last-admin lock and audit log. The stored role is
    now honored at login (see :func:`get_app_user_for_auth`); safety comes
    from bootstrap-first ordering in :func:`src.lib.auth.repo.authenticate`,
    not from forcing ``reader`` here.

Passwords are PBKDF2 hashes (:func:`src.lib.auth.passwords.hash_password`) —
never plaintext. ``authenticate()`` in ``repo.py`` checks the env/TOML source
FIRST and only falls back to this table for usernames not found there.

Schema lives in ``src/stage1_inventory/schema.py`` (``app_users``); the table
is created via ``CREATE TABLE IF NOT EXISTS`` by ``init_schema``, so these
accessors never run DDL themselves.
"""

from __future__ import annotations

from dataclasses import dataclass

from src.stage1_inventory.db import connect

from .models import User
from .passwords import hash_password


@dataclass(frozen=True)
class AppUser:
    """One ``app_users`` row, projected for the admin management UI.

    Never carries the password hash — the admin endpoints must not put it on
    the wire. Use :func:`get_app_user_for_auth` for the auth path that needs
    the stored hash.
    """

    username: str
    role: str
    display_name: str | None
    disabled: bool
    created_at: str | None


_VALID_ROLES = ("reader", "viewer", "admin")


def _normalize_username(username: str | None) -> str:
    """Lowercase + strip, mirroring the env/TOML credential rule in repo.py."""
    return (username or "").strip().lower()


def list_app_users() -> list[AppUser]:
    """Every DB-backed account, newest first. No password hashes."""
    with connect() as conn:
        rows = conn.execute(
            """
            SELECT username, role, display_name, disabled, created_at
              FROM app_users
             ORDER BY created_at DESC, username
            """
        ).fetchall()
    return [
        AppUser(
            username=r["username"],
            role=r["role"],
            display_name=r["display_name"],
            disabled=bool(r["disabled"]),
            created_at=r["created_at"],
        )
        for r in rows
    ]


def app_user_exists(username: str) -> bool:
    uname = _normalize_username(username)
    if not uname:
        return False
    with connect() as conn:
        row = conn.execute(
            "SELECT 1 FROM app_users WHERE username = ?", (uname,)
        ).fetchone()
    return row is not None


def create_app_user(
    username: str,
    password: str,
    display_name: str | None = None,
) -> AppUser:
    """Insert a new reader account and return its (hash-free) projection.

    The username is lowercased like the env/TOML credentials. Empty username
    or password is rejected with ``ValueError``; a duplicate username raises
    ``ValueError`` too (the admin router maps that to a 409). The password is
    hashed with PBKDF2 before storage — never plaintext.

    The role is always ``reader`` — this is a reader-only table by construction.
    The ``role`` parameter has been removed so no caller can accidentally or
    maliciously create an admin/viewer row via this path.
    """
    import sqlite3 as _sqlite3

    uname = _normalize_username(username)
    if not uname:
        raise ValueError("username must not be empty")
    if not password:
        raise ValueError("password must not be empty")
    if app_user_exists(uname):
        raise ValueError(f"user {uname!r} already exists")
    pw_hash = hash_password(password)
    display = (display_name or "").strip() or None
    try:
        with connect() as conn:
            conn.execute(
                """
                INSERT INTO app_users (username, password_hash, role, display_name, disabled)
                VALUES (?, ?, 'reader', ?, 0)
                """,
                (uname, pw_hash, display),
            )
            conn.commit()
    except _sqlite3.IntegrityError as exc:
        # Local SQLite backend: duplicate primary key raises IntegrityError.
        if app_user_exists(uname):
            raise ValueError(f"user {uname!r} already exists") from exc
        raise
    except Exception as exc:
        # Production (Turso/libsql) backend: the libsql client does NOT
        # subclass sqlite3 — a duplicate primary-key INSERT raises a bare
        # builtins.ValueError with message "UNIQUE constraint failed: …".
        # Catching sqlite3.IntegrityError alone misses this, so the raw SQL
        # error text would propagate to the router and be leaked to the client
        # as a 400 rather than the intended 409 USER_EXISTS (no SQL internals).
        # We match on the message instead of the exception class.
        msg = str(exc).lower()
        if "unique constraint" in msg or "already exists" in msg:
            if app_user_exists(uname):
                raise ValueError(f"user {uname!r} already exists") from exc
            # Concurrent-delete race: the uniqueness signal fired but the row
            # is gone by re-check time.  Still a conflict — raise a scrubbed
            # message so no SQL internals (table/column names) reach the client.
            raise ValueError("could not create the account due to a conflict") from exc
        raise
    return AppUser(
        username=uname,
        role="reader",
        display_name=display,
        disabled=False,
        created_at=None,
    )


def delete_app_user(username: str) -> None:
    """Hard-delete an account. Idempotent (missing user is a no-op)."""
    uname = _normalize_username(username)
    if not uname:
        return
    with connect() as conn:
        conn.execute("DELETE FROM app_users WHERE username = ?", (uname,))
        conn.commit()


def set_app_user_disabled(username: str, disabled: bool) -> None:
    """Enable/disable an account. A disabled account is refused at login."""
    uname = _normalize_username(username)
    if not uname:
        return
    with connect() as conn:
        conn.execute(
            "UPDATE app_users SET disabled = ? WHERE username = ?",
            (1 if disabled else 0, uname),
        )
        conn.commit()


def reset_app_user_password(username: str, password: str) -> None:
    """Replace an account's password with a fresh PBKDF2 hash."""
    uname = _normalize_username(username)
    if not uname:
        return
    if not password:
        raise ValueError("password must not be empty")
    with connect() as conn:
        conn.execute(
            "UPDATE app_users SET password_hash = ? WHERE username = ?",
            (hash_password(password), uname),
        )
        conn.commit()


def get_app_user_for_auth(username: str) -> User | None:
    """Resolve an ENABLED DB account to a :class:`User` for the login path.

    Returns ``None`` for unknown usernames AND for disabled accounts (so a
    suspended account can't log in). The returned ``User.password`` is the
    stored PBKDF2 hash — :func:`src.lib.auth.repo.authenticate` runs the
    constant-time :func:`verify_password` against it.

    The stored DB role is now honored (validated against
    :data:`_VALID_ROLES`, defaulting to ``reader`` for an unknown/garbage
    value). Safety no longer comes from forcing ``reader`` here; it comes
    from (1) bootstrap-first ordering in
    :func:`src.lib.auth.repo.authenticate` (an env/TOML admin is always
    checked FIRST, so a DB row can never shadow/impersonate it), and (2) the
    controlled promote path in the admin router, guarded by the last-admin
    lock. A tampered DB row can still write ``role='admin'`` directly in the
    table, but it cannot impersonate a bootstrap admin (username collision
    resolves to the bootstrap account) and promotion is audited.
    """
    uname = _normalize_username(username)
    if not uname:
        return None
    with connect() as conn:
        row = conn.execute(
            "SELECT username, password_hash, role, display_name, disabled "
            "FROM app_users WHERE username = ?",
            (uname,),
        ).fetchone()
    if row is None or bool(row["disabled"]):
        return None
    role = row["role"] if row["role"] in ("reader", "viewer", "admin") else "reader"
    return User(
        username=row["username"],
        password=row["password_hash"],
        role=role,
        display_name=row["display_name"],
    )


def set_app_user_role(username: str, role: str) -> None:
    """Set an existing account's role. Validates role; no-op on missing user."""
    if role not in _VALID_ROLES:
        raise ValueError(f"invalid role {role!r}")
    uname = _normalize_username(username)
    with connect() as conn:
        conn.execute("UPDATE app_users SET role = ? WHERE username = ?", (role, uname))
        conn.commit()


def get_app_user_role(username: str) -> str | None:
    """Stored role for a DB account, or None if no such row."""
    uname = _normalize_username(username)
    with connect() as conn:
        row = conn.execute("SELECT role FROM app_users WHERE username = ?", (uname,)).fetchone()
    return row["role"] if row else None


def display_names_for(usernames) -> dict[str, str]:
    """Map each username that HAS a non-empty ``app_users.display_name`` to it.

    One grouped ``IN (…)`` query for a batch of usernames (the passage-notes
    router resolves author/reply display names this way). Usernames absent from
    ``app_users`` — every bootstrap admin/viewer, whose credentials live in
    env/TOML, not this table — are simply not in the result, so the caller
    falls back to the raw username. Rows with a NULL/empty ``display_name`` are
    likewise omitted (same fallback). De-dupes and drops blanks first.
    """
    unames = sorted({u for u in usernames if u})
    if not unames:
        return {}
    ph = ",".join("?" for _ in unames)
    with connect() as conn:
        rows = conn.execute(
            f"SELECT username, display_name FROM app_users WHERE username IN ({ph})",
            tuple(unames),
        ).fetchall()
    return {r["username"]: r["display_name"] for r in rows if r["display_name"]}


def count_db_admins() -> int:
    """Enabled DB accounts with role='admin'."""
    with connect() as conn:
        row = conn.execute(
            "SELECT COUNT(*) AS n FROM app_users WHERE role='admin' AND disabled=0"
        ).fetchone()
    return int(row["n"])