diff --git a/backend/app/services/credential_service.py b/backend/app/services/credential_service.py new file mode 100644 index 0000000..a2cd324 --- /dev/null +++ b/backend/app/services/credential_service.py @@ -0,0 +1,199 @@ +"""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) + 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 _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" diff --git a/tests/test_credential_service.py b/tests/test_credential_service.py new file mode 100644 index 0000000..4d85f41 --- /dev/null +++ b/tests/test_credential_service.py @@ -0,0 +1,124 @@ +import pytest +from sqlalchemy import select + +from backend.app.models import Credential +from backend.app.services.credential_crypto import CredentialCrypto +from backend.app.services.credential_service import ( + CredentialService, + EmptyDataError, + UnknownPlatformError, + WrongAuthTypeError, +) + +pytestmark = pytest.mark.integration + + +_NETSCAPE = ( + "# Netscape HTTP Cookie File\n" + ".patreon.com\tTRUE\t/\tTRUE\t1700000000\tsession_id\tabc\n" +) + + +@pytest.fixture +def crypto(tmp_path): + return CredentialCrypto(tmp_path / "k") + + +@pytest.mark.asyncio +async def test_upsert_creates_then_updates(db, crypto): + svc = CredentialService(db, crypto) + rec = await svc.upsert( + platform="patreon", credential_type="cookies", data=_NETSCAPE, + ) + assert rec.platform == "patreon" + assert rec.credential_type == "cookies" + again = await svc.upsert( + platform="patreon", credential_type="cookies", data=_NETSCAPE + "x", + ) + assert again.platform == "patreon" + rows = (await db.execute( + select(Credential).where(Credential.platform == "patreon") + )).scalars().all() + assert len(rows) == 1 + + +@pytest.mark.asyncio +async def test_upsert_rejects_unknown_platform(db, crypto): + svc = CredentialService(db, crypto) + with pytest.raises(UnknownPlatformError): + await svc.upsert(platform="fanbox", credential_type="cookies", data="x") + + +@pytest.mark.asyncio +async def test_upsert_rejects_wrong_auth_type(db, crypto): + svc = CredentialService(db, crypto) + # discord is token-only + with pytest.raises(WrongAuthTypeError): + await svc.upsert(platform="discord", credential_type="cookies", data="x") + # patreon is cookies-only + with pytest.raises(WrongAuthTypeError): + await svc.upsert(platform="patreon", credential_type="token", data="x") + + +@pytest.mark.asyncio +async def test_upsert_rejects_empty_data(db, crypto): + svc = CredentialService(db, crypto) + with pytest.raises(EmptyDataError): + await svc.upsert(platform="patreon", credential_type="cookies", data=" ") + + +@pytest.mark.asyncio +async def test_list_excludes_blob(db, crypto): + svc = CredentialService(db, crypto) + await svc.upsert(platform="patreon", credential_type="cookies", data=_NETSCAPE) + records = await svc.list() + assert len(records) == 1 + d = records[0].to_dict() + assert "data" not in d + assert "encrypted_blob" not in d + + +@pytest.mark.asyncio +async def test_delete(db, crypto): + svc = CredentialService(db, crypto) + await svc.upsert(platform="patreon", credential_type="cookies", data=_NETSCAPE) + await svc.delete("patreon") + assert await svc.get("patreon") is None + + +@pytest.mark.asyncio +async def test_get_cookies_path_writes_usable_file(db, crypto, tmp_path): + svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies") + await svc.upsert(platform="patreon", credential_type="cookies", data=_NETSCAPE) + path = await svc.get_cookies_path("patreon") + assert path is not None and path.exists() + contents = path.read_text() + assert ".patreon.com" in contents + assert "session_id" in contents + + +@pytest.mark.asyncio +async def test_get_cookies_path_none_for_missing(db, crypto, tmp_path): + svc = CredentialService(db, crypto, cookies_dir=tmp_path) + assert await svc.get_cookies_path("patreon") is None + + +@pytest.mark.asyncio +async def test_get_cookies_path_none_for_token_kind(db, crypto, tmp_path): + svc = CredentialService(db, crypto, cookies_dir=tmp_path) + await svc.upsert(platform="discord", credential_type="token", data="tok") + assert await svc.get_cookies_path("discord") is None + + +@pytest.mark.asyncio +async def test_get_token_decrypts(db, crypto): + svc = CredentialService(db, crypto) + await svc.upsert(platform="discord", credential_type="token", data="my-token") + assert await svc.get_token("discord") == "my-token" + + +@pytest.mark.asyncio +async def test_get_token_none_for_cookies_kind(db, crypto): + svc = CredentialService(db, crypto) + await svc.upsert(platform="patreon", credential_type="cookies", data=_NETSCAPE) + assert await svc.get_token("patreon") is None