"""Admin CLI — escape hatch for local server administration. Entry point:: python -m scripts.admin_cli [args] Commands -------- reset-password [--password ] Reset the Argon2 password hash for the given admin user. If ``--password`` is omitted the new password is read interactively with ``getpass.getpass`` (no echo). Prompts twice for confirmation. unlock [--all | --ip | --username ] Delete rows from ``auth_login_throttle``. --all clear every row (all IPs and usernames) --ip clear the row for a specific IP address (scope='ip') --username clear the row for a specific username (scope='user') Prints the number of rows deleted. list-admin Print all users with their auth status columns (id, username, is_active, force_password_change). Design contract --------------- - Direct DB access via ``get_session_local()`` — no HTTP, no network. - Only touches ``auth_users`` (password_hash column) and ``auth_login_throttle`` (delete rows). Never touches any user-data table (location, poo, public_ip, app_config, etc.). - Non-zero exit code on user-facing errors (missing user, bad args, …). """ from __future__ import annotations import argparse import getpass import sys from pathlib import Path # Ensure project root is on sys.path when invoked as `python -m scripts.admin_cli` # or directly as a script. PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from sqlalchemy import delete, select from sqlalchemy.orm import Session from app.db import get_session_local from app.models.auth import AuthUser from app.models.auth_throttle import LoginThrottle from app.services.auth import hash_password, verify_password # noqa: F401 (used below) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _get_session() -> Session: """Return a new SQLAlchemy session for the configured app database.""" SessionLocal = get_session_local() return SessionLocal() def _find_user(session: Session, username: str) -> AuthUser | None: return session.scalar(select(AuthUser).where(AuthUser.username == username).limit(1)) def _die(message: str, code: int = 1) -> None: print(f"Error: {message}", file=sys.stderr) sys.exit(code) # --------------------------------------------------------------------------- # Sub-command implementations # --------------------------------------------------------------------------- def cmd_reset_password(args: argparse.Namespace) -> None: """Reset the password for .""" username: str = args.username if args.password is not None: new_password: str = args.password else: # Interactive prompt — getpass suppresses echo. new_password = getpass.getpass(f"New password for '{username}': ") confirm = getpass.getpass("Confirm new password: ") if new_password != confirm: _die("Passwords do not match.") if not new_password: _die("Password must not be empty.") session = _get_session() try: user = _find_user(session, username) if user is None: _die(f"User '{username}' not found.") user.password_hash = hash_password(new_password) session.commit() print(f"Password for '{username}' has been reset.") finally: session.close() def cmd_unlock(args: argparse.Namespace) -> None: """Delete rows from auth_login_throttle according to the given filter.""" use_all: bool = args.all ip: str | None = args.ip username: str | None = args.username # Validate: exactly one of --all / --ip / --username must be provided. flags = [use_all, ip is not None, username is not None] if sum(flags) == 0: _die("One of --all, --ip, or --username is required.") if sum(flags) > 1: _die("Only one of --all, --ip, or --username may be specified at a time.") session = _get_session() try: if use_all: stmt = delete(LoginThrottle) result = session.execute(stmt) count = result.rowcount elif ip is not None: stmt = delete(LoginThrottle).where( LoginThrottle.scope == "ip", LoginThrottle.key == ip, ) result = session.execute(stmt) count = result.rowcount else: # username is not None stmt = delete(LoginThrottle).where( LoginThrottle.scope == "user", LoginThrottle.key == username, ) result = session.execute(stmt) count = result.rowcount session.commit() noun = "row" if count == 1 else "rows" print(f"Cleared {count} throttle {noun}.") finally: session.close() def cmd_list_admin(args: argparse.Namespace) -> None: # noqa: ARG001 """List all admin users with their status columns.""" session = _get_session() try: users = session.scalars(select(AuthUser).order_by(AuthUser.id)).all() if not users: print("No users found.") return # Header print(f"{'ID':<5} {'USERNAME':<30} {'IS_ACTIVE':<10} {'FORCE_PW_CHANGE':<16}") print("-" * 65) for u in users: print( f"{u.id:<5} {u.username:<30} {str(u.is_active):<10} " f"{str(u.force_password_change):<16}" ) finally: session.close() # --------------------------------------------------------------------------- # Argument parser # --------------------------------------------------------------------------- def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="python -m scripts.admin_cli", description=( "Admin CLI — local escape hatch for password reset, throttle unlock, " "and user inspection. Connects directly to the configured app DB; " "no HTTP server required." ), ) subparsers = parser.add_subparsers(dest="command", metavar="") subparsers.required = True # -- reset-password -- rp = subparsers.add_parser( "reset-password", help="Reset a user's password (prompts interactively if --password is omitted).", ) rp.add_argument("username", help="Username whose password to reset.") rp.add_argument( "--password", default=None, help="New password (omit to be prompted interactively with no echo).", ) rp.set_defaults(func=cmd_reset_password) # -- unlock -- ul = subparsers.add_parser( "unlock", help="Clear login-throttle rows (escape when locked out by exponential back-off).", ) unlock_group = ul.add_mutually_exclusive_group(required=True) unlock_group.add_argument( "--all", action="store_true", default=False, help="Clear ALL throttle rows.", ) unlock_group.add_argument( "--ip", metavar="", default=None, help="Clear the throttle row for this specific IP address.", ) unlock_group.add_argument( "--username", metavar="", default=None, help="Clear the throttle row for this specific username.", ) ul.set_defaults(func=cmd_unlock) # -- list-admin -- la = subparsers.add_parser( "list-admin", help="List all users and their auth status (for diagnostics).", ) la.set_defaults(func=cmd_list_admin) return parser # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def main(argv: list[str] | None = None) -> None: parser = build_parser() args = parser.parse_args(argv) args.func(args) if __name__ == "__main__": main()