381 lines
18 KiB
Python
381 lines
18 KiB
Python
"""Magazine sources: parsing, host restriction, size/type checks. No network."""
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-test-config-"))
|
|
|
|
from core import magazine_sources as ms
|
|
|
|
|
|
class FakeResp:
|
|
def __init__(self, status=200, body=b"", headers=None, json_data=None):
|
|
self.status_code = status
|
|
self._body = body
|
|
self.headers = headers or {}
|
|
self._json = json_data
|
|
self.is_redirect = status in (301, 302, 303, 307, 308)
|
|
|
|
def json(self):
|
|
if self._json is None:
|
|
raise ValueError("no json")
|
|
return self._json
|
|
|
|
def iter_content(self, n):
|
|
for i in range(0, len(self._body), n):
|
|
yield self._body[i:i + n]
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, routes):
|
|
self.routes = routes
|
|
self.calls = []
|
|
|
|
def get(self, url, **kw):
|
|
self.calls.append(url)
|
|
self.kwargs = getattr(self, "kwargs", []) + [kw]
|
|
for prefix, resp in self.routes.items():
|
|
if url.startswith(prefix):
|
|
return resp() if callable(resp) else resp
|
|
return FakeResp(404)
|
|
|
|
|
|
OPDS = b"""<?xml version="1.0"?>
|
|
<feed xmlns="http://www.w3.org/2005/Atom">
|
|
<entry><title>c't 12/2026</title><author><name>Heise</name></author><summary>Issue</summary>
|
|
<link rel="http://opds-spec.org/acquisition" type="application/pdf" href="/dl/1.pdf"/>
|
|
<link rel="http://opds-spec.org/image/thumbnail" href="/thumb/1.jpg"/></entry>
|
|
<entry><title>Elsewhere</title>
|
|
<link rel="http://opds-spec.org/acquisition" type="application/pdf" href="http://evil.example/x.pdf"/></entry>
|
|
<entry><title>EPUB only</title>
|
|
<link rel="http://opds-spec.org/acquisition" type="application/epub+zip" href="/dl/2.epub"/></entry>
|
|
</feed>"""
|
|
|
|
|
|
class OpdsTests(unittest.TestCase):
|
|
def provider(self, routes):
|
|
return ms.OpdsProvider("opds-1", "NAS", "http://nas.local:8083/opds/", FakeSession(routes))
|
|
|
|
def test_search_keeps_only_same_host_pdf_entries(self):
|
|
# Most specific route first: the fake session matches by prefix.
|
|
p = self.provider({"http://nas.local:8083/opds/search": FakeResp(body=OPDS),
|
|
"http://nas.local:8083/opds/": FakeResp(body=b'<feed xmlns="http://www.w3.org/2005/Atom"/>')})
|
|
items = p.search("ct")
|
|
self.assertEqual([i.title for i in items], ["c't 12/2026"])
|
|
self.assertEqual(items[0].id, "http://nas.local:8083/dl/1.pdf")
|
|
self.assertEqual(items[0].thumb_url, "http://nas.local:8083/thumb/1.jpg")
|
|
|
|
def test_resolve_rejects_other_hosts(self):
|
|
p = self.provider({})
|
|
with self.assertRaises(ms.SourceError):
|
|
p.resolve_pdf("http://169.254.169.254/latest/meta-data")
|
|
|
|
def test_doctype_and_entities_are_refused(self):
|
|
evil = b'<?xml version="1.0"?><!DOCTYPE lol [<!ENTITY a "aaaa">]><feed xmlns="http://www.w3.org/2005/Atom"/>'
|
|
with self.assertRaises(ms.SourceError):
|
|
ms.parse_xml_safely(evil)
|
|
|
|
def test_malformed_feed_is_a_source_error(self):
|
|
with self.assertRaises(ms.SourceError):
|
|
ms.parse_xml_safely(b"<feed")
|
|
|
|
def test_empty_query_rejected(self):
|
|
with self.assertRaises(ms.SourceError):
|
|
self.provider({}).search(" ")
|
|
|
|
|
|
def feed(entries, next_href=None):
|
|
body = '<feed xmlns="http://www.w3.org/2005/Atom">'
|
|
for n in entries:
|
|
body += (f'<entry><title>Issue {n}</title><link rel="http://opds-spec.org/acquisition" type="application/pdf" '
|
|
f'href="/dl/{n}.pdf"/></entry>')
|
|
if next_href:
|
|
body += f'<link rel="next" href="{next_href}"/>'
|
|
return FakeResp(body=(body + "</feed>").encode())
|
|
|
|
|
|
ROOT = FakeResp(body=b'<feed xmlns="http://www.w3.org/2005/Atom"/>')
|
|
|
|
|
|
class OpdsPagingTests(unittest.TestCase):
|
|
def provider(self, routes, **kw):
|
|
routes = {**routes, "http://nas.local/opds/": ROOT}
|
|
self.session = FakeSession(routes)
|
|
return ms.OpdsProvider("opds-1", "NAS", "http://nas.local/opds/", self.session, **kw)
|
|
|
|
def test_a_second_results_page_follows_the_catalogues_next_link(self):
|
|
# Most specific route first: the fake session matches by prefix.
|
|
p = self.provider({"http://nas.local/opds/search?page=2": feed(range(16, 31)),
|
|
"http://nas.local/opds/search": feed(range(1, 16), "/opds/search?page=2")})
|
|
self.assertEqual(p.search("x", 1)[0].title, "Issue 1")
|
|
page2 = p.search("x", 2)
|
|
self.assertEqual([i.title for i in page2], [f"Issue {n}" for n in range(21, 31)])
|
|
|
|
def test_a_next_link_to_another_host_is_not_followed(self):
|
|
p = self.provider({"http://nas.local/opds/search": feed([1, 2], "http://evil.example/opds?page=2")})
|
|
self.assertEqual(len(p.search("x", 2)), 0)
|
|
self.assertFalse(any("evil.example" in c for c in self.session.calls))
|
|
|
|
def test_a_next_link_loop_ends(self):
|
|
p = self.provider({"http://nas.local/opds/search": feed([1], "/opds/search")})
|
|
self.assertEqual(len(p.search("x", 3)), 0)
|
|
self.assertLessEqual(len(self.session.calls), 3)
|
|
|
|
def test_paging_is_bounded(self):
|
|
routes = {f"http://nas.local/opds/search?p={n}": feed([n], f"/opds/search?p={n + 1}") for n in range(1, 50)}
|
|
routes["http://nas.local/opds/search"] = feed([0], "/opds/search?p=1")
|
|
p = self.provider(routes)
|
|
p.search("x", 9)
|
|
self.assertLessEqual(len(self.session.calls), 1 + ms.OpdsProvider.MAX_FEED_PAGES)
|
|
|
|
|
|
class OpdsLoginTests(unittest.TestCase):
|
|
def setUp(self):
|
|
import os
|
|
self.os = os
|
|
os.environ["TEST_OPDS_SECRET"] = "s3cret-value"
|
|
self.addCleanup(lambda: os.environ.pop("TEST_OPDS_SECRET", None))
|
|
|
|
def provider(self, routes, **kw):
|
|
self.session = FakeSession({**routes, "http://nas.local/opds/": ROOT})
|
|
return ms.OpdsProvider("opds-1", "NAS", "http://nas.local/opds/", self.session, **kw)
|
|
|
|
def test_credentials_come_from_the_named_environment_variable_and_go_to_the_catalogue(self):
|
|
p = self.provider({"http://nas.local/opds/search": feed([1])}, username="me", password_env="TEST_OPDS_SECRET")
|
|
self.assertEqual(len(p.search("x")), 1)
|
|
self.assertTrue(all(kw.get("auth") == ("me", "s3cret-value") for kw in self.session.kwargs))
|
|
|
|
def test_an_open_catalogue_sends_no_auth(self):
|
|
p = self.provider({"http://nas.local/opds/search": feed([1])})
|
|
p.search("x")
|
|
self.assertTrue(all("auth" not in kw for kw in self.session.kwargs))
|
|
|
|
def test_a_missing_password_is_a_clear_error_that_never_shows_a_secret(self):
|
|
p = self.provider({}, username="me", password_env="TEST_OPDS_UNSET")
|
|
with self.assertRaises(ms.SourceError) as ctx:
|
|
p.search("x")
|
|
self.assertIn("TEST_OPDS_UNSET", str(ctx.exception))
|
|
self.assertNotIn("s3cret", str(ctx.exception))
|
|
|
|
def test_downloads_are_authenticated_too(self):
|
|
import tempfile
|
|
from pathlib import Path
|
|
p = self.provider({"http://nas.local/dl": FakeResp(body=b"%PDF-1.7 x")}, username="me", password_env="TEST_OPDS_SECRET")
|
|
ms.download_pdf(p, "http://nas.local/dl/1.pdf", Path(tempfile.mkdtemp()) / "source.pdf", session=self.session)
|
|
self.assertEqual(self.session.kwargs[-1].get("auth"), ("me", "s3cret-value"))
|
|
|
|
def test_auth_is_not_sent_when_a_redirect_leaves_the_host(self):
|
|
import tempfile
|
|
from pathlib import Path
|
|
p = self.provider({"http://nas.local/dl": FakeResp(status=302, headers={"location": "http://evil.example/x"})},
|
|
username="me", password_env="TEST_OPDS_SECRET")
|
|
with self.assertRaises(ms.SourceError):
|
|
ms.download_pdf(p, "http://nas.local/dl/1.pdf", Path(tempfile.mkdtemp()) / "source.pdf", session=self.session)
|
|
self.assertFalse(any("evil.example" in c for c in self.session.calls))
|
|
|
|
def test_settings_with_an_implausible_variable_name_are_ignored(self):
|
|
provs = ms.build_providers([
|
|
{"name": "ok", "url": "http://nas.local/opds", "username": "me", "password_env": "NAS_PASSWORD"},
|
|
{"name": "bad", "url": "http://nas2.local/opds", "username": "me", "password_env": "x; rm -rf /"},
|
|
{"name": "lower", "url": "http://nas3.local/opds", "username": "me", "password_env": "lowercase"}])
|
|
self.assertEqual(sorted(provs), ["internet-archive", "opds-1"])
|
|
|
|
def test_the_source_list_never_exposes_credentials(self):
|
|
from unittest import mock
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from routes import magazine_sources as routes
|
|
app = FastAPI(); app.include_router(routes.router)
|
|
cats = [{"name": "NAS", "url": "http://nas.local/opds", "username": "me", "password_env": "TEST_OPDS_SECRET"}]
|
|
with mock.patch.object(routes, "_load_settings", return_value={"magazine_opds_catalogs": cats}):
|
|
body = TestClient(app).get("/api/magazine-sources").text
|
|
self.assertNotIn("s3cret", body)
|
|
self.assertNotIn("TEST_OPDS_SECRET", body)
|
|
self.assertNotIn("me\"", body.replace("name\"", ""))
|
|
|
|
|
|
class ArchiveTests(unittest.TestCase):
|
|
def test_search_builds_items_and_skips_bad_identifiers(self):
|
|
docs = [{"identifier": "ct-2026-01", "title": "c't", "creator": ["Heise"], "date": "2026-01-01T00:00:00Z"},
|
|
{"identifier": "../../etc/passwd", "title": "x"}]
|
|
s = FakeSession({"https://archive.org/advancedsearch.php": FakeResp(json_data={"response": {"docs": docs}})})
|
|
items = ms.InternetArchiveProvider(s).search("ct")
|
|
self.assertEqual([i.id for i in items], ["ct-2026-01"])
|
|
self.assertEqual(items[0].date, "2026-01-01")
|
|
self.assertEqual(items[0].author, "Heise")
|
|
|
|
def test_resolve_prefers_original_pdf(self):
|
|
meta = {"metadata": {}, "files": [
|
|
{"name": "a_text.pdf", "source": "derivative", "size": "9999"},
|
|
{"name": "issue.pdf", "source": "original", "size": "100"}]}
|
|
s = FakeSession({"https://archive.org/metadata/": FakeResp(json_data=meta)})
|
|
url = ms.InternetArchiveProvider(s).resolve_pdf("ct-2026-01")
|
|
self.assertEqual(url, "https://archive.org/download/ct-2026-01/issue.pdf")
|
|
|
|
def test_lending_only_is_refused(self):
|
|
s = FakeSession({"https://archive.org/metadata/": FakeResp(
|
|
json_data={"metadata": {"access-restricted-item": "true"}, "files": []})})
|
|
with self.assertRaises(ms.SourceError):
|
|
ms.InternetArchiveProvider(s).resolve_pdf("loan")
|
|
|
|
def test_rate_limit_message(self):
|
|
s = FakeSession({"https://archive.org/advancedsearch.php": FakeResp(status=429)})
|
|
with self.assertRaisesRegex(ms.SourceError, "rate-limiting"):
|
|
ms.InternetArchiveProvider(s).search("ct")
|
|
|
|
def test_bad_identifier_refused(self):
|
|
with self.assertRaises(ms.SourceError):
|
|
ms.InternetArchiveProvider(FakeSession({})).resolve_pdf("a/b")
|
|
|
|
|
|
class DownloadTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = Path(tempfile.mkdtemp(prefix="ttvc-dl-"))
|
|
self.p = ms.OpdsProvider("opds-1", "NAS", "http://nas.local/opds/", None)
|
|
|
|
def test_downloads_pdf(self):
|
|
s = FakeSession({"http://nas.local/dl": FakeResp(body=b"%PDF-1.7 hello")})
|
|
n = ms.download_pdf(self.p, "http://nas.local/dl/1.pdf", self.dir / "source.pdf", session=s)
|
|
self.assertEqual(n, 14)
|
|
self.assertEqual((self.dir / "source.pdf").read_bytes()[:5], b"%PDF-")
|
|
self.assertFalse((self.dir / "source.part").exists())
|
|
|
|
def test_rejects_non_pdf_and_cleans_up(self):
|
|
s = FakeSession({"http://nas.local/dl": FakeResp(body=b"<html>login</html>")})
|
|
with self.assertRaisesRegex(ms.SourceError, "PDF"):
|
|
ms.download_pdf(self.p, "http://nas.local/dl/1.pdf", self.dir / "source.pdf", session=s)
|
|
self.assertEqual(list(self.dir.iterdir()), [])
|
|
|
|
def test_redirect_to_other_host_is_refused(self):
|
|
s = FakeSession({"http://nas.local/dl": FakeResp(status=302, headers={"location": "http://evil.example/x"})})
|
|
with self.assertRaisesRegex(ms.SourceError, "different host"):
|
|
ms.download_pdf(self.p, "http://nas.local/dl/1.pdf", self.dir / "source.pdf", session=s)
|
|
|
|
def test_oversize_declared_is_refused(self):
|
|
s = FakeSession({"http://nas.local/dl": FakeResp(body=b"%PDF-", headers={"content-length": str(ms.MAX_DOWNLOAD_BYTES + 1)})})
|
|
with self.assertRaisesRegex(ms.SourceError, "larger"):
|
|
ms.download_pdf(self.p, "http://nas.local/dl/1.pdf", self.dir / "source.pdf", session=s)
|
|
|
|
def test_http_error(self):
|
|
s = FakeSession({"http://nas.local/dl": FakeResp(status=403)})
|
|
with self.assertRaisesRegex(ms.SourceError, "403"):
|
|
ms.download_pdf(self.p, "http://nas.local/dl/1.pdf", self.dir / "source.pdf", session=s)
|
|
|
|
|
|
class BuildProvidersTests(unittest.TestCase):
|
|
def test_skips_bad_and_credentialed_catalogues(self):
|
|
provs = ms.build_providers([
|
|
{"name": "NAS", "url": "http://nas.local/opds"},
|
|
{"name": "bad", "url": "ftp://x"},
|
|
{"name": "creds", "url": "http://user:pw@nas.local/opds"},
|
|
{"url": ""}])
|
|
self.assertEqual(sorted(provs), ["internet-archive", "opds-1"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class RouteTests(unittest.TestCase):
|
|
"""The HTTP layer over the providers, on a minimal app (no auth middleware, no network)."""
|
|
|
|
def setUp(self):
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from unittest import mock
|
|
from routes import magazine_sources as routes
|
|
self.routes = routes
|
|
self.app = FastAPI()
|
|
self.app.include_router(routes.router)
|
|
self.client = TestClient(self.app)
|
|
self.mock = mock
|
|
|
|
def test_lists_archive_and_configured_catalogues(self):
|
|
with self.mock.patch.object(self.routes, "_load_settings",
|
|
return_value={"magazine_opds_catalogs": [{"name": "NAS", "url": "http://nas.local/opds"}]}):
|
|
data = self.client.get("/api/magazine-sources").json()
|
|
self.assertEqual([s["key"] for s in data["sources"]], ["internet-archive", "opds-1"])
|
|
|
|
def test_unknown_source_is_404(self):
|
|
with self.mock.patch.object(self.routes, "_load_settings", return_value={}):
|
|
self.assertEqual(self.client.get("/api/magazine-sources/nope/search?q=x").status_code, 404)
|
|
self.assertEqual(self.client.get("/api/magazine-sources/..%2f/search?q=x").status_code, 404)
|
|
|
|
def test_source_errors_become_502_with_a_message(self):
|
|
class Boom:
|
|
key = "internet-archive"
|
|
label = "x"
|
|
def search(self, q, page=1):
|
|
raise ms.SourceError("The source is rate-limiting requests; try again in a minute")
|
|
with self.mock.patch.object(self.routes, "_providers", return_value={"internet-archive": Boom()}):
|
|
r = self.client.get("/api/magazine-sources/internet-archive/search?q=x")
|
|
self.assertEqual(r.status_code, 502)
|
|
self.assertIn("rate-limiting", r.json()["detail"])
|
|
|
|
def test_import_requires_an_item(self):
|
|
with self.mock.patch.object(self.routes, "_load_settings", return_value={}):
|
|
r = self.client.post("/api/magazine-sources/import", json={"source": "internet-archive"})
|
|
self.assertEqual(r.status_code, 400)
|
|
|
|
|
|
class CatalogueEndpointTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from unittest import mock
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from routes import magazine_sources as routes
|
|
self.store = {"magazine_opds_catalogs": []}
|
|
for name, fn in (("_load_settings", lambda: dict(self.store)), ("_save_settings", lambda s: self.store.update(s))):
|
|
p = mock.patch.object(routes, name, side_effect=fn)
|
|
p.start(); self.addCleanup(p.stop)
|
|
app = FastAPI(); app.include_router(routes.router)
|
|
self.client = TestClient(app)
|
|
|
|
def put(self, cats):
|
|
return self.client.put("/api/magazine-sources/catalogues", json={"catalogues": cats})
|
|
|
|
def test_a_valid_list_is_saved_and_read_back(self):
|
|
r = self.put([{"name": " My NAS ", "url": "http://nas.local:8083/opds/", "username": "me", "password_env": "NAS_PW"},
|
|
{"name": "", "url": "https://books.example/opds"}])
|
|
self.assertEqual(r.status_code, 200, r.text)
|
|
got = self.client.get("/api/magazine-sources/catalogues").json()["catalogues"]
|
|
self.assertEqual([c["name"] for c in got], ["My NAS", "books.example"])
|
|
self.assertEqual(got[0]["password_env"], "NAS_PW")
|
|
self.assertFalse(got[0]["password_set"]) # the variable is not set in this environment
|
|
|
|
def test_credentials_in_the_address_are_refused(self):
|
|
self.assertEqual(self.put([{"url": "http://me:pw@nas.local/opds"}]).status_code, 400)
|
|
|
|
def test_other_schemes_and_missing_hosts_are_refused(self):
|
|
for bad in ("ftp://nas/opds", "javascript:alert(1)", "file:///etc/passwd", "http://", "nas.local/opds", ""):
|
|
self.assertEqual(self.put([{"url": bad}]).status_code, 400, bad)
|
|
|
|
def test_a_login_needs_both_halves_and_a_plausible_variable_name(self):
|
|
self.assertEqual(self.put([{"url": "http://a.local/o", "username": "me"}]).status_code, 400)
|
|
self.assertEqual(self.put([{"url": "http://a.local/o", "password_env": "PW"}]).status_code, 400)
|
|
self.assertEqual(self.put([{"url": "http://a.local/o", "username": "me", "password_env": "lower; rm"}]).status_code, 400)
|
|
|
|
def test_duplicates_too_many_and_wrong_shapes_are_refused(self):
|
|
self.assertEqual(self.put([{"url": "http://a.local/o"}, {"url": "http://a.local/o"}]).status_code, 400)
|
|
self.assertEqual(self.put([{"url": f"http://h{n}.local/o"} for n in range(11)]).status_code, 400)
|
|
self.assertEqual(self.put("nope").status_code, 400)
|
|
self.assertEqual(self.put(["nope"]).status_code, 400)
|
|
|
|
def test_a_refused_list_changes_nothing(self):
|
|
self.put([{"url": "http://ok.local/o"}])
|
|
self.put([{"url": "http://ok.local/o"}, {"url": "ftp://bad"}])
|
|
self.assertEqual([c["url"] for c in self.store["magazine_opds_catalogs"]], ["http://ok.local/o"])
|
|
|
|
def test_the_password_itself_is_never_returned(self):
|
|
import os
|
|
os.environ["CAT_TEST_PW"] = "topsecret"
|
|
self.addCleanup(lambda: os.environ.pop("CAT_TEST_PW", None))
|
|
self.put([{"url": "http://a.local/o", "username": "me", "password_env": "CAT_TEST_PW"}])
|
|
r = self.client.get("/api/magazine-sources/catalogues")
|
|
self.assertNotIn("topsecret", r.text)
|
|
self.assertTrue(r.json()["catalogues"][0]["password_set"])
|