peter
96ad1ef306
The registration language is read today (preferences.language, the mail language); the Modal and the players order against popovers through layerStack, not the FloatingTree; the upload's auth dependency is current_user; the module of access tests covers onboarding and access, not only the canonical email. The dev proxy's WebSocket upgrade served the removed messages socket and goes; the test compose header says a pure-logic run needs SIFTLODE_TEST_REQUIRE_DB=0; the speed measure no longer reports an unreachable database as "no share link".
374 lines
16 KiB
Python
374 lines
16 KiB
Python
"""The onboarding and access invariants a review of that work found unpinned.
|
|
|
|
Every test here corresponds to a defect a reviewer could still introduce with the whole suite green:
|
|
a mail whose body never renders in some language, a link lifetime described in the wrong unit, a
|
|
registration that rewrites an existing password or kills a link already in flight, an activation
|
|
that only works for the exact spelling, a "we already wrote to them" stamp for a mail that was
|
|
dropped, an env allow-list that answers a different question than the gate it backs, a reset or
|
|
resend that reports success it did not have, and a throttle that leaks whether an account exists.
|
|
Approval's password revoke was called redundant because `password_login` refuses an unverified
|
|
account; that holds only until the address is verified, which is what the tests on pre-emption pin.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app import auth as auth_core
|
|
from app import email as email_mod
|
|
from app import mailtext
|
|
from app.config import settings
|
|
from app.models import AuthToken, Invite, User
|
|
from app.routes import auth as auth_routes
|
|
|
|
_PW = "correct horse battery"
|
|
|
|
|
|
@pytest.fixture
|
|
def wire(monkeypatch):
|
|
"""Capture what `_send` would have put on the wire. Mails are the one surface where a body that
|
|
never renders in a test is a body that first renders in someone's inbox."""
|
|
out: list[tuple[list[str], str, str]] = []
|
|
monkeypatch.setattr(
|
|
email_mod,
|
|
"_send",
|
|
lambda to, subject, body, reply_to=None: out.append((to, subject, body)) is None,
|
|
)
|
|
return out
|
|
|
|
|
|
@pytest.fixture
|
|
def smtp_on(monkeypatch):
|
|
monkeypatch.setattr(email_mod, "email_enabled", lambda db=None: True)
|
|
|
|
|
|
# --- every mail actually renders ------------------------------------------------------------------
|
|
|
|
# One call per `send_*`, with arguments of the right shape. The parity test compares HU against EN
|
|
# and can therefore never catch a template whose placeholders disagree with its CALLER — that
|
|
# mismatch is a KeyError inside a background task, i.e. a mail that silently never goes out. Three
|
|
# mailers this epic rewrote (`password_reset`, `role_changed`, `account_unsuspended`) had no test
|
|
# that executed their `.format()` at all.
|
|
_EVERY_MAIL = [
|
|
("send_access_approved", ("to@example.com", "LANG", "https://x/#verify=t", True, 168)),
|
|
("send_access_approved", ("to@example.com", "LANG", None, True, 24)),
|
|
("send_access_approved", ("to@example.com", "LANG", None, False, 24)),
|
|
("send_access_invitation", ("to@example.com", "LANG", "https://x/#verify=t", True, 168)),
|
|
("send_access_invitation", ("to@example.com", "LANG", None, False, 24)),
|
|
("send_access_denied", ("to@example.com", "LANG")),
|
|
("send_role_changed", ("to@example.com", "LANG", "admin", "op@example.com")),
|
|
("send_role_changed", ("to@example.com", "LANG", "user", None)),
|
|
("send_account_unsuspended", ("to@example.com", "LANG", "op@example.com")),
|
|
("send_account_deleted", ("to@example.com", "LANG", None)),
|
|
("send_account_suspended", ("to@example.com", "LANG", "op@example.com")),
|
|
("send_account_suspended", ("to@example.com", "LANG", None)),
|
|
("send_admin_new_request", (["a@example.com"], "LANG", "who@example.com")),
|
|
("send_verify_email", ("to@example.com", "LANG", "https://x/#verify=t", 24)),
|
|
("send_password_reset", ("to@example.com", "LANG", "https://x/#reset=t")),
|
|
("send_test", ("to@example.com", "LANG")),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("lang", mailtext.SUPPORTED)
|
|
@pytest.mark.parametrize("name,args", _EVERY_MAIL, ids=[f"{n}{a[2:3]}" for n, a in _EVERY_MAIL])
|
|
def test_every_mail_renders_in_every_language(wire, name, args, lang):
|
|
getattr(email_mod, name)(*[lang if a == "LANG" else a for a in args])
|
|
|
|
(_, subject, body) = wire[0]
|
|
assert subject and body
|
|
# A template that lost its interpolation renders the brace text verbatim.
|
|
assert "{" not in subject and "{" not in body
|
|
|
|
|
|
def test_a_weeks_link_is_described_in_days_not_hours(wire):
|
|
# The approval token lives seven days, and the mail used to say "expires in 168 hours".
|
|
email_mod.send_access_approved("to@example.com", "en", "https://x/#verify=t", True, 168)
|
|
assert "7 days" in wire[0][2] and "168" not in wire[0][2]
|
|
|
|
email_mod.send_verify_email("to@example.com", "hu", "https://x/#verify=t", 24)
|
|
assert "24 óra" in wire[1][2]
|
|
|
|
|
|
# --- what a second registration may and may not do -------------------------------------------------
|
|
|
|
|
|
def test_a_later_registration_never_touches_an_existing_password(db, client, smtp_on, wire):
|
|
"""The rule a review put back, having tried the opposite and found it worse.
|
|
|
|
"While the address is unverified, the newest registration owns it" looked like a fix for address
|
|
pre-emption. It is symmetric — an attacker can always be last — so it closed nothing, and it
|
|
handed an unauthenticated caller the power to overwrite a real registrant's chosen password and
|
|
delete the confirmation link already sitting in their inbox. On a SEC-10-stripped row it would
|
|
have installed a password where the whole point is that there is none. Reverted; the owner's
|
|
recourse is password-reset, which proves the mailbox. Residual logged as SEC-16.
|
|
"""
|
|
assert (
|
|
client.post(
|
|
"/auth/register", json={"email": "victim@example.com", "password": "the first password"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
db.expire_all()
|
|
user = db.execute(select(User).where(User.email == "victim@example.com")).scalar_one()
|
|
before = user.password_hash
|
|
|
|
assert (
|
|
client.post(
|
|
"/auth/register",
|
|
json={"email": "victim@example.com", "password": "the second password"},
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
db.expire_all()
|
|
assert db.get(User, user.id).password_hash == before
|
|
|
|
|
|
def test_a_later_registration_leaves_a_link_already_in_flight_alive(db, client, smtp_on, wire):
|
|
assert (
|
|
client.post(
|
|
"/auth/register", json={"email": "victim@example.com", "password": "the first password"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
db.expire_all()
|
|
user = db.execute(select(User).where(User.email == "victim@example.com")).scalar_one()
|
|
first = db.execute(select(AuthToken.id).where(AuthToken.user_id == user.id)).scalars().all()
|
|
# Not just a subset check below: `set() <= set()` is vacuously true, so without this the test
|
|
# would go green while asserting nothing the day registration stops issuing a token.
|
|
assert len(first) == 1
|
|
|
|
assert (
|
|
client.post(
|
|
"/auth/register",
|
|
json={"email": "victim@example.com", "password": "the second password"},
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
db.expire_all()
|
|
live = db.execute(select(AuthToken.id).where(AuthToken.user_id == user.id)).scalars().all()
|
|
# The re-send adds a link; it must not invalidate the one the owner may already be holding.
|
|
assert set(first) <= set(live) and len(live) == len(first) + 1
|
|
|
|
|
|
def test_a_stripped_account_keeps_its_recovery_stamp(db, client, smtp_on, make_user):
|
|
"""Prod carries accounts SEC-10 left with no password and a `password_revoked_at` stamp. That
|
|
stamp is the only reason password-reset still accepts them; an anonymous POST must not be able to
|
|
clear it, nor to put a password on a row whose whole state is "there is none"."""
|
|
stripped = make_user(
|
|
email="stripped@example.com",
|
|
password=None,
|
|
email_verified=False,
|
|
is_active=True,
|
|
password_revoked_at=datetime.now(timezone.utc),
|
|
)
|
|
|
|
assert (
|
|
client.post(
|
|
"/auth/register",
|
|
json={"email": "stripped@example.com", "password": "an attacker password"},
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
db.expire_all()
|
|
row = db.get(User, stripped.id)
|
|
assert row.password_hash is None and row.password_revoked_at is not None
|
|
|
|
|
|
def test_a_denied_sibling_is_not_reopened_while_a_pending_row_exists(db):
|
|
# Canonical siblings exist on prod. Picking the denied one reopens it and re-notifies every
|
|
# admin, putting one person in the queue twice — and the exact-spelling tie-break made that
|
|
# deterministic rather than merely possible.
|
|
db.add_all(
|
|
[
|
|
Invite(email="pair.one@gmail.com", status="pending"),
|
|
Invite(email="pairone@gmail.com", status="denied"),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
auth_routes.upsert_pending_invite(db, "pairone@gmail.com")
|
|
|
|
db.expire_all()
|
|
rows = {i.email: i.status for i in db.execute(select(Invite)).scalars().all()}
|
|
assert rows == {"pair.one@gmail.com": "pending", "pairone@gmail.com": "denied"}
|
|
|
|
|
|
# --- the env bootstrap answers the same question as the gate --------------------------------------
|
|
|
|
|
|
def test_the_env_allow_list_is_matched_canonically(db, monkeypatch):
|
|
# An operator writing the dotted spelling into ALLOWED_EMAILS was refused when Google answered
|
|
# with the undotted primary — the 2026-09-03 lockout, on the bootstrap path.
|
|
monkeypatch.setattr(settings, "allowed_emails", "first.last@gmail.com")
|
|
|
|
assert auth_core.is_allowed(db, "firstlast@gmail.com") is True
|
|
assert auth_core.is_allowed(db, "first.last+tag@gmail.com") is True
|
|
assert auth_core.is_allowed(db, "someone.else@gmail.com") is False
|
|
|
|
|
|
def test_an_env_granted_address_cannot_be_shadowed_by_a_stranger(db, client, monkeypatch):
|
|
# `is_allowed` short-circuits on ANY invite row for the canonical form, so an unauthenticated
|
|
# caller registering under a sibling spelling would open a pending row and permanently shadow an
|
|
# env grant. The env list is matched canonically, so the registration never opens a row.
|
|
monkeypatch.setattr(settings, "allowed_emails", "bob@gmail.com")
|
|
|
|
r = client.post("/auth/register", json={"email": "b.ob@gmail.com", "password": _PW})
|
|
assert r.status_code == 200
|
|
|
|
assert db.execute(select(Invite)).scalars().all() == []
|
|
assert auth_core.is_allowed(db, "bob@gmail.com") is True
|
|
|
|
|
|
def test_a_dotted_admin_emails_entry_still_bootstraps_its_admin(monkeypatch):
|
|
monkeypatch.setattr(settings, "admin_emails", "the.operator+siftlode@gmail.com")
|
|
|
|
assert "theoperator@gmail.com" in auth_core.env_admin_canonicals()
|
|
|
|
|
|
# --- approving decides about a MAILBOX, not a spelling --------------------------------------------
|
|
|
|
|
|
def test_approving_one_spelling_activates_the_account_registered_under_another(
|
|
db, make_user, as_user, smtp_on, wire
|
|
):
|
|
# The 2026-09-03 incident in one test: the admin approves the address as typed into the request,
|
|
# and the account exists under the other Gmail spelling of the same mailbox.
|
|
admin = make_user(email="admin@example.com", role="admin")
|
|
user = make_user(email="dotless@gmail.com", password=_PW, is_active=False)
|
|
user.email = "dotless@gmail.com"
|
|
inv = Invite(email="dot.less@gmail.com", status="pending")
|
|
db.add(inv)
|
|
db.commit()
|
|
|
|
r = as_user(admin).post(f"/api/admin/invites/{inv.id}/approve")
|
|
|
|
assert r.status_code == 200
|
|
db.expire_all()
|
|
assert db.get(User, user.id).is_active is True
|
|
|
|
|
|
def test_deleting_one_of_two_accounts_leaves_the_other_one_allowed(db, make_user, as_user):
|
|
# Prod carries two accounts for one Gmail mailbox. A canonical invite sweep on delete took the
|
|
# survivor's grant with it, refusing their next sign-in — this epic's founding bug, self-inflicted.
|
|
admin = make_user(email="admin@example.com", role="admin")
|
|
doomed = make_user(email="pair.one@gmail.com", password=_PW)
|
|
make_user(email="pairone@gmail.com", password=_PW)
|
|
db.add_all([Invite(email="pair.one@gmail.com", status="approved")])
|
|
db.commit()
|
|
|
|
r = as_user(admin).delete(f"/api/admin/users/{doomed.id}")
|
|
|
|
assert r.status_code == 200
|
|
assert auth_core.is_allowed(db, "pairone@gmail.com") is True
|
|
|
|
|
|
# --- a button an admin pressed must not report a send that did not happen -------------------------
|
|
|
|
|
|
def test_a_failed_invitation_changes_nothing_the_admin_was_told_about(
|
|
db, make_user, as_user, smtp_on, monkeypatch
|
|
):
|
|
monkeypatch.setattr(email_mod, "send_access_invitation", lambda *a, **kw: False)
|
|
admin = make_user(email="admin@example.com", role="admin")
|
|
stranger = make_user(
|
|
email="stranger@example.com", password=_PW, email_verified=False, is_active=False
|
|
)
|
|
inv = Invite(email="stranger@example.com", status="approved")
|
|
db.add(inv)
|
|
db.commit()
|
|
|
|
r = as_user(admin).post(f"/api/admin/invites/{inv.id}/invite", json={"lang": "en"})
|
|
|
|
assert r.status_code == 422 and r.json()["detail"]["reason"] == "send_failed"
|
|
db.expire_all()
|
|
# The column exists to answer "did I already write to them?", and the UI dims the button and
|
|
# warns about a second mail on the strength of it. A dropped send must not make it say yes.
|
|
assert db.get(Invite, inv.id).invited_at is None
|
|
# And a request the UI reports as a failure must not have let somebody in behind the admin's
|
|
# back — the activation belongs after a mail that actually left.
|
|
assert db.get(User, stranger.id).is_active is False
|
|
|
|
|
|
def test_a_failed_resend_says_so_instead_of_reporting_ok(
|
|
db, make_user, as_user, smtp_on, monkeypatch
|
|
):
|
|
monkeypatch.setattr(email_mod, "send_verify_email", lambda *a, **kw: False)
|
|
admin = make_user(email="admin@example.com", role="admin")
|
|
stuck = make_user(email="stuck@example.com", password=_PW, email_verified=False)
|
|
|
|
r = as_user(admin).post(f"/api/admin/users/{stuck.id}/resend-verification")
|
|
|
|
assert r.status_code == 422 and r.json()["detail"]["reason"] == "send_failed"
|
|
|
|
|
|
# --- refusals and links that are already dead ------------------------------------------------------
|
|
|
|
|
|
def test_an_expired_reset_link_is_reported_dead_without_being_spent(db, client, make_user):
|
|
# The endpoint exists because a spent or stale link used to draw the full "set a new password"
|
|
# form. Expiry is the reason its docstring gives, and it was the one case no test covered.
|
|
user = make_user(email="late@example.com", password=_PW)
|
|
raw = auth_routes._issue_token(db, user, "reset", timedelta(hours=1))
|
|
token = db.execute(select(AuthToken).where(AuthToken.user_id == user.id)).scalar_one()
|
|
token.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1)
|
|
db.commit()
|
|
|
|
assert client.post("/auth/password-reset/check", json={"token": raw}).json() == {"ok": False}
|
|
r = client.post(
|
|
"/auth/password-reset/confirm", json={"token": raw, "password": "a new one 123"}
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_a_verify_token_cannot_pass_as_a_reset_link(db, client, make_user):
|
|
user = make_user(email="mixed@example.com", password=_PW)
|
|
raw = auth_routes._issue_token(db, user, "verify", timedelta(hours=1))
|
|
|
|
assert client.post("/auth/password-reset/check", json={"token": raw}).json() == {"ok": False}
|
|
|
|
|
|
def test_a_throttled_login_says_nothing_about_the_account(db, client, make_user, monkeypatch):
|
|
# The two refusals that answer BEFORE the password is proven must stay byte-identical for a real
|
|
# account and an unknown one; only `invalid` was pinned.
|
|
make_user(email="real@example.com", password=_PW)
|
|
monkeypatch.setattr(auth_routes._login_limiter, "allow", lambda key: False)
|
|
|
|
real = client.post("/auth/password-login", json={"email": "real@example.com", "password": _PW})
|
|
unknown = client.post(
|
|
"/auth/password-login", json={"email": "nobody@example.com", "password": _PW}
|
|
)
|
|
|
|
assert real.status_code == unknown.status_code == 429
|
|
assert real.json() == unknown.json() == {"detail": {"code": "login", "reason": "rate_limited"}}
|
|
|
|
|
|
# --- admins hear about a request in their own language --------------------------------------------
|
|
|
|
|
|
def test_two_admins_two_languages_two_mails(db, client, make_user, smtp_on, wire):
|
|
make_user(email="hu.admin@example.com", role="admin", preferences={"language": "hu"})
|
|
make_user(email="en.admin@example.com", role="admin", preferences={"language": "en"})
|
|
|
|
r = client.post("/auth/register", json={"email": "asking@example.com", "password": _PW})
|
|
assert r.status_code == 200
|
|
|
|
notices = [w for w in wire if "asking@example.com" in w[1]]
|
|
assert len(notices) == 2
|
|
by_recipient = {w[0][0]: w[1] for w in notices}
|
|
assert by_recipient["hu.admin@example.com"] == mailtext.MAIL_TEXT["admin_new_request.subject"][
|
|
"hu"
|
|
].format(requester="asking@example.com")
|
|
assert by_recipient["en.admin@example.com"] == mailtext.MAIL_TEXT["admin_new_request.subject"][
|
|
"en"
|
|
].format(requester="asking@example.com")
|
|
|
|
|
|
def test_an_env_admin_without_an_account_is_written_to_in_the_default(db, monkeypatch):
|
|
monkeypatch.setattr(settings, "admin_emails", "bootstrap@example.com")
|
|
|
|
assert auth_routes.admin_notify_groups(db) == {mailtext.DEFAULT: ["bootstrap@example.com"]}
|