From 51da8c57160ada385877d0de79894b598b8982df Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sun, 21 Jun 2026 22:27:07 +0200 Subject: [PATCH] M4-T07: add admin_cli disable-totp and reissue-totp escape commands --- docs/design/m4-login-hardening.md | 2 +- scripts/admin_cli.py | 145 ++++++++++++++++++++- tests/test_admin_cli.py | 205 +++++++++++++++++++++++++++++- 3 files changed, 345 insertions(+), 7 deletions(-) diff --git a/docs/design/m4-login-hardening.md b/docs/design/m4-login-hardening.md index 7aeb867..8725f88 100644 --- a/docs/design/m4-login-hardening.md +++ b/docs/design/m4-login-hardening.md @@ -224,7 +224,7 @@ Phase B（TOTP，可选配） - **Reviewer checklist**: `totp_required` 时确未发 cookie；恢复码消费后不可复用；通用错误不泄露细节。 ### M4-T07 — CLI disable-totp / reissue-totp -- **Status**: `todo` · **Depends**: M4-T04（逻辑上配合 T03 的 CLI 骨架） +- **Status**: `done` · **Depends**: M4-T04（逻辑上配合 T03 的 CLI 骨架） - **Files**: `modify scripts/admin_cli.py`；`modify tests/test_admin_cli.py` - **Steps**: `disable-totp `：`totp_enabled=false` + 清 secret + 删恢复码（**不需任何恢复码即可执行**）；`reissue-totp `：生成新 secret（可选打印新 URI）。 - **Out of scope / 不要碰**: 不碰用户数据表。 diff --git a/scripts/admin_cli.py b/scripts/admin_cli.py index 619cfc6..5ccaa4e 100644 --- a/scripts/admin_cli.py +++ b/scripts/admin_cli.py @@ -18,6 +18,22 @@ unlock [--all | --ip | --username ] --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). @@ -25,9 +41,9 @@ list-admin 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.). +- 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, …). """ @@ -44,11 +60,13 @@ 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 +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) @@ -148,6 +166,103 @@ def cmd_unlock(args: argparse.Namespace) -> None: 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() @@ -225,6 +340,28 @@ def build_parser() -> argparse.ArgumentParser: ) 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", diff --git a/tests/test_admin_cli.py b/tests/test_admin_cli.py index 0da06d9..f17c59c 100644 --- a/tests/test_admin_cli.py +++ b/tests/test_admin_cli.py @@ -1,8 +1,10 @@ -"""Tests for M4-T03: scripts/admin_cli.py — CLI escape hatch. +"""Tests for M4-T03 + M4-T07: scripts/admin_cli.py — CLI escape hatch. Covers: - reset-password: new password verifies, old password fails. - unlock: clears specified/all throttle rows; prints count; idempotent. +- disable-totp: clears TOTP state + recovery codes without requiring any credential. +- reissue-totp: writes new secret; prints otpauth URI; secret changes. - list-admin: prints user table without crashing. - Non-existent user → friendly error + SystemExit with non-zero code. - Data-safety: CLI must never reference user-data tables in delete/drop. @@ -26,13 +28,15 @@ from sqlalchemy.orm import Session, sessionmaker from app.db import reset_db_caches from app.config import get_settings -from app.models.auth import AuthUser +from app.models.auth import AuthUser, RecoveryCode from app.models.auth_throttle import LoginThrottle from app.services.auth import hash_password, verify_password from scripts.admin_cli import ( build_parser, + cmd_disable_totp, cmd_list_admin, cmd_reset_password, + cmd_reissue_totp, cmd_unlock, ) @@ -129,6 +133,25 @@ def _throttle_count(session: Session) -> int: return session.query(LoginThrottle).count() +def _enable_totp_for_user(session: Session, username: str, secret: str = "JBSWY3DPEHPK3PXP") -> None: + """Helper: set TOTP enabled + secret for a user and add some recovery codes.""" + user = session.scalar(select(AuthUser).where(AuthUser.username == username)) + assert user is not None, f"User '{username}' not found in _enable_totp_for_user" + user.totp_secret = secret + user.totp_enabled = True + # Add two fake recovery codes so we can verify they get deleted. + session.add(RecoveryCode(user_id=user.id, code_hash=hash_password("aaaa-bbbb"))) + session.add(RecoveryCode(user_id=user.id, code_hash=hash_password("cccc-dddd"))) + session.commit() + session.expire_all() + + +def _recovery_code_count(session: Session, username: str) -> int: + user = session.scalar(select(AuthUser).where(AuthUser.username == username)) + assert user is not None + return session.query(RecoveryCode).filter(RecoveryCode.user_id == user.id).count() + + # --------------------------------------------------------------------------- # reset-password tests # --------------------------------------------------------------------------- @@ -342,6 +365,182 @@ class TestListAdmin: reset_db_caches() +# --------------------------------------------------------------------------- +# disable-totp tests +# --------------------------------------------------------------------------- + + +class TestDisableTotp: + """disable-totp sub-command (M4-T07).""" + + def test_disable_totp_clears_state(self, cli_db: str, cli_session: Session): + """After disable-totp: totp_enabled=False, totp_secret=None, recovery codes gone.""" + _enable_totp_for_user(cli_session, "admin") + + # Verify preconditions. + user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user is not None + assert user.totp_enabled is True + assert user.totp_secret is not None + assert _recovery_code_count(cli_session, "admin") == 2 + + # Run disable-totp — NO credential supplied (escape hatch semantics). + parser = build_parser() + args = parser.parse_args(["disable-totp", "admin"]) + cmd_disable_totp(args) + + # Verify state. + cli_session.expire_all() + user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user is not None + assert user.totp_enabled is False + assert user.totp_secret is None + assert _recovery_code_count(cli_session, "admin") == 0 + + def test_disable_totp_no_credential_required(self, cli_db: str, cli_session: Session): + """disable-totp must succeed without any recovery code or password argument.""" + _enable_totp_for_user(cli_session, "admin") + + # The argparse spec for disable-totp takes only username — no credential flags. + # Parsing must succeed with just the username. + parser = build_parser() + args = parser.parse_args(["disable-totp", "admin"]) + # No exception means no credential was needed. + cmd_disable_totp(args) # must not raise + + def test_disable_totp_already_disabled_is_noop(self, cli_db: str, cli_session: Session): + """disable-totp on a user who never had TOTP is a harmless no-op.""" + user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user is not None + assert user.totp_enabled is False + + parser = build_parser() + args = parser.parse_args(["disable-totp", "admin"]) + cmd_disable_totp(args) # must not raise + + cli_session.expire_all() + user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user is not None + assert user.totp_enabled is False + assert user.totp_secret is None + + def test_disable_totp_nonexistent_user_exits_nonzero(self, cli_db: str): + """disable-totp for an unknown user must SystemExit with code != 0.""" + parser = build_parser() + args = parser.parse_args(["disable-totp", "ghost-user"]) + with pytest.raises(SystemExit) as exc_info: + cmd_disable_totp(args) + assert exc_info.value.code != 0 + + def test_disable_totp_prints_confirmation( + self, cli_db: str, cli_session: Session, capsys + ): + """disable-totp must print a confirmation message mentioning the user.""" + _enable_totp_for_user(cli_session, "admin") + + parser = build_parser() + args = parser.parse_args(["disable-totp", "admin"]) + cmd_disable_totp(args) + + captured = capsys.readouterr() + assert "admin" in captured.out + + def test_disable_totp_does_not_touch_password( + self, cli_db: str, cli_session: Session + ): + """disable-totp must not alter the user's password_hash.""" + _enable_totp_for_user(cli_session, "admin") + + user_before = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user_before is not None + pw_hash_before = user_before.password_hash + + parser = build_parser() + args = parser.parse_args(["disable-totp", "admin"]) + cmd_disable_totp(args) + + cli_session.expire_all() + user_after = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user_after is not None + assert user_after.password_hash == pw_hash_before + + +# --------------------------------------------------------------------------- +# reissue-totp tests +# --------------------------------------------------------------------------- + + +class TestReissueTotp: + """reissue-totp sub-command (M4-T07).""" + + def test_reissue_totp_changes_secret(self, cli_db: str, cli_session: Session): + """After reissue-totp, totp_secret must be a new non-empty value.""" + old_secret = "JBSWY3DPEHPK3PXP" + _enable_totp_for_user(cli_session, "admin", secret=old_secret) + + parser = build_parser() + args = parser.parse_args(["reissue-totp", "admin"]) + cmd_reissue_totp(args) + + cli_session.expire_all() + user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user is not None + assert user.totp_secret is not None + assert user.totp_secret != old_secret, "Secret must be regenerated" + + def test_reissue_totp_prints_uri(self, cli_db: str, cli_session: Session, capsys): + """reissue-totp must print an otpauth:// URI and the secret.""" + _enable_totp_for_user(cli_session, "admin") + + parser = build_parser() + args = parser.parse_args(["reissue-totp", "admin"]) + cmd_reissue_totp(args) + + captured = capsys.readouterr() + assert "otpauth://" in captured.out + assert "Secret" in captured.out + + def test_reissue_totp_totp_enabled_unchanged(self, cli_db: str, cli_session: Session): + """reissue-totp must leave totp_enabled in its current state (True stays True).""" + _enable_totp_for_user(cli_session, "admin") + + user_before = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user_before is not None + assert user_before.totp_enabled is True + + parser = build_parser() + args = parser.parse_args(["reissue-totp", "admin"]) + cmd_reissue_totp(args) + + cli_session.expire_all() + user_after = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user_after is not None + assert user_after.totp_enabled is True, "totp_enabled must remain unchanged" + + def test_reissue_totp_disabled_user_stays_disabled( + self, cli_db: str, cli_session: Session + ): + """reissue-totp on a user with totp_enabled=False leaves it False.""" + # Don't enable TOTP — user is in disabled state. + parser = build_parser() + args = parser.parse_args(["reissue-totp", "admin"]) + cmd_reissue_totp(args) + + cli_session.expire_all() + user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user is not None + assert user.totp_enabled is False, "totp_enabled must remain False" + assert user.totp_secret is not None, "A new secret should have been written" + + def test_reissue_totp_nonexistent_user_exits_nonzero(self, cli_db: str): + """reissue-totp for an unknown user must SystemExit with code != 0.""" + parser = build_parser() + args = parser.parse_args(["reissue-totp", "ghost-user"]) + with pytest.raises(SystemExit) as exc_info: + cmd_reissue_totp(args) + assert exc_info.value.code != 0 + + # --------------------------------------------------------------------------- # Argument-parser edge-cases # --------------------------------------------------------------------------- @@ -387,4 +586,6 @@ def test_help_flag_exits_zero(): assert result.returncode == 0 assert "reset-password" in result.stdout assert "unlock" in result.stdout + assert "disable-totp" in result.stdout + assert "reissue-totp" in result.stdout assert "list-admin" in result.stdout