mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-07-30 15:26:14 +00:00
feat(scripts): add admin-password reset CLI
Provide an offline recovery command that sets a new admin password directly in the Secret store, for operators locked out of the admin UI. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
a024d5be5e
commit
8aa2fa5c4a
@@ -0,0 +1 @@
|
||||
"""Operational and recovery scripts for routstr-core (not a shipped package)."""
|
||||
@@ -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 <new-password>
|
||||
Hash and store <new-password> 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())
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user