bvandeusen
b54243a1ff
The cookie was pinned to .subscribestar.adult only; cookies are domain-scoped, so sources on subscribestar.art (Elasid, event #54116) never sent it and every poll 302d to /age_confirmation_warning. Emit one line per domain (.com/.adult/.art) with a per-domain presence check, and admit .art in the platform url_pattern. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
245 lines
10 KiB
Python
245 lines
10 KiB
Python
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", bootstrap_ok=True)
|
|
|
|
|
|
@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_cookies_path_subscribestar_injects_age_cookie(db, crypto, tmp_path):
|
|
"""SubscribeStar's server gates artist pages behind a _personalization_id
|
|
cookie; the browser-stored cookie expires annually and can't be easily
|
|
refreshed (the JS age popup is suppressed by localStorage). Mirror
|
|
gallery-dl's own login-flow workaround by injecting
|
|
`18_plus_agreement_generic=true` whenever cookies for subscribestar
|
|
are materialized — on EVERY SubscribeStar domain, since cookies are
|
|
domain-scoped and creators live on .com/.adult/.art alike (a .adult-only
|
|
line left .art sources stuck on the age wall — Elasid, event #54116)."""
|
|
netscape_in = (
|
|
"# Netscape HTTP Cookie File\n"
|
|
".subscribestar.adult\tTRUE\t/\tTRUE\t1700000000\tsession_id\txyz\n"
|
|
)
|
|
svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
|
|
await svc.upsert(platform="subscribestar", credential_type="cookies", data=netscape_in)
|
|
path = await svc.get_cookies_path("subscribestar")
|
|
contents = path.read_text()
|
|
for domain in (".subscribestar.com", ".subscribestar.adult", ".subscribestar.art"):
|
|
assert f"{domain}\tTRUE\t/\tTRUE\t4102444800\t18_plus_agreement_generic\ttrue" in contents
|
|
# Original session_id cookie preserved.
|
|
assert "session_id\txyz" in contents
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_cookies_path_subscribestar_idempotent_per_domain(db, crypto, tmp_path):
|
|
"""If the operator's captured cookies ALREADY contain the age cookie for
|
|
a domain (e.g. a manual paste, or a re-login), don't double-inject on
|
|
THAT domain — but a .adult-only capture must still get the cookie added
|
|
for .com and .art (per-domain check, not presence-anywhere)."""
|
|
netscape_in = (
|
|
"# Netscape HTTP Cookie File\n"
|
|
".subscribestar.adult\tTRUE\t/\tTRUE\t1700000000\t18_plus_agreement_generic\ttrue\n"
|
|
".subscribestar.adult\tTRUE\t/\tTRUE\t1700000000\tsession_id\txyz\n"
|
|
)
|
|
svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
|
|
await svc.upsert(platform="subscribestar", credential_type="cookies", data=netscape_in)
|
|
path = await svc.get_cookies_path("subscribestar")
|
|
contents = path.read_text()
|
|
assert contents.count("18_plus_agreement_generic") == 3
|
|
adult_lines = [
|
|
line for line in contents.splitlines()
|
|
if line.startswith(".subscribestar.adult\t")
|
|
and "18_plus_agreement_generic" in line
|
|
]
|
|
assert len(adult_lines) == 1
|
|
assert ".subscribestar.com\t" in contents
|
|
assert ".subscribestar.art\t" in contents
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_cookies_path_non_subscribestar_unchanged(db, crypto, tmp_path):
|
|
"""The age-cookie injection MUST NOT fire for non-subscribestar
|
|
platforms — Patreon/etc. don't need it and shouldn't carry a
|
|
foreign-domain cookie in their cookies.txt."""
|
|
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")
|
|
contents = path.read_text()
|
|
assert "18_plus_agreement_generic" not in contents
|
|
assert "subscribestar" not in contents
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_cookies_path_hf_injects_host_only_phpsessid(db, crypto, tmp_path):
|
|
"""HF: extension writes session cookies as subdomain-wide
|
|
(`.hentai-foundry.com`), but gallery-dl's extractor uses
|
|
`cookies.get(name, domain='www.hentai-foundry.com')` with EXACT
|
|
domain matching. Emit host-only duplicates of PHPSESSID +
|
|
YII_CSRF_TOKEN on `www.hentai-foundry.com` so the lookup matches."""
|
|
netscape_in = (
|
|
"# Netscape HTTP Cookie File\n"
|
|
".hentai-foundry.com\tTRUE\t/\tTRUE\t1900000000\tPHPSESSID\tsess123\n"
|
|
".hentai-foundry.com\tTRUE\t/\tTRUE\t1900000000\tYII_CSRF_TOKEN\ttoken456\n"
|
|
)
|
|
svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
|
|
await svc.upsert(platform="hentaifoundry", credential_type="cookies", data=netscape_in)
|
|
path = await svc.get_cookies_path("hentaifoundry")
|
|
contents = path.read_text()
|
|
# Subdomain-wide originals preserved.
|
|
assert ".hentai-foundry.com\tTRUE\t/\tTRUE\t1900000000\tPHPSESSID\tsess123" in contents
|
|
# Host-only duplicates appended for both names.
|
|
assert "www.hentai-foundry.com\tFALSE\t/\tTRUE\t1900000000\tPHPSESSID\tsess123" in contents
|
|
assert "www.hentai-foundry.com\tFALSE\t/\tTRUE\t1900000000\tYII_CSRF_TOKEN\ttoken456" in contents
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_cookies_path_hf_idempotent_when_host_only_present(db, crypto, tmp_path):
|
|
"""If the captured cookies already include a host-only PHPSESSID
|
|
on www.hentai-foundry.com (e.g. a future extension fix that
|
|
preserves browser hostOnly state), don't double-inject."""
|
|
netscape_in = (
|
|
"# Netscape HTTP Cookie File\n"
|
|
".hentai-foundry.com\tTRUE\t/\tTRUE\t1900000000\tPHPSESSID\tsess123\n"
|
|
"www.hentai-foundry.com\tFALSE\t/\tTRUE\t1900000000\tPHPSESSID\tsess123\n"
|
|
)
|
|
svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
|
|
await svc.upsert(platform="hentaifoundry", credential_type="cookies", data=netscape_in)
|
|
path = await svc.get_cookies_path("hentaifoundry")
|
|
contents = path.read_text()
|
|
# Should count PHPSESSID exactly twice — the original two lines, no third.
|
|
assert contents.count("PHPSESSID\tsess123") == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_cookies_path_hf_ignores_unrelated_cookies(db, crypto, tmp_path):
|
|
"""The injection should only target session/CSRF cookies. Other HF
|
|
cookies (e.g. analytics) stay subdomain-wide as captured."""
|
|
netscape_in = (
|
|
"# Netscape HTTP Cookie File\n"
|
|
".hentai-foundry.com\tTRUE\t/\tTRUE\t1900000000\t_ga\tGA1.2.x\n"
|
|
)
|
|
svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
|
|
await svc.upsert(platform="hentaifoundry", credential_type="cookies", data=netscape_in)
|
|
path = await svc.get_cookies_path("hentaifoundry")
|
|
contents = path.read_text()
|
|
assert "www.hentai-foundry.com" not in contents
|
|
assert contents.count("_ga") == 1
|
|
|
|
|
|
@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
|