bike-app/apps/api/velodrome/cli.py

"""Operator CLI — `velodrome <command>`.

Installed as a console script (`[project.scripts]` in pyproject.toml) into the same venv as
`alembic` and `uvicorn`, which `deploy/entrypoint.sh` already invokes by their installed names, so
the deployed container has this on PATH with no extra wiring:

    docker exec -it velodrome velodrome create-admin --email you@example.com

It exists because there is otherwise **no way to create the first user**. Registration requires a
valid invite (docs/PLAN.md "Auth": open signup does not exist as a setting), invites are created by
an existing admin, and a fresh database has neither — so a new deployment is unusable without a
path in from outside the HTTP API. docs/PLAN.md always called for this command; it was simply
never built during Phase 0. See docs/DECISIONS.md D18 for the three decisions recorded here: why
it refuses to touch an existing account, why it is *not* restricted to the very first user, and
why `role="admin"` is recorded but not yet enforced anywhere.

argparse rather than Typer/Click: one command with three options does not justify a runtime
dependency the deployed image would have to carry, and the stdlib covers this case completely.

There is deliberately **no `--password` flag** — see `_read_password`.
"""

import argparse
import asyncio
import getpass
import sys
from collections.abc import Sequence

from pydantic import BaseModel, EmailStr, ValidationError

from velodrome.auth import service
from velodrome.config import get_settings
from velodrome.schemas.auth import DisplayName, Password


class CliError(Exception):
    """An operator-facing failure: printed as `error: <message>`, exit code 1.

    Distinct from argparse's own usage errors, which exit 2 — so a script driving this can tell
    "you called it wrong" apart from "it ran and refused".
    """


class _CreateAdminInput(BaseModel):
    """The same constraints the HTTP register endpoint applies, reused rather than restated — a
    CLI-created account must not be able to hold a password the API would have rejected.
    """

    email: EmailStr
    password: Password
    display_name: DisplayName


def _validate(*, email: str, password: str, display_name: str) -> _CreateAdminInput:
    try:
        return _CreateAdminInput(email=email, password=password, display_name=display_name)
    except ValidationError as exc:
        # Deliberately not `str(exc)`: pydantic's rendered message embeds the offending value
        # ("... [type=string_too_short, input_value='hunter2', input_type=str]"), which for the
        # password field prints the password to the operator's terminal and into whatever
        # captures that output. CLAUDE.md invariant #5 — only `loc` and `msg` are safe to show.
        details = "; ".join(
            f"{'.'.join(str(part) for part in err['loc'])}: {err['msg']}"
            for err in exc.errors(include_url=False, include_input=False)
        )
        raise CliError(f"invalid input — {details}") from exc


def _read_password(*, from_stdin: bool) -> str:
    """Prompt for a password, or read one line from stdin.

    No `--password` flag exists on purpose: an argument lands in the operator's shell history, in
    `ps` output for as long as the process runs, and — because the realistic invocation here is
    `docker exec` — in the Docker daemon's own record of the exec'd command. A TTY prompt and a
    pipe are the two forms that avoid all three, and they're the same two forms `docker login`
    offers for exactly this reason.
    """
    if from_stdin:
        line = sys.stdin.readline()
        # Strip only the line ending, not surrounding whitespace — a trailing space in a password
        # is legitimate, and silently trimming it would create a password that can never be typed
        # back in correctly.
        password = line.rstrip("\r\n")
        if not password:
            raise CliError("--password-stdin was given but the first line of stdin was empty")
        return password

    if not sys.stdin.isatty():
        raise CliError(
            "no terminal available to prompt on. Either allocate one (note the -t):\n"
            "  docker exec -it velodrome velodrome create-admin --email you@example.com\n"
            "or pipe the password in:\n"
            "  printf '%s' \"$PASSWORD\" | docker exec -i velodrome \\\n"
            "    velodrome create-admin --email you@example.com --password-stdin"
        )

    password = getpass.getpass("Password: ")
    if password != getpass.getpass("Confirm password: "):
        raise CliError("passwords did not match")
    return password


async def _create_admin(args: argparse.Namespace) -> int:
    email: str = args.email
    # The local part is a reasonable default for a name nobody but the operator will see until
    # they change it in the UI; it keeps the common invocation to a single flag.
    display_name: str = args.name if args.name is not None else email.partition("@")[0]

    password = _read_password(from_stdin=args.password_stdin)
    validated = _validate(email=email, password=password, display_name=display_name)

    try:
        created = await service.create_admin(
            email=str(validated.email),
            password=validated.password,
            display_name=validated.display_name,
        )
    except service.EmailAlreadyRegistered as exc:
        raise CliError(
            f"an account already exists for {email} — refusing to modify it. This command only "
            "ever creates a new account; it will not reset an existing one's password (see "
            "docs/DECISIONS.md D18). To add a different admin, re-run with another --email."
        ) from exc

    print("Created admin user:")
    print(f"  id            {created.user_id}")
    print(f"  email         {created.email}")
    print(f"  display name  {created.display_name}")
    print(f"  role          {created.role}")
    print()
    print(f"Log in at {get_settings().public_url}")
    print(
        "Note: the admin role is recorded on the account but nothing enforces it yet — no "
        "admin-only endpoint exists (docs/DECISIONS.md D18)."
    )
    return 0


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="velodrome",
        description="Velodrome operator commands. Run inside the container, e.g. "
        "`docker exec -it velodrome velodrome create-admin --email you@example.com`.",
    )
    subcommands = parser.add_subparsers(dest="command", required=True)

    create_admin = subcommands.add_parser(
        "create-admin",
        help="create a user with the admin role, bypassing the invite requirement",
        description="Create a user with the admin role, bypassing the invite requirement. This is "
        "how the first account on a fresh deployment is made — registration needs an invite, and "
        "a fresh database has none. Refuses to modify an account that already exists.",
    )
    create_admin.add_argument("--email", required=True, help="the account's email address")
    create_admin.add_argument(
        "--name",
        default=None,
        help="display name (default: the part of the email address before the @)",
    )
    create_admin.add_argument(
        "--password-stdin",
        action="store_true",
        help="read the password from the first line of stdin instead of prompting for it",
    )
    return parser


async def run(argv: Sequence[str] | None = None) -> int:
    """The async entrypoint. `main` wraps this in `asyncio.run`; tests call it directly."""
    args = _build_parser().parse_args(argv)
    try:
        if args.command == "create-admin":
            return await _create_admin(args)
    except CliError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1
    raise AssertionError(f"unhandled command {args.command!r}")  # pragma: no cover


def main(argv: Sequence[str] | None = None) -> int:
    return asyncio.run(run(argv))