diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 00000000..5d5fdfa0 --- /dev/null +++ b/scripts/__init__.py @@ -0,0 +1 @@ +"""Operational and recovery scripts for routstr-core (not a shipped package).""" diff --git a/scripts/reset_admin_password.py b/scripts/reset_admin_password.py new file mode 100644 index 00000000..e2c8f46f --- /dev/null +++ b/scripts/reset_admin_password.py @@ -0,0 +1,104 @@ +"""Recover admin access by resetting the stored admin password (issue #553). + +The lockout escape hatch for an operator who has lost the admin password. It +talks to the ``secrets`` table directly and deliberately does *not* require +``ROUTSTR_SECRET_KEY``: the admin password is scrypt-hashed (key-independent), +so recovery works even when the encryption key is missing or has changed. + +Two explicit, mutually exclusive actions — running with no arguments only prints +help, so the password can't be reset by accident: + + python scripts/reset_admin_password.py --password + Hash and store now. + + python scripts/reset_admin_password.py --regenerate + Clear the stored hash; the next node startup generates a fresh random + password and logs it once (with the /admin URL). +""" + +import argparse +import asyncio +import sys +import time + +from sqlmodel.ext.asyncio.session import AsyncSession + +from routstr.core.db import create_session, get_secret, set_admin_password +from routstr.core.vault import MIN_PASSWORD_LENGTH + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="reset_admin_password", + description="Reset the node's admin password (recovery from lockout).", + ) + action = parser.add_mutually_exclusive_group() + action.add_argument( + "--password", + metavar="NEW_PASSWORD", + help=f"set this as the new admin password (min {MIN_PASSWORD_LENGTH} chars)", + ) + action.add_argument( + "--regenerate", + action="store_true", + help="clear the password so the next startup generates and logs a new one", + ) + return parser + + +async def apply_reset( + session: AsyncSession, + *, + password: str | None = None, + regenerate: bool = False, +) -> str: + """Perform the requested reset against ``session``; return a status message.""" + if password is not None: + if len(password) < MIN_PASSWORD_LENGTH: + raise ValueError( + f"New password must be at least {MIN_PASSWORD_LENGTH} characters" + ) + await set_admin_password(session, password) + return "Admin password updated." + + if regenerate: + secret = await get_secret(session) + secret.admin_password_hash = None + secret.updated_at = int(time.time()) + session.add(secret) + await session.commit() + return ( + "Admin password cleared. The next node startup will generate a new " + "one and log it once with the /admin URL." + ) + + return "" + + +async def _run(password: str | None, regenerate: bool) -> str: + async with create_session() as session: + return await apply_reset( + session, password=password, regenerate=regenerate + ) + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + if args.password is None and not args.regenerate: + parser.print_help() + return 0 + + try: + message = asyncio.run(_run(args.password, args.regenerate)) + except ValueError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + print(message) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/integration/test_reset_admin_password.py b/tests/integration/test_reset_admin_password.py new file mode 100644 index 00000000..d0e479be --- /dev/null +++ b/tests/integration/test_reset_admin_password.py @@ -0,0 +1,76 @@ +"""Tests for the ``reset_admin_password`` recovery script (issue #553). + +The script is the lockout escape hatch: it works without ``ROUTSTR_SECRET_KEY`` +(scrypt hashing is key-independent). Two explicit, mutually exclusive actions — +``--password`` sets a new hash now, ``--regenerate`` clears the hash so the next +boot generates and logs a fresh one. A bare invocation is informational only and +must never touch the database (so nobody resets their password by accident). +""" + +import pytest +from sqlmodel.ext.asyncio.session import AsyncSession + +from routstr.core import vault +from routstr.core.db import get_secret, set_admin_password +from scripts.reset_admin_password import apply_reset, build_parser, main + + +@pytest.mark.asyncio +async def test_password_sets_a_verifiable_hash( + integration_session: AsyncSession, +) -> None: + await apply_reset(integration_session, password="recover-me-123") + + secret = await get_secret(integration_session) + assert secret.admin_password_hash is not None + assert vault.verify_password("recover-me-123", secret.admin_password_hash) is True + assert secret.updated_at is not None + + +@pytest.mark.asyncio +async def test_regenerate_clears_the_hash( + integration_session: AsyncSession, +) -> None: + # Start from a node that already has an admin password set. + await set_admin_password(integration_session, "old-password-9") + assert (await get_secret(integration_session)).admin_password_hash is not None + + await apply_reset(integration_session, regenerate=True) + + secret = await get_secret(integration_session) + # Cleared -> the next boot's bootstrap_secrets generates and logs a new one. + assert secret.admin_password_hash is None + assert secret.updated_at is not None + + +@pytest.mark.asyncio +async def test_password_below_min_length_is_rejected( + integration_session: AsyncSession, +) -> None: + await set_admin_password(integration_session, "old-password-9") + + with pytest.raises(ValueError, match="8 characters"): + await apply_reset(integration_session, password="short") + + # The existing password is untouched by the rejected reset. + secret = await get_secret(integration_session) + assert vault.verify_password("old-password-9", secret.admin_password_hash or "") + + +def test_password_and_regenerate_are_mutually_exclusive() -> None: + parser = build_parser() + with pytest.raises(SystemExit): + parser.parse_args(["--password", "abcd1234", "--regenerate"]) + + +def test_no_args_prints_help_and_never_opens_a_session( + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + def _fail() -> None: + raise AssertionError("a bare invocation must not touch the database") + + monkeypatch.setattr("scripts.reset_admin_password.create_session", _fail) + + assert main([]) == 0 + assert "usage" in capsys.readouterr().out.lower()