FabledCurator/backend/app/services/credential_service.py

"""FC-3b: CRUD over Credential rows + on-demand cookies-file
materialisation for gallery-dl (consumed by FC-3c).
"""

from __future__ import annotations

import json
import os
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import Credential
from .credential_crypto import CredentialCrypto
from .platforms import PLATFORMS, auth_type_for


class CredentialServiceError(Exception):
    """Base."""


class UnknownPlatformError(CredentialServiceError):
    pass


class WrongAuthTypeError(CredentialServiceError):
    def __init__(self, expected: str):
        super().__init__(f"wrong credential_type for platform; expected {expected!r}")
        self.expected = expected


class EmptyDataError(CredentialServiceError):
    pass


@dataclass(frozen=True)
class CredentialRecord:
    platform: str
    credential_type: str
    captured_at: str
    expires_at: str | None
    last_verified: str | None

    def to_dict(self) -> dict:
        return {
            "platform": self.platform,
            "credential_type": self.credential_type,
            "captured_at": self.captured_at,
            "expires_at": self.expires_at,
            "last_verified": self.last_verified,
        }


def _to_record(row: Credential) -> CredentialRecord:
    return CredentialRecord(
        platform=row.platform,
        credential_type=row.credential_type,
        captured_at=row.captured_at.isoformat() if row.captured_at else "",
        expires_at=row.expires_at.isoformat() if row.expires_at else None,
        last_verified=row.last_verified.isoformat() if row.last_verified else None,
    )


# Default cookies dir under the image root; tests override.
_DEFAULT_COOKIES_DIR = Path("/images/cookies")


class CredentialService:
    def __init__(
        self,
        session: AsyncSession,
        crypto: CredentialCrypto,
        cookies_dir: Path | None = None,
    ):
        self.session = session
        self.crypto = crypto
        self.cookies_dir = Path(cookies_dir) if cookies_dir else _DEFAULT_COOKIES_DIR

    async def list(self) -> list[CredentialRecord]:
        rows = (await self.session.execute(
            select(Credential).order_by(Credential.platform.asc())
        )).scalars().all()
        return [_to_record(r) for r in rows]

    async def get(self, platform: str) -> CredentialRecord | None:
        row = (await self.session.execute(
            select(Credential).where(Credential.platform == platform)
        )).scalar_one_or_none()
        return _to_record(row) if row else None

    async def upsert(
        self,
        *,
        platform: str,
        credential_type: str,
        data: str,
        expires_at: datetime | None = None,
    ) -> CredentialRecord:
        if platform not in PLATFORMS:
            raise UnknownPlatformError(f"unknown platform: {platform!r}")
        expected = auth_type_for(platform)
        if credential_type != expected:
            raise WrongAuthTypeError(expected=expected or "")
        cleaned = (data or "").strip()
        if not cleaned:
            raise EmptyDataError("data must be a non-empty string")
        blob = self.crypto.encrypt(cleaned)

        row = (await self.session.execute(
            select(Credential).where(Credential.platform == platform)
        )).scalar_one_or_none()
        if row is None:
            row = Credential(
                platform=platform,
                credential_type=credential_type,
                encrypted_blob=blob,
                expires_at=expires_at,
            )
            self.session.add(row)
        else:
            row.credential_type = credential_type
            row.encrypted_blob = blob
            row.expires_at = expires_at
            row.last_verified = None  # invalidate prior verification
        await self.session.commit()
        await self.session.refresh(row)
        return _to_record(row)

    async def delete(self, platform: str) -> None:
        row = (await self.session.execute(
            select(Credential).where(Credential.platform == platform)
        )).scalar_one_or_none()
        if row is None:
            raise LookupError(f"no credential for platform {platform!r}")
        await self.session.delete(row)
        await self.session.commit()

    async def get_cookies_path(self, platform: str) -> Path | None:
        """Decrypt the credential and write a Netscape cookies.txt for
        gallery-dl. Returns None if no credential or wrong kind."""
        row = (await self.session.execute(
            select(Credential).where(Credential.platform == platform)
        )).scalar_one_or_none()
        if row is None or row.credential_type != "cookies":
            return None
        plaintext = self.crypto.decrypt(row.encrypted_blob)
        netscape = _to_netscape(plaintext)
        netscape = _augment_cookies(platform, netscape)
        self.cookies_dir.mkdir(parents=True, exist_ok=True)
        out = self.cookies_dir / f"{platform}_cookies.txt"
        out.write_text(netscape)
        os.chmod(out, 0o600)
        return out

    async def get_token(self, platform: str) -> str | None:
        row = (await self.session.execute(
            select(Credential).where(Credential.platform == platform)
        )).scalar_one_or_none()
        if row is None or row.credential_type != "token":
            return None
        return self.crypto.decrypt(row.encrypted_blob)


def _augment_cookies(platform: str, netscape: str) -> str:
    """Inject platform-specific synthetic cookies needed to bypass server
    gates or extractor quirks.

    subscribestar.adult: the server gates artist pages behind the
    `_personalization_id` age-confirmation cookie. The site's frontend JS
    uses localStorage to suppress the age popup once dismissed, so after
    the cookie's annual expiry the user can't easily get a fresh one —
    visiting the site in a logged-in session doesn't re-show the popup
    and doesn't re-issue the cookie. gallery-dl's own login flow (which
    FC doesn't use; we capture cookies via the extension instead)
    sidesteps this by manually setting `18_plus_agreement_generic=true`
    on `.subscribestar.adult` — the server accepts that as the
    age-confirmation marker.

    hentaifoundry: gallery-dl's extractor uses
    `self.cookies.get("PHPSESSID", domain="www.hentai-foundry.com")` to
    decide whether the user is logged in. `requests` does EXACT domain
    matching on .get(); the extension rewrites every captured cookie to
    a leading-dot subdomain-wide form (`.hentai-foundry.com`), which
    fails that exact match. The fallback path hits a HEAD
    `?enterAgree=1` that 401s. Emit host-only duplicates of PHPSESSID +
    YII_CSRF_TOKEN on `www.hentai-foundry.com` so the lookup succeeds.
    (The original `.hentai-foundry.com` entries stay — the actual HTTP
    requests use RFC 6265 subdomain matching, which works either way.)

    All injections are idempotent (no-op if the target cookie is already
    present) and platform-scoped.

    Operator-flagged 2026-05-27: subscribestar age-confirmation, then
    hentaifoundry 401 on /?enterAgree=1.
    """
    if platform == "subscribestar":
        return _augment_subscribestar(netscape)
    if platform == "hentaifoundry":
        return _augment_hentaifoundry(netscape)
    return netscape


def _augment_subscribestar(netscape: str) -> str:
    if "18_plus_agreement_generic" in netscape:
        return netscape
    # Far-future expiry — gallery-dl's own login flow sets this with no
    # explicit expiry; the server only checks presence/value.
    expiry = 4102444800  # 2100-01-01 UTC, opaque "far future"
    line = "\t".join([
        ".subscribestar.adult", "TRUE", "/", "TRUE",
        str(expiry), "18_plus_agreement_generic", "true",
    ])
    body = netscape.rstrip("\n")
    if not body:
        body = "# Netscape HTTP Cookie File"
    return body + "\n" + line + "\n"


_HF_HOST_ONLY_NAMES = ("PHPSESSID", "YII_CSRF_TOKEN")


def _augment_hentaifoundry(netscape: str) -> str:
    body = netscape.rstrip("\n")
    if not body:
        return netscape
    lines = body.split("\n")
    existing_host_only = set()
    by_name: dict[str, list[str]] = {}
    for raw in lines:
        if not raw or raw.startswith("#"):
            continue
        parts = raw.split("\t")
        if len(parts) < 7:
            continue
        domain, _flag, _path, _secure, _exp, name, _value = parts[:7]
        if name not in _HF_HOST_ONLY_NAMES:
            continue
        if domain == "www.hentai-foundry.com":
            existing_host_only.add(name)
        elif domain in (".hentai-foundry.com", "hentai-foundry.com"):
            by_name.setdefault(name, []).append(raw)

    appended = []
    for name in _HF_HOST_ONLY_NAMES:
        if name in existing_host_only or name not in by_name:
            continue
        # Duplicate the FIRST subdomain-wide line as host-only on
        # www.hentai-foundry.com. Same value + expiry; flag=FALSE marks
        # it host-only in netscape format.
        parts = by_name[name][0].split("\t")
        parts[0] = "www.hentai-foundry.com"
        parts[1] = "FALSE"
        appended.append("\t".join(parts[:7]))

    if not appended:
        return netscape
    return body + "\n" + "\n".join(appended) + "\n"


def _to_netscape(plaintext: str) -> str:
    """Accept either Netscape-format text (the extension's output) or a
    JSON array of cookie dicts (a manual-paste edge case); produce
    Netscape-format text suitable for gallery-dl --cookies."""
    stripped = plaintext.strip()
    if not stripped:
        return ""
    if stripped.startswith("#") or "\t" in stripped:
        return plaintext  # already Netscape
    try:
        loaded = json.loads(stripped)
    except json.JSONDecodeError:
        return plaintext  # write as-is and hope; gallery-dl will complain if invalid
    if isinstance(loaded, dict):
        loaded = [loaded]
    if not isinstance(loaded, list):
        return plaintext
    lines = ["# Netscape HTTP Cookie File"]
    for c in loaded:
        domain = str(c.get("domain", ""))
        if domain and not domain.startswith("."):
            domain = "." + domain
        flag = "TRUE"
        path = str(c.get("path", "/"))
        secure = "TRUE" if c.get("secure", False) else "FALSE"
        expiration = c.get("expiration") or c.get("expirationDate") or 0
        try:
            expiration_str = str(int(float(expiration)))
        except (TypeError, ValueError):
            expiration_str = "0"
        name = str(c.get("name", ""))
        value = str(c.get("value", ""))
        lines.append("\t".join([domain, flag, path, secure, expiration_str, name, value]))
    return "\n".join(lines) + "\n"