"""Password reset / user management CLI. Run inside the container: docker exec -it truenas-burnin python -m app.auth_cli reset docker exec -it truenas-burnin python -m app.auth_cli list docker exec -it truenas-burnin python -m app.auth_cli add Reads the password from a TTY prompt — never accept it on the command line so it doesn't leak into shell history. """ from __future__ import annotations import asyncio import getpass import sys import aiosqlite from app import auth from app.config import settings async def _reset(username: str) -> int: found = await auth.get_user_by_username(username) if not found: print(f"No such user: {username}", file=sys.stderr) return 1 pw1 = getpass.getpass(f"New password for {username}: ") pw2 = getpass.getpass("Confirm: ") if pw1 != pw2: print("Passwords don't match.", file=sys.stderr) return 2 if len(pw1) < 8: print("Password must be at least 8 characters.", file=sys.stderr) return 3 new_hash = auth.hash_password(pw1) async with aiosqlite.connect(settings.db_path) as db: await db.execute( "UPDATE users SET password_hash = ? WHERE username = ? COLLATE NOCASE", (new_hash, username), ) await db.commit() print(f"Password updated for {username}.") return 0 async def _list() -> int: async with aiosqlite.connect(settings.db_path) as db: db.row_factory = aiosqlite.Row cur = await db.execute( "SELECT id, username, full_name, is_admin, created_at, last_login_at " "FROM users ORDER BY username" ) rows = list(await cur.fetchall()) if not rows: print("(no users)") return 0 for r in rows: flag = "admin" if r["is_admin"] else "user " print(f" [{flag}] {r['username']:24s} created={r['created_at'][:19]} " f"last_login={(r['last_login_at'] or '-')[:19]}") return 0 async def _add(username: str) -> int: pw1 = getpass.getpass(f"Password for new user {username}: ") pw2 = getpass.getpass("Confirm: ") if pw1 != pw2: print("Passwords don't match.", file=sys.stderr) return 2 full = input("Full name (optional, press enter to skip): ").strip() or None is_admin = input("Admin? [y/N]: ").strip().lower() == "y" try: u = await auth.create_user(username, pw1, full, is_admin=is_admin) except ValueError as exc: print(f"Failed: {exc}", file=sys.stderr) return 1 print(f"Created user {u.username} (admin={u.is_admin}).") return 0 def main() -> int: if len(sys.argv) < 2: print(__doc__, file=sys.stderr) return 64 cmd = sys.argv[1] if cmd == "list": return asyncio.run(_list()) if cmd == "reset" and len(sys.argv) == 3: return asyncio.run(_reset(sys.argv[2])) if cmd == "add" and len(sys.argv) == 3: return asyncio.run(_add(sys.argv[2])) print(__doc__, file=sys.stderr) return 64 if __name__ == "__main__": sys.exit(main())