app.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
FastAPI Backend for Pieria.
Phase 4: Targeted WebSocket Routing for Multiple Displays.
"""
# NOTE — "kept-bound" imports (the dual-patch pattern). Several names in this file look unused *here*
# but MUST stay imported at module scope. Their call sites moved into core/* and routers/* during the
# app-split, yet tests still reach them via `app` (a.k.a. `app_module`): either monkeypatching
# `app_module.X` — which only works when `app` and the real call site share the *same* module object,
# so the name must remain bound here — or importing a helper/constant straight off `app`. Each such
# import is tagged with the test(s) that depend on it; don't drop one without checking them first.
import asyncio # noqa: F401 — tests/test_download.py patches app_module.asyncio.sleep
import logging
import pillow_heif
from dotenv import load_dotenv
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
# Load environment variables
load_dotenv()
# Teach Pillow to read HEIC/HEIF so iPhone photos (the default capture format) decode through the
# normal Image.open() path everywhere. Upload handlers transcode to a browser-renderable format —
# browsers can't display HEIC either, so decode alone isn't enough.
pillow_heif.register_heif_opener()
# -----------------------------------------------------------------------------
# 1. Configuration, Logging & Targeted WebSocket Manager
# -----------------------------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("artwork-display-api")
# Local imports (see the dual-patch NOTE above for why several look unused).
import httpx # noqa: F401 — tests/test_download.py + tests/test_catalog.py patch app_module.httpx.AsyncClient
import federation # noqa: F401 — tests/test_federation.py + tests/test_publisher_api.py patch app_module.federation.*
# ARTWORK_ROOT + STATIC_DIR are used below (static mounts); LIBRARY_DIR is kept bound because
# tests/test_display_image.py, test_factory_reset.py + test_playlist_resume.py monkeypatch it.
from config import APP_VERSION, ARTWORK_ROOT, DEMO_MODE, LIBRARY_DIR, STATIC_DIR # noqa: F401
# Targeted WebSocket connection registry (shared by the ws + remote push paths).
from core.connections import ConnectionManager, manager # noqa: F401
# Public demo mode (SD_DEMO_MODE=1) — default-DENY ASGI gate, no-op unless set. See core/demo.py.
from core.demo import DemoModeMiddleware # noqa: E402
# SSRF-safe downloader (see core/downloads.py); tests/test_download.py imports it off `app`.
from core.downloads import _download_image_to_library # noqa: F401,E402
# Boot machinery (leader election, migrations, filesystem sync, factory seed, Canvas warmer) — see
# core/lifespan.py. `lifespan` is passed to FastAPI() below.
from core.lifespan import (
lifespan,
sync_db_with_filesystem, # noqa: F401 — re-exported for tests/test_playlist_resume.py
)
# Derivative-image primitives (see core/media.py); tests/test_factory_reset.py + test_display_image.py
# monkeypatch DERIVATIVES_DIR and import DISPLAY_MAX_EDGE off `app`.
from core.media import DERIVATIVES_DIR, DISPLAY_MAX_EDGE, USER_UPLOAD_MAX_BYTES # noqa: F401,E402
# Origin/CORS trust checks used by the middleware below (see core/security.py).
from core.security import ( # noqa: E402
_PUBLIC_FEED_GET_PREFIXES,
_origin_allowed,
)
# Settings-table + schedule helpers (see core/settings_util.py); tests/test_schedule.py imports
# DEFAULT_SCHEDULE + resolve_schedule_state off `app`.
from core.settings_util import ( # noqa: E402
DEFAULT_SCHEDULE, # noqa: F401
resolve_schedule_state, # noqa: F401
)
# SessionLocal: tests/test_connection_manager.py monkeypatches app_module.SessionLocal (the /ws route
# reads its own binding in routers/ws.py).
from database import SessionLocal # noqa: F401,E402
# Leaf domain routers extracted from app.py (Phase 1 + Phase 2 + Phase 3 of the app-split refactor).
# Each is a plain APIRouter with no dependency on this module — see routers/__init__.py for the
# import rule.
from routers.admin import router as admin_router
from routers.api_tokens import router as api_tokens_router
from routers.api_v1 import v1_app
from routers.backup import router as backup_router
from routers.catalog import _read_local_json # noqa: F401 — re-exported for tests/test_cache.py
from routers.catalog import router as catalog_router
from routers.curation import router as curation_router
from routers.demo import router as demo_router
from routers.display import router as display_router
from routers.federation import router as federation_router
from routers.health import router as health_router
from routers.library import router as library_router
from routers.packs import router as packs_router
from routers.pages import router as pages_router
from routers.publisher import router as publisher_router
from routers.settings import router as settings_router
from routers.studio import PERSONAL_PLAYLIST_NAME # noqa: F401 — re-exported for tests/test_personal.py etc.
from routers.studio import router as studio_router
from routers.ws import router as ws_router
# Demo mode also turns off FastAPI's own introspection surface (/docs, /redoc, /openapi.json) — they're
# not app routes core/demo.py's gate can pattern-match, so the only clean "off" is never registering them.
_demo_docs_kwargs = {"docs_url": None, "redoc_url": None, "openapi_url": None} if DEMO_MODE else {}
app = FastAPI(title="Pieria", version=APP_VERSION, lifespan=lifespan, **_demo_docs_kwargs)
# --- M4: body-size cap enforced BEFORE Starlette parses/spools multipart (ADR-119 audit, 2026-09-22) --
# core/media.py's read_capped_upload runs inside the route handler, but by then Starlette has already
# parsed the multipart body (and, for large bodies, spooled it to disk) to hand FastAPI an UploadFile —
# the handler-level cap is too late to bound that work. This is a raw ASGI middleware (not
# BaseHTTPMiddleware, which itself buffers) sitting below routing: a Content-Length pre-check rejects
# oversized requests with 413 before any body is read, and a running byte count over the receive
# stream catches chunked/no-Content-Length bodies that lie about size. Scoped to the two upload routes
# only — every other endpoint is untouched. The per-handler check in core/media.py stays as
# defense-in-depth (e.g. a client whose declared boundary overhead undercounts).
_CAPPED_UPLOAD_PATHS = frozenset({"/upload", "/upload/personal"})
# Small multipart overhead allowance (boundary + per-part headers) atop the raw file cap.
_UPLOAD_BODY_CAP = USER_UPLOAD_MAX_BYTES + 64 * 1024
class UploadBodyCapMiddleware:
"""Pure ASGI middleware — rejects an oversized POST to a capped upload path with 413 before the
body reaches Starlette's multipart parser."""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if (scope["type"] != "http" or scope.get("method") != "POST"
or scope.get("path") not in _CAPPED_UPLOAD_PATHS):
await self.app(scope, receive, send)
return
headers = dict(scope.get("headers") or [])
content_length = headers.get(b"content-length")
if content_length is not None:
try:
if int(content_length) > _UPLOAD_BODY_CAP:
await self._reject(send)
return
except ValueError:
pass # malformed header — the running-total check below still applies
# Drain + count the body ourselves so a chunked/no-Content-Length request can't lie about
# size; buffer what we've read so it can be replayed to the real app once we know it fits.
buffered = []
total = 0
while True:
message = await receive()
buffered.append(message)
if message["type"] != "http.request":
break
total += len(message.get("body", b"") or b"")
if total > _UPLOAD_BODY_CAP:
await self._reject(send)
return
if not message.get("more_body", False):
break
idx = 0
async def replay_receive():
nonlocal idx
if idx < len(buffered):
message = buffered[idx]
idx += 1
return message
return await receive()
await self.app(scope, replay_receive, send)
@staticmethod
async def _reject(send):
await send({
"type": "http.response.start",
"status": 413,
"headers": [(b"content-type", b"application/json")],
})
await send({
"type": "http.response.body",
"body": (f'{{"detail":"File too large (max {USER_UPLOAD_MAX_BYTES // (1024 * 1024)} MB)."}}'
.encode()),
})
app.add_middleware(UploadBodyCapMiddleware)
# --- Public demo mode (SD_DEMO_MODE=1) — default-DENY ASGI gate, no-op unless set (core/demo.py) -----
app.add_middleware(DemoModeMiddleware)
# Leaf domain routers (Phase 1 + Phase 2 + Phase 3 + Phase 4 of the app-split refactor — see
# .ai/refactor_app_split_plan.md).
# Order matches the (alphabetical) import block above; these are leaf domain routers with distinct
# path prefixes, so registration order carries no route-matching significance.
app.include_router(admin_router)
app.include_router(api_tokens_router)
app.include_router(backup_router)
app.include_router(catalog_router)
app.include_router(curation_router)
app.include_router(demo_router)
app.include_router(display_router)
app.include_router(federation_router)
app.include_router(health_router)
app.include_router(library_router)
app.include_router(packs_router)
app.include_router(pages_router)
app.include_router(publisher_router)
app.include_router(settings_router)
app.include_router(studio_router)
app.include_router(ws_router)
# Public API v1 (ADR-147): a separate FastAPI instance with its own OpenAPI/docs, token-authenticated on
# every route (routers/api_v1.py). Mounted BEFORE the "/" static mount below, which would swallow it.
# The middleware added to `app` (origin guard, cache headers, demo gate) wraps mounted apps too.
app.mount("/api/v1", v1_app)
# M6 (INFRA-D075, 2026-09-25): these two were `@app.middleware("http")` (BaseHTTPMiddleware) —
# converted to pure ASGI, matching UploadBodyCapMiddleware/DemoModeMiddleware above. NOTE: a reviewer's
# load test showed BaseHTTPMiddleware did NOT leak pool connections on this FastAPI 0.141 / Starlette
# 0.52 stack, so that was never the thumbnail-storm incident's actual cause — see
# tests/test_middleware_pool_leak.py for the real mechanism (a synchronous DB query run directly inside
# `async def`, blocking the whole event loop — including the continuations that would free the pool —
# once the pool is exhausted; fixed by threading that query, core/media.py's lookup_artwork_filename).
# This conversion is kept anyway: pure ASGI middleware runs in the SAME task as the request instead of a
# separate one, which is both correct (a disconnect propagates straight through the endpoint's own
# exception handling, same as no middleware at all, instead of via BaseHTTPMiddleware's extra task
# indirection) and measurably faster. Behavior (headers, CORS/origin rules) and add_middleware ordering
# relative to UploadBodyCapMiddleware/DemoModeMiddleware are unchanged.
class CacheHeadersMiddleware:
"""Pure ASGI port of the former inject_aggressive_cache_headers — identical header logic."""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
path = scope.get("path", "")
is_media_cacheable = (
(path.startswith("/artworks/") and ("thumbnail" in path or "preview" in path))
or path.startswith("/media/")
or path.endswith((".svg", ".png", ".jpg", ".webp"))
)
is_code_asset = path.endswith((".css", ".js", ".json"))
is_html_asset = path.endswith(".html") or path in ("/admin", "/remote", "/studio", "/help", "/")
async def send_wrapper(message):
if message["type"] == "http.response.start":
headers = list(message.get("headers", []))
if path.startswith("/api/") or is_html_asset or path.startswith("/display/"):
# API/HTML and the per-display e-ink endpoint must never be cached.
# (/display/*.png must beat the is_media_cacheable .png rule below.)
headers.append((b"cache-control", b"no-store, no-cache, must-revalidate"))
elif is_media_cacheable:
# Images/media rarely change — cache aggressively
headers.append((b"cache-control", b"public, max-age=31536000, immutable"))
elif is_code_asset:
# JS/CSS/JSON change during development — short cache + revalidate
headers.append((b"cache-control", b"public, max-age=60, must-revalidate"))
# Security headers (L2). nosniff is defense-in-depth against the /media
# MIME-confusion XSS class (H1); Referrer-Policy keeps LAN paths out of any outbound
# Referer.
headers.append((b"x-content-type-options", b"nosniff"))
headers.append((b"referrer-policy", b"same-origin"))
message = dict(message)
message["headers"] = headers
await send(message)
await self.app(scope, receive, send_wrapper)
# --- CORS + cross-origin state-change guard (ADR-036) ------------------------------------------------
# The app is a no-login LAN kiosk (ADR-013/015): the trust boundary is "you are a device on my LAN".
# The old wildcard `CORSMiddleware(allow_origins=["*"], allow_credentials=True)` silently widened that
# to "any browser tab on the LAN can drive the full API cross-origin". We replace it with a policy that
# keeps the no-login model honest:
# * Cross-origin STATE CHANGES are blocked — a hostile page (bad ad, phishing tab) always sends an
# Origin header on a cross-origin state-changing fetch; curl/native integrations send none and are
# allowed (the accepted LAN-presence risk, unchanged).
# * The read-only public FEED (what integrations consume) stays cross-origin readable.
# * Admin/library GETs are NOT cross-origin readable (no ACAO) — a hostile tab can't exfiltrate them.
# * Same-origin (the kiosk's own page) and explicitly configured SD_ALLOWED_ORIGINS always pass.
# _origin_allowed, _PUBLIC_FEED_GET_PREFIXES now live in core/security.py (imported above).
_MUTATING_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"})
class CorsAndOriginGuardMiddleware:
"""Pure ASGI port of the former cors_and_origin_guard — identical CORS/origin logic."""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
# M6 nit: FIRST occurrence wins on a duplicated header — matches Starlette's own
# Request.headers.get() (which the old BaseHTTPMiddleware version read), and RFC 9110's
# guidance to treat extra copies as suspect rather than silently preferring the last one. A
# naive dict comprehension over the raw ASGI header list would let the LAST copy win instead.
headers_in: dict[str, str] = {}
for k, v in scope.get("headers") or []:
key = k.decode("latin-1").lower()
if key not in headers_in:
headers_in[key] = v.decode("latin-1")
origin = headers_in.get("origin", "")
host = headers_in.get("host", "")
path = scope.get("path", "")
method = scope.get("method", "GET")
allowed = _origin_allowed(origin, host)
# A CORS preflight advertises the real method it is clearing; judge against that, not OPTIONS.
effective_method = (headers_in.get("access-control-request-method", "GET").upper()
if method == "OPTIONS" else method)
is_public_read = effective_method in ("GET", "HEAD") and path.startswith(_PUBLIC_FEED_GET_PREFIXES)
# The teeth: refuse a cross-origin state change from a browser tab (blocks the preflight too).
if effective_method in _MUTATING_METHODS and origin and not allowed:
await send({
"type": "http.response.start",
"status": 403,
"headers": [(b"content-type", b"text/plain; charset=utf-8")],
})
await send({"type": "http.response.body", "body": b"cross-origin request blocked"})
return
extra_headers = []
if origin and (is_public_read or allowed):
extra_headers.append((b"access-control-allow-origin",
b"*" if is_public_read else origin.encode("latin-1")))
extra_headers.append((b"vary", b"Origin"))
if method == "OPTIONS":
extra_headers.append((b"access-control-allow-methods",
b"GET, POST, PUT, PATCH, DELETE, OPTIONS"))
req_headers = headers_in.get("access-control-request-headers") or "*"
extra_headers.append((b"access-control-allow-headers", req_headers.encode("latin-1")))
extra_headers.append((b"access-control-max-age", b"600"))
if method == "OPTIONS" and origin:
await send({"type": "http.response.start", "status": 204, "headers": list(extra_headers)})
await send({"type": "http.response.body", "body": b""})
return
async def send_wrapper(message):
if message["type"] == "http.response.start" and extra_headers:
message = dict(message)
message["headers"] = list(message.get("headers", [])) + extra_headers
await send(message)
await self.app(scope, receive, send_wrapper)
app.add_middleware(CacheHeadersMiddleware)
app.add_middleware(CorsAndOriginGuardMiddleware)
# POST /api/admin/factory-reset now lives in routers/admin.py.
# GET /next-image, /display/{display_id}/current.{ext}, /artworks/{artwork_id}/display.jpg,
# /api/telemetry/heartbeat, /api/displays/{display_id}/preferred-playlist, and
# /api/displays/{display_id}/schedule-state now live in routers/display.py.
#
# GET /api/remote/displays, GET /api/displays/{display_id}/now-playing, POST /api/remote/change,
# and WEBSOCKET /ws/{display_id} now live in routers/ws.py.
# -----------------------------------------------------------------------------
# 4.5 Settings (API Keys) — GET/POST /api/settings/keys* now live in routers/settings.py.
# -----------------------------------------------------------------------------
# POST /api/settings/keys/{source} now lives in routers/settings.py.
# -----------------------------------------------------------------------------
# 4.6 AI Engine (model provider configuration) — /api/settings/ai*, /api/settings/ai/oauth/* now
# live in routers/settings.py.
# -----------------------------------------------------------------------------
# -----------------------------------------------------------------------------
# 4.7 Catalog (browseable curated public-domain art; lazy high-res on add)
# -----------------------------------------------------------------------------
# CATALOG_DIR, _read_local_json, _subscribed_summaries, _subscribed_collection, _catalog_index,
# _catalog_collection, _download_and_create_artwork, and every /api/catalog* route now live in
# routers/catalog.py (imported above; _read_local_json is re-exported for tests/test_cache.py).
# ---------------------------------------------------------------------------
# §4.8 Samsung Frame TV push (Integrations)
# ---------------------------------------------------------------------------
# _frame_select (the selector shared with core/lifespan.py's frame_push_loop AND routers/settings.py's
# "Test / Push now" route) now lives in core/playback.py — see that module's
# docstring for why. The /api/settings/frame*, /api/settings/catalog* routes now live in
# routers/settings.py; the /api/catalog* browse + add routes now live in routers/catalog.py.
# Federation (/api/subscriptions*) now lives in routers/federation.py.
# Publisher Studio (/api/publisher/*) now lives in routers/publisher.py.
# -----------------------------------------------------------------------------
# 5. Static File Serving
# -----------------------------------------------------------------------------
if ARTWORK_ROOT.exists():
app.mount("/media", StaticFiles(directory=str(ARTWORK_ROOT)), name="media")
# The /admin, /help, /studio, /remote page routes + the /publisher redirect now live in
# routers/pages.py.
if STATIC_DIR.exists():
app.mount("/", StaticFiles(directory=str(STATIC_DIR), html=True), name="static")
if __name__ == "__main__":
import uvicorn
uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)