siftlode/backend/tests/test_download_credentials.py

"""S6 — per-user, per-host download cookies: host-scope, cache privacy, encryption, write-only API.

The two load-bearing security properties are covered as behaviour, not prose:
  * **cache privacy** — a cookie-authed download is a PRIVATE, salted asset; a second user without a
    cookie enqueuing the same URL gets the shared (unsalted) asset and can never be cache-served the
    authed one, and the worker attaches a cookie ONLY to that private identity.
  * **host scope (SEC-12)** — a cookie is resolved ONLY for a URL whose netloc the credential owns; a
    look-alike or unrelated host resolves to no host_key and gets no cookie.

Pure-logic parts run with no DB; the CRUD / enqueue-salting / API parts use the `db`/`as_user` lane.
"""

import os
from pathlib import Path

import pytest
from cryptography.fernet import Fernet

from app import security
from app.downloads import credentials, formats, service
from app.models import DownloadCredential, DownloadJob, MediaAsset, User


@pytest.fixture(autouse=True)
def _fernet(monkeypatch):
    # A throwaway key so encrypt/decrypt work without TOKEN_ENCRYPTION_KEY in the env (mirrors
    # test_security). Autouse: every credential write in this module encrypts.
    monkeypatch.setattr(security, "_fernet", Fernet(Fernet.generate_key()))


COOKIES = "# Netscape HTTP Cookie File\n.facebook.com\tTRUE\t/\tTRUE\t0\tc_user\t123\n"


# --- host scope (pure logic) ------------------------------------------------------------------


class TestHostKeyForUrl:
    def test_matches_a_supported_host(self):
        assert credentials.host_key_for_url("https://www.facebook.com/watch?v=1") == "facebook"
        assert credentials.host_key_for_url("https://fb.watch/abc/") == "facebook"
        assert credentials.host_key_for_url("https://x.com/i/status/1") == "twitter"

    def test_look_alike_host_never_matches(self):
        assert credentials.host_key_for_url("https://myfacebook.com/x") is None
        assert credentials.host_key_for_url("https://facebook.com.evil.test/x") is None

    def test_unrelated_or_empty_url_has_no_host_key(self):
        assert credentials.host_key_for_url("https://www.youtube.com/watch?v=abc") is None
        assert credentials.host_key_for_url("") is None
        assert credentials.host_key_for_url(None) is None


class TestValidateCookies:
    def test_accepts_a_netscape_file(self):
        assert credentials.validate_cookies(COOKIES).endswith("\n")

    def test_rejects_empty(self):
        with pytest.raises(credentials.InvalidCookies):
            credentials.validate_cookies("   ")

    def test_rejects_non_cookie_paste(self):
        # Comments only, no tab-separated data line -> not a usable cookies.txt.
        with pytest.raises(credentials.InvalidCookies):
            credentials.validate_cookies("just some pasted text\n# a comment")


def test_authed_sig_differs_per_user_and_from_the_shared_sig():
    spec = formats.normalize({})
    shared = formats.format_sig(spec)
    a = credentials.authed_format_sig(spec, 1, "facebook")
    b = credentials.authed_format_sig(spec, 2, "facebook")
    assert a != shared and b != shared and a != b


def test_write_cookiefile_is_private_and_survives_the_retry_wipe(tmp_path):
    staging = tmp_path / "staging"
    staging.mkdir()
    path = credentials._write_cookiefile(COOKIES, staging)
    assert path.read_text() == COOKIES
    # 0600 (owner-only) — POSIX perms only (the worker is Linux-only; Windows has no mode bits).
    if os.name == "posix":
        assert (path.stat().st_mode & 0o777) == 0o600
    # It lives in a SUBDIR so the download retry loop (which unlinks files in the staging ROOT) and
    # _produced_file (which scans the root) never see it.
    assert path.parent == staging / ".auth"
    assert not any(p.is_file() for p in staging.iterdir())


# --- CRUD + encryption at rest (DB) -----------------------------------------------------------


def _user(db, email="cred@example.com") -> User:
    u = User(email=email)
    db.add(u)
    db.flush()
    return u


class TestCrud:
    def test_upsert_stores_ciphertext_not_plaintext(self, db):
        u = _user(db)
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()
        row = db.get(DownloadCredential, 1)
        assert row.cookies_enc != COOKIES  # encrypted at rest
        assert COOKIES not in row.cookies_enc
        assert security.decrypt(row.cookies_enc) == COOKIES

    def test_upsert_replaces_in_place(self, db):
        u = _user(db)
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()
        credentials.upsert_credential(db, u.id, "facebook", COOKIES.replace("123", "999"))
        db.commit()
        rows = db.query(DownloadCredential).filter_by(user_id=u.id, host_key="facebook").all()
        assert len(rows) == 1
        assert "999" in security.decrypt(rows[0].cookies_enc)

    def test_status_list_never_leaks_the_blob(self, db):
        u = _user(db)
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()
        status = credentials.status_list(db, u.id)
        fb = next(h for h in status if h["host_key"] == "facebook")
        assert fb["configured"] is True and fb["updated_at"] is not None
        ig = next(h for h in status if h["host_key"] == "instagram")
        assert ig["configured"] is False and ig["updated_at"] is None
        # No field anywhere carries the cookie value.
        assert all(COOKIES not in str(v) for h in status for v in h.values())

    def test_delete(self, db):
        u = _user(db)
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()
        assert credentials.delete_credential(db, u.id, "facebook") is True
        db.commit()
        assert credentials.get_credential(db, u.id, "facebook") is None
        assert credentials.delete_credential(db, u.id, "facebook") is False


# --- the crux: enqueue salts privately, no cross-user cache serve (DB) -------------------------

FB_URL = "https://www.facebook.com/watch?v=100"


class TestCachePrivacy:
    def test_enqueue_with_a_cookie_creates_a_private_salted_asset(self, db):
        u = _user(db)
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()
        job = service.enqueue(db, u.id, "url", FB_URL, {})
        asset = db.get(MediaAsset, job.asset_id)
        spec = formats.normalize({})
        assert asset.format_sig == credentials.authed_format_sig(spec, u.id, "facebook")
        assert asset.format_sig != formats.format_sig(spec)

    def test_enqueue_without_a_cookie_uses_the_shared_sig(self, db):
        u = _user(db)
        job = service.enqueue(db, u.id, "url", FB_URL, {})
        asset = db.get(MediaAsset, job.asset_id)
        assert asset.format_sig == formats.format_sig(formats.normalize({}))

    def test_two_users_never_share_the_authed_asset(self, db):
        a = _user(db, "a@example.com")
        b = _user(db, "b@example.com")
        credentials.upsert_credential(db, a.id, "facebook", COOKIES)
        db.commit()
        ja = service.enqueue(db, a.id, "url", FB_URL, {})  # A: authed, private
        jb = service.enqueue(db, b.id, "url", FB_URL, {})  # B: no cookie, shared
        assert ja.asset_id != jb.asset_id
        # B's asset is the shared identity — A's private group video can never be served to B.
        assert db.get(MediaAsset, jb.asset_id).format_sig == formats.format_sig(
            formats.normalize({})
        )


# --- the worker gate: attach only to the private identity, host-scoped (DB) --------------------


class TestResolveCookiefile:
    def _asset_sig(self, sig: str) -> MediaAsset:
        return MediaAsset(
            source_kind="url", source_ref=FB_URL, format_sig=sig, status="downloading"
        )

    def test_attaches_to_the_private_salted_asset(self, db, tmp_path):
        u = _user(db)
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()
        spec = formats.normalize({})
        salted = credentials.authed_format_sig(spec, u.id, "facebook")
        path = credentials.resolve_cookiefile(
            db, u.id, FB_URL, self._asset_sig(salted), spec, tmp_path
        )
        assert path is not None and Path(path).read_text() == COOKIES

    def test_never_attaches_to_a_shared_asset(self, db, tmp_path):
        # Cookie exists, but the asset is the UNSALTED shared identity -> attaching would poison the
        # global cache with a private-group video. Must return None.
        u = _user(db)
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()
        spec = formats.normalize({})
        shared = formats.format_sig(spec)
        assert (
            credentials.resolve_cookiefile(
                db, u.id, FB_URL, self._asset_sig(shared), spec, tmp_path
            )
            is None
        )

    def test_non_matching_host_gets_no_cookie(self, db, tmp_path):
        u = _user(db)
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()
        spec = formats.normalize({})
        # A YouTube URL has no supported host_key -> no cookie regardless of what's stored.
        yt = "https://www.youtube.com/watch?v=abc"
        salted = credentials.authed_format_sig(spec, u.id, "facebook")
        assert (
            credentials.resolve_cookiefile(db, u.id, yt, self._asset_sig(salted), spec, tmp_path)
            is None
        )

    def test_no_credential_gets_no_cookie(self, db, tmp_path):
        u = _user(db)
        spec = formats.normalize({})
        salted = credentials.authed_format_sig(spec, u.id, "facebook")
        assert (
            credentials.resolve_cookiefile(
                db, u.id, FB_URL, self._asset_sig(salted), spec, tmp_path
            )
            is None
        )


# --- Retry / Re-download take a cookie added later (DB) -----------------------------------------
# Measured on dev (E11.5): a Facebook reel failed on the login wall, the user added their cookie as
# the failure copy tells them to, pressed Retry — and it failed again, because the job still held the
# shared asset the worker never attaches a cookie to. Re-download forked a salt the gate could not
# recognise, so even a job downloaded WITH a cookie rebuilt without it.


def _held_job(db, u, *, sig, asset_status="error", job_status="error"):
    a = MediaAsset(
        source_kind="url", source_ref=FB_URL, format_sig=sig, status=asset_status, ref_count=1
    )
    db.add(a)
    db.flush()
    j = DownloadJob(
        user_id=u.id,
        asset_id=a.id,
        source_kind="url",
        source_ref=FB_URL,
        status=job_status,
        queue_pos=0,
        profile_snapshot={},
    )
    db.add(j)
    db.commit()
    return a, j


class TestRequeueTakesTheCookie:
    spec = formats.normalize({})

    @pytest.mark.parametrize(
        ("asset_status", "job_status"),
        # A failed download; and a job paused before it ever started (its asset never downloaded).
        [("error", "error"), ("pending", "paused")],
    )
    def test_retry_after_adding_a_cookie_moves_the_job_to_the_private_asset(
        self, db, tmp_path, asset_status, job_status
    ):
        u = _user(db)
        shared, j = _held_job(
            db,
            u,
            sig=formats.format_sig(self.spec),
            asset_status=asset_status,
            job_status=job_status,
        )
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()

        service.resume(db, j)

        db.expire_all()
        job = db.get(DownloadJob, j.id)
        target = db.get(MediaAsset, job.asset_id)
        assert job.status == "queued" and target.id != shared.id
        assert target.format_sig == credentials.authed_format_sig(self.spec, u.id, "facebook")
        assert (target.status, target.ref_count) == ("pending", 1)
        assert db.get(MediaAsset, shared.id).ref_count == 0  # the old hold went with the move
        assert credentials.resolve_cookiefile(db, u.id, FB_URL, target, self.spec, tmp_path)

    def test_retry_without_a_cookie_keeps_the_shared_asset(self, db):
        u = _user(db)
        shared, j = _held_job(db, u, sig=formats.format_sig(self.spec))

        service.resume(db, j)

        db.expire_all()
        assert db.get(DownloadJob, j.id).asset_id == shared.id
        assert db.get(MediaAsset, shared.id).ref_count == 1

    @pytest.mark.parametrize(
        ("asset_status", "job_status"), [("ready", "error"), ("downloading", "paused")]
    )
    def test_retry_never_moves_off_a_file_or_a_download_in_flight(
        self, db, asset_status, job_status
    ):
        u = _user(db)
        shared, j = _held_job(
            db,
            u,
            sig=formats.format_sig(self.spec),
            asset_status=asset_status,
            job_status=job_status,
        )
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()

        service.resume(db, j)

        db.expire_all()
        assert db.get(DownloadJob, j.id).asset_id == shared.id

    def test_redownload_with_a_cookie_rebuilds_into_an_asset_the_cookie_goes_with(
        self, db, tmp_path
    ):
        u = _user(db)
        other = _user(db, "other@example.com")
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        credentials.upsert_credential(db, other.id, "facebook", COOKIES)
        authed = credentials.authed_format_sig(self.spec, u.id, "facebook")
        first, j = _held_job(db, u, sig=authed, asset_status="ready", job_status="done")

        seen = {first.id}
        for _ in range(2):  # a rebuild of a rebuild is still recognised
            service.redownload(db, j)
            db.expire_all()
            job = db.get(DownloadJob, j.id)
            target = db.get(MediaAsset, job.asset_id)
            assert target.id not in seen and target.format_sig != authed
            seen.add(target.id)
            assert credentials.resolve_cookiefile(db, u.id, FB_URL, target, self.spec, tmp_path)
            # Private: another user's cookie is never attached to it.
            assert not credentials.is_authed_asset(target, self.spec, other.id, FB_URL)

        # A Retry of a failed rebuild keeps the rebuild — it is already the cookie's identity.
        target.status, job.status = "error", "error"
        db.commit()
        service.resume(db, job)
        db.expire_all()
        assert db.get(DownloadJob, j.id).asset_id == target.id

    def test_a_rebuild_made_without_a_cookie_is_no_cookie_identity(self, db, tmp_path):
        u = _user(db)
        _, j = _held_job(db, u, sig=formats.format_sig(self.spec))
        service.redownload(db, j)
        credentials.upsert_credential(db, u.id, "facebook", COOKIES)
        db.commit()

        db.expire_all()
        target = db.get(MediaAsset, db.get(DownloadJob, j.id).asset_id)
        assert not credentials.is_authed_asset(target, self.spec, u.id, FB_URL)


# --- API: write-only, never echoes the blob (DB) ----------------------------------------------


class TestApi:
    def test_put_get_delete_roundtrip_never_returns_the_blob(self, db, as_user):
        u = _user(db)
        db.commit()
        client = as_user(u)

        r = client.put("/api/downloads/credentials/facebook", json={"cookies": COOKIES})
        assert r.status_code == 200
        body = r.json()
        assert body["ok"] is True
        assert COOKIES not in r.text  # the response must never carry the value
        fb = next(h for h in body["hosts"] if h["host_key"] == "facebook")
        assert fb["configured"] is True

        r = client.get("/api/downloads/credentials")
        assert r.status_code == 200 and COOKIES not in r.text

        r = client.delete("/api/downloads/credentials/facebook")
        assert r.status_code == 200
        assert credentials.get_credential(db, u.id, "facebook") is None

    def test_unknown_host_is_404(self, db, as_user):
        u = _user(db)
        db.commit()
        r = as_user(u).put("/api/downloads/credentials/nope", json={"cookies": COOKIES})
        assert r.status_code == 404

    def test_bad_cookies_are_rejected(self, db, as_user):
        u = _user(db)
        db.commit()
        r = as_user(u).put("/api/downloads/credentials/facebook", json={"cookies": "   "})
        assert r.status_code == 400
        assert r.json()["detail"]["code"] == "cookies"