feat(files): share-link tokens are kept only hashed and encrypted, and every link change is audited

A /watch link's token is the credential, and it sat in the database as written. Migration 0107
replaces the column with the token's SHA-256 (the lookup) and a Fernet-encrypted copy (so the owner
can copy the same link again); every existing address keeps working, and the downgrade decrypts
them back. Without a copy this key can read, a link still opens but its address cannot be shown:
"Copy link" then takes the newest link that has one or makes a new one, and the Share dialog says
so in that row instead of a broken URL. Making, changing and revoking a link are audited, never
with its token or password.

backend/alembic/versions/0107_link_token_at_rest.py

@@ -0,0 +1,66 @@
+"""Share-link tokens no longer stored in the clear
+
+Revision ID: 0107_link_token_at_rest
+Revises: 0106_api_tokens
+Create Date: 2026-10-15
+
+A `/watch` link's token is the credential itself, and it sat in `download_links.token` as written,
+so a copy of the database was every shared file. Now a link is found by the token's SHA-256
+(`token_hash`), and the token is kept only Fernet-encrypted (`token_enc`, TOKEN_ENCRYPTION_KEY) so its
+owner can copy the same link again. Without a key the copy is not kept: the link still opens, its
+address just cannot be shown again (`app/downloads/links.py`).
+
+Every existing link is converted here, so the addresses already sent keep working. The downgrade
+writes the tokens back from their encrypted copies; a link with none left (no key, or a key since
+changed) gets a fresh random token, which is the same as revoking it.
+"""
+
+import hashlib
+import secrets
+
+import sqlalchemy as sa
+from alembic import op
+
+from app import security
+
+revision: str = "0107_link_token_at_rest"
+down_revision: str | None = "0106_api_tokens"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+
+def upgrade() -> None:
+    op.add_column("download_links", sa.Column("token_hash", sa.String(64)))
+    op.add_column("download_links", sa.Column("token_enc", sa.Text()))
+    conn = op.get_bind()
+    sealing = security.can_encrypt()
+    for link_id, token in conn.execute(sa.text("SELECT id, token FROM download_links")).fetchall():
+        conn.execute(
+            sa.text("UPDATE download_links SET token_hash = :h, token_enc = :e WHERE id = :id"),
+            {
+                "id": link_id,
+                "h": hashlib.sha256(token.encode()).hexdigest(),
+                "e": security.encrypt(token) if sealing else None,
+            },
+        )
+    op.alter_column("download_links", "token_hash", nullable=False)
+    op.create_unique_constraint("uq_download_link_token_hash", "download_links", ["token_hash"])
+    op.drop_constraint("uq_download_link_token", "download_links", type_="unique")
+    op.drop_column("download_links", "token")
+
+
+def downgrade() -> None:
+    op.add_column("download_links", sa.Column("token", sa.String(64)))
+    conn = op.get_bind()
+    rows = conn.execute(sa.text("SELECT id, token_enc FROM download_links")).fetchall()
+    for link_id, enc in rows:
+        token = security.decrypt_optional(enc) or secrets.token_urlsafe(24)
+        conn.execute(
+            sa.text("UPDATE download_links SET token = :t WHERE id = :id"),
+            {"id": link_id, "t": token},
+        )
+    op.alter_column("download_links", "token", nullable=False)
+    op.create_unique_constraint("uq_download_link_token", "download_links", ["token"])
+    op.drop_constraint("uq_download_link_token_hash", "download_links", type_="unique")
+    op.drop_column("download_links", "token_enc")
+    op.drop_column("download_links", "token_hash")

backend/app/audit.py

@@ -73,6 +73,11 @@ class AuditAction:
     # What a token then did is the file rows above, their summary naming it.
     API_TOKEN_CREATE = "api_token_create"
     API_TOKEN_REVOKE = "api_token_revoke"
+    # A public `/watch` link to a file (`routes.downloads`): made, its settings changed, revoked.
+    # Never its token or password — only whether it has one.
+    DOWNLOAD_LINK_CREATE = "download_link_create"
+    DOWNLOAD_LINK_UPDATE = "download_link_update"
+    DOWNLOAD_LINK_REVOKE = "download_link_revoke"
 
 
 def record(

backend/app/downloads/links.py

@@ -1,10 +1,14 @@
 """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).
+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. So the token is never stored as written:
+the link is found by its SHA-256, and the owner's copy is Fernet-encrypted — a database copy alone
+opens nothing. Without that copy (no TOKEN_ENCRYPTION_KEY, or a key since changed) the link still
+opens; only its address cannot be shown again, and the owner makes a new link to share.
+
+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
@@ -12,6 +16,10 @@ import hmac
 import secrets
 from datetime import datetime, timezone
 
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app import security
 from app.config import settings
 from app.models import DownloadLink
 
@@ -22,6 +30,24 @@ def new_token() -> str:
     return secrets.token_urlsafe(24)  # ~32 chars, 192 bits
 
 
+def seal(link: DownloadLink, token: str) -> None:
+    """Store `token` on `link` the only ways it is kept: its hash, and its encrypted copy."""
+    link.token_hash = security.hash_token(token)
+    link.token_enc = security.encrypt(token) if security.can_encrypt() else None
+
+
+def find(db: Session, token: str) -> DownloadLink | None:
+    """The link a `/watch` token names, by its hash."""
+    return db.execute(
+        select(DownloadLink).where(DownloadLink.token_hash == security.hash_token(token))
+    ).scalar_one_or_none()
+
+
+def recover(link: DownloadLink) -> str | None:
+    """The link's token from its encrypted copy, or None when there is none this key can read."""
+    return security.decrypt_optional(link.token_enc)
+
+
 def is_expired(link: DownloadLink) -> bool:
     return link.expires_at is not None and link.expires_at <= datetime.now(timezone.utc)
 
@@ -61,12 +87,14 @@ def check_grant(token: str, grant: str | None, password_version: int) -> bool:
 # --- serialization -------------------------------------------------------------------------
 
 
-def owner_view(link: DownloadLink, base_url: str = "") -> dict:
-    """What the file owner sees when managing their links."""
+def owner_view(link: DownloadLink, base_url: str = "", token: str | None = None) -> dict:
+    """What the file owner sees when managing their links. `token` is the one just made (it is in
+    hand even where no copy is kept); else it is read back, and `url` is None when it cannot be."""
+    token = token or recover(link)
     return {
         "id": link.id,
-        "token": link.token,
-        "url": f"{base_url}/watch/{link.token}",
+        "token": token,
+        "url": f"{base_url}/watch/{token}" if token else None,
         "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,

backend/app/downloads/og.py

@@ -11,13 +11,12 @@ import html
 import re
 from pathlib import Path
 
-from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.config import settings
 from app.downloads import links as linksmod
 from app.downloads import storage
-from app.models import DownloadJob, DownloadLink, MediaAsset
+from app.models import DownloadJob, MediaAsset
 
 _OG_BLOCK = re.compile(r"<!--OG:START-->.*?<!--OG:END-->", re.DOTALL)
 
@@ -74,7 +73,7 @@ def _jpeg_size(path: Path) -> tuple[int, int] | None:
 
 def _watch_tags(token: str, db: Session) -> str | None:
     """Build the OG/Twitter meta block for a watch token, or None to keep the generic card."""
-    link = db.execute(select(DownloadLink).where(DownloadLink.token == token)).scalar_one_or_none()
+    link = linksmod.find(db, token)
     if link is None or linksmod.is_expired(link):
         return None
     url = f"{settings.app_base}/watch/{token}"

backend/app/models.py

@@ -1131,8 +1131,10 @@ 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`
+    Anyone holding the unguessable token can watch the file on the login-free `/watch/{token}`
+    player page (Google-Drive style). The token itself is never stored as written: a link is found
+    by its SHA-256 (`token_hash`), and `token_enc` is the Fernet-encrypted copy its owner copies the
+    link from again — absent without a key (`downloads/links.py`). 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."""
@@ -1143,7 +1145,8 @@ class DownloadLink(Base, TimestampMixin):
     job_id: Mapped[int] = mapped_column(
         ForeignKey("download_jobs.id", ondelete="CASCADE"), index=True
     )
-    token: Mapped[str] = mapped_column(String(64), unique=True)
+    token_hash: Mapped[str] = mapped_column(String(64), unique=True)
+    token_enc: Mapped[str | None] = mapped_column(Text)  # Fernet ciphertext — never returned/logged
     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)

backend/app/routes/downloads.py

@@ -62,6 +62,7 @@ from app.routes.files import (
 )
 from app.security import hash_password
 from app.userscope import is_messageable_user, messageable_clauses
+from app.utils import iso
 from app.refusal import refused
 
 router = APIRouter(prefix="/api/downloads", tags=["downloads"])
@@ -769,18 +770,37 @@ def create_link(
     if job.status != JobStatus.DONE or job.asset_id is None:
         raise refused(409, "share_unfinished")
     pw = body.password.strip()
+    token = linksmod.new_token()
     link = DownloadLink(
         job_id=job_id,
-        token=linksmod.new_token(),
         created_by=user.id,
         allow_download=body.allow_download,
         password_hash=hash_password(pw) if pw else None,
         expires_at=_link_expiry(body.expires_days),
     )
+    linksmod.seal(link, token)
     db.add(link)
+    db.flush()
+    audit.record(
+        db,
+        user.id,
+        audit.AuditAction.DOWNLOAD_LINK_CREATE,
+        target_type="download_link",
+        target_id=link.id,
+        after={"job_id": job_id, **_link_audit(link)},
+    )
     db.commit()
     db.refresh(link)
-    return linksmod.owner_view(link)
+    return linksmod.owner_view(link, token=token)
+
+
+def _link_audit(link: DownloadLink) -> dict:
+    """A link's settings as the audit records them: whether it has a password, never which."""
+    return {
+        "allow_download": link.allow_download,
+        "has_password": link.password_hash is not None,
+        "expires_at": iso(link.expires_at),
+    }
 
 
 class UpdateLinkIn(BaseModel):
@@ -797,6 +817,7 @@ def update_link(
     db: Session = Depends(get_db),
 ) -> dict:
     link = _own_link(db, user, link_id)
+    before = _link_audit(link)
     sent = body.model_fields_set
     if "allow_download" in sent:
         link.allow_download = bool(body.allow_download)
@@ -808,6 +829,17 @@ def update_link(
         # Bump the version so any grant issued under the previous password stops working now, rather
         # than living out its 6h TTL (setting/clearing/changing the password all count as a rotation).
         link.password_version += 1
+    after = _link_audit(link)
+    if after != before or "password" in sent:
+        audit.record(
+            db,
+            user.id,
+            audit.AuditAction.DOWNLOAD_LINK_UPDATE,
+            target_type="download_link",
+            target_id=link_id,
+            before=before,
+            after=after,
+        )
     db.commit()
     return linksmod.owner_view(link)
 
@@ -818,8 +850,16 @@ def revoke_link(
 ) -> dict:
     link = _own_link(db, user, link_id)
     db.delete(link)
+    audit.record(
+        db,
+        user.id,
+        audit.AuditAction.DOWNLOAD_LINK_REVOKE,
+        target_type="download_link",
+        target_id=link_id,
+        after={"job_id": link.job_id},
+    )
     db.commit()
-    return {"deleted": link_id}
+    return {"ok": True, "deleted": link_id}
 
 
 # --- admin ---------------------------------------------------------------------------------

backend/app/routes/public.py

@@ -9,7 +9,6 @@ a short-lived signed grant that the media URL carries (a `<video src>` can't sen
 from fastapi import APIRouter, Depends, Request
 from fastapi.responses import FileResponse
 from pydantic import BaseModel
-from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.config import settings
@@ -38,7 +37,7 @@ _unlock_failures = RateLimiter(max_events=50, 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()
+    link = linksmod.find(db, token)
     if link is None or linksmod.is_expired(link):
         raise refused(404, "link_gone")
     return link

backend/app/security.py

@@ -46,6 +46,12 @@ _fernet: Fernet | None = (
 )
 
 
+def can_encrypt() -> bool:
+    """Whether a value can be stored encrypted here: TOKEN_ENCRYPTION_KEY is set. Production
+    refuses to start without it (`config`); a local box may run without one."""
+    return _fernet is not None
+
+
 def _require_fernet() -> Fernet:
     if _fernet is None:
         raise RuntimeError("TOKEN_ENCRYPTION_KEY is not configured")

backend/tests/test_link_token_at_rest.py

@@ -0,0 +1,187 @@
+"""A `/watch` link's token at rest (downloads/links.py, migration 0107), DB-lane.
+
+The token is the credential, so the database keeps only its SHA-256 (to find the link by) and a
+Fernet-encrypted copy (for its owner to copy the same link again). Without a copy this key can read,
+the link still opens and only its address cannot be shown again. Making, changing and revoking a
+link is audited, never with its token or password.
+"""
+
+import json
+
+import pytest
+from alembic import command
+from alembic.config import Config
+from cryptography.fernet import Fernet
+from sqlalchemy import select, text
+
+from app import security
+from app.config import settings
+from app.db import engine
+from app.downloads import probe as probemod
+from app.models import AuditLog, DownloadLink
+
+_VIDEO = {
+    "streams": [
+        {"codec_type": "video", "codec_name": "h264", "pix_fmt": "yuv420p"},
+        {"codec_type": "audio", "codec_name": "aac"},
+    ],
+    "format": {"format_name": "mov,mp4", "duration": "5"},
+}
+
+
+@pytest.fixture
+def key(monkeypatch):
+    monkeypatch.setattr(security, "_fernet", Fernet(Fernet.generate_key()))
+
+
+@pytest.fixture
+def owner_job(as_user, make_user, tmp_path, monkeypatch, upload_headers):
+    monkeypatch.setattr(settings, "files_root", str(tmp_path))
+    monkeypatch.setattr(probemod, "probe_json", lambda _p: _VIDEO)
+    owner = as_user(make_user())
+    job = owner.post("/api/files/upload", content=b"fake-video", headers=upload_headers("clip.mp4"))
+    assert job.status_code == 200, job.text
+    return owner, job.json()["id"]
+
+
+def _row(db, link_id: int) -> dict:
+    db.expire_all()
+    row = db.execute(
+        text("SELECT * FROM download_links WHERE id = :id"), {"id": link_id}
+    ).mappings()
+    return dict(row.one())
+
+
+def _audits(db, action: str) -> list[AuditLog]:
+    db.expire_all()
+    return list(db.execute(select(AuditLog).where(AuditLog.action == action)).scalars())
+
+
+def test_the_token_is_kept_only_hashed_and_encrypted(db, key, owner_job, client):
+    owner, job_id = owner_job
+    made = owner.post(f"/api/downloads/{job_id}/links", json={}).json()
+    token = made["token"]
+    assert made["url"] == f"/watch/{token}"
+
+    row = _row(db, made["id"])
+    assert "token" not in row
+    assert token not in json.dumps(row, default=str)
+    assert row["token_hash"] == security.hash_token(token)
+    assert security.decrypt(row["token_enc"]) == token
+
+    # The address it was given opens it, and its owner can copy the same one again.
+    assert client.get(f"/api/public/watch/{token}").status_code == 200
+    listed = owner.get(f"/api/downloads/{job_id}/links").json()
+    assert [entry["url"] for entry in listed] == [f"/watch/{token}"]
+
+
+def test_a_token_naming_no_link_finds_none(db, key, owner_job, client):
+    owner, job_id = owner_job
+    token = owner.post(f"/api/downloads/{job_id}/links", json={}).json()["token"]
+    other = token[:-1] + ("A" if token[-1] != "A" else "B")
+    assert client.get(f"/api/public/watch/{other}").status_code == 404
+    # Its hash is no key either: the lookup hashes what it is given.
+    hashed = security.hash_token(token)
+    assert client.get(f"/api/public/watch/{hashed}").status_code == 404
+
+
+def test_without_a_key_the_link_opens_but_its_address_is_not_kept(db, owner_job, client):
+    assert not security.can_encrypt()
+    owner, job_id = owner_job
+    made = owner.post(f"/api/downloads/{job_id}/links", json={}).json()
+    # The one who made it has it in hand…
+    assert made["url"] == f"/watch/{made['token']}"
+    assert _row(db, made["id"])["token_enc"] is None
+    assert client.get(f"/api/public/watch/{made['token']}").status_code == 200
+    # …but it cannot be shown again.
+    listed = owner.get(f"/api/downloads/{job_id}/links").json()
+    assert [(entry["token"], entry["url"]) for entry in listed] == [(None, None)]
+
+
+def test_a_changed_key_hides_the_address_and_the_link_still_opens(
+    db, key, owner_job, client, monkeypatch
+):
+    owner, job_id = owner_job
+    token = owner.post(f"/api/downloads/{job_id}/links", json={}).json()["token"]
+    monkeypatch.setattr(security, "_fernet", Fernet(Fernet.generate_key()))
+    listed = owner.get(f"/api/downloads/{job_id}/links").json()
+    assert [entry["url"] for entry in listed] == [None]
+    assert client.get(f"/api/public/watch/{token}").status_code == 200
+
+
+def test_making_changing_and_revoking_a_link_is_audited_without_its_secrets(db, key, owner_job):
+    owner, job_id = owner_job
+    pw = "a secret word"
+    made = owner.post(
+        f"/api/downloads/{job_id}/links", json={"password": pw, "allow_download": False}
+    ).json()
+    link_id = made["id"]
+    patch = f"/api/downloads/links/{link_id}"
+    assert owner.patch(patch, json={"allow_download": True}).status_code == 200
+    assert owner.patch(patch, json={"password": "another one"}).status_code == 200
+    # Nothing changed: nothing to record.
+    assert owner.patch(patch, json={"allow_download": True}).status_code == 200
+    gone = owner.delete(f"/api/downloads/links/{link_id}")
+    assert gone.json() == {"ok": True, "deleted": link_id}
+
+    (create,) = _audits(db, "download_link_create")
+    assert create.target_id == str(link_id)
+    assert create.after == {
+        "job_id": job_id,
+        "allow_download": False,
+        "has_password": True,
+        "expires_at": None,
+    }
+    updates = _audits(db, "download_link_update")
+    assert [(u.before["allow_download"], u.after["allow_download"]) for u in updates] == [
+        (False, True),
+        (True, True),  # the password changed: a rotation is recorded though the flags are equal
+    ]
+    (revoke,) = _audits(db, "download_link_revoke")
+    assert revoke.target_id == str(link_id)
+    assert revoke.after == {"job_id": job_id}
+
+    trail = json.dumps(
+        [[a.summary, a.before, a.after] for a in [create, *updates, revoke]], default=str
+    )
+    for secret in (made["token"], pw, "another one"):
+        assert secret not in trail
+
+
+def test_the_migration_keeps_every_address_already_sent(db, key, owner_job, client):
+    owner, job_id = owner_job
+    db.commit()  # an open transaction's locks would hold the downgrade's ALTER TABLE forever
+    cfg = Config("alembic.ini")
+    command.downgrade(cfg, "0106_api_tokens")
+    try:
+        with engine.begin() as conn:
+            conn.execute(
+                text("INSERT INTO download_links (job_id, token) VALUES (:j, 'an-old-token-sent')"),
+                {"j": job_id},
+            )
+    finally:
+        command.upgrade(cfg, "head")
+
+    (row,) = db.execute(text("SELECT token_hash, token_enc FROM download_links")).mappings().all()
+    assert row["token_hash"] == security.hash_token("an-old-token-sent")
+    assert security.decrypt(row["token_enc"]) == "an-old-token-sent"
+    assert client.get("/api/public/watch/an-old-token-sent").status_code == 200
+    listed = owner.get(f"/api/downloads/{job_id}/links").json()
+    assert [entry["url"] for entry in listed] == ["/watch/an-old-token-sent"]
+
+
+def test_the_downgrade_writes_the_tokens_back(db, key, owner_job):
+    owner, job_id = owner_job
+    token = owner.post(f"/api/downloads/{job_id}/links", json={}).json()["token"]
+    db.commit()  # an open transaction's locks would hold the downgrade's ALTER TABLE forever
+    cfg = Config("alembic.ini")
+    command.downgrade(cfg, "0106_api_tokens")
+    try:
+        with engine.connect() as conn:
+            back = conn.execute(text("SELECT token FROM download_links")).scalars().all()
+    finally:
+        command.upgrade(cfg, "head")
+    assert back == [token]
+    assert db.execute(select(DownloadLink.token_hash)).scalars().all() == [
+        security.hash_token(token)
+    ]