fix(downloads): harden upload serving + ingest against stored-XSS and leaks (E7 review)

Deep-review remediation for the E7 upload path, all findings adversarially verified:

- Serve every library file with a Content-Type chosen from the ffprobe-verified media_kind
  + a strict container allow-list (storage.served_media_type), never guessed from the
  user-controlled file extension. Closes a stored-XSS vector where a media/HTML polyglot
  uploaded as "evil.html" was served inline as text/html on the login-free /watch surface.
  Applied to both the authenticated /file and the public watch_file endpoints.
- Classify uploaded still images (jpg/png/gif/webp/…) as media_kind='file' so they are
  force-downloaded and don't get a broken in-app video player.
- Roll back the on-disk side of an upload when the ingest transaction fails after place_file
  (unlink the placed file + its deterministic poster path), so a failed commit can't orphan
  bytes in the GC-skipped .uploads tree.
- Floor download_upload_max_bytes at min=1 so a stored 0 can't silently disable the per-file cap.
- Dedupe the ffprobe-JSON invocation into edit.run_ffprobe_json (shared by probe_codecs and
  upload._probe) and the upload extension into storage._clean_upload_ext; drop the now-unused
  parse_probe_streams.

Tests: public + authenticated force-download and the polyglot-not-served-as-HTML guard, the
real _probe classifier (image/cover-art/dims/duration), the ingest rollback/cleanup path, the
max_jobs branch, the cap-floor invariant, and run_ffprobe_json error handling.

backend/app/downloads/edit.py

@@ -438,18 +438,42 @@ _BROWSER_SAFE_VCODEC = ("h264",)
 _BROWSER_SAFE_ACODEC = ("aac", "mp3")
 
 
-def parse_probe_streams(payload: str) -> tuple[str | None, str | None]:
-    """(video_codec, audio_codec) out of ffprobe's JSON stream list, lowercased.
-
-    JSON, not the flat `default=nw=1` output: there the fields arrive in ffprobe's own struct
-    order, which puts `codec_name` BEFORE `codec_type`, so a line-by-line "remember the last
-    codec_type" parser attributes every name to the PREVIOUS stream. Measured on real downloads:
-    an ordinary H.264+AAC mp4 probed as `v=aac a=None` — the browser-playability check then
-    believed the video stream was AAC and re-encoded a perfectly fine file on every download."""
+def run_ffprobe_json(
+    path: Path,
+    *,
+    show_streams: bool = False,
+    show_format: bool = False,
+    entries: str | None = None,
+) -> dict | None:
+    """One ffprobe call → its parsed JSON dict, or None on any failure (timeout, non-zero exit,
+    missing binary, unparseable output). The SINGLE ffprobe-JSON-invocation point in the downloads
+    package: the browser-playability codec probe here and the upload-ingest probe (`upload._probe`)
+    both go through it, so the subprocess flags/timeout/error handling live in one place instead of
+    two drifting copies."""
+    args = ["ffprobe", "-v", "error"]
+    if show_streams:
+        args.append("-show_streams")
+    if show_format:
+        args.append("-show_format")
+    if entries:
+        args += ["-show_entries", entries]
+    args += ["-of", "json", str(path)]
     try:
-        streams = json.loads(payload or "{}").get("streams") or []
-    except ValueError:
-        return None, None
+        out = subprocess.run(args, capture_output=True, text=True, timeout=30, check=True).stdout
+        return json.loads(out)
+    except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError, ValueError):
+        return None
+
+
+def _codecs_from_streams(streams: list) -> tuple[str | None, str | None]:
+    """(video_codec, audio_codec) lowercased from an ffprobe JSON stream list — the first of each kind.
+
+    Parsed from the JSON `streams` list, NOT ffprobe's flat `default=nw=1` output: there the fields
+    arrive in ffprobe's own struct order, which puts `codec_name` BEFORE `codec_type`, so a line-by-line
+    "remember the last codec_type" parser attributes every name to the PREVIOUS stream. Measured on real
+    downloads: an ordinary H.264+AAC mp4 probed as `v=aac a=None` — the browser-playability check then
+    believed the video stream was AAC and re-encoded a perfectly fine file on every download. Reading each
+    stream dict by its own `codec_type` is immune to that ordering."""
     vcodec = acodec = None
     for s in streams:
         name = (s.get("codec_name") or "").lower() or None
@@ -464,26 +488,10 @@ def parse_probe_streams(payload: str) -> tuple[str | None, str | None]:
 def probe_codecs(path: Path) -> tuple[str | None, str | None]:
     """Return (video_codec, audio_codec) lowercased via one ffprobe call; None for a missing
     stream or if ffprobe fails (a failed probe is treated as 'leave it alone' upstream)."""
-    try:
-        out = subprocess.run(
-            [
-                "ffprobe",
-                "-v",
-                "error",
-                "-show_entries",
-                "stream=codec_type,codec_name",
-                "-of",
-                "json",
-                str(path),
-            ],
-            capture_output=True,
-            text=True,
-            timeout=30,
-            check=True,
-        ).stdout
-    except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError):
+    data = run_ffprobe_json(path, entries="stream=codec_type,codec_name")
+    if data is None:
         return None, None
-    return parse_probe_streams(out)
+    return _codecs_from_streams(data.get("streams") or [])
 
 
 def probe_duration(path: Path) -> int:

backend/app/downloads/storage.py

@@ -19,6 +19,8 @@ from datetime import date
 from pathlib import Path
 from xml.sax.saxutils import escape
 
+from app.downloads import mediakind
+
 # Reserve room for the "_id.ext" suffix + parent dirs within the 255-char component limit.
 _MAX_TITLE = 120
 # User-facing (Content-Disposition) names can be longer than an on-disk component.
@@ -147,6 +149,14 @@ def edit_rel_path(user_id: int, asset_id: int, ext: str) -> str:
     return f".edits/{user_id}/clip_{asset_id}.{ext}"
 
 
+def _clean_upload_ext(filename: str) -> str:
+    """The upload's extension reduced to a shell-safe, lowercased, bounded token (``""`` when the
+    upload has no extension). The SINGLE definition feeding both the on-disk suffix and the stored
+    `container` — they must agree (the served Content-Disposition suffix has to match the container),
+    so the sanitize/truncation rule lives in one place, not two drifting copies."""
+    return _UNSAFE.sub("", Path(filename).suffix.lstrip(".")).lower()[:16]
+
+
 def upload_rel_path(user_id: int, asset_id: int, filename: str) -> str:
     """On-disk path (relative to DOWNLOAD_ROOT) for a locally-uploaded file.
 
@@ -157,16 +167,61 @@ def upload_rel_path(user_id: int, asset_id: int, filename: str) -> str:
     preserved (shell-safe, bounded) so the served file + its Content-Disposition carry a correct
     suffix; an extensionless upload stays extensionless."""
     stem = sanitize(Path(filename).stem, 80)
-    ext = _UNSAFE.sub("", Path(filename).suffix.lstrip(".")).lower()[:16]
+    ext = _clean_upload_ext(filename)
     suffix = f".{ext}" if ext else ""
     return f".uploads/{user_id}/{stem}_a{asset_id}{suffix}"
 
 
 def upload_ext(filename: str) -> str | None:
     """The stored `container` for an uploaded file: its shell-safe lowercased extension, or None when
-    the upload has no extension (a bare name)."""
-    ext = _UNSAFE.sub("", Path(filename).suffix.lstrip(".")).lower()[:16]
-    return ext or None
+    the upload has no extension (a bare name). Shares `_clean_upload_ext` with `upload_rel_path` so the
+    stored container and the on-disk suffix can never diverge."""
+    return _clean_upload_ext(filename) or None
+
+
+# The Content-Type used to serve an inline-playable media asset, keyed by its stored `container`.
+# DELIBERATELY a strict allow-list of known audio/video containers: the served type is chosen from
+# the asset's ffprobe-verified media_kind + this table, NEVER guessed from the raw (user-controlled)
+# upload extension. Without this, an uploaded media/HTML polyglot named "evil.html" — a real MP3/MP4
+# that ffprobe classifies as media, so it dodges the generic-file force-download — would be served
+# with a Content-Type Starlette guesses from ".html" (text/html) and execute as stored XSS on the
+# serving origin. A media asset whose container isn't in this table falls back to a neutral
+# force-download attachment (see `served_media_type`), so nothing script-executable is ever inline.
+_MEDIA_MIME: dict[str, str] = {
+    # video containers
+    "mp4": "video/mp4",
+    "m4v": "video/mp4",
+    "mov": "video/quicktime",
+    "webm": "video/webm",
+    "mkv": "video/x-matroska",
+    "ogv": "video/ogg",
+    "avi": "video/x-msvideo",
+    # audio containers
+    "m4a": "audio/mp4",
+    "mp3": "audio/mpeg",
+    "aac": "audio/aac",
+    "ogg": "audio/ogg",
+    "oga": "audio/ogg",
+    "opus": "audio/ogg",
+    "weba": "audio/webm",
+    "wav": "audio/wav",
+    "flac": "audio/flac",
+}
+
+
+def served_media_type(media_kind: str | None, container: str | None) -> str | None:
+    """The Content-Type a library file MUST be served with — a real audio/video MIME for a recognised
+    media container, else ``None`` meaning "serve as a neutral force-download attachment".
+
+    Returns None for a generic ``file`` kind AND for any media whose container we don't positively
+    recognise as safe. Callers serve None as ``application/octet-stream`` + ``attachment`` so the
+    served Content-Type is NEVER derived from the (attacker-controlled) file extension — the guard
+    that stops uploaded HTML/SVG or a media/HTML polyglot from executing inline on the serving
+    origin. A ``None`` media_kind (a legacy asset before the discriminator was backfilled) is treated
+    as media so an old download keeps its inline player."""
+    if media_kind == mediakind.FILE:
+        return None
+    return _MEDIA_MIME.get((container or "").lower())
 
 
 def download_filename(display_name: str, container: str | None, path: Path) -> str:

backend/app/downloads/upload.py

@@ -16,9 +16,7 @@ source to re-download), so it doesn't age out on the retention clock the way a r
 does — it lives until the user deletes it.
 """
 
-import json
 import secrets
-import subprocess
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -29,32 +27,43 @@ from app.downloads import edit as editmod
 from app.downloads import mediakind, quota, service, storage
 from app.models import DownloadJob, MediaAsset
 
+# ffprobe container formats that mean "this is a still image / image sequence", NOT playable media.
+# A standalone jpg/png/gif/webp/bmp reports a real video stream (codec mjpeg/png/…) with no
+# `attached_pic` flag, so without this guard it would classify as media_kind='video' and get a Play
+# button + a broken in-app <video> player (and inline serving on the public share). Keyed on the
+# container format so an animated gif/webp also lands as a generic 'file' (no browser plays those in
+# a <video> either). SVG and anything else ffprobe can't decode already fails the probe → 'file'.
+_IMAGE_FORMATS = {
+    "image2",
+    "image2pipe",
+    "png_pipe",
+    "mjpeg",
+    "mjpeg_pipe",
+    "jpeg_pipe",
+    "gif",
+    "webp_pipe",
+    "bmp_pipe",
+    "tiff_pipe",
+    "ico",
+    "svg_pipe",
+}
+
 
 def _probe(path: Path) -> tuple[str | None, str | None, int | None, int | None, int | None]:
     """(vcodec, acodec, width, height, duration_s) from one ffprobe call; all None if ffprobe fails
-    or the file isn't media (a file we can't probe is treated as a generic non-media upload).
+    or the file isn't playable media (an unprobeable file — or a still image — is treated as a generic
+    non-media upload, so it becomes a download-only 'file' rather than broken 'video').
 
     A video stream flagged `attached_pic` (embedded cover art in an mp3/m4a) is skipped — it's a
     thumbnail, not a real video track, so an audio file with album art still classifies as audio."""
-    try:
-        out = subprocess.run(
-            [
-                "ffprobe",
-                "-v",
-                "error",
-                "-show_streams",
-                "-show_format",
-                "-of",
-                "json",
-                str(path),
-            ],
-            capture_output=True,
-            text=True,
-            timeout=30,
-            check=True,
-        ).stdout
-        data = json.loads(out)
-    except (subprocess.TimeoutExpired, subprocess.CalledProcessError, OSError, ValueError):
+    data = editmod.run_ffprobe_json(path, show_streams=True, show_format=True)
+    if data is None:
+        return None, None, None, None, None
+
+    fmt = data.get("format") or {}
+    fmt_names = {n.strip() for n in (fmt.get("format_name") or "").split(",")}
+    if fmt_names & _IMAGE_FORMATS:
+        # A still image / image sequence is a generic file, not video — no vcodec means media_kind='file'.
         return None, None, None, None, None
 
     vcodec = acodec = None
@@ -72,7 +81,7 @@ def _probe(path: Path) -> tuple[str | None, str | None, int | None, int | None,
 
     duration = None
     try:
-        raw = (data.get("format") or {}).get("duration")
+        raw = fmt.get("duration")
         if raw is not None:
             duration = int(float(raw))
     except (TypeError, ValueError):
@@ -117,42 +126,58 @@ def ingest_upload(
 
     rel = storage.upload_rel_path(user_id, asset.id, filename)
     storage.place_file(src_path, settings.download_root, rel)
-    final = storage.abs_path(settings.download_root, rel)
-    actual_size = final.stat().st_size if final.exists() else size
+    # From here the file is on disk under `.uploads/`. If the transaction fails before it commits, the
+    # asset/job rows roll back but the placed file would be ORPHANED — nothing references it, the
+    # per-user delete path keys off asset.rel_path (which is gone), and the GC orphan sweep skips the
+    # dot-prefixed `.uploads/` tree. So roll back the on-disk side too on any failure.
+    try:
+        final = storage.abs_path(settings.download_root, rel)
+        actual_size = final.stat().st_size if final.exists() else size
 
-    asset.status = "ready"
-    asset.rel_path = rel
-    asset.size_bytes = actual_size
-    asset.container = storage.upload_ext(filename)
-    asset.vcodec = vcodec
-    asset.acodec = acodec
-    asset.width = width
-    asset.height = height
-    asset.duration_s = duration
-    asset.media_kind = kind
-    asset.title = (display_name or "").strip() or _title_from_filename(filename)
-    asset.last_access_at = datetime.now(timezone.utc)
-    # No expires_at — see the module docstring (an upload is irreplaceable, so it doesn't age out).
-    if kind == mediakind.VIDEO:
-        # A frame poster gives the library row a thumbnail + the share link an og:image. Audio has no
-        # frame and a generic file no player, so neither gets one. Best-effort (ffmpeg decodes server
-        # side even for a codec the browser can't play).
-        asset.poster_path = editmod.ensure_poster(asset, settings.download_root)
+        asset.status = "ready"
+        asset.rel_path = rel
+        asset.size_bytes = actual_size
+        asset.container = storage.upload_ext(filename)
+        asset.vcodec = vcodec
+        asset.acodec = acodec
+        asset.width = width
+        asset.height = height
+        asset.duration_s = duration
+        asset.media_kind = kind
+        asset.title = (display_name or "").strip() or _title_from_filename(filename)
+        asset.last_access_at = datetime.now(timezone.utc)
+        # No expires_at — see the module docstring (an upload is irreplaceable, so it doesn't age out).
+        if kind == mediakind.VIDEO:
+            # A frame poster gives the library row a thumbnail + the share link an og:image. Audio has
+            # no frame and a generic file no player, so neither gets one. Best-effort (ffmpeg decodes
+            # server side even for a codec the browser can't play).
+            asset.poster_path = editmod.ensure_poster(asset, settings.download_root)
 
-    job = DownloadJob(
-        user_id=user_id,
-        asset_id=asset.id,
-        job_kind="upload",
-        source_kind="upload",
-        source_ref=ref,
-        profile_snapshot={},
-        display_name=(display_name or "").strip() or None,
-        status="done",
-        progress=100,
-        queue_pos=service.next_queue_pos(db),
-    )
-    db.add(job)
-    service.incr_ref_count(db, asset.id)
-    db.commit()
+        job = DownloadJob(
+            user_id=user_id,
+            asset_id=asset.id,
+            job_kind="upload",
+            source_kind="upload",
+            source_ref=ref,
+            profile_snapshot={},
+            display_name=(display_name or "").strip() or None,
+            status="done",
+            progress=100,
+            queue_pos=service.next_queue_pos(db),
+        )
+        db.add(job)
+        service.incr_ref_count(db, asset.id)
+        db.commit()
+    except Exception:
+        db.rollback()
+        # The rows are gone; remove the now-unreferenced file so no bytes leak. Clean the poster by its
+        # DETERMINISTIC path (rel with a .jpg suffix, the same path ensure_poster writes to) rather than
+        # ensure_poster's return value: it writes the frame directly to that path and returns None on an
+        # ffmpeg failure, so a partial poster can be on disk while the return is None — gating on the
+        # return would skip that orphan. unlink is missing_ok, so the poster path is a harmless no-op
+        # when none was generated (audio/file, or a video whose poster never started).
+        for leaked in (rel, str(Path(rel).with_suffix(".jpg"))):
+            storage.abs_path(settings.download_root, leaked).unlink(missing_ok=True)
+        raise
     db.refresh(job)
     return job

backend/app/routes/downloads.py

@@ -26,7 +26,6 @@ from app.db import get_db
 from app.downloads import credentials as dlcreds
 from app.downloads import edit as editmod
 from app.downloads import links as linksmod
-from app.downloads import mediakind
 from app.downloads import quota, service, storage
 from app.downloads import upload as uploadmod
 from app.models import (
@@ -675,17 +674,21 @@ def download_file(
 
     asset.last_access_at = datetime.now(timezone.utc)
     db.commit()
-    # A generic uploaded file is served as a neutral, force-download attachment — never with a
-    # sniffable content-type — so uploaded HTML/SVG can't execute inline on this origin. Media keeps
-    # its guessed type so the in-app player streams it. Both are range-aware (FileResponse → 206).
-    if asset.media_kind == mediakind.FILE:
+    # The served Content-Type comes from the ffprobe-verified media_kind + a container allow-list
+    # (storage.served_media_type), NEVER guessed from the file's own extension. A generic 'file' (and
+    # any media whose container we don't recognise — e.g. a media/HTML polyglot named "evil.html")
+    # returns None → a neutral force-download attachment, so uploaded HTML/SVG can't execute inline on
+    # this origin. Recognised media is served with its real type so the in-app player streams it. Both
+    # are range-aware (FileResponse → 206).
+    media_type = storage.served_media_type(asset.media_kind, asset.container)
+    if media_type is None:
         return FileResponse(
             path,
             filename=filename,
             media_type="application/octet-stream",
             content_disposition_type="attachment",
         )
-    return FileResponse(path, filename=filename)
+    return FileResponse(path, filename=filename, media_type=media_type)
 
 
 # --- editor filmstrip (scrub track) --------------------------------------------------------

backend/app/routes/public.py

@@ -15,7 +15,7 @@ from sqlalchemy.orm import Session
 from app.config import settings
 from app.db import get_db
 from app.downloads import links as linksmod
-from app.downloads import mediakind, storage
+from app.downloads import storage
 from app.models import DownloadJob, DownloadLink, MediaAsset
 from app.ratelimit import RateLimiter
 from app.security import verify_password
@@ -122,11 +122,15 @@ def watch_file(token: str, g: str | None = None, db: Session = Depends(get_db)):
         raise HTTPException(status_code=404, detail="This video is no longer available.")
 
     filename = storage.download_filename(asset.title or "video", asset.container, path)
-    # A generic uploaded file (media_kind='file') is ALWAYS a force-download attachment with a
-    # neutral type — never served inline — so arbitrary uploaded HTML/SVG can't execute on this
-    # login-free public origin. Media follows the link's own inline/attachment choice so it can play
+    # The served Content-Type comes from the asset's ffprobe-verified media_kind + a container
+    # allow-list (storage.served_media_type) — NEVER guessed from the file's own extension. A generic
+    # 'file' (and any media whose container we don't positively recognise, e.g. a media/HTML polyglot
+    # named "evil.html") returns None → a neutral force-download attachment, so arbitrary uploaded
+    # HTML/SVG can't execute inline on this login-free public origin. A recognised media container is
+    # served with its real audio/video type, inline unless the link forces a download so it can play
     # in the watch page. Both are range-aware (FileResponse → 206) for seeking/resume.
-    if asset.media_kind == mediakind.FILE:
+    media_type = storage.served_media_type(asset.media_kind, asset.container)
+    if media_type is None:
         return FileResponse(
             path,
             filename=filename,
@@ -134,7 +138,9 @@ def watch_file(token: str, g: str | None = None, db: Session = Depends(get_db)):
             content_disposition_type="attachment",
         )
     disposition = "attachment" if link.allow_download else "inline"
-    return FileResponse(path, filename=filename, content_disposition_type=disposition)
+    return FileResponse(
+        path, filename=filename, media_type=media_type, content_disposition_type=disposition
+    )
 
 
 @router.get("/watch/{token}/poster.jpg")

backend/app/sysconfig.py

@@ -127,7 +127,10 @@ SPECS: tuple[ConfigSpec, ...] = (
         min=1,
         max=100_000,
     ),
-    ConfigSpec("download_upload_max_bytes", "int", "downloads", "download_upload_max_bytes", min=0),
+    # min=1, NOT 0: the per-file cap is enforced as `if cap and size > cap`, so a stored 0 would read
+    # as falsy and SILENTLY disable the limit (an unbounded streamed write to the volume). Floor it at
+    # 1 so the cap is always active; the overall footprint is separately bounded by the byte quota.
+    ConfigSpec("download_upload_max_bytes", "int", "downloads", "download_upload_max_bytes", min=1),
     ConfigSpec(
         "download_retention_days", "int", "downloads", "download_retention_days", min=0, max=3_650
     ),

backend/tests/test_download_output.py

@@ -6,7 +6,7 @@ with "PosixPath('.') has an empty name" instead of downloading."""
 
 import pytest
 
-from app.downloads.edit import _concat_quote, parse_probe_streams
+from app.downloads.edit import _codecs_from_streams, _concat_quote, run_ffprobe_json
 from app.downloads.pipeline import _media_info, _produced_file
 from pathlib import Path, PurePosixPath
 
@@ -105,39 +105,76 @@ class TestMediaInfo:
         assert _media_info(info)["id"] == "vid"
 
 
-class TestParseProbeStreams:
+class TestCodecsFromStreams:
     # ffprobe emits codec_name BEFORE codec_type, so the old line parser shifted every name onto
     # the previous stream and an H.264+AAC mp4 came back as (aac, None) → a needless re-encode.
+    # _codecs_from_streams reads each stream dict by its own codec_type, immune to that ordering.
     def test_reads_each_stream_by_its_own_type(self):
-        payload = '{"streams":[{"codec_name":"h264","codec_type":"video"},{"codec_name":"aac","codec_type":"audio"}]}'
-        assert parse_probe_streams(payload) == ("h264", "aac")
+        streams = [
+            {"codec_name": "h264", "codec_type": "video"},
+            {"codec_name": "aac", "codec_type": "audio"},
+        ]
+        assert _codecs_from_streams(streams) == ("h264", "aac")
 
     def test_audio_first_ordering_is_handled(self):
-        payload = '{"streams":[{"codec_name":"AAC","codec_type":"audio"},{"codec_name":"H264","codec_type":"video"}]}'
-        assert parse_probe_streams(payload) == ("h264", "aac")
+        streams = [
+            {"codec_name": "AAC", "codec_type": "audio"},
+            {"codec_name": "H264", "codec_type": "video"},
+        ]
+        assert _codecs_from_streams(streams) == ("h264", "aac")
 
     def test_extra_streams_do_not_win_over_the_first_of_their_kind(self):
-        payload = (
-            '{"streams":[{"codec_name":"h264","codec_type":"video"},'
-            '{"codec_name":"vp9","codec_type":"video"},'
-            '{"codec_name":"mov_text","codec_type":"subtitle"},'
-            '{"codec_name":"opus","codec_type":"audio"}]}'
-        )
-        assert parse_probe_streams(payload) == ("h264", "opus")
+        streams = [
+            {"codec_name": "h264", "codec_type": "video"},
+            {"codec_name": "vp9", "codec_type": "video"},
+            {"codec_name": "mov_text", "codec_type": "subtitle"},
+            {"codec_name": "opus", "codec_type": "audio"},
+        ]
+        assert _codecs_from_streams(streams) == ("h264", "opus")
 
     def test_audio_only_and_video_only(self):
-        assert parse_probe_streams('{"streams":[{"codec_name":"opus","codec_type":"audio"}]}') == (
+        assert _codecs_from_streams([{"codec_name": "opus", "codec_type": "audio"}]) == (
             None,
             "opus",
         )
-        assert parse_probe_streams('{"streams":[{"codec_name":"av1","codec_type":"video"}]}') == (
-            "av1",
-            None,
-        )
+        assert _codecs_from_streams([{"codec_name": "av1", "codec_type": "video"}]) == ("av1", None)
 
-    def test_garbage_is_not_a_crash(self):
-        assert parse_probe_streams("") == (None, None)
-        assert parse_probe_streams("not json") == (None, None)
+    def test_empty_stream_list(self):
+        assert _codecs_from_streams([]) == (None, None)
+
+
+class TestRunFfprobeJson:
+    """The single ffprobe-JSON invocation point (shared by probe_codecs + upload._probe). A failed
+    probe or unparseable output must degrade to None, not raise — the malformed-input safety the old
+    parse_probe_streams string parser used to own now lives here."""
+
+    def test_unparseable_output_returns_none(self, monkeypatch):
+        import subprocess
+
+        class _R:
+            stdout = "not json"
+
+        monkeypatch.setattr(subprocess, "run", lambda *a, **k: _R())
+        assert run_ffprobe_json(Path("x")) is None
+
+    def test_ffprobe_failure_returns_none(self, monkeypatch):
+        import subprocess
+
+        def _boom(*a, **k):
+            raise subprocess.CalledProcessError(1, "ffprobe")
+
+        monkeypatch.setattr(subprocess, "run", _boom)
+        assert run_ffprobe_json(Path("x")) is None
+
+    def test_parses_streams_from_valid_json(self, monkeypatch):
+        import subprocess
+
+        class _R:
+            stdout = '{"streams":[{"codec_name":"h264","codec_type":"video"}]}'
+
+        monkeypatch.setattr(subprocess, "run", lambda *a, **k: _R())
+        data = run_ffprobe_json(Path("x"), entries="stream=codec_type,codec_name")
+        assert _codecs_from_streams(data["streams"]) == ("h264", None)
 
 
 class TestConcatQuote:

backend/tests/test_upload.py

@@ -6,10 +6,14 @@ size known up front, the per-file cap enforced while streaming, filename sanitiz
 guards that keep an upload out of paths that assume a re-fetchable source (edit / redownload).
 
 DB-backed (real Postgres) — skipped when no database is reachable. `settings.download_root` is
-pointed at a per-test tmp dir so ingest's on-disk moves are isolated; `_probe` is monkeypatched so
-the tests don't depend on ffprobe (its real behaviour is exercised by the download/edit suites).
+pointed at a per-test tmp dir so ingest's on-disk moves are isolated. The ingest/route tests
+monkeypatch `_probe` so they don't shell out to ffprobe; the classifier's OWN parsing (image →
+file, cover-art skip, dims, duration) is covered directly by `TestProbeClassifier` below, which
+stubs one level down (`run_ffprobe_json`) and exercises the real `_probe`.
 """
 
+from pathlib import Path
+
 import pytest
 
 from app.config import settings
@@ -137,6 +141,77 @@ def test_upload_route_generic_forces_download(as_user, make_user, _root, monkeyp
     assert "attachment" in fr.headers["content-disposition"].lower()
 
 
+def test_upload_route_media_html_extension_not_served_as_html(
+    as_user, make_user, _root, monkeypatch
+):
+    """The XSS guard: a MEDIA-classified upload (real codecs) named with an executable extension must
+    NOT be served with a guessed text/html type. The served Content-Type comes from the verified
+    media_kind + a container allow-list ('html' isn't one), so it force-downloads as octet-stream —
+    on both the authenticated file endpoint AND the login-free public watch endpoint. Without the fix
+    the public inline branch let a media/HTML polyglot named evil.html run as stored XSS."""
+    _probe(monkeypatch, ("h264", "aac", 320, 240, 5))  # ffprobe says it's video…
+    c = as_user(make_user())
+    r = c.post(
+        "/api/downloads/upload",
+        files={"file": ("evil.html", b"<script>alert(1)</script>", "text/html")},
+    )
+    assert r.status_code == 200, r.text
+    job = r.json()
+    assert job["asset"]["media_kind"] == "video" and job["asset"]["container"] == "html"
+
+    # Authenticated endpoint: never text/html.
+    fr = c.get(f"/api/downloads/{job['id']}/file")
+    assert fr.status_code == 200
+    assert fr.headers["content-type"].startswith("application/octet-stream")
+    assert "attachment" in fr.headers["content-disposition"].lower()
+
+    # Public login-free endpoint via a STREAM-ONLY link (the inline branch) — still not text/html.
+    link = c.post(f"/api/downloads/{job['id']}/links", json={"allow_download": False}).json()
+    pr = c.get(f"/api/public/watch/{link['token']}/file")
+    assert pr.status_code == 200
+    assert not pr.headers["content-type"].startswith("text/html")
+    assert pr.headers["content-type"].startswith("application/octet-stream")
+    assert "attachment" in pr.headers["content-disposition"].lower()
+
+
+def test_public_watch_file_forces_download_for_generic_file(as_user, make_user, _root, monkeypatch):
+    """The login-free public serve of a generic upload (media_kind='file') must be a neutral
+    force-download attachment — the branch that stops uploaded HTML/SVG executing on the public
+    origin. This is a SEPARATE code path from the authenticated /file route, so it needs its own
+    route-level assertion (previously only the authenticated branch was tested)."""
+    _probe(monkeypatch, (None, None, None, None, None))
+    c = as_user(make_user())
+    job = c.post(
+        "/api/downloads/upload",
+        files={"file": ("notes.html", b"<script>alert(1)</script>", "text/html")},
+    ).json()
+    assert job["asset"]["media_kind"] == "file"
+    link = c.post(f"/api/downloads/{job['id']}/links", json={"allow_download": True}).json()
+    pr = c.get(f"/api/public/watch/{link['token']}/file")
+    assert pr.status_code == 200
+    assert pr.headers["content-type"].startswith("application/octet-stream")
+    assert "attachment" in pr.headers["content-disposition"].lower()
+
+
+def test_public_watch_file_serves_media_inline_with_real_type(
+    as_user, make_user, _root, monkeypatch
+):
+    """A recognised media container IS still served inline with its real audio/video type (so it
+    plays in the watch page) — the fix must not break legitimate playback. A stream-only link →
+    inline; the type is video/mp4 (from the container allow-list), never text/* and never a guess."""
+    _probe(monkeypatch, ("h264", "aac", 640, 360, 12))
+    c = as_user(make_user())
+    job = c.post(
+        "/api/downloads/upload",
+        files={"file": ("clip.mp4", b"fake-mp4-bytes", "video/mp4")},
+    ).json()
+    link = c.post(f"/api/downloads/{job['id']}/links", json={"allow_download": False}).json()
+    pr = c.get(f"/api/public/watch/{link['token']}/file")
+    assert pr.status_code == 200
+    assert pr.headers["content-type"].startswith("video/mp4")
+    assert "inline" in pr.headers["content-disposition"].lower()
+
+
 def test_upload_route_enforces_per_file_cap(as_user, make_user, db, _root, monkeypatch):
     from app import sysconfig
 
@@ -151,6 +226,16 @@ def test_upload_route_enforces_per_file_cap(as_user, make_user, db, _root, monke
     assert r.json()["detail"]["code"] == "upload_too_large"
 
 
+def test_upload_cap_cannot_be_disabled_with_zero(db):
+    """download_upload_max_bytes floors at 1: a stored 0 would read falsy in the route's
+    `if cap and size > cap` guard and SILENTLY disable the per-file cap, so set_value must reject it.
+    Pins the min=1 invariant so a revert to min=0 can't pass the gate."""
+    from app import sysconfig
+
+    with pytest.raises(ValueError):
+        sysconfig.set_value(db, "download_upload_max_bytes", 0)
+
+
 def test_upload_route_rejects_empty_file(as_user, make_user, _root, monkeypatch):
     c = as_user(make_user())
     r = c.post(
@@ -182,3 +267,128 @@ def test_edit_rejects_a_generic_upload_source(as_user, make_user, _root, monkeyp
     )
     assert r.status_code == 400
     assert r.json()["detail"]["reason"] == "not_media"
+
+
+def test_ingest_rejects_over_max_jobs(db, make_user, _root, monkeypatch):
+    """The other check_upload rejection branch (holdings count, not bytes): once the user is at
+    max_jobs, a further upload is refused before any disk write."""
+    _probe(monkeypatch, (None, None, None, None, None))
+    u = make_user()
+    db.add(
+        DownloadQuota(user_id=u.id, max_bytes=10**12, max_concurrent=2, max_jobs=1, unlimited=False)
+    )
+    db.commit()
+    src1 = _root / "a.bin"
+    src1.write_bytes(b"a")
+    uploadmod.ingest_upload(db, u.id, filename="a.bin", src_path=src1, size=1)  # now 1 holding job
+    src2 = _root / "b.bin"
+    src2.write_bytes(b"b")
+    with pytest.raises(quota.QuotaExceeded) as e:
+        uploadmod.ingest_upload(db, u.id, filename="b.bin", src_path=src2, size=1)
+    assert e.value.reason == "max_jobs"
+    assert src2.exists()  # rejected before the move — nothing landed
+
+
+def test_ingest_cleans_up_placed_files_on_commit_failure(db, make_user, _root, monkeypatch):
+    """A failure AFTER storage.place_file has moved the file onto disk must roll back AND unlink the
+    placed file + any (even partial) poster — otherwise the orphan is permanent (the per-user delete
+    keys off the rolled-back asset.rel_path, and the GC orphan sweep skips the dot-prefixed .uploads/
+    tree). Exercises the rollback/cleanup branch that IS the leak remediation, incl. the poster case
+    where ensure_poster left a partial file but returned None."""
+    _probe(monkeypatch, ("h264", "aac", 320, 240, 5))  # video → the poster path runs
+    u = make_user()
+
+    # Simulate ffmpeg leaving a partial poster on disk, then failing (ensure_poster returns None).
+    def _partial_poster(asset, root):
+        storage.abs_path(root, str(Path(asset.rel_path).with_suffix(".jpg"))).write_bytes(
+            b"partial"
+        )
+        return None
+
+    monkeypatch.setattr(uploadmod.editmod, "ensure_poster", _partial_poster)
+
+    # Force the commit to fail after the file (and partial poster) are on disk.
+    def _boom():
+        raise RuntimeError("db down")
+
+    monkeypatch.setattr(db, "commit", _boom)
+
+    src = _root / "clip.mp4"
+    src.write_bytes(b"fake-video-bytes")
+    with pytest.raises(RuntimeError):
+        uploadmod.ingest_upload(db, u.id, filename="clip.mp4", src_path=src, size=16)
+
+    db.rollback()  # drop the monkeypatched-commit session state before querying
+    uploads = _root / ".uploads"
+    leftover = [p for p in uploads.rglob("*") if p.is_file()] if uploads.exists() else []
+    assert leftover == [], f"orphaned files left on disk: {leftover}"
+    assert db.query(MediaAsset).count() == 0  # the flushed asset rolled back — no dangling row
+
+
+# --- classifier: the REAL _probe parsing (stubs ffprobe one level down) ----------------------
+# The ingest/route tests above stub _probe wholesale; these exercise _probe's OWN logic — image
+# detection, the cover-art skip, dimension/duration parsing — by faking only run_ffprobe_json, so a
+# regression in the classifier the force-download guarantee leans on is actually caught.
+
+
+class TestProbeClassifier:
+    def _probe_with(self, monkeypatch, data):
+        monkeypatch.setattr(uploadmod.editmod, "run_ffprobe_json", lambda *a, **k: data)
+        return uploadmod._probe(Path("x"))
+
+    def test_probe_failure_is_a_generic_file(self, monkeypatch):
+        # run_ffprobe_json returns None (ffprobe failed / not media) → all None → derive() == FILE.
+        v, a, w, h, d = self._probe_with(monkeypatch, None)
+        assert (v, a, w, h, d) == (None, None, None, None, None)
+        assert mediakind.derive(v, a) == mediakind.FILE
+
+    @pytest.mark.parametrize("fmt", ["image2", "png_pipe", "jpeg_pipe", "webp_pipe", "gif"])
+    def test_still_image_classifies_as_file_not_video(self, monkeypatch, fmt):
+        # A standalone image exposes a real video stream, but its container format is an image demuxer
+        # — it must NOT become media_kind='video' (broken player + inline serving). Stays a file.
+        data = {
+            "format": {"format_name": fmt},
+            "streams": [{"codec_type": "video", "codec_name": "png", "width": 8, "height": 8}],
+        }
+        v, a, w, h, d = self._probe_with(monkeypatch, data)
+        assert v is None and mediakind.derive(v, a) == mediakind.FILE
+
+    def test_cover_art_video_stream_is_skipped_audio_stays_audio(self, monkeypatch):
+        # Album art embedded in an audio file is a video stream flagged attached_pic — skip it so the
+        # asset classifies as audio, not video.
+        data = {
+            "format": {"format_name": "mp3", "duration": "212.5"},
+            "streams": [
+                {
+                    "codec_type": "video",
+                    "codec_name": "mjpeg",
+                    "disposition": {"attached_pic": 1},
+                    "width": 600,
+                    "height": 600,
+                },
+                {"codec_type": "audio", "codec_name": "mp3"},
+            ],
+        }
+        v, a, w, h, d = self._probe_with(monkeypatch, data)
+        assert (v, a, d) == (None, "mp3", 212)
+        assert mediakind.derive(v, a) == mediakind.AUDIO
+
+    def test_real_video_records_codecs_dims_duration(self, monkeypatch):
+        data = {
+            "format": {"format_name": "mov,mp4,m4a", "duration": "60.0"},
+            "streams": [
+                {"codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080},
+                {"codec_type": "audio", "codec_name": "aac"},
+            ],
+        }
+        v, a, w, h, d = self._probe_with(monkeypatch, data)
+        assert (v, a, w, h, d) == ("h264", "aac", 1920, 1080, 60)
+        assert mediakind.derive(v, a) == mediakind.VIDEO
+
+    def test_unparseable_duration_is_none(self, monkeypatch):
+        data = {
+            "format": {"format_name": "wav", "duration": "N/A"},
+            "streams": [{"codec_type": "audio", "codec_name": "pcm_s16le"}],
+        }
+        v, a, w, h, d = self._probe_with(monkeypatch, data)
+        assert d is None and mediakind.derive(v, a) == mediakind.AUDIO