"""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. disable-totp Disable TOTP for the given user without requiring any existing credential (recovery codes, current TOTP code, or password). Sets totp_enabled=False, clears totp_secret, and deletes ALL recovery codes for that user. This is the final escape hatch when all recovery codes are lost. reissue-totp Generate a fresh TOTP secret for the given user and print the new otpauth:// URI so the user can re-enroll in their Authenticator app. totp_enabled is LEFT UNCHANGED (caller must re-run enable flow if needed). Existing recovery codes are NOT deleted. Recovery codes are independent random values with no cryptographic binding to the TOTP secret, so they remain valid after reissue and continue to serve as backup login credentials. Silently removing them would unexpectedly strip the user of that backup; use disable-totp for a full cleanup (secret + recovery codes). 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`` (totp_secret/totp_enabled/password_hash columns) and ``auth_login_throttle`` / ``auth_recovery_code`` (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)) import pyotp from sqlalchemy import delete, select from sqlalchemy.orm import Session from app.config import get_settings from app.db import get_session_local from app.models.auth import AuthUser, RecoveryCode 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_disable_totp(args: argparse.Namespace) -> None: """Disable TOTP for — no credential required (escape hatch). Sets totp_enabled=False, clears totp_secret, and deletes ALL recovery codes for this user. This is the final escape route when all recovery codes have been lost: whoever holds server CLI access can always turn off TOTP, restoring plain-password login. """ username: str = args.username session = _get_session() try: user = _find_user(session, username) if user is None: _die(f"User '{username}' not found.") assert user.id is not None # Delete all recovery codes for this user. deleted = session.execute( delete(RecoveryCode).where(RecoveryCode.user_id == user.id) ) rc_count = deleted.rowcount # Clear TOTP state. user.totp_enabled = False user.totp_secret = None session.commit() print(f"TOTP disabled for '{username}'. {rc_count} recovery code(s) deleted.") print("The user can now log in with password only.") finally: session.close() def cmd_reissue_totp(args: argparse.Namespace) -> None: """Generate a fresh TOTP secret for and print the otpauth URI. Design choices: - ``totp_enabled`` is LEFT UNCHANGED. If the user currently has TOTP enabled, the new secret takes effect immediately on the next login (old code invalid, new code accepted) — no web enable step required. If TOTP was disabled, the secret is updated but login remains password-only until the web setup/enable flow is completed. - Existing recovery codes are NOT deleted here. Recovery codes are independent random values stored as Argon2 hashes in ``auth_recovery_code.code_hash``; login verification compares only those hashes and has no dependency on ``totp_secret``. Reissuing the secret therefore leaves all existing recovery codes fully valid — they remain the user's backup login path. Silently deleting them would unexpectedly strip that backup; ``disable-totp`` is the explicit full cleanup path (secret + all recovery codes). - This command's purpose is to hand the new secret URI to the admin so they can re-enroll their Authenticator — useful when the device is lost but the account credentials are still intact. """ username: str = args.username session = _get_session() try: user = _find_user(session, username) if user is None: _die(f"User '{username}' not found.") # Generate a new base-32 secret using pyotp (same library as the web TOTP flow). new_secret = pyotp.random_base32() # Build the provisioning URI using the same issuer logic as the web service. settings = get_settings() issuer = settings.effective_totp_issuer otpauth_uri = pyotp.TOTP(new_secret).provisioning_uri( name=user.username, issuer_name=issuer, ) user.totp_secret = new_secret session.commit() print(f"New TOTP secret issued for '{username}'.") print(f"Secret : {new_secret}") print(f"URI : {otpauth_uri}") print() if user.totp_enabled: print( "NOTE: TOTP is still ENABLED. Scan the URI above in your Authenticator app;" " the new code is active immediately on next login — the old code is now invalid." " To fully disable TOTP instead, use 'disable-totp'." ) else: print( "NOTE: TOTP is currently DISABLED for this user — login remains password-only." " The new secret has been saved, but will only take effect once TOTP is" " enabled via the web settings page." ) 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) # -- disable-totp -- dt = subparsers.add_parser( "disable-totp", help=( "Disable TOTP for a user (escape hatch — no credential required). " "Clears totp_secret, sets totp_enabled=False, deletes all recovery codes." ), ) dt.add_argument("username", help="Username whose TOTP to disable.") dt.set_defaults(func=cmd_disable_totp) # -- reissue-totp -- rt = subparsers.add_parser( "reissue-totp", help=( "Generate a new TOTP secret for a user and print the otpauth:// URI. " "totp_enabled is left unchanged; re-enroll via the web UI." ), ) rt.add_argument("username", help="Username for whom to reissue the TOTP secret.") rt.set_defaults(func=cmd_reissue_totp) # -- 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()