"""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""" c't 12/2026HeiseIssue Elsewhere EPUB only """ 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'')}) 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']>' 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"Issue {n}') if next_href: body += f'' return FakeResp(body=(body + "").encode()) ROOT = FakeResp(body=b'') 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"login")}) 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"])