siftlode/backend/app/downloads/credentials.py

"""Per-user, per-host download cookies — the single source of truth for "does the acting user's
cookie apply to this download, and how does that keep the result private".

Two call sites depend on ONE consistent answer (house rule: an invariant lives in shared code, not
in prose):
  * `enqueue` salts the MediaAsset cache signature with the user id WHEN a cookie applies, so the
    authed download becomes a PRIVATE asset that is never cache-served to another user.
  * the worker attaches the cookie file to yt-dlp ONLY when a cookie still applies AND the asset it
    is downloading into is that private, salted identity — so a cookie can never poison a shared
    asset (the load-bearing cross-user privacy fix).

Both go through `cookie_salt` / `authed_format_sig` / `host_key_for_url` here, so they can't drift.

Security controls enforced here:
  * **host scope (SEC-12)** — `host_key_for_url` matches the download URL's netloc against a curated
    registry via the netloc-scoped `_host_is`, so a cookie is only ever offered to the exact site it
    belongs to; an arbitrary `?url=` (or a look-alike host) resolves to no host_key and gets no
    cookie.
  * **secret handling** — cookies are Fernet-encrypted at rest; the blob is only ever decrypted
    in-worker into a 0600 file that is unlinked in a `finally`, and is never returned by a route,
    logged, or written to a job's error/snapshot.
"""

import logging
import os
from pathlib import Path

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.downloads import formats
from app.downloads.adapters import _host_is
from app.models import DownloadCredential
from app.security import decrypt_optional, encrypt

log = logging.getLogger("siftlode.downloads.credentials")


class HostSpec:
    """A supported cookie host: a short canonical `key` (stored on the credential + folded into the
    cache salt), a human `label` for the UI, and the netloc `domains` it matches."""

    __slots__ = ("key", "label", "domains")

    def __init__(self, key: str, label: str, domains: tuple[str, ...]):
        self.key = key
        self.label = label
        self.domains = domains


# Curated registry (user's call: seed the common auth-walled hosts now). One `key` can cover several
# domains of the same site. Add a host by adding a row — the FE dropdown, host-scope guard, and salt
# all derive from this, so there is nothing else to wire up.
SUPPORTED_HOSTS: tuple[HostSpec, ...] = (
    HostSpec("facebook", "Facebook", ("facebook.com", "fb.watch")),
    HostSpec("instagram", "Instagram", ("instagram.com",)),
    HostSpec("twitter", "X (Twitter)", ("twitter.com", "x.com")),
    HostSpec("tiktok", "TikTok", ("tiktok.com",)),
    HostSpec("reddit", "Reddit", ("reddit.com",)),
)

_BY_KEY: dict[str, HostSpec] = {h.key: h for h in SUPPORTED_HOSTS}


def is_supported_host(host_key: str) -> bool:
    return host_key in _BY_KEY


def host_label(host_key: str) -> str:
    spec = _BY_KEY.get(host_key)
    return spec.label if spec else host_key


def host_key_for_url(url: str | None) -> str | None:
    """The supported host_key whose domains own `url`'s netloc, or None. Netloc-scoped (via
    `_host_is`), so a look-alike host ('myfacebook.com', 'facebook.com.evil') never matches. When
    several specs could match, the most-specific (longest matched domain) wins so the result is
    deterministic across the enqueue and worker call sites."""
    if not url:
        return None
    from urllib.parse import urlparse

    host = urlparse(url).netloc.lower()
    if not host:
        return None
    best_key: str | None = None
    best_len = -1
    for spec in SUPPORTED_HOSTS:
        for dom in spec.domains:
            if _host_is(host, dom) and len(dom) > best_len:
                best_key, best_len = spec.key, len(dom)
    return best_key


def cookie_salt(user_id: int, host_key: str) -> str:
    """The cache-signature salt for a cookie-authed download. Keyed by user id + host so the authed
    asset is private to that user (never dedup-shared) and stable across that user's own re-adds."""
    return f"cookie:{user_id}:{host_key}"


def authed_format_sig(spec: dict, user_id: int, host_key: str) -> str:
    """The salted `format_sig` an authed download for (user, host) resolves to — the private asset
    identity. Both enqueue (to create it) and the worker (to recognise it) compute it HERE."""
    return formats.format_sig(spec, salt=cookie_salt(user_id, host_key))


def salt_for_enqueue(db: Session, user_id: int, url: str | None) -> str | None:
    """At enqueue: the cache salt to apply if this user has a cookie for `url`'s host, else None (a
    normal shared download). Returns the salt string so the caller passes it straight to
    `format_sig(spec, salt=...)`."""
    host_key = host_key_for_url(url)
    if host_key is None:
        return None
    if get_credential(db, user_id, host_key) is None:
        return None
    return cookie_salt(user_id, host_key)


# --- storage (CRUD) ---------------------------------------------------------------------------


def get_credential(db: Session, user_id: int, host_key: str) -> DownloadCredential | None:
    return db.execute(
        select(DownloadCredential).where(
            DownloadCredential.user_id == user_id, DownloadCredential.host_key == host_key
        )
    ).scalar_one_or_none()


class InvalidCookies(ValueError):
    """The pasted cookies aren't a usable Netscape cookies.txt — surfaced as a 400 by the route."""


def validate_cookies(raw: str) -> str:
    """Sanity-check a pasted cookies.txt (Netscape format) and return the cleaned text. yt-dlp reads
    the Netscape format: comment/`#` lines plus TAB-separated cookie lines (domain, flag, path,
    secure, expiry, name, value). We don't fully parse — just reject obvious non-cookie paste (empty,
    or no tab-separated data line) so a mistaken paste fails HERE with a clear message rather than
    silently producing a cookie file yt-dlp ignores."""
    text = (raw or "").strip()
    if not text:
        raise InvalidCookies("empty")
    has_data_line = any(
        line.strip() and not line.lstrip().startswith("#") and "\t" in line
        for line in text.splitlines()
    )
    if not has_data_line:
        raise InvalidCookies("format")
    # Netscape cookie files must end with a newline for some parsers; normalise.
    return text if text.endswith("\n") else text + "\n"


def upsert_credential(db: Session, user_id: int, host_key: str, raw: str) -> DownloadCredential:
    """Create or replace this user's cookie blob for `host_key`. Caller has validated host_key is
    supported and `raw` via `validate_cookies`. Does NOT commit (the route commits with the audit
    row so the trail can't drift). The blob is encrypted here; the plaintext never leaves this call."""
    cred = get_credential(db, user_id, host_key)
    enc = encrypt(raw)
    if cred is None:
        cred = DownloadCredential(user_id=user_id, host_key=host_key, cookies_enc=enc)
        db.add(cred)
    else:
        cred.cookies_enc = enc
    return cred


def delete_credential(db: Session, user_id: int, host_key: str) -> bool:
    """Delete this user's cookie for `host_key`. Returns whether a row existed. Does NOT commit."""
    cred = get_credential(db, user_id, host_key)
    if cred is None:
        return False
    db.delete(cred)
    return True


def status_list(db: Session, user_id: int) -> list[dict]:
    """The supported hosts with this user's configured/updated state — the write-only Settings view.
    Never includes the blob."""
    have = {
        c.host_key: c
        for c in db.execute(
            select(DownloadCredential).where(DownloadCredential.user_id == user_id)
        ).scalars()
    }
    out = []
    for spec in SUPPORTED_HOSTS:
        cred = have.get(spec.key)
        out.append(
            {
                "host_key": spec.key,
                "label": spec.label,
                "configured": cred is not None,
                "updated_at": cred.updated_at.isoformat() if cred is not None else None,
            }
        )
    return out


# --- worker: resolve + write the cookie file --------------------------------------------------


def resolve_cookiefile(
    db: Session,
    user_id: int,
    url: str | None,
    asset_format_sig: str,
    spec: dict,
    staging: Path,
) -> Path | None:
    """In the worker child: the cookies.txt to hand yt-dlp for this download, or None.

    The double gate that prevents cross-user leakage:
      1. a credential must still exist for `url`'s host (host-scoped via `host_key_for_url`), AND
      2. the asset being downloaded into must BE this user's private, salted identity
         (`asset_format_sig == authed_format_sig(...)`).

    If (2) fails the asset is the SHARED (unsalted) one — attaching the cookie would poison the
    global cache with a private-group video, so we attach nothing. An unreadable blob (rotated key →
    InvalidToken) also yields None; the download then fails honestly on the login wall (S3's
    `login_required`)."""
    host_key = host_key_for_url(url)
    if host_key is None:
        return None
    cred = get_credential(db, user_id, host_key)
    if cred is None:
        return None
    if asset_format_sig != authed_format_sig(spec, user_id, host_key):
        return None  # shared asset — never attach a cookie to a non-private identity
    raw = decrypt_optional(cred.cookies_enc)
    if not raw:
        log.warning(
            "download cookie for user %s host %s could not be decrypted (rotated key?) — "
            "proceeding without it",
            user_id,
            host_key,
        )
        return None
    return _write_cookiefile(raw, staging)


def _write_cookiefile(raw: str, staging: Path) -> Path:
    """Write the cookie blob to a private 0600 file under `staging/.auth/` and return its path.

    Deliberately NOT in the staging root: the download retry loop unlinks every file it finds there,
    and `_produced_file` picks the largest file there as the output — either would corrupt a
    cookies.txt in the root. A `.auth` SUBDIR survives both (they skip directories) and is still
    removed by the parent worker's staging cleanup on a hard kill. The caller unlinks it in a
    `finally` on the normal/exception paths."""
    auth_dir = staging / ".auth"
    auth_dir.mkdir(mode=0o700, exist_ok=True)
    path = auth_dir / "cookies.txt"
    # Open with 0600 from the start (don't create world-readable then chmod).
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        os.write(fd, raw.encode("utf-8"))
    finally:
        os.close(fd)
    return path