"""Mockup upload, view-token minting, and the isolated /m/ serving route. /m/ is deliberately OUTSIDE the session boundary (an exact-prefix public path in the middleware): the signed, expiring bundle token in the path IS the auth, and every response carries the CSP sandbox header so the document is opaque-origin however it is reached (goal §3.1). This is not an /api/* exemption — see middleware.py. """ import logging from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile from fastapi.responses import HTMLResponse, Response from app import schemas from app.db import get_session_factory from app.identity import optional_user from app.services import mockups from app.services import projects as svc log = logging.getLogger(__name__) router = APIRouter(tags=["mockups"]) def _user(user: str | None = Depends(optional_user)) -> str: assert user is not None return user def _raise(exc: mockups.MockupError): raise HTTPException(status_code=exc.status, detail={"error_code": exc.code, "summary": str(exc)}) @router.post("/api/options/{option_id}/files", response_model=schemas.UploadResultOut) async def upload_bundle(option_id: int, files: list[UploadFile], user: str = Depends(_user)) -> schemas.UploadResultOut: """Upload one or more bundle files. Each file's rel_path is its (sanitized) client filename — a webkitdirectory upload preserves subdirs. If the option has no entry yet and an index.html (or the only .html) arrives, it becomes the entry.""" sf = get_session_factory() with sf() as db: try: option, project_id = svc.option_project(db, option_id) except svc.DomainError as exc: raise HTTPException(status_code=exc.status, detail={ "error_code": exc.code, "summary": str(exc)}) from exc if svc.side_of(user, project_id) != "manager": raise HTTPException(status_code=403, detail={ "error_code": "FORBIDDEN", "summary": "Uploading mockups needs the manager side."}) stored: list[schemas.UploadedFileOut] = [] html_paths: list[str] = [] for f in files: data = await f.read() try: row = mockups.store_file(db, option, f.filename or "", data) except mockups.MockupError as exc: db.rollback() _raise(exc) stored.append(schemas.UploadedFileOut( rel_path=row.rel_path, bytes=row.size, content_type=row.content_type)) if row.content_type.startswith("text/html"): html_paths.append(row.rel_path) all_paths = {x.rel_path for x in option.files} if option.entry_path not in all_paths and html_paths: option.entry_path = ("index.html" if "index.html" in html_paths else html_paths[0]) svc.record(db, project_id, user, "files_uploaded", {"option_id": option.id, "count": len(stored)}) db.commit() return schemas.UploadResultOut(files=stored, entry_path=option.entry_path) @router.get("/api/options/{option_id}/view", response_model=schemas.ViewTokenOut) def mint_view(option_id: int, user: str = Depends(_user)) -> schemas.ViewTokenOut: sf = get_session_factory() with sf() as db: try: option, project_id = svc.option_project(db, option_id) except svc.DomainError as exc: raise HTTPException(status_code=exc.status, detail={ "error_code": exc.code, "summary": str(exc)}) from exc if not svc.option_visible_to(db, user, option): raise HTTPException(status_code=404, detail={ "error_code": "NO_SUCH_OPTION", "summary": "No such option."}) if not option.files: # A manager opening an option they have not uploaded into yet: say so # rather than minting a token for a document that does not exist. raise HTTPException(status_code=409, detail={ "error_code": "NO_MOCKUP_YET", "summary": "This option has no mockup uploaded yet."}) token, nonce = mockups.mint_token(option.id) return schemas.ViewTokenOut( token=token, nonce=nonce, entry_url=f"/m/{token}/{option.entry_path}", expires_in=mockups.TOKEN_MAX_AGE) _BLOCKED_TOPLEVEL = """ Open in Easel

This file opens inside Easel

Mockups are viewed through the Easel reviewer, not as a standalone page.

""" @router.api_route("/m/{token}/{rel_path:path}", methods=["GET", "HEAD"], include_in_schema=False) def serve_mockup(token: str, rel_path: str, request: Request) -> Response: try: option_id, nonce = mockups.verify_token(token) rel = mockups.normalize_rel_path(rel_path) except mockups.MockupError as exc: return Response(str(exc), status_code=exc.status, media_type="text/plain", headers=dict(mockups.MOCKUP_HEADERS)) # Defense-in-depth (goal §3.1): a top-level navigation is refused with a # friendly page. The CSP sandbox header would already neutralize it — this # keeps the failure legible. Subresource dests (script/img/style/...) and # absent headers (older agents) pass; the sandbox is the real boundary. if request.headers.get("sec-fetch-dest", "").lower() == "document": log.info("mockup top-level navigation refused option=%s path=%s", option_id, rel) return HTMLResponse(_BLOCKED_TOPLEVEL, status_code=403, headers=dict(mockups.MOCKUP_HEADERS)) sf = get_session_factory() with sf() as db: from sqlalchemy import select from app.models import MockupFile row = db.scalar(select(MockupFile).where( MockupFile.option_id == option_id, MockupFile.rel_path == rel)) if row is None: return Response("Not found", status_code=404, media_type="text/plain", headers=dict(mockups.MOCKUP_HEADERS)) try: data = mockups.file_bytes(row) except FileNotFoundError: return Response("Not found", status_code=404, media_type="text/plain", headers=dict(mockups.MOCKUP_HEADERS)) headers = dict(mockups.MOCKUP_HEADERS) if row.content_type.startswith("text/html"): data = mockups.inject_bridge(data, nonce) headers["Cache-Control"] = "no-store" if request.method == "HEAD": return Response(status_code=200, media_type=row.content_type, headers=headers) return Response(content=data, media_type=row.content_type, headers=headers)