"""Native browser media authentication only; API authentication stays bearer-only.""" import mimetypes from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import FileResponse, Response from sqlalchemy.orm import Session from app.database import get_db from app.services import storage_service, thumbnails from app.utils.auth import get_current_user from app.utils.upload_access import ( LEGACY_LMS_PREFIXES, local_upload_path, upload_file, references, document_for_file, can_read_upload, card_deck_ids, ) router = APIRouter() @router.api_route("/uploads/{path:path}", methods=["GET", "HEAD"]) def read_upload(path: str, request: Request, attempt_id: int | None = None, w: int | None = None, db: Session = Depends(get_db)): """An upload, optionally at one of two smaller widths. `?w=256` and `?w=640` are the only sizes there are, and anything else is refused rather than honoured — an endpoint that resizes to whatever the query string asks for is a CPU sink anybody can point at. The authorisation below is unchanged and runs first: a thumbnail of a file you may not read is a file you may not read. These stay `private, no-store` like everything else here. They are behind authentication, so there is nothing for a shared cache to do with them; the win is that the bytes are a tenth the size. """ headers = {"Cache-Control": "private, no-store", "Vary": "Cookie, Authorization", "X-Content-Type-Options": "nosniff"} if w is not None and w not in thumbnails.WIDTHS: raise HTTPException(400, f"Width must be one of {', '.join(map(str, thumbnails.WIDTHS))}", headers=headers) try: return _read_upload(path, request, attempt_id, db, headers, w) except HTTPException as exc: exc.headers = {**(exc.headers or {}), **headers} raise def _read_upload(path, request, attempt_id, db, headers, width=None): try: path = local_upload_path(path) target = upload_file(path) except HTTPException: raise HTTPException(404, "File not found") questions = references(db, path) cards = card_deck_ids(db, path) # Default private also prevents orphaned question/card files becoming anonymous. protected = (not path.startswith(LEGACY_LMS_PREFIXES) or document_for_file(db, path) or questions or cards) if protected: authorization = request.headers.get("authorization", "") token = authorization[7:] if authorization.lower().startswith("bearer ") else request.cookies.get("pedshub_media", "") user = get_current_user(token, db) if not can_read_upload(db, path, user, questions, cards, attempt_id): raise HTTPException(404, "File not found") # Known legacy LMS directories retain their policy; this is not a whole-LMS audit. if target.suffix.lower() == ".svg": headers["Content-Security-Policy"] = "sandbox; default-src 'none'; style-src 'unsafe-inline'" # Asked for small, and small exists: the derivative is made on the first # ask and stored beside the original. `None` back means there is nothing # smaller worth serving — not an image, or already narrower than asked — # so the original goes out, which is what the caller wanted anyway. if width: small = thumbnails.get(path, width) if small is not None: if request.method == "HEAD": return Response(status_code=200, media_type="image/webp", headers={**headers, "Content-Length": str(len(small))}) return Response(content=small, media_type="image/webp", headers=headers) # Serving must go through the storage service, or object storage would be # write-only: the bytes would be in the bucket and still read from disk. # The path is already authorised and confined to the upload root above. if target.is_file(): return FileResponse(target, headers=headers) data = storage_service.s3_object(path) if data is None: raise HTTPException(404, "File not found") media_type = mimetypes.guess_type(path)[0] or "application/octet-stream" if request.method == "HEAD": return Response(status_code=200, headers={**headers, "Content-Length": str(len(data))}, media_type=media_type) return Response(content=data, media_type=media_type, headers=headers)