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]]