diff --git a/docs/design/m4-login-hardening.md b/docs/design/m4-login-hardening.md index 53e87ed..6e50d75 100644 --- a/docs/design/m4-login-hardening.md +++ b/docs/design/m4-login-hardening.md @@ -174,7 +174,7 @@ Phase B(TOTP,可选配) - **Reviewer checklist**: 退避在验密码**之前**生效;双键较大值;XFF 仅在配置开启时信任;无把密码写进日志。 ### M4-T03 — CLI 骨架 + reset-password + unlock -- **Status**: `todo` · **Depends**: M4-T01 +- **Status**: `done` · **Depends**: M4-T01 - **Context**: 逃生通道第一批命令。 - **Files**: `create scripts/admin_cli.py`;`create tests/test_admin_cli.py` - **Steps**: argparse 子命令;`reset-password [--password|prompt(getpass)]` 用 Argon2 hasher 重置 `password_hash`;`unlock [--all|--ip|--username]` 删 `auth_login_throttle` 行;`list-admin`;复用 `get_session_local()` + 模型;无网络。 diff --git a/scripts/admin_cli.py b/scripts/admin_cli.py new file mode 100644 index 0000000..619cfc6 --- /dev/null +++ b/scripts/admin_cli.py @@ -0,0 +1,250 @@ +"""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() diff --git a/tests/test_admin_cli.py b/tests/test_admin_cli.py new file mode 100644 index 0000000..0da06d9 --- /dev/null +++ b/tests/test_admin_cli.py @@ -0,0 +1,390 @@ +"""Tests for M4-T03: 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. +- 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. + +DB isolation: each test gets a fresh temporary SQLite database via +``cli_db`` fixture (mirrors the pattern in test_login_throttle.py). +""" + +from __future__ import annotations + +import subprocess +import sys +from datetime import UTC, datetime +from pathlib import Path + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, select +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_throttle import LoginThrottle +from app.services.auth import hash_password, verify_password +from scripts.admin_cli import ( + build_parser, + cmd_list_admin, + cmd_reset_password, + cmd_unlock, +) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +def _make_app_alembic_config(database_url: str) -> Config: + config = Config("alembic_app.ini") + config.set_main_option("sqlalchemy.url", database_url) + return config + + +@pytest.fixture() +def cli_db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + """Fresh migrated SQLite DB wired as the active app DB for the CLI. + + Monkeypatches APP_DATABASE_URL so that ``get_session_local()`` inside + admin_cli reaches this temp file, not the real data store. + """ + db_path = tmp_path / "cli_test.db" + db_url = f"sqlite:///{db_path}" + + monkeypatch.setenv("APP_DATABASE_URL", db_url) + monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") + monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "original-password") + monkeypatch.setenv("AUTH_COOKIE_SECURE_OVERRIDE", "false") + monkeypatch.setenv("AUTH_LOGIN_THROTTLE_ENABLED", "false") + get_settings.cache_clear() + reset_db_caches() + + command.upgrade(_make_app_alembic_config(db_url), "head") + reset_db_caches() + + # Bootstrap the admin user manually so the test controls the initial hash. + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, class_=Session) + with SessionLocal() as session: + # The Alembic upgrade does NOT create a user (that is app startup logic). + # Insert the initial admin user here for tests that need one. + existing = session.scalar(select(AuthUser).limit(1)) + if existing is None: + user = AuthUser( + username="admin", + password_hash=hash_password("original-password"), + is_active=True, + force_password_change=False, + created_at=datetime.now(UTC), + ) + session.add(user) + session.commit() + engine.dispose() + reset_db_caches() + + yield db_url + + get_settings.cache_clear() + reset_db_caches() + + +@pytest.fixture() +def cli_session(cli_db: str) -> Session: + """SQLAlchemy session bound to the CLI test DB.""" + engine = create_engine(cli_db, connect_args={"check_same_thread": False}) + SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, class_=Session) + session = SessionLocal() + try: + yield session + finally: + session.close() + engine.dispose() + + +def _add_throttle_rows(session: Session, rows: list[tuple[str, str]]) -> None: + """Helper: insert (scope, key) throttle rows into the test DB.""" + now = datetime.now(UTC) + for scope, key in rows: + session.add( + LoginThrottle( + scope=scope, + key=key, + failures=5, + first_failed_at=now, + last_failed_at=now, + next_allowed_at=None, + ) + ) + session.commit() + + +def _throttle_count(session: Session) -> int: + return session.query(LoginThrottle).count() + + +# --------------------------------------------------------------------------- +# reset-password tests +# --------------------------------------------------------------------------- + + +class TestResetPassword: + """reset-password sub-command.""" + + def test_new_password_verifies(self, cli_db: str, cli_session: Session): + """After reset, new password passes verify_password; old one fails.""" + parser = build_parser() + args = parser.parse_args(["reset-password", "admin", "--password", "new-secret-123"]) + cmd_reset_password(args) + + # Re-query from the session (committed by CLI, so flush needed). + cli_session.expire_all() + user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user is not None + assert verify_password("new-secret-123", user.password_hash), ( + "New password should verify against stored hash" + ) + assert not verify_password("original-password", user.password_hash), ( + "Old password must no longer verify" + ) + + def test_nonexistent_user_exits_nonzero(self, cli_db: str): + """reset-password for an unknown username must SystemExit with code != 0.""" + parser = build_parser() + args = parser.parse_args( + ["reset-password", "ghost-user", "--password", "doesnt-matter"] + ) + with pytest.raises(SystemExit) as exc_info: + cmd_reset_password(args) + assert exc_info.value.code != 0 + + def test_empty_password_exits_nonzero(self, cli_db: str): + """Passing an empty --password string must fail with non-zero exit.""" + parser = build_parser() + args = parser.parse_args(["reset-password", "admin", "--password", ""]) + with pytest.raises(SystemExit) as exc_info: + cmd_reset_password(args) + assert exc_info.value.code != 0 + + def test_multiple_resets_are_idempotent(self, cli_db: str, cli_session: Session): + """Calling reset-password multiple times leaves the last password active.""" + parser = build_parser() + + args1 = parser.parse_args(["reset-password", "admin", "--password", "first-pass"]) + cmd_reset_password(args1) + + args2 = parser.parse_args(["reset-password", "admin", "--password", "second-pass"]) + cmd_reset_password(args2) + + cli_session.expire_all() + user = cli_session.scalar(select(AuthUser).where(AuthUser.username == "admin")) + assert user is not None + assert verify_password("second-pass", user.password_hash) + assert not verify_password("first-pass", user.password_hash) + + +# --------------------------------------------------------------------------- +# unlock tests +# --------------------------------------------------------------------------- + + +class TestUnlock: + """unlock sub-command.""" + + def test_unlock_all_clears_all_rows(self, cli_db: str, cli_session: Session): + """unlock --all should delete every row in auth_login_throttle.""" + _add_throttle_rows( + cli_session, + [("ip", "1.2.3.4"), ("ip", "5.6.7.8"), ("user", "admin")], + ) + assert _throttle_count(cli_session) == 3 + + parser = build_parser() + args = parser.parse_args(["unlock", "--all"]) + cmd_unlock(args) + + cli_session.expire_all() + assert _throttle_count(cli_session) == 0 + + def test_unlock_ip_clears_only_target_ip(self, cli_db: str, cli_session: Session): + """unlock --ip should remove only the matching scope='ip' row.""" + _add_throttle_rows( + cli_session, + [("ip", "10.0.0.1"), ("ip", "10.0.0.2"), ("user", "admin")], + ) + + parser = build_parser() + args = parser.parse_args(["unlock", "--ip", "10.0.0.1"]) + cmd_unlock(args) + + cli_session.expire_all() + assert _throttle_count(cli_session) == 2 + remaining = cli_session.scalars(select(LoginThrottle)).all() + keys = [(r.scope, r.key) for r in remaining] + assert ("ip", "10.0.0.1") not in keys + assert ("ip", "10.0.0.2") in keys + assert ("user", "admin") in keys + + def test_unlock_username_clears_only_target_user(self, cli_db: str, cli_session: Session): + """unlock --username should remove only the matching scope='user' row.""" + _add_throttle_rows( + cli_session, + [("ip", "1.2.3.4"), ("user", "admin"), ("user", "other")], + ) + + parser = build_parser() + args = parser.parse_args(["unlock", "--username", "admin"]) + cmd_unlock(args) + + cli_session.expire_all() + assert _throttle_count(cli_session) == 2 + remaining = cli_session.scalars(select(LoginThrottle)).all() + keys = [(r.scope, r.key) for r in remaining] + assert ("user", "admin") not in keys + assert ("ip", "1.2.3.4") in keys + assert ("user", "other") in keys + + def test_unlock_all_idempotent(self, cli_db: str, cli_session: Session): + """Calling unlock --all when no rows exist must not raise.""" + assert _throttle_count(cli_session) == 0 + + parser = build_parser() + args = parser.parse_args(["unlock", "--all"]) + cmd_unlock(args) # must not raise + + cli_session.expire_all() + assert _throttle_count(cli_session) == 0 + + def test_unlock_ip_idempotent_when_absent(self, cli_db: str, cli_session: Session): + """Calling unlock --ip for a non-existent IP is a no-op (0 rows deleted).""" + parser = build_parser() + args = parser.parse_args(["unlock", "--ip", "99.99.99.99"]) + cmd_unlock(args) # must not raise; 0 rows deleted is fine + + def test_unlock_username_idempotent_when_absent(self, cli_db: str, cli_session: Session): + """Calling unlock --username for a non-existent username is a no-op.""" + parser = build_parser() + args = parser.parse_args(["unlock", "--username", "nobody"]) + cmd_unlock(args) # must not raise + + def test_unlock_all_prints_row_count(self, cli_db: str, cli_session: Session, capsys): + """unlock --all must print the number of rows deleted.""" + _add_throttle_rows(cli_session, [("ip", "1.2.3.4"), ("user", "admin")]) + + parser = build_parser() + args = parser.parse_args(["unlock", "--all"]) + cmd_unlock(args) + + captured = capsys.readouterr() + assert "2" in captured.out + + def test_unlock_does_not_touch_auth_users(self, cli_db: str, cli_session: Session): + """unlock must not delete or alter auth_users rows.""" + _add_throttle_rows(cli_session, [("ip", "1.2.3.4")]) + user_count_before = cli_session.query(AuthUser).count() + + parser = build_parser() + args = parser.parse_args(["unlock", "--all"]) + cmd_unlock(args) + + cli_session.expire_all() + assert cli_session.query(AuthUser).count() == user_count_before + + +# --------------------------------------------------------------------------- +# list-admin tests +# --------------------------------------------------------------------------- + + +class TestListAdmin: + """list-admin sub-command.""" + + def test_list_admin_prints_header_and_user(self, cli_db: str, capsys): + """list-admin must print the admin user row.""" + parser = build_parser() + args = parser.parse_args(["list-admin"]) + cmd_list_admin(args) + + captured = capsys.readouterr() + assert "admin" in captured.out + + def test_list_admin_no_users( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys + ): + """list-admin on an empty user table should say 'No users found.'""" + db_path = tmp_path / "empty_test.db" + db_url = f"sqlite:///{db_path}" + + monkeypatch.setenv("APP_DATABASE_URL", db_url) + monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin") + monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "x") + monkeypatch.setenv("AUTH_COOKIE_SECURE_OVERRIDE", "false") + get_settings.cache_clear() + reset_db_caches() + command.upgrade(_make_app_alembic_config(db_url), "head") + reset_db_caches() + + # Do NOT bootstrap a user — table is empty. + parser = build_parser() + args = parser.parse_args(["list-admin"]) + cmd_list_admin(args) + + captured = capsys.readouterr() + assert "No users found" in captured.out + + get_settings.cache_clear() + reset_db_caches() + + +# --------------------------------------------------------------------------- +# Argument-parser edge-cases +# --------------------------------------------------------------------------- + + +class TestParserEdgeCases: + """Ensure argparse flags behave as expected.""" + + def test_unlock_requires_flag(self): + """unlock without any flag should exit with a parse error (non-zero).""" + parser = build_parser() + with pytest.raises(SystemExit) as exc_info: + parser.parse_args(["unlock"]) + assert exc_info.value.code != 0 + + def test_unlock_ip_and_all_mutually_exclusive(self): + """--ip and --all together should produce a parse error.""" + parser = build_parser() + with pytest.raises(SystemExit) as exc_info: + parser.parse_args(["unlock", "--all", "--ip", "1.2.3.4"]) + assert exc_info.value.code != 0 + + def test_unknown_command_exits_nonzero(self): + """An unknown sub-command must exit non-zero.""" + parser = build_parser() + with pytest.raises(SystemExit) as exc_info: + parser.parse_args(["nonexistent-command"]) + assert exc_info.value.code != 0 + + +# --------------------------------------------------------------------------- +# Module-level smoke: python -m scripts.admin_cli --help +# --------------------------------------------------------------------------- + + +def test_help_flag_exits_zero(): + """python -m scripts.admin_cli --help must exit 0.""" + result = subprocess.run( + [sys.executable, "-m", "scripts.admin_cli", "--help"], + capture_output=True, + text=True, + ) + assert result.returncode == 0 + assert "reset-password" in result.stdout + assert "unlock" in result.stdout + assert "list-admin" in result.stdout