feat: phase 3 admin magic-link auth — tokens, sessions, rate limits, audit
End-to-end passwordless admin auth. /admin/login accepts an email, POSTs
mint a 256-bit magic-link token stored only as SHA-256 in
magic_link_tokens (15-min TTL, single-use via atomic rowcount UPDATE).
Resend delivers the link; in dev with no API key, EmailService logs a
structured magic_link_dev_fallback event with the URL so the flow works
offline. /admin/auth/consume/{token} verifies, upserts a users row
(display_name from email local-part), creates a sessions row, and drops
an itsdangerous-signed cb_session cookie (HttpOnly, SameSite=Lax, Secure
in prod). /admin renders a placeholder "Welcome, <name>" page pending
Phase 4 CMS. /admin/logout flips revoked_at rather than deleting the row
to preserve the audit trail.
Rate limits use SlowAPI's in-memory limiter (5/15min/IP on login,
20/15min/IP on consume) plus a DB per-email count to catch
IP-rotating abuse. ADMIN_EMAILS enforces allowlist; non-allowlisted
submissions return the same "check your inbox" page with no token
inserted and no email sent (anti-enumeration). Every event lands in
auth_events via AuditService: link_requested, link_consumed,
consume_failed, session_created, session_revoked, rate_limited.
Add a production config validator refusing empty RESEND_API_KEY,
RESEND_FROM, or ADMIN_EMAILS; add PUBLIC_BASE_URL for email link
construction. CSRF deferred to Phase 6 per roadmap scoping; logout
handler marked # TODO(phase-6-csrf).
Mark Phase 3 complete in docs/ROADMAP.md.
This commit is contained in:
152
tests/test_session_service.py
Normal file
152
tests/test_session_service.py
Normal file
@@ -0,0 +1,152 @@
|
||||
"""Tests for :class:`app.services.sessions.SessionService`.
|
||||
|
||||
Coverage:
|
||||
- Signed-cookie round-trip (create → lookup succeeds).
|
||||
- Revoke makes subsequent lookup return ``None``.
|
||||
- Bad-signature cookie returns ``None`` without raising.
|
||||
- Expired session row returns ``None``.
|
||||
- Cookie flags reflect ``app_env`` (``Secure`` off in dev / on in prod).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from itsdangerous import URLSafeTimedSerializer
|
||||
from sqlalchemy import Engine, text
|
||||
|
||||
from app.config import Settings
|
||||
from app.services.sessions import COOKIE_NAME, SessionService
|
||||
|
||||
|
||||
def _seed_user(engine: Engine) -> int:
|
||||
"""Insert a throwaway admin user; return its id."""
|
||||
now_iso = datetime.now(timezone.utc).isoformat()
|
||||
with engine.begin() as conn:
|
||||
result = conn.execute(
|
||||
text(
|
||||
"INSERT INTO users"
|
||||
" (email, display_name, created_at, last_login_at, active)"
|
||||
" VALUES (:e, :d, :c, :c, 1)"
|
||||
),
|
||||
{"e": "session-test@example.com", "d": "Session Tester", "c": now_iso},
|
||||
)
|
||||
return int(result.lastrowid) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def _build(
|
||||
engine: Engine,
|
||||
*,
|
||||
app_env: str = "development",
|
||||
secret: str = "a-very-long-test-secret-key-1234567890",
|
||||
) -> SessionService:
|
||||
"""Return a SessionService wired for a given app_env."""
|
||||
# In production mode the Settings validator requires these fields
|
||||
# to be non-empty; populate them with test values so we can still
|
||||
# exercise the prod-only ``Secure`` flag path.
|
||||
if app_env == "production":
|
||||
settings = Settings( # type: ignore[call-arg]
|
||||
app_env=app_env,
|
||||
secret_key=secret,
|
||||
resend_api_key="test-key",
|
||||
resend_from="no-reply@example.com",
|
||||
admin_emails="admin@example.com",
|
||||
)
|
||||
else:
|
||||
settings = Settings(app_env=app_env, secret_key=secret) # type: ignore[call-arg]
|
||||
signer = URLSafeTimedSerializer(secret, salt="session")
|
||||
return SessionService(engine, signer, settings)
|
||||
|
||||
|
||||
def test_create_then_lookup_roundtrip(clean_db_engine: Engine) -> None:
|
||||
"""Creating a session and then looking up the signed cookie returns it."""
|
||||
user_id = _seed_user(clean_db_engine)
|
||||
svc = _build(clean_db_engine)
|
||||
|
||||
session, cookie = svc.create(user_id=user_id, ip="10.0.0.1", user_agent="ua")
|
||||
|
||||
# Cookie is non-empty and not the raw session ID.
|
||||
assert cookie
|
||||
# DB row stores a sha256 hex hash, never the raw.
|
||||
assert len(session.token_hash) == 64
|
||||
|
||||
found = svc.lookup(cookie)
|
||||
assert found is not None
|
||||
assert found.id == session.id
|
||||
assert found.user_id == user_id
|
||||
|
||||
|
||||
def test_lookup_empty_cookie_returns_none(clean_db_engine: Engine) -> None:
|
||||
"""No cookie value resolves to None without raising."""
|
||||
svc = _build(clean_db_engine)
|
||||
assert svc.lookup(None) is None
|
||||
assert svc.lookup("") is None
|
||||
|
||||
|
||||
def test_lookup_bad_signature_returns_none(clean_db_engine: Engine) -> None:
|
||||
"""A cookie signed by a different key resolves to None."""
|
||||
user_id = _seed_user(clean_db_engine)
|
||||
svc = _build(clean_db_engine, secret="key-one-abcdefghijklmnopqrstuvwxyz")
|
||||
_, cookie = svc.create(user_id=user_id, ip="10.0.0.1", user_agent="ua")
|
||||
|
||||
# Different signer — same salt, different key — should fail verify.
|
||||
other = _build(clean_db_engine, secret="key-two-abcdefghijklmnopqrstuvwxyz")
|
||||
assert other.lookup(cookie) is None
|
||||
|
||||
|
||||
def test_revoke_marks_session_inactive(clean_db_engine: Engine) -> None:
|
||||
"""Revoking a session prevents further lookups."""
|
||||
user_id = _seed_user(clean_db_engine)
|
||||
svc = _build(clean_db_engine)
|
||||
session, cookie = svc.create(user_id=user_id, ip="", user_agent="")
|
||||
|
||||
assert svc.lookup(cookie) is not None
|
||||
svc.revoke(session)
|
||||
assert svc.lookup(cookie) is None
|
||||
|
||||
|
||||
def test_expired_session_returns_none(clean_db_engine: Engine) -> None:
|
||||
"""A session whose DB expires_at is in the past resolves to None."""
|
||||
user_id = _seed_user(clean_db_engine)
|
||||
svc = _build(clean_db_engine)
|
||||
_, cookie = svc.create(user_id=user_id, ip="", user_agent="")
|
||||
|
||||
# Force expires_at into the past without touching the cookie.
|
||||
past = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
|
||||
with clean_db_engine.begin() as conn:
|
||||
conn.execute(text("UPDATE sessions SET expires_at = :p"), {"p": past})
|
||||
|
||||
assert svc.lookup(cookie) is None
|
||||
|
||||
|
||||
def test_cookie_params_secure_flag_gated_by_env(clean_db_engine: Engine) -> None:
|
||||
"""``Secure`` is False in development and True in production."""
|
||||
dev = _build(clean_db_engine, app_env="development").cookie_params()
|
||||
prod = _build(clean_db_engine, app_env="production").cookie_params()
|
||||
|
||||
assert dev["secure"] is False
|
||||
assert prod["secure"] is True
|
||||
for params in (dev, prod):
|
||||
assert params["key"] == COOKIE_NAME
|
||||
assert params["httponly"] is True
|
||||
assert params["samesite"] == "lax"
|
||||
assert params["path"] == "/"
|
||||
|
||||
|
||||
def test_raw_token_not_in_cookie_or_row(clean_db_engine: Engine) -> None:
|
||||
"""The raw session token must never appear in the DB row."""
|
||||
user_id = _seed_user(clean_db_engine)
|
||||
svc = _build(clean_db_engine)
|
||||
session, cookie = svc.create(user_id=user_id, ip="", user_agent="")
|
||||
|
||||
# Cookie is itsdangerous-signed, not the raw. Decoding confirms it.
|
||||
signer = URLSafeTimedSerializer(
|
||||
"a-very-long-test-secret-key-1234567890", salt="session"
|
||||
)
|
||||
raw = signer.loads(cookie)
|
||||
|
||||
# DB row stores only the sha256(raw) hex digest.
|
||||
expected_hash = hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||||
assert session.token_hash == expected_hash
|
||||
assert raw not in session.token_hash
|
||||
Reference in New Issue
Block a user