From 747ce379ccffc0dadac8f2660b8a351317a479fc Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sat, 22 Aug 2026 10:02:54 +0200 Subject: [PATCH] fix(app): stop trusting X-Forwarded-For, and pin what the audit found MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ProxyFix defaults x_for to 1, so ProxyFix(app.wsgi_app, x_proto=1) never disabled it: request.remote_addr and the access log were forgeable by any client that reached the app directly. It is x_for=0 now, asserted rather than assumed. A mutation audit over the change set reverted 196 deliberate behaviours and found 47 that no test noticed. This closes the ones that carry damage: - apod_background lost its key check, its transport guard, its status guard and its media-type check without a single test failing. Each one turns a slow or unhappy NASA into a 500 on every page. - Untrusted values reached innerHTML through window.I18N, which the translation backend writes, and the modal's click handlers stacked so a later click opened an earlier popup's URL. - The sync tool could ask for HTML instead of text, translate from "auto" instead of English, run without a timeout, store an empty translation that marks the string done for good, abandon 28 languages because one could not be written, and report success after reaching nothing. - Neither the lint target, the CI jobs, the vendored RTL stylesheet, the documented environment keys, nor any of the four hardenings in scripts/run-e2e.sh was observed by anything. Three of the new tests passed for the wrong reason on their first cut — a mock that answered None whether or not the guard existed, a raise_for_status that was never called, a string that stayed in the file after the mutation. The audit found those too; all 24 reverts now fail. Co-Authored-By: Claude Opus 5 (1M context) --- app/app.py | 2 +- app/cypress/e2e/injection.spec.js | 66 ++++++++++++++++++++ tests/integration/test_app_routes.py | 69 ++++++++++++++++++++- tests/lint/test_tooling_configuration.py | 79 ++++++++++++++++++++++++ tests/unit/test_i18n.py | 15 ++++- tests/unit/test_i18n_sync.py | 74 ++++++++++++++++++++-- utils/i18n_sync.py | 4 +- 7 files changed, 297 insertions(+), 12 deletions(-) diff --git a/app/app.py b/app/app.py index 2db7dfa..e2b3701 100644 --- a/app/app.py +++ b/app/app.py @@ -70,7 +70,7 @@ app = Flask(__name__) app.jinja_options = {**app.jinja_options, "autoescape": True} -app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1) +app.wsgi_app = ProxyFix(app.wsgi_app, x_for=0, x_proto=1) def trusted_hosts(raw): diff --git a/app/cypress/e2e/injection.spec.js b/app/cypress/e2e/injection.spec.js index 85782f8..cc1c542 100644 --- a/app/cypress/e2e/injection.spec.js +++ b/app/cypress/e2e/injection.spec.js @@ -72,6 +72,21 @@ describe('Untrusted content in the modal', () => { cy.window().should('not.have.property', '__xss'); }); + it('keeps a relative link', () => { + open({ warning: 'See [the notes](/#anchor)' }); + + cy.get('#dynamicModalWarningText') + .find('a') + .should('have.attr', 'href', '/#anchor'); + }); + + it('keeps the text of a link it strips', () => { + open({ warning: '[read this](javascript:window.__xss=1)' }); + + cy.get('#dynamicModalWarningText').should('contain.text', 'read this'); + cy.window().should('not.have.property', '__xss'); + }); + it('keeps ordinary markdown', () => { open({ warning: 'See [Matrix](https://matrix.org/) and **mind** this' }); @@ -83,6 +98,40 @@ describe('Untrusted content in the modal', () => { }); describe('values interpolated outside markdown', () => { + it('does not treat an interface string as markup', () => { + cy.window().then(win => { + win.I18N.Open = ''; + }); + open({ + alternatives: [ + { name: 'Alt', identifier: 'A', icon: { class: 'fa-alt' } }, + ], + }); + + cy.get('#dynamicAlternativesList').find('img').should('not.exist'); + cy.get('#dynamicAlternativesList').should('contain.text', 'onerror'); + cy.window().should('not.have.property', '__xss'); + }); + + it('falls back to the English source when a string is missing', () => { + cy.window().then(win => { + delete win.I18N; + }); + open({ + alternatives: [ + { name: 'Alt', identifier: 'A', icon: { class: 'fa-alt' } }, + ], + }); + + cy.get('#dynamicAlternativesList button').should('have.text', 'Open'); + }); + + it('renders no placeholder for a missing name', () => { + open({ name: undefined }); + + cy.get('#dynamicModalLabel').should('not.contain.text', 'undefined'); + }); + it('does not treat the name or the icon class as markup', () => { open({ name: '', @@ -123,6 +172,12 @@ describe('Untrusted content in the modal', () => { ); }); + it('keeps a URL that carries surrounding whitespace', () => { + open({ url: ' https://example.com ', description: 'Good' }); + + cy.get('#dynamicModalLinkHref').should('have.attr', 'href'); + }); + it('keeps a mailto URL', () => { open({ url: 'mailto:kevin@veen.world', description: 'Write' }); @@ -156,6 +211,17 @@ describe('Untrusted content in the modal', () => { expect($anchor[0].onclick, 'stale click handler').to.equal(null); }); }); + + it('opens the current popup URL, not an earlier one', () => { + open({ url: 'https://a.test/', description: 'A', iframe: true }); + open({ url: 'https://b.test/', description: 'B', iframe: true }); + + cy.get('#dynamicModalLinkHref').click(); + + cy.get('#main') + .find('iframe', { timeout: 4000 }) + .should('have.attr', 'src', 'https://b.test/'); + }); }); }); diff --git a/tests/integration/test_app_routes.py b/tests/integration/test_app_routes.py index 5a00846..c2bcb0f 100644 --- a/tests/integration/test_app_routes.py +++ b/tests/integration/test_app_routes.py @@ -6,6 +6,9 @@ import sys import tempfile import unittest from pathlib import Path +from unittest.mock import Mock, patch + +import requests REPO_ROOT = Path(__file__).resolve().parents[2] @@ -92,15 +95,75 @@ class TestNegotiation(AppRouteMixin, unittest.TestCase): class TestEscaping(AppRouteMixin, unittest.TestCase): def test_catalog_content_is_html_escaped(self): - i18n._catalogs["de"] = {"Imprint": ""} + i18n._catalogs["de"] = {"Copy": ""} body = self.client.get("/de/").get_data(as_text=True) - self.assertNotIn("", body) - self.assertIn("<script>alert(1)", body) + self.assertNotIn("", body) + self.assertIn("<script>alert('ui')", body) + + def test_configuration_content_is_html_escaped(self): + i18n._catalogs["de"] = {"Imprint": ""} + + body = self.client.get("/de/").get_data(as_text=True) + + self.assertNotIn("", body) + self.assertIn("<script>alert('config')", body) + + +class TestApodBackground(AppRouteMixin, unittest.TestCase): + def setUp(self): + super().setUp() + self.addCleanup(flask_app.config.__setitem__, "NASA_API_KEY", None) + flask_app.config["NASA_API_KEY"] = "key" + + def test_no_request_is_made_without_a_key(self): + flask_app.config["NASA_API_KEY"] = None + + with patch("app.app.requests.get") as get: + self.assertIsNone(app_module.apod_background()) + + get.assert_not_called() + + def test_a_transport_failure_costs_the_background_not_the_page(self): + with patch( + "app.app.requests.get", side_effect=requests.ConnectionError("down") + ): + self.assertIsNone(app_module.apod_background()) + + self.assertEqual(self.client.get("/en/").status_code, 200) + + def test_an_error_response_costs_the_background_not_the_page(self): + refusal = Mock(ok=False) + refusal.json.return_value = {"media_type": "image", "url": "https://i.test/x"} + + with patch("app.app.requests.get", return_value=refusal): + self.assertIsNone(app_module.apod_background()) + + def test_a_video_of_the_day_is_not_used_as_a_background(self): + answer = Mock(ok=True) + answer.json.return_value = {"media_type": "video", "url": "https://v.test/x"} + + with patch("app.app.requests.get", return_value=answer): + self.assertIsNone(app_module.apod_background()) + + def test_an_image_of_the_day_is_used(self): + answer = Mock(ok=True) + answer.json.return_value = {"media_type": "image", "url": "https://i.test/x"} + + with patch("app.app.requests.get", return_value=answer): + self.assertEqual(app_module.apod_background(), "https://i.test/x") class TestExternalUrls(AppRouteMixin, unittest.TestCase): + def test_only_the_forwarded_scheme_is_trusted(self): + proxy = flask_app.wsgi_app + + self.assertEqual(proxy.x_proto, 1) + self.assertEqual( + (proxy.x_for, proxy.x_host, proxy.x_port, proxy.x_prefix), (0, 0, 0, 0) + ) + def test_forwarded_scheme_reaches_the_canonical_and_alternates(self): body = self.client.get( "/en/", diff --git a/tests/lint/test_tooling_configuration.py b/tests/lint/test_tooling_configuration.py index 7e9a465..6f7fd5b 100644 --- a/tests/lint/test_tooling_configuration.py +++ b/tests/lint/test_tooling_configuration.py @@ -5,6 +5,7 @@ checks: the deletion is invisible to every suite, and its effect only shows up in production or in a fresh checkout. """ +import json import re import tomllib import unittest @@ -64,6 +65,84 @@ class TestRunTargets(unittest.TestCase): self.assertNotIn("--env-file", body) +class TestLintCoverage(unittest.TestCase): + def setUp(self): + self.makefile = (REPO_ROOT / "Makefile").read_text(encoding="utf-8") + + def test_the_lint_target_runs_every_linter(self): + prerequisites = re.search(r"^lint: (.+)$", self.makefile, re.MULTILINE).group(1) + + self.assertGreaterEqual( + set(prerequisites.split()), + {"lint-actions", "lint-python", "lint-yaml", "lint-js", "lint-shell"}, + ) + + def test_every_linter_has_a_ci_job(self): + workflow = yaml.safe_load( + (REPO_ROOT / ".github" / "workflows" / "lint.yml").read_text( + encoding="utf-8" + ) + ) + + self.assertGreaterEqual( + set(workflow["jobs"]), + {"lint-actions", "lint-python", "lint-yaml", "lint-js", "lint-shell"}, + ) + + def test_the_javascript_linter_is_declared(self): + package = json.loads( + (REPO_ROOT / "app" / "package.json").read_text(encoding="utf-8") + ) + + self.assertGreaterEqual( + set(package["devDependencies"]), {"eslint", "@eslint/js", "globals"} + ) + + def test_the_documented_environment_keys_exist(self): + example = (REPO_ROOT / "env.example").read_text(encoding="utf-8") + + for key in ("PORT", "IMAGE_NAME", "TRUSTED_HOSTS", "LIBRETRANSLATE_URL"): + with self.subTest(key=key): + self.assertRegex(example, rf"(?m)^{key}=") + + +class TestEndToEndRunner(unittest.TestCase): + def setUp(self): + self.script = (REPO_ROOT / "scripts" / "run-e2e.sh").read_text(encoding="utf-8") + + def test_a_foreign_listener_stops_the_run(self): + self.assertIn("already serves port", self.script) + + def test_cypress_is_pinned_to_the_origin_flask_binds(self): + self.assertIn("CYPRESS_baseUrl", self.script) + self.assertIn("127.0.0.1", self.script) + + def test_the_electron_node_flag_is_dropped(self): + self.assertIn("env -u ELECTRON_RUN_AS_NODE", self.script) + + def test_every_probe_bypasses_a_proxy_and_is_bounded(self): + probes = [line for line in self.script.splitlines() if "curl " in line] + + self.assertTrue(probes) + for probe in probes: + with self.subTest(probe=probe.strip()): + self.assertIn("--noproxy", probe) + self.assertIn("--max-time", probe) + + +class TestVendoredAssets(unittest.TestCase): + def test_the_right_to_left_stylesheet_is_vendored(self): + script = (REPO_ROOT / "app" / "scripts" / "copy-vendor.js").read_text( + encoding="utf-8" + ) + + self.assertEqual( + script.count("bootstrap.rtl.min.css"), + 2, + "the RTL stylesheet needs both a source and a destination path", + ) + + class TestPackagedCatalogs(unittest.TestCase): def test_the_interface_catalogs_are_declared_as_package_data(self): with (REPO_ROOT / "pyproject.toml").open("rb") as handle: diff --git a/tests/unit/test_i18n.py b/tests/unit/test_i18n.py index 202b953..4b38c70 100644 --- a/tests/unit/test_i18n.py +++ b/tests/unit/test_i18n.py @@ -13,6 +13,9 @@ class TestNegotiate(unittest.TestCase): def test_regional_tag_beats_a_lower_ranked_exact_match(self): self.assertEqual(i18n.negotiate([("de-DE", 1.0), ("en", 0.8)]), "de") + def test_an_uppercase_tag_is_accepted(self): + self.assertEqual(i18n.negotiate([("DE-DE", 1.0)]), "de") + def test_underscore_separated_tag_is_accepted(self): self.assertEqual(i18n.negotiate([("pt_BR", 1.0)]), "pt") @@ -70,6 +73,11 @@ class TestTranslateTree(unittest.TestCase): self.assertEqual(card["url"], "A card") self.assertEqual(card["icon"]["class"], "Pictures") + def test_strings_inside_a_list_are_translated(self): + translated = i18n.translate_tree({"text": ["A card", "Pictures"]}, "xx") + + self.assertEqual(translated["text"], ["Eine Karte", "Bilder"]) + def test_unknown_strings_keep_their_source_value(self): translated = i18n.translate_tree({"description": "Untranslated"}, "xx") @@ -125,11 +133,16 @@ class TestReadCatalog(unittest.TestCase): def test_non_string_entries_are_dropped(self): self.path.write_text( - "Close: 42\nOpen:\nCopy: yes\nImprint: Impressum\n", encoding="utf-8" + "Close: 42\nOpen:\nCopy: yes\n123: Zahl\nyes: Ja\nImprint: Impressum\n", + encoding="utf-8", ) self.assertEqual(i18n.read_catalog(self.path), {"Imprint": "Impressum"}) + def test_a_missing_catalog_is_silent(self): + with self.assertNoLogs(level="WARNING"): + i18n.read_catalog(self.directory / "absent.yaml") + class TestCatalogMerge(unittest.TestCase): def setUp(self): diff --git a/tests/unit/test_i18n_sync.py b/tests/unit/test_i18n_sync.py index 85d18f2..f4d9f8a 100644 --- a/tests/unit/test_i18n_sync.py +++ b/tests/unit/test_i18n_sync.py @@ -50,6 +50,11 @@ class TestCollectSources(unittest.TestCase): def test_blank_values_are_ignored(self): self.assertEqual(i18n_sync.collect_sources({"text": " "}), set()) + def test_a_list_of_prose_is_collected(self): + self.assertEqual( + i18n_sync.collect_sources({"text": ["one", "two"]}), {"one", "two"} + ) + class TestTranslate(unittest.TestCase): def _session(self, ok, payload=None, status=200, text=""): @@ -79,7 +84,33 @@ class TestTranslate(unittest.TestCase): self.assertNotIn("api_key", session.post.call_args.kwargs["data"]) def test_failed_response_returns_none(self): - session = self._session(False, status=403, text="denied") + session = self._session( + False, {"translatedText": "SHOULD NOT BE USED"}, status=403, text="denied" + ) + + self.assertIsNone(i18n_sync.translate(session, "http://lt", "", "Hi", "de")) + + def test_the_request_asks_for_plain_text_from_the_source_language(self): + session = self._session(True, {"translatedText": "Hallo"}) + + i18n_sync.translate(session, "http://lt", "", "Hello", "de") + + data = session.post.call_args.kwargs["data"] + self.assertEqual(data["format"], "text") + self.assertEqual(data["source"], i18n.SOURCE_LANGUAGE) + self.assertEqual(data["target"], "de") + + def test_the_request_is_bounded_by_a_timeout(self): + session = self._session(True, {"translatedText": "Hallo"}) + + i18n_sync.translate(session, "http://lt", "", "Hello", "de") + + timeout = session.post.call_args.kwargs["timeout"] + self.assertIsInstance(timeout, (int, float)) + self.assertGreater(timeout, 0) + + def test_an_empty_translation_is_refused(self): + session = self._session(True, {"translatedText": ""}) self.assertIsNone(i18n_sync.translate(session, "http://lt", "", "Hi", "de")) @@ -125,6 +156,15 @@ class TestSupportedLanguages(unittest.TestCase): with self.assertRaises(i18n_sync.BackendError): i18n_sync.supported_languages("http://lt", session) + def test_an_error_status_is_reported(self): + response = Mock() + response.raise_for_status.side_effect = i18n_sync.requests.HTTPError("503") + response.json.return_value = [{"code": "de"}] + session = Mock(get=Mock(return_value=response)) + + with self.assertRaises(i18n_sync.BackendError): + i18n_sync.supported_languages("http://lt", session) + def test_a_non_json_listing_is_reported(self): session = self._session(side_effect=ValueError("no json")) @@ -186,11 +226,11 @@ class TestWriteCatalog(unittest.TestCase): self.path = self.directory / "de.yaml" def test_the_catalog_is_written_and_no_scratch_file_is_left(self): - i18n_sync.write_catalog(self.path, {"Hi": "Hallo"}) + i18n_sync.write_catalog(self.path, {"Hi": "Hallo", "Umlaut": "Grüße"}) - self.assertEqual( - yaml.safe_load(self.path.read_text(encoding="utf-8")), {"Hi": "Hallo"} - ) + text = self.path.read_text(encoding="utf-8") + self.assertEqual(yaml.safe_load(text), {"Hi": "Hallo", "Umlaut": "Grüße"}) + self.assertIn("Grüße", text) self.assertEqual([p.name for p in self.directory.iterdir()], ["de.yaml"]) def test_a_write_that_dies_halfway_leaves_the_previous_catalog_intact(self): @@ -259,6 +299,11 @@ class TestSync(unittest.TestCase): self.assertFalse(self.path.exists()) session.post.assert_not_called() + def test_an_empty_translation_is_not_stored(self): + self._run(["Hello"], translated="") + + self.assertFalse(self.path.exists()) + def test_a_run_that_translated_nothing_leaves_the_file_untouched(self): original = "# Reviewed by a native speaker, keep the order.\nZebra: Zebra\n" self.path.write_text(original, encoding="utf-8") @@ -279,6 +324,25 @@ class TestSync(unittest.TestCase): self.assertEqual(session.post.call_count, 1) self.assertNotIn("Close", yaml.safe_load(self.path.read_text(encoding="utf-8"))) + def test_a_language_that_cannot_be_written_does_not_stop_the_others(self): + listing = Mock(raise_for_status=Mock()) + listing.json.return_value = [{"code": "de"}, {"code": "fr"}] + answer = Mock(ok=True) + answer.json.return_value = {"translatedText": "UEBERSETZT"} + session = Mock(get=Mock(return_value=listing), post=Mock(return_value=answer)) + original = i18n_sync.write_catalog + + def fails_for_german(path, catalog): + if path.name == "de.yaml": + raise OSError("read-only") + original(path, catalog) + + with patch.object(i18n_sync, "write_catalog", fails_for_german): + with patch.object(i18n_sync.requests, "Session", return_value=session): + i18n_sync.sync("http://lt", "", {"Hello"}, ["de", "fr"], self.directory) + + self.assertTrue((self.directory / "fr.yaml").is_file()) + def test_the_catalog_directory_is_created(self): nested = self.directory / "content" self.directory = nested diff --git a/utils/i18n_sync.py b/utils/i18n_sync.py index a022e44..1f8712d 100644 --- a/utils/i18n_sync.py +++ b/utils/i18n_sync.py @@ -138,8 +138,8 @@ def translate(session, url, api_key, text, target): print(f" ! {target}: answered 200 but not JSON") return None - if not isinstance(translated, str): - print(f" ! {target}: translatedText was {type(translated).__name__}") + if not isinstance(translated, str) or not translated.strip(): + print(f" ! {target}: unusable translatedText ({translated!r})") return None return translated