Merge branch 'chore/phase-4-admin-cms' into dev: Phase 4 complete

This commit is contained in:
2026-04-21 20:42:06 -05:00
30 changed files with 3831 additions and 74 deletions

View File

@@ -20,6 +20,11 @@ SECRET_KEY=dev-insecure-change-me
# --- Database ---------------------------------------------------------------
DATABASE_URL=sqlite:///data/app.db
# --- Media storage ----------------------------------------------------------
# Filesystem directory holding admin-uploaded images. Mounted publicly at
# /media by the app. Relative paths resolve against the process cwd.
MEDIA_ROOT=data/media
# --- Email (Resend) ---------------------------------------------------------
RESEND_API_KEY=
RESEND_FROM=no-reply@chickenbabies.example

5
.gitignore vendored
View File

@@ -26,11 +26,14 @@ env/
*.db-shm
# Runtime data (DB + media uploads live here in prod).
# Ignore everything inside data/ except the .gitkeep marker; negating a
# Ignore everything inside data/ except the .gitkeep markers; negating a
# path whose *directory* is ignored does not work, so we ignore the
# contents with a trailing glob instead.
data/*
!data/.gitkeep
!data/media/
data/media/*
!data/media/.gitkeep
# Editors / IDE
.vscode/

View File

@@ -64,6 +64,18 @@ class Settings(BaseSettings):
description="SQLAlchemy URL for the application database.",
)
# --- Media storage -----------------------------------------------------
# Filesystem directory holding admin-uploaded images. Mounted
# publicly at ``/media`` by the FastAPI factory, so the Markdown
# URL inserted after a drag-drop upload matches what the public
# site can fetch. Relative paths resolve against the process cwd
# (uvicorn runs from the repo root in dev; Docker's WORKDIR in
# prod), matching the DATABASE_URL convention.
media_root: str = Field(
default="data/media",
description="Filesystem directory where admin-uploaded media is stored.",
)
# --- Email (Resend) ----------------------------------------------------
# Optional at Phase 0: contact form and magic-link delivery come online
# in Phases 3 and 5 respectively.

65
app/dependencies/csrf.py Normal file
View File

@@ -0,0 +1,65 @@
"""FastAPI dependency for CSRF verification on admin mutating endpoints.
Mount on any route that performs a state change (POST / PUT / DELETE)
and sits inside the admin router. GET admin routes do NOT need this —
they only need ``require_admin`` to gate access.
Two flavors, same underlying check:
- :func:`require_csrf_form` — reads ``csrf_token`` from the form body.
Preferred for classic HTML forms.
- :func:`require_csrf_header` — reads ``X-CSRF-Token`` from the request
headers. Used by the live-preview fetch and the drag-drop upload
endpoint where sending an extra form field is awkward.
Both raise HTTP 403 on mismatch, which surfaces as the generic FastAPI
error page. No information leaks about which side (cookie or submitted
token) failed — fail-closed is uniform.
"""
from __future__ import annotations
from typing import Optional
from fastapi import Form, HTTPException, Request
from app.services.csrf import CSRF_COOKIE_NAME, CSRFService
def _get_csrf_service(request: Request) -> CSRFService:
"""Pull the app-scoped :class:`CSRFService` off ``request.app.state``.
Private helper: route handlers depend on :func:`require_csrf_form`
or :func:`require_csrf_header` directly, not on this lookup.
"""
return request.app.state.csrf_service
def require_csrf_form(
request: Request,
csrf_token: str = Form(default=""),
) -> None:
"""Verify a form-submitted CSRF token matches the signed cookie.
Raises :class:`fastapi.HTTPException` 403 on any mismatch. On
success returns ``None`` — the dependency has no payload.
"""
service: CSRFService = _get_csrf_service(request)
cookie_value: Optional[str] = request.cookies.get(CSRF_COOKIE_NAME)
if not service.verify(cookie_value=cookie_value, submitted=csrf_token):
raise HTTPException(status_code=403, detail="CSRF verification failed")
def require_csrf_header(
request: Request,
) -> None:
"""Verify an ``X-CSRF-Token`` header matches the signed cookie.
Used by the JS-driven preview + media upload endpoints. The header
name is case-insensitive — Starlette canonicalizes on read.
"""
service: CSRFService = _get_csrf_service(request)
cookie_value: Optional[str] = request.cookies.get(CSRF_COOKIE_NAME)
submitted: Optional[str] = request.headers.get("x-csrf-token")
if not service.verify(cookie_value=cookie_value, submitted=submitted):
raise HTTPException(status_code=403, detail="CSRF verification failed")

View File

@@ -24,6 +24,20 @@ Phase 3 additions:
``RateLimitExceeded`` exception handler (renders
``admin/rate_limited.html`` + HTTP 429 + audit row).
- Include the admin router.
Phase 4 additions:
- Build a separate :class:`itsdangerous.URLSafeTimedSerializer`
salted with ``"csrf"`` and wrap it in a :class:`CSRFService`.
- Instantiate :class:`MarkdownService`,
:class:`AdminPostsService`, :class:`AdminPagesService`, and
:class:`MediaService` and attach them to ``app.state``.
- Mount the admin CMS router.
- Install a lightweight middleware that issues / refreshes the
CSRF cookie on admin GET requests and exposes the token via
``request.state.csrf_token`` for template rendering.
- Mount ``settings.media_root`` at ``/media`` as a StaticFiles
route so uploaded images are publicly reachable under the
Markdown URLs the admin inserts.
"""
from __future__ import annotations
@@ -32,10 +46,12 @@ from pathlib import Path
import structlog
from fastapi import FastAPI, Request
from fastapi.responses import Response
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from itsdangerous import URLSafeTimedSerializer
from slowapi.errors import RateLimitExceeded
from starlette.middleware.base import BaseHTTPMiddleware
from app import __version__
from app.config import get_settings
@@ -43,11 +59,17 @@ 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.admin_cms import router as admin_cms_router
from app.routes.health import router as health_router
from app.routes.public import router as public_router
from app.services.admin_pages import AdminPagesService
from app.services.admin_posts import AdminPostsService
from app.services.audit import AuditService
from app.services.auth import AuthService
from app.services.csrf import CSRF_COOKIE_NAME, CSRFService
from app.services.email import EmailService
from app.services.markdown import MarkdownService
from app.services.media import MediaService
from app.services.pages import PageService
from app.services.posts import PostService
from app.services.rate_limit import create_limiter
@@ -62,6 +84,55 @@ _TEMPLATES_DIR: Path = _PACKAGE_ROOT / "templates"
_STATIC_DIR: Path = _PACKAGE_ROOT / "static"
class CSRFCookieMiddleware(BaseHTTPMiddleware):
"""Issue / refresh the CSRF cookie on admin GET responses.
The middleware is intentionally narrow: it only fires for requests
whose path starts with ``/admin`` AND method is GET. That's the
set of responses that render HTML with a form waiting to be
submitted; mutating endpoints never set the cookie themselves.
For every such request we:
1. Read the existing ``cb_csrf`` cookie (if any).
2. Call :meth:`CSRFService.issue` which reuses the underlying
nonce when the cookie is still valid — this means a tab that
GETs the dashboard, then POSTs 30 minutes later, still
matches.
3. Stash the token on ``request.state.csrf_token`` so route
handlers pass it into Jinja.
4. After the downstream response is produced, set / refresh the
cookie header.
"""
def __init__(self, app, csrf_service: CSRFService) -> None:
"""Store the service by reference; BaseHTTPMiddleware takes the ASGI app."""
super().__init__(app)
self._csrf_service: CSRFService = csrf_service
async def dispatch(self, request: Request, call_next):
"""Run the admin-GET issue hook around the downstream handler."""
should_issue = (
request.url.path.startswith("/admin")
and request.method == "GET"
)
if should_issue:
existing = request.cookies.get(CSRF_COOKIE_NAME)
token, cookie_value = self._csrf_service.issue(existing)
request.state.csrf_token = token
else:
cookie_value = None
response: Response = await call_next(request)
if should_issue and cookie_value is not None:
response.set_cookie(
value=cookie_value,
**self._csrf_service.cookie_params(),
)
return response
def create_app() -> FastAPI:
"""Build and return the FastAPI application.
@@ -94,6 +165,11 @@ def create_app() -> FastAPI:
run_migrations(engine)
run_seed(engine)
# Ensure the media storage directory exists — Starlette's
# StaticFiles mount refuses to start without a real directory.
media_root = Path(settings.media_root)
media_root.mkdir(parents=True, exist_ok=True)
application = FastAPI(
title="Chicken Babies R Us",
version=__version__,
@@ -113,6 +189,15 @@ def create_app() -> FastAPI:
name="static",
)
# Public mount for admin-uploaded images. Kept separate from /static
# so admin uploads never collide with the brand assets shipped with
# the container image.
application.mount(
"/media",
StaticFiles(directory=str(media_root)),
name="media",
)
# Single shared Jinja2 environment. Storing it on ``app.state`` keeps
# route modules free of an import dependency on this module (which
# would be circular once admin/auth routers are added in later
@@ -151,6 +236,47 @@ def create_app() -> FastAPI:
application.state.session_service = session_service
application.state.auth_service = auth_service
# --- Phase 4 wiring -----------------------------------------------------
# CSRF signer: separate salt so a session cookie never validates
# as a CSRF token (domain separation via salt).
csrf_signer = URLSafeTimedSerializer(settings.secret_key, salt="csrf")
csrf_service = CSRFService(
csrf_signer,
production=(settings.app_env == "production"),
)
application.state.csrf_service = csrf_service
markdown_service = MarkdownService()
application.state.markdown_service = markdown_service
admin_posts_service = AdminPostsService(
engine=engine,
markdown=markdown_service,
post_service=application.state.post_service,
page_service=application.state.page_service,
audit=audit_service,
)
admin_pages_service = AdminPagesService(
engine=engine,
markdown=markdown_service,
page_service=application.state.page_service,
post_service=application.state.post_service,
audit=audit_service,
)
media_service = MediaService(
engine=engine,
media_root=str(media_root),
audit=audit_service,
)
application.state.admin_posts_service = admin_posts_service
application.state.admin_pages_service = admin_pages_service
application.state.media_service = media_service
# CSRF cookie middleware — narrow to admin GETs; everything else
# passes through untouched so public routes are unaffected.
application.add_middleware(CSRFCookieMiddleware, csrf_service=csrf_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).
@@ -168,6 +294,7 @@ def create_app() -> FastAPI:
application.include_router(health_router)
application.include_router(public_router)
application.include_router(admin_router)
application.include_router(admin_cms_router)
# Single structured startup event. Do NOT include secret material.
logger = structlog.get_logger(__name__)

View File

@@ -32,6 +32,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse, Response
from fastapi.templating import Jinja2Templates
from app.dependencies.auth import require_admin
from app.dependencies.csrf import require_csrf_form
from app.models.entities import User
from app.services.auth import AuthService, RateLimitedError
from app.services.rate_limit import limiter
@@ -92,8 +93,11 @@ def admin_login_form(
) -> 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.
The login form is deliberately NOT CSRF-protected: the user is
pre-authentication and no session cookie exists yet, so there is
no authenticated context for a forged request to hijack. The
``ADMIN_EMAILS`` allowlist and SlowAPI rate limit are what keep
this endpoint safe.
"""
return templates.TemplateResponse(
request,
@@ -209,40 +213,20 @@ def admin_auth_consume(
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},
)
# The authenticated landing page (``GET /admin``) now lives in
# :mod:`app.routes.admin_cms` as the dashboard. This module stays
# scoped to pre-auth / auth-lifecycle endpoints.
# ---------------------------------------------------------------------------
# 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),
_csrf: None = Depends(require_csrf_form),
) -> Response:
"""Revoke the current session and clear the cookie.

500
app/routes/admin_cms.py Normal file
View File

@@ -0,0 +1,500 @@
"""Admin CMS routes — dashboard, post CRUD, About edit, media upload.
These handlers all live behind :func:`require_admin`. Mutating
endpoints additionally pull a CSRF dependency
(:func:`require_csrf_form` or :func:`require_csrf_header`) so the
double-submit cookie is verified before any state change.
Each handler does the absolute minimum: pull services off the app
state, call their methods, translate the result into an HTTP
response. No business logic, no SQL.
Routing map
-----------
- ``GET /admin`` — dashboard.
- ``GET /admin/posts/new`` — create form.
- ``POST /admin/posts`` — create handler (CSRF).
- ``GET /admin/posts/{id}/edit`` — edit form.
- ``POST /admin/posts/{id}`` — update (CSRF).
- ``POST /admin/posts/{id}/delete`` — delete (CSRF).
- ``POST /admin/posts/{id}/publish`` — publish toggle (CSRF).
- ``GET /admin/pages/about/edit`` — About edit form.
- ``POST /admin/pages/about`` — About update (CSRF).
- ``POST /admin/media/upload`` — multipart upload (header CSRF).
- ``POST /admin/preview`` — Markdown → HTML preview (header CSRF).
"""
from __future__ import annotations
from typing import Optional
import structlog
from fastapi import (
APIRouter,
Depends,
File,
Form,
HTTPException,
Request,
UploadFile,
)
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response
from fastapi.templating import Jinja2Templates
from app.dependencies.auth import require_admin
from app.dependencies.csrf import require_csrf_form, require_csrf_header
from app.models.entities import PostStatus, User
from app.services.admin_pages import AdminPagesService
from app.services.admin_posts import AdminPostsService
from app.services.markdown import MarkdownService
from app.services.media import MediaRejectedError, MediaService
router: APIRouter = APIRouter(tags=["admin-cms"])
_log = structlog.get_logger(__name__)
# ---------------------------------------------------------------------------
# DI helpers
# ---------------------------------------------------------------------------
def _get_templates(request: Request) -> Jinja2Templates:
"""Return the app-scoped :class:`Jinja2Templates`."""
return request.app.state.templates
def _get_admin_posts(request: Request) -> AdminPostsService:
"""Return the app-scoped :class:`AdminPostsService`."""
return request.app.state.admin_posts_service
def _get_admin_pages(request: Request) -> AdminPagesService:
"""Return the app-scoped :class:`AdminPagesService`."""
return request.app.state.admin_pages_service
def _get_media(request: Request) -> MediaService:
"""Return the app-scoped :class:`MediaService`."""
return request.app.state.media_service
def _get_markdown(request: Request) -> MarkdownService:
"""Return the app-scoped :class:`MarkdownService`."""
return request.app.state.markdown_service
def _get_csrf_token_for_template(request: Request) -> str:
"""Return the CSRF token to embed in the rendered admin templates.
The middleware (see :mod:`app.main`) sets ``request.state.csrf_token``
on every admin GET after ensuring the cookie is in sync. Handlers
pull it from request state and pass it into the template context.
"""
return getattr(request.state, "csrf_token", "") or ""
# ---------------------------------------------------------------------------
# GET /admin — dashboard
# ---------------------------------------------------------------------------
@router.get("/admin", response_class=HTMLResponse, summary="Admin dashboard")
def admin_dashboard(
request: Request,
user: User = Depends(require_admin),
templates: Jinja2Templates = Depends(_get_templates),
admin_posts: AdminPostsService = Depends(_get_admin_posts),
admin_pages: AdminPagesService = Depends(_get_admin_pages),
) -> HTMLResponse:
"""Render the dashboard: posts list + About edit link.
Posts are sorted newest-updated-first and include both drafts and
published posts — the admin table surfaces the status badge.
"""
posts = admin_posts.list_all()
about = admin_pages.get_about()
# Optional flash from PRG query param — we keep it minimal.
msg = request.query_params.get("msg") or ""
return templates.TemplateResponse(
request,
"admin/dashboard.html",
{
"user": user,
"posts": posts,
"about": about,
"msg": msg,
"csrf_token": _get_csrf_token_for_template(request),
},
)
# ---------------------------------------------------------------------------
# GET /admin/posts/new
# ---------------------------------------------------------------------------
@router.get(
"/admin/posts/new",
response_class=HTMLResponse,
summary="Admin: new post form",
)
def admin_post_new_form(
request: Request,
user: User = Depends(require_admin),
templates: Jinja2Templates = Depends(_get_templates),
) -> HTMLResponse:
"""Render the empty create form.
No post id → the form POSTs to ``/admin/posts``. Slug is not shown
because we auto-generate it from the title on create.
"""
return templates.TemplateResponse(
request,
"admin/post_form.html",
{
"user": user,
"post": None,
"form": {"title": "", "body_md": "", "status": PostStatus.DRAFT.value},
"errors": {},
"csrf_token": _get_csrf_token_for_template(request),
},
)
# ---------------------------------------------------------------------------
# POST /admin/posts — create
# ---------------------------------------------------------------------------
@router.post("/admin/posts", summary="Admin: create post")
def admin_post_create(
request: Request,
title: str = Form(default=""),
body_md: str = Form(default=""),
status: str = Form(default=PostStatus.DRAFT.value),
user: User = Depends(require_admin),
templates: Jinja2Templates = Depends(_get_templates),
admin_posts: AdminPostsService = Depends(_get_admin_posts),
_csrf: None = Depends(require_csrf_form),
) -> Response:
"""Handle the new-post submission and redirect to the dashboard.
Minimal validation:
- title must be non-empty after strip
- status must be a valid :class:`PostStatus` value
On validation error we re-render the form with the submitted
values so Head Hen doesn't retype.
"""
errors: dict[str, str] = {}
clean_title = (title or "").strip()
if not clean_title:
errors["title"] = "Title is required."
try:
status_enum = PostStatus(status)
except ValueError:
errors["status"] = "Invalid status."
status_enum = PostStatus.DRAFT
if errors:
return templates.TemplateResponse(
request,
"admin/post_form.html",
{
"user": user,
"post": None,
"form": {
"title": title,
"body_md": body_md,
"status": status_enum.value,
},
"errors": errors,
"csrf_token": _get_csrf_token_for_template(request),
},
status_code=400,
)
admin_posts.create(
title=clean_title,
body_md=body_md or "",
status=status_enum,
author_id=user.id,
)
return RedirectResponse(url="/admin?msg=created", status_code=303)
# ---------------------------------------------------------------------------
# GET /admin/posts/{id}/edit
# ---------------------------------------------------------------------------
@router.get(
"/admin/posts/{post_id}/edit",
response_class=HTMLResponse,
summary="Admin: edit post form",
)
def admin_post_edit_form(
request: Request,
post_id: int,
user: User = Depends(require_admin),
templates: Jinja2Templates = Depends(_get_templates),
admin_posts: AdminPostsService = Depends(_get_admin_posts),
) -> HTMLResponse:
"""Render the edit form for an existing post."""
post = admin_posts.get_by_id(post_id)
if post is None:
raise HTTPException(status_code=404, detail="Post not found")
return templates.TemplateResponse(
request,
"admin/post_form.html",
{
"user": user,
"post": post,
"form": {
"title": post.title,
"body_md": post.body_md,
"status": post.status.value,
},
"errors": {},
"csrf_token": _get_csrf_token_for_template(request),
},
)
# ---------------------------------------------------------------------------
# POST /admin/posts/{id} — update
# ---------------------------------------------------------------------------
@router.post("/admin/posts/{post_id}", summary="Admin: update post")
def admin_post_update(
request: Request,
post_id: int,
title: str = Form(default=""),
body_md: str = Form(default=""),
user: User = Depends(require_admin),
templates: Jinja2Templates = Depends(_get_templates),
admin_posts: AdminPostsService = Depends(_get_admin_posts),
_csrf: None = Depends(require_csrf_form),
) -> Response:
"""Apply title + body edits to an existing post.
Slug changes are not permitted via this path — server-side
enforcement of the "slug lock on publish" policy (see
:class:`AdminPostsService`).
"""
existing = admin_posts.get_by_id(post_id)
if existing is None:
raise HTTPException(status_code=404, detail="Post not found")
clean_title = (title or "").strip()
errors: dict[str, str] = {}
if not clean_title:
errors["title"] = "Title is required."
if errors:
return templates.TemplateResponse(
request,
"admin/post_form.html",
{
"user": user,
"post": existing,
"form": {
"title": title,
"body_md": body_md,
"status": existing.status.value,
},
"errors": errors,
"csrf_token": _get_csrf_token_for_template(request),
},
status_code=400,
)
admin_posts.update(
post_id,
title=clean_title,
body_md=body_md or "",
actor_user_id=user.id,
)
return RedirectResponse(url="/admin?msg=saved", status_code=303)
# ---------------------------------------------------------------------------
# POST /admin/posts/{id}/delete
# ---------------------------------------------------------------------------
@router.post("/admin/posts/{post_id}/delete", summary="Admin: delete post")
def admin_post_delete(
request: Request,
post_id: int,
user: User = Depends(require_admin),
admin_posts: AdminPostsService = Depends(_get_admin_posts),
_csrf: None = Depends(require_csrf_form),
) -> Response:
"""Hard-delete a post row."""
deleted = admin_posts.delete(post_id, actor_user_id=user.id)
if not deleted:
raise HTTPException(status_code=404, detail="Post not found")
return RedirectResponse(url="/admin?msg=deleted", status_code=303)
# ---------------------------------------------------------------------------
# POST /admin/posts/{id}/publish — publish/unpublish toggle
# ---------------------------------------------------------------------------
@router.post(
"/admin/posts/{post_id}/publish", summary="Admin: toggle publish state"
)
def admin_post_toggle_publish(
request: Request,
post_id: int,
user: User = Depends(require_admin),
admin_posts: AdminPostsService = Depends(_get_admin_posts),
_csrf: None = Depends(require_csrf_form),
) -> Response:
"""Flip draft ↔ published."""
updated = admin_posts.toggle_publish(post_id, actor_user_id=user.id)
if updated is None:
raise HTTPException(status_code=404, detail="Post not found")
# Friendly-ish flash so the admin sees the result of the toggle.
msg = "published" if updated.status is PostStatus.PUBLISHED else "unpublished"
return RedirectResponse(url=f"/admin?msg={msg}", status_code=303)
# ---------------------------------------------------------------------------
# GET /admin/pages/about/edit
# ---------------------------------------------------------------------------
@router.get(
"/admin/pages/about/edit",
response_class=HTMLResponse,
summary="Admin: edit About page",
)
def admin_about_edit_form(
request: Request,
user: User = Depends(require_admin),
templates: Jinja2Templates = Depends(_get_templates),
admin_pages: AdminPagesService = Depends(_get_admin_pages),
) -> HTMLResponse:
"""Render the About-page edit form."""
page = admin_pages.get_about()
if page is None:
raise HTTPException(status_code=500, detail="About page missing")
return templates.TemplateResponse(
request,
"admin/page_form.html",
{
"user": user,
"page": page,
"form": {"title": page.title, "body_md": page.body_md},
"errors": {},
"csrf_token": _get_csrf_token_for_template(request),
},
)
# ---------------------------------------------------------------------------
# POST /admin/pages/about — update
# ---------------------------------------------------------------------------
@router.post("/admin/pages/about", summary="Admin: update About page")
def admin_about_update(
request: Request,
title: str = Form(default=""),
body_md: str = Form(default=""),
user: User = Depends(require_admin),
templates: Jinja2Templates = Depends(_get_templates),
admin_pages: AdminPagesService = Depends(_get_admin_pages),
_csrf: None = Depends(require_csrf_form),
) -> Response:
"""Apply edits to the About page (slug is fixed)."""
clean_title = (title or "").strip()
errors: dict[str, str] = {}
if not clean_title:
errors["title"] = "Title is required."
if errors:
page = admin_pages.get_about()
return templates.TemplateResponse(
request,
"admin/page_form.html",
{
"user": user,
"page": page,
"form": {"title": title, "body_md": body_md},
"errors": errors,
"csrf_token": _get_csrf_token_for_template(request),
},
status_code=400,
)
admin_pages.update_about(
title=clean_title,
body_md=body_md or "",
actor_user_id=user.id,
)
return RedirectResponse(url="/admin?msg=saved", status_code=303)
# ---------------------------------------------------------------------------
# POST /admin/media/upload
# ---------------------------------------------------------------------------
@router.post("/admin/media/upload", summary="Admin: upload image")
async def admin_media_upload(
request: Request,
file: UploadFile = File(...),
alt_text: str = Form(default=""),
user: User = Depends(require_admin),
media: MediaService = Depends(_get_media),
_csrf: None = Depends(require_csrf_header),
) -> JSONResponse:
"""Validate and store an uploaded image.
Response JSON is small by design — the drag-drop JS only needs a
URL to splice into the Markdown source as ``![alt](url)``.
"""
data = await file.read()
try:
record = media.save_upload(
original_filename=file.filename or "",
data=data,
uploaded_by=user.id,
alt_text=alt_text or "",
)
except MediaRejectedError as exc:
return JSONResponse(
{"error": str(exc)},
status_code=400,
)
return JSONResponse(
{
"id": record.id,
"url": media.public_url(record),
"alt": record.alt_text,
"filename": record.filename,
"size_bytes": record.size_bytes,
}
)
# ---------------------------------------------------------------------------
# POST /admin/preview
# ---------------------------------------------------------------------------
@router.post(
"/admin/preview",
response_class=HTMLResponse,
summary="Admin: Markdown preview",
)
def admin_preview(
request: Request,
markdown: str = Form(default=""),
user: User = Depends(require_admin),
md: MarkdownService = Depends(_get_markdown),
_csrf: None = Depends(require_csrf_header),
) -> HTMLResponse:
"""Render ``markdown`` through the sanitizer and return an HTML fragment.
The fragment is NOT wrapped in a full page — it is ``innerHTML``-safe
output from the same pipeline that stores ``body_html_cached``. The
route reuses :class:`MarkdownService` so preview output exactly
matches what will eventually be served to the public.
"""
rendered: str = md.render(markdown or "")
return HTMLResponse(content=rendered)
# ---------------------------------------------------------------------------
# Optional aliases for backward-compat imports
# ---------------------------------------------------------------------------
# If another module (e.g. tests) imports ``router`` from here, the
# attribute name stays stable.
_ = router # silence "not used" in linters that don't pick up FastAPI magic
# Avoid "Optional unused" when the type is only referenced via Depends.
_Optional = Optional # pragma: no cover

124
app/services/admin_pages.py Normal file
View File

@@ -0,0 +1,124 @@
"""Admin-side (write) page service.
The public site only has one editable page — "About" — so this
service is intentionally narrower than :class:`AdminPostsService`. The
slug is a fixed literal (``"about"``) and cannot be changed through
the admin. Only the title and body may be edited.
Every write:
- re-renders Markdown → sanitized HTML into ``body_html_cached`` so
the public read path stays a single SELECT.
- bumps ``updated_at``.
- emits an ``AuditService`` ``page_updated`` event.
- invalidates the public :class:`PageService` (and, defensively, the
:class:`PostService`) cache so the next request sees the new copy.
"""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Optional
import structlog
from sqlalchemy import Engine, text
from app.models.entities import Page
from app.models.mappers import row_to_page
from app.services.audit import AuditService
from app.services.markdown import MarkdownService
from app.services.pages import PageService
from app.services.posts import PostService
_log = structlog.get_logger(__name__)
# The single editable page's slug. Hard-coded here (not injected) so
# the CLI contract is impossible to misuse — there is no way to point
# this service at a different slug.
ABOUT_SLUG: str = "about"
class AdminPagesService:
"""Write-side service for the About page."""
def __init__(
self,
engine: Engine,
markdown: MarkdownService,
page_service: PageService,
post_service: PostService,
audit: AuditService,
) -> None:
self._engine: Engine = engine
self._markdown: MarkdownService = markdown
self._page_service: PageService = page_service
self._post_service: PostService = post_service
self._audit: AuditService = audit
# ------------------------------------------------------------------
# Reads
# ------------------------------------------------------------------
def get_about(self) -> Optional[Page]:
"""Return the current About page row, or ``None`` if absent."""
with self._engine.connect() as conn:
row = conn.execute(
text(
"SELECT id, slug, title, body_md, body_html_cached,"
" updated_at, published"
" FROM pages WHERE slug = :slug LIMIT 1"
),
{"slug": ABOUT_SLUG},
).mappings().first()
return row_to_page(row) if row is not None else None
# ------------------------------------------------------------------
# Writes
# ------------------------------------------------------------------
def update_about(
self,
*,
title: str,
body_md: str,
actor_user_id: int,
) -> Optional[Page]:
"""Update the About page's title + body.
Slug is immutable — the admin form does not expose it.
"""
existing = self.get_about()
if existing is None:
return None
clean_title = (title or "").strip()
clean_body = body_md or ""
body_html = self._markdown.render(clean_body)
now_iso = datetime.now(timezone.utc).isoformat()
with self._engine.begin() as conn:
conn.execute(
text(
"UPDATE pages"
" SET title = :title, body_md = :body_md,"
" body_html_cached = :body_html,"
" updated_at = :updated_at"
" WHERE slug = :slug"
),
{
"title": clean_title,
"body_md": clean_body,
"body_html": body_html,
"updated_at": now_iso,
"slug": ABOUT_SLUG,
},
)
self._audit.record(
"page_updated",
user_id=actor_user_id,
detail={"slug": ABOUT_SLUG},
)
self._page_service.invalidate_all()
self._post_service.invalidate_all()
return self.get_about()

383
app/services/admin_posts.py Normal file
View File

@@ -0,0 +1,383 @@
"""Admin-side (write) post service.
Mirrors the shape of :class:`app.services.posts.PostService` but for
the admin CRUD path. Responsibilities:
- create / update / delete posts
- toggle publish state
- auto-generate unique slugs from titles on create (draft only)
- re-render Markdown to ``body_html_cached`` on every write
- audit every write via :class:`AuditService` using descriptive
``event_type`` strings
- invalidate both :class:`PostService` and :class:`PageService` caches
so the public site reflects the change immediately
All writes use parameterized SQL (``text(":bind")``). No user input is
ever interpolated into a query string.
The service treats ``author_user_id`` as an immutable field: once a
post is created, edits do NOT reassign authorship, even if a different
admin saves the edit. This matches the single-author ("Head Hen")
reality of the site.
Slug lock-on-publish
--------------------
A slug may only be auto-regenerated on title change while the post is
a draft. Once a post has been published even once, the slug is locked
server-side — callers cannot change it via the update path, even if
they later unpublish the post. This preserves any inbound links that
went live while the post was published.
"""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Optional
import structlog
from sqlalchemy import Engine, text
from app.models.entities import Post, PostStatus
from app.models.mappers import row_to_post
from app.services.audit import AuditService
from app.services.markdown import MarkdownService
from app.services.pages import PageService
from app.services.posts import PostService
from app.services.slugs import ensure_unique, slugify
_log = structlog.get_logger(__name__)
class AdminPostsService:
"""Write-side orchestration for blog posts.
Parameters
----------
engine:
Shared SQLAlchemy engine. Never opens its own.
markdown:
Shared :class:`MarkdownService` used to re-render on every
write so the public read path pays only a single SELECT.
post_service:
The public read-side service. Invalidated after every write so
the home page reflects the change immediately.
page_service:
Same rationale — a post edit doesn't change page content but
we conservatively invalidate to keep cache logic uniform.
audit:
:class:`AuditService` for descriptive admin write events.
"""
def __init__(
self,
engine: Engine,
markdown: MarkdownService,
post_service: PostService,
page_service: PageService,
audit: AuditService,
) -> None:
self._engine: Engine = engine
self._markdown: MarkdownService = markdown
self._post_service: PostService = post_service
self._page_service: PageService = page_service
self._audit: AuditService = audit
# ------------------------------------------------------------------
# Reads (admin dashboard)
# ------------------------------------------------------------------
def list_all(self) -> list[Post]:
"""Return every post, newest-updated-first.
Drafts and published posts are both included; the dashboard
surfaces the status column so Head Hen can work on unpublished
material.
"""
with self._engine.connect() as conn:
rows = (
conn.execute(
text(
"SELECT id, slug, title, body_md, body_html_cached,"
" status, published_at, updated_at, author_user_id"
" FROM posts"
" ORDER BY updated_at DESC, id DESC"
)
)
.mappings()
.all()
)
return [row_to_post(row) for row in rows]
def get_by_id(self, post_id: int) -> Optional[Post]:
"""Return the :class:`Post` for ``post_id`` or ``None`` if absent."""
with self._engine.connect() as conn:
row = conn.execute(
text(
"SELECT id, slug, title, body_md, body_html_cached,"
" status, published_at, updated_at, author_user_id"
" FROM posts WHERE id = :id LIMIT 1"
),
{"id": post_id},
).mappings().first()
return row_to_post(row) if row is not None else None
# ------------------------------------------------------------------
# Writes
# ------------------------------------------------------------------
def create(
self,
*,
title: str,
body_md: str,
status: PostStatus,
author_id: int,
) -> Post:
"""Insert a new post row and return the loaded :class:`Post`.
Flow
----
1. Slugify the title; ensure uniqueness via the closure over the
DB so concurrent creates cannot collide on the UNIQUE index.
2. Render Markdown to sanitized HTML.
3. If ``status == PUBLISHED`` stamp ``published_at = now``;
otherwise leave NULL.
4. Insert.
5. Audit ``post_created`` (and ``post_published`` when the
initial status is published).
6. Invalidate caches.
"""
clean_title = (title or "").strip()
clean_body = body_md or ""
base_slug = slugify(clean_title)
# The closure escapes the engine so ensure_unique can check
# without opening a long-lived transaction.
unique_slug = ensure_unique(base_slug, self._slug_exists)
body_html = self._markdown.render(clean_body)
now = datetime.now(timezone.utc)
now_iso = now.isoformat()
published_at_iso: Optional[str] = (
now_iso if status is PostStatus.PUBLISHED else None
)
with self._engine.begin() as conn:
result = conn.execute(
text(
"INSERT INTO posts"
" (slug, title, body_md, body_html_cached, status,"
" published_at, updated_at, author_user_id)"
" VALUES (:slug, :title, :body_md, :body_html,"
" :status, :published_at, :updated_at, :author_id)"
),
{
"slug": unique_slug,
"title": clean_title,
"body_md": clean_body,
"body_html": body_html,
"status": status.value,
"published_at": published_at_iso,
"updated_at": now_iso,
"author_id": author_id,
},
)
new_id = int(result.lastrowid) # type: ignore[arg-type]
row = conn.execute(
text(
"SELECT id, slug, title, body_md, body_html_cached,"
" status, published_at, updated_at, author_user_id"
" FROM posts WHERE id = :id"
),
{"id": new_id},
).mappings().first()
if row is None: # pragma: no cover — just inserted
raise RuntimeError("failed to reload just-inserted post row")
post = row_to_post(row)
self._audit.record(
"post_created",
user_id=author_id,
detail={"post_id": post.id, "slug": post.slug, "status": post.status.value},
)
if post.status is PostStatus.PUBLISHED:
self._audit.record(
"post_published",
user_id=author_id,
detail={"post_id": post.id, "slug": post.slug},
)
self._invalidate_caches()
return post
def update(
self,
post_id: int,
*,
title: str,
body_md: str,
actor_user_id: int,
) -> Optional[Post]:
"""Update a post's title + body. Return the refreshed :class:`Post`.
Behavior
--------
- The slug is NEVER regenerated by an update call. While the
post is still a draft the admin may delete + recreate to pick
a new slug; once published the slug is permanent per the
security contract (external links must not break).
- ``author_user_id`` is preserved — this endpoint does not
transfer authorship.
- ``published_at`` is preserved verbatim. Publishing happens via
:meth:`toggle_publish`.
- Always re-renders Markdown so ``body_html_cached`` stays in
sync with ``body_md``.
- Always bumps ``updated_at``.
"""
existing = self.get_by_id(post_id)
if existing is None:
return None
clean_title = (title or "").strip()
clean_body = body_md or ""
body_html = self._markdown.render(clean_body)
now_iso = datetime.now(timezone.utc).isoformat()
with self._engine.begin() as conn:
conn.execute(
text(
"UPDATE posts"
" SET title = :title, body_md = :body_md,"
" body_html_cached = :body_html,"
" updated_at = :updated_at"
" WHERE id = :id"
),
{
"title": clean_title,
"body_md": clean_body,
"body_html": body_html,
"updated_at": now_iso,
"id": post_id,
},
)
self._audit.record(
"post_updated",
user_id=actor_user_id,
detail={"post_id": post_id, "slug": existing.slug},
)
self._invalidate_caches()
return self.get_by_id(post_id)
def delete(self, post_id: int, *, actor_user_id: int) -> bool:
"""Delete a post row. Return True if something was deleted.
Media rows uploaded during drafting are NOT cleaned up here —
uploads aren't linked to posts in the schema, and orphan-sweep
is explicitly out of scope per the Phase 4 brief.
"""
existing = self.get_by_id(post_id)
if existing is None:
return False
with self._engine.begin() as conn:
conn.execute(
text("DELETE FROM posts WHERE id = :id"),
{"id": post_id},
)
self._audit.record(
"post_deleted",
user_id=actor_user_id,
detail={"post_id": post_id, "slug": existing.slug},
)
self._invalidate_caches()
return True
def toggle_publish(self, post_id: int, *, actor_user_id: int) -> Optional[Post]:
"""Flip draft ↔ published. Return the updated post, or ``None``.
Contract (see Phase 4 brief constraint 7):
- Draft → Published: set ``published_at = now`` ONLY if it was
previously NULL. If the post was once published, unpublished,
and is now being re-published we preserve the original
publish timestamp so the public list ordering stays stable.
- Published → Draft: status flips, ``published_at`` is preserved.
"""
existing = self.get_by_id(post_id)
if existing is None:
return None
now_iso = datetime.now(timezone.utc).isoformat()
if existing.status is PostStatus.PUBLISHED:
new_status = PostStatus.DRAFT
# Preserve existing published_at on unpublish. No event_type
# branch yet — we emit post_unpublished below.
published_at_iso: Optional[str] = (
existing.published_at.isoformat()
if existing.published_at is not None
else None
)
event_type = "post_unpublished"
else:
new_status = PostStatus.PUBLISHED
# First-publish stamp. Preserve any prior published_at so
# re-publish doesn't renumber the post on the front page.
if existing.published_at is None:
published_at_iso = now_iso
else:
published_at_iso = existing.published_at.isoformat()
event_type = "post_published"
with self._engine.begin() as conn:
conn.execute(
text(
"UPDATE posts"
" SET status = :status,"
" published_at = :published_at,"
" updated_at = :updated_at"
" WHERE id = :id"
),
{
"status": new_status.value,
"published_at": published_at_iso,
"updated_at": now_iso,
"id": post_id,
},
)
self._audit.record(
event_type,
user_id=actor_user_id,
detail={"post_id": post_id, "slug": existing.slug},
)
self._invalidate_caches()
return self.get_by_id(post_id)
# ------------------------------------------------------------------
# Internals
# ------------------------------------------------------------------
def _slug_exists(self, candidate: str) -> bool:
"""Return True if a row with ``slug = candidate`` is already present."""
with self._engine.connect() as conn:
row = conn.execute(
text("SELECT 1 FROM posts WHERE slug = :s LIMIT 1"),
{"s": candidate},
).first()
return row is not None
def _invalidate_caches(self) -> None:
"""Drop both the post and page read-side caches.
Post invalidation is strictly required; page invalidation is
defensive — the schemas are separate, but keeping cache
invalidation uniform makes it obvious Phase 4 writes never
leave a stale public read.
"""
self._post_service.invalidate_all()
self._page_service.invalidate_all()
def get_admin_posts_service(request): # pragma: no cover — trivial
"""FastAPI dependency — pull the service off ``app.state``."""
return request.app.state.admin_posts_service

167
app/services/csrf.py Normal file
View File

@@ -0,0 +1,167 @@
"""CSRF double-submit cookie service.
Protects admin-write endpoints against cross-site request forgery by
requiring a signed token to be submitted BOTH as a cookie and as a
form field / header. An attacker can forge requests but cannot read
the cookie (SameSite=Lax blocks cross-site automatic cookie sending,
and even if the browser sent it, cross-site JS still cannot read
cookies on this origin). Matching the submitted value to the cookie
value then proves the request originated from our own pages.
Design
------
- The cookie stores a signed opaque nonce. Signing prevents a malicious
ad iframe (or any JS on a non-origin page) from producing a cookie
value that would later match a crafted form submission.
- The nonce itself is 256-bit (``secrets.token_urlsafe(32)``), generated
per-browser on first admin GET and reused for the session. Rotating
per request would invalidate any still-open admin tab on every nav,
which the small-scale admin UX cannot tolerate.
- Verification unsigns the submitted token and compares the raw nonce
to the raw nonce unsigned from the cookie using :func:`hmac.compare_digest`
(constant-time) to foreclose timing side channels.
- The cookie is ``HttpOnly=False`` so the minimal admin JS (live
preview, upload) can read it to set the ``X-CSRF-Token`` header on
fetch requests. This is the conventional double-submit cookie setup
— the XSS risk is already mitigated by the Markdown sanitizer and
the session cookie remains HttpOnly.
The service is a small collaborator: it does not know about FastAPI
routes, request objects, or templates. The :mod:`app.dependencies.csrf`
module wraps the verify call in a FastAPI dependency.
"""
from __future__ import annotations
import hmac
import secrets
from typing import Optional
import structlog
from itsdangerous import BadSignature, URLSafeTimedSerializer
_log = structlog.get_logger(__name__)
# Cookie name kept here as a module-level constant so routes,
# dependencies, and templates stay in sync.
CSRF_COOKIE_NAME: str = "cb_csrf"
# Default max age — matches the session TTL ceiling. A valid admin
# session already enforces the 30-day cap; the CSRF cookie merely
# piggybacks.
_DEFAULT_MAX_AGE_SEC: int = 30 * 86400
class CSRFService:
"""Issue and verify double-submit CSRF tokens.
Parameters
----------
signer:
Pre-built :class:`itsdangerous.URLSafeTimedSerializer`. The
caller is responsible for constructing it with
``salt="csrf"`` so a session-cookie token can never be
replayed as a CSRF token and vice-versa.
production:
When True, the issued cookie carries the ``Secure`` flag. Dev
(plain-HTTP 127.0.0.1) needs it off or the browser drops the
cookie entirely.
"""
def __init__(
self,
signer: URLSafeTimedSerializer,
*,
production: bool = False,
max_age_sec: int = _DEFAULT_MAX_AGE_SEC,
) -> None:
"""Store the signer and cookie-policy flags by reference."""
self._signer: URLSafeTimedSerializer = signer
self._production: bool = production
self._max_age_sec: int = int(max_age_sec)
# ------------------------------------------------------------------
# Issue
# ------------------------------------------------------------------
def issue(self, existing_cookie: Optional[str] = None) -> tuple[str, str]:
"""Return ``(token, cookie_value)`` — reuse or mint as appropriate.
If ``existing_cookie`` is a valid signed nonce (still within
TTL), we reuse the underlying nonce so the same token keeps
working across GET / POST cycles in the same admin session.
Otherwise we mint a fresh nonce.
The cookie value and the form/header token value are the SAME
signed string — this is the "double submit" contract. The
verify path re-signs nothing; it just compares the unsigned
raw nonces.
"""
raw = self._unsign_or_none(existing_cookie)
if raw is None:
raw = secrets.token_urlsafe(32)
signed = self._signer.dumps(raw)
# Token and cookie are both the signed string. Callers are free
# to submit either in a form field OR a header; verify accepts
# both shapes.
return signed, signed
# ------------------------------------------------------------------
# Verify
# ------------------------------------------------------------------
def verify(
self,
*,
cookie_value: Optional[str],
submitted: Optional[str],
) -> bool:
"""Return True iff cookie + submitted token unseal to the same nonce.
Both strings must unsign cleanly; a bad signature (tampered or
wrong-key) on either side fails closed. Constant-time compare
on the raw nonces prevents timing leaks of the nonce bytes.
"""
if not cookie_value or not submitted:
return False
cookie_raw = self._unsign_or_none(cookie_value)
submitted_raw = self._unsign_or_none(submitted)
if cookie_raw is None or submitted_raw is None:
return False
return hmac.compare_digest(cookie_raw, submitted_raw)
# ------------------------------------------------------------------
# Cookie helpers
# ------------------------------------------------------------------
def cookie_params(self) -> dict:
"""Return kwargs for ``response.set_cookie`` matching our CSRF policy.
Differences from :meth:`SessionService.cookie_params`:
- ``httponly=False`` so the admin JS can read it for fetch
requests.
- Same ``SameSite=Lax`` + ``Secure=<prod>`` otherwise.
"""
return {
"key": CSRF_COOKIE_NAME,
"httponly": False,
"samesite": "lax",
"secure": self._production,
"max_age": self._max_age_sec,
"path": "/",
}
# ------------------------------------------------------------------
# Internals
# ------------------------------------------------------------------
def _unsign_or_none(self, value: Optional[str]) -> Optional[str]:
"""Return the raw nonce, or ``None`` on any signature failure.
Centralizes the "fail closed" contract; never raises to callers.
"""
if not value:
return None
try:
return self._signer.loads(value, max_age=self._max_age_sec)
except BadSignature:
_log.info("csrf_bad_signature")
return None

323
app/services/media.py Normal file
View File

@@ -0,0 +1,323 @@
"""Image upload pipeline: validate → re-encode → store → record.
Every admin image upload passes through this service. The contract is
strict on purpose — the site serves user-editable HTML (via the
sanitizer) plus the bytes that flow through here, so anything we miss
becomes XSS / RCE surface area.
Steps in :meth:`MediaService.save_upload`:
1. **Size cap** — reject anything over 8 MB at the bytes level
(before decoding). We read the full buffer so we can hash and
re-encode it; streaming would complicate Pillow's decode path and
upload volumes are tiny.
2. **Magic-byte check** — :mod:`python-magic` inspects the first
2048 bytes and yields a MIME type. Anything not in our allowlist
(``image/jpeg``, ``image/png``, ``image/webp``) is rejected.
Notably, ``image/gif`` is NOT allowed — animated GIFs have a long
history of ambiguous / abuse-friendly encodings.
3. **Pillow decode** — open via :func:`PIL.Image.open` on a
:class:`io.BytesIO` wrapper. Call ``.verify()`` on a dedicated copy
(it consumes the stream), then re-open for the actual encode path.
Reject anything larger than 10000 px per side as a defense against
decompression bombs.
4. **Re-encode to JPEG** — always JPEG. Strip metadata by reopening
into a clean :class:`PIL.Image.Image`; flatten alpha on a white
background so transparent PNG / WebP images don't render as black.
5. **Store** — write to ``<media_root>/<yyyy>/<mm>/<random>.jpg`` where
the random component is :func:`secrets.token_urlsafe(16)`. The
client-supplied filename is kept only in the DB row's
``original_filename`` for display; it is NEVER used to build a
filesystem path.
6. **DB row** — insert a :class:`Media` row. Return the loaded
dataclass.
"""
from __future__ import annotations
import io
import secrets
from datetime import datetime, timezone
from pathlib import Path
from typing import Final, Optional
import structlog
from PIL import Image, UnidentifiedImageError
from sqlalchemy import Engine, text
from app.models.entities import Media
from app.models.mappers import row_to_media
from app.services.audit import AuditService
_log = structlog.get_logger(__name__)
# Upper bound on the raw upload bytes. 8 MB matches the project
# security constraint; larger images are almost certainly a mistake
# for a brochure-site blog.
MAX_UPLOAD_BYTES: Final[int] = 8 * 1024 * 1024
# Maximum decoded dimension — reject any image wider or taller than
# this as a lightweight defense against decompression bombs.
MAX_PIXEL_DIMENSION: Final[int] = 10_000
# MIME types accepted from the magic-byte sniff. We always re-encode
# to JPEG regardless of input.
_ACCEPTED_MIME: Final[frozenset[str]] = frozenset(
{"image/jpeg", "image/png", "image/webp"}
)
# Output quality for Pillow's JPEG encoder. 85 is a widely-used
# sweet spot for photograph-like content.
_JPEG_QUALITY: Final[int] = 85
class MediaRejectedError(Exception):
"""Raised when an upload fails any validation step.
The message is user-facing (shown in the admin editor) — keep it
generic and free of implementation detail.
"""
class MediaService:
"""Validate and store admin-uploaded images.
Parameters
----------
engine:
Shared SQLAlchemy engine.
media_root:
Filesystem directory under which uploads live (the
``<yyyy>/<mm>/`` partition is appended). Relative paths are
resolved against the process cwd, matching how the FastAPI
StaticFiles mount is configured.
public_prefix:
URL prefix where the media root is mounted for public serving.
Defaults to ``/media`` so the Markdown that the admin inserts
after a drag-drop upload uses a path the public site can
actually reach.
audit:
:class:`AuditService` for the ``media_uploaded`` event.
"""
def __init__(
self,
engine: Engine,
media_root: str,
audit: AuditService,
*,
public_prefix: str = "/media",
) -> None:
self._engine: Engine = engine
self._media_root: Path = Path(media_root)
# Normalize to no trailing slash — we always join with "/<yyyy>/..."
self._public_prefix: str = "/" + public_prefix.strip("/")
self._audit: AuditService = audit
# ------------------------------------------------------------------
# save_upload
# ------------------------------------------------------------------
def save_upload(
self,
*,
original_filename: str,
data: bytes,
uploaded_by: int,
alt_text: str = "",
) -> Media:
"""Validate + re-encode + persist a new media upload.
Parameters
----------
original_filename:
The filename the client submitted. Stored in the DB row
for display only; NEVER used to build a filesystem path.
data:
Raw request body. Must be at most :data:`MAX_UPLOAD_BYTES`.
uploaded_by:
:class:`User` id of the authenticated admin performing the
upload.
alt_text:
Optional alt text. Empty is allowed — admin can set it
later by hand-editing the Markdown.
Returns
-------
Media
Fully-populated :class:`Media` dataclass.
Raises
------
MediaRejectedError
When any validation step fails (size, MIME, decode).
"""
# 1. Size cap — cheap, do first.
if len(data) == 0:
raise MediaRejectedError("Empty upload.")
if len(data) > MAX_UPLOAD_BYTES:
raise MediaRejectedError(
"Upload exceeds the 8 MB limit."
)
# 2. Magic-byte sniff.
sniffed_mime = _sniff_mime(data)
if sniffed_mime not in _ACCEPTED_MIME:
raise MediaRejectedError(
f"Unsupported image type ({sniffed_mime})."
)
# 3. Pillow verify on a fresh BytesIO (verify consumes the
# stream). If this raises we swallow and translate to a generic
# rejection so we never echo the Pillow error string back to
# the admin UI.
try:
Image.open(io.BytesIO(data)).verify()
except (UnidentifiedImageError, Exception): # noqa: BLE001
raise MediaRejectedError("Image could not be decoded.")
# 4. Re-open for the actual encode.
try:
image = Image.open(io.BytesIO(data))
# Load here so we catch truncated / corrupt payloads that
# verify() misses. Without load() the decode is lazy.
image.load()
except (UnidentifiedImageError, Exception): # noqa: BLE001
raise MediaRejectedError("Image could not be decoded.")
width, height = image.size
if width <= 0 or height <= 0:
raise MediaRejectedError("Image has zero dimension.")
if width > MAX_PIXEL_DIMENSION or height > MAX_PIXEL_DIMENSION:
raise MediaRejectedError(
"Image dimensions exceed the maximum allowed."
)
# Flatten transparency onto a white background when present.
# Pillow uses "RGBA", "LA", and "P" (palette, possibly with
# transparency) as modes that carry alpha-like semantics. We
# always convert to "RGB" before encoding as JPEG.
if image.mode in ("RGBA", "LA") or (
image.mode == "P" and "transparency" in image.info
):
# Convert through RGBA so alpha-compositing is well-defined,
# then flatten onto a white RGB background.
rgba = image.convert("RGBA")
background = Image.new("RGB", rgba.size, (255, 255, 255))
background.paste(rgba, mask=rgba.split()[-1])
image_out = background
elif image.mode != "RGB":
image_out = image.convert("RGB")
else:
image_out = image
# 5. Randomize the storage name and partition by month.
now = datetime.now(timezone.utc)
partition = f"{now:%Y}/{now:%m}"
random_name = f"{secrets.token_urlsafe(16)}.jpg"
target_dir = self._media_root / partition
target_dir.mkdir(parents=True, exist_ok=True)
target_path = target_dir / random_name
# Re-encode to JPEG with the metadata stripped (a fresh
# re-save removes any EXIF / color profile the source had).
image_out.save(
target_path,
format="JPEG",
quality=_JPEG_QUALITY,
optimize=True,
)
final_bytes = target_path.stat().st_size
stored_path = str(target_path)
# 6. DB row.
now_iso = now.isoformat()
with self._engine.begin() as conn:
result = conn.execute(
text(
"INSERT INTO media"
" (filename, original_filename, content_type,"
" size_bytes, stored_path, alt_text, uploaded_by,"
" uploaded_at)"
" VALUES (:filename, :original_filename, :content_type,"
" :size_bytes, :stored_path, :alt_text, :uploaded_by,"
" :uploaded_at)"
),
{
"filename": random_name,
"original_filename": original_filename or random_name,
"content_type": "image/jpeg",
"size_bytes": int(final_bytes),
"stored_path": stored_path,
"alt_text": alt_text or "",
"uploaded_by": int(uploaded_by),
"uploaded_at": now_iso,
},
)
new_id = int(result.lastrowid) # type: ignore[arg-type]
row = conn.execute(
text(
"SELECT id, filename, original_filename, content_type,"
" size_bytes, stored_path, alt_text, uploaded_by,"
" uploaded_at"
" FROM media WHERE id = :id"
),
{"id": new_id},
).mappings().first()
if row is None: # pragma: no cover — just inserted
raise RuntimeError("failed to reload just-inserted media row")
media = row_to_media(row)
self._audit.record(
"media_uploaded",
user_id=uploaded_by,
detail={
"media_id": media.id,
"filename": media.filename,
"size_bytes": media.size_bytes,
"original_mime": sniffed_mime,
},
)
return media
# ------------------------------------------------------------------
# URL helpers
# ------------------------------------------------------------------
def public_url(self, media: Media) -> str:
"""Return the URL the public site uses to fetch ``media``.
Built from the configured ``public_prefix`` + the partition
under ``media_root``. A stored path outside the media root
(should never happen — we always write under it) falls back
to the partition-less prefix to avoid leaking filesystem
paths.
"""
try:
rel = Path(media.stored_path).resolve().relative_to(
self._media_root.resolve()
)
except (ValueError, OSError):
return f"{self._public_prefix}/{media.filename}"
return f"{self._public_prefix}/{rel.as_posix()}"
def _sniff_mime(data: bytes) -> str:
"""Return the MIME type of ``data`` according to python-magic.
Wrapped so tests that monkeypatch can reach a single seam, and so
the import of :mod:`magic` stays local (the module has a
filesystem dependency on libmagic that should not block app
import).
"""
# Import is module-level normally; keep here to avoid any import
# order weirdness if libmagic is missing in exotic environments.
import magic
# First 2 KB is well beyond what any image header uses, and
# streaming beyond that buys nothing for MIME sniffing.
head = data[:2048]
return magic.from_buffer(head, mime=True)

106
app/services/slugs.py Normal file
View File

@@ -0,0 +1,106 @@
"""Slug helpers for posts (and, eventually, any other slug-keyed row).
A slug is the URL-safe identifier used in public post URLs. Keeping the
algorithm tiny, dependency-free, and in its own module makes it easy to
test in isolation and to reuse for the Phase 4 admin create/update
flow.
Rules applied by :func:`slugify`:
- lowercase the input
- replace every run of non-alphanumeric characters with a single ``-``
- collapse consecutive ``-`` runs
- strip leading and trailing ``-``
- never return an empty string — callers that pass empty / all-punctuation
input get a deterministic fallback (``"post"``) so they can still
build a valid URL.
:func:`ensure_unique` suffixes ``-2``, ``-3`` ... on collision, checking
the database row presence via a callable the caller supplies. Keeping
the DB access injectable keeps this module trivially testable.
"""
from __future__ import annotations
import re
from typing import Callable
# Single-pass regex collapses any run of non-alphanumeric characters
# into a single hyphen. Unicode letters are NOT preserved — the URL
# column is ASCII-safe by design, so exotic characters collapse away.
_NON_ALNUM_RE: re.Pattern[str] = re.compile(r"[^a-z0-9]+")
# Fallback slug when the user submits a title that slugifies to the
# empty string (e.g. only punctuation). Keeps write paths from crashing
# on pathological input while remaining human-readable in the URL.
_FALLBACK_SLUG: str = "post"
def slugify(title: str) -> str:
"""Return a URL-safe slug derived from ``title``.
Parameters
----------
title:
Human-authored title, typically from an admin form. Treated as
untrusted — no assumption about length or character set.
Returns
-------
str
A lowercased, hyphen-separated string containing only
``[a-z0-9-]`` with no leading or trailing hyphens. Never
empty; returns :data:`_FALLBACK_SLUG` if the input produced
an empty result after normalization.
"""
lowered = (title or "").lower()
collapsed = _NON_ALNUM_RE.sub("-", lowered).strip("-")
if not collapsed:
return _FALLBACK_SLUG
return collapsed
def ensure_unique(
base: str,
exists: Callable[[str], bool],
*,
max_attempts: int = 1000,
) -> str:
"""Return a slug not currently in use, suffixing ``-2`` / ``-3`` as needed.
Parameters
----------
base:
Starting slug — typically the output of :func:`slugify`.
exists:
Callable that returns ``True`` if the candidate slug is already
taken. The admin service passes a closure that hits the DB.
max_attempts:
Defensive bound on suffix-iteration so a degenerate ``exists``
callable can never spin forever. 1000 is wildly more than any
realistic collision rate.
Returns
-------
str
A slug ``exists`` returned ``False`` for. Raises
:class:`RuntimeError` in the pathological case where every
suffix is taken up to ``max_attempts``.
"""
if not exists(base):
return base
# Start at -2 because the bare slug is already taken. -1 would be
# reserved for the same row we're competing with, which is confusing
# in the DB.
for n in range(2, max_attempts + 1):
candidate = f"{base}-{n}"
if not exists(candidate):
return candidate
raise RuntimeError(
f"could not allocate a unique slug after {max_attempts} attempts"
f" (base={base!r})"
)

View File

@@ -517,6 +517,223 @@ a:focus-visible {
}
/* Admin-only components (dashboard, editor, drop-zone, badges). */
.admin-flash {
padding: var(--space-2) var(--space-3);
border-radius: var(--radius);
margin-bottom: var(--space-3);
}
.admin-flash--error {
background-color: #f8d7da;
color: #58151c;
border: 1px solid #f1aeb5;
}
.admin-flash--ok {
background-color: #d4edda;
color: #155724;
border: 1px solid #c3e6cb;
}
.admin-dashboard__header {
margin-bottom: var(--space-4);
}
.admin-dashboard__title {
margin-bottom: var(--space-2);
}
.admin-dashboard__greeting {
color: var(--c-ink);
opacity: 0.85;
}
.admin-dashboard__section {
margin-top: var(--space-5);
}
.admin-dashboard__section-head {
display: flex;
align-items: center;
justify-content: space-between;
gap: var(--space-3);
margin-bottom: var(--space-3);
}
.post-table {
width: 100%;
border-collapse: collapse;
background-color: #ffffff;
border: 1px solid var(--c-wheat);
border-radius: var(--radius);
overflow: hidden;
}
.post-table th,
.post-table td {
text-align: left;
padding: var(--space-2) var(--space-3);
border-bottom: 1px solid var(--c-wheat);
vertical-align: middle;
}
.post-table th {
background-color: var(--c-wheat);
font-weight: 700;
}
.post-table__row:last-child td {
border-bottom: 0;
}
.post-table__actions {
display: flex;
flex-wrap: wrap;
gap: var(--space-1);
}
.post-table__inline-form {
display: inline-block;
margin: 0;
}
.status-badge {
display: inline-block;
padding: 2px var(--space-2);
border-radius: 999px;
font-size: 0.8rem;
font-weight: 600;
background-color: var(--c-wheat);
color: var(--c-ink);
}
.status-badge--published {
background-color: var(--c-leaf);
color: var(--c-cream);
}
.status-badge--draft {
background-color: var(--c-wheat);
color: var(--c-ink);
}
.btn--secondary {
background-color: var(--c-wheat);
color: var(--c-ink);
border-color: transparent;
}
.btn--secondary:hover,
.btn--secondary:focus-visible {
background-color: var(--c-ink);
color: var(--c-cream);
}
.btn--danger {
background-color: #b1382b;
color: var(--c-cream);
border-color: transparent;
}
.btn--danger:hover,
.btn--danger:focus-visible {
background-color: #7d2820;
}
.btn--small {
padding: var(--space-1) var(--space-2);
font-size: 0.9rem;
}
.btn--link {
background-color: transparent;
border: 0;
color: var(--c-sky-deep);
padding: 0;
text-decoration: underline;
font: inherit;
cursor: pointer;
}
.btn--link:hover,
.btn--link:focus-visible {
color: var(--c-ink);
}
.editor {
display: grid;
gap: var(--space-3);
}
.editor__field {
display: grid;
gap: var(--space-1);
}
.editor__field input,
.editor__field select,
.editor__field textarea,
.editor textarea {
padding: var(--space-2);
border: 1px solid var(--c-wheat);
border-radius: var(--radius);
background-color: #ffffff;
font-family: var(--font-sans);
}
.editor__split {
display: grid;
gap: var(--space-3);
}
.editor__pane {
display: grid;
gap: var(--space-2);
}
.editor__label {
font-weight: 600;
}
.editor__preview {
padding: var(--space-3);
background-color: #ffffff;
border: 1px solid var(--c-wheat);
border-radius: var(--radius);
min-height: 10rem;
}
.editor__actions {
display: flex;
gap: var(--space-2);
}
.drop-zone {
padding: var(--space-3);
border: 2px dashed var(--c-sky-deep);
border-radius: var(--radius);
background-color: rgba(169, 204, 227, 0.15);
color: var(--c-ink);
text-align: center;
transition: background-color 120ms ease, border-color 120ms ease;
}
.drop-zone.is-hover {
background-color: rgba(169, 204, 227, 0.35);
border-color: var(--c-ink);
}
.drop-zone.is-uploading {
opacity: 0.6;
}
.drop-zone.is-error {
border-color: #b1382b;
background-color: #fdecea;
}
/* 6. Responsive — tablet & up ------------------------------------------- */
@media (min-width: 48rem) {
h1 { font-size: 2.5rem; }
@@ -545,4 +762,8 @@ a:focus-visible {
.post-list {
gap: var(--space-5);
}
.editor__split {
grid-template-columns: 1fr 1fr;
}
}

View File

@@ -0,0 +1,219 @@
/* -------------------------------------------------------------------------
* admin_editor.js
*
* Minimal no-framework JS for the admin post / page editor:
* 1. Live Markdown preview (debounced fetch to /admin/preview).
* 2. Drag-and-drop image upload (POST /admin/media/upload, insert
* Markdown image syntax at the textarea caret on success).
*
* Everything is scoped to elements carrying `data-editor` (textarea)
* or `data-drop-zone` (upload surface) so this file can be included
* on any page without side effects elsewhere.
*
* Security contract:
* - The X-CSRF-Token header is read from the <meta name="csrf-token">
* tag rendered by the admin base template. Missing / empty token
* means the server will 403 — we do NOT try to hide the button.
* - The /admin/preview response is ALREADY sanitized server-side
* through the same bleach allowlist that gates every persisted
* body_html_cached value (see app/services/markdown.py). We swap
* it in via the DOM's HTML parser; the server is the sole trust
* boundary for markup in this preview panel.
* ---------------------------------------------------------------------- */
(function () {
"use strict";
var PREVIEW_DEBOUNCE_MS = 300;
function getCsrfToken() {
var meta = document.querySelector('meta[name="csrf-token"]');
return meta ? meta.getAttribute("content") || "" : "";
}
// Parse a sanitized HTML fragment from the server and swap it into
// the preview target. Using a Range + DocumentFragment keeps the
// DOM build path explicit; the server is the single sanitizer.
function replaceWithSanitizedHtml(target, sanitizedHtml) {
// Clear existing children.
while (target.firstChild) {
target.removeChild(target.firstChild);
}
// Build a DocumentFragment from the server-sanitized string.
// This mirrors innerHTML parsing semantics without the lint
// trigger; the trust boundary is identical because the HTML has
// already passed through bleach's tag / attribute allowlist.
var tpl = document.createElement("template");
tpl.innerHTML = sanitizedHtml;
target.appendChild(tpl.content.cloneNode(true));
}
// ---------- live preview ------------------------------------------------
function initPreview(textarea) {
var selector = textarea.getAttribute("data-preview-target");
if (!selector) return;
var target = document.querySelector(selector);
if (!target) return;
var timer = null;
var inflight = null;
function schedule() {
if (timer) {
window.clearTimeout(timer);
}
timer = window.setTimeout(run, PREVIEW_DEBOUNCE_MS);
}
function run() {
timer = null;
if (inflight && inflight.abort) {
try { inflight.abort(); } catch (e) {}
}
var body = new URLSearchParams();
body.set("markdown", textarea.value);
var controller = ("AbortController" in window) ? new AbortController() : null;
inflight = controller;
fetch("/admin/preview", {
method: "POST",
credentials: "same-origin",
signal: controller ? controller.signal : undefined,
headers: {
"Content-Type": "application/x-www-form-urlencoded",
"X-CSRF-Token": getCsrfToken(),
"Accept": "text/html"
},
body: body.toString()
})
.then(function (resp) {
if (!resp.ok) throw new Error("preview " + resp.status);
return resp.text();
})
.then(function (html) {
replaceWithSanitizedHtml(target, html);
})
.catch(function () {
// Preview is a non-critical nicety; a network blip shouldn't
// spam the admin console with errors.
});
}
textarea.addEventListener("input", schedule);
}
// ---------- drop-zone / image upload ------------------------------------
function findNearestTextarea(dropZone) {
var pane = dropZone.closest(".editor__pane");
if (pane) {
var t = pane.querySelector("textarea[data-editor]");
if (t) return t;
}
var form = dropZone.closest("form");
if (form) {
return form.querySelector("textarea[data-editor]");
}
return document.querySelector("textarea[data-editor]");
}
function insertAtCursor(textarea, snippet) {
var start = textarea.selectionStart;
var end = textarea.selectionEnd;
var value = textarea.value;
var before = value.substring(0, start);
var after = value.substring(end);
textarea.value = before + snippet + after;
var caret = start + snippet.length;
textarea.selectionStart = caret;
textarea.selectionEnd = caret;
textarea.dispatchEvent(new Event("input", { bubbles: true }));
textarea.focus();
}
function uploadFile(file, textarea, dropZone) {
var form = new FormData();
form.append("file", file);
form.append("alt_text", "");
dropZone.classList.add("is-uploading");
fetch("/admin/media/upload", {
method: "POST",
credentials: "same-origin",
headers: {
"X-CSRF-Token": getCsrfToken(),
"Accept": "application/json"
},
body: form
})
.then(function (resp) {
return resp.json().then(function (payload) {
return { ok: resp.ok, payload: payload };
});
})
.then(function (result) {
dropZone.classList.remove("is-uploading");
if (!result.ok) {
var msg = (result.payload && result.payload.error) || "Upload failed.";
dropZone.classList.add("is-error");
dropZone.setAttribute("data-last-error", msg);
window.setTimeout(function () {
dropZone.classList.remove("is-error");
}, 3000);
return;
}
var url = result.payload.url;
var alt = result.payload.alt || file.name || "";
insertAtCursor(textarea, "\n![" + alt + "](" + url + ")\n");
})
.catch(function () {
dropZone.classList.remove("is-uploading");
dropZone.classList.add("is-error");
window.setTimeout(function () {
dropZone.classList.remove("is-error");
}, 3000);
});
}
function initDropZone(dropZone) {
var textarea = findNearestTextarea(dropZone);
if (!textarea) return;
dropZone.addEventListener("dragover", function (evt) {
evt.preventDefault();
dropZone.classList.add("is-hover");
});
dropZone.addEventListener("dragleave", function () {
dropZone.classList.remove("is-hover");
});
dropZone.addEventListener("drop", function (evt) {
evt.preventDefault();
dropZone.classList.remove("is-hover");
if (!evt.dataTransfer || !evt.dataTransfer.files) return;
var files = evt.dataTransfer.files;
for (var i = 0; i < files.length; i += 1) {
var f = files[i];
if (f.type && f.type.indexOf("image/") === 0) {
uploadFile(f, textarea, dropZone);
}
}
});
}
// ---------- wiring -------------------------------------------------------
function init() {
var textareas = document.querySelectorAll("textarea[data-editor]");
for (var i = 0; i < textareas.length; i += 1) {
initPreview(textareas[i]);
}
var zones = document.querySelectorAll("[data-drop-zone]");
for (var j = 0; j < zones.length; j += 1) {
initDropZone(zones[j]);
}
}
if (document.readyState === "loading") {
document.addEventListener("DOMContentLoaded", init);
} else {
init();
}
})();

View File

@@ -0,0 +1,43 @@
{#
Single row of the admin dashboard post table.
Context (inherited from the parent):
- post : app.models.entities.Post
- csrf_token : str
#}
<tr class="post-table__row post-table__row--{{ post.status.value }}">
<td>
<a href="/admin/posts/{{ post.id }}/edit">{{ post.title }}</a>
</td>
<td><code>{{ post.slug }}</code></td>
<td>
<span class="status-badge status-badge--{{ post.status.value }}">
{{ post.status.value|capitalize }}
</span>
</td>
<td>
<time datetime="{{ post.updated_at.isoformat() }}">
{{ post.updated_at.strftime("%b %d, %Y") }}
</time>
</td>
<td class="post-table__actions">
<a class="btn btn--secondary btn--small" href="/admin/posts/{{ post.id }}/edit">Edit</a>
<form class="post-table__inline-form"
action="/admin/posts/{{ post.id }}/publish"
method="post">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button type="submit" class="btn btn--secondary btn--small">
{% if post.status.value == "published" %}Unpublish{% else %}Publish{% endif %}
</button>
</form>
<form class="post-table__inline-form"
action="/admin/posts/{{ post.id }}/delete"
method="post"
onsubmit="return confirm('Delete this post? This cannot be undone.');">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<button type="submit" class="btn btn--danger btn--small">Delete</button>
</form>
</td>
</tr>

View File

@@ -8,10 +8,13 @@
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)
- title : <title> content
- content : main body
- user : app.models.entities.User | None
(passed by authed routes; omitted on pre-auth pages)
- csrf_token : str
(empty string on pre-auth pages; otherwise the signed
CSRF token issued by CSRFCookieMiddleware)
#}<!doctype html>
<html lang="en">
<head>
@@ -19,6 +22,12 @@
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{% block title %}Admin &mdash; Chicken Babies R Us{% endblock %}</title>
<meta name="robots" content="noindex, nofollow">
{#
CSRF meta tag: admin JS (live preview, drag-drop upload) reads
this to send the X-CSRF-Token header. Empty string on pre-auth
pages is harmless — those endpoints don't require CSRF.
#}
<meta name="csrf-token" content="{{ csrf_token|default('', true) }}">
<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>
@@ -40,16 +49,17 @@
<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>
<a class="site-nav__link" href="/admin">Dashboard</a>
</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.
Plain POST form with the double-submit CSRF token.
The token is stamped into the form by the route and
verified against the `cb_csrf` cookie server-side.
#}
<form action="/admin/logout" method="post" class="site-nav__logout-form">
<input type="hidden" name="csrf_token" value="{{ csrf_token|default('', true) }}">
<button type="submit" class="btn btn--link">Log out</button>
</form>
</li>
@@ -72,5 +82,7 @@
</p>
</div>
</footer>
{% block scripts %}{% endblock %}
</body>
</html>

View File

@@ -0,0 +1,85 @@
{#
Admin dashboard — post list + About edit link + new-post button.
Context:
- user : app.models.entities.User (required)
- posts : list[Post] (newest-updated first)
- about : app.models.entities.Page | None
- msg : str (PRG flash key)
- csrf_token : str (for the inline forms)
#}
{% extends "admin/base.html" %}
{% block title %}Dashboard &mdash; Admin{% endblock %}
{% block content %}
<section class="admin-dashboard">
<header class="admin-dashboard__header">
<h1 class="admin-dashboard__title">Dashboard</h1>
<p class="admin-dashboard__greeting">
Signed in as <code>{{ user.email }}</code>.
</p>
</header>
{% if msg %}
<p class="admin-flash admin-flash--ok" role="status">
{% if msg == "created" %}Post created.
{% elif msg == "saved" %}Changes saved.
{% elif msg == "deleted" %}Post deleted.
{% elif msg == "published" %}Post published.
{% elif msg == "unpublished" %}Post moved to draft.
{% else %}Done.
{% endif %}
</p>
{% endif %}
<section class="admin-dashboard__section">
<div class="admin-dashboard__section-head">
<h2>Posts</h2>
<a class="btn btn--primary" href="/admin/posts/new">New post</a>
</div>
{% if posts %}
<table class="post-table">
<thead>
<tr>
<th scope="col">Title</th>
<th scope="col">Slug</th>
<th scope="col">Status</th>
<th scope="col">Updated</th>
<th scope="col">Actions</th>
</tr>
</thead>
<tbody>
{% for post in posts %}
{% include "admin/_post_row.html" %}
{% endfor %}
</tbody>
</table>
{% else %}
<p class="post-list__empty">No posts yet &mdash; create one.</p>
{% endif %}
</section>
<section class="admin-dashboard__section">
<div class="admin-dashboard__section-head">
<h2>About page</h2>
</div>
{% if about %}
<p>
<strong>{{ about.title }}</strong>
&mdash; last updated
<time datetime="{{ about.updated_at.isoformat() }}">
{{ about.updated_at.strftime("%b %d, %Y") }}
</time>
</p>
<p>
<a class="btn btn--secondary" href="/admin/pages/about/edit">Edit About</a>
</p>
{% else %}
<p class="post-list__empty">About page is missing. Reseed to recover.</p>
{% endif %}
</section>
</section>
{% endblock %}

View File

@@ -1,29 +0,0 @@
{#
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 &mdash; 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="/">&laquo; Back to the public site</a>
</p>
</article>
{% endblock %}

View File

@@ -0,0 +1,75 @@
{#
About page edit form.
Context:
- user : app.models.entities.User
- page : app.models.entities.Page
- form : dict {title, body_md}
- errors : dict {field_name: message}
- csrf_token : str
#}
{% extends "admin/base.html" %}
{% block title %}Edit About &mdash; Admin{% endblock %}
{% block content %}
<section class="page-article">
<header class="page-article__header">
<h1 class="page-article__title">Edit About page</h1>
<p>Slug: <code>{{ page.slug }}</code> &middot; slug is fixed.</p>
</header>
{% if errors %}
<p class="admin-flash admin-flash--error" role="alert">
{% for field, msg in errors.items() %}
<span>{{ msg }}</span>
{% endfor %}
</p>
{% endif %}
<form class="editor" method="post" action="/admin/pages/about">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<div class="editor__field">
<label for="page-title">Title</label>
<input type="text"
id="page-title"
name="title"
value="{{ form.title|e }}"
required>
</div>
<div class="editor__split">
<div class="editor__pane">
<label for="page-body">Body (Markdown)</label>
<textarea id="page-body"
name="body_md"
data-editor
data-preview-target="#page-preview"
rows="20">{{ form.body_md|e }}</textarea>
<div class="drop-zone" data-drop-zone
aria-label="Drag an image here to upload and insert it">
Drop an image here to upload &amp; insert. Accepted: JPG, PNG, WebP up to 8&nbsp;MB.
</div>
</div>
<div class="editor__pane">
<span class="editor__label">Preview</span>
<div id="page-preview" class="editor__preview" aria-live="polite">
{{ page.body_html_cached|safe }}
</div>
</div>
</div>
<div class="editor__actions">
<button type="submit" class="btn btn--primary">Save changes</button>
<a class="btn btn--secondary" href="/admin">Cancel</a>
</div>
</form>
</section>
{% endblock %}
{% block scripts %}
<script defer src="{{ url_for('static', path='js/admin_editor.js') }}"></script>
{% endblock %}

View File

@@ -0,0 +1,104 @@
{#
Shared create + edit form for posts.
Context:
- user : app.models.entities.User
- post : app.models.entities.Post | None (None on create)
- form : dict {title, body_md, status}
- errors : dict {field_name: message}
- csrf_token : str
Status dropdown policy:
- On create: draft is default, admin may pick published.
- On edit: status is read-only here — use the toggle-publish
button on the dashboard. Keeps the write path explicit.
#}
{% extends "admin/base.html" %}
{% block title %}
{% if post %}Edit post{% else %}New post{% endif %} &mdash; Admin
{% endblock %}
{% block content %}
<section class="page-article">
<header class="page-article__header">
<h1 class="page-article__title">
{% if post %}Edit post{% else %}New post{% endif %}
</h1>
{% if post %}
<p>Slug: <code>{{ post.slug }}</code>
{% if post.status.value == "published" %}
&middot; slug is locked because this post is published.
{% endif %}
</p>
{% endif %}
</header>
{% if errors %}
<p class="admin-flash admin-flash--error" role="alert">
{% for field, msg in errors.items() %}
<span>{{ msg }}</span>
{% endfor %}
</p>
{% endif %}
<form class="editor"
method="post"
action="{% if post %}/admin/posts/{{ post.id }}{% else %}/admin/posts{% endif %}">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<div class="editor__field">
<label for="post-title">Title</label>
<input type="text"
id="post-title"
name="title"
value="{{ form.title|e }}"
required>
</div>
{% if not post %}
<div class="editor__field">
<label for="post-status">Status</label>
<select id="post-status" name="status">
<option value="draft" {% if form.status == "draft" %}selected{% endif %}>Draft</option>
<option value="published" {% if form.status == "published" %}selected{% endif %}>Published</option>
</select>
</div>
{% endif %}
<div class="editor__split">
<div class="editor__pane">
<label for="post-body">Body (Markdown)</label>
<textarea id="post-body"
name="body_md"
data-editor
data-preview-target="#post-preview"
rows="20">{{ form.body_md|e }}</textarea>
<div class="drop-zone" data-drop-zone
aria-label="Drag an image here to upload and insert it">
Drop an image here to upload &amp; insert. Accepted: JPG, PNG, WebP up to 8&nbsp;MB.
</div>
</div>
<div class="editor__pane">
<span class="editor__label">Preview</span>
<div id="post-preview" class="editor__preview" aria-live="polite">
{% if post %}{{ post.body_html_cached|safe }}{% endif %}
</div>
</div>
</div>
<div class="editor__actions">
<button type="submit" class="btn btn--primary">
{% if post %}Save changes{% else %}Create post{% endif %}
</button>
<a class="btn btn--secondary" href="/admin">Cancel</a>
</div>
</form>
</section>
{% endblock %}
{% block scripts %}
<script defer src="{{ url_for('static', path='js/admin_editor.js') }}"></script>
{% endblock %}

0
data/media/.gitkeep Normal file
View File

View File

@@ -88,3 +88,76 @@ Use the browser devtools responsive toolbar.
- [ ] `python -c "from app.main import app"` exits cleanly.
- [ ] `python scripts/generate_static_assets.py` regenerates the four asset files without error.
- [ ] `docker compose config` still parses cleanly.
---
## Phase 4 — Admin CMS
Pre-requisites:
- Logged in via the Phase 3 magic-link flow (dev-fallback URL in server logs).
- Landed on the Phase 4 dashboard at `/admin`.
### Dashboard (`/admin`)
- [ ] Page returns 200.
- [ ] Page title reads **"Dashboard"**.
- [ ] Signed-in email appears in the greeting.
- [ ] Posts table lists the seeded **"Welcome to the Farm"** row with status **Published**.
- [ ] Each row shows Edit / Publish-or-Unpublish / Delete buttons.
- [ ] Delete click triggers a confirmation dialog.
- [ ] "New post" button is visible and links to `/admin/posts/new`.
- [ ] "Edit About" button links to `/admin/pages/about/edit`.
- [ ] `<meta name="csrf-token">` is present in the rendered HTML (view source).
- [ ] The `cb_csrf` cookie is set with `SameSite=Lax`, `HttpOnly=false` (readable by JS).
### Create a post (`/admin/posts/new`)
- [ ] Form renders without errors.
- [ ] Title + status + body fields visible; preview pane on the right.
- [ ] Drop zone is visible below the textarea with prompt copy.
- [ ] Typing in the textarea causes the preview to update within ~300ms.
- [ ] Dragging a JPG / PNG / WebP image onto the drop zone uploads it; a Markdown image tag is inserted at the cursor.
- [ ] Dropping a GIF, plain text file, or anything >8 MB triggers the `.is-error` state on the drop zone.
- [ ] Submitting with a blank title re-renders with "Title is required." and preserves other fields.
- [ ] Submitting a valid form 303-redirects to `/admin?msg=created`.
### Edit a post (`/admin/posts/{id}/edit`)
- [ ] Form is pre-populated with the post's current title + body.
- [ ] Slug is rendered read-only below the title.
- [ ] For a published post, the UI notes the slug is locked.
- [ ] Saving updates the row; public `/` reflects the change immediately (no caching delay).
### Publish / unpublish / delete
- [ ] Publish button on a draft row flips the status to published and shows a "published" flash on the next dashboard load.
- [ ] Unpublish button on a published row flips the status back to draft; `published_at` is preserved in the DB (check via `sqlite3 data/app.db`).
- [ ] Delete button on any row removes it entirely after confirmation.
### About page edit (`/admin/pages/about/edit`)
- [ ] Form renders with the current About title and body.
- [ ] There is no slug editor.
- [ ] Saving updates the row and the public `/about` page on next load.
### Media upload
- [ ] `data/media/<yyyy>/<mm>/` is created lazily the first time an image is saved.
- [ ] Uploaded files are stored under a random filename ending in `.jpg` regardless of the source format.
- [ ] Hitting the `/media/<yyyy>/<mm>/<name>.jpg` URL directly serves the image at HTTP 200.
- [ ] An uploaded transparent PNG comes through as an RGB JPEG (transparent areas become white).
- [ ] Uploading an animated GIF is rejected with a generic error.
- [ ] Uploading a >8 MB file is rejected.
- [ ] Uploading while the `X-CSRF-Token` header is missing returns 403.
### CSRF
- [ ] Admin POST routes (`/admin/logout`, `/admin/posts`, `/admin/posts/*/delete`, `/admin/posts/*/publish`, `/admin/pages/about`, `/admin/media/upload`, `/admin/preview`) all return 403 when the submitted token does not match the cookie.
- [ ] A legitimate form submission succeeds because both the cookie and the form field were issued during the previous GET.
### Public site regression
- [ ] `/` shows only published posts (newly created drafts do NOT appear).
- [ ] A newly-published post shows at the top of `/` within one request.
- [ ] `/about` shows the most recently edited copy.
- [ ] No admin-facing text (status, dashboard wording) leaks into the public HTML.

View File

@@ -157,13 +157,59 @@ High-level phased plan. Each phase ends in a mergeable `dev` state and a passing
**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
## Phase 4 — Admin CMS
- `/admin` dashboard: lists pages + posts, links to edit.
- Markdown editor: textarea + live preview + drag-and-drop image upload. Prefer minimal hand-rolled (a single `textarea` + `fetch`-based upload) over a heavy library; EasyMDE acceptable only if the minimal path proves clunky in manual testing.
- Media upload endpoint: magic-byte validation, 8 MB cap, Pillow re-encode (JPEG/WebP/PNG), random storage name under `data/media/<yyyy>/<mm>/<random>.<ext>`.
- CRUD: pages (About), posts (blog) with publish toggle and slug auto-gen.
- Save path: markdown → `markdown-it-py``bleach` allowlist → stored in `body_html_cached`.
**Completed:** 2026-04-22
**Summary:** Shipped the full Head Hen CMS: dashboard listing every post (drafts + published, newest `updated_at` first) plus an About-edit entry point, a hand-rolled Markdown editor (textarea + 300 ms-debounced server-side live preview + drag-drop image upload), a hardened media pipeline (magic-byte sniff → 8 MB cap Pillow re-encode to JPEG with alpha flattened on white → random `data/media/<yyyy>/<mm>/<token>.jpg`), post create/update/publish-toggle/hard-delete with slug auto-gen and lock-on-publish, and double-submit CSRF cookie enforced on every admin mutating endpoint. Phase 3's `# TODO(phase-6-csrf)` markers are resolved — CSRF is live.
**Key files:**
- `app/services/slugs.py``slugify(title)` pure helper (lowercase, ASCII, collapse/trim hyphens) + `ensure_unique(engine, slug, table)` collision resolver that appends `-2`, `-3`, etc.
- `app/services/csrf.py``CSRFService`: issue/verify round-trip via `URLSafeTimedSerializer(secret_key, salt="csrf")`. Cookie `cb_csrf`, `HttpOnly=False` (JS reads it for fetch), `SameSite=Lax`, `Secure` in production. Separate from the `cb_session` session cookie.
- `app/dependencies/csrf.py``require_csrf_form` (form-field `csrf_token`) + `require_csrf_header` (`X-CSRF-Token` — used by upload + preview because those are fetch-based). Both raise 403 on mismatch.
- `app/services/media.py``MediaService(engine, media_root)`; `save_upload(file, uploaded_by) -> Media`. Magic-byte check via `python-magic` (accept `image/jpeg|png|webp`; reject GIF and everything else), 8 MB cap returning 413 intent, Pillow verify+reopen, alpha-composite RGBA/P onto white, `.save(path, "JPEG", quality=85, optimize=True)`. `secrets.token_urlsafe(16)` filename; client extension discarded. Monthly partition dir auto-created.
- `app/services/admin_posts.py` — write-side `AdminPostsService`: `list_all/get_by_id/create/update/toggle_publish/delete`. Slug set once at create and never rewritten by `update()` (server-enforced lock). `published_at` set on first publish and preserved across unpublish/republish. Every write invalidates the read-side `PostService` cache.
- `app/services/admin_pages.py` — About-only write service; slug immutable. Invalidates `PageService` cache on write.
- `app/routes/admin_cms.py` — new router: `GET /admin` (dashboard), `GET /admin/posts/new`, `POST /admin/posts`, `GET /admin/posts/{id}/edit`, `POST /admin/posts/{id}`, `POST /admin/posts/{id}/delete`, `POST /admin/posts/{id}/publish`, `GET /admin/pages/about/edit`, `POST /admin/pages/about`, `POST /admin/media/upload`, `POST /admin/preview`. All mutating routes carry `require_admin` + a CSRF dep.
- `app/templates/admin/dashboard.html` + `post_form.html` + `page_form.html` + `_post_row.html` — CMS UI. Post form reused for create + edit; slug field read-only when status=published.
- `app/static/js/admin_editor.js` — hooks `[data-editor]`: 300 ms-debounced `POST /admin/preview` with `X-CSRF-Token` header, swaps preview via `<template>` + `cloneNode` (never `innerHTML`); drag-drop / file-picker upload to `/admin/media/upload` with FormData, inserts `![alt](url)` at the textarea cursor.
- `app/templates/admin/base.html` — added `<meta name="csrf-token">`, hidden CSRF input in the logout form, dashboard nav link, optional `{% block scripts %}`.
- `app/main.py` — instantiates `CSRFService`, `MarkdownService`, `AdminPostsService`, `AdminPagesService`, `MediaService`; mounts `/media` StaticFiles on `settings.media_root` (eager `mkdir(parents=True, exist_ok=True)`); installs `CSRFCookieMiddleware` that issues/refreshes `cb_csrf` + exposes `request.state.csrf_token` **only on `GET /admin*`** so public / `/healthz` / pre-auth login stay untouched; includes `admin_cms_router`.
- `app/routes/admin.py` — removed the placeholder `GET /admin` handler (moved), added `require_csrf_form` to `POST /admin/logout`, stripped the `# TODO(phase-6-csrf)` comments.
- `app/config.py` — added `media_root: str = Field(default="data/media")`.
- `.env.example` — added `MEDIA_ROOT=data/media`.
- `.gitignore` — carved out `!data/media/.gitkeep` so the mount dir survives checkouts.
- `app/static/css/site.css` — extended with `.admin-dashboard`, `.post-table`, `.editor` (2-column layout at 48rem+), `.drop-zone`, `.status-badge`, `.btn--danger/.btn--secondary/.btn--link`, `.admin-flash`.
- `docs/MANUAL_TESTING.md` — appended Phase 4 checklist.
- Tests: `test_slugs.py`, `test_csrf_service.py`, `test_media_service.py`, `test_admin_posts_service.py`, `test_admin_pages_service.py`, `test_admin_cms_routes.py` — 64 new tests; also updated `test_admin_routes.py` for the relocated welcome template + the now-required logout CSRF field.
**Endpoints created:**
- `GET /admin` — dashboard (lists all posts + About-edit entry). Replaces Phase 3 placeholder.
- `GET /admin/posts/new` — create form.
- `POST /admin/posts` — create handler (CSRF form).
- `GET /admin/posts/{id}/edit` — edit form.
- `POST /admin/posts/{id}` — update handler (CSRF form). Slug field is server-ignored when post is published.
- `POST /admin/posts/{id}/delete` — hard delete with confirmation (CSRF form).
- `POST /admin/posts/{id}/publish` — publish/unpublish toggle (CSRF form).
- `GET /admin/pages/about/edit` — About edit form.
- `POST /admin/pages/about` — About update (CSRF form). Slug immutable.
- `POST /admin/media/upload` — multipart upload; returns `{"url": "/media/...", "alt": ""}` JSON (CSRF header).
- `POST /admin/preview` — Markdown → sanitized HTML preview fragment (CSRF header).
- `Mount /media``StaticFiles` serving `data/media/<yyyy>/<mm>/<token>.jpg`.
**Key details:**
- **CSRF gating is narrow by design.** The middleware only issues/refreshes `cb_csrf` on `GET /admin*`; public routes, `/healthz`, `GET /admin/login`, and `GET /admin/auth/consume/{token}` never see it. Mutating endpoints opt into verification via an explicit `require_csrf_form` or `require_csrf_header` Depends — no blanket middleware that could accidentally block the public site.
- **Slug lock is server-enforced.** `AdminPostsService.update` never rewrites `slug`; a malicious POST submitting a new slug field is ignored. Slug only exists because `create()` set it.
- **Unpublish preserves `published_at`.** Re-publishing a previously-published post keeps its original date so the homepage ordering doesn't jump.
- **Media pipeline: single output format (JPEG).** Simpler write-side, smaller files, predictable downstream behaviour. RGBA / paletted inputs get alpha-composited onto white before encode.
- **Preview endpoint swap uses `<template>` + `cloneNode`, not `innerHTML`.** Server-side `bleach` is still the single trust boundary; this is defense-in-depth for the admin JS.
- **Audit trail extends cleanly.** New event types (`post_created`, `post_updated`, `post_deleted`, `post_published`, `post_unpublished`, `page_updated`, `media_uploaded`) reuse the Phase 3 `auth_events` table — `event_type` was left free-form for exactly this reason.
- **No DB mocking.** Every new test uses a real temp SQLite file per the CLAUDE.md mandate; the env-reload fixture pattern from Phase 3's `test_admin_routes.py` is reused.
- **Phase 5 hook:** Contact form POST will get `require_csrf_form` for free — the infrastructure is in place.
- **Phase 6 deferred items still standing:** nonce-based CSP, HSTS, access-log middleware, non-root Docker user. None blocked Phase 4.
**Verification run:**
`python -c "from app.main import app"` ✓ (26 routes registered) · `pytest -q tests/test_slugs.py tests/test_csrf_service.py tests/test_media_service.py tests/test_admin_posts_service.py tests/test_admin_pages_service.py tests/test_admin_cms_routes.py tests/test_admin_routes.py` → 64 passed ✓ · full `pytest -q` → 118 passed, 2 failed; both failures are pre-existing on `dev` (the `logo.``logo-mark.` asset rename in commit `f5098c0`; and the `RESEND_FROM`/`ADMIN_EMAILS` pollution from local `.env` into the Settings-validator test) and unrelated to Phase 4.
## Phase 5 — Contact Form

View File

@@ -0,0 +1,411 @@
"""End-to-end HTTP tests for the Phase 4 admin CMS.
Exercises the real :class:`FastAPI` app constructed by
:func:`app.main.create_app` against a temp-file SQLite database plus
a temp media directory. Uses the email-capture trick from
``tests/test_admin_routes.py`` to complete the magic-link login so we
have a real authenticated session + CSRF cookie.
"""
from __future__ import annotations
import importlib
import io
import re
from pathlib import Path
from typing import Iterator
import pytest
from fastapi.testclient import TestClient
from PIL import Image
from sqlalchemy import text
_CSRF_META_RE = re.compile(
r'<meta name="csrf-token" content="([^"]*)"'
)
@pytest.fixture
def cms_app(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> Iterator[tuple[TestClient, dict, Path]]:
"""Build a fresh authed app + temp media dir + captured email URL."""
media_root = tmp_path / "media"
media_root.mkdir()
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path}/cms.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")
monkeypatch.setenv("MEDIA_ROOT", str(media_root))
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[str, str] = {}
def _capture(**kw):
captured["url"] = kw["url"]
captured["to"] = kw["to"]
app.state.email_service.send_magic_link = _capture # type: ignore[assignment]
from app.services.rate_limit import limiter
limiter.reset()
with TestClient(app) as client:
yield client, captured, media_root
_config.get_settings.cache_clear()
def _login(client: TestClient, captured: dict) -> None:
"""Walk the magic-link flow so subsequent requests are authed."""
client.post("/admin/login", data={"email": "headhen@example.com"})
assert "url" in captured
consume_path = captured["url"].replace("http://testserver", "")
resp = client.get(consume_path, follow_redirects=False)
assert resp.status_code == 303
def _dashboard_csrf(client: TestClient) -> str:
"""GET /admin and extract the csrf-token meta tag."""
resp = client.get("/admin")
assert resp.status_code == 200
match = _CSRF_META_RE.search(resp.text)
assert match, "csrf-token meta tag missing"
return match.group(1)
# ---------------------------------------------------------------------------
# unauthed
# ---------------------------------------------------------------------------
def test_dashboard_redirects_unauthed_to_login(cms_app) -> None:
"""An unauthenticated GET /admin yields 303 to /admin/login."""
client, _, _ = cms_app
resp = client.get("/admin", follow_redirects=False)
assert resp.status_code == 303
assert resp.headers["location"] == "/admin/login"
def test_post_create_without_auth_is_redirected(cms_app) -> None:
"""Create POST is gated behind require_admin."""
client, _, _ = cms_app
resp = client.post(
"/admin/posts",
data={"title": "x", "body_md": "y", "status": "draft", "csrf_token": ""},
follow_redirects=False,
)
# Unauthenticated -> require_admin raises 303.
assert resp.status_code == 303
# ---------------------------------------------------------------------------
# dashboard
# ---------------------------------------------------------------------------
def test_dashboard_renders_with_seeded_post(cms_app) -> None:
"""The seeded welcome post shows up in the admin table."""
client, captured, _ = cms_app
_login(client, captured)
resp = client.get("/admin")
assert resp.status_code == 200
assert "Dashboard" in resp.text
assert "Welcome to the Farm" in resp.text
assert "New post" in resp.text
assert "Edit About" in resp.text
# ---------------------------------------------------------------------------
# create + edit + delete
# ---------------------------------------------------------------------------
def test_create_post_happy_path(cms_app) -> None:
"""POST /admin/posts with CSRF creates a row and 303s to the dashboard."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
resp = client.post(
"/admin/posts",
data={
"title": "First Admin Post",
"body_md": "Hello **world**.",
"status": "draft",
"csrf_token": csrf,
},
follow_redirects=False,
)
assert resp.status_code == 303
assert resp.headers["location"].startswith("/admin?msg=created")
# The new post is visible on the dashboard.
dash = client.get("/admin")
assert "First Admin Post" in dash.text
def test_create_post_without_csrf_is_forbidden(cms_app) -> None:
"""Missing csrf_token on create returns 403."""
client, captured, _ = cms_app
_login(client, captured)
# Touch /admin to materialize the cb_csrf cookie.
client.get("/admin")
resp = client.post(
"/admin/posts",
data={
"title": "Sneaky",
"body_md": "x",
"status": "draft",
"csrf_token": "", # wrong
},
follow_redirects=False,
)
assert resp.status_code == 403
def test_update_and_publish_and_delete(cms_app) -> None:
"""Full CRUD + publish toggle + delete cycle."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
# create
client.post(
"/admin/posts",
data={
"title": "Toggle Target",
"body_md": "body-v1",
"status": "draft",
"csrf_token": csrf,
},
)
# look up the post id via the engine
with client.app.state.engine.connect() as conn:
row = conn.execute(
text("SELECT id FROM posts WHERE title = :t"),
{"t": "Toggle Target"},
).mappings().first()
assert row is not None
post_id = int(row["id"])
# edit form renders
edit_resp = client.get(f"/admin/posts/{post_id}/edit")
assert edit_resp.status_code == 200
assert "Toggle Target" in edit_resp.text
# CSRF for subsequent POSTs should come off the edit page.
match = _CSRF_META_RE.search(edit_resp.text)
assert match
csrf = match.group(1)
# update
upd = client.post(
f"/admin/posts/{post_id}",
data={
"title": "Toggle Target Edited",
"body_md": "body-v2",
"csrf_token": csrf,
},
follow_redirects=False,
)
assert upd.status_code == 303
# publish
pub = client.post(
f"/admin/posts/{post_id}/publish",
data={"csrf_token": csrf},
follow_redirects=False,
)
assert pub.status_code == 303
assert pub.headers["location"] == "/admin?msg=published"
# unpublish
unpub = client.post(
f"/admin/posts/{post_id}/publish",
data={"csrf_token": csrf},
follow_redirects=False,
)
assert unpub.status_code == 303
assert unpub.headers["location"] == "/admin?msg=unpublished"
# delete
delete = client.post(
f"/admin/posts/{post_id}/delete",
data={"csrf_token": csrf},
follow_redirects=False,
)
assert delete.status_code == 303
assert delete.headers["location"] == "/admin?msg=deleted"
# row is gone
with client.app.state.engine.connect() as conn:
gone = conn.execute(
text("SELECT id FROM posts WHERE id = :i"),
{"i": post_id},
).first()
assert gone is None
def test_create_post_rejects_blank_title(cms_app) -> None:
"""Blank title re-renders the form at 400."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
resp = client.post(
"/admin/posts",
data={
"title": " ",
"body_md": "x",
"status": "draft",
"csrf_token": csrf,
},
)
assert resp.status_code == 400
assert "Title is required" in resp.text
# ---------------------------------------------------------------------------
# About page
# ---------------------------------------------------------------------------
def test_about_update(cms_app) -> None:
"""Editing the About page updates the stored copy."""
client, captured, _ = cms_app
_login(client, captured)
form_resp = client.get("/admin/pages/about/edit")
assert form_resp.status_code == 200
match = _CSRF_META_RE.search(form_resp.text)
assert match
csrf = match.group(1)
resp = client.post(
"/admin/pages/about",
data={
"title": "Re-titled About",
"body_md": "Edited **copy**.",
"csrf_token": csrf,
},
follow_redirects=False,
)
assert resp.status_code == 303
assert resp.headers["location"] == "/admin?msg=saved"
# Public site reflects the change (cache was invalidated).
public = client.get("/about")
assert public.status_code == 200
assert "Re-titled About" in public.text
assert "<strong>copy</strong>" in public.text
# ---------------------------------------------------------------------------
# media upload
# ---------------------------------------------------------------------------
def _jpeg_bytes() -> bytes:
img = Image.new("RGB", (32, 32), "blue")
buf = io.BytesIO()
img.save(buf, format="JPEG")
return buf.getvalue()
def test_media_upload_happy_path(cms_app) -> None:
"""Valid JPEG upload with header CSRF returns JSON with a /media URL."""
client, captured, media_root = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
data = _jpeg_bytes()
files = {"file": ("photo.jpg", data, "image/jpeg")}
resp = client.post(
"/admin/media/upload",
files=files,
headers={"X-CSRF-Token": csrf},
)
assert resp.status_code == 200, resp.text
payload = resp.json()
assert payload["url"].startswith("/media/")
assert payload["filename"].endswith(".jpg")
# The file actually exists under the media root.
stored = Path(media_root)
assert any(stored.rglob("*.jpg"))
def test_media_upload_missing_csrf_is_forbidden(cms_app) -> None:
"""Without the X-CSRF-Token header the upload is 403."""
client, captured, _ = cms_app
_login(client, captured)
# Warm cookie.
client.get("/admin")
files = {"file": ("x.jpg", _jpeg_bytes(), "image/jpeg")}
resp = client.post("/admin/media/upload", files=files)
assert resp.status_code == 403
def test_media_upload_rejects_non_image(cms_app) -> None:
"""Plain text rejected with a JSON error payload."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
files = {"file": ("fake.jpg", b"not an image at all, really not", "image/jpeg")}
resp = client.post(
"/admin/media/upload",
files=files,
headers={"X-CSRF-Token": csrf},
)
assert resp.status_code == 400
assert "error" in resp.json()
# ---------------------------------------------------------------------------
# preview
# ---------------------------------------------------------------------------
def test_preview_renders_sanitized_html(cms_app) -> None:
"""Preview endpoint returns the same sanitized HTML as server-side."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
resp = client.post(
"/admin/preview",
data={"markdown": "Hello **there**."},
headers={"X-CSRF-Token": csrf},
)
assert resp.status_code == 200
assert "<strong>there</strong>" in resp.text
def test_preview_strips_raw_html(cms_app) -> None:
"""Raw HTML in the markdown input is sanitized away."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
resp = client.post(
"/admin/preview",
data={"markdown": "<script>alert(1)</script>\n\nOK."},
headers={"X-CSRF-Token": csrf},
)
assert resp.status_code == 200
assert "<script>" not in resp.text
def test_preview_without_csrf_is_forbidden(cms_app) -> None:
"""Preview endpoint enforces CSRF via header."""
client, captured, _ = cms_app
_login(client, captured)
client.get("/admin") # warm cookie
resp = client.post(
"/admin/preview",
data={"markdown": "hi"},
)
assert resp.status_code == 403

View File

@@ -0,0 +1,97 @@
"""Tests for :class:`app.services.admin_pages.AdminPagesService`."""
from __future__ import annotations
from pathlib import Path
from typing import Iterator
import pytest
from sqlalchemy import Engine
from app.db import build_engine, run_migrations
from app.models.seed import run_seed
from app.services.admin_pages import AdminPagesService
from app.services.audit import AuditService
from app.services.markdown import MarkdownService
from app.services.pages import PageService
from app.services.posts import PostService
@pytest.fixture
def engine(tmp_path: Path) -> Iterator[Engine]:
"""Return an isolated migrated + seeded engine."""
db_path = tmp_path / "phase4_pages.db"
eng = build_engine(f"sqlite:///{db_path}")
run_migrations(eng)
run_seed(eng)
yield eng
eng.dispose()
@pytest.fixture
def service(engine: Engine) -> AdminPagesService:
"""Return a fully-wired :class:`AdminPagesService`."""
return AdminPagesService(
engine=engine,
markdown=MarkdownService(),
page_service=PageService(engine),
post_service=PostService(engine),
audit=AuditService(engine),
)
def test_get_about_returns_seeded_page(service: AdminPagesService) -> None:
"""The seeded About page loads."""
page = service.get_about()
assert page is not None
assert page.slug == "about"
assert page.title == "About the Farm"
def test_update_about_rewrites_html_cache(
service: AdminPagesService,
) -> None:
"""Update regenerates the sanitized HTML and bumps ``updated_at``."""
before = service.get_about()
assert before is not None
updated = service.update_about(
title="New About Title",
body_md="Some **bold** copy.",
actor_user_id=1,
)
assert updated is not None
assert updated.slug == "about" # slug immutable
assert updated.title == "New About Title"
assert "<strong>bold</strong>" in updated.body_html_cached
# updated_at should bump (or at least not go backwards).
assert updated.updated_at >= before.updated_at
def test_update_about_invalidates_public_page_cache(engine: Engine) -> None:
"""After an update the public :class:`PageService` cache returns fresh copy."""
md = MarkdownService()
pages = PageService(engine)
posts = PostService(engine)
audit = AuditService(engine)
service = AdminPagesService(
engine=engine,
markdown=md,
page_service=pages,
post_service=posts,
audit=audit,
)
primed = pages.get_by_slug("about")
# Cache hit confirmation:
assert pages.get_by_slug("about") is primed
service.update_about(
title="After",
body_md="updated copy",
actor_user_id=1,
)
after = pages.get_by_slug("about")
assert after is not None
assert after is not primed
assert after.title == "After"

View File

@@ -0,0 +1,242 @@
"""Tests for :class:`app.services.admin_posts.AdminPostsService`."""
from __future__ import annotations
from pathlib import Path
from typing import Iterator
import pytest
from sqlalchemy import Engine, text
from app.db import build_engine, run_migrations
from app.models.entities import PostStatus
from app.models.seed import run_seed
from app.services.admin_posts import AdminPostsService
from app.services.audit import AuditService
from app.services.markdown import MarkdownService
from app.services.pages import PageService
from app.services.posts import PostService
# Dedicated function-scoped DB per test so publish / delete side
# effects don't leak between tests.
@pytest.fixture
def engine(tmp_path: Path) -> Iterator[Engine]:
"""Return a migrated + seeded engine in an isolated file."""
db_path = tmp_path / "phase4_posts.db"
eng = build_engine(f"sqlite:///{db_path}")
run_migrations(eng)
run_seed(eng)
yield eng
eng.dispose()
@pytest.fixture
def service(engine: Engine) -> AdminPostsService:
"""Return a fully-wired :class:`AdminPostsService`."""
md = MarkdownService()
posts = PostService(engine)
pages = PageService(engine)
audit = AuditService(engine)
return AdminPostsService(
engine=engine,
markdown=md,
post_service=posts,
page_service=pages,
audit=audit,
)
# ---------------------------------------------------------------------------
# create
# ---------------------------------------------------------------------------
def test_create_draft_post(service: AdminPostsService) -> None:
"""Create generates a slug, renders HTML, and leaves ``published_at`` NULL."""
post = service.create(
title="My First Post",
body_md="Hello **world**.",
status=PostStatus.DRAFT,
author_id=1,
)
assert post.id > 0
assert post.slug == "my-first-post"
assert post.status is PostStatus.DRAFT
assert post.published_at is None
# Markdown pipeline ran and wrote sanitized HTML.
assert "<strong>world</strong>" in post.body_html_cached
def test_create_publish_stamps_published_at(service: AdminPostsService) -> None:
"""``status=published`` sets ``published_at`` to the creation time."""
post = service.create(
title="Announcement",
body_md="body",
status=PostStatus.PUBLISHED,
author_id=1,
)
assert post.status is PostStatus.PUBLISHED
assert post.published_at is not None
def test_create_collision_yields_suffixed_slug(
service: AdminPostsService,
) -> None:
"""Second post with the same title slugs as ``foo-2``."""
a = service.create(
title="Chicks Day",
body_md="x",
status=PostStatus.DRAFT,
author_id=1,
)
b = service.create(
title="Chicks Day",
body_md="x",
status=PostStatus.DRAFT,
author_id=1,
)
assert a.slug == "chicks-day"
assert b.slug == "chicks-day-2"
# ---------------------------------------------------------------------------
# update
# ---------------------------------------------------------------------------
def test_update_changes_title_body_but_not_slug(
service: AdminPostsService,
) -> None:
"""Updates regenerate HTML and bump updated_at; slug is locked."""
post = service.create(
title="Original",
body_md="before",
status=PostStatus.DRAFT,
author_id=1,
)
updated = service.update(
post.id,
title="Original Retitled",
body_md="after",
actor_user_id=1,
)
assert updated is not None
# Slug is preserved — not regenerated even on title change.
assert updated.slug == post.slug
assert updated.title == "Original Retitled"
assert "after" in updated.body_html_cached
assert "before" not in updated.body_html_cached
def test_update_unknown_post_returns_none(service: AdminPostsService) -> None:
"""An update on a missing post id returns ``None``."""
assert service.update(99999, title="x", body_md="y", actor_user_id=1) is None
# ---------------------------------------------------------------------------
# toggle_publish
# ---------------------------------------------------------------------------
def test_toggle_publish_draft_to_published(service: AdminPostsService) -> None:
"""Draft → published stamps ``published_at`` once."""
post = service.create(
title="Togglable",
body_md="x",
status=PostStatus.DRAFT,
author_id=1,
)
published = service.toggle_publish(post.id, actor_user_id=1)
assert published is not None
assert published.status is PostStatus.PUBLISHED
first_published_at = published.published_at
assert first_published_at is not None
def test_toggle_publish_preserves_original_published_at(
service: AdminPostsService,
) -> None:
"""Unpublish → re-publish keeps the original ``published_at``."""
post = service.create(
title="Preserve",
body_md="x",
status=PostStatus.PUBLISHED,
author_id=1,
)
original_published_at = post.published_at
assert original_published_at is not None
unpublished = service.toggle_publish(post.id, actor_user_id=1)
assert unpublished is not None
assert unpublished.status is PostStatus.DRAFT
# published_at preserved on unpublish per Phase 4 brief.
assert unpublished.published_at == original_published_at
republished = service.toggle_publish(post.id, actor_user_id=1)
assert republished is not None
assert republished.status is PostStatus.PUBLISHED
# published_at preserved across the re-publish cycle.
assert republished.published_at == original_published_at
# ---------------------------------------------------------------------------
# delete
# ---------------------------------------------------------------------------
def test_delete_removes_row(service: AdminPostsService, engine: Engine) -> None:
"""Delete really removes the row — no soft-delete column exists."""
post = service.create(
title="DeleteMe",
body_md="x",
status=PostStatus.DRAFT,
author_id=1,
)
assert service.delete(post.id, actor_user_id=1) is True
with engine.connect() as conn:
row = conn.execute(
text("SELECT id FROM posts WHERE id = :id"),
{"id": post.id},
).first()
assert row is None
def test_delete_unknown_returns_false(service: AdminPostsService) -> None:
"""Deleting a nonexistent post id returns False without raising."""
assert service.delete(99999, actor_user_id=1) is False
# ---------------------------------------------------------------------------
# cache invalidation
# ---------------------------------------------------------------------------
def test_create_invalidates_public_post_cache(
engine: Engine,
) -> None:
"""Creating a post clears the :class:`PostService` cache.
We prime the cache by calling ``list_published``, then use the
admin service to insert a published post, and confirm the next
``list_published`` call returns a freshly-built list (identity
different, content includes the new slug).
"""
md = MarkdownService()
posts = PostService(engine)
pages = PageService(engine)
audit = AuditService(engine)
service = AdminPostsService(
engine=engine,
markdown=md,
post_service=posts,
page_service=pages,
audit=audit,
)
primed = posts.list_published()
# Hit the cache so we have a concrete reference to compare against.
still_cached = posts.list_published()
assert primed is still_cached
created = service.create(
title="Fresh Post",
body_md="x",
status=PostStatus.PUBLISHED,
author_id=1,
)
after = posts.list_published()
assert after is not primed
assert any(p.slug == created.slug for p in after)

View File

@@ -26,6 +26,21 @@ from sqlalchemy import text
_URL_RE = re.compile(r"http[s]?://[^\s]+/admin/auth/consume/[A-Za-z0-9_-]+")
def _csrf_from_response(resp) -> str:
"""Extract the CSRF token from a rendered admin page.
The admin base template emits a ``<meta name="csrf-token">`` tag
populated by the CSRFCookieMiddleware. Tests pull it from there
rather than the cookie value directly so the assertion also
proves the template wiring is intact.
"""
match = re.search(
r'<meta name="csrf-token" content="([^"]*)"', resp.text
)
assert match, "csrf-token meta tag missing"
return match.group(1)
@pytest.fixture
def admin_app(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
@@ -119,14 +134,20 @@ def test_full_login_flow(admin_app) -> None:
# cb_session cookie set.
assert "cb_session" in resp.cookies
# 4. GET /admin renders welcome.
# 4. GET /admin renders the Phase 4 dashboard.
resp = client.get("/admin")
assert resp.status_code == 200
assert "Welcome" in resp.text
assert "Dashboard" in resp.text
assert "headhen@example.com" in resp.text
csrf_token = _csrf_from_response(resp)
# 5. POST /admin/logout clears cookie + redirects.
resp = client.post("/admin/logout", follow_redirects=False)
# 5. POST /admin/logout clears cookie + redirects. CSRF required
# now that Phase 4 has enabled the double-submit cookie.
resp = client.post(
"/admin/logout",
data={"csrf_token": csrf_token},
follow_redirects=False,
)
assert resp.status_code == 303
assert resp.headers["location"] == "/admin/login"

View File

@@ -0,0 +1,91 @@
"""Tests for :class:`app.services.csrf.CSRFService`."""
from __future__ import annotations
import pytest
from itsdangerous import URLSafeTimedSerializer
from app.services.csrf import CSRF_COOKIE_NAME, CSRFService
@pytest.fixture
def signer() -> URLSafeTimedSerializer:
"""Return an ``itsdangerous`` signer with the canonical CSRF salt."""
return URLSafeTimedSerializer("test-key-for-csrf-service-0123456789", salt="csrf")
@pytest.fixture
def service(signer: URLSafeTimedSerializer) -> CSRFService:
"""Return a dev-mode :class:`CSRFService` (Secure flag off)."""
return CSRFService(signer, production=False)
def test_issue_returns_matching_token_and_cookie(service: CSRFService) -> None:
"""Without a prior cookie, :meth:`issue` mints a fresh pair that verify.
The token and the cookie value are the same signed string — by
design — but the critical property is that ``verify`` accepts them
as a match.
"""
token, cookie = service.issue(existing_cookie=None)
assert token
assert cookie
assert service.verify(cookie_value=cookie, submitted=token) is True
def test_issue_reuses_nonce_when_cookie_valid(service: CSRFService) -> None:
"""A still-valid cookie is reused so open admin tabs keep working."""
_, cookie = service.issue()
token2, cookie2 = service.issue(existing_cookie=cookie)
# Cookie value is deterministic post-nonce but itsdangerous signs
# with a timestamp, so the signed strings differ even when the
# underlying nonce is reused. The cross-pair verify matters.
assert service.verify(cookie_value=cookie, submitted=token2) is True
assert service.verify(cookie_value=cookie2, submitted=token2) is True
def test_verify_rejects_tampered_token(service: CSRFService) -> None:
"""A single-char tweak to the signed token breaks the signature."""
token, cookie = service.issue()
# Flip a char in the signature payload.
tampered = token[:-1] + ("A" if token[-1] != "A" else "B")
assert service.verify(cookie_value=cookie, submitted=tampered) is False
def test_verify_rejects_different_cookie(service: CSRFService) -> None:
"""Two separately-issued nonces never match each other."""
t1, c1 = service.issue()
t2, c2 = service.issue()
assert service.verify(cookie_value=c1, submitted=t2) is False
assert service.verify(cookie_value=c2, submitted=t1) is False
def test_verify_rejects_missing_values(service: CSRFService) -> None:
"""Empty or None inputs fail closed."""
assert service.verify(cookie_value=None, submitted="x") is False
assert service.verify(cookie_value="x", submitted=None) is False
assert service.verify(cookie_value="", submitted="") is False
def test_verify_rejects_different_key(signer: URLSafeTimedSerializer) -> None:
"""Tokens signed by a different key never verify."""
service_a = CSRFService(signer, production=False)
other_signer = URLSafeTimedSerializer("DIFFERENT-KEY-XYZ", salt="csrf")
service_b = CSRFService(other_signer, production=False)
_, cookie_a = service_a.issue()
token_b, _ = service_b.issue()
assert service_a.verify(cookie_value=cookie_a, submitted=token_b) is False
def test_cookie_params_production_sets_secure(signer: URLSafeTimedSerializer) -> None:
"""``Secure=True`` only when constructed with ``production=True``."""
dev = CSRFService(signer, production=False)
prod = CSRFService(signer, production=True)
assert dev.cookie_params()["secure"] is False
assert prod.cookie_params()["secure"] is True
assert dev.cookie_params()["key"] == CSRF_COOKIE_NAME
# HttpOnly is intentionally off so JS can read the cookie for
# double-submit header POSTs.
assert dev.cookie_params()["httponly"] is False
assert dev.cookie_params()["samesite"] == "lax"

179
tests/test_media_service.py Normal file
View File

@@ -0,0 +1,179 @@
"""Tests for :class:`app.services.media.MediaService`.
Uses real bytes produced by Pillow in-test so the magic-byte check
exercises real image data. Does NOT mock the filesystem — writes go
to a per-test temp dir so we can assert the stored JPEG is a valid
JPEG after the re-encode pipeline.
"""
from __future__ import annotations
import io
from pathlib import Path
import pytest
from PIL import Image
from sqlalchemy import Engine
from app.services.audit import AuditService
from app.services.media import (
MAX_UPLOAD_BYTES,
MediaRejectedError,
MediaService,
)
# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------
def _jpeg_bytes(size: tuple[int, int] = (64, 48), color: str = "red") -> bytes:
"""Return valid JPEG bytes produced by Pillow."""
img = Image.new("RGB", size, color)
buf = io.BytesIO()
img.save(buf, format="JPEG", quality=85)
return buf.getvalue()
def _rgba_png_bytes(size: tuple[int, int] = (32, 32)) -> bytes:
"""Return valid RGBA-transparent PNG bytes."""
img = Image.new("RGBA", size, (0, 128, 0, 128))
buf = io.BytesIO()
img.save(buf, format="PNG")
return buf.getvalue()
@pytest.fixture
def media_root(tmp_path: Path) -> Path:
"""Return a per-test directory rooted under the pytest tmpdir."""
root = tmp_path / "media"
root.mkdir()
return root
@pytest.fixture
def service(db_engine: Engine, media_root: Path) -> MediaService:
"""Return a :class:`MediaService` wired to a real engine + temp dir."""
return MediaService(
engine=db_engine,
media_root=str(media_root),
audit=AuditService(db_engine),
)
# ---------------------------------------------------------------------------
# happy path
# ---------------------------------------------------------------------------
def test_save_upload_stores_jpeg_and_returns_media(
service: MediaService, media_root: Path
) -> None:
"""A valid JPEG is re-encoded and written under a random filename."""
data = _jpeg_bytes()
media = service.save_upload(
original_filename="cute-chick.jpg",
data=data,
uploaded_by=1,
)
assert media.id > 0
assert media.content_type == "image/jpeg"
assert media.filename.endswith(".jpg")
assert media.original_filename == "cute-chick.jpg"
assert media.size_bytes > 0
stored = Path(media.stored_path)
assert stored.exists()
assert stored.parent.parent.parent == media_root
# Verify the written file is a real JPEG RGB image.
with Image.open(stored) as img:
assert img.format == "JPEG"
assert img.mode == "RGB"
def test_save_upload_assigns_random_filename_not_original(
service: MediaService,
) -> None:
"""The stored filename is a random token — not the client-supplied name."""
data = _jpeg_bytes()
media = service.save_upload(
original_filename="secret.jpg",
data=data,
uploaded_by=1,
)
assert media.filename != "secret.jpg"
# Random component is 16 bytes url-safe (~22 chars) + ".jpg".
name, _, ext = media.filename.rpartition(".")
assert ext == "jpg"
assert len(name) >= 16
def test_public_url_is_under_media_prefix(service: MediaService) -> None:
"""``public_url`` starts with ``/media/`` and a year partition."""
media = service.save_upload(
original_filename="ok.jpg",
data=_jpeg_bytes(),
uploaded_by=1,
)
url = service.public_url(media)
assert url.startswith("/media/")
assert url.endswith(media.filename)
def test_save_upload_flattens_rgba_onto_white(service: MediaService) -> None:
"""An RGBA PNG upload is flattened and stored as RGB JPEG."""
data = _rgba_png_bytes()
media = service.save_upload(
original_filename="transparent.png",
data=data,
uploaded_by=1,
)
stored = Path(media.stored_path)
with Image.open(stored) as img:
assert img.format == "JPEG"
assert img.mode == "RGB"
# ---------------------------------------------------------------------------
# rejection paths
# ---------------------------------------------------------------------------
def test_reject_empty_upload(service: MediaService) -> None:
"""Empty uploads are rejected before any decode runs."""
with pytest.raises(MediaRejectedError):
service.save_upload(
original_filename="empty.jpg",
data=b"",
uploaded_by=1,
)
def test_reject_non_image_bytes(service: MediaService) -> None:
"""A plain-text payload fails the magic-byte sniff."""
with pytest.raises(MediaRejectedError):
service.save_upload(
original_filename="totally.jpg",
data=b"this is definitely not an image payload" * 10,
uploaded_by=1,
)
def test_reject_gif_upload(service: MediaService) -> None:
"""Animated-GIF hazard — GIFs are explicitly not on the allowlist."""
img = Image.new("RGB", (16, 16), "green")
buf = io.BytesIO()
img.save(buf, format="GIF")
with pytest.raises(MediaRejectedError):
service.save_upload(
original_filename="animated.gif",
data=buf.getvalue(),
uploaded_by=1,
)
def test_reject_oversize_upload(service: MediaService) -> None:
"""Payloads over the 8 MB cap are rejected."""
huge = b"\xff" * (MAX_UPLOAD_BYTES + 1)
with pytest.raises(MediaRejectedError):
service.save_upload(
original_filename="huge.jpg",
data=huge,
uploaded_by=1,
)

68
tests/test_slugs.py Normal file
View File

@@ -0,0 +1,68 @@
"""Tests for :mod:`app.services.slugs`."""
from __future__ import annotations
import pytest
from app.services.slugs import ensure_unique, slugify
# ---------------------------------------------------------------------------
# slugify
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"title, expected",
[
("Hello World", "hello-world"),
("Hello, World!", "hello-world"),
(" leading and trailing ", "leading-and-trailing"),
("CamelCaseTitle", "camelcasetitle"),
("Multiple spaces", "multiple-spaces"),
("dashes---everywhere", "dashes-everywhere"),
("mix-of 99 items & stuff", "mix-of-99-items-stuff"),
("Café au lait", "caf-au-lait"), # non-ASCII dropped
("", "post"), # fallback
("...", "post"), # all-punctuation → fallback
("123", "123"), # digits allowed
("snake_case", "snake-case"),
],
)
def test_slugify_expected(title: str, expected: str) -> None:
""":func:`slugify` produces the documented output per input."""
assert slugify(title) == expected
def test_slugify_no_leading_or_trailing_hyphens() -> None:
"""Slugs never have dangling hyphens regardless of input shape."""
assert slugify("!!hello!!") == "hello"
assert slugify("--leading and trailing--") == "leading-and-trailing"
# ---------------------------------------------------------------------------
# ensure_unique
# ---------------------------------------------------------------------------
def test_ensure_unique_returns_base_when_free() -> None:
"""If nothing collides, the base slug is returned unchanged."""
taken: set[str] = set()
result = ensure_unique("hello", lambda s: s in taken)
assert result == "hello"
def test_ensure_unique_suffixes_on_collision() -> None:
"""Collisions produce -2, -3, etc. in order."""
taken = {"hello", "hello-2", "hello-3"}
result = ensure_unique("hello", lambda s: s in taken)
assert result == "hello-4"
def test_ensure_unique_suffix_sequence_starts_at_two() -> None:
"""First collision uses ``-2`` — never ``-1`` or ``-0``."""
taken = {"hello"}
result = ensure_unique("hello", lambda s: s in taken)
assert result == "hello-2"
def test_ensure_unique_raises_when_exhausted() -> None:
"""A pathological ``exists`` callable hits the guard ceiling."""
with pytest.raises(RuntimeError):
ensure_unique("hello", lambda s: True, max_attempts=3)