Ben Stull
8f3a8ec33a
After a successful OTC sign-in, a contributor can set a passcode (4-20 characters, bcrypt-hashed) and use email + passcode for subsequent sign-ins. OTC remains the structural fallback: five consecutive failed verifies lock the passcode path for 15 minutes (HTTP 423), and a forgotten passcode is recovered by requesting a fresh code. Migration 015_passcode.sql adds four nullable columns to the users table; existing rows pass through as OTC-only and can opt into a passcode from a new Sign-in tab in /settings/notifications. The /login surface is extended to a five-step flow (email → either passcode or OTC code → optional post-OTC passcode offer → optional set-passcode). SPEC corrections per §19.3 rule 2: §6 names the three auth paths, §14.1 documents the stepped login flow, §17 lists the four new /auth/passcode/* endpoints, §19.2 surfaces four new candidates and refreshes the cross-refs. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
533 lines
19 KiB
Python
533 lines
19 KiB
Python
"""End-to-end integration tests for the v0.10.0 user-set passcode
|
|
vertical (§6.2, roadmap item #8).
|
|
|
|
After a successful OTC sign-in the user can set a passcode and use
|
|
email + passcode for subsequent sign-ins. OTC remains the structural
|
|
fallback — these tests prove:
|
|
|
|
* `/auth/passcode/set` requires an active session.
|
|
* `/auth/passcode/check` returns `has_passcode` without leaking the
|
|
hash, the set-at stamp, or the lockout state.
|
|
* Happy path: OTC sign-in → set passcode → sign out → email +
|
|
passcode signs in (no OTC roundtrip).
|
|
* Wrong passcode increments the failure counter without locking.
|
|
* Five consecutive failures lock the passcode path (HTTP 423) and
|
|
persist `passcode_locked_until` on the user row.
|
|
* The lockout expires after `passcode_locked_until`; a verify
|
|
attempt past the window succeeds again and clears the counter.
|
|
* The OTC path is unaffected by the passcode lockout — a user
|
|
whose passcode is locked can still request and verify a fresh
|
|
OTC to sign in.
|
|
* Clearing the passcode wipes the hash; subsequent verify refuses
|
|
with the no-passcode failure shape.
|
|
* Setting a new passcode replaces the prior one (and resets the
|
|
failure counter / lockout state).
|
|
* `passcode_set_at` updates on every set call.
|
|
* The validation denylist refuses obvious patterns (e.g. `0000`,
|
|
`1234`).
|
|
* Passcode length is enforced (4-20).
|
|
|
|
The fakes from `test_propose_vertical` give us a working app harness.
|
|
The OTC envelope buffer from `test_otc_vertical` is reused for the
|
|
OTC roundtrips this suite needs.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from test_propose_vertical import ( # noqa: F401
|
|
FakeGitea,
|
|
app_with_fake_gitea,
|
|
provision_user_row,
|
|
tmp_env,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers — mirror the OTC suite's outbound-buffer helpers.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _reset_outbound():
|
|
from app import email as email_mod
|
|
email_mod.reset_sent_envelopes()
|
|
|
|
|
|
def _outbound_otc_codes(to_address: str | None = None) -> list[str]:
|
|
from app import email as email_mod
|
|
out = []
|
|
for env in email_mod.sent_envelopes():
|
|
if env.get("kind") != "otc":
|
|
continue
|
|
if to_address is not None and env["to"] != to_address:
|
|
continue
|
|
for line in env["body"].splitlines():
|
|
tok = line.strip()
|
|
if tok.isdigit() and len(tok) == 6:
|
|
out.append(tok)
|
|
break
|
|
return out
|
|
|
|
|
|
def _sign_in_via_otc(client, email: str) -> None:
|
|
"""Run an OTC request+verify so the client carries an authenticated
|
|
session. The cooldown is irrelevant on a fresh email; we don't
|
|
need to drop it."""
|
|
r = client.post("/auth/otc/request", json={"email": email})
|
|
assert r.status_code == 200, r.text
|
|
code = _outbound_otc_codes(email)[-1]
|
|
r = client.post("/auth/otc/verify", json={"email": email, "code": code})
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Set passcode — auth-required, happy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_set_passcode_requires_session(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
r = client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_set_passcode_after_otc_landing_persists_hash(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "alice@example.com")
|
|
|
|
r = client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
assert r.status_code == 200, r.text
|
|
|
|
row = db.conn().execute(
|
|
"SELECT passcode_hash, passcode_set_at FROM users WHERE email = ? COLLATE NOCASE",
|
|
("alice@example.com",),
|
|
).fetchone()
|
|
assert row is not None
|
|
assert row["passcode_hash"] is not None
|
|
# Not the plaintext.
|
|
assert row["passcode_hash"] != "secret123"
|
|
assert row["passcode_set_at"] is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Check endpoint — leak-free shape
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_check_endpoint_returns_false_for_unknown_email(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
r = client.get("/auth/passcode/check", params={"email": "nobody@example.com"})
|
|
assert r.status_code == 200
|
|
assert r.json() == {"has_passcode": False}
|
|
|
|
|
|
def test_check_endpoint_returns_false_for_user_without_passcode(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "bob@example.com")
|
|
|
|
r = client.get("/auth/passcode/check", params={"email": "bob@example.com"})
|
|
assert r.status_code == 200
|
|
assert r.json() == {"has_passcode": False}
|
|
|
|
|
|
def test_check_endpoint_returns_true_after_set(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "carol@example.com")
|
|
client.post("/auth/passcode/set", json={"passcode": "letmein9"})
|
|
|
|
# Drop the session so the check is read in the anonymous shape.
|
|
client.cookies.clear()
|
|
r = client.get("/auth/passcode/check", params={"email": "carol@example.com"})
|
|
assert r.status_code == 200
|
|
assert r.json() == {"has_passcode": True}
|
|
# The response carries ONLY the boolean — no hash, no stamp.
|
|
assert set(r.json().keys()) == {"has_passcode"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Verify path — happy path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_verify_passcode_signs_in_user(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "dave@example.com")
|
|
client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
client.cookies.clear()
|
|
|
|
r = client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "dave@example.com", "passcode": "secret123"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
me = client.get("/api/auth/me").json()
|
|
assert me["authenticated"] is True
|
|
assert me["user"]["email"] == "dave@example.com"
|
|
assert me["user"]["has_passcode"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Verify path — failure modes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_verify_passcode_wrong_increments_counter_without_locking(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "erin@example.com")
|
|
client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
client.cookies.clear()
|
|
|
|
# Three bad attempts — under the lockout threshold.
|
|
for _ in range(3):
|
|
r = client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "erin@example.com", "passcode": "wrongwrong"},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
row = db.conn().execute(
|
|
"SELECT passcode_failed_attempts, passcode_locked_until FROM users WHERE email = ?",
|
|
("erin@example.com",),
|
|
).fetchone()
|
|
assert row["passcode_failed_attempts"] == 3
|
|
assert row["passcode_locked_until"] is None
|
|
|
|
|
|
def test_verify_passcode_locks_after_five_failures(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "frank@example.com")
|
|
client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
client.cookies.clear()
|
|
|
|
# Five bad attempts — the last crosses the threshold and the
|
|
# response shape flips to 423.
|
|
statuses = []
|
|
for _ in range(5):
|
|
r = client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "frank@example.com", "passcode": "wrongwrong"},
|
|
)
|
|
statuses.append(r.status_code)
|
|
# First four are 400, the fifth (threshold-crossing) is 423.
|
|
assert statuses == [400, 400, 400, 400, 423]
|
|
|
|
row = db.conn().execute(
|
|
"SELECT passcode_failed_attempts, passcode_locked_until FROM users WHERE email = ?",
|
|
("frank@example.com",),
|
|
).fetchone()
|
|
assert row["passcode_failed_attempts"] >= 5
|
|
assert row["passcode_locked_until"] is not None
|
|
|
|
# Sixth attempt — still locked, still 423, even with the correct
|
|
# passcode (lockout overrides the verify).
|
|
r = client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "frank@example.com", "passcode": "secret123"},
|
|
)
|
|
assert r.status_code == 423
|
|
|
|
|
|
def test_verify_passcode_lockout_expires(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "gina@example.com")
|
|
client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
client.cookies.clear()
|
|
|
|
for _ in range(5):
|
|
client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "gina@example.com", "passcode": "wrongwrong"},
|
|
)
|
|
|
|
# Backdate the lockout to the past so the next attempt clears it.
|
|
db.conn().execute(
|
|
"""
|
|
UPDATE users
|
|
SET passcode_locked_until = datetime('now', '-1 minute')
|
|
WHERE email = ?
|
|
""",
|
|
("gina@example.com",),
|
|
)
|
|
|
|
r = client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "gina@example.com", "passcode": "secret123"},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
|
|
# Lockout cleared, counter reset.
|
|
row = db.conn().execute(
|
|
"SELECT passcode_failed_attempts, passcode_locked_until FROM users WHERE email = ?",
|
|
("gina@example.com",),
|
|
).fetchone()
|
|
assert row["passcode_failed_attempts"] == 0
|
|
assert row["passcode_locked_until"] is None
|
|
|
|
|
|
def test_otc_path_unaffected_by_passcode_lockout(app_with_fake_gitea, monkeypatch):
|
|
from fastapi.testclient import TestClient
|
|
|
|
# Drop the OTC cooldown so the second request lands without a 429.
|
|
# The cooldown is re-read from env on every `request_code` call so
|
|
# this takes effect mid-process.
|
|
monkeypatch.setenv("OTC_REQUEST_COOLDOWN_SECONDS", "0")
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "harvey@example.com")
|
|
client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
client.cookies.clear()
|
|
|
|
# Lock the passcode path.
|
|
for _ in range(5):
|
|
client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "harvey@example.com", "passcode": "wrongwrong"},
|
|
)
|
|
|
|
# The OTC path is unaffected by the passcode lockout: the user
|
|
# can still request and verify a fresh code to sign in.
|
|
r = client.post("/auth/otc/request", json={"email": "harvey@example.com"})
|
|
assert r.status_code == 200
|
|
code = _outbound_otc_codes("harvey@example.com")[-1]
|
|
r = client.post("/auth/otc/verify", json={"email": "harvey@example.com", "code": code})
|
|
assert r.status_code == 200
|
|
|
|
# The user is now signed in via OTC even though the passcode
|
|
# path is locked. The /api/auth/me payload reflects this.
|
|
me = client.get("/api/auth/me").json()
|
|
assert me["authenticated"] is True
|
|
assert me["user"]["email"] == "harvey@example.com"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Clear + replace
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_clear_passcode_wipes_the_hash(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "ivy@example.com")
|
|
client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
|
|
r = client.delete("/auth/passcode")
|
|
assert r.status_code == 200
|
|
|
|
row = db.conn().execute(
|
|
"SELECT passcode_hash, passcode_set_at FROM users WHERE email = ?",
|
|
("ivy@example.com",),
|
|
).fetchone()
|
|
assert row["passcode_hash"] is None
|
|
assert row["passcode_set_at"] is None
|
|
|
|
# Verify against the cleared passcode refuses (no-passcode shape
|
|
# collapses to a generic 400).
|
|
client.cookies.clear()
|
|
r = client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "ivy@example.com", "passcode": "secret123"},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_setting_new_passcode_replaces_old(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "jane@example.com")
|
|
client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
# Replace.
|
|
r = client.post("/auth/passcode/set", json={"passcode": "newsecret9"})
|
|
assert r.status_code == 200
|
|
|
|
client.cookies.clear()
|
|
|
|
# Old passcode refuses.
|
|
r = client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "jane@example.com", "passcode": "secret123"},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
# New passcode signs in.
|
|
r = client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "jane@example.com", "passcode": "newsecret9"},
|
|
)
|
|
assert r.status_code == 200
|
|
|
|
|
|
def test_setting_new_passcode_resets_lockout(app_with_fake_gitea, monkeypatch):
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
|
|
monkeypatch.setenv("OTC_REQUEST_COOLDOWN_SECONDS", "0")
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "kate@example.com")
|
|
client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
|
|
# Lock the passcode path with bad attempts (drop session first).
|
|
client.cookies.clear()
|
|
for _ in range(5):
|
|
client.post(
|
|
"/auth/passcode/verify",
|
|
json={"email": "kate@example.com", "passcode": "wrongwrong"},
|
|
)
|
|
|
|
row = db.conn().execute(
|
|
"SELECT passcode_locked_until FROM users WHERE email = ?",
|
|
("kate@example.com",),
|
|
).fetchone()
|
|
assert row["passcode_locked_until"] is not None
|
|
|
|
# Sign back in via OTC and reset the passcode.
|
|
r = client.post("/auth/otc/request", json={"email": "kate@example.com"})
|
|
assert r.status_code == 200
|
|
code = _outbound_otc_codes("kate@example.com")[-1]
|
|
r = client.post("/auth/otc/verify", json={"email": "kate@example.com", "code": code})
|
|
assert r.status_code == 200
|
|
|
|
r = client.post("/auth/passcode/set", json={"passcode": "freshcode9"})
|
|
assert r.status_code == 200
|
|
|
|
# Lockout cleared on set.
|
|
row = db.conn().execute(
|
|
"SELECT passcode_locked_until, passcode_failed_attempts FROM users WHERE email = ?",
|
|
("kate@example.com",),
|
|
).fetchone()
|
|
assert row["passcode_locked_until"] is None
|
|
assert row["passcode_failed_attempts"] == 0
|
|
|
|
|
|
def test_passcode_set_at_updates_on_each_set(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
from app import db
|
|
import time
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "luke@example.com")
|
|
|
|
client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
first_stamp = db.conn().execute(
|
|
"SELECT passcode_set_at FROM users WHERE email = ?",
|
|
("luke@example.com",),
|
|
).fetchone()["passcode_set_at"]
|
|
assert first_stamp is not None
|
|
|
|
# SQLite's datetime('now') has second precision; sleep so the
|
|
# stamp visibly advances on the next set.
|
|
time.sleep(1.1)
|
|
|
|
client.post("/auth/passcode/set", json={"passcode": "newcode99"})
|
|
second_stamp = db.conn().execute(
|
|
"SELECT passcode_set_at FROM users WHERE email = ?",
|
|
("luke@example.com",),
|
|
).fetchone()["passcode_set_at"]
|
|
assert second_stamp is not None
|
|
assert second_stamp >= first_stamp
|
|
# Lexicographic compare on ISO-8601 datetime strings works for
|
|
# the SQLite shape.
|
|
assert second_stamp > first_stamp
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_set_passcode_refuses_too_short(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "mia@example.com")
|
|
r = client.post("/auth/passcode/set", json={"passcode": "abc"})
|
|
assert r.status_code == 422
|
|
|
|
|
|
def test_set_passcode_refuses_denylist_pattern(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "nick@example.com")
|
|
for bad in ["0000", "1234", "aaaa", "qwerty", "password"]:
|
|
r = client.post("/auth/passcode/set", json={"passcode": bad})
|
|
assert r.status_code == 422, f"expected 422 for {bad!r}, got {r.status_code}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth me payload
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_auth_me_carries_has_passcode_flag(app_with_fake_gitea):
|
|
from fastapi.testclient import TestClient
|
|
|
|
app, _fake = app_with_fake_gitea
|
|
with TestClient(app) as client:
|
|
_reset_outbound()
|
|
_sign_in_via_otc(client, "olga@example.com")
|
|
|
|
me = client.get("/api/auth/me").json()
|
|
assert me["user"]["has_passcode"] is False
|
|
assert me["user"]["passcode_set_at"] is None
|
|
|
|
client.post("/auth/passcode/set", json={"passcode": "secret123"})
|
|
me = client.get("/api/auth/me").json()
|
|
assert me["user"]["has_passcode"] is True
|
|
assert me["user"]["passcode_set_at"] is not None
|