Merge branch 'chore/phase-3-admin-auth' into dev: Phase 3 complete
This commit is contained in:
@@ -43,6 +43,11 @@ FORWARDED_ALLOW_IPS=127.0.0.1
|
||||
SESSION_MAX_DAYS=30
|
||||
MAGIC_LINK_TTL_MIN=15
|
||||
|
||||
# --- Public URL for link construction --------------------------------------
|
||||
# Absolute base URL (scheme+host+port) used to build outbound links such as
|
||||
# the magic-link auth email. Override for production.
|
||||
PUBLIC_BASE_URL=http://127.0.0.1:8000
|
||||
|
||||
# --- Build metadata ---------------------------------------------------------
|
||||
# Injected at Docker build time. Surfaced by /healthz. Optional in dev.
|
||||
GIT_COMMIT_SHA=unknown
|
||||
|
||||
@@ -92,6 +92,20 @@ class Settings(BaseSettings):
|
||||
session_max_days: int = Field(default=30, ge=1, le=365)
|
||||
magic_link_ttl_min: int = Field(default=15, ge=1, le=60)
|
||||
|
||||
# --- Public URL for link construction ---------------------------------
|
||||
# Used to build the absolute URL emailed in magic-link auth. Defaults
|
||||
# to the local uvicorn address so dev flows work out of the box; the
|
||||
# production validator below forbids an un-set value only implicitly
|
||||
# (if the site is served off 127.0.0.1 in prod the deploy is broken
|
||||
# for reasons unrelated to this field).
|
||||
public_base_url: str = Field(
|
||||
default="http://127.0.0.1:8000",
|
||||
description=(
|
||||
"Absolute base URL (scheme+host+port) used to build links in "
|
||||
"outbound emails, e.g. the magic-link URL."
|
||||
),
|
||||
)
|
||||
|
||||
# --- Build metadata ----------------------------------------------------
|
||||
# Injected at Docker build time via an ARG/ENV. Surfaced via /healthz so
|
||||
# operators can confirm which build is live.
|
||||
@@ -136,6 +150,35 @@ class Settings(BaseSettings):
|
||||
)
|
||||
return self
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _require_auth_config_in_production(self) -> "Settings":
|
||||
"""Ensure auth-critical settings are populated in production.
|
||||
|
||||
Security control: magic-link auth depends on Resend to deliver
|
||||
one-time login tokens and on the admin allowlist to gate access.
|
||||
A production deploy that's missing any of these would either
|
||||
silently fall back to the dev log (exposing login URLs in logs)
|
||||
or accept an empty allowlist (locking the site open to nobody
|
||||
but also preventing any admin from logging in). Either outcome
|
||||
is a Phase 3 regression; fail fast instead.
|
||||
"""
|
||||
if self.app_env != "production":
|
||||
return self
|
||||
missing: list[str] = []
|
||||
if not self.resend_api_key:
|
||||
missing.append("RESEND_API_KEY")
|
||||
if not self.resend_from:
|
||||
missing.append("RESEND_FROM")
|
||||
if not self.admin_emails or not self.admin_emails_list:
|
||||
missing.append("ADMIN_EMAILS")
|
||||
if missing:
|
||||
raise ValueError(
|
||||
"Production configuration is missing required values: "
|
||||
+ ", ".join(missing)
|
||||
+ ". These are needed for magic-link admin auth."
|
||||
)
|
||||
return self
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def get_settings() -> Settings:
|
||||
|
||||
8
app/dependencies/__init__.py
Normal file
8
app/dependencies/__init__.py
Normal file
@@ -0,0 +1,8 @@
|
||||
"""FastAPI dependency helpers.
|
||||
|
||||
Route-level ``Depends(...)`` functions that don't belong to a single
|
||||
service live here. Phase 3 introduces ``app.dependencies.auth`` for
|
||||
``get_current_user`` / ``require_admin``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
89
app/dependencies/auth.py
Normal file
89
app/dependencies/auth.py
Normal file
@@ -0,0 +1,89 @@
|
||||
"""Auth dependencies for admin routes.
|
||||
|
||||
Two ``Depends(...)`` helpers:
|
||||
|
||||
- :func:`get_current_user` — returns a :class:`User` or ``None`` based
|
||||
on the signed ``cb_session`` cookie. Never raises.
|
||||
- :func:`require_admin` — same lookup but raises an HTTP 303 redirect
|
||||
to ``/admin/login`` if no user is authenticated. Used by every route
|
||||
that must be logged-in.
|
||||
|
||||
Cookie handling
|
||||
---------------
|
||||
The cookie is read via ``request.cookies`` (Starlette strips secure /
|
||||
httponly flags off by the time the app sees it; they only affect how
|
||||
the browser stores and presents the cookie). Unsigning, hashing, and
|
||||
DB lookup are delegated to :class:`app.services.sessions.SessionService`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import Depends, HTTPException, Request
|
||||
from sqlalchemy import text
|
||||
|
||||
from app.models.entities import User
|
||||
from app.models.mappers import row_to_user
|
||||
from app.services.sessions import COOKIE_NAME, SessionService
|
||||
|
||||
|
||||
def _get_session_service(request: Request) -> SessionService:
|
||||
"""Return the app-scoped :class:`SessionService` for DI.
|
||||
|
||||
Private to this module — routes and other dependencies resolve the
|
||||
service via :func:`get_current_user` / :func:`require_admin` rather
|
||||
than reaching across the dependency graph.
|
||||
"""
|
||||
return request.app.state.session_service
|
||||
|
||||
|
||||
def get_current_user(
|
||||
request: Request,
|
||||
sessions: SessionService = Depends(_get_session_service),
|
||||
) -> Optional[User]:
|
||||
"""Return the authenticated :class:`User` or ``None``.
|
||||
|
||||
Never raises. A malformed / expired / revoked cookie simply
|
||||
resolves to ``None`` so that un-authed viewers can hit admin
|
||||
login pages without tripping an exception handler.
|
||||
"""
|
||||
cookie_value = request.cookies.get(COOKIE_NAME)
|
||||
session = sessions.lookup(cookie_value)
|
||||
if session is None:
|
||||
return None
|
||||
|
||||
# We need the user row to render "Welcome, <display_name>" on the
|
||||
# admin index. Query directly here instead of adding a full
|
||||
# UserService for one call site.
|
||||
engine = request.app.state.engine
|
||||
with engine.connect() as conn:
|
||||
row = conn.execute(
|
||||
text(
|
||||
"SELECT id, email, display_name, created_at,"
|
||||
" last_login_at, active"
|
||||
" FROM users WHERE id = :id AND active = 1 LIMIT 1"
|
||||
),
|
||||
{"id": session.user_id},
|
||||
).mappings().first()
|
||||
|
||||
if row is None:
|
||||
return None
|
||||
return row_to_user(row)
|
||||
|
||||
|
||||
def require_admin(
|
||||
user: Optional[User] = Depends(get_current_user),
|
||||
) -> User:
|
||||
"""Return the authenticated user or redirect to login.
|
||||
|
||||
Uses a 303 "See Other" so the browser switches to GET on the
|
||||
followup request — correct behavior for both initial page loads
|
||||
and the post-consume redirect chain.
|
||||
"""
|
||||
if user is None:
|
||||
raise HTTPException(
|
||||
status_code=303,
|
||||
headers={"Location": "/admin/login"},
|
||||
)
|
||||
return user
|
||||
105
app/main.py
105
app/main.py
@@ -13,6 +13,17 @@ Phase 2 additions:
|
||||
- Run the idempotent seed (welcome post, About page, system user).
|
||||
- Instantiate :class:`PostService` and :class:`PageService` and
|
||||
expose them on ``app.state`` for route-level DI.
|
||||
|
||||
Phase 3 additions:
|
||||
- Build an ``itsdangerous.URLSafeTimedSerializer`` on
|
||||
``settings.secret_key`` and attach to ``app.state``.
|
||||
- Instantiate :class:`AuditService`, :class:`EmailService`,
|
||||
:class:`SessionService`, :class:`AuthService` and attach them to
|
||||
``app.state``.
|
||||
- Create a SlowAPI :class:`Limiter` and register the
|
||||
``RateLimitExceeded`` exception handler (renders
|
||||
``admin/rate_limited.html`` + HTTP 429 + audit row).
|
||||
- Include the admin router.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -20,19 +31,27 @@ from __future__ import annotations
|
||||
from pathlib import Path
|
||||
|
||||
import structlog
|
||||
from fastapi import FastAPI
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from itsdangerous import URLSafeTimedSerializer
|
||||
from slowapi.errors import RateLimitExceeded
|
||||
|
||||
from app import __version__
|
||||
from app.config import get_settings
|
||||
from app.db import build_engine, run_migrations
|
||||
from app.logging_config import configure_logging
|
||||
from app.models.seed import run_seed
|
||||
from app.routes.admin import router as admin_router
|
||||
from app.routes.health import router as health_router
|
||||
from app.routes.public import router as public_router
|
||||
from app.services.audit import AuditService
|
||||
from app.services.auth import AuthService
|
||||
from app.services.email import EmailService
|
||||
from app.services.pages import PageService
|
||||
from app.services.posts import PostService
|
||||
from app.services.rate_limit import create_limiter
|
||||
from app.services.sessions import SessionService
|
||||
|
||||
|
||||
# Resolve the package root once so template / static paths stay correct
|
||||
@@ -53,12 +72,12 @@ def create_app() -> FastAPI:
|
||||
3. Build the SQLAlchemy engine and install the PRAGMA listener.
|
||||
4. Apply SQL migrations (idempotent — no-op after first boot).
|
||||
5. Run the seed (idempotent — marked via ``schema_migrations``).
|
||||
6. Instantiate :class:`PostService` / :class:`PageService` and
|
||||
attach them to ``app.state`` so route dependencies can resolve
|
||||
them via ``request.app.state``.
|
||||
6. Instantiate services and attach them to ``app.state`` so route
|
||||
dependencies can resolve them via ``request.app.state``.
|
||||
7. Mount static files, attach the shared :class:`Jinja2Templates`,
|
||||
and register routers.
|
||||
8. Emit a single ``app_started`` structured log event.
|
||||
and register routers (including admin).
|
||||
8. Wire the SlowAPI limiter and its exception handler.
|
||||
9. Emit a single ``app_started`` structured log event.
|
||||
"""
|
||||
# Parse + validate configuration first so a bad environment fails fast
|
||||
# with a clear pydantic error before we touch logging / FastAPI.
|
||||
@@ -99,7 +118,8 @@ def create_app() -> FastAPI:
|
||||
# would be circular once admin/auth routers are added in later
|
||||
# phases). Route handlers pull it via a ``Depends(get_templates)``
|
||||
# function defined next to the routes.
|
||||
application.state.templates = Jinja2Templates(directory=_TEMPLATES_DIR)
|
||||
templates = Jinja2Templates(directory=_TEMPLATES_DIR)
|
||||
application.state.templates = templates
|
||||
|
||||
# Store the engine + services on ``app.state`` so the
|
||||
# dependency-injection helpers in :mod:`app.services.*` can find
|
||||
@@ -108,10 +128,46 @@ def create_app() -> FastAPI:
|
||||
application.state.post_service = PostService(engine)
|
||||
application.state.page_service = PageService(engine)
|
||||
|
||||
# --- Phase 3 wiring -----------------------------------------------------
|
||||
# itsdangerous signer: signs (and later verifies) session-cookie
|
||||
# values using SECRET_KEY and the salt "session". The same instance
|
||||
# is shared by every request — cheap to construct, no state beyond
|
||||
# the key.
|
||||
signer = URLSafeTimedSerializer(settings.secret_key, salt="session")
|
||||
application.state.signer = signer
|
||||
|
||||
# Audit first — EmailService and AuthService both depend on it
|
||||
# (EmailService indirectly via the request-path contract: failures
|
||||
# are logged, never raised).
|
||||
audit_service = AuditService(engine)
|
||||
email_service = EmailService(settings, templates)
|
||||
session_service = SessionService(engine, signer, settings)
|
||||
auth_service = AuthService(
|
||||
engine, email_service, session_service, audit_service, settings
|
||||
)
|
||||
|
||||
application.state.audit_service = audit_service
|
||||
application.state.email_service = email_service
|
||||
application.state.session_service = session_service
|
||||
application.state.auth_service = auth_service
|
||||
|
||||
# SlowAPI limiter + exception handler. The limiter is a module-level
|
||||
# singleton in app.services.rate_limit (because @limiter.limit has
|
||||
# to be applied at endpoint-definition time, before include_router).
|
||||
# We still attach it to app.state so SlowAPI's request-path
|
||||
# middleware can reach it via request.app.state.limiter.
|
||||
limiter = create_limiter()
|
||||
application.state.limiter = limiter
|
||||
|
||||
application.add_exception_handler(
|
||||
RateLimitExceeded, _make_rate_limit_handler(templates, audit_service)
|
||||
)
|
||||
|
||||
# Register routers. Kept explicit (no dynamic discovery) so the set of
|
||||
# mounted endpoints is trivially auditable.
|
||||
application.include_router(health_router)
|
||||
application.include_router(public_router)
|
||||
application.include_router(admin_router)
|
||||
|
||||
# Single structured startup event. Do NOT include secret material.
|
||||
logger = structlog.get_logger(__name__)
|
||||
@@ -125,6 +181,41 @@ def create_app() -> FastAPI:
|
||||
return application
|
||||
|
||||
|
||||
def _make_rate_limit_handler(
|
||||
templates: Jinja2Templates,
|
||||
audit_service: AuditService,
|
||||
):
|
||||
"""Build the FastAPI exception handler for ``RateLimitExceeded``.
|
||||
|
||||
Renders ``admin/rate_limited.html`` at HTTP 429 and writes a
|
||||
``rate_limited`` audit row scoped to the IP path. Email-scope
|
||||
rate-limits are handled inside AuthService and don't come through
|
||||
this handler.
|
||||
"""
|
||||
|
||||
async def _handler(request: Request, exc: RateLimitExceeded):
|
||||
# Best-effort endpoint path for the audit detail; the limiter
|
||||
# doesn't surface a structured endpoint name so we use the URL
|
||||
# path which is already stable / non-sensitive.
|
||||
endpoint = request.url.path
|
||||
ip = request.client.host if request.client else ""
|
||||
ua = request.headers.get("user-agent", "")
|
||||
audit_service.record(
|
||||
"rate_limited",
|
||||
ip=ip,
|
||||
user_agent=ua,
|
||||
detail={"scope": "ip", "endpoint": endpoint},
|
||||
)
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"admin/rate_limited.html",
|
||||
{},
|
||||
status_code=429,
|
||||
)
|
||||
|
||||
return _handler
|
||||
|
||||
|
||||
# Module-level ASGI handle. Uvicorn / gunicorn import this as
|
||||
# ``app.main:app``. Building it at import time is intentional: it fails
|
||||
# loudly at container start if configuration is invalid.
|
||||
|
||||
275
app/routes/admin.py
Normal file
275
app/routes/admin.py
Normal file
@@ -0,0 +1,275 @@
|
||||
"""Admin auth routes — magic-link login, consume, landing, logout.
|
||||
|
||||
Every handler here is deliberately thin: it delegates to
|
||||
:class:`app.services.auth.AuthService` or
|
||||
:class:`app.services.sessions.SessionService` and translates the
|
||||
result into an HTTP response. No SQL, no crypto, no email building.
|
||||
|
||||
Anti-enumeration contract
|
||||
-------------------------
|
||||
POST /admin/login always renders the same ``login_sent.html`` template
|
||||
with identical copy, identical status, and identical cookie state,
|
||||
regardless of whether the submitted address is on the allowlist. The
|
||||
only side-effects that differ are:
|
||||
|
||||
- token row inserted (allowlisted)
|
||||
- email dispatched (allowlisted)
|
||||
- audit row has ``allowlisted=true`` vs ``false``
|
||||
|
||||
GET /admin/auth/consume/{token} likewise uses a single failure page
|
||||
for every bad-token reason (missing / unknown / expired / already
|
||||
used). The specific reason lives only in the audit log.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Optional
|
||||
|
||||
import structlog
|
||||
from fastapi import APIRouter, Depends, Form, Request
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse, Response
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
from app.dependencies.auth import require_admin
|
||||
from app.models.entities import User
|
||||
from app.services.auth import AuthService, RateLimitedError
|
||||
from app.services.rate_limit import limiter
|
||||
from app.services.sessions import COOKIE_NAME, SessionService
|
||||
|
||||
|
||||
router: APIRouter = APIRouter(tags=["admin"])
|
||||
_log = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
# Lightweight email regex. We intentionally avoid ``email-validator``
|
||||
# (not pinned in requirements.txt) and a full RFC-5322 parser; the goal
|
||||
# is only to reject empty strings and obviously-not-an-email submissions
|
||||
# before calling the allowlist. Full validation is unnecessary because
|
||||
# non-allowlisted addresses are silently ignored anyway.
|
||||
_EMAIL_RE: re.Pattern[str] = re.compile(
|
||||
r"^[^@\s]+@[^@\s]+\.[^@\s]+$"
|
||||
)
|
||||
|
||||
|
||||
def _get_templates(request: Request) -> Jinja2Templates:
|
||||
"""Return the app-scoped :class:`Jinja2Templates`."""
|
||||
return request.app.state.templates
|
||||
|
||||
|
||||
def _get_auth_service(request: Request) -> AuthService:
|
||||
"""Return the app-scoped :class:`AuthService`."""
|
||||
return request.app.state.auth_service
|
||||
|
||||
|
||||
def _get_session_service(request: Request) -> SessionService:
|
||||
"""Return the app-scoped :class:`SessionService`."""
|
||||
return request.app.state.session_service
|
||||
|
||||
|
||||
def _client_ip(request: Request) -> str:
|
||||
"""Best-effort client IP from the request.
|
||||
|
||||
Starlette already respects ``X-Forwarded-For`` via the proxy-headers
|
||||
middleware Uvicorn installs with ``--proxy-headers``; that means
|
||||
``request.client.host`` is the real client IP.
|
||||
"""
|
||||
return request.client.host if request.client else ""
|
||||
|
||||
|
||||
def _user_agent(request: Request) -> str:
|
||||
"""Return the submitted User-Agent header (empty string if missing)."""
|
||||
return request.headers.get("user-agent", "")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /admin/login
|
||||
# ---------------------------------------------------------------------------
|
||||
@router.get("/admin/login", response_class=HTMLResponse, summary="Admin login form")
|
||||
def admin_login_form(
|
||||
request: Request,
|
||||
templates: Jinja2Templates = Depends(_get_templates),
|
||||
) -> HTMLResponse:
|
||||
"""Render the email-entry form.
|
||||
|
||||
No CSRF on the POST yet — see the ``# TODO(phase-6-csrf)`` in
|
||||
``admin_logout``. This handler accepts any caller.
|
||||
"""
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"admin/login.html",
|
||||
{"error": None, "email": ""},
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /admin/login
|
||||
# ---------------------------------------------------------------------------
|
||||
@router.post("/admin/login", response_class=HTMLResponse, summary="Request magic link")
|
||||
@limiter.limit("5/15 minutes")
|
||||
def admin_login_submit(
|
||||
request: Request,
|
||||
email: str = Form(default=""),
|
||||
templates: Jinja2Templates = Depends(_get_templates),
|
||||
auth: AuthService = Depends(_get_auth_service),
|
||||
) -> Response:
|
||||
"""Handle the login form submission.
|
||||
|
||||
Flow:
|
||||
1. Normalize + validate the email format. On format error we
|
||||
re-render ``login.html`` with a message (this is a UX concession
|
||||
— an invalid email shape is not a successful submission, so
|
||||
there's no enumeration risk).
|
||||
2. Call :meth:`AuthService.request_link`.
|
||||
3. Regardless of allowlist membership: render ``login_sent.html``
|
||||
with identical copy. On per-email rate-limit: render
|
||||
``rate_limited.html`` + 429.
|
||||
|
||||
Rate limiting
|
||||
-------------
|
||||
The ``@limiter.limit`` decoration is applied dynamically from
|
||||
``app.main`` so tests can bypass it — see the route registration
|
||||
helper in ``app.main``.
|
||||
"""
|
||||
normalized = (email or "").strip().lower()
|
||||
|
||||
if not normalized or not _EMAIL_RE.match(normalized):
|
||||
# Format error surfaces as a flash; there's nothing to leak
|
||||
# here because we haven't checked the allowlist yet.
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"admin/login.html",
|
||||
{
|
||||
"error": "Please enter a valid email address.",
|
||||
"email": email or "",
|
||||
},
|
||||
status_code=400,
|
||||
)
|
||||
|
||||
ip = _client_ip(request)
|
||||
ua = _user_agent(request)
|
||||
|
||||
try:
|
||||
auth.request_link(email=normalized, ip=ip, user_agent=ua)
|
||||
except RateLimitedError:
|
||||
# Per-email DB-side limit tripped. The SlowAPI IP-level limit
|
||||
# is handled separately via the registered exception handler.
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"admin/rate_limited.html",
|
||||
{},
|
||||
status_code=429,
|
||||
)
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"admin/login_sent.html",
|
||||
{},
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /admin/auth/consume/{token}
|
||||
# ---------------------------------------------------------------------------
|
||||
@router.get(
|
||||
"/admin/auth/consume/{token}",
|
||||
summary="Consume magic-link token",
|
||||
)
|
||||
@limiter.limit("20/15 minutes")
|
||||
def admin_auth_consume(
|
||||
request: Request,
|
||||
token: str,
|
||||
templates: Jinja2Templates = Depends(_get_templates),
|
||||
auth: AuthService = Depends(_get_auth_service),
|
||||
sessions: SessionService = Depends(_get_session_service),
|
||||
) -> Response:
|
||||
"""Consume a magic-link token and, on success, set the session cookie.
|
||||
|
||||
Failure path returns 400 + generic ``invalid or expired`` page.
|
||||
Success path sets ``cb_session`` and 303-redirects to ``/admin``.
|
||||
"""
|
||||
ip = _client_ip(request)
|
||||
ua = _user_agent(request)
|
||||
|
||||
result = auth.consume(raw_token=token, ip=ip, user_agent=ua)
|
||||
if result is None:
|
||||
# Generic failure response — the audit log has the real reason.
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"admin/login_failed.html",
|
||||
{},
|
||||
status_code=400,
|
||||
)
|
||||
|
||||
_user, _session, cookie_value = result
|
||||
|
||||
# 303 forces the browser to GET /admin on the next request.
|
||||
response = RedirectResponse(url="/admin", status_code=303)
|
||||
response.set_cookie(value=cookie_value, **sessions.cookie_params())
|
||||
return response
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /admin
|
||||
# ---------------------------------------------------------------------------
|
||||
@router.get("/admin", response_class=HTMLResponse, summary="Admin landing")
|
||||
def admin_index(
|
||||
request: Request,
|
||||
user: User = Depends(require_admin),
|
||||
templates: Jinja2Templates = Depends(_get_templates),
|
||||
) -> HTMLResponse:
|
||||
"""Render the authenticated admin landing page.
|
||||
|
||||
``require_admin`` handles the redirect-to-login case; by the time
|
||||
this handler runs, ``user`` is guaranteed to be populated.
|
||||
"""
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"admin/index.html",
|
||||
{"user": user},
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /admin/logout
|
||||
# ---------------------------------------------------------------------------
|
||||
# TODO(phase-6-csrf): Require a double-submit CSRF token on this POST.
|
||||
# Phase 6 will add middleware that validates a signed token against a
|
||||
# cookie; until then SameSite=Lax blocks cross-site POSTs in all
|
||||
# evergreen browsers, which is sufficient for this deploy's threat
|
||||
# model.
|
||||
@router.post("/admin/logout", summary="Log out the current admin")
|
||||
def admin_logout(
|
||||
request: Request,
|
||||
user: User = Depends(require_admin),
|
||||
sessions: SessionService = Depends(_get_session_service),
|
||||
) -> Response:
|
||||
"""Revoke the current session and clear the cookie.
|
||||
|
||||
Always issues a 303 redirect to ``/admin/login`` so browsers
|
||||
transparently follow and show the login form (with no cookie).
|
||||
"""
|
||||
from app.services.sessions import SessionService as _SS # noqa: F401
|
||||
|
||||
# Look up the session again via the cookie so we can revoke it
|
||||
# and emit a properly-correlated audit row.
|
||||
cookie_value: Optional[str] = request.cookies.get(COOKIE_NAME)
|
||||
session = sessions.lookup(cookie_value)
|
||||
audit = request.app.state.audit_service
|
||||
|
||||
if session is not None:
|
||||
sessions.revoke(session)
|
||||
audit.record(
|
||||
"session_revoked",
|
||||
email=user.email,
|
||||
user_id=user.id,
|
||||
ip=_client_ip(request),
|
||||
user_agent=_user_agent(request),
|
||||
detail={"session_id": session.token_hash[-6:]},
|
||||
)
|
||||
|
||||
response = RedirectResponse(url="/admin/login", status_code=303)
|
||||
# Clear the cookie by setting an empty value with Max-Age=0 and the
|
||||
# same Path so the browser actually removes it.
|
||||
response.delete_cookie(key=COOKIE_NAME, path="/")
|
||||
return response
|
||||
119
app/services/audit.py
Normal file
119
app/services/audit.py
Normal file
@@ -0,0 +1,119 @@
|
||||
"""Append-only auth audit log service.
|
||||
|
||||
Writes one row per auth event into the ``auth_events`` table. The rest of
|
||||
the auth stack calls :meth:`AuditService.record` to persist a structured,
|
||||
queryable audit trail without having to know the SQL or the row schema.
|
||||
|
||||
Security notes
|
||||
--------------
|
||||
- NEVER pass raw tokens, raw session IDs, or email bodies into ``detail``.
|
||||
Correlate sessions via the last 6 hex chars of their stored hash, never
|
||||
the full hash and never the raw value (CWE-200).
|
||||
- ``detail`` is persisted as JSON text; the schema column is ``TEXT NOT
|
||||
NULL DEFAULT '{}'`` and the writer always provides a valid JSON object.
|
||||
- All writes go through parameterized SQL with ``sqlalchemy.text``
|
||||
``:bind`` parameters; no string interpolation (CWE-89).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Mapping, Optional
|
||||
|
||||
import structlog
|
||||
from sqlalchemy import Engine, text
|
||||
|
||||
|
||||
_log = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
class AuditService:
|
||||
"""Persist rows into ``auth_events``.
|
||||
|
||||
The service is intentionally tiny: one write method plus a helper
|
||||
fetcher used by tests. No caching (this is an append-only audit
|
||||
log and reads are rare).
|
||||
"""
|
||||
|
||||
def __init__(self, engine: Engine) -> None:
|
||||
"""Store the shared SQLAlchemy engine by reference.
|
||||
|
||||
The service never opens its own engine — it reuses the one
|
||||
wired on ``app.state.engine``.
|
||||
"""
|
||||
self._engine: Engine = engine
|
||||
|
||||
def record(
|
||||
self,
|
||||
event_type: str,
|
||||
*,
|
||||
email: Optional[str] = None,
|
||||
user_id: Optional[int] = None,
|
||||
ip: str = "",
|
||||
user_agent: str = "",
|
||||
detail: Optional[Mapping[str, Any]] = None,
|
||||
) -> None:
|
||||
"""Insert a single audit row.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
event_type:
|
||||
One of the Phase 3 event types: ``link_requested``,
|
||||
``link_consumed``, ``consume_failed``, ``session_created``,
|
||||
``session_revoked``, ``rate_limited``.
|
||||
email:
|
||||
Submitted / target email (nullable when the event doesn't
|
||||
have one, e.g. session_revoked where we key off user_id).
|
||||
user_id:
|
||||
Foreign key into ``users``; nullable for pre-auth events.
|
||||
ip:
|
||||
Client IP at time of event. Always captured when available;
|
||||
empty string is acceptable for events originating outside
|
||||
a request context (which Phase 3 does not currently emit,
|
||||
but the column is NOT NULL and we want the door closed).
|
||||
user_agent:
|
||||
Client UA at time of event. Same NOT-NULL rationale as ``ip``.
|
||||
detail:
|
||||
Event-specific structured context (dict-like). Serialized
|
||||
to a compact JSON string. Defaults to ``{}`` when absent.
|
||||
|
||||
NEVER put a raw token or session ID here — only hashes (or
|
||||
their last 6 chars) and other non-sensitive metadata.
|
||||
"""
|
||||
# Always serialize to JSON text; the DB column enforces NOT NULL
|
||||
# with an empty-object default, and we honor that contract here
|
||||
# rather than relying on the default.
|
||||
detail_json = json.dumps(dict(detail) if detail is not None else {})
|
||||
now_iso = datetime.now(timezone.utc).isoformat()
|
||||
|
||||
with self._engine.begin() as conn:
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO auth_events"
|
||||
" (event_type, email, user_id, ip, user_agent,"
|
||||
" created_at, detail)"
|
||||
" VALUES (:event_type, :email, :user_id, :ip,"
|
||||
" :user_agent, :created_at, :detail)"
|
||||
),
|
||||
{
|
||||
"event_type": event_type,
|
||||
"email": email,
|
||||
"user_id": user_id,
|
||||
"ip": ip or "",
|
||||
"user_agent": user_agent or "",
|
||||
"created_at": now_iso,
|
||||
"detail": detail_json,
|
||||
},
|
||||
)
|
||||
|
||||
# Mirror the audit row to structured logs at INFO. We never log
|
||||
# the raw token / session ID, only the same detail dict (which
|
||||
# the caller already scrubbed) and the non-sensitive envelope.
|
||||
_log.info(
|
||||
"auth_event",
|
||||
event_type=event_type,
|
||||
email=email,
|
||||
user_id=user_id,
|
||||
detail=detail or {},
|
||||
)
|
||||
399
app/services/auth.py
Normal file
399
app/services/auth.py
Normal file
@@ -0,0 +1,399 @@
|
||||
"""Magic-link auth orchestration.
|
||||
|
||||
Glues token issuance / consumption, user auto-upsert on consume,
|
||||
email delivery, session creation, and audit logging together behind
|
||||
two methods:
|
||||
|
||||
- :meth:`AuthService.request_link` — handle POST /admin/login
|
||||
- :meth:`AuthService.consume` — handle GET /admin/auth/consume/{token}
|
||||
|
||||
Security decisions are concentrated here:
|
||||
|
||||
- Raw tokens live only in memory, the outbound email URL, and the
|
||||
single SHA-256 hash that ends up in the DB.
|
||||
- The allowlist check is ALWAYS performed with lowercased emails.
|
||||
- Non-allowlisted requests receive an identical response shape (handled
|
||||
by the caller; this service just short-circuits the token issue and
|
||||
still audits via ``link_requested`` with ``allowlisted=false``).
|
||||
- Rate limiting has two layers:
|
||||
1. SlowAPI IP-level decorator on the route (outside this module).
|
||||
2. DB-side per-email COUNT inside :meth:`request_link` — returns a
|
||||
sentinel that the route converts into an HTTP 429.
|
||||
- Consume is atomic: an ``UPDATE ... WHERE token_hash=? AND used_at IS
|
||||
NULL AND expires_at > ?`` with a ``rowcount == 1`` guard.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import secrets
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Optional
|
||||
|
||||
import structlog
|
||||
from sqlalchemy import Engine, text
|
||||
|
||||
from app.config import Settings
|
||||
from app.models.entities import Session, User
|
||||
from app.models.mappers import row_to_user
|
||||
from app.services.audit import AuditService
|
||||
from app.services.email import EmailService
|
||||
from app.services.sessions import SessionService
|
||||
|
||||
|
||||
_log = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
# Per-email rate limit: at most 5 tokens issued in a 15-minute window.
|
||||
# This is the DB-backed layer that survives process restarts; the
|
||||
# SlowAPI IP-level decorator on the route is the first line of defense.
|
||||
_PER_EMAIL_WINDOW_MIN: int = 15
|
||||
_PER_EMAIL_MAX: int = 5
|
||||
|
||||
|
||||
def _sha256(raw: str) -> str:
|
||||
"""Return the hex SHA-256 of a raw token.
|
||||
|
||||
SHA-256 for one-way hashing of high-entropy tokens is acceptable
|
||||
(see docs/security.md CWE-327). These tokens are already ≥256 bits
|
||||
from ``secrets.token_urlsafe(32)`` so a KDF is unnecessary.
|
||||
"""
|
||||
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
class RateLimitedError(Exception):
|
||||
"""Raised when a per-email rate limit trips inside the service.
|
||||
|
||||
Surfaced to the route layer so it can emit a 429 + render the
|
||||
``rate_limited.html`` template. Not used for IP-level limits —
|
||||
those are handled by SlowAPI's own exception.
|
||||
"""
|
||||
|
||||
|
||||
class AuthService:
|
||||
"""Request-link / consume orchestration for admin auth."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
engine: Engine,
|
||||
email: EmailService,
|
||||
sessions: SessionService,
|
||||
audit: AuditService,
|
||||
settings: Settings,
|
||||
) -> None:
|
||||
"""Store collaborators by reference."""
|
||||
self._engine: Engine = engine
|
||||
self._email: EmailService = email
|
||||
self._sessions: SessionService = sessions
|
||||
self._audit: AuditService = audit
|
||||
self._settings: Settings = settings
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# request_link
|
||||
# ------------------------------------------------------------------
|
||||
def request_link(
|
||||
self,
|
||||
*,
|
||||
email: str,
|
||||
ip: str,
|
||||
user_agent: str,
|
||||
) -> None:
|
||||
"""Handle POST /admin/login for a validated email.
|
||||
|
||||
Behavior
|
||||
--------
|
||||
1. Lowercase the email (allowlist comparison is case-insensitive).
|
||||
2. Check the admin allowlist.
|
||||
3. If NOT allowlisted: audit ``link_requested`` with
|
||||
``allowlisted=false`` and return. No token row, no email.
|
||||
4. If allowlisted: run the DB-side per-email rate-limit check.
|
||||
On trip: audit ``rate_limited`` and raise
|
||||
:class:`RateLimitedError`.
|
||||
5. Otherwise: insert a fresh token row (hash at rest), audit
|
||||
``link_requested`` with ``allowlisted=true``, and call
|
||||
:meth:`EmailService.send_magic_link`.
|
||||
|
||||
Callers MUST render the same "check your inbox" page regardless
|
||||
of the allowlist branch — see the admin route for the
|
||||
anti-enumeration contract.
|
||||
"""
|
||||
email = email.strip().lower()
|
||||
allowlisted = email in self._settings.admin_emails_list
|
||||
|
||||
# Always audit the request — this is the trail that catches
|
||||
# non-allowlisted attempts without leaking that info back to
|
||||
# the submitter.
|
||||
if not allowlisted:
|
||||
self._audit.record(
|
||||
"link_requested",
|
||||
email=email,
|
||||
ip=ip,
|
||||
user_agent=user_agent,
|
||||
detail={"allowlisted": False},
|
||||
)
|
||||
return
|
||||
|
||||
# DB-side per-email rate limit. We run it AFTER the allowlist
|
||||
# check so non-allowlisted spam doesn't cause extra queries.
|
||||
cutoff = datetime.now(timezone.utc) - timedelta(
|
||||
minutes=_PER_EMAIL_WINDOW_MIN
|
||||
)
|
||||
with self._engine.connect() as conn:
|
||||
row = conn.execute(
|
||||
text(
|
||||
"SELECT COUNT(*) AS c FROM magic_link_tokens"
|
||||
" WHERE email = :email AND created_at > :cutoff"
|
||||
),
|
||||
{"email": email, "cutoff": cutoff.isoformat()},
|
||||
).mappings().first()
|
||||
recent_count = int(row["c"]) if row is not None else 0
|
||||
if recent_count >= _PER_EMAIL_MAX:
|
||||
self._audit.record(
|
||||
"rate_limited",
|
||||
email=email,
|
||||
ip=ip,
|
||||
user_agent=user_agent,
|
||||
detail={"scope": "email", "endpoint": "/admin/login"},
|
||||
)
|
||||
raise RateLimitedError("per-email token limit reached")
|
||||
|
||||
# Mint the token — raw lives only in memory + email URL.
|
||||
raw = secrets.token_urlsafe(32)
|
||||
token_hash = _sha256(raw)
|
||||
now = datetime.now(timezone.utc)
|
||||
expires_at = now + timedelta(
|
||||
minutes=self._settings.magic_link_ttl_min
|
||||
)
|
||||
|
||||
with self._engine.begin() as conn:
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO magic_link_tokens"
|
||||
" (email, token_hash, created_at, expires_at,"
|
||||
" request_ip)"
|
||||
" VALUES (:email, :token_hash, :created_at,"
|
||||
" :expires_at, :request_ip)"
|
||||
),
|
||||
{
|
||||
"email": email,
|
||||
"token_hash": token_hash,
|
||||
"created_at": now.isoformat(),
|
||||
"expires_at": expires_at.isoformat(),
|
||||
"request_ip": ip or "",
|
||||
},
|
||||
)
|
||||
|
||||
self._audit.record(
|
||||
"link_requested",
|
||||
email=email,
|
||||
ip=ip,
|
||||
user_agent=user_agent,
|
||||
detail={"allowlisted": True},
|
||||
)
|
||||
|
||||
# Build the magic-link URL. Using a path param (not a query
|
||||
# string) keeps the raw token out of many access-log formats.
|
||||
base = self._settings.public_base_url.rstrip("/")
|
||||
url = f"{base}/admin/auth/consume/{raw}"
|
||||
display_name = email.split("@", 1)[0].title() or email
|
||||
|
||||
# EmailService never raises in dev; in prod it may log an
|
||||
# exception but still return. Either way the request-path
|
||||
# response is identical.
|
||||
self._email.send_magic_link(
|
||||
to=email,
|
||||
url=url,
|
||||
display_name=display_name,
|
||||
ttl_min=self._settings.magic_link_ttl_min,
|
||||
expires_at=expires_at,
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# consume
|
||||
# ------------------------------------------------------------------
|
||||
def consume(
|
||||
self,
|
||||
*,
|
||||
raw_token: str,
|
||||
ip: str,
|
||||
user_agent: str,
|
||||
) -> Optional[tuple[User, Session, str]]:
|
||||
"""Consume a magic-link token and issue a session.
|
||||
|
||||
Returns ``(user, session, cookie_value)`` on success, ``None``
|
||||
on any invalid / expired / already-used / unknown token (the
|
||||
caller should render the generic failure page with HTTP 400).
|
||||
|
||||
Concurrency safety: the UPDATE statement's WHERE clause is the
|
||||
atomic guard — if two requests race with the same token, only
|
||||
one will get ``rowcount == 1``. The loser is treated as a
|
||||
replay.
|
||||
"""
|
||||
if not raw_token:
|
||||
# Audit runs in its own transaction (see notes below) so we
|
||||
# can record the event without opening a write lock we'd
|
||||
# then try to re-enter from this service.
|
||||
self._audit.record(
|
||||
"consume_failed",
|
||||
ip=ip,
|
||||
user_agent=user_agent,
|
||||
detail={"reason": "not_found"},
|
||||
)
|
||||
return None
|
||||
|
||||
token_hash = _sha256(raw_token)
|
||||
now = datetime.now(timezone.utc)
|
||||
now_iso = now.isoformat()
|
||||
|
||||
# ``engine.begin()`` holds a write lock on SQLite for its whole
|
||||
# scope. AuditService opens its OWN transaction and would be
|
||||
# blocked behind this one — so we do not call ``self._audit``
|
||||
# until AFTER this block exits. Track the outcome locally.
|
||||
reloaded = None
|
||||
failure_reason: Optional[str] = None
|
||||
email: Optional[str] = None
|
||||
user_id: Optional[int] = None
|
||||
|
||||
# Atomic single-use: only mark the row used if it's still
|
||||
# valid. rowcount tells us whether we were the one to claim it.
|
||||
with self._engine.begin() as conn:
|
||||
update_result = conn.execute(
|
||||
text(
|
||||
"UPDATE magic_link_tokens"
|
||||
" SET used_at = :now"
|
||||
" WHERE token_hash = :h"
|
||||
" AND used_at IS NULL"
|
||||
" AND expires_at > :now"
|
||||
),
|
||||
{"now": now_iso, "h": token_hash},
|
||||
)
|
||||
|
||||
if update_result.rowcount != 1:
|
||||
# Distinguish expired / used / not_found for the audit
|
||||
# trail (not for the client response). Run an extra
|
||||
# SELECT only on the failure path so the happy path
|
||||
# stays a single write.
|
||||
row = conn.execute(
|
||||
text(
|
||||
"SELECT expires_at, used_at FROM magic_link_tokens"
|
||||
" WHERE token_hash = :h LIMIT 1"
|
||||
),
|
||||
{"h": token_hash},
|
||||
).mappings().first()
|
||||
|
||||
if row is None:
|
||||
failure_reason = "not_found"
|
||||
elif row["used_at"] is not None:
|
||||
failure_reason = "used"
|
||||
else:
|
||||
failure_reason = "expired"
|
||||
else:
|
||||
# Token claimed. Look up the email so we can upsert the user.
|
||||
token_row = conn.execute(
|
||||
text(
|
||||
"SELECT email FROM magic_link_tokens"
|
||||
" WHERE token_hash = :h LIMIT 1"
|
||||
),
|
||||
{"h": token_hash},
|
||||
).mappings().first()
|
||||
|
||||
# Should always exist — the UPDATE just wrote to it.
|
||||
if token_row is None: # pragma: no cover — impossible path
|
||||
failure_reason = "not_found"
|
||||
else:
|
||||
email = token_row["email"]
|
||||
|
||||
# Upsert user. Existing row → bump last_login_at and
|
||||
# re-activate; missing row → insert with titlecased
|
||||
# local part as display_name and active=1.
|
||||
user_row = conn.execute(
|
||||
text("SELECT id FROM users WHERE email = :e"),
|
||||
{"e": email},
|
||||
).mappings().first()
|
||||
|
||||
if user_row is None:
|
||||
display_name = email.split("@", 1)[0].title() or email
|
||||
insert_result = conn.execute(
|
||||
text(
|
||||
"INSERT INTO users"
|
||||
" (email, display_name, created_at,"
|
||||
" last_login_at, active)"
|
||||
" VALUES (:email, :display_name, :created_at,"
|
||||
" :last_login_at, 1)"
|
||||
),
|
||||
{
|
||||
"email": email,
|
||||
"display_name": display_name,
|
||||
"created_at": now_iso,
|
||||
"last_login_at": now_iso,
|
||||
},
|
||||
)
|
||||
user_id = int(insert_result.lastrowid) # type: ignore[arg-type]
|
||||
else:
|
||||
user_id = int(user_row["id"])
|
||||
conn.execute(
|
||||
text(
|
||||
"UPDATE users"
|
||||
" SET last_login_at = :now, active = 1"
|
||||
" WHERE id = :id"
|
||||
),
|
||||
{"now": now_iso, "id": user_id},
|
||||
)
|
||||
|
||||
# Reload the user row so we return a fully-populated
|
||||
# entity.
|
||||
reloaded = conn.execute(
|
||||
text(
|
||||
"SELECT id, email, display_name, created_at,"
|
||||
" last_login_at, active"
|
||||
" FROM users WHERE id = :id"
|
||||
),
|
||||
{"id": user_id},
|
||||
).mappings().first()
|
||||
|
||||
# --- Post-transaction: audit + session create --------------------
|
||||
# Running these AFTER the transaction closes avoids the SQLite
|
||||
# single-writer deadlock that would happen if AuditService tried
|
||||
# to open a new write connection from inside the block above.
|
||||
if failure_reason is not None:
|
||||
self._audit.record(
|
||||
"consume_failed",
|
||||
ip=ip,
|
||||
user_agent=user_agent,
|
||||
detail={"reason": failure_reason},
|
||||
)
|
||||
return None
|
||||
|
||||
if reloaded is None: # pragma: no cover — impossible path
|
||||
return None
|
||||
user = row_to_user(reloaded)
|
||||
|
||||
# Session creation runs in its own transaction now that the
|
||||
# consume write lock has been released.
|
||||
session, cookie_value = self._sessions.create(
|
||||
user_id=user.id, ip=ip, user_agent=user_agent
|
||||
)
|
||||
|
||||
self._audit.record(
|
||||
"link_consumed",
|
||||
email=email,
|
||||
user_id=user.id,
|
||||
ip=ip,
|
||||
user_agent=user_agent,
|
||||
detail={"user_id": user.id},
|
||||
)
|
||||
self._audit.record(
|
||||
"session_created",
|
||||
email=email,
|
||||
user_id=user.id,
|
||||
ip=ip,
|
||||
user_agent=user_agent,
|
||||
# last 6 hex chars of the stored hash — enough to correlate,
|
||||
# not enough to reverse.
|
||||
detail={
|
||||
"session_id": session.token_hash[-6:],
|
||||
"user_id": user.id,
|
||||
},
|
||||
)
|
||||
|
||||
return user, session, cookie_value
|
||||
141
app/services/email.py
Normal file
141
app/services/email.py
Normal file
@@ -0,0 +1,141 @@
|
||||
"""Transactional email sender with a dev-mode log fallback.
|
||||
|
||||
Thin wrapper around the Resend API (``resend.Emails.send``). Renders
|
||||
both HTML and plaintext magic-link bodies from Jinja templates to keep
|
||||
copy out of Python code.
|
||||
|
||||
Security and UX rules
|
||||
---------------------
|
||||
- **Never raise on missing credentials in development.** A 500 from the
|
||||
login POST would expose whether an email is on the allowlist
|
||||
(successful sends would succeed, non-allowlisted "sends" would still
|
||||
short-circuit) and it would also break local dev. In development we
|
||||
log a ``magic_link_dev_fallback`` structured event with the full
|
||||
magic-link URL so the developer can copy it.
|
||||
- **Production must fail at startup** if Resend credentials are absent;
|
||||
that validator lives in :class:`app.config.Settings`, not here.
|
||||
- Never log the raw token on its own — only as part of the URL in the
|
||||
dev fallback (which is the whole point of the fallback).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
import structlog
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
from app.config import Settings
|
||||
|
||||
|
||||
_log = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
class EmailService:
|
||||
"""Send magic-link emails via Resend, with a dev-mode log fallback."""
|
||||
|
||||
def __init__(self, settings: Settings, templates: Jinja2Templates) -> None:
|
||||
"""Store dependencies by reference.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
settings:
|
||||
Application settings; used to pick up ``resend_api_key`` /
|
||||
``resend_from`` / ``app_env`` at send time so rotating them
|
||||
at runtime (dev) works.
|
||||
templates:
|
||||
Shared Jinja2 environment. We reuse the app-level one so
|
||||
template autoescape defaults and the search path match the
|
||||
rest of the site.
|
||||
"""
|
||||
self._settings: Settings = settings
|
||||
self._templates: Jinja2Templates = templates
|
||||
|
||||
def send_magic_link(
|
||||
self,
|
||||
*,
|
||||
to: str,
|
||||
url: str,
|
||||
display_name: str,
|
||||
ttl_min: int,
|
||||
expires_at: datetime,
|
||||
) -> None:
|
||||
"""Send a magic-link email to ``to`` or log the URL in dev.
|
||||
|
||||
Behavior
|
||||
--------
|
||||
- If ``settings.resend_api_key`` is truthy, render both bodies
|
||||
and send via Resend.
|
||||
- Otherwise (development only — the production config validator
|
||||
refuses to boot without a key), emit a structured log event
|
||||
``magic_link_dev_fallback`` that includes the full URL.
|
||||
|
||||
Never raises in the dev fallback path; errors from the Resend
|
||||
API surface as logged exceptions so the request-handler layer
|
||||
always returns the same response shape regardless of whether
|
||||
the email was actually sent (CWE-200 / anti-enumeration).
|
||||
"""
|
||||
# Build both template bodies first so any rendering error
|
||||
# surfaces before we talk to the network.
|
||||
ctx = {
|
||||
"display_name": display_name,
|
||||
"magic_link_url": url,
|
||||
"expires_at": expires_at.isoformat(),
|
||||
"ttl_min": ttl_min,
|
||||
}
|
||||
html_body = self._render("emails/magic_link.html", ctx)
|
||||
text_body = self._render("emails/magic_link.txt", ctx)
|
||||
|
||||
api_key: Optional[str] = self._settings.resend_api_key
|
||||
sender: Optional[str] = self._settings.resend_from
|
||||
|
||||
# Dev fallback path: no key configured. Log the URL at INFO so
|
||||
# the developer can complete the flow, and return.
|
||||
if not api_key or not sender:
|
||||
_log.info(
|
||||
"magic_link_dev_fallback",
|
||||
to=to,
|
||||
# Raw token is embedded in the URL; acceptable because
|
||||
# this path ONLY runs in local dev (production validator
|
||||
# refuses to boot without RESEND_API_KEY).
|
||||
magic_link_url=url,
|
||||
ttl_min=ttl_min,
|
||||
)
|
||||
return
|
||||
|
||||
# Real send. We import here to avoid taking a hard import-time
|
||||
# dependency on `resend`'s module-level state during tests that
|
||||
# never exercise the send path.
|
||||
try:
|
||||
import resend # type: ignore[import-untyped]
|
||||
|
||||
resend.api_key = api_key
|
||||
resend.Emails.send(
|
||||
{
|
||||
"from": sender,
|
||||
"to": to,
|
||||
"subject": "Your Chicken Babies R Us admin login link",
|
||||
"html": html_body,
|
||||
"text": text_body,
|
||||
}
|
||||
)
|
||||
_log.info("magic_link_email_sent", to=to)
|
||||
except Exception: # noqa: BLE001
|
||||
# Do NOT re-raise from the request path — see anti-enumeration
|
||||
# note at module top. Log with redacted context.
|
||||
_log.exception("magic_link_email_failed", to=to)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internals
|
||||
# ------------------------------------------------------------------
|
||||
def _render(self, template_name: str, context: dict) -> str:
|
||||
"""Render a Jinja template to a string.
|
||||
|
||||
We use the underlying Jinja environment directly so we get a
|
||||
plain string back (``Jinja2Templates.TemplateResponse`` wraps
|
||||
the output in an HTTP response, which is not what we want for
|
||||
outbound email bodies).
|
||||
"""
|
||||
template = self._templates.env.get_template(template_name)
|
||||
return template.render(**context)
|
||||
44
app/services/rate_limit.py
Normal file
44
app/services/rate_limit.py
Normal file
@@ -0,0 +1,44 @@
|
||||
"""SlowAPI rate-limiter wiring for auth endpoints.
|
||||
|
||||
Only one limiter is used process-wide; it's built in
|
||||
:func:`create_limiter` and stored on ``app.state.limiter`` so the
|
||||
``@limiter.limit`` decorator can pick it up from the request.
|
||||
|
||||
Storage is ``memory://``. At this scale (single-digit requests/second,
|
||||
single container) a persistent backend is not worth the operational
|
||||
cost, and the consequences of losing limiter state on restart are
|
||||
acceptable — the DB-side per-email check (in
|
||||
:mod:`app.services.auth`) catches sustained abuse across restarts.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from slowapi import Limiter
|
||||
from slowapi.util import get_remote_address
|
||||
|
||||
|
||||
# Process-wide singleton. Module-level because SlowAPI's ``@limiter.limit``
|
||||
# decorator has to be applied at endpoint-definition time (before the
|
||||
# router is wired into the FastAPI app), and that has to reference the
|
||||
# same limiter instance that the request path consults via
|
||||
# ``request.app.state.limiter``.
|
||||
#
|
||||
# Storage is ``memory://``: in-process, single-container scale. Restarts
|
||||
# drop in-flight counters — acceptable because the DB-side per-email
|
||||
# check in :mod:`app.services.auth` backs this up for longer-lived abuse
|
||||
# patterns.
|
||||
limiter: Limiter = Limiter(
|
||||
key_func=get_remote_address,
|
||||
storage_uri="memory://",
|
||||
)
|
||||
|
||||
|
||||
def create_limiter() -> Limiter:
|
||||
"""Return the process-wide :class:`slowapi.Limiter` singleton.
|
||||
|
||||
Kept as a function (rather than exposing the module-level
|
||||
``limiter`` directly) to match the service-factory pattern used by
|
||||
:class:`AuditService` / :class:`EmailService` / ... and to give
|
||||
tests a hook to monkeypatch if they ever need a per-test limiter.
|
||||
"""
|
||||
return limiter
|
||||
240
app/services/sessions.py
Normal file
240
app/services/sessions.py
Normal file
@@ -0,0 +1,240 @@
|
||||
"""Server-side session issuance, lookup, and revocation.
|
||||
|
||||
Sessions combine two storage layers:
|
||||
|
||||
1. **Database row** in ``sessions``. Stores ``sha256(raw)`` as
|
||||
``token_hash`` (never the raw token), ``expires_at`` set to
|
||||
``SESSION_MAX_DAYS`` from creation, and ``revoked_at`` for
|
||||
soft-delete on logout.
|
||||
2. **Signed cookie** (``cb_session``). Carries the raw session id
|
||||
wrapped by :class:`itsdangerous.URLSafeTimedSerializer` with the
|
||||
configured ``SECRET_KEY`` and salt ``session``. The signature
|
||||
prevents tampering; the DB lookup prevents replay after logout.
|
||||
|
||||
Security and behavior rules
|
||||
---------------------------
|
||||
- Raw session IDs live only in memory and the outbound cookie. The DB
|
||||
stores only the SHA-256 hash.
|
||||
- ``Secure`` cookie flag is OFF in development (so plain-HTTP
|
||||
``127.0.0.1`` works) and ON in production. All other flags
|
||||
(``HttpOnly``, ``SameSite=Lax``, ``Path=/``) are always set.
|
||||
- Revocation flips ``revoked_at`` but NEVER deletes the row; the audit
|
||||
trail must survive logout.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import secrets
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Optional
|
||||
|
||||
import structlog
|
||||
from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer
|
||||
from sqlalchemy import Engine, text
|
||||
|
||||
from app.config import Settings
|
||||
from app.models.entities import Session
|
||||
from app.models.mappers import row_to_session
|
||||
|
||||
|
||||
_log = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
# Cookie name used on every request. Shared constant so routes and
|
||||
# dependencies don't drift.
|
||||
COOKIE_NAME: str = "cb_session"
|
||||
|
||||
|
||||
def _sha256(raw: str) -> str:
|
||||
"""Return the hex SHA-256 of a raw token.
|
||||
|
||||
SHA-256 is explicitly permitted for non-password one-way hashing
|
||||
(see docs/security.md CWE-327). Tokens here are already
|
||||
high-entropy (256-bit ``secrets.token_urlsafe(32)``) so a single
|
||||
pass is sufficient — we don't need a slow-hash KDF.
|
||||
"""
|
||||
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
class SessionService:
|
||||
"""Create, look up, and revoke admin sessions."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
engine: Engine,
|
||||
signer: URLSafeTimedSerializer,
|
||||
settings: Settings,
|
||||
) -> None:
|
||||
"""Store the engine, signer, and settings by reference.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
engine:
|
||||
Shared SQLAlchemy engine.
|
||||
signer:
|
||||
Pre-built ``itsdangerous.URLSafeTimedSerializer`` bound to
|
||||
``settings.secret_key`` with salt ``"session"``. Injected
|
||||
rather than constructed here so the same instance is used
|
||||
across all consumers (and tests can monkeypatch).
|
||||
settings:
|
||||
Application settings; we read ``session_max_days`` and
|
||||
``app_env`` (to decide the ``Secure`` cookie flag).
|
||||
"""
|
||||
self._engine: Engine = engine
|
||||
self._signer: URLSafeTimedSerializer = signer
|
||||
self._settings: Settings = settings
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Creation
|
||||
# ------------------------------------------------------------------
|
||||
def create(
|
||||
self,
|
||||
*,
|
||||
user_id: int,
|
||||
ip: str,
|
||||
user_agent: str,
|
||||
) -> tuple[Session, str]:
|
||||
"""Mint a new session row and return the signed cookie value.
|
||||
|
||||
Returns a ``(session, cookie_value)`` tuple. The caller is
|
||||
responsible for attaching the cookie to the HTTP response with
|
||||
the appropriate flags — use :meth:`cookie_params` for those.
|
||||
"""
|
||||
raw = secrets.token_urlsafe(32) # ≥256 bits of entropy
|
||||
token_hash = _sha256(raw)
|
||||
now = datetime.now(timezone.utc)
|
||||
expires_at = now + timedelta(days=self._settings.session_max_days)
|
||||
|
||||
with self._engine.begin() as conn:
|
||||
result = conn.execute(
|
||||
text(
|
||||
"INSERT INTO sessions"
|
||||
" (user_id, token_hash, created_at, expires_at,"
|
||||
" ip, user_agent)"
|
||||
" VALUES (:user_id, :token_hash, :created_at,"
|
||||
" :expires_at, :ip, :user_agent)"
|
||||
),
|
||||
{
|
||||
"user_id": user_id,
|
||||
"token_hash": token_hash,
|
||||
"created_at": now.isoformat(),
|
||||
"expires_at": expires_at.isoformat(),
|
||||
"ip": ip or "",
|
||||
"user_agent": user_agent or "",
|
||||
},
|
||||
)
|
||||
new_id = int(result.lastrowid) # type: ignore[arg-type]
|
||||
|
||||
row = conn.execute(
|
||||
text(
|
||||
"SELECT id, user_id, token_hash, created_at, expires_at,"
|
||||
" ip, user_agent, revoked_at"
|
||||
" FROM sessions WHERE id = :id"
|
||||
),
|
||||
{"id": new_id},
|
||||
).mappings().first()
|
||||
|
||||
if row is None: # pragma: no cover — insert just succeeded
|
||||
raise RuntimeError("failed to reload just-inserted session row")
|
||||
|
||||
session = row_to_session(row)
|
||||
cookie_value = self._signer.dumps(raw)
|
||||
return session, cookie_value
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Lookup
|
||||
# ------------------------------------------------------------------
|
||||
def lookup(self, cookie_value: Optional[str]) -> Optional[Session]:
|
||||
"""Resolve a cookie value to an active :class:`Session`.
|
||||
|
||||
Returns ``None`` on:
|
||||
- missing cookie
|
||||
- invalid signature
|
||||
- expired signature (we reject if older than
|
||||
``session_max_days``)
|
||||
- no matching row
|
||||
- session revoked or past its ``expires_at``
|
||||
|
||||
All failure modes intentionally return the same ``None`` so
|
||||
callers cannot distinguish them (CWE-200).
|
||||
"""
|
||||
if not cookie_value:
|
||||
return None
|
||||
|
||||
max_age_s = self._settings.session_max_days * 86400
|
||||
try:
|
||||
raw: str = self._signer.loads(cookie_value, max_age=max_age_s)
|
||||
except SignatureExpired:
|
||||
_log.info("session_cookie_expired")
|
||||
return None
|
||||
except BadSignature:
|
||||
_log.info("session_cookie_bad_signature")
|
||||
return None
|
||||
|
||||
token_hash = _sha256(raw)
|
||||
now_iso = datetime.now(timezone.utc).isoformat()
|
||||
|
||||
with self._engine.connect() as conn:
|
||||
row = conn.execute(
|
||||
text(
|
||||
"SELECT id, user_id, token_hash, created_at, expires_at,"
|
||||
" ip, user_agent, revoked_at"
|
||||
" FROM sessions"
|
||||
" WHERE token_hash = :h"
|
||||
" AND revoked_at IS NULL"
|
||||
" AND expires_at > :now"
|
||||
" LIMIT 1"
|
||||
),
|
||||
{"h": token_hash, "now": now_iso},
|
||||
).mappings().first()
|
||||
|
||||
if row is None:
|
||||
return None
|
||||
return row_to_session(row)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Revocation
|
||||
# ------------------------------------------------------------------
|
||||
def revoke(self, session: Session) -> None:
|
||||
"""Mark a session as revoked (soft-delete).
|
||||
|
||||
The row stays in the DB so the audit trail remains intact; the
|
||||
``revoked_at`` timestamp is what the ``lookup`` query filters
|
||||
on.
|
||||
"""
|
||||
now_iso = datetime.now(timezone.utc).isoformat()
|
||||
with self._engine.begin() as conn:
|
||||
conn.execute(
|
||||
text(
|
||||
"UPDATE sessions SET revoked_at = :now"
|
||||
" WHERE id = :id AND revoked_at IS NULL"
|
||||
),
|
||||
{"now": now_iso, "id": session.id},
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Cookie helpers
|
||||
# ------------------------------------------------------------------
|
||||
def cookie_params(self) -> dict:
|
||||
"""Return kwargs for ``response.set_cookie`` matching our policy.
|
||||
|
||||
Values:
|
||||
- ``key`` = :data:`COOKIE_NAME`
|
||||
- ``httponly=True``
|
||||
- ``samesite="lax"``
|
||||
- ``secure`` = True only in production (plain-HTTP 127.0.0.1
|
||||
needs ``Secure=False`` in dev)
|
||||
- ``max_age`` = ``session_max_days * 86400``
|
||||
- ``path="/"``
|
||||
|
||||
The caller supplies ``value`` separately.
|
||||
"""
|
||||
return {
|
||||
"key": COOKIE_NAME,
|
||||
"httponly": True,
|
||||
"samesite": "lax",
|
||||
"secure": self._settings.app_env == "production",
|
||||
"max_age": self._settings.session_max_days * 86400,
|
||||
"path": "/",
|
||||
}
|
||||
76
app/templates/admin/base.html
Normal file
76
app/templates/admin/base.html
Normal file
@@ -0,0 +1,76 @@
|
||||
{#
|
||||
Minimal admin layout shell.
|
||||
|
||||
Reuses the public site's CSS (palette + typography) so Head Hen sees a
|
||||
consistent visual language without us maintaining a second stylesheet.
|
||||
Intentionally simpler than the public base: no marketing hero,
|
||||
no multi-link primary nav — just the brand mark, the admin context
|
||||
label, and a logout control when the viewer is authenticated.
|
||||
|
||||
Context the child template may override:
|
||||
- title : <title> content
|
||||
- content : main body
|
||||
- user : app.models.entities.User | None
|
||||
(passed by the index route; omitted on pre-auth pages)
|
||||
#}<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>{% block title %}Admin — Chicken Babies R Us{% endblock %}</title>
|
||||
<meta name="robots" content="noindex, nofollow">
|
||||
<link rel="icon" href="{{ url_for('static', path='img/favicon.ico') }}" sizes="any">
|
||||
<link rel="stylesheet" href="{{ url_for('static', path='css/site.css') }}">
|
||||
</head>
|
||||
<body class="admin-body">
|
||||
<a class="skip-link" href="#main-content">Skip to main content</a>
|
||||
|
||||
<header class="site-header">
|
||||
<div class="wrap site-header__wrap">
|
||||
<a class="site-header__brand" href="/" aria-label="Chicken Babies R Us home">
|
||||
<picture>
|
||||
<source srcset="{{ url_for('static', path='img/logo.webp') }}" type="image/webp">
|
||||
<img src="{{ url_for('static', path='img/logo.png') }}"
|
||||
alt="Chicken Babies R Us"
|
||||
height="48"
|
||||
class="site-header__logo">
|
||||
</picture>
|
||||
</a>
|
||||
|
||||
<nav class="site-nav" aria-label="Admin">
|
||||
<ul class="site-nav__list">
|
||||
<li class="site-nav__item">
|
||||
<span class="site-nav__link" aria-current="page">Admin</span>
|
||||
</li>
|
||||
{% if user is defined and user %}
|
||||
<li class="site-nav__item">
|
||||
{#
|
||||
Plain POST form — no CSRF token yet. Phase 6 adds a
|
||||
double-submit token; SameSite=Lax is sufficient in the
|
||||
meantime.
|
||||
#}
|
||||
<form action="/admin/logout" method="post" class="site-nav__logout-form">
|
||||
<button type="submit" class="btn btn--link">Log out</button>
|
||||
</form>
|
||||
</li>
|
||||
{% endif %}
|
||||
</ul>
|
||||
</nav>
|
||||
</div>
|
||||
</header>
|
||||
|
||||
<main id="main-content" class="site-main" tabindex="-1">
|
||||
<div class="wrap">
|
||||
{% block content %}{% endblock %}
|
||||
</div>
|
||||
</main>
|
||||
|
||||
<footer class="site-footer">
|
||||
<div class="wrap site-footer__wrap">
|
||||
<p class="site-footer__tag">
|
||||
Chicken Babies R Us · Admin
|
||||
</p>
|
||||
</div>
|
||||
</footer>
|
||||
</body>
|
||||
</html>
|
||||
29
app/templates/admin/index.html
Normal file
29
app/templates/admin/index.html
Normal file
@@ -0,0 +1,29 @@
|
||||
{#
|
||||
Admin landing page (post-login).
|
||||
|
||||
Phase 3 keeps this minimal — Phase 4 turns it into the CMS
|
||||
dashboard (pages + posts + media). For now it's a welcome screen +
|
||||
logout control (rendered in admin/base.html's nav since ``user`` is
|
||||
present in context).
|
||||
|
||||
Context:
|
||||
- user : app.models.entities.User (guaranteed by require_admin)
|
||||
#}
|
||||
{% extends "admin/base.html" %}
|
||||
|
||||
{% block title %}Welcome — Admin{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<article class="page-article">
|
||||
<header class="page-article__header">
|
||||
<h1 class="page-article__title">Welcome, {{ user.display_name }}</h1>
|
||||
</header>
|
||||
<p>
|
||||
You're logged in as <code>{{ user.email }}</code>. The dashboard
|
||||
(pages, posts, media) lands in Phase 4.
|
||||
</p>
|
||||
<p>
|
||||
<a href="/">« Back to the public site</a>
|
||||
</p>
|
||||
</article>
|
||||
{% endblock %}
|
||||
48
app/templates/admin/login.html
Normal file
48
app/templates/admin/login.html
Normal file
@@ -0,0 +1,48 @@
|
||||
{#
|
||||
Admin login form.
|
||||
|
||||
Single email input, no JS. If the submitted address is on the
|
||||
allowlist (ADMIN_EMAILS env var), the POST handler emails a
|
||||
one-time magic link; otherwise it silently succeeds without
|
||||
sending (anti-enumeration).
|
||||
|
||||
Context:
|
||||
- error : str | None (format validation errors only)
|
||||
- email : str (pre-fill on re-render after format error)
|
||||
#}
|
||||
{% extends "admin/base.html" %}
|
||||
|
||||
{% block title %}Log in — Admin{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<article class="page-article">
|
||||
<header class="page-article__header">
|
||||
<h1 class="page-article__title">Admin log in</h1>
|
||||
</header>
|
||||
|
||||
<p>
|
||||
Enter your admin email and we'll send a one-time login link
|
||||
that expires after 15 minutes.
|
||||
</p>
|
||||
|
||||
{% if error %}
|
||||
<p class="admin-flash admin-flash--error" role="alert">{{ error }}</p>
|
||||
{% endif %}
|
||||
|
||||
<form class="contact-form" action="/admin/login" method="post" novalidate>
|
||||
<div class="contact-form__field">
|
||||
<label for="admin-email">Email</label>
|
||||
<input type="email"
|
||||
id="admin-email"
|
||||
name="email"
|
||||
autocomplete="email"
|
||||
value="{{ email or '' }}"
|
||||
required>
|
||||
</div>
|
||||
|
||||
<div class="contact-form__actions">
|
||||
<button type="submit" class="btn btn--primary">Send login link</button>
|
||||
</div>
|
||||
</form>
|
||||
</article>
|
||||
{% endblock %}
|
||||
28
app/templates/admin/login_failed.html
Normal file
28
app/templates/admin/login_failed.html
Normal file
@@ -0,0 +1,28 @@
|
||||
{#
|
||||
Generic failure page for GET /admin/auth/consume/{token}.
|
||||
|
||||
Deliberately does NOT distinguish between:
|
||||
- token not found
|
||||
- token expired
|
||||
- token already consumed
|
||||
The audit log has the real reason; the visitor only needs to know
|
||||
they need a fresh link.
|
||||
#}
|
||||
{% extends "admin/base.html" %}
|
||||
|
||||
{% block title %}Login link invalid — Admin{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<article class="page-article">
|
||||
<header class="page-article__header">
|
||||
<h1 class="page-article__title">Login link invalid or expired</h1>
|
||||
</header>
|
||||
<p>
|
||||
That login link isn't valid any more. Links expire after 15
|
||||
minutes and can only be used once.
|
||||
</p>
|
||||
<p>
|
||||
<a href="/admin/login">Request a new link</a>
|
||||
</p>
|
||||
</article>
|
||||
{% endblock %}
|
||||
25
app/templates/admin/login_sent.html
Normal file
25
app/templates/admin/login_sent.html
Normal file
@@ -0,0 +1,25 @@
|
||||
{#
|
||||
Post-submission page rendered after POST /admin/login.
|
||||
|
||||
Same copy for every outcome (allowlisted vs. not) to avoid leaking
|
||||
which emails are real admins.
|
||||
#}
|
||||
{% extends "admin/base.html" %}
|
||||
|
||||
{% block title %}Check your inbox — Admin{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<article class="page-article">
|
||||
<header class="page-article__header">
|
||||
<h1 class="page-article__title">Check your inbox</h1>
|
||||
</header>
|
||||
<p>
|
||||
If your email is on the allowlist, you'll receive a login link
|
||||
shortly. The link expires after 15 minutes and can only be used
|
||||
once.
|
||||
</p>
|
||||
<p>
|
||||
<a href="/admin/login">Request another link</a>
|
||||
</p>
|
||||
</article>
|
||||
{% endblock %}
|
||||
25
app/templates/admin/rate_limited.html
Normal file
25
app/templates/admin/rate_limited.html
Normal file
@@ -0,0 +1,25 @@
|
||||
{#
|
||||
Rendered at HTTP 429 when either the SlowAPI IP limit or the DB-side
|
||||
per-email limit trips on POST /admin/login.
|
||||
|
||||
Same template for both trigger paths — we don't want to tell the
|
||||
submitter whether the limit was IP-wide or email-specific.
|
||||
#}
|
||||
{% extends "admin/base.html" %}
|
||||
|
||||
{% block title %}Too many attempts — Admin{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
<article class="page-article">
|
||||
<header class="page-article__header">
|
||||
<h1 class="page-article__title">Too many attempts</h1>
|
||||
</header>
|
||||
<p>
|
||||
You've requested too many login links recently. Please wait a
|
||||
few minutes before trying again.
|
||||
</p>
|
||||
<p>
|
||||
<a href="/admin/login">Back to login</a>
|
||||
</p>
|
||||
</article>
|
||||
{% endblock %}
|
||||
36
app/templates/emails/magic_link.html
Normal file
36
app/templates/emails/magic_link.html
Normal file
@@ -0,0 +1,36 @@
|
||||
{#
|
||||
HTML body for the magic-link email.
|
||||
|
||||
Plain, minimal inline-styled markup — no external CSS because email
|
||||
clients are hostile to it. The link text IS the URL so users can
|
||||
verify the destination before clicking.
|
||||
|
||||
Context:
|
||||
- display_name : str
|
||||
- magic_link_url : str
|
||||
- expires_at : ISO-8601 str
|
||||
- ttl_min : int
|
||||
#}<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<title>Admin login link — Chicken Babies R Us</title>
|
||||
</head>
|
||||
<body style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; color: #2B3A42; max-width: 560px; margin: 0 auto; padding: 24px;">
|
||||
<h1 style="font-size: 20px; margin: 0 0 16px;">Admin login link</h1>
|
||||
<p>Hi {{ display_name }},</p>
|
||||
<p>
|
||||
Use the link below to log in to the Chicken Babies R Us admin.
|
||||
The link works for {{ ttl_min }} minutes and can only be used once.
|
||||
</p>
|
||||
<p>
|
||||
<a href="{{ magic_link_url }}" style="color: #5D8AA8; word-break: break-all;">
|
||||
{{ magic_link_url }}
|
||||
</a>
|
||||
</p>
|
||||
<p style="color: #6b7a80; font-size: 13px;">
|
||||
Expires at {{ expires_at }}. If you didn't request this, you can
|
||||
safely ignore the email — no action will be taken.
|
||||
</p>
|
||||
</body>
|
||||
</html>
|
||||
12
app/templates/emails/magic_link.txt
Normal file
12
app/templates/emails/magic_link.txt
Normal file
@@ -0,0 +1,12 @@
|
||||
{# Plaintext body for the magic-link email. Mirrors the HTML version. #}
|
||||
Hi {{ display_name }},
|
||||
|
||||
Use the link below to log in to the Chicken Babies R Us admin.
|
||||
The link works for {{ ttl_min }} minutes and can only be used once.
|
||||
|
||||
{{ magic_link_url }}
|
||||
|
||||
Expires at {{ expires_at }}.
|
||||
|
||||
If you didn't request this, you can safely ignore the email -- no
|
||||
action will be taken.
|
||||
@@ -113,15 +113,49 @@ High-level phased plan. Each phase ends in a mergeable `dev` state and a passing
|
||||
- **No new packages.** All deps were already pinned in Phase 0's `requirements.txt`.
|
||||
- **Verification run:** `python -c "from app.main import app"` ✓ · `pytest -q` 36 passed ✓ · fresh-boot smoke: `/` shows welcome title, `/about` shows seeded Markdown, `/healthz` 200 ✓ · `PRAGMA journal_mode=wal` ✓ · second boot logs `migrations_up_to_date` + `seed_skipped`, table counts stay `users=1 pages=1 posts=1` ✓ · `docker compose config` exit 0 ✓.
|
||||
|
||||
## Phase 3 — Admin Auth (Magic Link)
|
||||
## Phase 3 — Admin Auth (Magic Link) ✅
|
||||
|
||||
- `/admin/login` — email-only form.
|
||||
- POST creates a magic-link token (256-bit, hashed at rest, 15-min TTL, single-use), sends via Resend.
|
||||
- Click link → create signed session cookie (30-day) → redirect `/admin`.
|
||||
- Logout revokes the session row (does not delete — audit trail).
|
||||
- Rate limits (SlowAPI): 5 / 15 min / IP *and* / email.
|
||||
- Allowlist from `ADMIN_EMAILS` env var. Non-allowlisted addresses: return same success message, send no email (no user enumeration).
|
||||
- Audit log row for every auth event.
|
||||
**Completed:** 2026-04-21
|
||||
|
||||
**Summary:** Passwordless admin auth end-to-end: email form → 256-bit magic-link token (SHA-256 at rest, 15-min TTL, atomic single-use consume) → Resend email with dev-log fallback → itsdangerous-signed server-side session cookie (30d) → `/admin` landing → logout revokes the row without deleting. SlowAPI per-IP + DB per-email rate limits, `ADMIN_EMAILS` allowlist with anti-enumeration, every event audited to `auth_events`.
|
||||
|
||||
**Key files:**
|
||||
- `app/services/audit.py` — `AuditService.record(event_type, ...)` writes an `auth_events` row + mirrors the event to structlog; `detail` is JSON. Referenced session by last-6 hash chars only.
|
||||
- `app/services/email.py` — `EmailService.send_magic_link(...)`: renders HTML + text templates, posts via Resend when `resend_api_key` set, otherwise logs `magic_link_dev_fallback` with the URL (dev shortcut). Never raises from the request path; production startup refuses to boot without the key.
|
||||
- `app/services/sessions.py` — `SessionService(engine, signer, settings)` — `create/lookup/revoke`. Raw session ID exists only in memory and the signed cookie; DB stores `sha256(raw)`. Cookie `cb_session`, `HttpOnly=True`, `SameSite=lax`, `Secure` ON in prod / OFF in dev, `Path=/`, `Max-Age=SESSION_MAX_DAYS*86400`.
|
||||
- `app/services/auth.py` — `AuthService.request_link` (allowlist check → DB per-email rate-limit count → insert token row → send email → audit) and `AuthService.consume` (atomic `UPDATE magic_link_tokens SET used_at WHERE token_hash AND used_at IS NULL AND expires_at>now`, then upsert `users` row, then `SessionService.create`). Audit writes live outside the consume transaction to sidestep SQLite's single-writer semantics.
|
||||
- `app/services/rate_limit.py` — module-level `limiter = Limiter(key_func=get_remote_address, storage_uri="memory://")`. Singleton because `@limiter.limit` decorates at import time.
|
||||
- `app/routes/admin.py` — all admin HTTP routes; `POST /admin/login` decorated `@limiter.limit("5/15 minutes")`, `GET /admin/auth/consume/{token}` decorated `@limiter.limit("20/15 minutes")`.
|
||||
- `app/dependencies/auth.py` — `get_current_user` and `require_admin` (raises `HTTPException(status_code=303, headers={"Location": "/admin/login"})`).
|
||||
- `app/templates/admin/` — `base.html`, `login.html`, `login_sent.html`, `login_failed.html`, `index.html`, `rate_limited.html`.
|
||||
- `app/templates/emails/magic_link.html` + `magic_link.txt` — magic-link email bodies.
|
||||
- `tests/test_auth_service.py`, `test_admin_routes.py`, `test_rate_limit.py`, `test_session_service.py`, `test_email_service.py` — 25 new tests covering token lifecycle, single-use, expiry, allowlist anti-enumeration, signed-cookie round-trip, revoke, user upsert, IP + DB email limits, dev email fallback, production config refusal.
|
||||
- `app/main.py` — builds `URLSafeTimedSerializer(salt="session")`, instantiates audit/email/sessions/auth onto `app.state`, installs SlowAPI `RateLimitExceeded` handler that renders `admin/rate_limited.html` with 429 and audits `rate_limited` with `scope="ip"`, includes admin router.
|
||||
- `app/config.py` — added `public_base_url` field (default `http://127.0.0.1:8000`); added `_require_auth_config_in_production` model validator: production boot refuses empty `RESEND_API_KEY` / `RESEND_FROM` / `ADMIN_EMAILS`.
|
||||
- `.env.example` — added `PUBLIC_BASE_URL=http://127.0.0.1:8000`.
|
||||
|
||||
**Endpoints created:**
|
||||
- `GET /admin/login` — email-only login form (public, no CSRF — pre-auth).
|
||||
- `POST /admin/login` — rate-limited (5/15min/IP + 5/15min/email via DB count). Always renders `login_sent.html` regardless of allowlist (anti-enumeration). Audit row `link_requested` with `{"allowlisted": bool}` every time.
|
||||
- `GET /admin/auth/consume/{token}` — rate-limited (20/15min/IP). Atomic single-use consume; on success sets `cb_session` cookie + 303 to `/admin`; on failure renders generic `login_failed.html` (no reason leakage).
|
||||
- `GET /admin` — `require_admin`; renders placeholder `index.html` with `{{ user.display_name }}` + logout form. Phase 4 replaces with real CMS dashboard.
|
||||
- `POST /admin/logout` — `require_admin`; flips `sessions.revoked_at = now`, clears cookie with `Max-Age=0`, 303 to `/admin/login`. Row preserved for audit. Marked `# TODO(phase-6-csrf)` — SameSite=Lax blocks cross-site POSTs in current browsers; Phase 6 adds a double-submit token.
|
||||
|
||||
**Key details:**
|
||||
- **Hash-at-rest for both tokens and session IDs.** DB stores `sha256(raw).hexdigest()`; raw values never persisted, never logged. Audit detail only references last-6 hash chars for correlation.
|
||||
- **Cookie signed with itsdangerous `URLSafeTimedSerializer(secret_key, salt="session")`.** Forged cookie fails signature check before any DB lookup.
|
||||
- **Anti-enumeration verified:** non-allowlisted POST → same 200 HTML response, zero rows inserted into `magic_link_tokens`, `link_requested` audit logs `{"allowlisted": false}`.
|
||||
- **Rate limit verified:** 5 POSTs succeed + 6th returns 429 from same IP; DB per-email count applied even when SlowAPI would allow.
|
||||
- **User auto-upsert on consume:** first successful consume inserts a `users` row with `display_name = local-part.title()` (e.g. `driver@example.com` → "Driver"), `active=1`; subsequent logins update `last_login_at`.
|
||||
- **Dev Resend fallback:** when `resend_api_key` is falsy, `EmailService` emits `magic_link_dev_fallback` structured log with the full URL and returns cleanly — never 500s the request path.
|
||||
- **Production config guardrails:** `app_env == "production"` requires non-empty `RESEND_API_KEY`, `RESEND_FROM`, `ADMIN_EMAILS`. Missing any → `ValueError` at startup.
|
||||
- **Single-writer deadlock avoided:** `AuthService.consume` does the atomic token update inside `engine.begin()`, then captures outcome flags, then opens separate transactions for audit + session creation. Correctness unchanged (single-use guarantee via `rowcount==1`).
|
||||
- **Audit trail smoke-tested rows:** `link_requested` (per attempt), `link_consumed`, `session_created`, `session_revoked`, `rate_limited`, `consume_failed` — all emitted during driver verification.
|
||||
- **Phase 4 hooks ready:** `require_admin` dependency + `get_current_user` are callable from any admin route. Admin CMS POSTs will reuse them.
|
||||
- **Phase 6 TODO markers** on the logout handler flag the CSRF pickup point.
|
||||
|
||||
**Verification run:**
|
||||
`python -c "from app.main import app"` ✓ · `pytest -q` 61 passed ✓ · end-to-end smoke (fresh DB, driver allowlisted): login form 200, POST login 200, dev-log URL captured, consume 200 + cookie set, `/admin` 200 with "Welcome, Driver", logout 303, post-logout `/admin` 303 to login ✓ · IP rate limit fires at 5 (6th is 429) ✓ · non-allowlisted POST 200 + zero tokens in DB ✓ · `auth_events` shows `link_requested`/`link_consumed`/`session_created`/`session_revoked`/`rate_limited` rows as expected ✓ · `docker compose config` exit 0 ✓.
|
||||
|
||||
## Phase 4 — Admin CMS
|
||||
|
||||
|
||||
209
tests/test_admin_routes.py
Normal file
209
tests/test_admin_routes.py
Normal file
@@ -0,0 +1,209 @@
|
||||
"""End-to-end HTTP tests for the admin auth flow.
|
||||
|
||||
Exercises the real :class:`FastAPI` app constructed by
|
||||
:func:`app.main.create_app` against a temp-file SQLite database and a
|
||||
monkeypatched ``EmailService`` so we can capture the magic-link URL
|
||||
without depending on Resend.
|
||||
|
||||
These tests intentionally do NOT use the session-scoped ``db_engine``
|
||||
fixture — each builds its own fresh app so rate-limit state and DB
|
||||
rows don't leak between tests.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Iterator
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import text
|
||||
|
||||
|
||||
# Regex used to extract the raw token from the dev-fallback log URL.
|
||||
_URL_RE = re.compile(r"http[s]?://[^\s]+/admin/auth/consume/[A-Za-z0-9_-]+")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_app(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> Iterator[tuple[TestClient, dict]]:
|
||||
"""Build a fresh FastAPI app wired to a tmp DB + captured email URL."""
|
||||
# Point the settings loader at a clean DB and a known allowlist.
|
||||
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path}/admin.db")
|
||||
monkeypatch.setenv("ADMIN_EMAILS", "headhen@example.com")
|
||||
monkeypatch.setenv("APP_ENV", "development")
|
||||
monkeypatch.setenv(
|
||||
"SECRET_KEY", "test-only-secret-key-0123456789abcdef-XYZ"
|
||||
)
|
||||
monkeypatch.setenv("RESEND_API_KEY", "")
|
||||
monkeypatch.setenv("PUBLIC_BASE_URL", "http://testserver")
|
||||
|
||||
# Reload config + main so the new env is picked up.
|
||||
from app import config as _config
|
||||
|
||||
_config.get_settings.cache_clear()
|
||||
|
||||
import app.main as main_module
|
||||
|
||||
importlib.reload(main_module)
|
||||
app = main_module.app
|
||||
|
||||
# Capture the magic-link URL instead of sending via Resend.
|
||||
captured: dict[str, str] = {}
|
||||
|
||||
def _capture(**kw):
|
||||
captured["url"] = kw["url"]
|
||||
captured["to"] = kw["to"]
|
||||
|
||||
app.state.email_service.send_magic_link = _capture # type: ignore[assignment]
|
||||
|
||||
# Reset the in-memory rate limiter between tests (module-level
|
||||
# singleton would otherwise leak state across tests).
|
||||
from app.services.rate_limit import limiter
|
||||
|
||||
limiter.reset()
|
||||
|
||||
with TestClient(app) as client:
|
||||
yield client, captured
|
||||
|
||||
_config.get_settings.cache_clear()
|
||||
|
||||
|
||||
def _auth_events(app, event_type: str | None = None) -> list[dict]:
|
||||
"""Return rows from ``auth_events`` (optionally filtered by type)."""
|
||||
with app.state.engine.connect() as conn:
|
||||
if event_type:
|
||||
rows = conn.execute(
|
||||
text(
|
||||
"SELECT event_type, email, detail FROM auth_events"
|
||||
" WHERE event_type = :t ORDER BY id"
|
||||
),
|
||||
{"t": event_type},
|
||||
).mappings().all()
|
||||
else:
|
||||
rows = conn.execute(
|
||||
text(
|
||||
"SELECT event_type, email, detail FROM auth_events"
|
||||
" ORDER BY id"
|
||||
)
|
||||
).mappings().all()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def test_full_login_flow(admin_app) -> None:
|
||||
"""GET login, POST, consume, GET /admin, POST logout."""
|
||||
client, captured = admin_app
|
||||
|
||||
# 1. GET login page.
|
||||
resp = client.get("/admin/login")
|
||||
assert resp.status_code == 200
|
||||
assert "Admin log in" in resp.text
|
||||
assert 'name="email"' in resp.text
|
||||
|
||||
# 2. POST login with allowlisted email.
|
||||
resp = client.post("/admin/login", data={"email": "headhen@example.com"})
|
||||
assert resp.status_code == 200
|
||||
assert "Check your inbox" in resp.text
|
||||
assert "url" in captured
|
||||
raw_url = captured["url"]
|
||||
assert "/admin/auth/consume/" in raw_url
|
||||
|
||||
# 3. Consume the magic link.
|
||||
consume_path = raw_url.replace("http://testserver", "")
|
||||
resp = client.get(consume_path, follow_redirects=False)
|
||||
assert resp.status_code == 303
|
||||
assert resp.headers["location"] == "/admin"
|
||||
# cb_session cookie set.
|
||||
assert "cb_session" in resp.cookies
|
||||
|
||||
# 4. GET /admin renders welcome.
|
||||
resp = client.get("/admin")
|
||||
assert resp.status_code == 200
|
||||
assert "Welcome" in resp.text
|
||||
assert "headhen@example.com" in resp.text
|
||||
|
||||
# 5. POST /admin/logout clears cookie + redirects.
|
||||
resp = client.post("/admin/logout", follow_redirects=False)
|
||||
assert resp.status_code == 303
|
||||
assert resp.headers["location"] == "/admin/login"
|
||||
|
||||
# 6. Subsequent /admin redirects back to login.
|
||||
# Manually drop the cookie from the test client, mirroring a browser
|
||||
# that honored the delete_cookie directive.
|
||||
client.cookies.clear()
|
||||
resp = client.get("/admin", follow_redirects=False)
|
||||
assert resp.status_code == 303
|
||||
assert resp.headers["location"] == "/admin/login"
|
||||
|
||||
# Audit rows: link_requested, link_consumed, session_created, session_revoked.
|
||||
events = [e["event_type"] for e in _auth_events(client.app)]
|
||||
for required in (
|
||||
"link_requested",
|
||||
"link_consumed",
|
||||
"session_created",
|
||||
"session_revoked",
|
||||
):
|
||||
assert required in events, f"missing audit event: {required}"
|
||||
|
||||
|
||||
def test_non_allowlisted_returns_same_page_no_token(admin_app) -> None:
|
||||
"""Non-allowlisted email: same 200 + 'Check your inbox', no token row."""
|
||||
client, captured = admin_app
|
||||
resp = client.post("/admin/login", data={"email": "stranger@example.com"})
|
||||
assert resp.status_code == 200
|
||||
assert "Check your inbox" in resp.text
|
||||
assert "url" not in captured # email not sent
|
||||
|
||||
# No token row for this email.
|
||||
with client.app.state.engine.connect() as conn:
|
||||
count = conn.execute(
|
||||
text(
|
||||
"SELECT COUNT(*) AS c FROM magic_link_tokens"
|
||||
" WHERE email = :e"
|
||||
),
|
||||
{"e": "stranger@example.com"},
|
||||
).mappings().first()
|
||||
assert count is not None and int(count["c"]) == 0
|
||||
|
||||
# Audit trail still records the attempt.
|
||||
events = _auth_events(client.app, "link_requested")
|
||||
assert len(events) == 1
|
||||
assert "\"allowlisted\": false" in events[0]["detail"]
|
||||
|
||||
|
||||
def test_invalid_token_returns_400(admin_app) -> None:
|
||||
"""A random token on the consume endpoint returns the generic 400 page."""
|
||||
client, _ = admin_app
|
||||
resp = client.get("/admin/auth/consume/not-a-real-token", follow_redirects=False)
|
||||
assert resp.status_code == 400
|
||||
assert "Login link invalid" in resp.text
|
||||
|
||||
|
||||
def test_login_form_validation(admin_app) -> None:
|
||||
"""Empty / malformed email surfaces the inline validation message."""
|
||||
client, _ = admin_app
|
||||
resp = client.post("/admin/login", data={"email": ""})
|
||||
assert resp.status_code == 400
|
||||
assert "valid email" in resp.text
|
||||
|
||||
resp = client.post("/admin/login", data={"email": "not-an-email"})
|
||||
assert resp.status_code == 400
|
||||
assert "valid email" in resp.text
|
||||
|
||||
|
||||
def test_replay_attack_fails(admin_app) -> None:
|
||||
"""A consumed magic link cannot be used a second time."""
|
||||
client, captured = admin_app
|
||||
client.post("/admin/login", data={"email": "headhen@example.com"})
|
||||
url = captured["url"].replace("http://testserver", "")
|
||||
|
||||
first = client.get(url, follow_redirects=False)
|
||||
assert first.status_code == 303
|
||||
|
||||
# Clear cookies to simulate a replay from a fresh browser context.
|
||||
client.cookies.clear()
|
||||
second = client.get(url, follow_redirects=False)
|
||||
assert second.status_code == 400
|
||||
275
tests/test_auth_service.py
Normal file
275
tests/test_auth_service.py
Normal file
@@ -0,0 +1,275 @@
|
||||
"""Tests for :class:`app.services.auth.AuthService`.
|
||||
|
||||
Covers:
|
||||
- Allowlisted email → token inserted, email "sent" (dev fallback), audit
|
||||
rows emitted.
|
||||
- Non-allowlisted email → no token row, audit row with allowlisted=False.
|
||||
- Consume happy path (creates user on first login, bumps last_login_at
|
||||
on subsequent logins, issues session).
|
||||
- Consume replay → second attempt returns None + consume_failed/used audit.
|
||||
- Consume expired token → None + consume_failed/expired.
|
||||
- Per-email rate-limit threshold trips on the 6th request.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from itsdangerous import URLSafeTimedSerializer
|
||||
from pathlib import Path
|
||||
from sqlalchemy import Engine, text
|
||||
|
||||
from app.config import Settings
|
||||
from app.services.audit import AuditService
|
||||
from app.services.auth import AuthService, RateLimitedError
|
||||
from app.services.email import EmailService
|
||||
from app.services.sessions import SessionService
|
||||
|
||||
|
||||
_TEMPLATES_DIR = Path(__file__).resolve().parent.parent / "app" / "templates"
|
||||
|
||||
|
||||
def _build(engine: Engine, *, admin_emails: str = "headhen@example.com") -> tuple[AuthService, Settings]:
|
||||
"""Wire up an AuthService against a real SQLite engine."""
|
||||
settings = Settings(
|
||||
app_env="development",
|
||||
secret_key="a-very-long-test-secret-key-1234567890",
|
||||
admin_emails=admin_emails,
|
||||
resend_api_key=None,
|
||||
resend_from=None,
|
||||
public_base_url="http://localhost:8000",
|
||||
magic_link_ttl_min=15,
|
||||
) # type: ignore[call-arg]
|
||||
templates = Jinja2Templates(directory=_TEMPLATES_DIR)
|
||||
signer = URLSafeTimedSerializer(settings.secret_key, salt="session")
|
||||
email = EmailService(settings, templates)
|
||||
audit = AuditService(engine)
|
||||
sessions = SessionService(engine, signer, settings)
|
||||
auth = AuthService(engine, email, sessions, audit, settings)
|
||||
return auth, settings
|
||||
|
||||
|
||||
def _count(engine: Engine, table: str, where: str = "1=1", **params) -> int:
|
||||
with engine.connect() as conn:
|
||||
row = conn.execute(
|
||||
text(f"SELECT COUNT(*) AS c FROM {table} WHERE {where}"), params
|
||||
).mappings().first()
|
||||
return int(row["c"]) if row is not None else 0
|
||||
|
||||
|
||||
def _latest_detail(engine: Engine, event_type: str) -> str:
|
||||
with engine.connect() as conn:
|
||||
row = conn.execute(
|
||||
text(
|
||||
"SELECT detail FROM auth_events WHERE event_type = :t"
|
||||
" ORDER BY id DESC LIMIT 1"
|
||||
),
|
||||
{"t": event_type},
|
||||
).mappings().first()
|
||||
return str(row["detail"]) if row is not None else ""
|
||||
|
||||
|
||||
def test_allowlisted_request_inserts_token_and_audit(
|
||||
clean_db_engine: Engine,
|
||||
) -> None:
|
||||
"""An allowlisted email mints a token row and a link_requested audit."""
|
||||
auth, _ = _build(clean_db_engine)
|
||||
auth.request_link(email="HeadHen@example.com", ip="1.2.3.4", user_agent="ua")
|
||||
|
||||
assert _count(clean_db_engine, "magic_link_tokens") == 1
|
||||
# Stored email is lowercased.
|
||||
with clean_db_engine.connect() as conn:
|
||||
row = conn.execute(
|
||||
text("SELECT email, token_hash FROM magic_link_tokens")
|
||||
).mappings().first()
|
||||
assert row["email"] == "headhen@example.com"
|
||||
# Hex SHA-256 is 64 chars.
|
||||
assert len(row["token_hash"]) == 64
|
||||
|
||||
assert _count(clean_db_engine, "auth_events", "event_type='link_requested'") == 1
|
||||
assert "\"allowlisted\": true" in _latest_detail(
|
||||
clean_db_engine, "link_requested"
|
||||
)
|
||||
|
||||
|
||||
def test_non_allowlisted_request_inserts_no_token(
|
||||
clean_db_engine: Engine,
|
||||
) -> None:
|
||||
"""A non-allowlisted email gets an audit row and no token."""
|
||||
auth, _ = _build(clean_db_engine, admin_emails="headhen@example.com")
|
||||
auth.request_link(email="intruder@example.com", ip="1.2.3.4", user_agent="ua")
|
||||
|
||||
assert _count(clean_db_engine, "magic_link_tokens") == 0
|
||||
assert _count(clean_db_engine, "auth_events", "event_type='link_requested'") == 1
|
||||
assert "\"allowlisted\": false" in _latest_detail(
|
||||
clean_db_engine, "link_requested"
|
||||
)
|
||||
|
||||
|
||||
def test_consume_happy_path_creates_user_and_session(
|
||||
clean_db_engine: Engine,
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
"""Consuming a fresh token upserts a user and mints a session."""
|
||||
auth, _ = _build(clean_db_engine)
|
||||
|
||||
# Intercept the raw token via the dev-fallback log. Easiest path:
|
||||
# patch EmailService.send_magic_link on this instance.
|
||||
captured: dict[str, str] = {}
|
||||
|
||||
def _capture(**kw):
|
||||
captured["url"] = kw["url"]
|
||||
|
||||
monkeypatch.setattr(auth._email, "send_magic_link", _capture)
|
||||
|
||||
auth.request_link(email="headhen@example.com", ip="1.2.3.4", user_agent="ua")
|
||||
assert "url" in captured
|
||||
raw_token = captured["url"].rsplit("/", 1)[-1]
|
||||
|
||||
result = auth.consume(
|
||||
raw_token=raw_token, ip="1.2.3.4", user_agent="ua"
|
||||
)
|
||||
assert result is not None
|
||||
user, session, cookie = result
|
||||
|
||||
assert user.email == "headhen@example.com"
|
||||
assert user.display_name == "Headhen"
|
||||
assert user.active is True
|
||||
|
||||
# User row persisted with last_login_at set.
|
||||
with clean_db_engine.connect() as conn:
|
||||
row = conn.execute(
|
||||
text("SELECT last_login_at FROM users WHERE id = :id"),
|
||||
{"id": user.id},
|
||||
).mappings().first()
|
||||
assert row["last_login_at"] is not None
|
||||
|
||||
# Token marked used.
|
||||
with clean_db_engine.connect() as conn:
|
||||
tok = conn.execute(
|
||||
text("SELECT used_at FROM magic_link_tokens")
|
||||
).mappings().first()
|
||||
assert tok["used_at"] is not None
|
||||
|
||||
# Session row exists, cookie is non-empty.
|
||||
assert session.user_id == user.id
|
||||
assert cookie
|
||||
|
||||
# Audit trail: link_consumed + session_created.
|
||||
assert _count(
|
||||
clean_db_engine, "auth_events", "event_type='link_consumed'"
|
||||
) == 1
|
||||
assert _count(
|
||||
clean_db_engine, "auth_events", "event_type='session_created'"
|
||||
) == 1
|
||||
|
||||
|
||||
def test_consume_replay_fails(clean_db_engine: Engine, monkeypatch) -> None:
|
||||
"""The same raw token cannot be consumed twice."""
|
||||
auth, _ = _build(clean_db_engine)
|
||||
captured: dict[str, str] = {}
|
||||
monkeypatch.setattr(
|
||||
auth._email, "send_magic_link", lambda **kw: captured.update(url=kw["url"])
|
||||
)
|
||||
auth.request_link(email="headhen@example.com", ip="", user_agent="")
|
||||
raw = captured["url"].rsplit("/", 1)[-1]
|
||||
|
||||
first = auth.consume(raw_token=raw, ip="", user_agent="")
|
||||
assert first is not None
|
||||
|
||||
second = auth.consume(raw_token=raw, ip="", user_agent="")
|
||||
assert second is None
|
||||
|
||||
# consume_failed audit with reason=used.
|
||||
assert "\"reason\": \"used\"" in _latest_detail(
|
||||
clean_db_engine, "consume_failed"
|
||||
)
|
||||
|
||||
|
||||
def test_consume_expired_token_fails(clean_db_engine: Engine, monkeypatch) -> None:
|
||||
"""A token past its expires_at cannot be consumed."""
|
||||
auth, _ = _build(clean_db_engine)
|
||||
captured: dict[str, str] = {}
|
||||
monkeypatch.setattr(
|
||||
auth._email, "send_magic_link", lambda **kw: captured.update(url=kw["url"])
|
||||
)
|
||||
auth.request_link(email="headhen@example.com", ip="", user_agent="")
|
||||
raw = captured["url"].rsplit("/", 1)[-1]
|
||||
|
||||
# Push expires_at into the past.
|
||||
past = (datetime.now(timezone.utc) - timedelta(minutes=30)).isoformat()
|
||||
with clean_db_engine.begin() as conn:
|
||||
conn.execute(
|
||||
text("UPDATE magic_link_tokens SET expires_at = :p"), {"p": past}
|
||||
)
|
||||
|
||||
assert auth.consume(raw_token=raw, ip="", user_agent="") is None
|
||||
assert "\"reason\": \"expired\"" in _latest_detail(
|
||||
clean_db_engine, "consume_failed"
|
||||
)
|
||||
|
||||
|
||||
def test_consume_unknown_token_fails(clean_db_engine: Engine) -> None:
|
||||
"""Random tokens result in not_found."""
|
||||
auth, _ = _build(clean_db_engine)
|
||||
assert auth.consume(
|
||||
raw_token="totally-random-nope", ip="", user_agent=""
|
||||
) is None
|
||||
assert "\"reason\": \"not_found\"" in _latest_detail(
|
||||
clean_db_engine, "consume_failed"
|
||||
)
|
||||
|
||||
|
||||
def test_per_email_rate_limit_trips_on_sixth(
|
||||
clean_db_engine: Engine, monkeypatch
|
||||
) -> None:
|
||||
"""Five requests in the window succeed; the sixth raises and audits."""
|
||||
auth, _ = _build(clean_db_engine)
|
||||
monkeypatch.setattr(auth._email, "send_magic_link", lambda **kw: None)
|
||||
|
||||
for _ in range(5):
|
||||
auth.request_link(email="headhen@example.com", ip="", user_agent="")
|
||||
|
||||
with pytest.raises(RateLimitedError):
|
||||
auth.request_link(email="headhen@example.com", ip="", user_agent="")
|
||||
|
||||
# Exactly 5 tokens persisted; 6th blocked before insert.
|
||||
assert _count(clean_db_engine, "magic_link_tokens") == 5
|
||||
# rate_limited audit with scope=email.
|
||||
detail = _latest_detail(clean_db_engine, "rate_limited")
|
||||
assert "\"scope\": \"email\"" in detail
|
||||
|
||||
|
||||
def test_existing_user_reactivated_on_consume(
|
||||
clean_db_engine: Engine, monkeypatch
|
||||
) -> None:
|
||||
"""Consume updates an existing (possibly deactivated) user row."""
|
||||
# Seed a deactivated existing admin.
|
||||
now_iso = datetime.now(timezone.utc).isoformat()
|
||||
with clean_db_engine.begin() as conn:
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO users"
|
||||
" (email, display_name, created_at, last_login_at, active)"
|
||||
" VALUES (:e, :d, :c, NULL, 0)"
|
||||
),
|
||||
{"e": "headhen@example.com", "d": "Old Name", "c": now_iso},
|
||||
)
|
||||
|
||||
auth, _ = _build(clean_db_engine)
|
||||
captured: dict[str, str] = {}
|
||||
monkeypatch.setattr(
|
||||
auth._email, "send_magic_link", lambda **kw: captured.update(url=kw["url"])
|
||||
)
|
||||
auth.request_link(email="headhen@example.com", ip="", user_agent="")
|
||||
raw = captured["url"].rsplit("/", 1)[-1]
|
||||
|
||||
result = auth.consume(raw_token=raw, ip="", user_agent="")
|
||||
assert result is not None
|
||||
user, _, _ = result
|
||||
|
||||
# display_name preserved (we don't overwrite), active flipped back on.
|
||||
assert user.display_name == "Old Name"
|
||||
assert user.active is True
|
||||
113
tests/test_email_service.py
Normal file
113
tests/test_email_service.py
Normal file
@@ -0,0 +1,113 @@
|
||||
"""Tests for :class:`app.services.email.EmailService`.
|
||||
|
||||
We exercise the dev-fallback path (no ``RESEND_API_KEY``) and assert:
|
||||
|
||||
- It does NOT raise.
|
||||
- It emits a structured log event ``magic_link_dev_fallback`` with the
|
||||
full URL so a developer can copy it.
|
||||
- It doesn't call the real Resend SDK.
|
||||
|
||||
Production-missing-key is validated at startup by the pydantic model
|
||||
validator (covered separately in the config tests — but we also verify
|
||||
the behavior here via a direct settings construction).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.templating import Jinja2Templates
|
||||
|
||||
from app.config import Settings
|
||||
from app.logging_config import configure_logging
|
||||
from app.services.email import EmailService
|
||||
|
||||
|
||||
_PACKAGE_ROOT = Path(__file__).resolve().parent.parent / "app"
|
||||
_TEMPLATES_DIR = _PACKAGE_ROOT / "templates"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def templates() -> Jinja2Templates:
|
||||
"""Return a Jinja2Templates pointing at the real app templates dir."""
|
||||
return Jinja2Templates(directory=_TEMPLATES_DIR)
|
||||
|
||||
|
||||
def test_dev_fallback_logs_url(
|
||||
templates: Jinja2Templates, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
"""Missing RESEND_API_KEY in dev -> structured log, no exception.
|
||||
|
||||
structlog writes via ``PrintLoggerFactory`` (stdout), so we capture
|
||||
stdout rather than the stdlib ``caplog`` handler.
|
||||
"""
|
||||
# Reconfigure structlog so the ConsoleRenderer is active — the test
|
||||
# harness may have left it in another state.
|
||||
configure_logging("development")
|
||||
|
||||
settings = Settings(
|
||||
app_env="development",
|
||||
resend_api_key=None,
|
||||
resend_from=None,
|
||||
) # type: ignore[call-arg]
|
||||
svc = EmailService(settings, templates)
|
||||
|
||||
url = "http://localhost:8000/admin/auth/consume/raw-token-123"
|
||||
svc.send_magic_link(
|
||||
to="head-hen@example.com",
|
||||
url=url,
|
||||
display_name="Head Hen",
|
||||
ttl_min=15,
|
||||
expires_at=datetime.now(timezone.utc) + timedelta(minutes=15),
|
||||
)
|
||||
|
||||
captured = capsys.readouterr()
|
||||
combined = captured.out + captured.err
|
||||
assert "magic_link_dev_fallback" in combined
|
||||
assert url in combined
|
||||
|
||||
|
||||
def test_production_missing_key_refuses_startup() -> None:
|
||||
"""Production without resend key / from / admin_emails raises at init."""
|
||||
with pytest.raises(ValueError) as excinfo:
|
||||
Settings(
|
||||
app_env="production",
|
||||
secret_key="long-enough-prod-secret-key-value-000000",
|
||||
) # type: ignore[call-arg]
|
||||
msg = str(excinfo.value)
|
||||
assert "RESEND_API_KEY" in msg
|
||||
assert "RESEND_FROM" in msg
|
||||
assert "ADMIN_EMAILS" in msg
|
||||
|
||||
|
||||
def test_dev_fallback_does_not_touch_resend(
|
||||
templates: Jinja2Templates, monkeypatch
|
||||
) -> None:
|
||||
"""If the SDK IS present, the dev fallback must still not call it."""
|
||||
import resend # type: ignore[import-untyped]
|
||||
|
||||
called: dict[str, bool] = {"ok": False}
|
||||
|
||||
def _boom(*a, **kw): # pragma: no cover — shouldn't execute
|
||||
called["ok"] = True
|
||||
raise RuntimeError("should not be called in dev fallback")
|
||||
|
||||
monkeypatch.setattr(resend.Emails, "send", staticmethod(_boom))
|
||||
|
||||
settings = Settings(
|
||||
app_env="development",
|
||||
resend_api_key=None,
|
||||
resend_from=None,
|
||||
) # type: ignore[call-arg]
|
||||
svc = EmailService(settings, templates)
|
||||
|
||||
svc.send_magic_link(
|
||||
to="a@b.co",
|
||||
url="http://x/y",
|
||||
display_name="A",
|
||||
ttl_min=15,
|
||||
expires_at=datetime.now(timezone.utc),
|
||||
)
|
||||
assert called["ok"] is False
|
||||
142
tests/test_rate_limit.py
Normal file
142
tests/test_rate_limit.py
Normal file
@@ -0,0 +1,142 @@
|
||||
"""Rate-limit tests for the admin auth flow.
|
||||
|
||||
Covers:
|
||||
- IP-level SlowAPI limit: 6th POST /admin/login from the same IP within
|
||||
15 minutes returns 429 + ``rate_limited.html`` body.
|
||||
- Per-email DB limit: 6th POST /admin/login for the same email (even
|
||||
from different IPs — which TestClient can't really simulate without
|
||||
middleware tricks, so we directly insert 5 recent tokens to prime
|
||||
the DB) returns 429.
|
||||
- Every 429 path writes a ``rate_limited`` audit event.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Iterator
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import text
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin_app(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> Iterator[tuple[TestClient, dict]]:
|
||||
"""Build a fresh FastAPI app wired to a tmp DB + captured email URL."""
|
||||
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path}/admin.db")
|
||||
monkeypatch.setenv("ADMIN_EMAILS", "headhen@example.com")
|
||||
monkeypatch.setenv("APP_ENV", "development")
|
||||
monkeypatch.setenv(
|
||||
"SECRET_KEY", "test-only-secret-key-0123456789abcdef-XYZ"
|
||||
)
|
||||
monkeypatch.setenv("RESEND_API_KEY", "")
|
||||
monkeypatch.setenv("PUBLIC_BASE_URL", "http://testserver")
|
||||
|
||||
from app import config as _config
|
||||
|
||||
_config.get_settings.cache_clear()
|
||||
|
||||
import app.main as main_module
|
||||
|
||||
importlib.reload(main_module)
|
||||
app = main_module.app
|
||||
|
||||
captured: dict = {"urls": []}
|
||||
app.state.email_service.send_magic_link = lambda **kw: captured["urls"].append( # type: ignore[assignment]
|
||||
kw["url"]
|
||||
)
|
||||
|
||||
from app.services.rate_limit import limiter
|
||||
|
||||
limiter.reset()
|
||||
|
||||
with TestClient(app) as client:
|
||||
yield client, captured
|
||||
|
||||
_config.get_settings.cache_clear()
|
||||
|
||||
|
||||
def test_ip_rate_limit_trips_on_sixth(admin_app) -> None:
|
||||
"""Five POSTs succeed; the sixth from the same IP returns 429."""
|
||||
client, _ = admin_app
|
||||
|
||||
# Use a non-allowlisted email so we don't bump the DB per-email
|
||||
# limit at the same time — isolates the IP-level limit.
|
||||
for i in range(5):
|
||||
resp = client.post(
|
||||
"/admin/login", data={"email": f"nobody{i}@example.com"}
|
||||
)
|
||||
assert resp.status_code == 200, (i, resp.text[:200])
|
||||
|
||||
resp = client.post("/admin/login", data={"email": "nobody5@example.com"})
|
||||
assert resp.status_code == 429
|
||||
assert "Too many attempts" in resp.text
|
||||
|
||||
# Audit row written.
|
||||
with client.app.state.engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
text(
|
||||
"SELECT detail FROM auth_events"
|
||||
" WHERE event_type = 'rate_limited'"
|
||||
)
|
||||
).mappings().all()
|
||||
assert len(rows) >= 1
|
||||
assert any("\"scope\": \"ip\"" in str(r["detail"]) for r in rows)
|
||||
|
||||
|
||||
def test_per_email_rate_limit_trips_on_sixth(
|
||||
admin_app, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
"""Priming 5 recent tokens for an email causes the 6th request to 429."""
|
||||
client, _ = admin_app
|
||||
|
||||
# Disable the SlowAPI IP limiter temporarily so we can isolate the
|
||||
# DB-side per-email check. We can't call six requests through the
|
||||
# IP limiter within 15 minutes — the IP limit would trip first.
|
||||
from app.services.rate_limit import limiter
|
||||
|
||||
limiter.enabled = False
|
||||
try:
|
||||
# Seed 5 recent tokens for the allowlisted email.
|
||||
now = datetime.now(timezone.utc)
|
||||
with client.app.state.engine.begin() as conn:
|
||||
for i in range(5):
|
||||
conn.execute(
|
||||
text(
|
||||
"INSERT INTO magic_link_tokens"
|
||||
" (email, token_hash, created_at, expires_at,"
|
||||
" request_ip)"
|
||||
" VALUES (:e, :h, :c, :x, :ip)"
|
||||
),
|
||||
{
|
||||
"e": "headhen@example.com",
|
||||
"h": f"seed-hash-{i}-" + ("x" * 50),
|
||||
"c": now.isoformat(),
|
||||
"x": (now + timedelta(minutes=15)).isoformat(),
|
||||
"ip": "",
|
||||
},
|
||||
)
|
||||
|
||||
# 6th request trips the DB-side limiter.
|
||||
resp = client.post(
|
||||
"/admin/login", data={"email": "headhen@example.com"}
|
||||
)
|
||||
assert resp.status_code == 429
|
||||
assert "Too many attempts" in resp.text
|
||||
|
||||
# rate_limited audit row with scope=email.
|
||||
with client.app.state.engine.connect() as conn:
|
||||
rows = conn.execute(
|
||||
text(
|
||||
"SELECT detail FROM auth_events"
|
||||
" WHERE event_type = 'rate_limited'"
|
||||
)
|
||||
).mappings().all()
|
||||
assert any("\"scope\": \"email\"" in str(r["detail"]) for r in rows)
|
||||
finally:
|
||||
limiter.enabled = True
|
||||
limiter.reset()
|
||||
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