feat(downloads): public share-by-link + watch page backend

Share a download by a capability URL (/watch/{token}) that anyone can play on a login-free page —
distinct from the registered-user ACL share. Per-link controls: optional expiry, allow-download
toggle (stream-only vs downloadable), optional argon2 password. Revoke = delete.
- migration 0042 + DownloadLink model; app/downloads/links.py (token, HMAC grant sign/verify)
- owner endpoints (POST/GET/PATCH/DELETE links) + /recipients for the internal user picker
- public router (no auth): watch meta / password unlock→signed grant / range-aware file serve
  (inline vs attachment); unlock is rate-limited; meta verifies the file exists on disk
Verified end-to-end: 206 range, inline vs attachment, wrong-password 403, tampered-grant 403.

backend/alembic/versions/0042_download_links.py

@@ -0,0 +1,45 @@
+"""public watch links for downloads (share-by-link)
+
+Revision ID: 0042_download_links
+Revises: 0041_download_edit
+Create Date: 2026-07-04
+
+A `download_links` row is a capability URL for one download: anyone holding the unguessable token
+can watch the file on the public `/watch/{token}` player page — no account needed. Optional
+per-link controls: an expiry, an "allow download" toggle, and an (argon2-hashed) password. Revoke
+= delete the row. Distinct from `download_shares` (an ACL grant to another *registered* user).
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0042_download_links"
+down_revision: Union[str, None] = "0041_download_edit"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "download_links",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("job_id", sa.Integer(), nullable=False),
+        sa.Column("token", sa.String(length=64), nullable=False),
+        sa.Column("created_by", sa.Integer(), nullable=True),
+        sa.Column("allow_download", sa.Boolean(), server_default="false", nullable=False),
+        sa.Column("password_hash", sa.Text(), nullable=True),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("view_count", sa.Integer(), server_default="0", nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["job_id"], ["download_jobs.id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["created_by"], ["users.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("token", name="uq_download_link_token"),
+    )
+    op.create_index("ix_download_links_job_id", "download_links", ["job_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_download_links_job_id", table_name="download_links")
+    op.drop_table("download_links")

backend/app/downloads/links.py

@@ -0,0 +1,81 @@
+"""Public "share by link" helpers for the download center.
+
+A DownloadLink is a capability URL: the unguessable `token` grants anyone read access to one
+download's file on the login-free `/watch/{token}` page. Password-protected links additionally
+require a short-lived signed **grant** on the file request — because a `<video src>` can't send a
+header, the password is exchanged once at `/unlock` for an HMAC grant appended to the file URL,
+so the raw password never rides in the media URL (or server logs).
+"""
+import hashlib
+import hmac
+import secrets
+from datetime import datetime, timezone
+
+from app.config import settings
+from app.models import DownloadLink
+
+_GRANT_TTL = 6 * 3600  # a watch session's grant is good for 6h, then the viewer re-unlocks
+
+
+def new_token() -> str:
+    return secrets.token_urlsafe(24)  # ~32 chars, 192 bits
+
+
+def is_expired(link: DownloadLink) -> bool:
+    return link.expires_at is not None and link.expires_at <= datetime.now(timezone.utc)
+
+
+# --- signed grants (password-protected file access) ----------------------------------------
+
+def _sign(msg: str) -> str:
+    return hmac.new(
+        settings.secret_key.encode(), msg.encode(), hashlib.sha256
+    ).hexdigest()[:32]
+
+
+def make_grant(token: str, now: float | None = None) -> str:
+    """A short-lived HMAC grant proving the viewer unlocked `token`. Format: `<exp>.<sig>`."""
+    exp = int((now if now is not None else datetime.now(timezone.utc).timestamp())) + _GRANT_TTL
+    return f"{exp}.{_sign(f'{token}.{exp}')}"
+
+
+def check_grant(token: str, grant: str | None) -> bool:
+    if not grant or "." not in grant:
+        return False
+    exp_s, sig = grant.split(".", 1)
+    try:
+        exp = int(exp_s)
+    except ValueError:
+        return False
+    if exp < datetime.now(timezone.utc).timestamp():
+        return False
+    return hmac.compare_digest(sig, _sign(f"{token}.{exp}"))
+
+
+# --- serialization -------------------------------------------------------------------------
+
+def owner_view(link: DownloadLink, base_url: str = "") -> dict:
+    """What the file owner sees when managing their links."""
+    return {
+        "id": link.id,
+        "token": link.token,
+        "url": f"{base_url}/watch/{link.token}",
+        "allow_download": link.allow_download,
+        "has_password": link.password_hash is not None,
+        "expires_at": link.expires_at.isoformat() if link.expires_at else None,
+        "view_count": link.view_count,
+        "created_at": link.created_at.isoformat() if link.created_at else None,
+    }
+
+
+def public_meta(link: DownloadLink, asset, file_url: str) -> dict:
+    """Metadata for the public watch page (only ever returned once access is authorized)."""
+    return {
+        "title": asset.title,
+        "uploader": asset.uploader,
+        "duration_s": asset.duration_s,
+        "width": asset.width,
+        "height": asset.height,
+        "allow_download": link.allow_download,
+        "file_url": file_url,
+    }

backend/app/main.py

@@ -38,6 +38,7 @@ from app.routes import (
     messages,
     notifications,
     playlists,
+    public as public_routes,
     quota,
     saved_views,
     scheduler as scheduler_routes,
@@ -125,6 +126,7 @@ app.include_router(playlists.router)
 app.include_router(saved_views.router)
 app.include_router(downloads.router)
 app.include_router(downloads.admin_router)
+app.include_router(public_routes.router)
 app.include_router(admin.router)
 app.include_router(config_routes.router)
 app.include_router(scheduler_routes.router)

backend/app/models.py

@@ -867,6 +867,33 @@ class DownloadQuota(Base, UpdatedAtMixin):
     )
 
 
+class DownloadLink(Base, TimestampMixin):
+    """A public capability link to one download — the "share by link" surface.
+
+    Anyone holding the unguessable `token` can watch the file on the login-free `/watch/{token}`
+    player page (Google-Drive style). Optional per-link controls: `expires_at`, `allow_download`
+    (stream-only vs downloadable), and a `password_hash` (argon2, like a user password). Revoke =
+    delete the row. Distinct from `DownloadShare`, which grants a *registered* user access via the
+    in-app "Shared with me" list."""
+
+    __tablename__ = "download_links"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    job_id: Mapped[int] = mapped_column(
+        ForeignKey("download_jobs.id", ondelete="CASCADE"), index=True
+    )
+    token: Mapped[str] = mapped_column(String(64), unique=True)
+    created_by: Mapped[int | None] = mapped_column(
+        ForeignKey("users.id", ondelete="CASCADE")
+    )
+    allow_download: Mapped[bool] = mapped_column(
+        Boolean, default=False, server_default="false"
+    )
+    password_hash: Mapped[str | None] = mapped_column(Text)
+    expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    view_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
+
+
 class MessageKey(Base, TimestampMixin):
     """A user's end-to-end-encryption key material for direct messaging (phase 2).
 

backend/app/routes/downloads.py

@@ -6,7 +6,7 @@ worker is a separate process); this module never downloads — it enqueues, mana
 serves finished files (range-aware, with the user's custom display name), and shares them.
 """
 import re
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from urllib.parse import parse_qs, urlparse
 
@@ -19,15 +19,18 @@ from app.auth import admin_user, require_human
 from app.config import settings
 from app.db import get_db
 from app.downloads import edit as editmod
+from app.downloads import links as linksmod
 from app.downloads import quota, service, storage
 from app.models import (
     DownloadJob,
+    DownloadLink,
     DownloadProfile,
     DownloadQuota,
     DownloadShare,
     MediaAsset,
     User,
 )
+from app.security import hash_password
 
 router = APIRouter(prefix="/api/downloads", tags=["downloads"])
 admin_router = APIRouter(prefix="/api/admin/downloads", tags=["admin-downloads"])
@@ -616,6 +619,119 @@ def unshare_download(
     return {"ok": True}
 
 
+# --- recipients (for the internal "share with a user" picker) ------------------------------
+
+@router.get("/recipients")
+def share_recipients(
+    user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> list[dict]:
+    """Registered users this user can share a download with (excludes self + the demo account)."""
+    rows = (
+        db.execute(
+            select(User)
+            .where(User.id != user.id, User.is_demo.is_(False))
+            .order_by(func.lower(User.email))
+        )
+        .scalars()
+        .all()
+    )
+    return [{"id": u.id, "email": u.email, "display_name": u.display_name} for u in rows]
+
+
+# --- public watch links (share by link) ----------------------------------------------------
+
+def _own_link(db: Session, user: User, link_id: int) -> DownloadLink:
+    link = db.get(DownloadLink, link_id)
+    if link is None:
+        raise HTTPException(status_code=404, detail="Unknown link")
+    job = db.get(DownloadJob, link.job_id)
+    if job is None or job.user_id != user.id:
+        raise HTTPException(status_code=404, detail="Unknown link")
+    return link
+
+
+def _link_expiry(payload: dict) -> datetime | None:
+    days = payload.get("expires_days")
+    if days in (None, "", 0, "0"):
+        return None
+    try:
+        d = int(days)
+    except (TypeError, ValueError):
+        raise HTTPException(status_code=400, detail="expires_days must be a number")
+    if d <= 0:
+        return None
+    return datetime.now(timezone.utc) + timedelta(days=min(d, 3650))
+
+
+@router.get("/{job_id}/links")
+def list_links(
+    job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> list[dict]:
+    _own_job(db, user, job_id)
+    rows = (
+        db.execute(
+            select(DownloadLink).where(DownloadLink.job_id == job_id).order_by(DownloadLink.id.desc())
+        )
+        .scalars()
+        .all()
+    )
+    return [linksmod.owner_view(l) for l in rows]
+
+
+@router.post("/{job_id}/links")
+def create_link(
+    job_id: int,
+    payload: dict,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    job = _own_job(db, user, job_id)
+    if job.status != "done" or job.asset_id is None:
+        raise HTTPException(status_code=409, detail="Only a finished download can be shared.")
+    pw = (payload.get("password") or "").strip()
+    link = DownloadLink(
+        job_id=job_id,
+        token=linksmod.new_token(),
+        created_by=user.id,
+        allow_download=bool(payload.get("allow_download")),
+        password_hash=hash_password(pw) if pw else None,
+        expires_at=_link_expiry(payload),
+    )
+    db.add(link)
+    db.commit()
+    db.refresh(link)
+    return linksmod.owner_view(link)
+
+
+@router.patch("/links/{link_id}")
+def update_link(
+    link_id: int,
+    payload: dict,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    link = _own_link(db, user, link_id)
+    if "allow_download" in payload:
+        link.allow_download = bool(payload["allow_download"])
+    if "expires_days" in payload:
+        link.expires_at = _link_expiry(payload)
+    if "password" in payload:
+        pw = (payload.get("password") or "").strip()
+        link.password_hash = hash_password(pw) if pw else None
+    db.commit()
+    return linksmod.owner_view(link)
+
+
+@router.delete("/links/{link_id}")
+def revoke_link(
+    link_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    link = _own_link(db, user, link_id)
+    db.delete(link)
+    db.commit()
+    return {"deleted": link_id}
+
+
 # --- admin ---------------------------------------------------------------------------------
 
 @admin_router.get("")

backend/app/routes/public.py

@@ -0,0 +1,109 @@
+"""Public, login-free surface for shared download links (the `/watch/{token}` player page).
+
+These routes are deliberately OUTSIDE the auth/`require_human` gate — the unguessable token IS the
+credential (a capability URL). They only ever expose one file's stream + minimal metadata, never
+the app or any account data. Password-protected links exchange the password once at `/unlock` for
+a short-lived signed grant that the media URL carries (a `<video src>` can't send a header).
+"""
+from pathlib import Path
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import FileResponse
+from sqlalchemy import select
+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 storage
+from app.models import DownloadJob, DownloadLink, MediaAsset
+from app.ratelimit import RateLimiter
+from app.security import verify_password
+
+router = APIRouter(prefix="/api/public", tags=["public"])
+
+# Slow password guessing per link (the token itself is 192-bit, so it needs no throttle).
+_unlock_limit = RateLimiter(max_events=10, window_seconds=300)
+
+
+def _link_or_404(db: Session, token: str) -> DownloadLink:
+    link = db.execute(
+        select(DownloadLink).where(DownloadLink.token == token)
+    ).scalar_one_or_none()
+    if link is None or linksmod.is_expired(link):
+        raise HTTPException(status_code=404, detail="This link is no longer available.")
+    return link
+
+
+def _ready_asset(db: Session, link: DownloadLink) -> MediaAsset:
+    job = db.get(DownloadJob, link.job_id)
+    asset = db.get(MediaAsset, job.asset_id) if job and job.asset_id else None
+    if asset is None or asset.status != "ready" or not asset.rel_path:
+        raise HTTPException(status_code=404, detail="This video is no longer available.")
+    # Verify the file is actually on disk here (not just flagged ready), so the watch page never
+    # loads a player for a missing file — meta and file agree.
+    path = storage.abs_path(settings.download_root, asset.rel_path).resolve()
+    root = Path(settings.download_root).resolve()
+    if root not in path.parents or not path.exists():
+        raise HTTPException(status_code=404, detail="This video is no longer available.")
+    return asset
+
+
+def _file_url(token: str, grant: str | None = None) -> str:
+    base = f"/api/public/watch/{token}/file"
+    return f"{base}?g={grant}" if grant else base
+
+
+@router.get("/watch/{token}")
+def watch_meta(token: str, db: Session = Depends(get_db)) -> dict:
+    """Public metadata for the watch page. For a password link, returns only `needs_password`
+    until the viewer unlocks it (no title/thumbnail leak)."""
+    link = _link_or_404(db, token)
+    if link.password_hash:
+        return {"needs_password": True}
+    asset = _ready_asset(db, link)
+    link.view_count += 1
+    db.commit()
+    return {"needs_password": False, **linksmod.public_meta(link, asset, _file_url(token))}
+
+
+@router.post("/watch/{token}/unlock")
+def watch_unlock(token: str, payload: dict, db: Session = Depends(get_db)) -> dict:
+    """Exchange the link password for a signed grant carried by the media URL."""
+    if not _unlock_limit.allow(token):
+        raise HTTPException(status_code=429, detail="Too many attempts. Try again later.")
+    link = _link_or_404(db, token)
+    if not link.password_hash:
+        # Not a password link — nothing to unlock; just hand back the plain meta.
+        asset = _ready_asset(db, link)
+        return {"needs_password": False, **linksmod.public_meta(link, asset, _file_url(token))}
+    if not verify_password((payload.get("password") or ""), link.password_hash):
+        raise HTTPException(status_code=403, detail="Wrong password.")
+    asset = _ready_asset(db, link)
+    link.view_count += 1
+    db.commit()
+    grant = linksmod.make_grant(token)
+    return {"needs_password": False, **linksmod.public_meta(link, asset, _file_url(token, grant))}
+
+
+@router.get("/watch/{token}/file")
+def watch_file(token: str, g: str | None = None, db: Session = Depends(get_db)):
+    """Range-aware stream of the shared file. Inline by default (plays in the page); an
+    `allow_download` link serves it as an attachment. Password links require a valid grant."""
+    link = _link_or_404(db, token)
+    if link.password_hash and not linksmod.check_grant(token, g):
+        raise HTTPException(status_code=403, detail="This link needs a password.")
+    asset = _ready_asset(db, link)
+    path = storage.abs_path(settings.download_root, asset.rel_path).resolve()
+    root = Path(settings.download_root).resolve()
+    if root not in path.parents or not path.exists():
+        raise HTTPException(status_code=404, detail="This video is no longer available.")
+
+    ext = asset.container or path.suffix.lstrip(".")
+    base = storage.display_filename(asset.title or "video")
+    if base.lower().endswith(f".{ext.lower()}"):
+        base = base[: -(len(ext) + 1)]
+    filename = f"{base}.{ext}"
+    disposition = "attachment" if link.allow_download else "inline"
+    # Starlette's FileResponse honours the Range header (206) for seeking/scrubbing.
+    return FileResponse(path, filename=filename, content_disposition_type=disposition)