diff --git a/lib/crewai-tools/pyproject.toml b/lib/crewai-tools/pyproject.toml index bd2e638f5a..51f6563df3 100644 --- a/lib/crewai-tools/pyproject.toml +++ b/lib/crewai-tools/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "python-docx~=1.2.0", "youtube-transcript-api~=1.2.2", "pymupdf~=1.26.6", + "defusedxml>=0.7.1", ] diff --git a/lib/crewai-tools/src/crewai_tools/rag/loaders/xml_loader.py b/lib/crewai-tools/src/crewai_tools/rag/loaders/xml_loader.py index f7f9a6d00d..8f48ea30ff 100644 --- a/lib/crewai-tools/src/crewai_tools/rag/loaders/xml_loader.py +++ b/lib/crewai-tools/src/crewai_tools/rag/loaders/xml_loader.py @@ -1,5 +1,8 @@ from typing import Any -from xml.etree.ElementTree import ParseError, fromstring, parse +from xml.etree.ElementTree import ParseError + +from defusedxml.ElementTree import fromstring +from defusedxml.common import DefusedXmlException from crewai_tools.rag.base_loader import BaseLoader, LoaderResult from crewai_tools.rag.loaders.utils import load_from_url @@ -39,10 +42,7 @@ def _load_from_file(path: str) -> str: def _parse_xml(self, content: str, source_ref: str) -> LoaderResult: try: - if content.strip().startswith("<"): - root = fromstring(content) # noqa: S314 - else: - root = parse(source_ref).getroot() # noqa: S314 + root = fromstring(content) text_parts = [] for text_content in root.itertext(): @@ -51,6 +51,9 @@ def _parse_xml(self, content: str, source_ref: str) -> LoaderResult: text = "\n".join(text_parts) metadata = {"format": "xml", "root_tag": root.tag} + except DefusedXmlException as e: + text = "" + metadata = {"format": "xml", "security_error": str(e)} except ParseError as e: text = content metadata = {"format": "xml", "parse_error": str(e)} diff --git a/lib/crewai-tools/tests/rag/test_xml_loader_security.py b/lib/crewai-tools/tests/rag/test_xml_loader_security.py new file mode 100644 index 0000000000..1f85f55c53 --- /dev/null +++ b/lib/crewai-tools/tests/rag/test_xml_loader_security.py @@ -0,0 +1,120 @@ +"""Regression tests for XMLLoader against XXE and entity-expansion attacks.""" + +from unittest.mock import patch + +from crewai_tools.rag.loaders.xml_loader import XMLLoader +from crewai_tools.rag.source_content import SourceContent + + +class TestXMLLoaderSecurity: + """Ensure XMLLoader rejects XML documents that would trigger XXE or entity + expansion attacks when parsed with the standard library. + + The loader keeps its "return a LoaderResult with an error in metadata" + contract for malformed input, so malicious documents are contained at the + loader boundary rather than aborting the caller (e.g. RAG.add). + """ + + def test_rejects_external_entity_expansion(self): + """A document referencing an external entity must be refused. + + defusedxml refuses to process any declared entity by default, + including this external `SYSTEM` reference. XMLLoader captures + that refusal and surfaces `security_error` in metadata instead + of letting the exception escape into `RAG.add()`. + """ + malicious = ( + '' + '' + ']>' + '&xxe;' + ) + + loader = XMLLoader() + result = loader.load(SourceContent(malicious)) + + assert result.content == "" + assert "security_error" in result.metadata + assert result.metadata["format"] == "xml" + assert "root_tag" not in result.metadata + + def test_rejects_billion_laughs(self): + """Nested entity expansion (billion laughs) must be refused.""" + malicious = ( + '' + '' + '' + '' + ']>' + '&lol3;' + ) + + loader = XMLLoader() + result = loader.load(SourceContent(malicious)) + + assert result.content == "" + assert "security_error" in result.metadata + assert result.metadata["format"] == "xml" + + def test_parses_safe_xml(self): + """A benign XML document without a DOCTYPE must still parse.""" + safe = 'helloworld' + + loader = XMLLoader() + result = loader.load(SourceContent(safe)) + + assert "hello" in result.content + assert "world" in result.content + assert result.metadata["format"] == "xml" + assert result.metadata["root_tag"] == "root" + assert "security_error" not in result.metadata + + def test_malformed_xml_still_returns_parse_error(self): + """Non-security parse failures still land in `parse_error`, not + `security_error`, so existing consumers keep working. + """ + broken = "" + + loader = XMLLoader() + result = loader.load(SourceContent(broken)) + + assert "parse_error" in result.metadata + assert "security_error" not in result.metadata + + @patch("crewai_tools.rag.loaders.xml_loader.load_from_url") + def test_url_payload_with_bom_is_parsed_not_treated_as_path(self, mock_load): + """Content with a BOM or leading whitespace must still be parsed + as XML, not routed to a filesystem `parse(source_ref)` call that + would raise `FileNotFoundError` on a URL source_ref. + """ + payload = ( + "\ufeffbom-safe" + ) + mock_load.return_value = payload + + loader = XMLLoader() + result = loader.load(SourceContent("https://example.com/feed.xml")) + + assert "bom-safe" in result.content + assert result.metadata["root_tag"] == "root" + assert "parse_error" not in result.metadata + assert "security_error" not in result.metadata + + @patch("crewai_tools.rag.loaders.xml_loader.load_from_url") + def test_rejects_xxe_from_url_source(self, mock_load): + """XML fetched from a remote URL is still validated.""" + mock_load.return_value = ( + '' + '' + ']>' + '&xxe;' + ) + + loader = XMLLoader() + result = loader.load(SourceContent("https://example.com/feed.xml")) + + assert result.content == "" + assert "security_error" in result.metadata