feat(fc3b): CredentialService — CRUD + on-demand cookies.txt materialisation
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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"
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user