feat(auth): CLI users subcommand

This commit is contained in:
Nora
2026-06-22 15:01:35 -06:00
parent 9c57b493a7
commit 6c80bf0512
3 changed files with 317 additions and 0 deletions
+183
View File
@@ -0,0 +1,183 @@
"""CLI subcommand: ``python -m cyclone users ...``.
Click-based to match the existing parse-837 / parse-835 convention in
``cyclone.cli``. Provides operator-side user management without going
through the admin HTTP API.
Subcommands
-----------
- ``users create USERNAME --role {admin,user,viewer} [--password PW]``
Create a user. If ``--password`` is omitted, prompts (with
confirmation) on the controlling terminal.
- ``users list`` — tab-separated ``id / username / role / state``.
- ``users disable USERNAME`` — set ``disabled_at`` to now.
- ``users reset-password USERNAME [--password PW]`` — replace the hash.
- ``users set-role USERNAME --role {admin,user,viewer}`` — change role.
Exit codes
----------
* 0 — success
* 1 — validation error (unknown username, duplicate username)
* 2 — usage error (missing arg, bad role, short password, unknown
subcommand). Click itself uses 2 for usage errors so the conventional
shell tools (``set -e``, etc.) recognize them.
Passwords shorter than 12 chars are rejected everywhere they appear.
"""
from __future__ import annotations
import getpass
import sys
import click
from cyclone.auth import users
from cyclone.auth.permissions import Role
from cyclone.db import SessionLocal
ROLE_CHOICES = [Role.ADMIN.value, Role.USER.value, Role.VIEWER.value]
MIN_PASSWORD_LEN = 12
def _prompt_password(label: str) -> str:
pw = getpass.getpass(f"{label}: ")
if not pw:
click.echo("Password required.", err=True)
sys.exit(2)
return pw
def _validate_password(pw: str) -> None:
if len(pw) < MIN_PASSWORD_LEN:
click.echo(
f"Password must be at least {MIN_PASSWORD_LEN} characters.",
err=True,
)
sys.exit(2)
# --------------------------------------------------------------------------- #
# Group
# --------------------------------------------------------------------------- #
@click.group(name="users")
def users_cli() -> None:
"""Manage Cyclone users from the command line."""
# --------------------------------------------------------------------------- #
# create
# --------------------------------------------------------------------------- #
@users_cli.command("create")
@click.argument("username")
@click.option(
"--role",
required=True,
type=click.Choice(ROLE_CHOICES, case_sensitive=False),
help="Role to grant the new user.",
)
@click.option(
"--password",
default=None,
help=f"Password (min {MIN_PASSWORD_LEN} chars). Prompts if omitted.",
)
def create_user(username: str, role: str, password: str | None) -> None:
"""Create a new user with the given USERNAME and ROLE."""
pw = password or _prompt_password("Password")
_validate_password(pw)
with SessionLocal()() as db:
if users.get_by_username(db, username) is not None:
click.echo(f"User '{username}' already exists.", err=True)
sys.exit(1)
u = users.create(db, username=username, password=pw, role=role)
click.echo(f"Created user '{u.username}' with role '{u.role}'.")
# --------------------------------------------------------------------------- #
# list
# --------------------------------------------------------------------------- #
@users_cli.command("list")
def list_users() -> None:
"""List all users as id / username / role / state."""
from cyclone.db import User
with SessionLocal()() as db:
for u in db.query(User).order_by(User.id.asc()).all():
state = "disabled" if u.disabled_at else "active"
click.echo(f"{u.id}\t{u.username}\t{u.role}\t{state}")
# --------------------------------------------------------------------------- #
# disable
# --------------------------------------------------------------------------- #
@users_cli.command("disable")
@click.argument("username")
def disable_user(username: str) -> None:
"""Disable USERNAME (sets disabled_at to now)."""
with SessionLocal()() as db:
u = users.get_by_username(db, username)
if u is None:
click.echo(f"No such user: {username}", err=True)
sys.exit(1)
users.disable(db, u.id)
click.echo(f"Disabled '{username}'.")
# --------------------------------------------------------------------------- #
# reset-password
# --------------------------------------------------------------------------- #
@users_cli.command("reset-password")
@click.argument("username")
@click.option(
"--password",
default=None,
help=f"New password (min {MIN_PASSWORD_LEN} chars). Prompts if omitted.",
)
def reset_password(username: str, password: str | None) -> None:
"""Replace USERNAME's password."""
pw = password or _prompt_password("New password")
_validate_password(pw)
with SessionLocal()() as db:
u = users.get_by_username(db, username)
if u is None:
click.echo(f"No such user: {username}", err=True)
sys.exit(1)
users.update_password(db, u.id, pw)
click.echo(f"Password reset for '{username}'.")
# --------------------------------------------------------------------------- #
# set-role
# --------------------------------------------------------------------------- #
@users_cli.command("set-role")
@click.argument("username")
@click.option(
"--role",
required=True,
type=click.Choice(ROLE_CHOICES, case_sensitive=False),
help="Role to grant.",
)
def set_role(username: str, role: str) -> None:
"""Change USERNAME's role."""
with SessionLocal()() as db:
u = users.get_by_username(db, username)
if u is None:
click.echo(f"No such user: {username}", err=True)
sys.exit(1)
users.update_role(db, u.id, role)
click.echo(f"Role for '{username}' set to '{role}'.")
+6
View File
@@ -74,6 +74,12 @@ def main(ctx: click.Context, log_format: str | None, log_file: Path | None) -> N
ctx.ensure_object(dict) ctx.ensure_object(dict)
# Register the auth users subgroup. Imported here (not at module top) to
# avoid pulling passlib / bcrypt at CLI parse-only import time.
from cyclone.auth.cli import users_cli # noqa: E402
main.add_command(users_cli)
@main.command("parse-837") @main.command("parse-837")
@click.argument("input_file", type=click.Path(exists=True, dir_okay=False, path_type=Path)) @click.argument("input_file", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.option("--output-dir", required=True, type=click.Path(file_okay=False, path_type=Path)) @click.option("--output-dir", required=True, type=click.Path(file_okay=False, path_type=Path))
+128
View File
@@ -0,0 +1,128 @@
"""CLI: python -m cyclone users {create,list,disable,reset-password,set-role}.
Uses Click's CliRunner (matches the existing parse-837/parse-835 CLI tests).
The full CLI group lives in ``cyclone.cli.main`` — we exercise the
``users`` subgroup through the top-level ``main`` so the wiring is real.
"""
from __future__ import annotations
import pytest
from sqlalchemy import delete, select
from cyclone.auth import users
from cyclone.cli import main as cli_main
from cyclone.db import Session as DbSession
from cyclone.db import SessionLocal, User
@pytest.fixture(autouse=True)
def _clear():
with SessionLocal()() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
yield
with SessionLocal()() as db:
db.execute(delete(DbSession))
db.execute(delete(User))
db.commit()
def test_create_user_via_cli():
from click.testing import CliRunner
runner = CliRunner()
result = runner.invoke(
cli_main,
[
"users", "create", "cli-user",
"--role", "viewer",
"--password", "clipassword1",
],
input="", # don't prompt
)
assert result.exit_code == 0, result.output
with SessionLocal()() as db:
u = users.get_by_username(db, "cli-user")
assert u is not None
assert u.role == "viewer"
def test_list_users_via_cli():
from click.testing import CliRunner
with SessionLocal()() as db:
users.create(db, username="listed", password="hunter2hunter2", role="user")
runner = CliRunner()
result = runner.invoke(cli_main, ["users", "list"])
assert result.exit_code == 0, result.output
assert "listed" in result.output
def test_disable_user_via_cli():
from click.testing import CliRunner
with SessionLocal()() as db:
users.create(db, username="todie", password="hunter2hunter2", role="user")
runner = CliRunner()
result = runner.invoke(cli_main, ["users", "disable", "todie"])
assert result.exit_code == 0, result.output
with SessionLocal()() as db:
u = users.get_by_username(db, "todie")
assert u.disabled_at is not None
def test_reset_password_via_cli():
from click.testing import CliRunner
with SessionLocal()() as db:
users.create(db, username="pwchange", password="oldpassword1", role="user")
runner = CliRunner()
result = runner.invoke(
cli_main,
[
"users", "reset-password", "pwchange",
"--password", "newpassword1",
],
)
assert result.exit_code == 0, result.output
with SessionLocal()() as db:
u = users.get_by_username(db, "pwchange")
assert users.verify_password("newpassword1", u.password_hash)
def test_set_role_via_cli():
from click.testing import CliRunner
with SessionLocal()() as db:
users.create(db, username="promote", password="hunter2hunter2", role="viewer")
runner = CliRunner()
result = runner.invoke(
cli_main,
["users", "set-role", "promote", "--role", "admin"],
)
assert result.exit_code == 0, result.output
with SessionLocal()() as db:
u = users.get_by_username(db, "promote")
assert u.role == "admin"
def test_create_rejects_short_password():
from click.testing import CliRunner
runner = CliRunner()
result = runner.invoke(
cli_main,
[
"users", "create", "weak",
"--role", "viewer",
"--password", "short",
],
)
# Click surfaces validation failures with a non-zero exit code.
assert result.exit_code != 0, result.output
with SessionLocal()() as db:
rows = db.execute(select(User).where(User.username == "weak")).scalars().all()
assert rows == []
def test_disable_unknown_user_exits_nonzero():
from click.testing import CliRunner
runner = CliRunner()
result = runner.invoke(cli_main, ["users", "disable", "ghost"])
assert result.exit_code != 0, result.output