diff --git a/.env.example b/.env.example
index 543d49a..d2a054d 100644
--- a/.env.example
+++ b/.env.example
@@ -18,6 +18,13 @@ LOG_LEVEL=INFO
# python -c "import secrets; print(secrets.token_urlsafe(32))"
SECRET_KEY=changeme
+# Optional. A way back into the dashboard if you forget your password and have
+# no recovery code (every account is shown one when it is created). Put it in
+# the deployment's environment, then give it on the Forgot password page with
+# your email. At least 24 characters; leave it unset to turn it off.
+# python -c "import secrets; print(secrets.token_urlsafe(32))"
+# RECOVERY_KEY=
+
# Apply migrations when the container starts, before the app does.
#
# Off by default, and left off under Compose: the one-shot "migrate" service
diff --git a/apps/api/pyproject.toml b/apps/api/pyproject.toml
index 00afba5..f29a30c 100644
--- a/apps/api/pyproject.toml
+++ b/apps/api/pyproject.toml
@@ -19,6 +19,11 @@ dependencies = [
"alembic>=1.13",
]
+[project.scripts]
+# `docker compose exec api seskit reset-password`. Named for the product rather
+# than the package, because it is what an operator types.
+seskit = "seskit_api.cli:main"
+
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
diff --git a/apps/api/src/seskit_api/cli.py b/apps/api/src/seskit_api/cli.py
new file mode 100644
index 0000000..6d541d0
--- /dev/null
+++ b/apps/api/src/seskit_api/cli.py
@@ -0,0 +1,165 @@
+"""The ``seskit`` command (§24): what an operator does with a shell and no browser.
+
+ seskit reset-password
+ docker compose exec api seskit reset-password
+ docker compose exec api python -m seskit_api.cli reset-password
+
+**Why this is the first command.** A forgotten password is the one problem the
+dashboard cannot solve for itself. There is no emailed reset link, on purpose: a
+fresh instance may not be able to send email until somebody can log in and
+connect AWS, which is exactly when a password gets forgotten.
+
+This is for an operator with a shell. Someone without one - a deployment made by
+pushing to Git has none - recovers from the Forgot password page with a recovery
+code or the ``RECOVERY_KEY``; see ``seskit_core.security.recovery``.
+
+**Why it asks rather than accepts.** The password is read from a hidden prompt
+and is deliberately not an argument or an environment variable: either would put
+it in shell history or a process listing. The email may be an argument, because
+it is not a secret.
+
+**Why it ends the account's sessions.** Someone resetting a password has either
+lost it or lost control of it, and in the second case every session already open
+is the thing being reset away. Ending them costs a legitimate user one login.
+
+It does not reactivate a deactivated account, and says so.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import getpass
+import sys
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from redis.asyncio import Redis
+from seskit_core.db import dispose_engine, get_session_factory
+from seskit_core.redis import close_redis, get_redis
+from seskit_core.security.sessions import delete_user_sessions
+from seskit_core.services import MIN_PASSWORD_LENGTH, PasswordTooShort, reset_password
+from sqlalchemy.ext.asyncio import AsyncSession
+
+EXIT_OK = 0
+EXIT_NO_ACCOUNT = 1
+EXIT_USAGE = 2
+
+
+@dataclass(frozen=True, slots=True)
+class Reset:
+ """What a reset did, for the command to report."""
+
+ is_active: bool
+ sessions_ended: int
+
+
+async def reset_account_password(
+ session: AsyncSession, redis: Redis, *, email: str, password: str
+) -> Reset | None:
+ """Set the password, then end the account's sessions. None when there is no account.
+
+ In that order: the new password is committed before anything is revoked, so a
+ Redis failure leaves the password changed and the old sessions standing -
+ which is recoverable by running it again - rather than sessions revoked and
+ the old password still in force.
+
+ Raises :class:`PasswordTooShort`.
+ """
+ user = await reset_password(session, email=email, password=password)
+ if user is None:
+ return None
+ await session.commit()
+
+ ended = await delete_user_sessions(redis, user.id)
+ return Reset(is_active=user.is_active, sessions_ended=ended)
+
+
+def _fail(message: str, code: int) -> int:
+ print(message, file=sys.stderr)
+ return code
+
+
+#: Shown when the address matches nothing. A sentence, not a query the program
+#: runs; it assumes the Compose stack because that is how SESKit is installed.
+LIST_ACCOUNTS_HINT = (
+ "The addresses on this instance: "
+ 'docker compose exec db psql -U seskit seskit -c "select email from users;"'
+)
+
+
+def _prompt(email: str | None) -> tuple[str, str] | int:
+ """The address and the new password, or an exit code when they cannot be had.
+
+ Synchronous and before the event loop starts: ``input`` and ``getpass``
+ block, and nothing else is waiting on this process.
+ """
+ address = (email or input("Account email: ")).strip()
+ if not address:
+ return _fail("An email address is required.", EXIT_USAGE)
+
+ password = getpass.getpass("New password: ")
+ if getpass.getpass("Again: ") != password:
+ return _fail("The two passwords differ. Nothing was changed.", EXIT_USAGE)
+ return address, password
+
+
+async def _reset_password_command(address: str, password: str) -> int:
+ try:
+ async with get_session_factory()() as session:
+ result = await reset_account_password(
+ session, get_redis(), email=address, password=password
+ )
+ except PasswordTooShort:
+ return _fail(
+ f"Use at least {MIN_PASSWORD_LENGTH} characters. Nothing was changed.", EXIT_USAGE
+ )
+ finally:
+ await dispose_engine()
+ await close_redis()
+
+ if result is None:
+ return _fail(
+ f"There is no account for {address}. Nothing was changed.\n{LIST_ACCOUNTS_HINT}",
+ EXIT_NO_ACCOUNT,
+ )
+
+ print("Password changed.")
+ if result.sessions_ended:
+ print(f"Signed out {result.sessions_ended} open session(s).")
+ if not result.is_active:
+ print(
+ "This account is deactivated, so it still cannot sign in. "
+ "Resetting a password does not reactivate it."
+ )
+ return EXIT_OK
+
+
+def build_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(prog="seskit", description="Operate a SESKit instance.")
+ commands = parser.add_subparsers(dest="command", required=True)
+
+ reset = commands.add_parser(
+ "reset-password",
+ help="set a new dashboard password for an account",
+ description=(
+ "Set a new dashboard password. The password is asked for, never "
+ "passed as an argument, so it stays out of shell history."
+ ),
+ )
+ reset.add_argument("--email", help="the account's email address (asked for if omitted)")
+ return parser
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+ args = build_parser().parse_args(argv)
+ if args.command == "reset-password":
+ answers = _prompt(args.email)
+ if isinstance(answers, int):
+ return answers
+ return asyncio.run(_reset_password_command(*answers))
+ return EXIT_USAGE # pragma: no cover - argparse rejects an unknown command first
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/apps/api/src/seskit_api/main.py b/apps/api/src/seskit_api/main.py
index 488ffc9..6159889 100644
--- a/apps/api/src/seskit_api/main.py
+++ b/apps/api/src/seskit_api/main.py
@@ -27,6 +27,7 @@
)
from seskit_api.queue import create_queue
from seskit_api.routes import (
+ account,
api_keys,
auth,
aws,
@@ -259,6 +260,7 @@ async def _unhandled(request: Request, exc: Exception) -> JSONResponse | Respons
app.include_router(domains.router)
app.include_router(emails.router)
app.include_router(webhooks.router)
+ app.include_router(account.router)
app.include_router(suppressions.router)
# Public and unauthenticated, unlike everything above it.
app.include_router(unsubscribe.router)
diff --git a/apps/api/src/seskit_api/routes/account.py b/apps/api/src/seskit_api/routes/account.py
new file mode 100644
index 0000000..2e26401
--- /dev/null
+++ b/apps/api/src/seskit_api/routes/account.py
@@ -0,0 +1,88 @@
+"""The Settings page (``/settings``): your account, and the way back into it (§7).
+
+Today that is one thing - the recovery code. The nav has linked here since the
+shell was built, and until now it led nowhere; a page that holds the one setting
+that has to exist before it is needed is a reasonable first occupant.
+
+**Why regenerating asks for the password.** A signed-in session is not proof of
+ownership: it is also exactly what a stolen cookie looks like, and a new
+recovery code would let whoever holds the cookie take the account for good,
+including after the real owner changes their password. Asking for the password
+again makes the code something only the owner can replace.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Form, Request, Response, status
+from fastapi.responses import HTMLResponse
+from seskit_core.db import get_session
+from seskit_core.logging import get_logger
+from seskit_core.services import list_projects, regenerate_recovery_code
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from seskit_api.dependencies import CurrentUser, require_user, verify_csrf
+from seskit_api.routes.auth import show_recovery_code
+from seskit_api.templating import render
+
+logger = get_logger(__name__)
+
+router = APIRouter(tags=["settings"], include_in_schema=False)
+
+
+async def _page(
+ request: Request,
+ db: AsyncSession,
+ current: CurrentUser,
+ *,
+ error: str | None = None,
+ status_code: int = status.HTTP_200_OK,
+) -> HTMLResponse:
+ return render(
+ request,
+ "pages/settings.html",
+ current=current,
+ nav_active="settings",
+ projects=await list_projects(db, current.user.id),
+ has_recovery_code=current.user.recovery_code_hash is not None,
+ error=error,
+ status_code=status_code,
+ )
+
+
+@router.get("/settings", response_class=HTMLResponse, summary="Settings")
+async def settings_page(
+ request: Request,
+ db: Annotated[AsyncSession, Depends(get_session)],
+ current: Annotated[CurrentUser, Depends(require_user)],
+) -> HTMLResponse:
+ return await _page(request, db, current)
+
+
+@router.post(
+ "/settings/recovery-code",
+ response_class=HTMLResponse,
+ dependencies=[Depends(verify_csrf)],
+ summary="Make a new recovery code",
+)
+async def new_recovery_code(
+ request: Request,
+ db: Annotated[AsyncSession, Depends(get_session)],
+ current: Annotated[CurrentUser, Depends(require_user)],
+ password: Annotated[str, Form()],
+) -> Response:
+ code = await regenerate_recovery_code(db, current.user, password=password)
+ if code is None:
+ logger.info("recovery_code_refused", user_id=current.user.id)
+ return await _page(
+ request,
+ db,
+ current,
+ error="That is not your current password.",
+ status_code=status.HTTP_400_BAD_REQUEST,
+ )
+
+ await db.commit()
+ logger.info("recovery_code_replaced", user_id=current.user.id)
+ return show_recovery_code(request, code, reason="regenerated")
diff --git a/apps/api/src/seskit_api/routes/auth.py b/apps/api/src/seskit_api/routes/auth.py
index b42f896..25fb268 100644
--- a/apps/api/src/seskit_api/routes/auth.py
+++ b/apps/api/src/seskit_api/routes/auth.py
@@ -10,19 +10,25 @@
from seskit_core.config import Settings
from seskit_core.db import get_session
from seskit_core.logging import get_logger
+from seskit_core.models import normalise_email
from seskit_core.redis import get_redis
from seskit_core.security.sessions import (
create_session,
delete_session,
+ delete_user_sessions,
set_current_project,
)
+from seskit_core.security.throttle import RECOVERY_SCOPE, is_throttled, record_failure
from seskit_core.security.throttle import clear as clear_attempts
-from seskit_core.security.throttle import is_throttled, record_failure
from seskit_core.services import (
+ MIN_PASSWORD_LENGTH,
EmailAlreadyRegistered,
+ PasswordTooShort,
SignupClosed,
authenticate,
get_owned_project,
+ issue_recovery_code,
+ recover_password,
register_user,
signup_allowed,
)
@@ -40,10 +46,6 @@
router = APIRouter(tags=["auth"], include_in_schema=False)
logger = get_logger(__name__)
-#: Long enough to be worth hashing. OWASP's floor is 8; this is an email
-#: platform holding AWS access, so it asks for a little more.
-MIN_PASSWORD_LENGTH = 12
-
#: One message for every failure. Naming which half was wrong would let anyone
#: with the login form discover which addresses have accounts.
INVALID_CREDENTIALS = "Email or password is incorrect."
@@ -61,6 +63,18 @@ def _client_address(request: Request) -> str:
return request.client.host if request.client else "unknown"
+def _set_session_cookie(response: Response, token: str, settings: Settings) -> None:
+ response.set_cookie(
+ key=settings.SESSION_COOKIE_NAME,
+ value=token,
+ max_age=settings.session_ttl_seconds,
+ httponly=True, # unreadable from JavaScript, so XSS cannot lift it
+ samesite="lax", # not sent on cross-site POSTs
+ secure=settings.session_cookie_secure,
+ path="/",
+ )
+
+
def _start_session_response(
redirect_to: str,
token: str,
@@ -72,15 +86,35 @@ def _start_session_response(
on the landing page does not repost the credentials.
"""
response = RedirectResponse(url=redirect_to, status_code=status.HTTP_303_SEE_OTHER)
- response.set_cookie(
- key=settings.SESSION_COOKIE_NAME,
- value=token,
- max_age=settings.session_ttl_seconds,
- httponly=True, # unreadable from JavaScript, so XSS cannot lift it
- samesite="lax", # not sent on cross-site POSTs
- secure=settings.session_cookie_secure,
- path="/",
- )
+ _set_session_cookie(response, token, settings)
+ return response
+
+
+def show_recovery_code(
+ request: Request,
+ code: str,
+ *,
+ reason: str,
+ token: str | None = None,
+ settings: Settings | None = None,
+) -> Response:
+ """The page that shows a recovery code, once.
+
+ Rendered as the answer to the POST rather than redirected to: the code
+ exists in the clear only in this response, and a redirect would need
+ somewhere to leave it in the meantime. The price is that a refresh asks the
+ browser to repeat the POST; the page says to save the code before leaving.
+
+ ``Cache-Control: no-store`` is not optional here. The back button and a
+ shared machine's history are exactly how a credential outlives the moment
+ it was meant to be seen.
+
+ ``token`` is given when this response is also what starts the session.
+ """
+ response = render(request, "pages/recovery_code.html", recovery_code=code, reason=reason)
+ response.headers["Cache-Control"] = "no-store, private"
+ if token is not None and settings is not None:
+ _set_session_cookie(response, token, settings)
return response
@@ -138,12 +172,16 @@ def fail(message: str, *, allowed: bool = True) -> Response:
except EmailAlreadyRegistered:
return fail("That email is already registered.")
+ code = issue_recovery_code(user)
await db.commit()
token, _ = await create_session(redis, user.id, settings.session_ttl_seconds)
logger.info("user_registered", user_id=user.id, is_owner=user.is_owner)
- return _start_session_response("/", token, settings)
+ # Not a redirect to the dashboard: this is the one moment the code can be
+ # shown, and an owner who skips past it has no way back in short of the
+ # operator's key.
+ return show_recovery_code(request, code, reason="signup", token=token, settings=settings)
# ------------------------------------------------------------------ login ---
@@ -209,6 +247,97 @@ async def fail(message: str, code: int = status.HTTP_400_BAD_REQUEST) -> Respons
return _start_session_response("/", token, settings)
+# --------------------------------------------------------------- recovery ---
+
+#: One message for every way recovery can fail, for the reason
+#: INVALID_CREDENTIALS is one message: naming the wrong half would let anyone
+#: with the form discover which addresses have accounts.
+INVALID_RECOVERY = "That email and recovery code do not match."
+
+
+@router.get("/forgot-password", response_class=HTMLResponse)
+async def forgot_password_form(
+ request: Request,
+ current: Annotated[CurrentUser | None, Depends(get_optional_user)],
+) -> Response:
+ if current is not None:
+ return RedirectResponse(url="/settings", status_code=status.HTTP_303_SEE_OTHER)
+ return render(request, "pages/forgot_password.html", min_password_length=MIN_PASSWORD_LENGTH)
+
+
+@router.post("/forgot-password", response_class=HTMLResponse)
+async def forgot_password(
+ request: Request,
+ db: Annotated[AsyncSession, Depends(get_session)],
+ redis: Annotated[Redis, Depends(get_redis)],
+ settings: Annotated[Settings, Depends(get_app_settings)],
+ email: Annotated[str, Form()],
+ recovery_code: Annotated[str, Form()],
+ password: Annotated[str, Form()],
+) -> Response:
+ """Set a new password, given a recovery code or the operator's key.
+
+ No CSRF token, for the reason the login form has none: it is reachable
+ without a session, and all a forged request can do is attempt a recovery
+ for an address the attacker already knows, against their own limit.
+
+ A success signs the account out everywhere before starting a new session:
+ someone recovering an account may be doing so because it was taken, and the
+ sessions they are recovering from are the thing to end.
+ """
+
+ def fail(message: str, code: int = status.HTTP_400_BAD_REQUEST) -> Response:
+ return render(
+ request,
+ "pages/forgot_password.html",
+ min_password_length=MIN_PASSWORD_LENGTH,
+ error=message,
+ email=email,
+ status_code=code,
+ )
+
+ # Counted against the email alone - see RECOVERY_SCOPE for why not the address.
+ scope_email = normalise_email(email)
+
+ if await is_throttled(redis, scope_email, RECOVERY_SCOPE, settings.LOGIN_MAX_ATTEMPTS):
+ logger.warning("recovery_throttled")
+ return fail(
+ "Too many attempts. Wait a few minutes and try again.",
+ status.HTTP_429_TOO_MANY_REQUESTS,
+ )
+
+ try:
+ recovered = await recover_password(
+ db,
+ email=email,
+ proof=recovery_code,
+ new_password=password,
+ recovery_key=settings.RECOVERY_KEY,
+ )
+ except PasswordTooShort:
+ # About what was typed and nothing else, so it is not a failed attempt
+ # and says exactly what to change.
+ return fail(f"Use at least {MIN_PASSWORD_LENGTH} characters.")
+
+ if recovered is None:
+ await record_failure(
+ redis, scope_email, RECOVERY_SCOPE, settings.LOGIN_ATTEMPT_WINDOW_SECONDS
+ )
+ logger.info("recovery_failed")
+ return fail(INVALID_RECOVERY)
+
+ await db.commit()
+ await delete_user_sessions(redis, recovered.user.id)
+ await clear_attempts(redis, scope_email, RECOVERY_SCOPE)
+
+ token, _ = await create_session(redis, recovered.user.id, settings.session_ttl_seconds)
+ logger.info("password_recovered", user_id=recovered.user.id)
+
+ return show_recovery_code(
+ request, recovered.recovery_code, reason="recovered", token=token, settings=settings
+ )
+
+
# ----------------------------------------------------------------- logout ---
diff --git a/apps/api/src/seskit_api/templates/pages/forgot_password.html b/apps/api/src/seskit_api/templates/pages/forgot_password.html
new file mode 100644
index 0000000..65c8eba
--- /dev/null
+++ b/apps/api/src/seskit_api/templates/pages/forgot_password.html
@@ -0,0 +1,43 @@
+{% extends "base_auth.html" %}
+{% from "components/ui.html" import button, field %}
+
+{% block title %}Forgot password · SESKit{% endblock %}
+
+{% block content %}
+
+
Forgot password
+
+ Enter your email, the recovery code you were shown when you created the
+ account, and a new password.
+
+
+
+ {% if error %}
+
{{ error }}
+ {% endif %}
+
+ {# No CSRF token, for the reason the login form has none: there is no session
+ to bind one to, and a forged request can only attempt a recovery against
+ the attacker's own limit. #}
+
+
+
+ No recovery code? The person who runs this instance can set a
+ RECOVERY_KEY, which works in the same box.
+
once an owner exists the link would be a dead end, so it goes away. #}
{% if signup_allowed %}
Not set up yet? Create the owner account.
- {% else %}
- Self-hosted SESKit. Forgotten passwords are reset from the command line.
{% endif %}
+ {# Always offered, including while registration is open: ALLOW_SIGNUP can
+ reopen it on an instance that already has accounts. #}
+ Forgot your password?
{% endblock %}
diff --git a/apps/api/src/seskit_api/templates/pages/recovery_code.html b/apps/api/src/seskit_api/templates/pages/recovery_code.html
new file mode 100644
index 0000000..f8e879b
--- /dev/null
+++ b/apps/api/src/seskit_api/templates/pages/recovery_code.html
@@ -0,0 +1,62 @@
+{% extends "base_auth.html" %}
+{% from "components/ui.html" import button %}
+
+{% block title %}Your recovery code · SESKit{% endblock %}
+
+{# The one and only time this code is displayed - the same contract as a newly
+ created API key, and it borrows that component's look on purpose: a user who
+ has learnt that this red-bordered box means "copy it now" should not have to
+ learn it twice.
+
+ Three ways to arrive, told apart by `reason`, because what the reader has just
+ done decides what they need to hear first. #}
+
+{% block content %}
+
+ {% if reason == "signup" %}
+
Save your recovery code
+
+ Your account is created. Keep this code somewhere safe - it is the way
+ back in if you forget your password.
+
+ {% elif reason == "recovered" %}
+
Password changed
+
+ You are signed in. The code you just used is spent, so here is the
+ next one.
+
+ {% else %}
+
New recovery code
+
+ The old code no longer works. Keep this one somewhere safe.
+
+ {% endif %}
+
+
+
+
+ Recovery code
+ Copy it now
+
+
+
+ This is the only time it is shown. It is stored hashed, so it cannot be
+ looked up again - if you lose it you can make a new one from Settings,
+ while you can still sign in.
+
+
+
+ {{ recovery_code }}
+
+
+
+
+
+ Use it on the Forgot password page, with your email, to set
+ a new password. Each code works once. Do not reload this page before you
+ have saved it.
+
+ {% if has_recovery_code %}
+ {{ badge("Set", tone="success") }}
+ You have a recovery code. It lets you set a new password from the
+ Forgot password page if you ever lose
+ this one, without email.
+ {% else %}
+ {{ badge("Not set", tone="warning") }}
+ This account has no recovery code, so there is no way back in if you
+ forget your password. Make one now.
+ {% endif %}
+
+
+ Making a new code replaces the old one. A code is shown once and stored
+ only as a hash, so it cannot be looked up later.
+
+
+ {% if error %}
+
{{ error }}
+ {% endif %}
+
+ {# The password is asked for even though you are signed in: a session is
+ also what a stolen cookie looks like, and a new code would let whoever
+ holds it keep the account after you change your password. #}
+
+ {{ card_close() }}
+{% endblock %}
diff --git a/docs/getting-started/first-email.md b/docs/getting-started/first-email.md
index 16e9952..6e596ee 100644
--- a/docs/getting-started/first-email.md
+++ b/docs/getting-started/first-email.md
@@ -12,7 +12,8 @@ depended on that, most people would never get there.
## Three steps
1. Open and create the owner account. The first
- registration claims the instance; signup closes behind you.
+ registration claims the instance; signup closes behind you. Save the
+ recovery code it shows — it is the way back in if you forget your password.
2. Go to **API Keys**, create one, and copy it — **it is shown once**. SESKit
stores a SHA-256 hash, so a lost key cannot be recovered, only replaced.
3. Send:
diff --git a/docs/getting-started/installation.md b/docs/getting-started/installation.md
index 8988794..4669c30 100644
--- a/docs/getting-started/installation.md
+++ b/docs/getting-started/installation.md
@@ -142,6 +142,11 @@ an open sign-up page someone else can find.
Set `ALLOW_SIGNUP=true` if you want it to stay open.
+**Save the recovery code it shows you.** Right after registering, SESKit shows a
+code like `K7QD-4M2X-9PHB-TWE6`, once. It is how you get back in if you forget
+your password, without email and without a shell. See
+[I have forgotten my dashboard password](../operating/troubleshooting.md#i-have-forgotten-my-dashboard-password).
+
## What next
- [Your first email](first-email.md) — about a minute, still no AWS account.
diff --git a/docs/operating/troubleshooting.md b/docs/operating/troubleshooting.md
index 1439a4f..0b29361 100644
--- a/docs/operating/troubleshooting.md
+++ b/docs/operating/troubleshooting.md
@@ -102,6 +102,58 @@ in the selected range, and `0%` would assert something untrue.
If open and click read **"Not tracked"**, tracking is off for that project,
which is the default. See [reading your metrics](../guides/metrics.md).
+## I have forgotten my dashboard password
+
+There is no emailed reset link: a new instance may not be able to send email
+until somebody can sign in and connect AWS, which is when a password gets
+forgotten. And a deployment made by pushing to Git gives you no shell to run a
+command in. So the way back in is something you already hold.
+
+**With your recovery code.** When you created the account, SESKit showed a
+code like `K7QD-4M2X-9PHB-TWE6`, once. On the sign-in page choose **Forgot your
+password?**, then give your email, that code, and a new password. Capitals and
+hyphens are optional.
+
+- The code works **once**. Using it signs you in and shows the next one, so
+ save that too.
+- It signs the account out everywhere, which matters if you are recovering
+ because somebody else may have had access.
+- The new password must be at least 12 characters, the same as at sign-up.
+- After 10 wrong attempts for an email, recovery pauses for 15 minutes.
+ Signing in is not affected. The number and window are `LOGIN_MAX_ATTEMPTS` and
+ `LOGIN_ATTEMPT_WINDOW_SECONDS`.
+
+**If you have lost the code, or the account never had one** (accounts created
+before recovery codes existed have none), use the recovery key. Whoever runs
+the deployment sets `RECOVERY_KEY` in its environment, redeploys, and gives that
+value in the same box in place of a code. Make it at least 24 characters:
+
+```bash
+python -c "import secrets; print(secrets.token_urlsafe(32))"
+```
+
+An instance refuses to start if `RECOVERY_KEY` is set but shorter than that,
+rather than quietly ignoring it. The key is not used up by recovering, so remove
+it from the environment once you are back in, unless you want it kept as a
+standing way in. See [configuration](../reference/configuration.md).
+
+**Make sure you have a code.** Once signed in, **Settings** says whether the
+account has one and makes a new one, after asking for your password again. Do
+that now if you have an older account.
+
+**With a shell**, there is also a command, for instance on a machine you run
+yourself:
+
+```bash
+docker compose exec api seskit reset-password
+```
+
+It asks for the email and the new password at a hidden prompt, so the password
+stays out of your shell history, and signs the account out everywhere. It does
+not reactivate a deactivated account. Without Docker, run `seskit
+reset-password` from the environment the API runs in, so it reads the same
+`.env`.
+
## It will not start
| Symptom | Cause |
diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md
index d990c74..ac02bcd 100644
--- a/docs/reference/configuration.md
+++ b/docs/reference/configuration.md
@@ -116,8 +116,9 @@ so a value you find in the code has somewhere to be looked up.
| `ALLOW_SIGNUP` | `false` | Registration opens anyway while no account exists, then closes behind the first one |
| `SESSION_COOKIE_NAME` | `seskit_session` | |
| `SESSION_TTL_DAYS` | `14` | Idle timeout, not absolute — reading a session refreshes it, so an active user is never signed out mid-task |
-| `LOGIN_MAX_ATTEMPTS` | `10` | Failed logins per email and IP before the door closes for a while |
+| `LOGIN_MAX_ATTEMPTS` | `10` | Failed logins per email and IP before the door closes for a while. Also the number of failed [password recoveries](../operating/troubleshooting.md#i-have-forgotten-my-dashboard-password) per email |
| `LOGIN_ATTEMPT_WINDOW_SECONDS` | `900` | The window those attempts are counted in |
+| `RECOVERY_KEY` | *(empty)* | A secret that lets you set a new dashboard password for an account with no recovery code, or a lost one. At least 24 characters, or empty to turn it off. Anyone who can set it already controls the deployment, so it grants nothing they did not have; but treat it like `SECRET_KEY` |
### The public API
diff --git a/migrations/versions/47ac84b79629_add_recovery_codes.py b/migrations/versions/47ac84b79629_add_recovery_codes.py
new file mode 100644
index 0000000..e0dd2d4
--- /dev/null
+++ b/migrations/versions/47ac84b79629_add_recovery_codes.py
@@ -0,0 +1,36 @@
+"""add recovery codes
+
+One nullable column on `users`: the SHA-256 of the account's recovery code
+(§7). The code is how a forgotten dashboard password is recovered without email
+or a shell, and it is shown once, so only its hash is kept.
+
+NULL for every existing user, which is correct: they were never shown a code.
+They recover through the operator's `RECOVERY_KEY`, or by signing in and being
+issued one from Settings. The downgrade drops the column, and with it every
+code, which leaves those accounts exactly as they were before.
+
+Revision ID: 47ac84b79629
+Revises: c4e81a7d5b29
+Create Date: 2026-10-13 12:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "47ac84b79629"
+down_revision: str | Sequence[str] | None = "c4e81a7d5b29"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+ op.add_column(
+ "users",
+ sa.Column("recovery_code_hash", sa.String(length=64), nullable=True),
+ )
+
+
+def downgrade() -> None:
+ op.drop_column("users", "recovery_code_hash")
diff --git a/packages/core/src/seskit_core/config.py b/packages/core/src/seskit_core/config.py
index 7a48034..e121e8b 100644
--- a/packages/core/src/seskit_core/config.py
+++ b/packages/core/src/seskit_core/config.py
@@ -25,6 +25,11 @@
#: because it is a page a person opens, not an API a program calls.
UNSUBSCRIBE_PATH = "/u"
+#: Shortest ``RECOVERY_KEY`` that will be accepted (§7). Here and not in
+#: ``security.recovery`` because the settings validator needs it, and settings
+#: cannot import the security package without a cycle.
+MIN_RECOVERY_KEY_LENGTH = 24
+
# Placeholder used in .env.example. Refusing to boot on this value is what stops
# it reaching a real deployment.
INSECURE_PLACEHOLDER = "changeme"
@@ -96,6 +101,13 @@ class Settings(BaseSettings):
LOGIN_MAX_ATTEMPTS: int = 10
LOGIN_ATTEMPT_WINDOW_SECONDS: int = 900
+ #: A secret the operator sets so a forgotten dashboard password can be
+ #: recovered on an account that has no recovery code, or whose code is lost
+ #: (§7). Presented on the forgot-password page together with the account's
+ #: email. Empty turns the feature off; an empty value is how most platforms
+ #: represent "declared, not set", so it must not mean "accept anything".
+ RECOVERY_KEY: str = ""
+
# -- Public API (§7, §20) ------------------------------------------------
#: How long a verified key stays resolved in Redis. Short, and only a
@@ -421,6 +433,21 @@ def _require_async_driver(self) -> Self:
)
return self
+ @model_validator(mode="after")
+ def _require_a_strong_recovery_key(self) -> Self:
+ """A short key is a guessable password to every account.
+
+ Refused at boot rather than ignored: a key that silently does nothing
+ leaves the operator believing they have a way back in.
+ """
+ if self.RECOVERY_KEY and len(self.RECOVERY_KEY) < MIN_RECOVERY_KEY_LENGTH:
+ raise ValueError(
+ f"RECOVERY_KEY must be at least {MIN_RECOVERY_KEY_LENGTH} characters, "
+ "or empty to turn recovery by key off. "
+ "Generate one: python -c 'import secrets; print(secrets.token_urlsafe(32))'"
+ )
+ return self
+
@model_validator(mode="after")
def _reject_insecure_defaults(self) -> Self:
"""Refuse to boot outside local with placeholder secrets still in place."""
diff --git a/packages/core/src/seskit_core/models/user.py b/packages/core/src/seskit_core/models/user.py
index 141a56c..65117af 100644
--- a/packages/core/src/seskit_core/models/user.py
+++ b/packages/core/src/seskit_core/models/user.py
@@ -38,6 +38,12 @@ class User(Base, TimestampMixin):
#: Argon2id. Never the password itself (§22).
password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
+ #: SHA-256 of the account's recovery code, or NULL when it has none (§7).
+ #: The code itself exists only on the page that shows it, once. NULL is every
+ #: account created before recovery codes existed, and those recover through
+ #: the operator's ``RECOVERY_KEY`` or by being issued a code from Settings.
+ recovery_code_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
+
#: The first account on an install. Set once, at registration.
is_owner: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
diff --git a/packages/core/src/seskit_core/security/recovery.py b/packages/core/src/seskit_core/security/recovery.py
new file mode 100644
index 0000000..bea62fc
--- /dev/null
+++ b/packages/core/src/seskit_core/security/recovery.py
@@ -0,0 +1,112 @@
+"""Recovery codes and the operator's recovery key (§7, §22).
+
+A forgotten dashboard password cannot be solved with email: a fresh instance
+may be unable to send any until somebody can sign in and connect AWS, and that
+is when a password gets forgotten. It cannot be solved with a shell either,
+because the platforms people deploy to by pushing to Git do not give one. So the
+way back in is something the account holder already holds.
+
+**The code.** Sixteen characters from a 32 character alphabet is 80 bits, shown
+as ``K7QD-4M2X-9PHB-TWE6``. It is shown once, at sign-up, and stored only as a
+hash. Using it replaces it: a recovery code that survived being spent would be a
+password with no expiry.
+
+**SHA-256, not Argon2id,** for the reason ``api_keys.py`` gives at length and
+which holds here for the same arithmetic: Argon2 makes a low-entropy secret
+expensive to guess, and 80 random bits are not guessable. The attempt limit in
+``throttle.py`` is what bounds an online guess, and an offline one against a
+stolen database is 2^80 hashes.
+
+**The alphabet** leaves out the characters people misread - ``I``, ``L``, ``O``
+and ``U`` - and :func:`normalise_code` maps the look-alikes onto what was meant,
+so a code copied by eye from paper or typed in lowercase still works. It also
+drops spaces and hyphens, which chat clients and password managers add and
+remove freely.
+
+**The key** is a different thing: one long secret the *operator* puts in the
+deployment's environment as ``RECOVERY_KEY``. It exists because a recovery code
+only helps an account that has one, and every account created before this
+feature does not. Whoever controls the environment controls the instance, so
+accepting it as proof of ownership grants nothing they did not already hold.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import secrets
+
+from seskit_core.config import MIN_RECOVERY_KEY_LENGTH
+
+#: Crockford's base32: digits and letters without ``I``, ``L``, ``O`` and ``U``.
+ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+
+#: Characters of entropy in a code. 16 x 5 bits = 80 bits.
+CODE_LENGTH = 16
+
+#: Size of the hyphen-separated groups, for reading aloud and copying by eye.
+GROUP_SIZE = 4
+
+#: What a mistyped character is most likely to have been.
+_LOOKALIKES = str.maketrans({"O": "0", "I": "1", "L": "1", "U": "V"})
+
+
+__all__ = [
+ "ALPHABET",
+ "CODE_LENGTH",
+ "MIN_RECOVERY_KEY_LENGTH",
+ "code_matches",
+ "generate_code",
+ "hash_code",
+ "key_matches",
+ "normalise_code",
+]
+
+
+def generate_code() -> str:
+ """A new recovery code, grouped for display.
+
+ ``secrets`` rather than ``random``, as for every credential here.
+ """
+ raw = "".join(secrets.choice(ALPHABET) for _ in range(CODE_LENGTH))
+ return "-".join(raw[i : i + GROUP_SIZE] for i in range(0, CODE_LENGTH, GROUP_SIZE))
+
+
+def normalise_code(value: str) -> str:
+ """What a person typed, reduced to the form that is hashed.
+
+ Uppercase, without separators, with look-alikes resolved. Applied to both
+ sides - once when the code is issued, once when it is presented - so the two
+ cannot disagree about what counts as the same code.
+ """
+ kept = "".join(ch for ch in value.upper() if ch.isalnum())
+ return kept.translate(_LOOKALIKES)
+
+
+def hash_code(code: str) -> str:
+ """The SHA-256 hex digest stored for a code."""
+ return hashlib.sha256(normalise_code(code).encode("utf-8")).hexdigest()
+
+
+def code_matches(code: str, stored_hash: str | None) -> bool:
+ """Whether ``code`` is the one ``stored_hash`` was made from.
+
+ Constant-time. Takes a missing hash and answers False after the same work,
+ so an account with no code and one with the wrong code look the same.
+ """
+ expected = stored_hash if stored_hash is not None else "0" * 64
+ matched = hmac.compare_digest(hash_code(code), expected)
+ return matched and stored_hash is not None
+
+
+def key_matches(presented: str, configured: str) -> bool:
+ """Whether ``presented`` is the operator's recovery key.
+
+ Always False when no key is configured, whatever was presented - an empty
+ string must not match an empty setting. Compared as bytes: ``compare_digest``
+ raises ``TypeError`` on non-ASCII ``str``, which would turn a hostile form
+ field into a 500.
+ """
+ if not configured:
+ return False
+ return hmac.compare_digest(presented.strip().encode("utf-8"), configured.encode("utf-8"))
diff --git a/packages/core/src/seskit_core/security/throttle.py b/packages/core/src/seskit_core/security/throttle.py
index 322bfe8..6e64f85 100644
--- a/packages/core/src/seskit_core/security/throttle.py
+++ b/packages/core/src/seskit_core/security/throttle.py
@@ -18,6 +18,20 @@
THROTTLE_KEY_PREFIX = "login_attempts:"
+#: Stands in for a client address when recovery attempts are counted (§7).
+#:
+#: Recovery is limited per **email alone**, where login is per email *and*
+#: address, and the difference is deliberate. The address is whatever the
+#: caller's ``X-Forwarded-For`` says it is, so a limit that includes it is a
+#: limit the attacker chooses the size of. Keyed on the email alone, an
+#: attacker cannot buy more guesses; what it costs is that they can use up the
+#: allowance for a known email and delay that person's recovery for one window.
+#: That is the right way round for a limit that guards account takeover: signing
+#: in is unaffected, and the delay ends by itself.
+#:
+#: It cannot collide with a real address, so the two never share a counter.
+RECOVERY_SCOPE = "recovery"
+
def _key(email: str, client: str) -> str:
return f"{THROTTLE_KEY_PREFIX}{email}:{client}"
diff --git a/packages/core/src/seskit_core/services/__init__.py b/packages/core/src/seskit_core/services/__init__.py
index f3557ef..c95833d 100644
--- a/packages/core/src/seskit_core/services/__init__.py
+++ b/packages/core/src/seskit_core/services/__init__.py
@@ -103,13 +103,20 @@
suppressed_among,
)
from seskit_core.services.users import (
+ MIN_PASSWORD_LENGTH,
EmailAlreadyRegistered,
+ PasswordTooShort,
+ Recovered,
SignupClosed,
authenticate,
count_users,
get_user_by_email,
get_user_by_id,
+ issue_recovery_code,
+ recover_password,
+ regenerate_recovery_code,
register_user,
+ reset_password,
signup_allowed,
)
from seskit_core.services.webhooks import (
@@ -134,6 +141,7 @@
__all__ = [
"ACKNOWLEDGEMENT",
"MAX_CONTACTS",
+ "MIN_PASSWORD_LENGTH",
"ActivityPoint",
"EmailAlreadyRegistered",
"InboundProvisionerFactory",
@@ -141,11 +149,13 @@
"IssuedKey",
"Metrics",
"Outgoing",
+ "PasswordTooShort",
"PolledInbox",
"PolledQueue",
"ProviderFactory",
"ProvisionerFactory",
"Readiness",
+ "Recovered",
"SetupStep",
"SignupClosed",
"TimeRange",
@@ -192,6 +202,7 @@
"is_complete",
"is_recheck_due",
"is_retryable",
+ "issue_recovery_code",
"list_api_keys",
"list_deliveries",
"list_endpoints",
@@ -209,13 +220,16 @@
"record_delivery_success",
"record_failure",
"record_sent",
+ "recover_password",
"refresh_connection",
"refresh_identity",
"refuse_suppressed",
+ "regenerate_recovery_code",
"register_user",
"remove_identity",
"remove_suppression",
"request_production_access",
+ "reset_password",
"revoke_api_key",
"rotate_secret",
"rule_name_for",
diff --git a/packages/core/src/seskit_core/services/users.py b/packages/core/src/seskit_core/services/users.py
index 863114c..4bd8246 100644
--- a/packages/core/src/seskit_core/services/users.py
+++ b/packages/core/src/seskit_core/services/users.py
@@ -6,11 +6,29 @@
from __future__ import annotations
+from dataclasses import dataclass
+
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from seskit_core.models import DEFAULT_PROJECT_NAME, Project, User, normalise_email
-from seskit_core.security.passwords import burn_dummy_hash, hash_password, verify_and_update
+from seskit_core.security.passwords import (
+ burn_dummy_hash,
+ hash_password,
+ verify_and_update,
+ verify_password,
+)
+from seskit_core.security.recovery import code_matches, generate_code, hash_code, key_matches
+
+#: Long enough to be worth hashing. OWASP's floor is 8; this is an email
+#: platform holding AWS access, so it asks for a little more. Here rather than in
+#: a route because the sign-up form and the command line must agree on it - a
+#: password the dashboard would refuse is not one a reset should accept.
+MIN_PASSWORD_LENGTH = 12
+
+
+class PasswordTooShort(Exception):
+ """A password below :data:`MIN_PASSWORD_LENGTH`."""
class SignupClosed(Exception):
@@ -118,3 +136,112 @@ async def authenticate(session: AsyncSession, *, email: str, password: str) -> U
await session.flush()
return user
+
+
+async def reset_password(session: AsyncSession, *, email: str, password: str) -> User | None:
+ """Set a new password for an account, or return None when there is none.
+
+ For the operator, not the account holder: there is no emailed reset link,
+ because a self-hosted instance may be unable to send email until somebody
+ can log in and connect AWS - the situation a forgotten password is most
+ likely to arrive in. Whoever can run this already holds the database and
+ ``SECRET_KEY``, so naming an unknown address here discloses nothing.
+
+ Flushes and does not commit; the caller owns the transaction. It also does
+ not end sessions, which live in Redis - see
+ :func:`seskit_core.security.sessions.delete_user_sessions`.
+ """
+ if len(password) < MIN_PASSWORD_LENGTH:
+ raise PasswordTooShort
+
+ user = await get_user_by_email(session, email)
+ if user is None:
+ return None
+
+ user.password_hash = hash_password(password)
+ await session.flush()
+ return user
+
+
+# --------------------------------------------------------------- recovery ---
+
+
+@dataclass(frozen=True, slots=True)
+class Recovered:
+ """A successful recovery: who, and the replacement code to show them once."""
+
+ user: User
+ recovery_code: str
+
+
+def issue_recovery_code(user: User) -> str:
+ """Give an account a new recovery code, replacing any it had.
+
+ Returns the code, which is the only moment it exists in the clear; the row
+ keeps its hash. The caller flushes or commits.
+ """
+ code = generate_code()
+ user.recovery_code_hash = hash_code(code)
+ return code
+
+
+async def recover_password(
+ session: AsyncSession,
+ *,
+ email: str,
+ proof: str,
+ new_password: str,
+ recovery_key: str = "",
+) -> Recovered | None:
+ """Set a new password for someone who can prove they own the account.
+
+ ``proof`` is either the account's recovery code or the operator's recovery
+ key, accepted from the same field so the page asks for one thing. Returns
+ None for every way of being wrong - no such account, a deactivated one, no
+ code on file, a wrong code, a wrong key - so a caller cannot tell them
+ apart and the page cannot be used to find out which addresses have accounts.
+
+ Spending a code issues the next one, in the same step: an account must not
+ be left without a way back in by having used it. The operator's key is not
+ spent, because it is not the account's.
+
+ Raises :class:`PasswordTooShort` before looking at anything else, which
+ reveals nothing: it depends only on what the caller typed.
+
+ Flushes and does not commit. Sessions live in Redis and are the caller's to
+ end.
+ """
+ if len(new_password) < MIN_PASSWORD_LENGTH:
+ raise PasswordTooShort
+
+ user = await get_user_by_email(session, email)
+
+ # Both checks always run, whatever the first says, and run against a stand-in
+ # when there is no account: the cost of a wrong answer must not depend on
+ # which half was wrong.
+ code_ok = code_matches(proof, user.recovery_code_hash if user else None)
+ key_ok = key_matches(proof, recovery_key)
+
+ if user is None or not user.is_active or not (code_ok or key_ok):
+ return None
+
+ user.password_hash = hash_password(new_password)
+ next_code = issue_recovery_code(user)
+ await session.flush()
+ return Recovered(user=user, recovery_code=next_code)
+
+
+async def regenerate_recovery_code(
+ session: AsyncSession, user: User, *, password: str
+) -> str | None:
+ """Replace an account's recovery code, after the current password is given.
+
+ The password is asked for because a signed-in session is not proof of
+ ownership - it is also what a stolen cookie looks like, and a new code would
+ let the thief take the account for good. None when the password is wrong.
+ """
+ if not verify_password(password, user.password_hash):
+ return None
+ code = issue_recovery_code(user)
+ await session.flush()
+ return code
diff --git a/tests/conftest.py b/tests/conftest.py
index 46fe3f9..c1e3a91 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -345,11 +345,14 @@ async def signed_in_client(app_client: AsyncClient) -> AsyncClient:
Since Phase 2 the dashboard is not reachable while signed out, so a test
about how a page renders needs an account before it can see a page at all.
+
+ Signing up answers with the page that shows the recovery code, so it is a
+ 200 and not the redirect it once was; the session cookie is on that response.
"""
response = await app_client.post(
"/signup",
data={"email": "owner@example.com", "password": "correct-horse-battery"},
follow_redirects=False,
)
- assert response.status_code == 303, response.text
+ assert response.status_code == 200, response.text
return app_client
diff --git a/tests/test_api_keys.py b/tests/test_api_keys.py
index 114db33..1010810 100644
--- a/tests/test_api_keys.py
+++ b/tests/test_api_keys.py
@@ -378,7 +378,7 @@ async def _sign_in(client: AsyncClient, email: str = "owner@example.com") -> Non
response = await client.post(
"/signup", data={"email": email, "password": PASSWORD}, follow_redirects=False
)
- assert response.status_code == 303, response.text
+ assert response.status_code == 200, response.text
def _csrf(html: str) -> str:
diff --git a/tests/test_auth.py b/tests/test_auth.py
index 7c0922c..5115e71 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -17,7 +17,7 @@ async def _sign_in(client: AsyncClient, email: str = "owner@example.com") -> Non
response = await client.post(
"/signup", data={"email": email, "password": PASSWORD}, follow_redirects=False
)
- assert response.status_code == 303, response.text
+ assert response.status_code == 200, response.text
# --------------------------------------------------------- access control ---
diff --git a/tests/test_aws.py b/tests/test_aws.py
index a7f4d92..28c53b6 100644
--- a/tests/test_aws.py
+++ b/tests/test_aws.py
@@ -55,7 +55,7 @@ async def _sign_in(client: AsyncClient, email: str = "owner@example.com") -> Non
response = await client.post(
"/signup", data={"email": email, "password": PASSWORD}, follow_redirects=False
)
- assert response.status_code == 303, response.text
+ assert response.status_code == 200, response.text
async def _csrf(client: AsyncClient) -> str:
diff --git a/tests/test_cli.py b/tests/test_cli.py
new file mode 100644
index 0000000..f2bdb3a
--- /dev/null
+++ b/tests/test_cli.py
@@ -0,0 +1,159 @@
+"""`seskit reset-password` - the way back in after a forgotten password.
+
+The login page promises that forgotten passwords are reset from the command
+line. What is worth testing is the part that decides things: that the new
+password works and the old one does not, that nobody is left signed in on a
+session opened under the old one, that a short password changes nothing, and
+that an unknown address is reported rather than invented.
+
+The prompts are checked without a database; the reset itself needs one.
+"""
+
+from __future__ import annotations
+
+import pytest
+from redis.asyncio import Redis
+from seskit_api import cli
+from seskit_core.security.sessions import create_session, read_session
+from seskit_core.services import (
+ MIN_PASSWORD_LENGTH,
+ PasswordTooShort,
+ authenticate,
+ register_user,
+)
+from sqlalchemy.ext.asyncio import AsyncSession
+
+OLD = "the-old-password-1"
+NEW = "a-brand-new-password-2"
+
+
+async def _user(session: AsyncSession, email: str = "owner@example.com"): # type: ignore[no-untyped-def]
+ return await register_user(session, email=email, password=OLD, allow_signup=True)
+
+
+# ------------------------------------------------------------- the reset ---
+
+
+async def test_the_new_password_works_and_the_old_one_does_not(
+ db_session: AsyncSession, redis_client: Redis
+) -> None:
+ await _user(db_session)
+
+ result = await cli.reset_account_password(
+ db_session, redis_client, email="owner@example.com", password=NEW
+ )
+
+ assert result is not None
+ assert await authenticate(db_session, email="owner@example.com", password=NEW) is not None
+ assert await authenticate(db_session, email="owner@example.com", password=OLD) is None
+
+
+async def test_the_address_is_matched_the_way_login_matches_it(
+ db_session: AsyncSession, redis_client: Redis
+) -> None:
+ """Capitalisation and stray spaces are what a person types at a prompt."""
+ await _user(db_session)
+
+ result = await cli.reset_account_password(
+ db_session, redis_client, email=" Owner@Example.COM ", password=NEW
+ )
+
+ assert result is not None
+
+
+async def test_sessions_opened_under_the_old_password_are_ended(
+ db_session: AsyncSession, redis_client: Redis
+) -> None:
+ user = await _user(db_session)
+ first, _ = await create_session(redis_client, user.id, 3600)
+ second, _ = await create_session(redis_client, user.id, 3600)
+
+ result = await cli.reset_account_password(
+ db_session, redis_client, email=user.email, password=NEW
+ )
+
+ assert result is not None
+ assert result.sessions_ended == 2
+ assert await read_session(redis_client, first, 3600) is None
+ assert await read_session(redis_client, second, 3600) is None
+
+
+async def test_another_accounts_sessions_are_left_alone(
+ db_session: AsyncSession, redis_client: Redis
+) -> None:
+ target = await _user(db_session)
+ other = await _user(db_session, "other@example.com")
+ token, _ = await create_session(redis_client, other.id, 3600)
+
+ await cli.reset_account_password(db_session, redis_client, email=target.email, password=NEW)
+
+ assert await read_session(redis_client, token, 3600) is not None
+
+
+async def test_an_unknown_address_is_reported_not_created(
+ db_session: AsyncSession, redis_client: Redis
+) -> None:
+ result = await cli.reset_account_password(
+ db_session, redis_client, email="nobody@example.com", password=NEW
+ )
+
+ assert result is None
+
+
+async def test_a_short_password_changes_nothing(
+ db_session: AsyncSession, redis_client: Redis
+) -> None:
+ user = await _user(db_session)
+ token, _ = await create_session(redis_client, user.id, 3600)
+
+ with pytest.raises(PasswordTooShort):
+ await cli.reset_account_password(
+ db_session, redis_client, email=user.email, password="x" * (MIN_PASSWORD_LENGTH - 1)
+ )
+
+ assert await authenticate(db_session, email=user.email, password=OLD) is not None
+ assert await read_session(redis_client, token, 3600) is not None
+
+
+async def test_a_deactivated_account_is_reset_but_not_reactivated(
+ db_session: AsyncSession, redis_client: Redis
+) -> None:
+ user = await _user(db_session)
+ user.is_active = False
+ await db_session.flush()
+
+ result = await cli.reset_account_password(
+ db_session, redis_client, email=user.email, password=NEW
+ )
+
+ assert result is not None
+ assert result.is_active is False
+ assert await authenticate(db_session, email=user.email, password=NEW) is None
+
+
+# ------------------------------------------------------------ the prompts ---
+
+
+def test_two_different_passwords_change_nothing_and_never_reach_the_database(
+ monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+ answers = iter(["first-password-here", "a-different-one"])
+ monkeypatch.setattr("getpass.getpass", lambda prompt="": next(answers))
+
+ code = cli.main(["reset-password", "--email", "owner@example.com"])
+
+ assert code == cli.EXIT_USAGE
+ assert "differ" in capsys.readouterr().err
+
+
+def test_the_password_is_never_an_argument() -> None:
+ """On the command line it would be in shell history and `ps`."""
+ parser = cli.build_parser()
+
+ with pytest.raises(SystemExit):
+ parser.parse_args(["reset-password", "--password", "hunter2hunter2"])
+
+
+def test_a_command_is_required() -> None:
+ with pytest.raises(SystemExit):
+ cli.build_parser().parse_args([])
diff --git a/tests/test_domains_page.py b/tests/test_domains_page.py
index 7515c29..7a41fa3 100644
--- a/tests/test_domains_page.py
+++ b/tests/test_domains_page.py
@@ -25,7 +25,7 @@ async def _sign_in(client: AsyncClient, email: str = "owner@example.com") -> Non
response = await client.post(
"/signup", data={"email": email, "password": PASSWORD}, follow_redirects=False
)
- assert response.status_code == 303, response.text
+ assert response.status_code == 200, response.text
async def _csrf(client: AsyncClient) -> str:
diff --git a/tests/test_emails_page.py b/tests/test_emails_page.py
index ea39471..13bd846 100644
--- a/tests/test_emails_page.py
+++ b/tests/test_emails_page.py
@@ -34,7 +34,7 @@ async def _sign_in(client: AsyncClient, email: str = "owner@example.com") -> Non
response = await client.post(
"/signup", data={"email": email, "password": PASSWORD}, follow_redirects=False
)
- assert response.status_code == 303, response.text
+ assert response.status_code == 200, response.text
async def _key_for_signed_in_project(client: AsyncClient, session: AsyncSession) -> str:
diff --git a/tests/test_recovery.py b/tests/test_recovery.py
new file mode 100644
index 0000000..34da471
--- /dev/null
+++ b/tests/test_recovery.py
@@ -0,0 +1,581 @@
+"""Recovering a forgotten dashboard password without email or a shell (§7).
+
+The login page cannot ask for an emailed link, because a fresh instance may not
+be able to send one yet, and a deployment made by pushing to Git offers no shell
+to run a command in. So the way back in is something the owner already holds: a
+recovery code shown once at sign-up, or a key the operator puts in the
+environment.
+
+What is worth testing is not that a form posts. It is that the code is spent
+when it is used and replaced in the same step, that a wrong answer looks the same
+whichever half was wrong, that the limit holds when the caller lies about their
+address, and that a stolen session cannot mint a code without the password.
+
+The first group needs no database.
+"""
+
+from __future__ import annotations
+
+import re
+
+import pytest
+from httpx import AsyncClient
+from pydantic import ValidationError
+from redis.asyncio import Redis
+from seskit_core.config import MIN_RECOVERY_KEY_LENGTH, Settings
+from seskit_core.security.recovery import (
+ ALPHABET,
+ CODE_LENGTH,
+ code_matches,
+ generate_code,
+ hash_code,
+ key_matches,
+ normalise_code,
+)
+from seskit_core.security.sessions import create_session, read_session
+from seskit_core.services import (
+ MIN_PASSWORD_LENGTH,
+ PasswordTooShort,
+ authenticate,
+ get_user_by_email,
+ issue_recovery_code,
+ recover_password,
+ regenerate_recovery_code,
+ register_user,
+)
+from sqlalchemy.ext.asyncio import AsyncSession
+
+OLD = "the-old-password-1"
+NEW = "a-brand-new-password-2"
+EMAIL = "owner@example.com"
+KEY = "k" * MIN_RECOVERY_KEY_LENGTH
+CODE_RE = re.compile(r"[0-9A-Z]{4}(?:-[0-9A-Z]{4}){3}")
+
+
+# ------------------------------------------------------------- the code ---
+
+
+def test_a_code_is_sixteen_unambiguous_characters_in_groups() -> None:
+ code = generate_code()
+
+ assert CODE_RE.fullmatch(code)
+ assert len(code.replace("-", "")) == CODE_LENGTH
+ assert set(code.replace("-", "")) <= set(ALPHABET)
+
+
+def test_a_code_never_contains_a_character_people_misread() -> None:
+ assert not set("ILOU") & set(ALPHABET)
+
+
+def test_codes_are_not_repeated() -> None:
+ assert len({generate_code() for _ in range(500)}) == 500
+
+
+@pytest.mark.parametrize(
+ "retyped",
+ [
+ lambda c: c.lower(),
+ lambda c: c.replace("-", " "),
+ lambda c: c.replace("-", ""),
+ lambda c: f" {c}\n",
+ ],
+ ids=["lowercase", "spaces", "no separators", "padded"],
+)
+def test_a_code_typed_by_hand_is_read_the_way_it_was_meant(retyped) -> None: # type: ignore[no-untyped-def]
+ """Case, spacing and hyphens are what chat clients and paper change."""
+ code = generate_code()
+
+ assert hash_code(retyped(code)) == hash_code(code)
+
+
+def test_look_alike_letters_resolve_to_the_character_intended() -> None:
+ """The alphabet has no O, I, L or U, so one typed is a misreading of the
+ digit or letter that is: O is 0, I and L are 1, U is V."""
+ assert normalise_code("O0IL1U") == "00111V"
+
+
+def test_two_different_codes_never_normalise_alike() -> None:
+ assert hash_code("0000-0000-0000-0000") != hash_code("0000-0000-0000-0001")
+
+
+def test_a_code_matches_only_its_own_hash() -> None:
+ code = generate_code()
+
+ assert code_matches(code, hash_code(code))
+ assert not code_matches(generate_code(), hash_code(code))
+
+
+def test_no_hash_on_file_is_never_a_match() -> None:
+ """An account with no code and one with the wrong code look the same."""
+ assert not code_matches(generate_code(), None)
+ assert not code_matches("", None)
+ assert not code_matches("0" * CODE_LENGTH, None)
+
+
+# -------------------------------------------------------------- the key ---
+
+
+def test_the_key_matches_itself_and_only_itself() -> None:
+ assert key_matches(KEY, KEY)
+ assert key_matches(f" {KEY}\n", KEY)
+ assert not key_matches(KEY + "x", KEY)
+
+
+def test_nothing_matches_an_unset_key() -> None:
+ """An empty form field must not equal an empty setting."""
+ assert not key_matches("", "")
+ assert not key_matches("anything", "")
+
+
+def test_a_key_with_non_ascii_text_is_a_miss_not_a_crash() -> None:
+ """`hmac.compare_digest` raises on non-ASCII `str`, which would turn a
+ hostile form field into a 500."""
+ assert not key_matches("ключ-восстановления-" * 3, KEY)
+
+
+def _settings(**overrides: str) -> Settings:
+ """Settings from explicit values only, as `test_config.py` builds them.
+
+ `RECOVERY_KEY` is cleared in the base so an operator's real one in the
+ environment running the tests cannot change what these assert.
+ """
+ base = {
+ "SECRET_KEY": "a-real-secret",
+ "DATABASE_URL": "postgresql+asyncpg://u:p@localhost:5432/db",
+ "REDIS_URL": "redis://localhost:6379/0",
+ "RECOVERY_KEY": "",
+ }
+ return Settings(_env_file=None, **(base | overrides)) # type: ignore[arg-type]
+
+
+def test_recovery_by_key_is_off_unless_one_is_set() -> None:
+ assert _settings().RECOVERY_KEY == ""
+
+
+def test_a_short_recovery_key_stops_the_instance_booting() -> None:
+ """Refused rather than ignored: a key that silently does nothing leaves the
+ operator believing they have a way back in."""
+ with pytest.raises(ValidationError, match="RECOVERY_KEY"):
+ _settings(RECOVERY_KEY="short")
+
+
+def test_a_long_enough_recovery_key_is_accepted() -> None:
+ assert _settings(RECOVERY_KEY=KEY).RECOVERY_KEY == KEY
+
+
+# ------------------------------------------------------------ the service ---
+
+
+async def _account(session: AsyncSession, *, with_code: bool = True) -> tuple[str, str | None]:
+ user = await register_user(session, email=EMAIL, password=OLD, allow_signup=True)
+ code = issue_recovery_code(user) if with_code else None
+ await session.flush()
+ return user.id, code
+
+
+async def test_the_code_sets_a_new_password_and_the_old_one_stops_working(
+ db_session: AsyncSession,
+) -> None:
+ _, code = await _account(db_session)
+ assert code is not None
+
+ recovered = await recover_password(db_session, email=EMAIL, proof=code, new_password=NEW)
+
+ assert recovered is not None
+ assert await authenticate(db_session, email=EMAIL, password=NEW) is not None
+ assert await authenticate(db_session, email=EMAIL, password=OLD) is None
+
+
+async def test_using_a_code_replaces_it_in_the_same_step(db_session: AsyncSession) -> None:
+ """Otherwise recovering would leave the account with no way to do it again."""
+ _, code = await _account(db_session)
+ assert code is not None
+
+ recovered = await recover_password(db_session, email=EMAIL, proof=code, new_password=NEW)
+
+ assert recovered is not None
+ assert recovered.recovery_code != code
+ assert CODE_RE.fullmatch(recovered.recovery_code)
+
+ again = await recover_password(
+ db_session, email=EMAIL, proof=code, new_password="another-new-password-3"
+ )
+ assert again is None, "a spent code must not work twice"
+
+ fresh = await recover_password(
+ db_session,
+ email=EMAIL,
+ proof=recovered.recovery_code,
+ new_password="another-new-password-3",
+ )
+ assert fresh is not None
+
+
+async def test_the_code_is_stored_only_as_a_hash(db_session: AsyncSession) -> None:
+ _, code = await _account(db_session)
+ assert code is not None
+
+ user = await get_user_by_email(db_session, EMAIL)
+
+ assert user is not None
+ assert user.recovery_code_hash == hash_code(code)
+ assert code not in (user.recovery_code_hash or "")
+ assert code not in repr(user)
+
+
+async def test_a_code_is_accepted_however_it_was_typed(db_session: AsyncSession) -> None:
+ _, code = await _account(db_session)
+ assert code is not None
+
+ recovered = await recover_password(
+ db_session,
+ email=f" {EMAIL.upper()} ",
+ proof=code.lower().replace("-", " "),
+ new_password=NEW,
+ )
+
+ assert recovered is not None
+
+
+async def test_the_operator_key_recovers_an_account_that_has_no_code(
+ db_session: AsyncSession,
+) -> None:
+ """Every account made before recovery codes existed is in this position."""
+ await _account(db_session, with_code=False)
+
+ recovered = await recover_password(
+ db_session, email=EMAIL, proof=KEY, new_password=NEW, recovery_key=KEY
+ )
+
+ assert recovered is not None
+ assert await authenticate(db_session, email=EMAIL, password=NEW) is not None
+ assert CODE_RE.fullmatch(recovered.recovery_code), "and leaves the account with a code"
+
+
+async def test_the_operator_key_is_not_spent_by_use(db_session: AsyncSession) -> None:
+ await _account(db_session, with_code=False)
+
+ first = await recover_password(
+ db_session, email=EMAIL, proof=KEY, new_password=NEW, recovery_key=KEY
+ )
+ second = await recover_password(
+ db_session, email=EMAIL, proof=KEY, new_password="yet-another-password-4", recovery_key=KEY
+ )
+
+ assert first is not None and second is not None
+
+
+async def test_the_key_means_nothing_when_none_is_configured(db_session: AsyncSession) -> None:
+ await _account(db_session, with_code=False)
+
+ assert (
+ await recover_password(
+ db_session, email=EMAIL, proof=KEY, new_password=NEW, recovery_key=""
+ )
+ is None
+ )
+ assert (
+ await recover_password(db_session, email=EMAIL, proof="", new_password=NEW, recovery_key="")
+ is None
+ )
+
+
+@pytest.mark.parametrize("kind", ["wrong code", "unknown account", "deactivated"])
+async def test_every_way_of_being_wrong_gets_the_same_answer(
+ db_session: AsyncSession, kind: str
+) -> None:
+ """None for all of them, so the form cannot be used to learn which
+ addresses have accounts."""
+ user_id, code = await _account(db_session)
+ assert code is not None
+ email, proof = EMAIL, code
+
+ if kind == "wrong code":
+ proof = generate_code()
+ elif kind == "unknown account":
+ email = "nobody@example.com"
+ else:
+ user = await get_user_by_email(db_session, EMAIL)
+ assert user is not None and user.id == user_id
+ user.is_active = False
+ await db_session.flush()
+
+ assert await recover_password(db_session, email=email, proof=proof, new_password=NEW) is None
+ assert await authenticate(db_session, email=EMAIL, password=NEW) is None
+
+
+async def test_a_short_password_changes_nothing_and_does_not_spend_the_code(
+ db_session: AsyncSession,
+) -> None:
+ _, code = await _account(db_session)
+ assert code is not None
+
+ with pytest.raises(PasswordTooShort):
+ await recover_password(
+ db_session, email=EMAIL, proof=code, new_password="x" * (MIN_PASSWORD_LENGTH - 1)
+ )
+
+ assert await authenticate(db_session, email=EMAIL, password=OLD) is not None
+ assert await recover_password(db_session, email=EMAIL, proof=code, new_password=NEW)
+
+
+async def test_regenerating_needs_the_current_password(db_session: AsyncSession) -> None:
+ """A session is also what a stolen cookie looks like."""
+ _, code = await _account(db_session)
+ user = await get_user_by_email(db_session, EMAIL)
+ assert user is not None and code is not None
+
+ assert await regenerate_recovery_code(db_session, user, password="not-the-password") is None
+ assert user.recovery_code_hash == hash_code(code)
+
+ new = await regenerate_recovery_code(db_session, user, password=OLD)
+
+ assert new is not None and new != code
+ assert user.recovery_code_hash == hash_code(new)
+ assert not code_matches(code, user.recovery_code_hash), "the old code is dead"
+
+
+# --------------------------------------------------------------- the pages ---
+
+COOKIE = "seskit_session"
+
+
+def _csrf(html: str) -> str:
+ marker = 'name="csrf_token" value="'
+ start = html.index(marker) + len(marker)
+ return html[start : html.index('"', start)]
+
+
+def _code_on_page(html: str) -> str:
+ match = re.search(r"keyreveal__key\">([^<]+)<", html)
+ assert match, "no recovery code on the page"
+ return match.group(1)
+
+
+def _error_on_page(html: str) -> str:
+ match = re.search(r'role="alert">([^<]+)<', html)
+ assert match, "no error on the page"
+ return match.group(1)
+
+
+async def _sign_up(client: AsyncClient) -> str:
+ """Create the owner through the app and return the code the page showed."""
+ response = await client.post("/signup", data={"email": EMAIL, "password": OLD})
+ assert response.status_code == 200, response.text
+ return _code_on_page(response.text)
+
+
+async def test_signing_up_shows_a_recovery_code_once(app_client: AsyncClient) -> None:
+ response = await app_client.post("/signup", data={"email": EMAIL, "password": OLD})
+
+ assert response.status_code == 200
+ assert "Save your recovery code" in response.text
+ assert CODE_RE.fullmatch(_code_on_page(response.text))
+ assert COOKIE in response.cookies, "and the same response starts the session"
+ assert "no-store" in response.headers["cache-control"]
+
+
+async def test_the_code_is_not_on_the_dashboard_afterwards(app_client: AsyncClient) -> None:
+ code = await _sign_up(app_client)
+
+ dashboard = await app_client.get("/")
+ settings_page = await app_client.get("/settings")
+
+ assert code not in dashboard.text
+ assert code not in settings_page.text
+
+
+async def test_the_login_page_offers_the_way_back(app_client: AsyncClient) -> None:
+ assert 'href="/forgot-password"' in (await app_client.get("/login")).text
+ assert (await app_client.get("/forgot-password")).status_code == 200
+
+
+async def test_the_code_recovers_the_account_through_the_page(
+ app_client: AsyncClient, redis_client: Redis
+) -> None:
+ code = await _sign_up(app_client)
+ old_cookie = app_client.cookies[COOKIE]
+ app_client.cookies.clear()
+
+ response = await app_client.post(
+ "/forgot-password", data={"email": EMAIL, "recovery_code": code, "password": NEW}
+ )
+
+ assert response.status_code == 200, response.text
+ assert "Password changed" in response.text
+ assert _code_on_page(response.text) != code, "the spent code is replaced"
+ assert await read_session(redis_client, old_cookie, 3600) is None, "old sessions are ended"
+ assert (
+ await app_client.post(
+ "/login", data={"email": EMAIL, "password": NEW}, follow_redirects=False
+ )
+ ).status_code == 303
+
+
+async def test_a_wrong_code_and_an_unknown_email_look_identical(app_client: AsyncClient) -> None:
+ await _sign_up(app_client)
+ app_client.cookies.clear()
+
+ wrong_code = await app_client.post(
+ "/forgot-password",
+ data={"email": EMAIL, "recovery_code": generate_code(), "password": NEW},
+ )
+ unknown = await app_client.post(
+ "/forgot-password",
+ data={"email": "nobody@example.com", "recovery_code": generate_code(), "password": NEW},
+ )
+
+ assert wrong_code.status_code == unknown.status_code == 400
+ assert _error_on_page(wrong_code.text) == _error_on_page(unknown.text)
+
+
+async def test_a_short_new_password_is_explained_and_costs_no_attempt(
+ app_client: AsyncClient, settings: Settings
+) -> None:
+ code = await _sign_up(app_client)
+ app_client.cookies.clear()
+
+ for _ in range(settings.LOGIN_MAX_ATTEMPTS + 2):
+ response = await app_client.post(
+ "/forgot-password", data={"email": EMAIL, "recovery_code": code, "password": "short"}
+ )
+ assert response.status_code == 400
+ assert "at least" in response.text
+
+ ok = await app_client.post(
+ "/forgot-password", data={"email": EMAIL, "recovery_code": code, "password": NEW}
+ )
+ assert ok.status_code == 200
+
+
+async def test_guessing_stops_however_the_caller_claims_to_be_addressed(
+ app_client: AsyncClient, settings: Settings
+) -> None:
+ """The limit is per email, not per address, because the address is whatever
+ `X-Forwarded-For` says. An attacker who varies it must not get fresh guesses."""
+ await _sign_up(app_client)
+ app_client.cookies.clear()
+
+ statuses = []
+ for attempt in range(settings.LOGIN_MAX_ATTEMPTS + 3):
+ response = await app_client.post(
+ "/forgot-password",
+ data={"email": EMAIL, "recovery_code": generate_code(), "password": NEW},
+ headers={"X-Forwarded-For": f"203.0.113.{attempt}"},
+ )
+ statuses.append(response.status_code)
+
+ assert statuses[: settings.LOGIN_MAX_ATTEMPTS] == [400] * settings.LOGIN_MAX_ATTEMPTS
+ assert set(statuses[settings.LOGIN_MAX_ATTEMPTS :]) == {429}
+
+
+@pytest.fixture
+def recovery_key_in_env(monkeypatch: pytest.MonkeyPatch) -> None:
+ monkeypatch.setenv("RECOVERY_KEY", KEY)
+
+
+@pytest.fixture
+async def keyed_client(recovery_key_in_env: None, app_client: AsyncClient) -> AsyncClient:
+ """An app built after `RECOVERY_KEY` is in the environment.
+
+ The `settings` fixture reads the environment when it first runs, so the
+ variable has to be set before `app_client` is requested. Pytest sets up a
+ fixture's arguments left to right, which is what makes the order here
+ load-bearing - and why it is spelt out as arguments rather than fetched with
+ `getfixturevalue`, which cannot start an async fixture from inside another.
+ """
+ return app_client
+
+
+async def test_the_key_signs_an_owner_with_no_code_back_in(
+ keyed_client: AsyncClient, db_session: AsyncSession
+) -> None:
+ await _account(db_session, with_code=False)
+
+ response = await keyed_client.post(
+ "/forgot-password", data={"email": EMAIL, "recovery_code": KEY, "password": NEW}
+ )
+
+ assert response.status_code == 200, response.text
+ assert "Password changed" in response.text
+ assert CODE_RE.fullmatch(_code_on_page(response.text))
+
+
+async def test_a_wrong_key_is_refused(keyed_client: AsyncClient, db_session: AsyncSession) -> None:
+ await _account(db_session, with_code=False)
+
+ response = await keyed_client.post(
+ "/forgot-password", data={"email": EMAIL, "recovery_code": KEY + "x", "password": NEW}
+ )
+
+ assert response.status_code == 400
+
+
+# -------------------------------------------------------------- settings ---
+
+
+async def test_settings_needs_a_session(app_client: AsyncClient) -> None:
+ response = await app_client.get("/settings", follow_redirects=False)
+
+ assert response.status_code == 303
+ assert response.headers["location"] == "/login"
+
+
+async def test_settings_says_whether_the_account_has_a_code(app_client: AsyncClient) -> None:
+ await _sign_up(app_client)
+
+ page = await app_client.get("/settings")
+
+ assert page.status_code == 200
+ assert "You have a recovery code" in page.text
+
+
+async def test_a_new_code_needs_the_password_and_replaces_the_old_one(
+ app_client: AsyncClient,
+) -> None:
+ code = await _sign_up(app_client)
+ csrf = _csrf((await app_client.get("/settings")).text)
+
+ refused = await app_client.post(
+ "/settings/recovery-code", data={"csrf_token": csrf, "password": "not-my-password"}
+ )
+ assert refused.status_code == 400
+ assert "not your current password" in refused.text
+
+ issued = await app_client.post(
+ "/settings/recovery-code", data={"csrf_token": csrf, "password": OLD}
+ )
+ assert issued.status_code == 200
+ assert "no-store" in issued.headers["cache-control"]
+ new_code = _code_on_page(issued.text)
+ assert new_code != code
+
+ app_client.cookies.clear()
+ old_use = await app_client.post(
+ "/forgot-password", data={"email": EMAIL, "recovery_code": code, "password": NEW}
+ )
+ assert old_use.status_code == 400, "the replaced code is dead"
+
+
+async def test_a_new_code_cannot_be_made_without_a_csrf_token(app_client: AsyncClient) -> None:
+ await _sign_up(app_client)
+
+ response = await app_client.post("/settings/recovery-code", data={"password": OLD})
+
+ assert response.status_code == 403
+
+
+async def test_another_accounts_session_is_untouched_by_a_recovery(
+ app_client: AsyncClient, db_session: AsyncSession, redis_client: Redis
+) -> None:
+ _, code = await _account(db_session)
+ other = await register_user(
+ db_session, email="other@example.com", password=OLD, allow_signup=True
+ )
+ token, _ = await create_session(redis_client, other.id, 3600)
+
+ await app_client.post(
+ "/forgot-password", data={"email": EMAIL, "recovery_code": code, "password": NEW}
+ )
+
+ assert await read_session(redis_client, token, 3600) is not None