siftlode/backend/tests/test_link_token_at_rest.py

"""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)
    ]