peter
59ec4a28ba
The session held a wallet of up to six accounts, and a request picked one by a header or an ?account= parameter. It now holds one: user_id and its epoch. A sign-in replaces whoever was signed in, sign-out and account deletion end the session, and an account refused mid-session (revoked, suspended, deactivated) clears it. /api/me/accounts and /api/me/switch are gone, and with them their refusal codes and the UI copy for them. A cookie the multi-account build wrote carries no single epoch and is refused: its holder signs in once more. A Google link finishing in a browser now signed in as someone else is refused, as before for an account outside the wallet. A test holds that no app code reads the wallet keys or the account header again.
666 lines
28 KiB
Python
666 lines
28 KiB
Python
"""Auth / session security core (TEST-1), DB-lane.
|
|
|
|
One account per browser session, epoch-based session revocation, single-use tokens and the
|
|
enumeration-safe uniform 401 are the security mag. Starlette's SessionMiddleware round-trips the
|
|
signed cookie under TestClient, so the whole browser-less flow is exercisable. Uses the real auth
|
|
stack via the `client` fixture (NO current_user override) so login, the session and epoch revoke are
|
|
the code under test.
|
|
|
|
Note: `/api/me` is the bootstrap probe — it returns 200 `{authenticated: False}` when logged out, not a
|
|
401. A revoked session is therefore checked on a PROTECTED endpoint (`/api/feed`), which does 401.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from itsdangerous import TimestampSigner
|
|
|
|
from app.config import settings
|
|
from app.main import app
|
|
|
|
_PW = "correct horse battery"
|
|
|
|
# The `client`/`as_user` fixtures bake in the post-setup `_configured` state, so no per-test patch here.
|
|
|
|
|
|
def _login(client, email, password):
|
|
return client.post("/auth/password-login", json={"email": email, "password": password})
|
|
|
|
|
|
def test_password_login_then_me_is_authenticated(client, make_user):
|
|
make_user(email="alice@example.com", password=_PW)
|
|
assert _login(client, "alice@example.com", _PW).status_code == 200
|
|
me = client.get("/api/me")
|
|
assert me.status_code == 200
|
|
assert me.json()["authenticated"] is True
|
|
assert me.json()["email"] == "alice@example.com"
|
|
|
|
|
|
def test_wrong_password_and_unknown_email_give_a_byte_identical_401(client, make_user):
|
|
make_user(email="bob@example.com", password=_PW)
|
|
wrong_pw = _login(client, "bob@example.com", "not the password")
|
|
unknown = _login(client, "nobody@example.com", _PW)
|
|
assert wrong_pw.status_code == 401 and unknown.status_code == 401
|
|
# Identical body: a wrong password and a non-existent account must be indistinguishable (no
|
|
# account-enumeration oracle).
|
|
assert wrong_pw.json() == unknown.json()
|
|
|
|
|
|
def _cookie(payload: dict) -> str:
|
|
"""A session cookie signed the way SessionMiddleware signs one — for a cookie an older build wrote."""
|
|
data = base64.b64encode(json.dumps(payload).encode())
|
|
return TimestampSigner(str(settings.secret_key)).sign(data).decode()
|
|
|
|
|
|
def test_signing_in_as_another_account_replaces_the_first(client, make_user):
|
|
# One account per browser session: a second sign-in is a fresh session, not a second account
|
|
# beside the first — so signing out leaves nobody signed in here.
|
|
make_user(email="alice3@example.com", password=_PW)
|
|
make_user(email="bob3@example.com", password=_PW)
|
|
assert _login(client, "alice3@example.com", _PW).status_code == 200
|
|
assert _login(client, "bob3@example.com", _PW).status_code == 200
|
|
assert client.get("/api/me").json()["email"] == "bob3@example.com"
|
|
|
|
assert client.post("/auth/logout").status_code == 200
|
|
assert client.get("/api/feed").status_code == 401
|
|
assert client.get("/api/me").json() == {"authenticated": False}
|
|
|
|
|
|
def test_no_header_or_parameter_picks_another_account(client, make_user):
|
|
# Alice signed in here before Bob: nothing a request carries makes it act as her any more.
|
|
alice = make_user(email="alice4@example.com", password=_PW)
|
|
make_user(email="bob4@example.com", password=_PW)
|
|
assert _login(client, "alice4@example.com", _PW).status_code == 200
|
|
assert _login(client, "bob4@example.com", _PW).status_code == 200
|
|
me = client.get(f"/api/me?account={alice.id}", headers={"X-Siftlode-Account": str(alice.id)})
|
|
assert me.json()["email"] == "bob4@example.com"
|
|
|
|
|
|
@pytest.mark.parametrize("path", ["/api/me/accounts", "/api/me/switch"])
|
|
def test_there_is_no_account_list_or_switch(client, make_user, path):
|
|
make_user(email="alice5@example.com", password=_PW)
|
|
assert _login(client, "alice5@example.com", _PW).status_code == 200
|
|
r = client.get(path) if path.endswith("accounts") else client.post(path, json={"user_id": 1})
|
|
assert r.status_code in (404, 405)
|
|
|
|
|
|
@pytest.mark.parametrize("ended", ["revoked", "suspended", "deactivated"])
|
|
def test_a_session_whose_account_ended_signs_the_browser_out(client, make_user, db, ended):
|
|
alice = make_user(email="alice6@example.com", password=_PW)
|
|
assert _login(client, "alice6@example.com", _PW).status_code == 200
|
|
if ended == "revoked":
|
|
alice.session_epoch = (alice.session_epoch or 0) + 1
|
|
elif ended == "suspended":
|
|
alice.is_suspended = True
|
|
else:
|
|
alice.is_active = False
|
|
db.commit()
|
|
assert client.get("/api/feed").status_code == 401
|
|
assert "user_id" not in _session(client) # the refusal ended the session, not just this request
|
|
assert client.get("/api/me").json() == {"authenticated": False}
|
|
|
|
|
|
def test_a_cookie_from_the_multi_account_build_signs_in_again(client, make_user):
|
|
# The old session held a wallet and one epoch per account. It is not read: whoever holds one
|
|
# signs in once more.
|
|
alice = make_user(email="alice7@example.com", password=_PW)
|
|
old = {"user_id": alice.id, "account_ids": [alice.id], "epochs": {str(alice.id): 0}}
|
|
client.cookies.set("session", _cookie(old))
|
|
assert client.get("/api/feed").status_code == 401
|
|
assert client.get("/api/me").json() == {"authenticated": False}
|
|
|
|
|
|
def test_deleting_the_account_signs_the_browser_out(client, make_user):
|
|
make_user(email="alice8@example.com", password=_PW)
|
|
assert _login(client, "alice8@example.com", _PW).status_code == 200
|
|
assert client.delete("/api/me/account").status_code == 200
|
|
# The cookie names nobody now: the probe answers signed out without looking the account up.
|
|
assert "user_id" not in _session(client)
|
|
|
|
|
|
def _session(client) -> dict:
|
|
"""The decoded payload of the client's session cookie."""
|
|
raw = client.cookies.get("session")
|
|
if not raw:
|
|
return {}
|
|
data = TimestampSigner(str(settings.secret_key)).unsign(raw.encode())
|
|
return json.loads(base64.b64decode(data))
|
|
|
|
|
|
@pytest.mark.parametrize("signed_in_as", ["starter", "someone_else"])
|
|
def test_a_google_link_lands_only_on_the_account_still_signed_in(db, make_user, signed_in_as):
|
|
# The link marker names who started it; a sign-in as someone else before the callback must not
|
|
# get that Google identity attached to the account that started it from the other person's session.
|
|
from starlette.requests import Request
|
|
|
|
from app.routes.auth import _complete_link
|
|
|
|
starter, other = make_user(), make_user()
|
|
session = {"user_id": starter.id if signed_in_as == "starter" else other.id}
|
|
request = Request({"type": "http", "session": session, "headers": []})
|
|
resp = _complete_link(request, db, starter.id, True, "sub-123", {"email": ""}, {"scope": ""})
|
|
db.refresh(starter)
|
|
if signed_in_as == "starter":
|
|
assert resp.headers["location"] == "/?link=ok"
|
|
assert starter.google_sub == "sub-123"
|
|
else:
|
|
assert resp.headers["location"] == "/?link=error"
|
|
assert starter.google_sub is None
|
|
|
|
|
|
def test_no_app_code_reads_a_second_account():
|
|
# The multi-account machinery is gone; a module reading its session keys or its header again
|
|
# would bring back a session that names more than one account.
|
|
app_dir = Path(__file__).parent.parent / "app"
|
|
hits = [
|
|
f"{path.relative_to(app_dir)}: {word}"
|
|
for path in app_dir.rglob("*.py")
|
|
for word in ("account_ids", '"epochs"', "x-siftlode-account", 'query_params.get("account")')
|
|
if word in path.read_text(encoding="utf-8").lower()
|
|
]
|
|
assert hits == []
|
|
|
|
|
|
def test_logout_others_revokes_the_other_session_but_keeps_the_caller(client, make_user):
|
|
# Two independent cookie jars (browsers) signed into the same account. `logout-others` from one
|
|
# bumps the account epoch so the OTHER's cookie is rejected on its next request, while the caller
|
|
# is re-stamped and stays in. Checked on /api/feed (a protected endpoint that 401s on revocation).
|
|
make_user(email="carol@example.com", password=_PW)
|
|
# A second, independent cookie jar; the `client` fixture already routed get_db to the test DB on
|
|
# the shared app, so this instance uses it too — it just has its own session cookie.
|
|
other = TestClient(app)
|
|
assert _login(client, "carol@example.com", _PW).status_code == 200
|
|
assert _login(other, "carol@example.com", _PW).status_code == 200
|
|
assert other.get("/api/feed").status_code == 200 # both valid to start
|
|
|
|
assert client.post("/auth/logout-others").status_code == 200
|
|
assert other.get("/api/feed").status_code == 401 # its epoch is now stale
|
|
assert client.get("/api/feed").status_code == 200 # the caller stays signed in
|
|
|
|
|
|
def test_reset_confirm_is_single_use_and_kills_existing_sessions(client, make_user, db):
|
|
from app.routes.auth import _issue_token
|
|
|
|
user = make_user(email="dave@example.com", password=_PW)
|
|
assert _login(client, "dave@example.com", _PW).status_code == 200
|
|
assert client.get("/api/feed").status_code == 200 # session valid at the pre-reset epoch
|
|
|
|
raw = _issue_token(db, user, "reset", timedelta(hours=1))
|
|
db.commit()
|
|
ok = client.post(
|
|
"/auth/password-reset/confirm", json={"token": raw, "password": "brand new pw!"}
|
|
)
|
|
assert ok.status_code == 200
|
|
|
|
# A reset bumps the epoch (compromise response) -> the pre-reset cookie is now rejected.
|
|
assert client.get("/api/feed").status_code == 401
|
|
# The token is single-use: a replay is a clean 400, not a second reset.
|
|
replay = client.post(
|
|
"/auth/password-reset/confirm", json={"token": raw, "password": "another pw!!"}
|
|
)
|
|
assert replay.status_code == 400
|
|
|
|
|
|
def test_approving_an_unverified_account_keeps_its_password(db, make_user):
|
|
# This inverts an earlier rule. Approval used to drop a password set while the email was
|
|
# unverified (SEC-10, account pre-emption). It was dropped because it destroyed the password a
|
|
# legitimate registrant had just chosen, in what is the NORMAL ordering — the admin decides
|
|
# before the user reads their mail — making "your password was cleared for security" the default
|
|
# experience.
|
|
#
|
|
# It was ALSO argued that the revoke was redundant, because password_login refuses an
|
|
# unverified account before it reads is_active. The epic's closing review refuted that: the gate
|
|
# holds only while the address is unverified, and the approval mail exists to get the real
|
|
# owner to end that state. Do not restore that reasoning from this comment — the residual is
|
|
# SEC-16, and the same review tried and reverted a guard for it (see test_review_round.py).
|
|
from app.routes.admin import _activate_user_by_email
|
|
|
|
user = make_user(
|
|
email="approved-first@example.com",
|
|
password="chosen at registration",
|
|
email_verified=False,
|
|
is_active=False,
|
|
)
|
|
epoch0 = user.session_epoch or 0
|
|
|
|
_activate_user_by_email(db, "approved-first@example.com")
|
|
|
|
assert user.is_active is True
|
|
assert user.password_hash is not None
|
|
assert user.password_revoked_at is None
|
|
assert (user.session_epoch or 0) == epoch0 # nothing to revoke, so no epoch bump either
|
|
|
|
|
|
def test_an_approved_but_unverified_account_still_cannot_sign_in(client, make_user):
|
|
# The gate that stops a pre-emptor BEFORE the address is confirmed, whatever the admin decided.
|
|
# It is not the whole story — confirming the address opens it, which is SEC-16 — but it is the
|
|
# invariant this test owns, and a change that let an unverified account sign in would be a
|
|
# straightforwardly worse bug than the residual.
|
|
make_user(
|
|
email="notyet@example.com",
|
|
password="attacker-chosen",
|
|
email_verified=False,
|
|
is_active=True,
|
|
)
|
|
|
|
r = _login(client, "notyet@example.com", "attacker-chosen")
|
|
|
|
assert r.status_code == 403
|
|
assert r.json()["detail"]["reason"] == "email_unverified"
|
|
|
|
|
|
def test_verifying_afterwards_lets_the_original_password_in(client, db, make_user):
|
|
# The payoff for the legitimate registrant: the password chosen at registration still works once
|
|
# the mailbox is proven — no "your password was cleared for security" detour, no reset round-trip.
|
|
from app.routes.auth import _issue_token
|
|
|
|
user = make_user(
|
|
email="ordering@example.com",
|
|
password="chosen at registration",
|
|
email_verified=False,
|
|
is_active=False,
|
|
)
|
|
from app.routes.admin import _activate_user_by_email
|
|
|
|
_activate_user_by_email(db, "ordering@example.com") # admin decides FIRST — the common ordering
|
|
raw = _issue_token(db, user, "verify", timedelta(hours=1))
|
|
db.commit()
|
|
|
|
verified = client.post("/auth/verify", json={"token": raw})
|
|
|
|
assert verified.status_code == 200
|
|
assert verified.json() == {
|
|
"ok": True,
|
|
"password_reset_required": False,
|
|
"email": "ordering@example.com",
|
|
}
|
|
assert _login(client, "ordering@example.com", "chosen at registration").status_code == 200
|
|
|
|
|
|
def test_activating_a_verified_account_keeps_its_password(db, make_user):
|
|
# The no-SMTP shape: registration marks such an account verified up front, so it was never in
|
|
# scope for the SEC-10 guard even back then. Kept as the other half of the pair.
|
|
from app.routes.admin import _activate_user_by_email
|
|
|
|
user = make_user(
|
|
email="legit@example.com", password="my own password", email_verified=True, is_active=False
|
|
)
|
|
_activate_user_by_email(db, "legit@example.com")
|
|
assert user.is_active is True
|
|
assert user.password_hash is not None
|
|
|
|
|
|
def _register_and_fetch(db, email: str):
|
|
"""Run the /register background worker, then read the users/invites rows back through a FRESH
|
|
query. The worker owns its own session, so anything it failed to commit is simply not there."""
|
|
from sqlalchemy import select as _select
|
|
|
|
from app.models import Invite, User
|
|
from app.routes.auth import _register_account
|
|
|
|
_register_account(email, _PW)
|
|
db.expire_all()
|
|
user = db.execute(_select(User).where(User.email == email)).scalar_one_or_none()
|
|
invite = db.execute(_select(Invite).where(Invite.email == email)).scalar_one_or_none()
|
|
return user, invite
|
|
|
|
|
|
def _no_smtp(monkeypatch):
|
|
"""Put the worker on the no-SMTP path and capture the admin notice instead of sending it."""
|
|
from app.routes import auth as auth_mod
|
|
|
|
notified: list[str] = []
|
|
monkeypatch.setattr(auth_mod.email_mod, "email_enabled", lambda: False)
|
|
monkeypatch.setattr(
|
|
auth_mod.email_mod,
|
|
"send_admin_new_request",
|
|
lambda admins, lang, to: notified.append(to) or True,
|
|
)
|
|
return notified
|
|
|
|
|
|
def test_registration_of_a_pre_approved_email_persists_the_account_without_smtp(
|
|
db, make_user, monkeypatch
|
|
):
|
|
# THE regression. An admin whitelists an address, the person registers, and the account silently
|
|
# never existed: upsert_pending_invite leaves an already-approved row alone and returns WITHOUT
|
|
# committing, and the only other commit on this path (_issue_token) is gated on SMTP — so with no
|
|
# working mail the flushed user was rolled back at session close. The sign-up answered "ok",
|
|
# created nothing, and re-registering did the same again.
|
|
from app.models import Invite
|
|
|
|
notified = _no_smtp(monkeypatch)
|
|
make_user(email="admin@example.com", role="admin")
|
|
db.add(Invite(email="roland@example.com", status="approved", decided_by="admin@example.com"))
|
|
db.commit()
|
|
|
|
user, invite = _register_and_fetch(db, "roland@example.com")
|
|
|
|
assert user is not None, "pre-approved registration was rolled back — the account is gone"
|
|
assert user.password_hash is not None
|
|
assert user.email_verified is True # no SMTP => ownership can't be proven by link
|
|
assert (
|
|
user.is_active is False
|
|
) # approval is bound to the account by an admin, not by signing up
|
|
# The admin's standing decision is left exactly as it was — is_allowed() still reads "approved",
|
|
# so their Google sign-in keeps working and an anonymous sign-up cannot revoke access.
|
|
assert invite.status == "approved"
|
|
assert invite.decided_by == "admin@example.com"
|
|
# The notify path still fires for an already-approved address — nothing lands in the queue, so
|
|
# it is the only announcement there is. (Whether it is DELIVERED is separate: on a real no-SMTP
|
|
# instance `_send` drops it, which is the open half of the problem, roadmap UX-25.)
|
|
assert notified == ["roland@example.com"]
|
|
|
|
|
|
def test_registration_with_an_already_pending_request_persists_the_account(
|
|
db, make_user, monkeypatch
|
|
):
|
|
# The same rollback, reached the other way: someone uses Request access, then registers before
|
|
# the admin has decided. upsert_pending_invite leaves the pending row alone and does not commit.
|
|
from app.models import Invite
|
|
|
|
notified = _no_smtp(monkeypatch)
|
|
make_user(email="admin@example.com", role="admin")
|
|
db.add(Invite(email="pia@example.com", status="pending"))
|
|
db.commit()
|
|
|
|
user, invite = _register_and_fetch(db, "pia@example.com")
|
|
|
|
assert user is not None, "registration was rolled back — the account is gone"
|
|
assert user.is_active is False
|
|
assert invite.status == "pending"
|
|
assert notified == ["pia@example.com"]
|
|
|
|
|
|
def test_an_admin_can_activate_a_pre_approved_registrant(db, monkeypatch):
|
|
# The recovery path for the account above, pinned so it cannot quietly break: re-entering the
|
|
# address in the Access-requests whitelist box (POST /api/admin/invites) runs
|
|
# _activate_user_by_email.
|
|
from app.models import Invite
|
|
from app.routes.admin import _activate_user_by_email
|
|
|
|
_no_smtp(monkeypatch)
|
|
db.add(Invite(email="roland@example.com", status="approved", decided_by="admin@example.com"))
|
|
db.commit()
|
|
user, _ = _register_and_fetch(db, "roland@example.com")
|
|
|
|
_activate_user_by_email(db, "roland@example.com")
|
|
|
|
db.expire_all()
|
|
assert user.is_active is True
|
|
assert user.password_hash is not None # approval never touches the password
|
|
|
|
|
|
def test_registration_of_an_unknown_email_still_persists_account_and_pending_invite(
|
|
db, make_user, monkeypatch
|
|
):
|
|
# Guard, not a regression test: this branch always worked (upsert INSERTs the invite and commits
|
|
# the same transaction). It must keep working — and it is the case that DOES notify the admin.
|
|
notified = _no_smtp(monkeypatch)
|
|
make_user(email="admin@example.com", role="admin")
|
|
|
|
user, invite = _register_and_fetch(db, "stranger@example.com")
|
|
|
|
assert user is not None
|
|
assert user.is_active is False
|
|
assert invite is not None and invite.status == "pending"
|
|
assert notified == ["stranger@example.com"]
|
|
|
|
|
|
def test_registration_of_a_previously_denied_email_revives_the_request(db, make_user, monkeypatch):
|
|
# Guard: a denied person may ask again — the invite flips back to pending (and commits) so the
|
|
# admin sees a fresh request. The stale decision must not leak into the new account's state.
|
|
from app.models import Invite
|
|
|
|
notified = _no_smtp(monkeypatch)
|
|
make_user(email="admin@example.com", role="admin")
|
|
db.add(Invite(email="denis@example.com", status="denied", decided_by="admin@example.com"))
|
|
db.commit()
|
|
|
|
user, invite = _register_and_fetch(db, "denis@example.com")
|
|
|
|
assert user is not None
|
|
assert user.is_active is False
|
|
assert invite.status == "pending"
|
|
assert invite.decided_at is None and invite.decided_by is None
|
|
assert notified == ["denis@example.com"]
|
|
|
|
|
|
def test_registration_of_a_pre_approved_email_persists_the_account_with_smtp(db, monkeypatch):
|
|
# Guard, not a regression test: with SMTP on, _issue_token's own commit already persisted the
|
|
# user, so this branch was never broken and must stay unbroken.
|
|
from app.models import Invite
|
|
from app.routes import auth as auth_mod
|
|
|
|
sent: list[str] = []
|
|
monkeypatch.setattr(auth_mod.email_mod, "email_enabled", lambda: True)
|
|
monkeypatch.setattr(
|
|
auth_mod.email_mod, "send_verify_email", lambda to, lang, link, **kw: sent.append(to)
|
|
)
|
|
monkeypatch.setattr(auth_mod.email_mod, "send_admin_new_request", lambda admins, lang, to: True)
|
|
db.add(Invite(email="rita@example.com", status="approved", decided_by="admin@example.com"))
|
|
db.commit()
|
|
|
|
user, invite = _register_and_fetch(db, "rita@example.com")
|
|
|
|
assert user is not None
|
|
assert user.email_verified is False # must confirm the address
|
|
assert user.is_active is False
|
|
assert invite.status == "approved"
|
|
assert sent == ["rita@example.com"]
|
|
|
|
|
|
def test_verifying_an_email_never_grants_access_on_its_own(client, db, make_user):
|
|
# Verification attests the address; it does not decide access. The two gates stay separate.
|
|
from app.models import Invite
|
|
from app.routes.auth import _issue_token
|
|
|
|
user = make_user(email="rita@example.com", password=_PW, email_verified=False, is_active=False)
|
|
db.add(Invite(email="rita@example.com", status="approved", decided_by="admin@example.com"))
|
|
db.commit()
|
|
raw = _issue_token(db, user, "verify", timedelta(hours=1))
|
|
|
|
assert client.post("/auth/verify", json={"token": raw}).status_code == 200
|
|
|
|
db.expire_all()
|
|
assert user.email_verified is True
|
|
assert user.is_active is False
|
|
assert _login(client, "rita@example.com", _PW).status_code == 403
|
|
|
|
|
|
def test_register_route_creates_the_account_end_to_end(client, db, monkeypatch):
|
|
# The tests above call _register_account directly, so nothing covered the ROUTE: its
|
|
# allow_registration gate, its rate limiter and the background.add_task wiring that makes the
|
|
# whole flow reachable at all. TestClient runs BackgroundTasks after the response, so a single
|
|
# POST exercises the real path end to end.
|
|
from sqlalchemy import select as _select
|
|
|
|
from app.models import User
|
|
|
|
_no_smtp(monkeypatch)
|
|
|
|
r = client.post("/auth/register", json={"email": "newbie@example.com", "password": _PW})
|
|
|
|
assert r.status_code == 200 and r.json() == {"ok": True, "status": "ok"}
|
|
db.expire_all()
|
|
user = db.execute(_select(User).where(User.email == "newbie@example.com")).scalar_one_or_none()
|
|
assert user is not None, "POST /auth/register answered ok but created nothing"
|
|
assert user.password_hash is not None
|
|
|
|
|
|
def test_register_route_is_refused_when_registration_is_closed(client, db, monkeypatch):
|
|
# The gate is an explicit 403 with nothing created — not a uniform "ok" that silently drops it.
|
|
from sqlalchemy import select as _select
|
|
|
|
from app import sysconfig
|
|
from app.models import User
|
|
|
|
_no_smtp(monkeypatch)
|
|
sysconfig.set_value(db, "allow_registration", False)
|
|
|
|
r = client.post("/auth/register", json={"email": "closed@example.com", "password": _PW})
|
|
|
|
assert r.status_code == 403
|
|
db.expire_all()
|
|
assert (
|
|
db.execute(_select(User).where(User.email == "closed@example.com")).scalar_one_or_none()
|
|
is None
|
|
)
|
|
|
|
|
|
# --- SEC-10 lockout: a revoked preset password must stay RECOVERABLE ---------------------------
|
|
#
|
|
# Regression pack for the prod incident of 2026-09-02. An account registered with a password, then
|
|
# stripped of it by the SEC-10 guard, ended up with password_hash = NULL. That is also what a
|
|
# Google-only account looks like, and reset eligibility keyed off `password_hash` alone — so the
|
|
# account could neither sign in (no password) nor obtain a reset link, and every response on the way
|
|
# was a deliberately uniform "ok"/401. `password_revoked_at` is the trace that tells the two apart;
|
|
# these pin each step of the way back in.
|
|
#
|
|
# The admin-approval caller is gone, so Google adoption is the only path that still produces
|
|
# this state — plus every account stripped on prod before that change, which is why the whole recovery
|
|
# route below must keep working. These tests manufacture the state through the surviving producer.
|
|
|
|
|
|
def _revoke_as_google_adoption(db, user) -> None:
|
|
"""Put an account into the SEC-10-revoked state the way the ONLY remaining producer does: the
|
|
Google callback adopting an account whose email was never verified."""
|
|
from app.routes.auth import revoke_untrusted_preset_password
|
|
|
|
revoke_untrusted_preset_password(db, user)
|
|
db.commit()
|
|
|
|
|
|
def _capture_reset_links(monkeypatch) -> list[str]:
|
|
"""Patch the reset mailer and collect the recipients it is asked to write to."""
|
|
from app.routes import auth as auth_mod
|
|
|
|
sent: list[str] = []
|
|
monkeypatch.setattr(
|
|
auth_mod.email_mod, "send_password_reset", lambda to, lang, link: sent.append(to)
|
|
)
|
|
return sent
|
|
|
|
|
|
def test_the_google_adoption_revoke_stamps_password_revoked_at(db, make_user):
|
|
# The guard still fires where it earns its keep: the real owner has just proven ownership through
|
|
# Google, so a password set on the address while it was unverified must not survive.
|
|
victim = make_user(
|
|
email="stamped@example.com",
|
|
password="attacker-chosen",
|
|
email_verified=False,
|
|
is_active=False,
|
|
)
|
|
assert victim.password_revoked_at is None
|
|
epoch0 = victim.session_epoch or 0
|
|
|
|
_revoke_as_google_adoption(db, victim)
|
|
|
|
assert victim.password_hash is None
|
|
assert victim.password_revoked_at is not None # …so the reset flow can tell it HAD a password
|
|
assert (victim.session_epoch or 0) == epoch0 + 1 # any session that password minted is killed
|
|
|
|
|
|
def test_an_account_whose_password_was_revoked_can_still_get_a_reset_link(
|
|
db, make_user, monkeypatch
|
|
):
|
|
# THE lockout bug: without the stamp this sent nothing, and the account was unrecoverable.
|
|
from app.routes.auth import _send_reset_if_eligible
|
|
|
|
victim = make_user(
|
|
email="lockedout@example.com",
|
|
password="chosen at registration",
|
|
email_verified=False,
|
|
is_active=False,
|
|
)
|
|
_revoke_as_google_adoption(db, victim)
|
|
sent = _capture_reset_links(monkeypatch)
|
|
|
|
_send_reset_if_eligible("lockedout@example.com")
|
|
|
|
assert sent == ["lockedout@example.com"]
|
|
|
|
|
|
def test_a_google_only_account_is_still_not_reset_eligible(db, make_user, monkeypatch):
|
|
# The other side of the rule: an account that NEVER had a password gets no reset link — its owner
|
|
# adds one from Settings while signed in (/auth/set-password), which proves identity by session.
|
|
from app.routes.auth import _send_reset_if_eligible
|
|
|
|
make_user(email="googler@example.com", password=None, google_sub="sub-123")
|
|
sent = _capture_reset_links(monkeypatch)
|
|
|
|
_send_reset_if_eligible("googler@example.com")
|
|
|
|
assert sent == []
|
|
|
|
|
|
def test_completing_a_reset_clears_the_revocation_stamp(client, db, make_user):
|
|
from app.routes.auth import _issue_token
|
|
|
|
user = make_user(
|
|
email="recovered@example.com",
|
|
password="chosen at registration",
|
|
email_verified=False,
|
|
is_active=True, # already approved; reset-confirm verifies the email but never activates
|
|
)
|
|
_revoke_as_google_adoption(db, user)
|
|
raw = _issue_token(db, user, "reset", timedelta(hours=1))
|
|
db.commit()
|
|
|
|
ok = client.post(
|
|
"/auth/password-reset/confirm", json={"token": raw, "password": "a brand new one"}
|
|
)
|
|
|
|
assert ok.status_code == 200
|
|
db.expire_all()
|
|
assert user.password_hash is not None
|
|
assert user.password_revoked_at is None # spent — not a permanently reset-eligible account
|
|
assert _login(client, "recovered@example.com", "a brand new one").status_code == 200
|
|
|
|
|
|
def test_verify_tells_a_revoked_account_that_it_needs_a_new_password(client, db, make_user):
|
|
# The banner's input. Saying so leaks nothing: the caller just proved mailbox control by holding a
|
|
# valid single-use verify token. Without it the user meets only the uniform "Invalid email or
|
|
# password" and has no way to learn why the password they chose stopped working.
|
|
from app.routes.auth import _issue_token
|
|
|
|
user = make_user(
|
|
email="needspw@example.com",
|
|
password="chosen at registration",
|
|
email_verified=False,
|
|
is_active=False,
|
|
)
|
|
_revoke_as_google_adoption(db, user)
|
|
raw = _issue_token(db, user, "verify", timedelta(hours=1))
|
|
db.commit()
|
|
|
|
r = client.post("/auth/verify", json={"token": raw})
|
|
|
|
assert r.status_code == 200
|
|
assert r.json() == {
|
|
"ok": True,
|
|
"password_reset_required": True,
|
|
"email": "needspw@example.com",
|
|
}
|
|
|
|
|
|
def test_verify_reports_no_reset_needed_for_an_intact_account(client, db, make_user):
|
|
from app.routes.auth import _issue_token
|
|
|
|
user = make_user(email="intact@example.com", email_verified=False, is_active=False)
|
|
raw = _issue_token(db, user, "verify", timedelta(hours=1))
|
|
db.commit()
|
|
|
|
r = client.post("/auth/verify", json={"token": raw})
|
|
|
|
assert r.status_code == 200
|
|
assert r.json() == {
|
|
"ok": True,
|
|
"password_reset_required": False,
|
|
"email": "intact@example.com",
|
|
}
|