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.
153 lines
5.6 KiB
Python
153 lines
5.6 KiB
Python
"""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
|