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, 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, 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, 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, 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, salted, spec, tmp_path) is None


# --- 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"