tests.py 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544
  1. from __future__ import print_function
  2. import os
  3. import sys
  4. import unittest
  5. import io
  6. import re
  7. from xml.sax.saxutils import XMLGenerator
  8. from xml.sax import SAXParseException
  9. from pyexpat import ExpatError
  10. from defusedxml import cElementTree, ElementTree, minidom, pulldom, sax, xmlrpc
  11. from defusedxml import defuse_stdlib
  12. from defusedxml import (DefusedXmlException, DTDForbidden, EntitiesForbidden,
  13. ExternalReferenceForbidden, NotSupportedError)
  14. from defusedxml.common import PY3, PY26, PY31
  15. try:
  16. import gzip
  17. except ImportError:
  18. gzip = None
  19. try:
  20. from defusedxml import lxml
  21. from lxml.etree import XMLSyntaxError
  22. LXML3 = lxml.LXML3
  23. except ImportError:
  24. lxml = None
  25. XMLSyntaxError = None
  26. LXML3 = False
  27. HERE = os.path.dirname(os.path.abspath(__file__))
  28. # prevent web access
  29. # based on Debian's rules, Port 9 is discard
  30. os.environ["http_proxy"] = "http://127.0.9.1:9"
  31. os.environ["https_proxy"] = os.environ["http_proxy"]
  32. os.environ["ftp_proxy"] = os.environ["http_proxy"]
  33. if PY26 or PY31:
  34. class _AssertRaisesContext(object):
  35. def __init__(self, expected, test_case, expected_regexp=None):
  36. self.expected = expected
  37. self.failureException = test_case.failureException
  38. self.expected_regexp = expected_regexp
  39. def __enter__(self):
  40. return self
  41. def __exit__(self, exc_type, exc_value, tb):
  42. if exc_type is None:
  43. try:
  44. exc_name = self.expected.__name__
  45. except AttributeError:
  46. exc_name = str(self.expected)
  47. raise self.failureException(
  48. "{0} not raised".format(exc_name))
  49. if not issubclass(exc_type, self.expected):
  50. # let unexpected exceptions pass through
  51. return False
  52. self.exception = exc_value # store for later retrieval
  53. if self.expected_regexp is None:
  54. return True
  55. expected_regexp = self.expected_regexp
  56. if isinstance(expected_regexp, basestring):
  57. expected_regexp = re.compile(expected_regexp)
  58. if not expected_regexp.search(str(exc_value)):
  59. raise self.failureException('"%s" does not match "%s"' %
  60. (expected_regexp.pattern, str(exc_value)))
  61. return True
  62. class DefusedTestCase(unittest.TestCase):
  63. if PY3:
  64. content_binary = False
  65. else:
  66. content_binary = True
  67. xml_dtd = os.path.join(HERE, "xmltestdata", "dtd.xml")
  68. xml_external = os.path.join(HERE, "xmltestdata", "external.xml")
  69. xml_external_file = os.path.join(HERE, "xmltestdata", "external_file.xml")
  70. xml_quadratic = os.path.join(HERE, "xmltestdata", "quadratic.xml")
  71. xml_simple = os.path.join(HERE, "xmltestdata", "simple.xml")
  72. xml_simple_ns = os.path.join(HERE, "xmltestdata", "simple-ns.xml")
  73. xml_bomb = os.path.join(HERE, "xmltestdata", "xmlbomb.xml")
  74. xml_bomb2 = os.path.join(HERE, "xmltestdata", "xmlbomb2.xml")
  75. xml_cyclic = os.path.join(HERE, "xmltestdata", "cyclic.xml")
  76. if PY26 or PY31:
  77. # old Python versions don't have these useful test methods
  78. def assertRaises(self, excClass, callableObj=None, *args, **kwargs):
  79. context = _AssertRaisesContext(excClass, self)
  80. if callableObj is None:
  81. return context
  82. with context:
  83. callableObj(*args, **kwargs)
  84. def assertIn(self, member, container, msg=None):
  85. if member not in container:
  86. standardMsg = '%s not found in %s' % (repr(member),
  87. repr(container))
  88. self.fail(self._formatMessage(msg, standardMsg))
  89. def get_content(self, xmlfile):
  90. mode = "rb" if self.content_binary else "r"
  91. with io.open(xmlfile, mode) as f:
  92. data = f.read()
  93. return data
  94. class BaseTests(DefusedTestCase):
  95. module = None
  96. dtd_external_ref = False
  97. external_ref_exception = ExternalReferenceForbidden
  98. cyclic_error = None
  99. iterparse = None
  100. def test_simple_parse(self):
  101. self.parse(self.xml_simple)
  102. self.parseString(self.get_content(self.xml_simple))
  103. if self.iterparse:
  104. self.iterparse(self.xml_simple)
  105. def test_simple_parse_ns(self):
  106. self.parse(self.xml_simple_ns)
  107. self.parseString(self.get_content(self.xml_simple_ns))
  108. if self.iterparse:
  109. self.iterparse(self.xml_simple_ns)
  110. def test_entities_forbidden(self):
  111. self.assertRaises(EntitiesForbidden, self.parse, self.xml_bomb)
  112. self.assertRaises(EntitiesForbidden, self.parse, self.xml_quadratic)
  113. self.assertRaises(EntitiesForbidden, self.parse, self.xml_external)
  114. self.assertRaises(EntitiesForbidden, self.parseString,
  115. self.get_content(self.xml_bomb))
  116. self.assertRaises(EntitiesForbidden, self.parseString,
  117. self.get_content(self.xml_quadratic))
  118. self.assertRaises(EntitiesForbidden, self.parseString,
  119. self.get_content(self.xml_external))
  120. if self.iterparse:
  121. self.assertRaises(EntitiesForbidden, self.iterparse,
  122. self.xml_bomb)
  123. self.assertRaises(EntitiesForbidden, self.iterparse,
  124. self.xml_quadratic)
  125. self.assertRaises(EntitiesForbidden, self.iterparse,
  126. self.xml_external)
  127. def test_entity_cycle(self):
  128. self.assertRaises(self.cyclic_error, self.parse, self.xml_cyclic,
  129. forbid_entities=False)
  130. def test_dtd_forbidden(self):
  131. self.assertRaises(DTDForbidden, self.parse, self.xml_bomb,
  132. forbid_dtd=True)
  133. self.assertRaises(DTDForbidden, self.parse, self.xml_quadratic,
  134. forbid_dtd=True)
  135. self.assertRaises(DTDForbidden, self.parse, self.xml_external,
  136. forbid_dtd=True)
  137. self.assertRaises(DTDForbidden, self.parse, self.xml_dtd,
  138. forbid_dtd=True)
  139. self.assertRaises(DTDForbidden, self.parseString,
  140. self.get_content(self.xml_bomb),
  141. forbid_dtd=True)
  142. self.assertRaises(DTDForbidden, self.parseString,
  143. self.get_content(self.xml_quadratic),
  144. forbid_dtd=True)
  145. self.assertRaises(DTDForbidden, self.parseString,
  146. self.get_content(self.xml_external),
  147. forbid_dtd=True)
  148. self.assertRaises(DTDForbidden, self.parseString,
  149. self.get_content(self.xml_dtd),
  150. forbid_dtd=True)
  151. if self.iterparse:
  152. self.assertRaises(DTDForbidden, self.iterparse,
  153. self.xml_bomb, forbid_dtd=True)
  154. self.assertRaises(DTDForbidden, self.iterparse,
  155. self.xml_quadratic, forbid_dtd=True)
  156. self.assertRaises(DTDForbidden, self.iterparse,
  157. self.xml_external, forbid_dtd=True)
  158. self.assertRaises(DTDForbidden, self.iterparse,
  159. self.xml_dtd, forbid_dtd=True)
  160. def test_dtd_with_external_ref(self):
  161. if self.dtd_external_ref:
  162. self.assertRaises(self.external_ref_exception, self.parse,
  163. self.xml_dtd)
  164. else:
  165. self.parse(self.xml_dtd)
  166. def test_external_ref(self):
  167. self.assertRaises(self.external_ref_exception, self.parse,
  168. self.xml_external, forbid_entities=False)
  169. def test_external_file_ref(self):
  170. content = self.get_content(self.xml_external_file)
  171. if isinstance(content, bytes):
  172. here = HERE.encode(sys.getfilesystemencoding())
  173. content = content.replace(b"/PATH/TO", here)
  174. else:
  175. content = content.replace("/PATH/TO", HERE)
  176. self.assertRaises(self.external_ref_exception, self.parseString,
  177. content, forbid_entities=False)
  178. def test_allow_expansion(self):
  179. self.parse(self.xml_bomb2, forbid_entities=False)
  180. self.parseString(self.get_content(self.xml_bomb2),
  181. forbid_entities=False)
  182. class TestDefusedElementTree(BaseTests):
  183. module = ElementTree
  184. ## etree doesn't do external ref lookup
  185. #external_ref_exception = ElementTree.ParseError
  186. cyclic_error = ElementTree.ParseError
  187. def parse(self, xmlfile, **kwargs):
  188. tree = self.module.parse(xmlfile, **kwargs)
  189. return self.module.tostring(tree.getroot())
  190. def parseString(self, xmlstring, **kwargs):
  191. tree = self.module.fromstring(xmlstring, **kwargs)
  192. return self.module.tostring(tree)
  193. def iterparse(self, source, **kwargs):
  194. return list(self.module.iterparse(source, **kwargs))
  195. class TestDefusedcElementTree(TestDefusedElementTree):
  196. module = cElementTree
  197. class TestDefusedMinidom(BaseTests):
  198. module = minidom
  199. cyclic_error = ExpatError
  200. iterparse = None
  201. def parse(self, xmlfile, **kwargs):
  202. doc = self.module.parse(xmlfile, **kwargs)
  203. return doc.toxml()
  204. def parseString(self, xmlstring, **kwargs):
  205. doc = self.module.parseString(xmlstring, **kwargs)
  206. return doc.toxml()
  207. class TestDefusedPulldom(BaseTests):
  208. module = pulldom
  209. cyclic_error = SAXParseException
  210. dtd_external_ref = True
  211. def parse(self, xmlfile, **kwargs):
  212. events = self.module.parse(xmlfile, **kwargs)
  213. return list(events)
  214. def parseString(self, xmlstring, **kwargs):
  215. events = self.module.parseString(xmlstring, **kwargs)
  216. return list(events)
  217. class TestDefusedSax(BaseTests):
  218. module = sax
  219. cyclic_error = SAXParseException
  220. content_binary = True
  221. dtd_external_ref = True
  222. def parse(self, xmlfile, **kwargs):
  223. if PY3:
  224. result = io.StringIO()
  225. else:
  226. result = io.BytesIO()
  227. handler = XMLGenerator(result)
  228. self.module.parse(xmlfile, handler, **kwargs)
  229. return result.getvalue()
  230. def parseString(self, xmlstring, **kwargs):
  231. if PY3:
  232. result = io.StringIO()
  233. else:
  234. result = io.BytesIO()
  235. handler = XMLGenerator(result)
  236. self.module.parseString(xmlstring, handler, **kwargs)
  237. return result.getvalue()
  238. def test_exceptions(self):
  239. if PY26:
  240. # Python 2.6 unittest doesn't support with self.assertRaises()
  241. return
  242. with self.assertRaises(EntitiesForbidden) as ctx:
  243. self.parse(self.xml_bomb)
  244. msg = "EntitiesForbidden(name='a', system_id=None, public_id=None)"
  245. self.assertEqual(str(ctx.exception), msg)
  246. self.assertEqual(repr(ctx.exception), msg)
  247. with self.assertRaises(ExternalReferenceForbidden) as ctx:
  248. self.parse(self.xml_external, forbid_entities=False)
  249. msg = ("ExternalReferenceForbidden"
  250. "(system_id='http://www.w3schools.com/xml/note.xml', public_id=None)")
  251. self.assertEqual(str(ctx.exception), msg)
  252. self.assertEqual(repr(ctx.exception), msg)
  253. with self.assertRaises(DTDForbidden) as ctx:
  254. self.parse(self.xml_bomb, forbid_dtd=True)
  255. msg = "DTDForbidden(name='xmlbomb', system_id=None, public_id=None)"
  256. self.assertEqual(str(ctx.exception), msg)
  257. self.assertEqual(repr(ctx.exception), msg)
  258. class TestDefusedLxml(BaseTests):
  259. module = lxml
  260. cyclic_error = XMLSyntaxError
  261. content_binary = True
  262. def parse(self, xmlfile, **kwargs):
  263. tree = self.module.parse(xmlfile, **kwargs)
  264. return self.module.tostring(tree)
  265. def parseString(self, xmlstring, **kwargs):
  266. tree = self.module.fromstring(xmlstring, **kwargs)
  267. return self.module.tostring(tree)
  268. if not LXML3:
  269. def test_entities_forbidden(self):
  270. self.assertRaises(NotSupportedError, self.parse, self.xml_bomb)
  271. def test_dtd_with_external_ref(self):
  272. self.assertRaises(NotSupportedError, self.parse, self.xml_dtd)
  273. def test_external_ref(self):
  274. pass
  275. def test_external_file_ref(self):
  276. pass
  277. def test_restricted_element1(self):
  278. tree = self.module.parse(self.xml_bomb, forbid_dtd=False,
  279. forbid_entities=False)
  280. root = tree.getroot()
  281. self.assertEqual(root.text, None)
  282. self.assertEqual(list(root), [])
  283. self.assertEqual(root.getchildren(), [])
  284. self.assertEqual(list(root.iter()), [root])
  285. self.assertEqual(list(root.iterchildren()), [])
  286. self.assertEqual(list(root.iterdescendants()), [])
  287. self.assertEqual(list(root.itersiblings()), [])
  288. self.assertEqual(list(root.getiterator()), [root])
  289. self.assertEqual(root.getnext(), None)
  290. def test_restricted_element2(self):
  291. tree = self.module.parse(self.xml_bomb2, forbid_dtd=False,
  292. forbid_entities=False)
  293. root = tree.getroot()
  294. bomb, tag = root
  295. self.assertEqual(root.text, "text")
  296. self.assertEqual(list(root), [bomb, tag])
  297. self.assertEqual(root.getchildren(), [bomb, tag])
  298. self.assertEqual(list(root.iter()), [root, bomb, tag])
  299. self.assertEqual(list(root.iterchildren()), [bomb, tag])
  300. self.assertEqual(list(root.iterdescendants()), [bomb, tag])
  301. self.assertEqual(list(root.itersiblings()), [])
  302. self.assertEqual(list(root.getiterator()), [root, bomb, tag])
  303. self.assertEqual(root.getnext(), None)
  304. self.assertEqual(root.getprevious(), None)
  305. self.assertEqual(list(bomb.itersiblings()), [tag])
  306. self.assertEqual(bomb.getnext(), tag)
  307. self.assertEqual(bomb.getprevious(), None)
  308. self.assertEqual(tag.getnext(), None)
  309. self.assertEqual(tag.getprevious(), bomb)
  310. def test_xpath_injection(self):
  311. # show XPath injection vulnerability
  312. xml = """<root><tag id="one" /><tag id="two"/></root>"""
  313. expr = "one' or @id='two"
  314. root = lxml.fromstring(xml)
  315. # insecure way
  316. xp = "tag[@id='%s']" % expr
  317. elements = root.xpath(xp)
  318. self.assertEqual(len(elements), 2)
  319. self.assertEqual(elements, list(root))
  320. # proper and safe way
  321. xp = "tag[@id=$idname]"
  322. elements = root.xpath(xp, idname=expr)
  323. self.assertEqual(len(elements), 0)
  324. self.assertEqual(elements, [])
  325. elements = root.xpath(xp, idname="one")
  326. self.assertEqual(len(elements), 1)
  327. self.assertEqual(elements, list(root)[:1])
  328. class XmlRpcTarget(object):
  329. def __init__(self):
  330. self._data = []
  331. def __str__(self):
  332. return "".join(self._data)
  333. def xml(self, encoding, standalone):
  334. pass
  335. def start(self, tag, attrs):
  336. self._data.append("<%s>" % tag)
  337. def data(self, text):
  338. self._data.append(text)
  339. def end(self, tag):
  340. self._data.append("</%s>" % tag)
  341. class TestXmlRpc(DefusedTestCase):
  342. module = xmlrpc
  343. def parse(self, xmlfile, **kwargs):
  344. target = XmlRpcTarget()
  345. parser = self.module.DefusedExpatParser(target, **kwargs)
  346. data = self.get_content(xmlfile)
  347. parser.feed(data)
  348. parser.close()
  349. return target
  350. def parse_unpatched(self, xmlfile):
  351. target = XmlRpcTarget()
  352. parser = self.module.ExpatParser(target)
  353. data = self.get_content(xmlfile)
  354. parser.feed(data)
  355. parser.close()
  356. return target
  357. def test_xmlrpc(self):
  358. self.assertRaises(EntitiesForbidden, self.parse, self.xml_bomb)
  359. self.assertRaises(EntitiesForbidden, self.parse, self.xml_quadratic)
  360. self.parse(self.xml_dtd)
  361. self.assertRaises(DTDForbidden, self.parse, self.xml_dtd,
  362. forbid_dtd=True)
  363. #def test_xmlrpc_unpatched(self):
  364. # for fname in (self.xml_external, self.xml_dtd):
  365. # print(self.parse_unpatched(fname))
  366. def test_monkeypatch(self):
  367. try:
  368. xmlrpc.monkey_patch()
  369. finally:
  370. xmlrpc.unmonkey_patch()
  371. class TestDefusedGzip(DefusedTestCase):
  372. def get_gzipped(self, length):
  373. f = io.BytesIO()
  374. gzf = gzip.GzipFile(mode="wb", fileobj=f)
  375. gzf.write(b"d" * length)
  376. gzf.close()
  377. f.seek(0)
  378. return f
  379. def decode_response(self, response, limit=None, readlength=1024):
  380. dec = xmlrpc.DefusedGzipDecodedResponse(response, limit)
  381. acc = []
  382. while True:
  383. data = dec.read(readlength)
  384. if not data:
  385. break
  386. acc.append(data)
  387. return b"".join(acc)
  388. def test_defused_gzip_decode(self):
  389. data = self.get_gzipped(4096).getvalue()
  390. result = xmlrpc.defused_gzip_decode(data)
  391. self.assertEqual(result, b"d" *4096)
  392. result = xmlrpc.defused_gzip_decode(data, -1)
  393. self.assertEqual(result, b"d" *4096)
  394. result = xmlrpc.defused_gzip_decode(data, 4096)
  395. self.assertEqual(result, b"d" *4096)
  396. with self.assertRaises(ValueError):
  397. result = xmlrpc.defused_gzip_decode(data, 4095)
  398. with self.assertRaises(ValueError):
  399. result = xmlrpc.defused_gzip_decode(data, 0)
  400. def test_defused_gzip_response(self):
  401. clen = len(self.get_gzipped(4096).getvalue())
  402. response = self.get_gzipped(4096)
  403. data = self.decode_response(response)
  404. self.assertEqual(data, b"d" *4096)
  405. with self.assertRaises(ValueError):
  406. response = self.get_gzipped(4096)
  407. xmlrpc.DefusedGzipDecodedResponse(response, clen - 1)
  408. with self.assertRaises(ValueError):
  409. response = self.get_gzipped(4096)
  410. self.decode_response(response, 4095)
  411. with self.assertRaises(ValueError):
  412. response = self.get_gzipped(4096)
  413. self.decode_response(response, 4095, 8192)
  414. def test_main():
  415. suite = unittest.TestSuite()
  416. suite.addTests(unittest.makeSuite(TestDefusedcElementTree))
  417. suite.addTests(unittest.makeSuite(TestDefusedElementTree))
  418. suite.addTests(unittest.makeSuite(TestDefusedMinidom))
  419. suite.addTests(unittest.makeSuite(TestDefusedPulldom))
  420. suite.addTests(unittest.makeSuite(TestDefusedSax))
  421. suite.addTests(unittest.makeSuite(TestXmlRpc))
  422. if lxml is not None:
  423. suite.addTests(unittest.makeSuite(TestDefusedLxml))
  424. if gzip is not None:
  425. suite.addTests(unittest.makeSuite(TestDefusedGzip))
  426. return suite
  427. if __name__ == "__main__":
  428. suite = test_main()
  429. result = unittest.TextTestRunner(verbosity=1).run(suite)
  430. # TODO: test that it actually works
  431. defuse_stdlib()
  432. sys.exit(not result.wasSuccessful())