"""``/admin/users/*`` — admin management of DB-backed ``reader`` accounts. All endpoints are ``Depends(require_admin)``. They manage the ``app_users`` store (see :mod:`src.lib.auth.user_store`) — the preview / Reading-Room-only accounts an admin creates from the UI. The admins themselves keep coming from env/TOML (the bootstrap layer), so nothing here can lock the operator out and nothing here can mint another admin (the create path forces role ``reader``). Endpoints: - ``GET /admin/users`` → list accounts (no password hashes). - ``POST /admin/users`` → create a reader account; 409 on dup. - ``DELETE /admin/users/{username}`` → hard-delete (idempotent → 204). - ``PATCH /admin/users/{username}`` → enable/disable and/or reset password. - ``PATCH /admin/users/{username}/role`` → promote/demote (last-admin guarded, audited). Error conventions match the rest of the API: typed errors from :mod:`src.api.errors` serialize to the canonical ``{code, message, details}`` envelope via the global handler in ``src/api/app.py``. """ from __future__ import annotations from fastapi import APIRouter, Depends, Path, Response, status from pydantic import Field from src.api.deps import require_admin from src.api.dto.admin_users import ( AppUserDTO, AppUserListResponse, CreateAppUserRequest, UpdateAppUserRequest, ) from src.api.dto.common import ApiModel from src.api.errors import ApiError, BadRequest, UserExists, UserNotFound from src.lib.auth.models import User from src.lib.auth.repo import get_credentials from src.lib.auth.user_store import ( AppUser, app_user_exists, create_app_user, delete_app_user, get_app_user_role, list_app_users, reset_app_user_password, set_app_user_disabled, set_app_user_role, ) from src.lib.auth.book_access_repo import ( grant as grant_book_access, list_book_ids_for_user, revoke as revoke_book_access, ) from src.stage1_inventory.db import connect router = APIRouter(prefix="/admin/users", tags=["admin-users"]) # Minimum password length for created/reset reader accounts. Low-friction but # enough to keep a one-character preview password out of the DB. _MIN_PASSWORD_LEN = 8 class _LastAdminLocked(ApiError): """Refusing to demote/disable/delete the last remaining admin (409). Raised when an action would remove the final admin account. Env/TOML bootstrap admins remain the recovery floor — the lock only blocks losing the *last* admin overall (DB + bootstrap), not the last DB admin. """ code = "LAST_ADMIN_LOCKED" status_code = 409 class UpdateRoleRequest(ApiModel): """``PATCH /admin/users/{username}/role`` body. ``role`` is validated to the three valid roles; the router additionally enforces the last-admin lock before applying it. """ role: str = Field(pattern=r"^(reader|viewer|admin)$") def _total_admins() -> int: from src.lib.auth.user_store import count_db_admins from src.lib.auth.repo import count_bootstrap_admins return count_db_admins() + count_bootstrap_admins() def _to_dto(u: AppUser) -> AppUserDTO: return AppUserDTO( username=u.username, role=u.role, display_name=u.display_name, disabled=u.disabled, created_at=u.created_at, ) def _is_bootstrap_username(username: str) -> bool: """True if ``username`` is an env/TOML (admin/viewer bootstrap) account. ``authenticate()`` checks the bootstrap source FIRST, so a DB row with the same id would be unreachable AND would let an admin think they'd created a working reader when logins actually resolve to the bootstrap account. We refuse the create up-front. Fail closed: if the credential source can't be read, treat the name as taken. """ try: return username in get_credentials() except Exception: # noqa: BLE001 — unreadable creds → refuse to be safe return True def _validate_password(password: str) -> None: if not password or len(password) < _MIN_PASSWORD_LEN: raise BadRequest( f"Password must be at least {_MIN_PASSWORD_LEN} characters." ) @router.get("", response_model=AppUserListResponse) def list_users(_: User = Depends(require_admin)) -> AppUserListResponse: """List every DB-backed account. Never returns password hashes.""" return AppUserListResponse(items=[_to_dto(u) for u in list_app_users()]) @router.post("", response_model=AppUserDTO, status_code=status.HTTP_201_CREATED) def create_user( req: CreateAppUserRequest, _: User = Depends(require_admin), ) -> AppUserDTO: """Create a ``reader`` account. 409 ``USER_EXISTS`` on a duplicate. The role is forced to ``reader`` (this body has no role field). The username is lowercased to match the login normalization. Validates a non-empty username and a sane minimum password length; rejects a username that collides with a bootstrap (env/TOML) account. """ username = (req.username or "").strip().lower() if not username: raise BadRequest("Username must not be empty.") _validate_password(req.password) if _is_bootstrap_username(username) or app_user_exists(username): raise UserExists(username=username) try: # role defaults to 'reader' inside create_app_user — never overridden. created = create_app_user( username=username, password=req.password, display_name=req.display_name, ) except ValueError as e: # Lost a create race (duplicate) or an empty-field edge → 409 / 400. if "already exists" in str(e): raise UserExists(username=username) raise BadRequest(str(e)) return _to_dto(created) @router.delete("/{username}", status_code=status.HTTP_204_NO_CONTENT) def delete_user( username: str = Path(...), _: User = Depends(require_admin), ) -> Response: """Hard-delete a DB account. Idempotent — a missing user still returns 204. Last-admin lock: deleting the last admin (DB + bootstrap) is refused with 409 ``LAST_ADMIN_LOCKED`` so an admin can never lock the operator out. """ uname = (username or "").strip().lower() if get_app_user_role(uname) == "admin" and _total_admins() <= 1: raise _LastAdminLocked( f"Cannot delete {uname!r}: it is the last admin account.", details={"username": uname}, ) delete_app_user(username) return Response(status_code=status.HTTP_204_NO_CONTENT) @router.patch("/{username}/role", response_model=AppUserDTO) def update_user_role( req: UpdateRoleRequest, username: str = Path(...), admin: User = Depends(require_admin), ) -> AppUserDTO: """Promote/demote a DB account. Last-admin guarded; audited. 404 ``USER_NOT_FOUND`` if the target isn't a DB account (bootstrap accounts live in server secrets and aren't editable here). Demoting the last admin is refused with 409 ``LAST_ADMIN_LOCKED``. Every change writes a ``role_change_audit`` row (actor = the acting admin). """ uname = (username or "").strip().lower() old_role = get_app_user_role(uname) if old_role is None: # Not a DB account — bootstrap (env/TOML) accounts are not editable here. raise UserNotFound(username=uname) new_role = req.role if old_role == "admin" and new_role != "admin" and _total_admins() <= 1: raise _LastAdminLocked( f"Cannot demote {uname!r}: it is the last admin account.", details={"username": uname, "oldRole": old_role, "newRole": new_role}, ) set_app_user_role(uname, new_role) action = "promote" if new_role == "admin" else "demote" with connect() as conn: conn.execute( """ INSERT INTO role_change_audit (actor, target, old_role, new_role, action) VALUES (?, ?, ?, ?, ?) """, (admin.username, uname, old_role, new_role, action), ) conn.commit() # Re-read to return the authoritative current row (no hash). for u in list_app_users(): if u.username == uname: return _to_dto(u) # Race: deleted between the role set and the re-read. raise UserNotFound(username=uname) @router.patch("/{username}", response_model=AppUserDTO) def update_user( req: UpdateAppUserRequest, username: str = Path(...), _: User = Depends(require_admin), ) -> AppUserDTO: """Enable/disable and/or reset the password of a DB account. 404 ``USER_NOT_FOUND`` when the account doesn't exist. Both body fields are optional; sending neither returns the current row unchanged. A password reset is validated for minimum length and re-hashed server-side. Last-admin lock: disabling the last admin is refused with 409 ``LAST_ADMIN_LOCKED`` (deleting/disabling the last admin would lock the operator out; env bootstrap admin remains the recovery floor). """ uname = (username or "").strip().lower() if not app_user_exists(uname): raise UserNotFound(username=uname) if req.password is not None: _validate_password(req.password) reset_app_user_password(uname, req.password) if req.disabled is not None: if req.disabled and get_app_user_role(uname) == "admin" and _total_admins() <= 1: raise _LastAdminLocked( f"Cannot disable {uname!r}: it is the last admin account.", details={"username": uname}, ) set_app_user_disabled(uname, req.disabled) # Re-read to return the authoritative current row (no hash). for u in list_app_users(): if u.username == uname: return _to_dto(u) # Race: deleted between the existence check and the re-read. raise UserNotFound(username=uname) # ---------- Per-account book grants (D2) ---------------------------------- # A reader sees public books ∪ books explicitly granted to them. Grants only # ADD access; they never restrict a public book. These endpoints let an admin # manage the explicit grants. Response shapes are plain dicts (no new DTO is # added) to stay within the D2 file-scope; the file's existing typed-DTO # convention is kept for the account endpoints above. @router.get("/{username}/book-access") def list_user_book_access( username: str = Path(...), _: User = Depends(require_admin), ) -> dict: """List the ``book_id``s explicitly granted to a reader account.""" uname = (username or "").strip().lower() return {"username": uname, "book_ids": list_book_ids_for_user(uname)} @router.put("/{username}/book-access/{book_id}", status_code=status.HTTP_204_NO_CONTENT) def grant_user_book_access( username: str = Path(...), book_id: str = Path(...), admin: User = Depends(require_admin), ) -> Response: """Grant one book to a reader. Idempotent upsert. ``granted_by`` is the acting admin's username (audited on the row). Returns 204 on success.""" uname = (username or "").strip().lower() grant_book_access(uname, book_id, granted_by=admin.username) return Response(status_code=status.HTTP_204_NO_CONTENT) @router.delete( "/{username}/book-access/{book_id}", status_code=status.HTTP_204_NO_CONTENT ) def revoke_user_book_access( username: str = Path(...), book_id: str = Path(...), _: User = Depends(require_admin), ) -> Response: """Revoke one book grant from a reader. Idempotent (missing grant is a no-op) and still returns 204.""" uname = (username or "").strip().lower() revoke_book_access(uname, book_id) return Response(status_code=status.HTTP_204_NO_CONTENT)