Learning Hub

How to Build a Secure Password Reset Flow in FastAPI and Block Token Guessing, Enumeration, and Host Header Poisoning

Build a password reset flow in FastAPI, break the naive version with twelve attack scripts, then harden it with random hashed tokens, one-time claims and trusted hosts, and check every defense with...

October 22, 2026 39 Min Read
48

Most web apps with accounts have a “Forgot your password?” link, and that link is a second front door. A person who cannot remember a password still has to get back in, so the server hands out a short-lived secret and accepts it in place of the password. In this tutorial you will build that flow in a small FastAPI app twice: first the way most of us write it on the first try, then the way it should be written. You will attack the first version with twelve small scripts (guessing tokens, spotting which addresses have accounts, poisoning the emailed link, replaying and racing tokens, and more), and every one of them will work. Then you will build the second version, run the same twelve scripts again, and watch all of them fail while the legitimate reset keeps working. Finally you will check that each defense earns its place by removing them one at a time.

Table Of Content

No security background is assumed. You should be able to read Python, know what an HTTP request and a cookie are, and have seen a FastAPI route before. Every term is defined when it first appears, and nothing in the lab sends real email: messages are written to a file.

What a password reset is and why attackers like it

A password reset swaps “something you know” for “something you can receive”. The usual flow has five moves:

  1. The person types their email address into a “forgot password” form.
  2. The server creates a reset token, a secret that stands for “this request may change the password of account X”.
  3. The server emails a link that contains the token.
  4. The person opens the link and chooses a new password.
  5. The server checks the token, changes the password and, ideally, throws the token away.

Whoever holds the token can set the password, so for as long as it lives the token is as powerful as the password itself. MITRE’s weakness entry CWE-640 describes the problem as a mechanism “for users to recover or change their passwords without knowing the original password, but the mechanism is weak”, and adds that “Weak password recovery schemes completely undermine a strong password authentication scheme.” You can hash every password with Argon2, as in our password hashing tutorial, and still lose accounts through the reset form.

Four ideas come back again and again, so here they are up front:

  • Entropy is how many guesses an attacker needs. A token derived from the clock has almost none. A token from Python’s secrets module has 256 bits.
  • Account enumeration means learning which email addresses have accounts from the wording, status code or speed of an answer. It turns a reset form into a lookup service for phishing lists.
  • Host header poisoning means tricking the app into building the emailed link from the Host header of the request. PortSwigger defines “password reset poisoning” as a technique “whereby an attacker manipulates a vulnerable website into generating a password reset link pointing to a domain under their control.”
  • Single use and expiry mean a token works once and only for a short time, so a copy found later in a log, a backup or an old mailbox is worthless.

The checklist we follow is the OWASP Forgot Password Cheat Sheet. I quote it where a rule comes from it and say so when a choice is mine.

The twelve attacks you will run

Each attack has an ID so you can follow it through the tutorial. The last column is the defense you will build for it.

ID Attack What the attacker does Defense
A01 Predictable token Requests a reset for the victim, then guesses the token from the clock Random token from secrets
A02 Enumeration by response Compares the answers for a known and an unknown address One answer for both
A03 Enumeration by timing Times the answers Send the mail after responding
A04 Host header poisoning Sends the request with Host set to a domain they own Configured base URL plus trusted hosts
A05 Token reuse Submits a token that was already used Single use
A06 Expired token Uses a token that is hours old Expiry
A07 Stolen database Reads the token table and submits what it finds Store only a digest
A08 Session survives reset Keeps a stolen session after the owner resets Revoke sessions on reset
A09 Link scanner A mail scanner opens the link before the person does GET reads, POST spends
A10 Email flood Requests 15 resets for one victim at the same moment Silent per-account throttle, checked and recorded in one statement
A11 Token in the access log Reads the server log Redact the token
A12 Double spend Submits one token 8 times at the same moment Atomic claim

What you need before you start

  • Python 3.10 or newer. Every output below was produced on Python 3.13.14 on Windows 11 with FastAPI 0.159.1, Starlette 1.9.0, uvicorn 0.54.0, python-multipart 0.0.32, argon2-cffi 25.1.0, requests 2.34.2, httpx2 2.16.0 and pytest 9.1.1. Pinning those versions keeps your output identical to mine, apart from generated tokens and timings.
  • A terminal. The commands use bash syntax; on Windows use Git Bash or adapt the activation line.
  • SQLite ships with Python, so there is nothing else to install. Each attack run takes about ten seconds; the mutation matrix in Step 9 starts fourteen servers one after another and takes a couple of minutes.
mkdir reset-lab && cd reset-lab
python -m venv .venv
source .venv/bin/activate        # on Windows: .venv\Scripts\activate
pip install fastapi==0.159.1 uvicorn==0.54.0 python-multipart==0.0.32 argon2-cffi==25.1.0 requests==2.34.2 httpx2==2.16.0 pytest==9.1.1

FastAPI is the web framework, uvicorn runs it, python-multipart lets FastAPI read HTML form posts, and argon2-cffi hashes passwords. requests plays the attacker, and httpx2 is what Starlette’s test client uses inside pytest. Put every file in the reset-lab folder. Each Python block starts with a comment that names its file. When the comment says “(continued)”, append the block to the end of that file.

Step 1: Write the plumbing both versions share

The reset code is what we want to measure, so everything else is identical in the naive and the hardened version. Start with the database helpers. Users and sessions live in SQLite, passwords are hashed with Argon2id at OWASP’s minimum cost (19 MiB of memory, 2 passes, 1 lane), and every connection runs in autocommit mode so that we choose where transactions begin.

# store.py - users, sessions and the password hasher, shared by the naive and the hardened app
import os
import sqlite3

from argon2 import PasswordHasher

DB_PATH = os.environ.get("RESET_DB", "lab.db")
hasher = PasswordHasher(time_cost=2, memory_cost=19456, parallelism=1)  # Argon2id at OWASP's minimum cost


def connect():
    con = sqlite3.connect(DB_PATH, timeout=10, isolation_level=None)  # autocommit; we write BEGIN ourselves
    con.row_factory = sqlite3.Row
    return con


def init_users():
    con = connect()
    con.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, email TEXT UNIQUE NOT NULL, pw_hash TEXT NOT NULL)")
    con.execute("CREATE TABLE IF NOT EXISTS sessions (sid TEXT PRIMARY KEY, user_id INTEGER NOT NULL, created_at REAL NOT NULL)")
    con.close()

Next comes the fake mailbox. Every message is appended to a JSON lines file instead of being sent. A real mail server round trip is much slower than a database lookup, so the lab simulates it with an 80 millisecond sleep. That delay matters later, because a difference in speed can leak information.

# mailer.py - a fake outbox, so the lab never sends real email
import json
import os
import time

OUTBOX = os.environ.get("RESET_OUTBOX", "outbox.jsonl")
SMTP_DELAY = float(os.environ.get("SMTP_DELAY", "0.08"))  # a real mail server round trip is slower than a database lookup


def send_mail(to, subject, body):
    time.sleep(SMTP_DELAY)
    with open(OUTBOX, "a", encoding="utf-8", newline="\n") as fh:
        fh.write(json.dumps({"to": to, "subject": subject, "body": body, "at": time.time()}) + "\n")

Finally the account routes: sign up, sign in and a /me page that says who you are. A session is a random 256-bit id stored in a table and handed to the browser in a cookie. These routes are plumbing, not the subject of this tutorial, but notice that login fills the sessions table, because a reset has to deal with those rows later.

# accounts.py - signup, login and /me, shared by both apps (not the subject of this tutorial)
import secrets
import sqlite3
import time

from argon2.exceptions import VerifyMismatchError
from fastapi import APIRouter, Cookie, Form
from fastapi.responses import JSONResponse

from store import connect, hasher

router = APIRouter()


@router.post("/signup")
def signup(email: str = Form(), password: str = Form()):
    try:
        connect().execute("INSERT INTO users (email, pw_hash) VALUES (?, ?)", (email.strip().lower(), hasher.hash(password)))
    except sqlite3.IntegrityError:
        return JSONResponse({"message": "taken"}, status_code=409)
    return {"message": "created"}


@router.post("/login")
def login(email: str = Form(), password: str = Form()):
    con = connect()
    user = con.execute("SELECT id, pw_hash FROM users WHERE email = ?", (email.strip().lower(),)).fetchone()
    try:
        if user is None:
            raise VerifyMismatchError()
        hasher.verify(user["pw_hash"], password)
    except VerifyMismatchError:
        return JSONResponse({"message": "wrong email or password"}, status_code=401)
    sid = secrets.token_urlsafe(32)
    con.execute("INSERT INTO sessions (sid, user_id, created_at) VALUES (?, ?, ?)", (sid, user["id"], time.time()))
    response = JSONResponse({"message": "signed in"})
    response.set_cookie("sid", sid, httponly=True, samesite="lax")
    return response


@router.get("/me")
def me(sid: str | None = Cookie(default=None)):
    row = connect().execute(
        "SELECT u.email FROM sessions s JOIN users u ON u.id = s.user_id WHERE s.sid = ?", (sid,)
    ).fetchone()
    if row is None:
        return JSONResponse({"message": "not signed in"}, status_code=401)
    return {"email": row["email"]}

Step 2: Write the reset the way most of us write it first

Here is a complete password reset in about 50 lines. It works. A person who forgets their password can get back in. Read it as a reviewer and see how many problems you can spot before Step 4 lists them.

# app_v1.py - the first version of password reset that most of us write: it works, and it is full of holes
import hashlib
import time

from fastapi import FastAPI, Form, Request
from fastapi.responses import HTMLResponse, JSONResponse

from accounts import router
from mailer import send_mail
from store import connect, hasher, init_users

init_users()
connect().execute(
    "CREATE TABLE IF NOT EXISTS reset_tokens (token TEXT PRIMARY KEY, user_id INTEGER NOT NULL, "
    "created_at REAL NOT NULL, used INTEGER NOT NULL DEFAULT 0)"
)
app = FastAPI()
app.include_router(router)


@app.post("/forgot")
def forgot(request: Request, email: str = Form()):
    con = connect()
    user = con.execute("SELECT id FROM users WHERE email = ?", (email,)).fetchone()
    if user is None:
        return JSONResponse({"message": "No account with that email address."}, status_code=404)
    token = hashlib.md5(f"{email}{int(time.time())}".encode()).hexdigest()
    con.execute("INSERT OR REPLACE INTO reset_tokens (token, user_id, created_at) VALUES (?, ?, ?)",
                (token, user["id"], time.time()))
    send_mail(email, "Reset your password", f"Open this link to choose a new password: {request.base_url}reset?token={token}")
    return {"message": "We emailed you a reset link."}


@app.get("/reset", response_class=HTMLResponse)
def reset_page(token: str):
    con = connect()
    row = con.execute("SELECT user_id FROM reset_tokens WHERE token = ? AND used = 0", (token,)).fetchone()
    if row is None:
        return HTMLResponse("This link is invalid or has already been used.", status_code=400)
    con.execute("UPDATE reset_tokens SET used = 1 WHERE token = ?", (token,))
    return (f'<form method="post" action="/reset"><input type="hidden" name="token" value="{token}">'
            '<input type="password" name="new_password"><button>Change password</button></form>')


@app.post("/reset")
def reset_submit(token: str = Form(), new_password: str = Form()):
    con = connect()
    row = con.execute("SELECT user_id FROM reset_tokens WHERE token = ?", (token,)).fetchone()
    if row is None:
        return JSONResponse({"message": "Invalid token."}, status_code=400)
    con.execute("UPDATE users SET pw_hash = ? WHERE id = ?", (hasher.hash(new_password), row["user_id"]))
    return {"message": "Password updated."}

The flow is the five moves from earlier. /forgot looks the address up, builds a token, stores it, and mails a link whose start comes from request.base_url. GET /reset checks the token and shows a form with the token in a hidden field. POST /reset checks the token again and sets the new password. The used column is meant to make the link single use.

A password reset that works for the legitimate user is the easy half. Before attacking anything, build the harness that will run the attacks, and make sure it can drive the happy path.

Step 3: Build a harness that attacks a real running server

The attacks need real HTTP, real headers, a real access log and a real database file, so each run starts a throwaway copy of the app. labkit.py launches uvicorn on a port you choose, gives it its own database, outbox and log file under runs/, waits until it answers, and stops the whole process tree afterwards (on Windows the virtual environment’s python.exe is a launcher, so it uses taskkill with the tree flag).

# labkit.py - run a throwaway copy of an app with its own port, database, outbox and server log
import os
import pathlib
import shutil
import subprocess
import sys
import time
import urllib.request

LAB = pathlib.Path(__file__).resolve().parent


class Server:
    def __init__(self, module, port, cwd=LAB):
        self.module, self.port, self.cwd = module, port, pathlib.Path(cwd)
        self.dir = LAB / "runs" / f"{module}_{port}"
        self.base = f"http://127.0.0.1:{port}"
        self.db, self.outbox, self.log = (self.dir / "lab.db", self.dir / "outbox.jsonl", self.dir / "server.log")

    def __enter__(self):
        shutil.rmtree(self.dir, ignore_errors=True)
        self.dir.mkdir(parents=True)
        env = dict(os.environ, PYTHONUTF8="1", RESET_DB=str(self.db), RESET_OUTBOX=str(self.outbox),
                   PUBLIC_BASE_URL=self.base, ALLOWED_HOSTS="127.0.0.1,localhost", SMTP_DELAY="0.08")
        self.handle = open(self.log, "w", encoding="utf-8")
        self.proc = subprocess.Popen([sys.executable, "-m", "uvicorn", f"{self.module}:app", "--host", "127.0.0.1",
                                      "--port", str(self.port)], cwd=self.cwd, env=env, stdout=self.handle,
                                     stderr=subprocess.STDOUT)
        for _ in range(200):
            try:
                urllib.request.urlopen(self.base + "/openapi.json", timeout=1)
                return self
            except Exception:
                time.sleep(0.1)
        raise RuntimeError("server did not start; see " + str(self.log))

    def __exit__(self, *exc):
        if sys.platform == "win32":  # the venv python.exe is a launcher, so end the whole process tree
            subprocess.run(["taskkill", "/PID", str(self.proc.pid), "/T", "/F"], capture_output=True)
        else:
            self.proc.terminate()
        self.proc.wait()
        self.handle.close()

Now the attacker’s toolbox. The Lab class wraps the HTTP calls an attacker makes (signup, login, forgot, reset). It also has four helpers that need a word of explanation, because the attacker is not supposed to have them for free:

  • mails, wait_mail, link_for and token_for read the outbox. That is what the victim sees in their inbox, so attacks use it only when the script is playing the victim.
  • sql and age_tokens touch the database file. sql models a leaked backup, and age_tokens moves the timestamps back so we do not have to wait two hours.
# attacks.py - twelve attacks on a password reset flow. Each returns (succeeded, detail).
import hashlib
import json
import pathlib
import re
import sqlite3
import statistics
import threading
import time

import requests

PASSWORD = "Correct-Horse-Battery-1"


class Lab:
    """What an attack needs: the base URL, plus the lab's database, outbox and server log."""

    def __init__(self, base, db, outbox, log):
        self.base, self.db, self.outbox, self.log = base, db, outbox, log

    def post(self, path, data, **kwargs):
        return requests.post(self.base + path, data=data, **kwargs)  # a new connection every time

    def signup(self, n):
        email = f"victim{n}@example.com"
        self.post("/signup", {"email": email, "password": PASSWORD})
        return email

    def login(self, email, password):
        r = self.post("/login", {"email": email, "password": password})
        return r.cookies.get("sid") if r.status_code == 200 else None

    def forgot(self, email, **kwargs):
        return self.post("/forgot", {"email": email}, **kwargs)

    def reset(self, token, new_password):
        return self.post("/reset", {"token": token, "new_password": new_password})

    def mails(self, email):
        path = pathlib.Path(self.outbox)
        if not path.exists():
            return []
        rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
        return [m for m in rows if m["to"] == email and m["subject"].startswith("Reset")]

    def wait_mail(self, email, wait=3.0):
        end = time.time() + wait
        while time.time() < end:
            if self.mails(email):
                break
            time.sleep(0.05)
        return self.mails(email)

    def link_for(self, email):
        mails = self.wait_mail(email)
        return re.search(r"https?://\S+", mails[-1]["body"]).group(0) if mails else None

    def token_for(self, email):
        link = self.link_for(email)
        return re.search(r"token=([\w-]+)", link).group(1) if link else None

    def sql(self, query, args=()):
        con = sqlite3.connect(self.db)
        try:
            return con.execute(query, args).fetchall()
        finally:
            con.close()

    def age_tokens(self, seconds):
        con = sqlite3.connect(self.db)
        columns = [row[1] for row in con.execute("PRAGMA table_info(reset_tokens)")]
        for column in ("created_at", "expires_at"):
            if column in columns:
                con.execute(f"UPDATE reset_tokens SET {column} = {column} - ?", (seconds,))
        con.commit()
        con.close()


ATTACKS = []


def attack(code, name):
    """Register an attack function in ATTACKS, in the order it is defined."""

    def register(fn):
        ATTACKS.append((code, name, fn))
        return fn

    return register

Every attack returns a pair: did it succeed, and a one-line detail. The small attack decorator registers each attack in the ATTACKS list, in the order the functions are defined, so the runner can find them. First comes the one that is not an attack at all. a00_happy_path runs the legitimate flow: request a reset, open the link, submit a new password, and check that the new password works and the old one does not. A hardened flow that nobody can use is not a fix, so this check stays in the suite for every version.

# attacks.py (continued)
def a00_happy_path(lab):
    """Not an attack: the legitimate flow has to keep working, or 'secure' just means 'broken'."""
    email = lab.signup(0)
    lab.forgot(email)
    page = requests.get(lab.link_for(email))
    token = re.search(r'name="token" value="([^"]+)"', page.text).group(1)  # what the browser form posts back
    done = lab.reset(token, "Brand-New-Password-1")
    ok = (page.status_code, done.status_code) == (200, 200) and lab.login(email, "Brand-New-Password-1") and not lab.login(email, PASSWORD)
    return bool(ok), f"form page HTTP {page.status_code}, reset HTTP {done.status_code}, new password works, old one does not"

The runner starts a server, runs the happy path, then runs the registered attacks and prints one line each. Pass --only with a comma-separated list of IDs to run just some of them; --only A00 runs the happy path alone, because no attack has that ID.

# run_attacks.py MODULE [PORT] [--only A01,A04] - start MODULE:app, run the attacks, print one line per attack
import sys

from attacks import ATTACKS, Lab, a00_happy_path
from labkit import Server


def run_suite(module, port, only=None, cwd=None, quiet=False):
    results = {}
    kwargs = {"cwd": cwd} if cwd else {}
    with Server(module, port, **kwargs) as srv:
        lab = Lab(srv.base, srv.db, srv.outbox, srv.log)
        ok, detail = a00_happy_path(lab)
        results["A00"] = ok
        if not quiet:
            print(f"A00 legitimate reset         {'works' if ok else 'BROKEN':9} {detail}")
        for code, name, fn in ATTACKS:
            if only and code not in only:
                continue
            succeeded, detail = fn(lab)
            results[code] = succeeded
            if not quiet:
                print(f"{code} {name:25} {'SUCCEEDED' if succeeded else 'blocked':9} {detail}")
    return results


if __name__ == "__main__":
    args = [a for a in sys.argv[1:] if not a.startswith("--")]
    only = set(sys.argv[sys.argv.index("--only") + 1].split(",")) if "--only" in sys.argv else None
    if only:
        args = [a for a in args if a not in only and "," not in a]
    port = int(args[1]) if len(args) > 1 else 8000
    res = run_suite(args[0], port, only)
    if len(res) > 1:
        wins = sum(1 for k, v in res.items() if k != "A00" and v)
        print(f"{wins} of {len(res) - 1} attacks succeeded against {args[0]}")

Create attacks.py from the first two blocks above, and run the happy path against the naive app:

python run_attacks.py app_v1 8101 --only A00
A00 legitimate reset         works     form page HTTP 200, reset HTTP 200, new password works, old one does not

Port 8101 is arbitrary; any free port works. The line says works, so a person can reset a password with this app. That is the baseline: from here on, a defense counts only if this line keeps saying works.

Step 4: Attack the naive version

Add the twelve attacks below to attacks.py, in this order. Each one is a short function. Read them as a story: the attacker knows a victim’s email address and nothing else.

Guess the token (A01)

The naive token is md5(email + current_unix_second). The attacker knows the victim’s address and roughly when they pressed the button, so the token has about as much secrecy as the time of day. The script never reads the email. It requests a reset, waits a moment for the server to finish, then tries every second in a small window with two formulas that developers reach for first, submitting each guess as a token together with a new password.

# attacks.py (continued)
@attack("A01", "predictable token")
def a01_predictable_token(lab):
    email = lab.signup(1)
    first = int(time.time())
    lab.forgot(email)
    time.sleep(0.3)  # the token is issued in the background, so give the server a moment
    last = int(time.time()) + 1
    tries = 0
    for ts in range(first - 1, last + 1):
        for guess in (hashlib.md5(f"{email}{ts}".encode()).hexdigest(), hashlib.md5(str(ts).encode()).hexdigest()):
            tries += 1
            if lab.reset(guess, "Attacker-Chosen-Pass-1").status_code == 200:
                taken = lab.login(email, "Attacker-Chosen-Pass-1") is not None
                return taken, f"guessed the token on try {tries} without seeing the email; logged in as the victim: {taken}"
    return False, f"{tries} guesses from the clock, none accepted"

A success here is complete account takeover with no access to the victim’s mailbox. The fix is to stop deriving tokens from guessable facts and draw them from the operating system’s random number generator. The Python documentation says secrets “should be used in preference to the default pseudo-random number generator in the random module, which is designed for modelling and simulation, not security or cryptography”, and that “it is believed that 32 bytes (256 bits) of randomness is sufficient for the typical use-case expected for the secrets module.”

Learn who has an account (A02 and A03)

The naive /forgot answers 404 with “No account with that email address.” for an unknown address and 200 with a different sentence for a known one. That is the obvious leak, and A02 compares the two answers. A03 is the subtle one. For a known address the handler sends mail before answering, which takes about 80 milliseconds here; for an unknown address it returns immediately. Even if you make both answers identical, a stopwatch tells the difference. A03 uses a fresh registered address for every measurement so that nothing else (like a rate limit that kicks in after a few requests) can hide the gap.

# attacks.py (continued)
@attack("A02", "enumeration by response")
def a02_enumeration_by_response(lab):
    known = lab.forgot(lab.signup(2))
    unknown = lab.forgot("[email protected]")
    differs = (known.status_code, known.json()) != (unknown.status_code, unknown.json())
    return differs, f"known: {known.status_code} {known.json()['message']!r} | unknown: {unknown.status_code} {unknown.json()['message']!r}"


@attack("A03", "enumeration by timing")
def a03_enumeration_by_timing(lab, rounds=20):
    known, unknown = [], []
    for i in range(rounds):
        email = f"timing{i}@example.com"
        lab.post("/signup", {"email": email, "password": PASSWORD})
        for address, bucket in ((email, known), (f"nobody{i}@example.com", unknown)):
            start = time.perf_counter()
            lab.forgot(address)
            bucket.append((time.perf_counter() - start) * 1000)
    k, u = statistics.median(known), statistics.median(unknown)
    return (k - u) > 30, f"median response: registered {k:.1f} ms, unregistered {u:.1f} ms over {rounds} first requests each"

OWASP’s rules for this are short: “Return a consistent message for both existent and non-existent accounts.” and “Ensure that responses return in a consistent amount of time to prevent an attacker enumerating which accounts exist. This could be achieved by using asynchronous calls or by making sure that the same logic is followed, instead of using a quick exit method.”

Poison the link (A04)

The naive app builds the link with request.base_url, which comes from the Host header. The attacker asks for a reset for the victim’s address but sends Host: evil.example. The victim receives a genuine email from the genuine site, containing a link that points at the attacker’s domain. When they click it, the attacker’s server receives the token in the URL. The script plays both parts: it reads the poisoned email, then replays the token against the real site.

# attacks.py (continued)
@attack("A04", "Host header poisoning")
def a04_host_header_poisoning(lab):
    email = lab.signup(4)
    lab.forgot(email, headers={"Host": "evil.example"})
    poisoned = [m for m in lab.wait_mail(email, 1.5) if "evil.example" in m["body"]]
    if not poisoned:
        return False, "no email carried the attacker's host"
    link = re.search(r"https?://\S+", poisoned[-1]["body"]).group(0)
    token = re.search(r"token=([\w-]+)", link).group(1)
    r = lab.reset(token, "Attacker-Chosen-Pass-1")  # the victim clicked, the attacker's server logged the token
    return r.status_code == 200, f"email link {link[:40]}...; token replayed on the real site: HTTP {r.status_code}"

The OWASP sheet is direct about the fix: “Don’t rely on the Host header while creating the reset URLs to avoid Host Header Injection attacks. The URL should either be hard-coded, or validated against a list of trusted domains.”

Abuse the token’s lifecycle (A05, A06 and A07)

These three attacks treat the token as an object with a lifespan. A05 submits a token twice, as someone would who found a used link in a log or an old email. A06 uses a token two hours after it was issued. A07 plays an attacker who got hold of a copy of the database (a backup, a leaky debug endpoint, an injection bug elsewhere) and submits the value stored in the token table as if it were the token.

# attacks.py (continued)
@attack("A05", "token reuse")
def a05_token_reuse(lab):
    email = lab.signup(5)
    lab.forgot(email)
    token = lab.token_for(email)
    first = lab.reset(token, "First-New-Password-1")
    second = lab.reset(token, "Attacker-Reuse-Pass-1")
    return second.status_code == 200, f"first use HTTP {first.status_code}, second use of the same token HTTP {second.status_code}"


@attack("A06", "expired token")
def a06_expired_token(lab):
    email = lab.signup(6)
    lab.forgot(email)
    token = lab.token_for(email)
    lab.age_tokens(2 * 3600)  # pretend two hours went by
    r = lab.reset(token, "Late-Attacker-Pass-1")
    return r.status_code == 200, f"token used two hours after it was issued: HTTP {r.status_code}"


@attack("A07", "stolen database")
def a07_stolen_database(lab):
    email = lab.signup(7)
    lab.forgot(email)
    lab.wait_mail(email)
    stored = lab.sql("SELECT * FROM reset_tokens ORDER BY created_at DESC LIMIT 1")[0][0]  # first column
    r = lab.reset(stored, "Thief-Chosen-Pass-1")
    return r.status_code == 200, f"value stored in the table ({stored[:10]}...) submitted as a token: HTTP {r.status_code}"

The OWASP sheet lists what a reset token should be: “Generated using a cryptographically secure random number generator.”, “Long enough to protect against brute-force attacks.”, “Linked to an individual user in the database.”, “Invalidated after they have been used.” and “Stored in a secure manner, as discussed in the Password Storage Cheat Sheet.” Our naive app fails four of the five: its token is linked to a user in the table, but it is predictable, reusable and stored in plain text.

Look at everything around the reset (A08 to A12)

A08 asks what happens to a session an attacker already holds when the real owner resets the password to lock them out. A09 simulates a mail security scanner. Automated clients fetch URLs with GET all the time, and a mail security tool that follows the links in incoming messages is one example (I did not test any real product; the script stands in for one with a plain GET). A naive link that changes state on GET gets used up by that visit. HTTP itself says GET is not supposed to cause such changes. RFC 9110 defines GET, HEAD, OPTIONS and TRACE as safe methods and explains that “The purpose of distinguishing between safe and unsafe methods is to allow automated retrieval processes (spiders) and cache performance optimization (pre-fetching) to work without fear of causing harm.” A10 fires fifteen reset requests for one victim at the same moment to flood their inbox. A11 checks whether the token ends up in the server’s access log, where anyone with log access, including a log-shipping vendor, can read it. Last, A12 submits one token eight times at the same instant, which is how you test single use under concurrency.

# attacks.py (continued)
@attack("A08", "session survives reset")
def a08_session_survives_reset(lab):
    email = lab.signup(8)
    stolen = lab.login(email, PASSWORD)  # the attacker already holds this session
    lab.forgot(email)
    lab.reset(lab.token_for(email), "Owner-Recovers-Pass-1")  # the real owner resets to lock them out
    r = requests.get(lab.base + "/me", cookies={"sid": stolen})
    return r.status_code == 200, f"stolen session after the owner's reset: HTTP {r.status_code}"


@attack("A09", "link scanner burns token")
def a09_link_scanner(lab):
    email = lab.signup(9)
    lab.forgot(email)
    link = lab.link_for(email)
    requests.get(link)  # a mail scanner opens the link first
    r = requests.get(link)  # then the person clicks it
    return r.status_code >= 400, f"after a scanner fetched the link, the person got HTTP {r.status_code}"


@attack("A10", "email flood")
def a10_email_flood(lab, sent=15):
    email = lab.signup(10)
    statuses, gate = [], threading.Barrier(sent)

    def fire():
        gate.wait()  # all requests leave at the same moment
        statuses.append(lab.forgot(email).status_code)

    pool = [threading.Thread(target=fire) for _ in range(sent)]
    [t.start() for t in pool]
    [t.join() for t in pool]
    time.sleep(1.5)
    count = len(lab.mails(email))
    return count > 5, f"{sent} simultaneous requests (statuses {sorted(set(statuses))}) put {count} emails in the victim's inbox"


@attack("A11", "token in access log")
def a11_token_in_access_log(lab):
    email = lab.signup(11)
    lab.forgot(email)
    token = lab.token_for(email)
    requests.get(lab.base + "/reset", params={"token": token})  # the person opens the link
    time.sleep(0.3)
    found = token in pathlib.Path(lab.log).read_text(encoding="utf-8", errors="replace")
    return found, f"raw token {'found' if found else 'not found'} in the server's access log"


@attack("A12", "double spend race")
def a12_double_spend(lab, threads=8):
    email = lab.signup(12)
    lab.forgot(email)
    token = lab.token_for(email)
    codes, gate = [], threading.Barrier(threads)

    def submit(i):
        gate.wait()
        codes.append(lab.reset(token, f"Racer-Number-{i}-Pass").status_code)

    pool = [threading.Thread(target=submit, args=(i,)) for i in range(threads)]
    [t.start() for t in pool]
    [t.join() for t in pool]
    return codes.count(200) > 1, f"{threads} simultaneous submissions of one token: {codes.count(200)} accepted"

Run all twelve against the naive app

python run_attacks.py app_v1 8102
A00 legitimate reset         works     form page HTTP 200, reset HTTP 200, new password works, old one does not
A01 predictable token         SUCCEEDED guessed the token on try 3 without seeing the email; logged in as the victim: True
A02 enumeration by response   SUCCEEDED known: 200 'We emailed you a reset link.' | unknown: 404 'No account with that email address.'
A03 enumeration by timing     SUCCEEDED median response: registered 102.8 ms, unregistered 6.4 ms over 20 first requests each
A04 Host header poisoning     SUCCEEDED email link http://evil.example/reset?token=d40109c5...; token replayed on the real site: HTTP 200
A05 token reuse               SUCCEEDED first use HTTP 200, second use of the same token HTTP 200
A06 expired token             SUCCEEDED token used two hours after it was issued: HTTP 200
A07 stolen database           SUCCEEDED value stored in the table (f3ff6f9ea8...) submitted as a token: HTTP 200
A08 session survives reset    SUCCEEDED stolen session after the owner's reset: HTTP 200
A09 link scanner burns token  SUCCEEDED after a scanner fetched the link, the person got HTTP 400
A10 email flood               SUCCEEDED 15 simultaneous requests (statuses [200]) put 15 emails in the victim's inbox
A11 token in access log       SUCCEEDED raw token found in the server's access log
A12 double spend race         SUCCEEDED 8 simultaneous submissions of one token: 8 accepted
12 of 12 attacks succeeded against app_v1

Every attack succeeded. Read a few of the lines closely. A01 took over an account in 3 guesses without ever seeing the email. A03 shows 102.8 milliseconds for a registered address against 6.4 for an unregistered one, a gap you can measure from the other side of the internet. A04 produced a link to evil.example in a real email. A12 shows that eight simultaneous submissions of the same token were all accepted, because the POST handler never looks at the used flag. A09 shows the reverse problem: the GET handler spends the token, so a scanner’s visit leaves the real person with “This link is invalid or has already been used.”

A07 deserves a closer look, because the fix is a design decision. A tiny script shows what a thief who copies the database sees in the token table (it prints the column names and the size of the first column, not the secret itself):

# peek_db.py DB - what someone holding a copy of the database sees in the reset_tokens table
import sqlite3
import sys

con = sqlite3.connect(sys.argv[1])
columns = [row[1] for row in con.execute("PRAGMA table_info(reset_tokens)")]
first = con.execute("SELECT * FROM reset_tokens ORDER BY created_at DESC LIMIT 1").fetchone()[0]
print("columns:", ", ".join(columns))
print(f"first column holds {len(first)} characters, all hex digits: {all(c in '0123456789abcdef' for c in first)}")
python peek_db.py runs/app_v1_8102/lab.db
columns: token, user_id, created_at, used
first column holds 32 characters, all hex digits: True

The first column is the token itself, in plain text. Anyone who reads the table can reset any account that has a pending reset. You will fix that in the next step.

Step 5: Build the token module

All token handling now lives in one small module, tokens.py. Read it first, then the explanation of each decision below.

# tokens.py - everything about reset tokens in one place
import hashlib
import secrets
import time

from store import connect

TOKEN_TTL = 30 * 60      # seconds a link stays valid
MAX_PER_HOUR = 3         # reset emails we will send to one account per hour


def init():
    con = connect()
    con.execute(
        "CREATE TABLE IF NOT EXISTS reset_tokens (token_hash TEXT PRIMARY KEY, user_id INTEGER NOT NULL, "
        "created_at REAL NOT NULL, expires_at REAL NOT NULL, used_at REAL)"
    )
    con.execute("CREATE INDEX IF NOT EXISTS reset_tokens_user ON reset_tokens (user_id, created_at)")
    con.close()


def digest(raw):
    return hashlib.sha256(raw.encode()).hexdigest()


def issue(con, user_id, now=None):
    """Create a token and return the raw value, or None if the account already got MAX_PER_HOUR in the last hour.

    Only the SHA-256 digest is stored. The limit check and the insert are one statement, so a burst of
    simultaneous requests cannot all squeeze under the limit."""
    now = time.time() if now is None else now
    raw = secrets.token_urlsafe(32)
    cur = con.execute(
        "INSERT INTO reset_tokens (token_hash, user_id, created_at, expires_at) "
        "SELECT ?, ?, ?, ? WHERE (SELECT COUNT(*) FROM reset_tokens WHERE user_id = ? AND created_at > ?) < ?",
        (digest(raw), user_id, now, now + TOKEN_TTL, user_id, now - 3600, MAX_PER_HOUR))
    return raw if cur.rowcount == 1 else None


def peek(raw, now=None):
    """Read-only check: whose token is this, if it is still good? Nothing is spent."""
    now = time.time() if now is None else now
    row = connect().execute("SELECT user_id FROM reset_tokens WHERE token_hash = ? AND used_at IS NULL AND expires_at > ?",
                            (digest(raw), now)).fetchone()
    return row["user_id"] if row else None


def claim(con, raw, now=None):
    """Spend a token in one conditional UPDATE. At most one caller can ever get a user id back."""
    now = time.time() if now is None else now
    rows = con.execute("UPDATE reset_tokens SET used_at = ? WHERE token_hash = ? AND used_at IS NULL AND expires_at > ? "
                       "RETURNING user_id", (now, digest(raw), now)).fetchall()
    return rows[0]["user_id"] if rows else None

Unguessable. issue draws 32 random bytes with secrets.token_urlsafe(32), which is 43 URL-safe characters and 256 bits of entropy. Nothing about the account or the time goes into it, so there is nothing to guess (A01).

Only a digest is stored. The raw value goes into the email and is never written anywhere on the server; the table holds its SHA-256 digest, and every lookup hashes the submitted token first. A stolen table now contains values that cannot be turned back into tokens (A07). OWASP says tokens should be “Stored in a secure manner, as discussed in the Password Storage Cheat Sheet.” For passwords that means a deliberately slow hash such as Argon2id, because people choose guessable passwords. A reset token is different: it is 256 random bits, so there is nothing realistic to brute-force, and a fast hash is enough. That distinction is my reasoning; the sheet does not say which hash to use for tokens. Here is the same table after the hardened app has run:

python peek_db.py runs/app_8103/lab.db
columns: token_hash, user_id, created_at, expires_at, used_at
first column holds 64 characters, all hex digits: True

Expires. Each row has an expires_at time 30 minutes after it was created, and both peek and claim compare it with the current time (A06). The comparison is strict, so a token is dead at exactly the expiry second.

Spent once. claim is the heart of the module. It spends a token with one conditional statement: UPDATE ... SET used_at = ? WHERE token_hash = ? AND used_at IS NULL AND expires_at > ? RETURNING user_id. The database decides, in a single step, whether this caller was the first. There is no “look, then update” gap for a second request to slip through (A05 and A12). If it returns a user id you won; if it returns nothing the token was unknown, already spent or expired, and the caller cannot tell which.

Read without spending. peek answers “is this token currently good, and whose is it?” without changing anything. The page that shows the form uses it, and so does the first check in the submit handler. That split is what makes the link survive a mail scanner (A09).

Throttled. issue also limits how many tokens an account can get: it inserts a row only if fewer than MAX_PER_HOUR rows were created for that account in the last hour, and otherwise returns None (A10). The count and the insert are a single INSERT ... SELECT ... WHERE statement, and that detail is a lesson I learned the hard way. My first version counted the rows in one statement and inserted in the next. It passed every sequential test, and then A10 came back in unrelated rows of the mutation matrix on a busy run, because requests that arrive together all count before any of them inserts. The attack now fires its fifteen requests at the same moment so that it would catch the old version, and the matrix has a mutant that restores it. The issued tokens are rows in the same table, so no extra storage is needed. I only tested this on SQLite, where one statement is atomic.

Step 6: Rebuild the endpoints

The hardened app.py has six parts. Start with the configuration. The public base URL comes from the environment, never from the request. ALLOWED_HOSTS lists the host names the app answers to. The form is a constant. RedactToken is a logging filter: uvicorn’s access log prints the request line, query string included, so without a filter the line GET /reset?token=... puts the secret in the log file (A11). The filter rewrites token=... to token=[redacted] before the line is written.

# app.py - the hardened password reset flow
import html
import logging
import os
import re
import time

from fastapi import BackgroundTasks, FastAPI, Form, Request
from fastapi.responses import HTMLResponse, JSONResponse
from starlette.middleware.trustedhost import TrustedHostMiddleware

import tokens
from accounts import router
from mailer import send_mail
from store import connect, hasher, init_users

PUBLIC_BASE_URL = os.environ.get("PUBLIC_BASE_URL", "http://127.0.0.1:8000")
ALLOWED_HOSTS = os.environ.get("ALLOWED_HOSTS", "127.0.0.1,localhost").split(",")
MIN_PASSWORD_LENGTH = 12
FORM = """<!doctype html>
<title>Choose a new password</title>
<form method="post" action="/reset">
  <input type="hidden" name="token" value="{token}">
  <label>New password <input type="password" name="new_password" minlength="12" autocomplete="new-password" required></label>
  <button>Change password</button>
</form>"""


class RedactToken(logging.Filter):
    """Keep reset tokens out of the access log."""

    def filter(self, record):
        if isinstance(record.args, tuple):
            record.args = tuple(re.sub(r"(token=)[^&\s\"]+", r"\1[redacted]", a) if isinstance(a, str) else a
                                for a in record.args)
        return True


logging.getLogger("uvicorn.access").addFilter(RedactToken())

Then the app itself, with two pieces of middleware. TrustedHostMiddleware comes with Starlette (FastAPI re-exports it). The FastAPI documentation describes it as enforcing “that all incoming requests have a correctly set Host header, in order to guard against HTTP Host Header attacks.” A request that arrives with Host: evil.example is answered with 400 and never reaches our code. The second piece adds headers to the reset page. OWASP asks that “the reset password page adds the Referrer Policy tag with the no-referrer value in order to avoid referrer leakage”, so the browser sends no Referer when the page loads images or links to other sites (MDN documents strict-origin-when-cross-origin as the default policy when none is specified, which already sends only the origin to other sites, so the header is a guarantee that does not depend on the default), and we add Cache-Control: no-store so the page, which has the token in its URL, is not kept in a cache.

# app.py (continued)
init_users()
tokens.init()
app = FastAPI()
app.add_middleware(TrustedHostMiddleware, allowed_hosts=ALLOWED_HOSTS)
app.include_router(router)


@app.middleware("http")
async def reset_page_headers(request: Request, call_next):
    response = await call_next(request)
    if request.url.path == "/reset":
        response.headers["Referrer-Policy"] = "no-referrer"
        response.headers["Cache-Control"] = "no-store"
    return response

Now the request handler. Two helpers come first. invalid builds the one error answer that the submit handler uses for every bad token, so it never says whether a token was unknown, used or expired. start_reset does all the account-dependent work: look the address up, create the token (the throttle lives inside issue), send the mail. The handler itself only schedules it with BackgroundTasks, so FastAPI runs it after the response has been sent. Both a registered and an unregistered address get the same answer in the same time, because the work that differs happens after the answer has gone out (A02, A03). The throttle is silent: when an account already has three tokens in the last hour, issue returns None, start_reset simply returns without sending, and the caller still sees 202 with the same message (A10). The link in the mail is built from PUBLIC_BASE_URL (A04).

# app.py (continued)
def invalid():
    return JSONResponse({"message": "This link is invalid or has expired. Please request a new one."}, status_code=400)


def start_reset(email):
    """Runs after the response has been sent, so the caller cannot time it."""
    con = connect()
    user = con.execute("SELECT id FROM users WHERE email = ?", (email,)).fetchone()
    raw = tokens.issue(con, user["id"]) if user else None  # None: no such account, or the throttle said no
    if raw is None:
        return
    send_mail(email, "Reset your password",
              f"Open this link within 30 minutes to choose a new password: {PUBLIC_BASE_URL}/reset?token={raw}")


@app.post("/forgot", status_code=202)
def forgot(background: BackgroundTasks, email: str = Form()):
    background.add_task(start_reset, email.strip().lower())
    return {"message": "If that address has an account, we sent a reset link."}

The page that the link opens only reads. It calls peek, shows the form if the token is good, and spends nothing, so opening the link twice (or letting a scanner open it first) changes nothing (A09). The token is escaped with html.escape before it goes into the hidden field.

# app.py (continued)
@app.get("/reset", response_class=HTMLResponse)
def reset_page(token: str):
    if tokens.peek(token) is None:
        return HTMLResponse("This link is invalid or has expired.", status_code=400)
    return HTMLResponse(FORM.format(token=html.escape(token)))

Django shows another way to keep the token out of Referer headers. In PasswordResetConfirmView its developers wrote a comment that explains the trick: “Store the token in the session and redirect to the password reset form at a URL without the token. That avoids the possibility of leaking the token in the HTTP Referer header.” Our no-referrer header is the control OWASP asks for; a redirect to a token-free URL is a worthwhile extra if you want a second layer.

Finally the handler that changes the password. The order of the steps matters, and each one has a reason:

  1. Check the new password first. A too-short password must not use up the token, or a typo would force the person to start again (test_a_weak_password_does_not_burn_the_token checks this).
  2. Peek before hashing. Argon2 costs real CPU time. Without this cheap check, anyone could make the server run it by posting junk tokens.
  3. Hash outside the transaction. We do not want to hold the database write lock while Argon2 runs.
  4. Claim, change the password, burn the other links and delete the sessions in one transaction. BEGIN IMMEDIATE takes the write lock first. If anything fails, ROLLBACK leaves the token unspent, so the person can try again. If it succeeds, every other outstanding link for the account is dead (UPDATE reset_tokens SET used_at ...) and every session is gone (DELETE FROM sessions, A08). OWASP’s recovery guidance says to “invalidate existing sessions and outstanding password reset and recovery links or codes so they cannot restore an attacker’s access.”
  5. Tell the owner, and do not log them in. The notification email contains no token and no password. OWASP’s rules: “Send the user an email informing them that their password has been reset (do not send the password in the email!)” and “Don’t automatically log the user in, as this introduces additional complexity to the authentication and session handling code, and increases the likelihood of introducing vulnerabilities.”
# app.py (continued)
@app.post("/reset")
def reset_submit(background: BackgroundTasks, token: str = Form(), new_password: str = Form()):
    if len(new_password) < MIN_PASSWORD_LENGTH:
        return JSONResponse({"message": f"Use at least {MIN_PASSWORD_LENGTH} characters."}, status_code=422)
    if tokens.peek(token) is None:
        return invalid()
    new_hash = hasher.hash(new_password)
    now = time.time()
    con = connect()
    con.execute("BEGIN IMMEDIATE")
    user_id = tokens.claim(con, token, now)
    if user_id is None:
        con.execute("ROLLBACK")
        return invalid()
    con.execute("UPDATE users SET pw_hash = ? WHERE id = ?", (new_hash, user_id))
    con.execute("UPDATE reset_tokens SET used_at = ? WHERE user_id = ? AND used_at IS NULL", (now, user_id))
    con.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,))
    email = con.execute("SELECT email FROM users WHERE id = ?", (user_id,)).fetchone()["email"]
    con.execute("COMMIT")
    background.add_task(send_mail, email, "Your password was changed",
                        "Your password was just changed and every device was signed out. If this was not you, contact support.")
    return {"message": "Password updated. Please sign in with your new password."}

One more thing about claim and the single-statement update. SQLite lets only one connection write at a time, so inside BEGIN IMMEDIATE the race is already closed. The conditional UPDATE is still the line that makes the claim correct, and it is the form that stays correct on other databases, where two requests can run at once. I only tested SQLite here, and the mutation matrix in Step 9 shows what happens to the concurrent attack when that condition is removed.

Step 7: Run the same twelve attacks against the hardened app

Nothing in attacks.py changes. Only the module name does:

python run_attacks.py app 8103
A00 legitimate reset         works     form page HTTP 200, reset HTTP 200, new password works, old one does not
A01 predictable token         blocked   6 guesses from the clock, none accepted
A02 enumeration by response   blocked   known: 202 'If that address has an account, we sent a reset link.' | unknown: 202 'If that address has an account, we sent a reset link.'
A03 enumeration by timing     blocked   median response: registered 3.0 ms, unregistered 2.5 ms over 20 first requests each
A04 Host header poisoning     blocked   no email carried the attacker's host
A05 token reuse               blocked   first use HTTP 200, second use of the same token HTTP 400
A06 expired token             blocked   token used two hours after it was issued: HTTP 400
A07 stolen database           blocked   value stored in the table (b7e56a8cb0...) submitted as a token: HTTP 400
A08 session survives reset    blocked   stolen session after the owner's reset: HTTP 401
A09 link scanner burns token  blocked   after a scanner fetched the link, the person got HTTP 200
A10 email flood               blocked   15 simultaneous requests (statuses [202]) put 3 emails in the victim's inbox
A11 token in access log       blocked   raw token not found in the server's access log
A12 double spend race         blocked   8 simultaneous submissions of one token: 1 accepted
0 of 12 attacks succeeded against app

All twelve are blocked, and the happy path still works. Compare the lines with Step 4. The two enumeration attacks now see the same 202 with the same sentence, and the timing gap is gone: 3.0 milliseconds for a registered address and 2.5 for an unregistered one, against 102.8 and 6.4 before. The poisoned request is refused before any mail exists. The token that was submitted twice is accepted once and rejected with 400 the second time. The expired token, the stolen table value and the late session are all rejected. After a scanner opened the link the person still gets 200. Fifteen simultaneous requests produced three emails and fifteen 202 responses. The raw token is not in the log, and of eight simultaneous submissions exactly one was accepted.

Step 8: Write the tests that keep it that way

The attack harness is an integration test, but it takes several seconds and needs a real server. Unit tests give faster feedback while you work and pin down each rule by name. conftest.py points the app at a throwaway database and outbox before anything imports it, switches the mail delay off, and sets the base URL and allowed hosts that the tests expect (testserver is the host name Starlette’s test client uses).

# conftest.py - point the app at a throwaway database and outbox before anything imports it
import os
import pathlib
import tempfile

_tmp = pathlib.Path(tempfile.mkdtemp(prefix="reset_tests_"))
os.environ["RESET_DB"] = str(_tmp / "test.db")
os.environ["RESET_OUTBOX"] = str(_tmp / "outbox.jsonl")
os.environ["SMTP_DELAY"] = "0"
os.environ["PUBLIC_BASE_URL"] = "https://app.example.test"
os.environ["ALLOWED_HOSTS"] = "testserver,localhost"

import pytest  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402


@pytest.fixture()
def client():
    import app as app_module
    from store import connect

    con = connect()
    for table in ("users", "sessions", "reset_tokens"):
        con.execute(f"DELETE FROM {table}")
    pathlib.Path(os.environ["RESET_OUTBOX"]).unlink(missing_ok=True)
    return TestClient(app_module.app)

The tests follow the rules of the flow one by one. A few things to notice: the token tests call tokens.issue with an explicit now, so the expiry boundary is tested without sleeping; the Host header test passes headers={"host": "evil.example"} and expects 400 and an empty outbox; the collision test releases eight threads at the same instant, each with its own database connection, and expects exactly one winner; and a second concurrency test releases twelve threads at start_reset and expects exactly three emails.

# test_reset.py - the rules of the reset flow, one test per rule
import json
import os
import pathlib
import re
import threading

import tokens
from store import connect

PASSWORD, NEW = "Correct-Horse-Battery-1", "Brand-New-Password-1"
EMAIL = "[email protected]"


def mails(subject_start="Reset", to=EMAIL):
    path = pathlib.Path(os.environ["RESET_OUTBOX"])
    rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()] if path.exists() else []
    return [m for m in rows if m["to"] == to and m["subject"].startswith(subject_start)]


def signup(client):
    client.post("/signup", data={"email": EMAIL, "password": PASSWORD})


def request_token(client):
    client.post("/forgot", data={"email": EMAIL})
    return re.search(r"token=([\w-]+)", mails()[-1]["body"]).group(1)


def user_id():
    return connect().execute("SELECT id FROM users WHERE email = ?", (EMAIL,)).fetchone()["id"]


def test_tokens_are_long_and_unique(client):
    signup(client)
    con = connect()
    issued = {tokens.issue(con, user_id(), now=1000.0 + 7200 * i) for i in range(200)}  # spaced out, so the throttle stays quiet
    assert len(issued) == 200 and all(len(t) == 43 for t in issued)


def test_only_a_digest_is_stored(client):
    signup(client)
    token = request_token(client)
    stored = [str(tuple(row)) for row in connect().execute("SELECT * FROM reset_tokens")]
    assert token not in " ".join(stored) and tokens.digest(token) in " ".join(stored)


def test_known_and_unknown_addresses_get_the_same_answer(client):
    signup(client)
    known = client.post("/forgot", data={"email": EMAIL})
    unknown = client.post("/forgot", data={"email": "[email protected]"})
    assert (known.status_code, known.json()) == (unknown.status_code, unknown.json())
    assert mails(to="[email protected]") == []


def test_link_uses_the_configured_base_url(client):
    signup(client)
    request_token(client)
    assert mails()[-1]["body"].count("https://app.example.test/reset?token=") == 1


def test_unknown_host_header_is_refused(client):
    signup(client)
    r = client.post("/forgot", data={"email": EMAIL}, headers={"host": "evil.example"})
    assert r.status_code == 400 and mails() == []


def test_opening_the_link_does_not_spend_it(client):
    signup(client)
    token = request_token(client)
    assert client.get("/reset", params={"token": token}).status_code == 200
    assert client.get("/reset", params={"token": token}).status_code == 200
    assert client.post("/reset", data={"token": token, "new_password": NEW}).status_code == 200


def test_submitting_spends_it(client):
    signup(client)
    token = request_token(client)
    assert client.post("/reset", data={"token": token, "new_password": NEW}).status_code == 200
    assert client.post("/reset", data={"token": token, "new_password": "Another-Password-2"}).status_code == 400


def test_expiry_boundary(client):
    signup(client)
    raw = tokens.issue(connect(), user_id(), now=1000.0)
    assert tokens.peek(raw, now=1000.0 + tokens.TOKEN_TTL - 1) == user_id()
    assert tokens.peek(raw, now=1000.0 + tokens.TOKEN_TTL) is None
    assert tokens.claim(connect(), raw, now=1000.0 + tokens.TOKEN_TTL) is None


def test_a_weak_password_does_not_burn_the_token(client):
    signup(client)
    token = request_token(client)
    assert client.post("/reset", data={"token": token, "new_password": "short"}).status_code == 422
    assert client.post("/reset", data={"token": token, "new_password": NEW}).status_code == 200


def test_reset_signs_everyone_out_and_notifies(client):
    signup(client)
    client.post("/login", data={"email": EMAIL, "password": PASSWORD})
    assert client.get("/me").status_code == 200
    token = request_token(client)
    client.post("/reset", data={"token": token, "new_password": NEW})
    assert client.get("/me").status_code == 401
    notice = mails("Your password was changed")
    assert len(notice) == 1 and "token" not in notice[0]["body"] and NEW not in notice[0]["body"]
    assert client.post("/login", data={"email": EMAIL, "password": PASSWORD}).status_code == 401
    assert client.post("/login", data={"email": EMAIL, "password": NEW}).status_code == 200


def test_other_outstanding_links_die_with_the_first_use(client):
    signup(client)
    first, second = request_token(client), request_token(client)
    assert client.post("/reset", data={"token": first, "new_password": NEW}).status_code == 200
    assert client.post("/reset", data={"token": second, "new_password": "Another-Password-2"}).status_code == 400


def test_the_throttle_is_silent(client):
    signup(client)
    answers = {client.post("/forgot", data={"email": EMAIL}).status_code for _ in range(6)}
    assert answers == {202} and len(mails()) == tokens.MAX_PER_HOUR


def test_parallel_requests_cannot_beat_the_throttle(client):
    import app as app_module

    signup(client)
    gate = threading.Barrier(12)

    def go():
        gate.wait()
        app_module.start_reset(EMAIL)

    pool = [threading.Thread(target=go) for _ in range(12)]
    [t.start() for t in pool]
    [t.join() for t in pool]
    assert len(mails()) == tokens.MAX_PER_HOUR


def test_reset_page_headers(client):
    r = client.get("/reset", params={"token": "x"})
    assert r.headers["referrer-policy"] == "no-referrer" and r.headers["cache-control"] == "no-store"


def test_one_winner_when_claims_collide(client):
    signup(client)
    raw = tokens.issue(connect(), user_id())
    wins, gate = [], threading.Barrier(8)

    def go():
        con = connect()
        gate.wait()
        wins.append(tokens.claim(con, raw))

    pool = [threading.Thread(target=go) for _ in range(8)]
    [t.start() for t in pool]
    [t.join() for t in pool]
    assert [w for w in wins if w is not None] == [user_id()]
python -m pytest -q
...............                                                                                                                                                    [100%]
15 passed in 3.46s

Step 9: Check that every defense matters

A suite that passes does not show that each check is needed. The check I use is a mutation: remove one defense, run all twelve attacks again, and see which ones come back. mutations.py copies the hardened app into a scratch folder, applies one exact text replacement per mutant (and fails loudly if the text to replace is not found exactly once), starts that copy and runs the suite.

# mutations.py - remove one defense at a time from the hardened app and see which attacks come back
import ast
import pathlib
import shutil

from run_attacks import run_suite

LAB = pathlib.Path(__file__).resolve().parent

RACY_ISSUE = '''def issue(con, user_id, now=None):
    now = time.time() if now is None else now
    recent = con.execute("SELECT COUNT(*) FROM reset_tokens WHERE user_id = ? AND created_at > ?", (user_id, now - 3600)).fetchone()[0]
    if recent >= MAX_PER_HOUR:
        return None
    time.sleep(0.01)  # widen the gap between the check and the insert so the race shows every time
    raw = secrets.token_urlsafe(32)
    con.execute("INSERT INTO reset_tokens (token_hash, user_id, created_at, expires_at) VALUES (?, ?, ?, ?)",
                (digest(raw), user_id, now, now + TOKEN_TTL))
    return raw
'''

LINK_FROM_HOST = [
    ("app.py", "def start_reset(email):", "def start_reset(email, base):"),
    ("app.py", "{PUBLIC_BASE_URL}/reset?token={raw}", "{base}/reset?token={raw}"),
    ("app.py", "def forgot(background: BackgroundTasks, email: str = Form()):",
     "def forgot(request: Request, background: BackgroundTasks, email: str = Form()):"),
    ("app.py", "background.add_task(start_reset, email.strip().lower())",
     'background.add_task(start_reset, email.strip().lower(), str(request.base_url).rstrip("/"))'),
]
NO_TRUSTED_HOST = [("app.py", "app.add_middleware(TrustedHostMiddleware, allowed_hosts=ALLOWED_HOSTS)", "pass")]

MUTANTS = [
    ("tokens stored in plain text", [("tokens.py", "    return hashlib.sha256(raw.encode()).hexdigest()", "    return raw")]),
    ("links never expire", [("tokens.py", "TOKEN_TTL = 30 * 60", "TOKEN_TTL = 10 ** 9")]),
    ("claim ignores used_at (peek still checks)", [("tokens.py", "AND used_at IS NULL AND expires_at > ? \"", "AND expires_at > ? \"")]),
    ("claim and peek ignore used_at", [("tokens.py", "AND used_at IS NULL AND expires_at > ? \"", "AND expires_at > ? \""),
                                       ("tokens.py", "AND used_at IS NULL AND expires_at > ?\",", "AND expires_at > ?\",")]),
    ("GET spends the token", [("app.py", "def reset_page(token: str):\n    if tokens.peek(token) is None:",
                               "def reset_page(token: str):\n    if tokens.claim(connect(), token) is None:")]),
    ("sessions kept after reset", [("app.py", '    con.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,))\n', "")]),
    ("TrustedHostMiddleware removed", NO_TRUSTED_HOST),
    ("link built from the Host header", LINK_FROM_HOST),
    ("both of the two above", NO_TRUSTED_HOST + LINK_FROM_HOST),
    ("throttle counts, then inserts", [("tokens.py", "def issue", RACY_ISSUE)]),
    ("no per-account throttle", [("tokens.py", "MAX_PER_HOUR = 3", "MAX_PER_HOUR = 10 ** 6")]),
    ("unknown address gets a 404", [("app.py", "    background.add_task(start_reset, email.strip().lower())\n",
                                     '    if connect().execute("SELECT 1 FROM users WHERE email = ?", (email.strip().lower(),)).fetchone() is None:\n'
                                     '        return JSONResponse({"message": "No account with that email address."}, status_code=404)\n'
                                     "    background.add_task(start_reset, email.strip().lower())\n")]),
    ("mail sent before the response", [("app.py", "background.add_task(start_reset, email.strip().lower())", "start_reset(email.strip().lower())")]),
    ("no token redaction in the log", [("app.py", 'logging.getLogger("uvicorn.access").addFilter(RedactToken())', "pass")]),
]


def build(index, edits):
    folder = LAB / "runs" / f"mutant_{index:02d}"
    shutil.rmtree(folder, ignore_errors=True)
    folder.mkdir(parents=True)
    for name in ("app.py", "tokens.py", "accounts.py", "store.py", "mailer.py"):
        shutil.copy(LAB / name, folder / name)
    for name, old, new in edits:
        text = (folder / name).read_text(encoding="utf-8")
        if old == "def issue":  # replace a whole function: find its lines with ast
            node = next(n for n in ast.parse(text).body if isinstance(n, ast.FunctionDef) and n.name == "issue")
            rows = text.split("\n")
            text = "\n".join(rows[: node.lineno - 1] + new.rstrip("\n").split("\n") + rows[node.end_lineno:])
        else:
            assert text.count(old) == 1, f"{name}: {old!r} occurs {text.count(old)} times"
            text = text.replace(old, new)
        (folder / name).write_text(text, encoding="utf-8", newline="\n")
    return folder


if __name__ == "__main__":
    print(f"{'defense removed':44} {'login':6} attacks that came back")
    for index, (name, edits) in enumerate(MUTANTS, start=1):
        results = run_suite("app", 8200 + index, cwd=build(index, edits), quiet=True)
        back = [code for code, hit in results.items() if code != "A00" and hit]
        print(f"{name:44} {'works' if results['A00'] else 'BROKEN':6} {', '.join(back) or '(none)'}", flush=True)
python mutations.py
defense removed                              login  attacks that came back
tokens stored in plain text                  works  A07
links never expire                           works  A06
claim ignores used_at (peek still checks)    works  A12
claim and peek ignore used_at                works  A05, A12
GET spends the token                         BROKEN A09
sessions kept after reset                    works  A08
TrustedHostMiddleware removed                works  (none)
link built from the Host header              works  (none)
both of the two above                        works  A04
throttle counts, then inserts                works  A10
no per-account throttle                      works  A10
unknown address gets a 404                   works  A02
mail sent before the response                works  A03
no token redaction in the log                works  A11

Each defense except the token generator has a row, and most rows bring back exactly the attack you expect. The interesting rows are the ones that do not:

  • “claim ignores used_at (peek still checks)” brings back only A12. Sequential replays (A05) are still stopped by the read-only check in front of the transaction, but eight simultaneous submissions all pass that check before any of them spends the token. Only the conditional update in claim closes the window. Remove both conditions and A05 returns as well.
  • “TrustedHostMiddleware removed” and “link built from the Host header” each bring back nothing, because each of the two defenses covers for the other. Remove both and A04 returns. That is defense in depth, and it comes with a warning: a defense that nothing tests becomes invisible when its twin hides the failure. test_unknown_host_header_is_refused and test_link_uses_the_configured_base_url exist so each layer is checked on its own.
  • “GET spends the token” brings back A09, and the login column says BROKEN. The legitimate flow dies too, because the form page uses up the token before the person can submit it. This is why A00 stays in the suite: an attack that is blocked because the feature no longer works is not a win.
  • “throttle counts, then inserts” and “no per-account throttle” both bring back A10. The first mutant is the code I originally wrote, with a 10 millisecond sleep between the count and the insert so that the race shows up every time instead of only on a busy machine.
  • “unknown address gets a 404” brings back A02 only, and “mail sent before the response” brings back A03 only. The two enumeration leaks are independent, and fixing the wording does not fix the timing.

There is no row for the token generator (A01). Any weak generator I swapped in would be my guess about which weak scheme an attacker tries, so A01 and test_tokens_are_long_and_unique cover that defense instead.

Common mistakes and gotchas

  • Different answers for different outcomes. Any difference in status code, wording, headers or speed between “account exists” and “account does not exist” is an enumeration leak. A rate-limit answer such as “429, too many resets for this address” leaks too, which is why the throttle in this lab drops the mail quietly and keeps the 202. Rate-limit by account, as OWASP says: “Implement protections against excessive automated submissions such as rate-limiting on a per-account basis, requiring a CAPTCHA, or other controls.”
  • Locking accounts. OWASP warns that “Accounts should not be locked out in response to a forgotten password attack, as this can be used to deny access to users with known usernames.” Throttle the emails, not the account.
  • Background tasks are not a queue. BackgroundTasks runs inside the web process, so if the process stops after the response, the email is never sent. For production, put the work in a durable queue, for example the one from our SQLite job queue tutorial, and enqueue the same job for known and unknown addresses so the timing stays equal.
  • Plain HTTP links. The lab uses http://127.0.0.1 only because it runs on your machine. In production PUBLIC_BASE_URL must be an https address.
  • A reset does not repair a compromised account. OWASP notes that “A password reset alone may not restore control of a compromised account. An attacker may still have an active session or may have changed recovery information or MFA methods.” Our reset deletes sessions but does not touch recovery addresses or second factors. If your app has them, review them as part of the recovery, and never let a reset skip a second factor that the account has enabled.
  • The login form has its own enumeration leak. In accounts.py, login returns before hashing when the address is unknown, so it should answer faster than for a known address (I did not measure it). The fix is to verify the submitted password against a dummy hash in that case.
  • Stateless tokens are a different design, not a free upgrade. Django’s PasswordResetTokenGenerator keeps no token table. Its docstring says it hashes “the user’s primary key, email (if available), and some user state that’s sure to change after a password reset to produce a token that is invalidated when it’s used”, and that running this data through salted_hmac() “prevents password cracking attempts using the reset token, provided the secret isn’t compromised.” That removes the table and the cleanup job. In exchange, single use comes from the password change rather than from a flag, and everything depends on the server’s secret key. Pick one design and test it with the same attacks.
  • Old rows pile up. The table never deletes anything. Add a job that removes tokens that expired days ago.

Confirm everything works

You are done when three commands agree. The attack run reports that none of the twelve succeeded and the happy path works, the unit tests pass, and the mutation matrix shows each defense bringing back its own attack. In my run the three results were 0 of 12 attacks succeeded against app, 15 passed, and the matrix above.

Where to go next

Tags:

Application SecurityAuthenticationFastAPIPassword ResetpytestPython

Share

No Comment! Be the first one.

Leave a Reply

Your email address will not be published. Required fields are marked *

Related Posts