From cae678102afd02cc32cfbc80b2a02e1e36fb92e2 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 16 Jul 2026 02:50:04 +0800 Subject: [PATCH] Harden upstream RSS XML parsing Co-Authored-By: Codex --- pyproject.toml | 4 +- .../rss_source_fetch.py | 23 ++++++- tests/test_rss_source_fetch.py | 66 +++++++++++++++++++ uv.lock | 17 ++++- 4 files changed, 105 insertions(+), 5 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 2e90dca..a42ebcb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,7 +8,9 @@ version = "0.1.0" description = "Research-only political event and public disclosure tracking for US equities." readme = "README.md" requires-python = ">=3.11" -dependencies = [] +dependencies = [ + "defusedxml==0.7.1", +] license = { text = "MIT" } [project.optional-dependencies] diff --git a/src/political_event_tracking_research/rss_source_fetch.py b/src/political_event_tracking_research/rss_source_fetch.py index 98e6b13..0884243 100644 --- a/src/political_event_tracking_research/rss_source_fetch.py +++ b/src/political_event_tracking_research/rss_source_fetch.py @@ -8,11 +8,13 @@ import json import re import urllib.request -import xml.etree.ElementTree as ET from collections.abc import Callable from dataclasses import dataclass from pathlib import Path +import defusedxml.ElementTree as ET +from defusedxml.common import DefusedXmlException + from .csv_utils import read_csv_rows, write_csv_rows @@ -20,6 +22,7 @@ "Mozilla/5.0 (compatible; QuantStrategyLabSourceIngest/0.1; " "+https://github.com/QuantStrategyLab/PoliticalEventTrackingResearch)" ) +MAX_XML_BYTES = 1024 * 1024 @dataclass(frozen=True) @@ -48,6 +51,10 @@ def to_json(self) -> dict[str, object]: } +class FeedXmlError(ValueError): + """Sanitized producer-boundary XML failure.""" + + def load_feed_config(path: str | Path) -> list[FeedConfig]: feeds: list[FeedConfig] = [] for row in read_csv_rows(path): @@ -71,7 +78,10 @@ def fetch_url(url: str) -> bytes: }, ) with urllib.request.urlopen(request, timeout=20) as response: - return response.read() + payload = response.read(MAX_XML_BYTES + 1) + if len(payload) > MAX_XML_BYTES: + raise FeedXmlError("feed_xml_oversize") + return payload def strip_html(value: str) -> str: @@ -122,7 +132,14 @@ def stable_item_id(feed_id: str, link: str, title: str) -> str: def parse_feed_items(feed_bytes: bytes, feed: FeedConfig, *, max_items: int = 25) -> list[dict[str, str]]: - root = ET.fromstring(feed_bytes) + if type(feed_bytes) is not bytes: + raise FeedXmlError("feed_xml_invalid") + if len(feed_bytes) > MAX_XML_BYTES: + raise FeedXmlError("feed_xml_oversize") + try: + root = ET.fromstring(feed_bytes, forbid_dtd=True, forbid_entities=True, forbid_external=True) + except (DefusedXmlException, ET.ParseError, LookupError, UnicodeError, ValueError, RecursionError): + raise FeedXmlError("feed_xml_invalid") from None rows: list[dict[str, str]] = [] rss_items = root.findall("./channel/item") diff --git a/tests/test_rss_source_fetch.py b/tests/test_rss_source_fetch.py index a883395..d1f324e 100644 --- a/tests/test_rss_source_fetch.py +++ b/tests/test_rss_source_fetch.py @@ -2,9 +2,11 @@ import json from pathlib import Path +from unittest.mock import patch import pytest +from political_event_tracking_research import rss_source_fetch from political_event_tracking_research.rss_source_fetch import FeedConfig, fetch_rss_sources, parse_feed_items @@ -68,6 +70,70 @@ def test_parse_atom_feed_items_to_source_items() -> None: assert "EVT2" in rows[0]["text"] +@pytest.mark.parametrize( + "payload", + [ + b"]>", + b"]>", + b"", + b"", + ], +) +def test_defused_parser_rejects_entities_and_dtd(payload: bytes) -> None: + with pytest.raises(ValueError, match="feed_xml_"): + parse_feed_items(payload, FeedConfig("x", "https://example.test", "official", "")) + + +def test_utf16_entity_is_rejected_before_row_conversion() -> None: + payload = """ + ]> + &x;""".encode("utf-16") + with pytest.raises(ValueError, match="feed_xml_"): + parse_feed_items(payload, FeedConfig("x", "https://example.test", "official", "")) + + +def test_utf16_valid_feed_is_supported_without_fallback() -> None: + payload = """ + UTF16 + https://example.test/utf16Fri, 01 May 2026 12:30:00 GMT + """.encode("utf-16") + rows = parse_feed_items(payload, FeedConfig("x", "https://example.test", "official", "")) + assert rows[0]["source_url"] == "https://example.test/utf16" + + +def test_xml_payload_size_is_bounded_before_parse() -> None: + feed = FeedConfig("x", "https://example.test", "official", "") + with pytest.raises(ValueError, match="feed_xml_oversize"): + parse_feed_items(b"x" * (rss_source_fetch.MAX_XML_BYTES + 1), feed) + + +def test_network_read_is_bounded_and_oversize_is_sanitized() -> None: + class Response: + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def read(self, size: int) -> bytes: + assert size == rss_source_fetch.MAX_XML_BYTES + 1 + return b"x" * size + + with patch.object(rss_source_fetch.urllib.request, "urlopen", return_value=Response()): + with pytest.raises(ValueError, match="feed_xml_oversize"): + rss_source_fetch.fetch_url("https://example.test/feed") + + +def test_network_timeout_is_not_retried_or_parsed() -> None: + with patch.object(rss_source_fetch.urllib.request, "urlopen", side_effect=TimeoutError("timeout")): + with pytest.raises(TimeoutError, match="timeout"): + rss_source_fetch.fetch_url("https://example.test/feed") + + +def test_parser_has_no_stdlib_xml_fallback() -> None: + assert rss_source_fetch.ET.__name__ == "defusedxml.ElementTree" + + def test_fetch_rss_sources_can_continue_and_write_status(tmp_path: Path) -> None: feeds_path = tmp_path / "feeds.csv" diff --git a/uv.lock b/uv.lock index 4991d47..09d96ce 100644 --- a/uv.lock +++ b/uv.lock @@ -11,6 +11,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, ] +[[package]] +name = "defusedxml" +version = "0.7.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/0f/d5/c66da9b79e5bdb124974bfe172b4daf3c984ebd9c2a06e2b8a4dc7331c72/defusedxml-0.7.1.tar.gz", hash = "sha256:1bb3032db185915b62d7c6209c5a8792be6a32ab2fedacc84e01b52c51aa3e69", size = 75520, upload-time = "2021-03-08T10:59:26.269Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/07/6c/aa3f2f849e01cb6a001cd8554a88d4c77c5c1a31c95bdf1cf9301e6d9ef4/defusedxml-0.7.1-py2.py3-none-any.whl", hash = "sha256:a352e7e428770286cc899e2542b6cdaedb2b4953ff269a210103ec58f6198a61", size = 25604, upload-time = "2021-03-08T10:59:24.45Z" }, +] + [[package]] name = "iniconfig" version = "2.3.0" @@ -42,6 +51,9 @@ wheels = [ name = "political-event-tracking-research" version = "0.1.0" source = { editable = "." } +dependencies = [ + { name = "defusedxml" }, +] [package.optional-dependencies] test = [ @@ -49,7 +61,10 @@ test = [ ] [package.metadata] -requires-dist = [{ name = "pytest", marker = "extra == 'test'", specifier = ">=8" }] +requires-dist = [ + { name = "defusedxml", specifier = "==0.7.1" }, + { name = "pytest", marker = "extra == 'test'", specifier = ">=8" }, +] provides-extras = ["test"] [[package]]