"""The admin roster (spec admin-user-management). Super Admin only, all of it. This is where scope itself is assigned — a Host who could reach these endpoints could grant themselves every event in the system, which would make `services/scope.py` decorative. `require(Action.MANAGE_ADMINS)` guards the whole router. **No password value crosses this boundary except once.** `issue_temporary_password` returns one, to the super admin who asked for it, in that one response. It is not stored in clear, not returned by any read, and not written to the audit log — the log records that a temporary password was issued, by whom and for whom, and nothing else. `AdminUserRead` has no field capable of carrying a hash, which is the mechanical reason it cannot leak rather than a promise that nobody will add one. """ import logging import uuid from datetime import UTC, datetime from fastapi import APIRouter, Body, Depends, HTTPException, Request, status from pydantic import BaseModel, EmailStr, Field from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import get_settings from app.db import get_session from app.models import AdminUser from app.models.enums import AdminRole, AdminStatus, AuthMethod from app.models.wedding import Event from app.services import audit, auth, passwords from app.services.auth import CurrentAdmin, require from app.services.policy import Action logger = logging.getLogger(__name__) router = APIRouter(tags=["admin-users"], dependencies=[Depends(require(Action.MANAGE_ADMINS))]) # ------------------------------------------------------------------------------ schemas class AdminUserRead(BaseModel): """What the roster screen shows. Deliberately has nowhere to put a credential.""" id: uuid.UUID email: str name: str | None username: str | None role: AdminRole status: AdminStatus auth_method: AuthMethod #: True while a super-admin-issued password is still waiting to be replaced — an #: onboarding somebody started and nobody finished. temporary_password_outstanding: bool locked: bool first_seen_at: datetime | None last_login_at: datetime | None owned_event_count: int class AdminUserCreate(BaseModel): email: EmailStr name: str | None = Field(default=None, max_length=200) role: AdminRole = AdminRole.HOST auth_method: AuthMethod = AuthMethod.GOOGLE #: Password accounts only, and required for them — an account that cannot sign in has no #: reason to exist yet (spec admin-user-management). username: str | None = Field(default=None, min_length=3, max_length=64) temporary_password: str | None = Field(default=None, max_length=256) class AdminUserUpdate(BaseModel): """`auth_method` is absent on purpose: kind is fixed at creation (design D16).""" name: str | None = Field(default=None, max_length=200) email: EmailStr | None = None username: str | None = Field(default=None, min_length=3, max_length=64) role: AdminRole | None = None class TemporaryPasswordRead(BaseModel): """The one response in this system that carries a password. Shown once, never stored.""" password: str expires_at: datetime class OwnershipTransfer(BaseModel): new_owner_id: uuid.UUID # ------------------------------------------------------------------------------ helpers def _client_ip(request: Request) -> str | None: forwarded = request.headers.get("x-forwarded-for") if forwarded: return forwarded.split(",")[0].strip() return request.client.host if request.client else None def _to_read(admin: AdminUser, owned: int) -> AdminUserRead: now = datetime.now(UTC) return AdminUserRead( id=admin.id, email=admin.email, name=admin.name, username=admin.username, role=admin.role, status=admin.status, auth_method=admin.auth_method, temporary_password_outstanding=admin.must_change_password, locked=admin.locked_until is not None and admin.locked_until > now, first_seen_at=admin.first_seen_at, last_login_at=admin.last_login_at, owned_event_count=owned, ) async def _owned_counts(session: AsyncSession) -> dict[uuid.UUID, int]: rows = await session.execute( select(Event.owner_admin_id, func.count(Event.id)).group_by(Event.owner_admin_id) ) return dict(rows.all()) # type: ignore[arg-type] async def _get_or_404(session: AsyncSession, admin_id: uuid.UUID) -> AdminUser: admin = await session.get(AdminUser, admin_id) if admin is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found") return admin async def _assert_username_free( session: AsyncSession, username: str, *, exclude: uuid.UUID | None = None ) -> None: """Belt to the database's braces. The functional unique index is what actually holds.""" stmt = select(AdminUser.id).where(func.lower(AdminUser.username) == username.lower()) if exclude is not None: stmt = stmt.where(AdminUser.id != exclude) if await session.scalar(stmt) is not None: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="That username is already taken." ) async def _assert_not_last_super_admin(session: AsyncSession, target: AdminUser) -> None: """Refuse anything that would leave zero active super admins. `with_for_update` on the count is what makes this hold under concurrency: two remaining super admins demoting each other in overlapping requests would each read "2 active" and each proceed, and the roster would end with none. The lock serialises them, so the second one reads 1 and is refused (spec admin-user-management). """ if target.role is not AdminRole.SUPER_ADMIN or target.status is not AdminStatus.ACTIVE: return others = await session.scalars( select(AdminUser.id) .where( AdminUser.role == AdminRole.SUPER_ADMIN, AdminUser.status == AdminStatus.ACTIVE, AdminUser.id != target.id, ) .with_for_update() ) if others.first() is None: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=( "This is the last active Super Admin. Promote another account first, " "or the system would be left with nobody who can manage it." ), ) def _set_temporary_password(admin: AdminUser, raw: str) -> datetime: """Apply a temporary password, with every consequence it drags along.""" settings = get_settings() now = datetime.now(UTC) expires = passwords.temporary_expiry(ttl_hours=settings.temp_password_ttl_hours) admin.password_hash = passwords.hash_password(raw) admin.password_set_at = now admin.password_expires_at = expires admin.must_change_password = True # Whoever held a session on this account no longer should — the credential changed # underneath them, and the usual reason for issuing one is that it was compromised. auth.revoke_sessions(admin) admin.failed_login_count = 0 admin.locked_until = None return expires # -------------------------------------------------------------------------------- routes @router.get("/admin/users", response_model=list[AdminUserRead]) async def list_admin_users( session: AsyncSession = Depends(get_session), ) -> list[AdminUserRead]: admins = (await session.scalars(select(AdminUser).order_by(AdminUser.created_at))).all() counts = await _owned_counts(session) return [_to_read(a, counts.get(a.id, 0)) for a in admins] @router.post("/admin/users", response_model=AdminUserRead, status_code=status.HTTP_201_CREATED) async def create_admin_user( payload: AdminUserCreate, request: Request, actor: CurrentAdmin = Depends(require(Action.MANAGE_ADMINS)), session: AsyncSession = Depends(get_session), ) -> AdminUserRead: settings = get_settings() email = str(payload.email).lower() if await session.scalar(select(AdminUser.id).where(func.lower(AdminUser.email) == email)): raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="An account with that email exists." ) admin = AdminUser( email=email, name=payload.name, role=payload.role, # Created by a human who decided, so it is usable immediately — the pending queue is # for accounts that arrived by themselves. status=AdminStatus.ACTIVE, auth_method=payload.auth_method, first_seen_at=datetime.now(UTC), ) if payload.auth_method is AuthMethod.PASSWORD: if not payload.username: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="A password account needs a username.", ) if not payload.temporary_password: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="A password account needs a temporary password to sign in with.", ) await _assert_username_free(session, payload.username) try: validated = passwords.validate_quality( payload.temporary_password, min_length=settings.password_min_length ) except passwords.PasswordRejectedError as exc: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc admin.username = payload.username _set_temporary_password(admin, validated) else: if payload.temporary_password or payload.username: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="A Google account has no username or password to set.", ) session.add(admin) await session.flush() audit.record( session, action=audit.Actions.ADMIN_CREATE, admin_user_id=actor.id, actor_email=actor.email, entity_type="admin_user", entity_id=admin.id, after={ "email": admin.email, "role": str(admin.role), "auth_method": str(admin.auth_method), }, ip=_client_ip(request), ) return _to_read(admin, 0) @router.patch("/admin/users/{admin_id}", response_model=AdminUserRead) async def update_admin_user( admin_id: uuid.UUID, payload: AdminUserUpdate, request: Request, actor: CurrentAdmin = Depends(require(Action.MANAGE_ADMINS)), session: AsyncSession = Depends(get_session), ) -> AdminUserRead: admin = await _get_or_404(session, admin_id) before = {"role": str(admin.role), "email": admin.email, "username": admin.username} if payload.role is not None and payload.role is not admin.role: # Demoting the last super admin is the same loss as withdrawing them. if admin.role is AdminRole.SUPER_ADMIN: await _assert_not_last_super_admin(session, admin) admin.role = payload.role if payload.username is not None: if admin.auth_method is not AuthMethod.PASSWORD: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="A Google account does not sign in with a username.", ) await _assert_username_free(session, payload.username, exclude=admin.id) admin.username = payload.username if payload.email is not None: admin.email = str(payload.email).lower() if payload.name is not None: admin.name = payload.name audit.record( session, action=audit.Actions.ADMIN_UPDATE, admin_user_id=actor.id, actor_email=actor.email, entity_type="admin_user", entity_id=admin.id, before=before, after={"role": str(admin.role), "email": admin.email, "username": admin.username}, ip=_client_ip(request), ) counts = await _owned_counts(session) return _to_read(admin, counts.get(admin.id, 0)) @router.post("/admin/users/{admin_id}/activate", response_model=AdminUserRead) async def activate_admin_user( admin_id: uuid.UUID, request: Request, role: AdminRole = Body(embed=True, default=AdminRole.HOST), actor: CurrentAdmin = Depends(require(Action.MANAGE_ADMINS)), session: AsyncSession = Depends(get_session), ) -> AdminUserRead: """Approve a pending account, choosing its role in the same action.""" admin = await _get_or_404(session, admin_id) before = {"status": str(admin.status), "role": str(admin.role)} admin.status = AdminStatus.ACTIVE admin.role = role audit.record( session, action=audit.Actions.ADMIN_ACTIVATE, admin_user_id=actor.id, actor_email=actor.email, entity_type="admin_user", entity_id=admin.id, before=before, after={"status": str(admin.status), "role": str(admin.role)}, ip=_client_ip(request), ) counts = await _owned_counts(session) return _to_read(admin, counts.get(admin.id, 0)) @router.post("/admin/users/{admin_id}/reject", response_model=AdminUserRead) async def reject_admin_user( admin_id: uuid.UUID, request: Request, actor: CurrentAdmin = Depends(require(Action.MANAGE_ADMINS)), session: AsyncSession = Depends(get_session), ) -> AdminUserRead: """Turn a pending account away. Withdrawn rather than deleted, deliberately. A deleted row would be recreated as a fresh pending account by the next sign-in from that address, and the super admin who already said no would have to say it again every time. """ admin = await _get_or_404(session, admin_id) if admin.status is not AdminStatus.PENDING: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="That account is not pending." ) admin.status = AdminStatus.WITHDRAWN audit.record( session, action=audit.Actions.ADMIN_REJECT, admin_user_id=actor.id, actor_email=actor.email, entity_type="admin_user", entity_id=admin.id, before={"status": str(AdminStatus.PENDING)}, after={"status": str(admin.status)}, ip=_client_ip(request), ) return _to_read(admin, 0) @router.post("/admin/users/{admin_id}/withdraw", response_model=AdminUserRead) async def withdraw_admin_user( admin_id: uuid.UUID, request: Request, actor: CurrentAdmin = Depends(require(Action.MANAGE_ADMINS)), session: AsyncSession = Depends(get_session), ) -> AdminUserRead: """End an active account's access, immediately — not at their next sign-in.""" admin = await _get_or_404(session, admin_id) await _assert_not_last_super_admin(session, admin) admin.status = AdminStatus.WITHDRAWN # `get_current_admin` already refuses a non-active row, so this is belt and braces; it # also means a later reactivation does not silently revive their old open sessions. auth.revoke_sessions(admin) audit.record( session, action=audit.Actions.ADMIN_DEACTIVATE, admin_user_id=actor.id, actor_email=actor.email, entity_type="admin_user", entity_id=admin.id, after={"status": str(admin.status)}, ip=_client_ip(request), ) counts = await _owned_counts(session) return _to_read(admin, counts.get(admin.id, 0)) @router.post("/admin/users/{admin_id}/temporary-password", response_model=TemporaryPasswordRead) async def issue_temporary_password( admin_id: uuid.UUID, request: Request, actor: CurrentAdmin = Depends(require(Action.MANAGE_ADMINS)), session: AsyncSession = Depends(get_session), ) -> TemporaryPasswordRead: """Issue a fresh temporary password. The only recovery path for a password account. There is no self-service forgotten-password flow: the only outbound mail this system has is guest-facing, with suppression lists and quiet hours, and routing admin reset links through it would put them behind a guest's unsubscribe state (design D14). The value is returned here and nowhere else, ever. """ admin = await _get_or_404(session, admin_id) if admin.auth_method is not AuthMethod.PASSWORD: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="That account signs in with Google and has no password.", ) raw = passwords.generate_temporary() expires = _set_temporary_password(admin, raw) audit.record( session, action=audit.Actions.ADMIN_TEMP_PASSWORD, admin_user_id=actor.id, actor_email=actor.email, entity_type="admin_user", entity_id=admin.id, # The fact, the parties and the expiry. Never the value — see the module docstring. after={"expires_at": expires.isoformat()}, ip=_client_ip(request), ) logger.info("temporary password issued for an admin account") return TemporaryPasswordRead(password=raw, expires_at=expires) @router.post("/admin/events/{event_id}/transfer", response_model=AdminUserRead) async def transfer_event_ownership( event_id: uuid.UUID, payload: OwnershipTransfer, request: Request, actor: CurrentAdmin = Depends(require(Action.MANAGE_ADMINS)), session: AsyncSession = Depends(get_session), ) -> AdminUserRead: """Move an event, and everything under it, to another active account. Guests, invitations, the card design and the message history all follow without being touched, because every one of them is reachable through the event rather than owned separately (design D1). """ event = await session.get(Event, event_id) if event is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found") new_owner = await _get_or_404(session, payload.new_owner_id) if new_owner.status is not AdminStatus.ACTIVE: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="An event can only be owned by an active account.", ) previous = event.owner_admin_id event.owner_admin_id = new_owner.id audit.record( session, action=audit.Actions.EVENT_TRANSFER, admin_user_id=actor.id, actor_email=actor.email, entity_type="event", entity_id=event.id, before={"owner_admin_id": str(previous)}, after={"owner_admin_id": str(new_owner.id)}, ip=_client_ip(request), ) counts = await _owned_counts(session) return _to_read(new_owner, counts.get(new_owner.id, 0)) @router.delete("/admin/users/{admin_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_admin_user( admin_id: uuid.UUID, request: Request, actor: CurrentAdmin = Depends(require(Action.MANAGE_ADMINS)), session: AsyncSession = Depends(get_session), ) -> None: """Remove an account outright. Refused while it still owns anything. The FK is `ON DELETE RESTRICT`, so the database would refuse this anyway — but it would refuse with an integrity error, and the super admin deserves to be told how many events they need to transfer first. """ admin = await _get_or_404(session, admin_id) await _assert_not_last_super_admin(session, admin) owned = await session.scalar( select(func.count(Event.id)).where(Event.owner_admin_id == admin.id) ) if owned: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=( f"That account still owns {owned} event(s). Transfer them to another account first." ), ) audit.record( session, action=audit.Actions.ADMIN_DEACTIVATE, admin_user_id=actor.id, actor_email=actor.email, entity_type="admin_user", entity_id=admin.id, before={"email": admin.email, "role": str(admin.role)}, after={"deleted": True}, ip=_client_ip(request), ) await session.delete(admin)