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