Files
UmbraTrans/umbratrans_web/routes_setup.py
2026-10-18 19:01:05 -07:00

143 lines
4.7 KiB
Python

# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0
"""First-run setup routes.
These are the only unauthenticated endpoints that can create an account, and they
exist only while no platform administrator exists and only for the holder of the
token printed in the journal. See :mod:`umbratrans_web.setup` for why the token
is required and why these paths are exempt from CSRF.
There is no route here that returns the token, and none that reports whether a
particular guess is closer to correct.
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Request
from pydantic import BaseModel, Field
from .errors import ConflictError
from .service import AccountService
LOGGER = logging.getLogger("umbratrans_web.setup")
router = APIRouter()
class SetupStatusResponse(BaseModel):
"""Whether first-run setup is outstanding.
Reports only whether setup is needed. It never carries the token, a count of
accounts, or anything else that would help someone decide whether to keep
trying.
"""
setup_required: bool
class InitialAdministratorRequest(BaseModel):
"""Create the first platform administrator.
The password arrives from the caller, unlike every other enrollment path,
where the service generates the one-time password. That inversion is
deliberate and confined to this one endpoint: there is no administrator yet,
so there is nobody to hand a generated password to, and a browser form has to
be able to set a password the account owner will actually type.
"""
token: str = Field(min_length=16, max_length=256)
email: str = Field(min_length=3, max_length=254)
display_name: str = Field(min_length=2, max_length=120)
password: str = Field(min_length=1, max_length=1024)
username: str | None = Field(default=None, max_length=64)
class InitialAdministratorResponse(BaseModel):
"""The new administrator's identifier and email, never the password.
The password was chosen by the caller and is not returned, because returning
a credential that the caller already holds would widen the places it can leak
from for no benefit.
"""
user_id: str
email: str
display_name: str
message: str
def _service(request: Request) -> AccountService:
service: AccountService = request.app.state.service
return service
def _gate(request: Request): # type: ignore[no-untyped-def]
gate = request.app.state.setup_gate
return gate
@router.get("/status", response_model=SetupStatusResponse, tags=["setup"])
def setup_status(request: Request) -> SetupStatusResponse:
"""Report whether the deployment still needs its first administrator."""
service = _service(request)
return SetupStatusResponse(setup_required=service.setup_is_required())
@router.post(
"/initial-administrator",
response_model=InitialAdministratorResponse,
status_code=201,
tags=["setup"],
)
def create_initial_administrator(
payload: InitialAdministratorRequest,
request: Request,
) -> InitialAdministratorResponse:
"""Create the first platform administrator.
Refused once any platform administrator exists, and refused without the
journal token. Both checks happen here rather than being left to the service,
because this is the unauthenticated surface and the refusal has to be
unconditional.
"""
service = _service(request)
gate = _gate(request)
# Checked before the token so a configured deployment does not even look at
# a presented secret, and so the refusal a caller sees after setup is the
# accurate one. The same error the gate raises, so this path has one shape.
if not service.setup_is_required():
LOGGER.warning("refused first-run setup: a platform administrator already exists")
raise ConflictError("setup_not_required", "This deployment is already configured.")
# The gate raises ConflictError and InputError, the same types the rest of the
# API uses, so letting them propagate keeps one error shape across the app
# rather than inventing a second one for this path.
gate.consume(payload.token)
issued = service.create_initial_administrator(
email=payload.email,
display_name=payload.display_name,
password=payload.password,
username=payload.username,
request_id=request.headers.get("x-request-id"),
)
gate.close()
return InitialAdministratorResponse(
user_id=str(issued.account.id),
email=issued.account.email,
display_name=issued.account.display_name,
message=(
"Administrator created. Sign in with the email address and password you just chose, "
"then replace the password when prompted."
),
)
__all__ = ["router"]