H0: harden upstream RSS XML parsing

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]

src/political_event_tracking_research/rss_source_fetch.py

@@ -8,18 +8,21 @@
 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
 
 
 USER_AGENT = (
     "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")

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"<!DOCTYPE rss [<!ENTITY x SYSTEM 'file:///etc/passwd'>]><rss version='2.0'><channel/></rss>",
+        b"<?xml version='1.0'?><!DOCTYPE rss [<!ENTITY laugh 'x'>]><rss version='2.0'><channel/></rss>",
+        b"<?xml version='1.0' encoding='UTF-16'?><!DOCTYPE rss><rss version='2.0'><channel/></rss>",
+        b"<?xml version='1.0' encoding='x-unknown'?><rss version='2.0'><channel/></rss>",
+    ],
+)
+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 = """<?xml version='1.0' encoding='UTF-16'?>
+    <!DOCTYPE rss [<!ENTITY x SYSTEM 'file:///etc/passwd'>]>
+    <rss version='2.0'><channel><item><title>&x;</title></item></channel></rss>""".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 = """<?xml version='1.0' encoding='UTF-16'?>
+    <rss version='2.0'><channel><item><title>UTF16</title>
+    <link>https://example.test/utf16</link><pubDate>Fri, 01 May 2026 12:30:00 GMT</pubDate>
+    </item></channel></rss>""".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"