199 lines
8.3 KiB
Python
199 lines
8.3 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""First-run setup: creating the first platform administrator.
|
|
|
|
A deployment cannot be configured by signing in, because signing in needs an
|
|
administrator, and the operator command that used to fill that gap refuses to run
|
|
in production. This module closes the loop without loosening either end.
|
|
|
|
The gate has two independent conditions, and both must hold for setup to be
|
|
offered:
|
|
|
|
* **No platform administrator exists.** Checked against the store on every
|
|
request, not cached at boot, so the window closes the moment the first
|
|
administrator is created even if several processes are running.
|
|
* **A token generated at boot is presented.** Without it, setup would be
|
|
first-requester-wins: anyone who reached the URL before the operator would
|
|
become the administrator. The token is written to the process journal, so
|
|
control of the server still implies control of the first administrator, while
|
|
a stray request from the internet does not.
|
|
|
|
The token is never returned by an endpoint, never written to the store, and never
|
|
persisted to disk by this module. It exists in memory for the life of the
|
|
process; restarting regenerates it, which is the intended behaviour, since the
|
|
journal is where an operator looks for it.
|
|
|
|
Why the endpoint is exempt from CSRF: there is no ambient authority to ride on.
|
|
No cookie identifies anybody at this point, so a cross-site request carries
|
|
nothing a legitimate caller would not also have to send. The token is the
|
|
authenticator, and it is the only thing standing between an anonymous request and
|
|
a platform administrator.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import secrets
|
|
import threading
|
|
|
|
from .errors import ConflictError, InputError
|
|
from .models import RoleName
|
|
|
|
LOGGER = logging.getLogger("umbratrans_web.setup")
|
|
|
|
#: Generated per process when setup is still outstanding. ``secrets`` rather than
|
|
#: ``random``: this value is the whole authorisation for creating a platform
|
|
#: administrator.
|
|
TOKEN_BYTES = 32
|
|
|
|
#: Attempts before the token is burned and setup needs a restart. A token that
|
|
#: can be guessed slowly is not a token, but the point is that an online guessing
|
|
#: attempt should not be unlimited either.
|
|
MAX_TOKEN_ATTEMPTS = 5
|
|
|
|
|
|
def announce_setup_token(token: str) -> None:
|
|
"""Write the first-run setup token to the journal.
|
|
|
|
The one place in this codebase that emits a secret to the log deliberately.
|
|
|
|
The log filter rewrites ``name_token=value`` because that shape is what a
|
|
verification or session token looks like in a record, and those must never be
|
|
written down. This value is different in kind: it is not a stored credential,
|
|
it authorises exactly one action, it is useless the moment that action is
|
|
taken, and there is no other channel to deliver it on. An operator who cannot
|
|
read it cannot complete setup, and the alternative is a shell command that
|
|
creates the first administrator, which is the thing this module exists to
|
|
avoid.
|
|
|
|
It is emitted on its own line with no ``key=value`` assignment, so the filter
|
|
is not weakened, disabled, or special-cased to let it through: the shape it
|
|
looks for simply is not used here.
|
|
"""
|
|
|
|
LOGGER.warning(
|
|
"FIRST-RUN SETUP IS REQUIRED: no platform administrator exists on this deployment.\n"
|
|
"Open the setup page to create one. The single-use code below is shown here and "
|
|
"nowhere else, is not stored, and is regenerated on restart.\n"
|
|
" setup code: %s\n"
|
|
"After it is used the deployment refuses further setup attempts.",
|
|
token,
|
|
)
|
|
|
|
|
|
class SetupGate:
|
|
"""Tracks whether first-run setup is outstanding, and authorises one attempt.
|
|
|
|
One instance per process. The attempts counter is deliberately per process
|
|
rather than shared, because nothing durable is written: a restart resets it,
|
|
and a restart also regenerates the token, so the pair cannot be used to get
|
|
more attempts at the same secret.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Lock()
|
|
self._token: str = ""
|
|
self._attempts = 0
|
|
self._spent = False
|
|
|
|
# -- lifecycle -----------------------------------------------------------
|
|
|
|
def begin_if_needed(self, *, platform_administrators: int) -> str | None:
|
|
"""Mint a token when setup is outstanding, and announce it to the journal.
|
|
|
|
Returns ``None`` when an administrator already exists, so a configured
|
|
deployment prints nothing and holds no secret.
|
|
|
|
The token has to reach the operator, and the log filter will not carry it
|
|
for them: :func:`umbratrans_web.security.redact_text` rewrites anything
|
|
shaped like ``<name>token<name> = value``, which is exactly the right
|
|
behaviour for a verification or session token and exactly wrong here.
|
|
Rather than weaken the filter, the announcement uses
|
|
:func:`announce_setup_token`, which is the single documented place in
|
|
the codebase that emits a secret to the log on purpose. The filter itself
|
|
is untouched, and a test pins both halves of that: this value gets
|
|
through, and ``token=`` assignments still do not.
|
|
"""
|
|
|
|
with self._lock:
|
|
if self._token or self._spent:
|
|
return None
|
|
if platform_administrators > 0:
|
|
self._spent = True
|
|
self._token = ""
|
|
return None
|
|
self._token = secrets.token_urlsafe(TOKEN_BYTES)
|
|
announce_setup_token(self._token)
|
|
return self._token
|
|
|
|
# -- state ---------------------------------------------------------------
|
|
|
|
@property
|
|
def is_outstanding(self) -> bool:
|
|
with self._lock:
|
|
return bool(self._token) and not self._spent
|
|
|
|
@property
|
|
def attempts_remaining(self) -> int:
|
|
with self._lock:
|
|
if not self._token:
|
|
return 0
|
|
return max(MAX_TOKEN_ATTEMPTS - self._attempts, 0)
|
|
|
|
def close(self) -> None:
|
|
"""Mark setup finished and drop the token from memory."""
|
|
|
|
with self._lock:
|
|
self._token = ""
|
|
self._spent = True
|
|
|
|
# -- authorisation -------------------------------------------------------
|
|
|
|
def consume(self, presented: str) -> None:
|
|
"""Validate a presented token, or raise.
|
|
|
|
Comparison is constant-time. A length or content difference must not be
|
|
observable in timing, because the whole purpose of the token is to be
|
|
unguessable and a timing oracle reintroduces guessing.
|
|
"""
|
|
|
|
with self._lock:
|
|
if not self._token or self._spent:
|
|
raise ConflictError(
|
|
"setup_not_required",
|
|
"First-run setup is not available on this deployment.",
|
|
)
|
|
if self._attempts >= MAX_TOKEN_ATTEMPTS:
|
|
self._token = ""
|
|
self._spent = True
|
|
raise ConflictError(
|
|
"setup_locked",
|
|
"Too many failed setup attempts. Restart the service to generate a new token.",
|
|
)
|
|
if not secrets.compare_digest(presented or "", self._token):
|
|
self._attempts += 1
|
|
remaining = max(MAX_TOKEN_ATTEMPTS - self._attempts, 0)
|
|
if remaining == 0:
|
|
self._token = ""
|
|
self._spent = True
|
|
raise ConflictError(
|
|
"setup_locked",
|
|
"Too many failed setup attempts. Restart the service to generate a new token.",
|
|
)
|
|
raise InputError(
|
|
"setup_token_invalid",
|
|
f"That setup token is not correct. {remaining} attempt(s) remain before this "
|
|
"token is discarded and the service must be restarted.",
|
|
)
|
|
# The token is spent the moment it authorises, whether or not the
|
|
# account creation that follows succeeds. Otherwise a failure part
|
|
# way through, such as a rejected password, would leave a token that
|
|
# has already been handed out still usable.
|
|
self._token = ""
|
|
self._spent = True
|
|
LOGGER.warning("first-run setup token accepted; no further setup attempts are possible")
|
|
|
|
|
|
__all__ = ["MAX_TOKEN_ATTEMPTS", "SetupGate", "RoleName"]
|