"""The second factor, and the ways back in when it goes wrong.
A second factor is only worth having if it cannot be walked around, so most of
these tests are about the walk-arounds:
* a password alone must not reach a single page — the half-authenticated
session is refused at `load_session`, the one place every route resolves a
cookie, rather than by each route remembering to check;
* the code stage must be inside the lockout, or the factor is a six-digit
secret that anyone holding the password can brute-force at leisure;
* enrolment must be confirmed with a real code, or people switch it on with a
secret they never scanned and lock themselves out;
* every recovery path (recovery codes, an admin, the CLI) must actually work,
because the failure mode is "you can never see your own portfolio again".
TOTP codes are generated with pyotp here rather than stubbed: the point is that
a real authenticator's output is accepted, and a stub would only prove the code
calls itself.
"""
from __future__ import annotations
import datetime as dt
import re
import pyotp
import pytest
from freezegun import freeze_time
from sqlalchemy import select
import factories as fac
from app import auth as auth_mod
from app import twofactor
from app.models import RecoveryCode, User, UserSession
from test_routes import PASSWORD, make_login, pre_auth_csrf, session_csrf
HTML = {"accept": "text/html"}
def code_for(secret: str) -> str:
return pyotp.TOTP(secret).now()
@pytest.fixture
def enrolled(client, session_factory):
"""A signed-in user with 2FA already on. Returns their secret."""
make_login(client, session_factory)
secret = pyotp.random_base32()
with session_factory() as s:
user = s.scalars(select(User)).one()
twofactor.enable(s, user, secret)
s.commit()
return secret
# --------------------------------------------------------------------------- #
# The primitives
# --------------------------------------------------------------------------- #
def test_a_current_code_verifies(pf):
secret = twofactor.new_secret()
assert twofactor.matching_step(secret, code_for(secret)) is not None
def test_a_wrong_code_does_not(pf):
secret = twofactor.new_secret()
assert twofactor.matching_step(secret, "000000") is None
@pytest.mark.parametrize("code", ["", " ", "abcdef", "12345", None, "²²²²²²"])
def test_junk_is_rejected_without_reaching_the_library(code):
"""Empty and non-numeric input is refused here rather than letting pyotp
decide what an empty code means."""
assert twofactor.matching_step("JBSWY3DPEHPK3PXP", code) is None
def test_a_code_with_spaces_in_it_still_works():
"""People paste "123 456" out of their phone."""
secret = twofactor.new_secret()
spaced = code_for(secret)
assert twofactor.matching_step(secret, f"{spaced[:3]} {spaced[3:]}") is not None
def test_no_secret_means_no_code_is_ever_valid():
"""A user who never enrolled must not be verifiable by anything."""
assert twofactor.matching_step(None, "000000") is None
assert twofactor.matching_step("", "000000") is None
@pytest.mark.parametrize("offset, accepted", [(-2, False), (-1, True), (0, True), (1, True),
(2, False)])
def test_a_code_is_good_for_its_own_step_and_one_either_side(offset, accepted):
"""Phones drift either way. A 30-second window either side is the standard
trade-off: stricter generates support requests, looser widens replay for
no gain. Built from step numbers rather than from "now minus 90 seconds",
which is three steps back, not two."""
secret = twofactor.new_secret()
totp = pyotp.TOTP(secret)
with freeze_time("2026-08-02 12:00:10"): # one clock for both sides
now = totp.timecode(dt.datetime.now(dt.timezone.utc))
step = twofactor.matching_step(secret, totp.generate_otp(now + offset))
assert (step == now + offset) if accepted else step is None
# --------------------------------------------------------------------------- #
# Replay — GHSA-28h6-x798-2qv5
#
# A code is valid for about 90 seconds, so without a record of what was used,
# anyone who saw one go by could use it again in that time. Frozen clocks
# throughout: these are about which STEP a code belongs to, and a test that
# straddled a 30-second boundary would be testing the boundary instead.
# --------------------------------------------------------------------------- #
NOW = "2026-10-03 01:00:10"
def frozen_code(secret: str) -> str:
"""The current code under a frozen clock. Not `code_for`: pyotp's `now()`
reads a naive datetime and converts it as LOCAL time, and freezegun's
naive now is UTC — so off a UTC machine it is a code for the wrong step."""
return pyotp.TOTP(secret).at(dt.datetime.now(dt.timezone.utc))
def _user_with_secret(pf):
user = fac.make_user(pf, "replay@example.test")
user.totp_secret = twofactor.new_secret()
pf.flush()
return user
def test_a_code_is_accepted_once(pf):
user = _user_with_secret(pf)
with freeze_time(NOW):
code = frozen_code(user.totp_secret)
assert twofactor.accept_code(pf, user, code)
assert not twofactor.accept_code(pf, user, code), "the same code was accepted twice"
def test_an_earlier_code_is_refused_once_a_later_one_was_used(pf):
"""The previous step's code is still inside the drift window, but anyone
holding it saw it before the one just used."""
user = _user_with_secret(pf)
totp = pyotp.TOTP(user.totp_secret)
with freeze_time(NOW):
now = dt.datetime.now(dt.timezone.utc)
assert twofactor.accept_code(pf, user, totp.at(now))
assert not twofactor.accept_code(pf, user, totp.at(now - dt.timedelta(seconds=30)))
def test_the_next_code_is_still_accepted(pf):
"""Refusing replays must not refuse the person: the code after the one they
just used works."""
user = _user_with_secret(pf)
totp = pyotp.TOTP(user.totp_secret)
with freeze_time(NOW):
now = dt.datetime.now(dt.timezone.utc)
assert twofactor.accept_code(pf, user, totp.at(now - dt.timedelta(seconds=30)))
assert twofactor.accept_code(pf, user, totp.at(now))
def test_two_requests_racing_with_one_code_cannot_both_pass(session_factory):
"""The relay case: the same code arrives twice before either request has
committed. Each loaded the user before the other wrote, so a read-then-write
check would pass both. The UPDATE's own condition is what refuses one."""
with session_factory() as s:
user = _user_with_secret(s)
s.commit()
user_id = user.id
with freeze_time(NOW), session_factory() as first, session_factory() as second:
mine = first.get(User, user_id)
theirs = second.get(User, user_id)
code = frozen_code(mine.totp_secret)
assert twofactor.accept_code(first, mine, code)
first.commit()
assert not twofactor.accept_code(second, theirs, code), "both racers got in"
def test_a_new_secret_forgets_the_old_ones_steps(pf):
"""A fresh enrolment's first code can share a step with the last code from
the old secret, and must not be refused for it."""
user = _user_with_secret(pf)
with freeze_time(NOW):
assert twofactor.accept_code(pf, user, frozen_code(user.totp_secret))
user.totp_enabled_at = dt.datetime.now(dt.timezone.utc)
twofactor.disable(pf, user)
assert user.totp_last_step is None
twofactor.begin_enrolment(user)
assert twofactor.accept_code(pf, user, frozen_code(user.totp_secret))
def test_issuing_a_secret_starts_with_no_step_recorded(pf):
"""Wherever the secret is replaced, the record goes with it — not only on
the disable path that happens to come first today."""
user = _user_with_secret(pf)
user.totp_enabled_at = dt.datetime.now(dt.timezone.utc)
user.totp_last_step = 99_999_999
old = user.totp_secret
twofactor.begin_enrolment(user)
assert user.totp_secret != old
assert user.totp_last_step is None
def test_a_code_used_to_sign_in_cannot_sign_in_again(client, session_factory, enrolled):
"""End to end, through the login form: somebody who watched the code go in
and also has the password still does not get a session out of it."""
code = code_for(enrolled)
client.cookies.clear()
login_password_only(client)
first = client.post("/login/code", data={"code": code, "_csrf": pre_auth_csrf(client)},
headers=HTML, follow_redirects=False)
assert first.headers["location"] == "/"
client.cookies.clear()
login_password_only(client)
again = client.post("/login/code", data={"code": code, "_csrf": pre_auth_csrf(client)},
headers=HTML, follow_redirects=False)
assert again.status_code == 200
assert "isn't right" in again.text or "isn't right" in again.text
assert client.get("/", headers=HTML, follow_redirects=False).status_code == 303
def test_the_provisioning_uri_names_this_install(pf):
secret = twofactor.new_secret()
uri = twofactor.provisioning_uri(secret, "someone@example.test", "portfolio")
assert uri.startswith("otpauth://totp/")
assert "portfolio" in uri
assert secret in uri
def test_the_qr_is_rendered_locally_as_inline_svg(pf):
"""The obvious shortcut is a chart URL, which would post the TOTP secret to
a third party in a query string."""
svg = twofactor.qr_svg("otpauth://totp/x?secret=JBSWY3DPEHPK3PXP")
assert svg.lstrip().startswith("" in page
assert 'name="secret"' not in page
def test_a_secret_posted_with_the_form_is_ignored(client, session_factory):
"""Only the secret this account was issued counts.
While it rode in the form, anything posted alongside a matching code turned
2FA on — so a caller could enrol an authenticator this account had never
scanned, and the real owner would be locked behind somebody else's phone.
Holding it on the row closed that; this is what keeps it closed.
"""
make_login(client, session_factory)
client.get("/profile/2fa", headers=HTML) # issues the real one
theirs = pyotp.random_base32()
client.post("/profile/2fa/enable", headers=HTML,
data={"secret": theirs, "code": code_for(theirs),
"_csrf": session_csrf(session_factory)})
with session_factory() as s:
user = s.scalars(select(User)).one()
assert not twofactor.is_enabled(user)
assert user.totp_secret != theirs
def test_returning_to_the_setup_page_shows_the_same_secret(client, session_factory):
"""The bug this replaced: a secret generated per render meant a refresh —
or anything else that re-fetched the page — silently invalidated the QR
somebody had just scanned, and their code stopped matching with nothing on
screen to explain it. Found in sundries first.
"""
make_login(client, session_factory)
first = client.get("/profile/2fa", headers=HTML).text
again = client.get("/profile/2fa", headers=HTML).text
key = re.search(r"([A-Z2-7]+)", first).group(1)
assert f"{key}" in again
def test_enabling_needs_a_code_generated_from_that_secret(client, session_factory):
"""Skipping this check is how people switch 2FA on with a secret they never
successfully scanned, and lock themselves out of their own portfolio."""
make_login(client, session_factory)
secret = pyotp.random_base32()
resp = client.post("/profile/2fa/enable",
data={"secret": secret, "code": "000000",
"_csrf": session_csrf(session_factory)},
headers=HTML, follow_redirects=False)
assert "error=" in resp.headers["location"]
with session_factory() as s:
assert not twofactor.is_enabled(s.scalars(select(User)).one())
def test_enabling_with_a_real_code_turns_it_on_and_shows_the_codes_once(client, session_factory):
make_login(client, session_factory)
# Through the page, because that is what issues the secret. Posting one of
# its own used to work, which is what let a caller enrol an authenticator
# this account had never scanned.
page = client.get("/profile/2fa", headers=HTML).text
secret = re.search(r"([A-Z2-7]+)", page).group(1)
resp = client.post("/profile/2fa/enable",
data={"code": code_for(secret),
"_csrf": session_csrf(session_factory)},
headers=HTML)
assert "only time these are shown" in resp.text
with session_factory() as s:
user = s.scalars(select(User)).one()
assert twofactor.is_enabled(user)
assert twofactor.remaining_recovery_codes(s, user) == twofactor.RECOVERY_CODE_COUNT
# And they are not shown again on a later visit.
assert "only time these are shown" not in client.get("/profile", headers=HTML).text
def test_an_abandoned_enrolment_leaves_nothing_ENABLED(client, session_factory):
"""It leaves a secret now — that is what makes the QR survive a refresh —
but nothing that can be signed in with.
`is_enabled` wants `totp_enabled_at` as well, so an enrolment nobody
finished is inert: the login flow never asks for a code, and the account is
exactly as it was.
"""
make_login(client, session_factory)
client.get("/profile/2fa", headers=HTML)
with session_factory() as s:
user = s.scalars(select(User)).one()
assert user.totp_secret is not None # so returning shows the same QR
assert user.totp_enabled_at is None # and it turns nothing on
assert twofactor.is_enabled(user) is False
def test_turning_it_off_needs_both_the_password_and_a_code(client, session_factory, enrolled):
for data in (
{"password": "wrong-password-entirely", "code": code_for(enrolled)},
{"password": PASSWORD, "code": "000000"},
):
resp = client.post("/profile/2fa/disable",
data={**data, "_csrf": session_csrf(session_factory)},
headers=HTML, follow_redirects=False)
assert "error=" in resp.headers["location"]
with session_factory() as s:
assert twofactor.is_enabled(s.scalars(select(User)).one())
def test_turning_it_off_with_both_works(client, session_factory, enrolled):
resp = client.post("/profile/2fa/disable",
data={"password": PASSWORD, "code": code_for(enrolled),
"_csrf": session_csrf(session_factory)},
headers=HTML, follow_redirects=False)
assert resp.headers["location"] == "/profile?saved=2fa-off"
with session_factory() as s:
assert not twofactor.is_enabled(s.scalars(select(User)).one())
def test_a_recovery_code_also_turns_it_off(client, session_factory, enrolled):
"""Someone whose phone is gone still needs to be able to switch it off."""
with session_factory() as s:
codes = twofactor.generate_recovery_codes(s, s.scalars(select(User)).one())
s.commit()
resp = client.post("/profile/2fa/disable",
data={"password": PASSWORD, "code": codes[0],
"_csrf": session_csrf(session_factory)},
headers=HTML, follow_redirects=False)
assert resp.headers["location"] == "/profile?saved=2fa-off"
# --------------------------------------------------------------------------- #
# Recovery: admin, and the command line
# --------------------------------------------------------------------------- #
def test_an_admin_can_clear_someone_elses_2fa(client, session_factory):
make_login(client, session_factory, admin=True)
with session_factory() as s:
other = fac.make_user(s, "other@example.test",
password_hash=auth_mod.hash_password(PASSWORD))
twofactor.enable(s, other, twofactor.new_secret())
s.commit()
other_id = other.id
token = session_csrf(session_factory)
with session_factory() as s:
# A session to end: without one, "their sessions went" was true anyway.
from app.settings import PortfolioSettings
auth_mod.create_session(s, s.get(User, other_id), PortfolioSettings())
s.commit()
resp = client.post(f"/users/{other_id}/2fa/clear", data={"_csrf": token},
headers=HTML, follow_redirects=False)
assert resp.headers["location"] == "/users?reset=2fa"
with session_factory() as s:
assert not twofactor.is_enabled(s.get(User, other_id))
# Their sessions went too — the account's security just changed.
assert s.scalars(
select(UserSession).where(UserSession.user_id == other_id)
).all() == []
def test_a_non_admin_cannot_clear_anyone_elses_2fa(client, session_factory):
make_login(client, session_factory, admin=False)
with session_factory() as s:
other = fac.make_user(s, "other@example.test")
twofactor.enable(s, other, twofactor.new_secret())
s.commit()
other_id = other.id
resp = client.post(f"/users/{other_id}/2fa/clear",
data={"_csrf": session_csrf(session_factory)}, headers=HTML)
assert resp.status_code == 403
with session_factory() as s:
assert twofactor.is_enabled(s.get(User, other_id))
def _methods_table(client) -> str:
page = client.get("/profile", headers=HTML).text
return re.search(r'