peter
24c93cbb83
Round 6's two remaining nits. The operator note said "Configuration → Email" but the group renders as "Email / SMTP" in both locales, and test_an_admin_can_activate_a_pre_approved_registrant requested make_user without using it — a leftover from the reverted rounds that made the test look like it seeded an admin when it did not.
344 lines
15 KiB
Python
344 lines
15 KiB
Python
"""Auth / session security core (TEST-1), DB-lane.
|
|
|
|
The wallet impersonation gate, epoch-based session revocation, single-use tokens and the
|
|
enumeration-safe uniform 401 are the security mag, and had zero tests (only `_oauth_redirect_uri` was
|
|
covered). 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 wallet gate 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.
|
|
"""
|
|
|
|
from datetime import timedelta
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.auth import ACTIVE_ACCOUNT_HEADER
|
|
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 test_a_non_wallet_account_header_falls_back_to_the_default(client, make_user):
|
|
# The X-Siftlode-Account header may only re-target an account already signed into THIS browser
|
|
# (in the session wallet). A header naming someone else must be ignored, not honored — otherwise
|
|
# it's an impersonation gate hole.
|
|
make_user(email="alice2@example.com", password=_PW)
|
|
mallory = make_user(email="mallory@example.com", password=_PW)
|
|
assert _login(client, "alice2@example.com", _PW).status_code == 200
|
|
me = client.get("/api/me", headers={ACTIVE_ACCOUNT_HEADER: str(mallory.id)})
|
|
assert me.status_code == 200
|
|
assert me.json()["email"] == "alice2@example.com" # still Alice, not Mallory
|
|
|
|
|
|
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_activating_an_unverified_account_drops_its_preset_password(db, make_user):
|
|
# SEC-10 (account pre-emption): an attacker registers victim@x with a password before the approved
|
|
# owner arrives. The address was never verified, so admin approval must NOT hand that pre-set
|
|
# password a working login — it is dropped and any session it minted is revoked.
|
|
from app.routes.admin import _activate_user_by_email
|
|
|
|
victim = make_user(
|
|
email="victim@example.com",
|
|
password="attacker-chosen",
|
|
email_verified=False,
|
|
is_active=False,
|
|
)
|
|
epoch0 = victim.session_epoch or 0
|
|
_activate_user_by_email(db, "victim@example.com")
|
|
assert victim.is_active is True
|
|
assert victim.password_hash is None
|
|
assert (victim.session_epoch or 0) == epoch0 + 1
|
|
|
|
|
|
def test_activating_a_verified_account_keeps_its_password(db, make_user):
|
|
# A legit account whose email IS verified (e.g. a no-SMTP registration) keeps its password on
|
|
# approval — the guard must not break the normal flow.
|
|
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, 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, which is also what keeps the SEC-10 preset-password guard in play.
|
|
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 # a verified account keeps its own 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, link: sent.append(to))
|
|
monkeypatch.setattr(auth_mod.email_mod, "send_admin_new_request", lambda admins, 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() == {"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
|
|
)
|