:exc:`ExpatError` to be raised with the :attr:`code` attribute set to
``errors.codes[errors.XML_ERROR_CANT_CHANGE_FEATURE_ONCE_PARSING]``.
+.. method:: xmlparser.SetReparseDeferralEnabled(enabled)
+
+ .. warning::
+
+ Calling ``SetReparseDeferralEnabled(False)`` has security implications,
+ as detailed below; please make sure to understand these consequences
+ prior to using the ``SetReparseDeferralEnabled`` method.
+
+ Expat 2.6.0 introduced a security mechanism called "reparse deferral"
+ where instead of causing denial of service through quadratic runtime
+ from reparsing large tokens, reparsing of unfinished tokens is now delayed
+ by default until a sufficient amount of input is reached.
+ Due to this delay, registered handlers may — depending of the sizing of
+ input chunks pushed to Expat — no longer be called right after pushing new
+ input to the parser. Where immediate feedback and taking over responsiblity
+ of protecting against denial of service from large tokens are both wanted,
+ calling ``SetReparseDeferralEnabled(False)`` disables reparse deferral
+ for the current Expat parser instance, temporarily or altogether.
+ Calling ``SetReparseDeferralEnabled(True)`` allows re-enabling reparse
+ deferral.
+
+ Note that :meth:`SetReparseDeferralEnabled` has been backported to some
+ prior releases of CPython as a security fix. Check for availability of
+ :meth:`SetReparseDeferralEnabled` using :func:`hasattr` if used in code
+ running across a variety of Python versions.
+
+ .. versionadded:: 3.11.9
+
+.. method:: xmlparser.GetReparseDeferralEnabled()
+
+ Returns whether reparse deferral is currently enabled for the given
+ Expat parser instance.
+
+ .. versionadded:: 3.11.9
+
+
:class:`xmlparser` objects have the following attributes:
at :func:`iterparse`. It can be useful when you're reading a large XML document
and don't want to hold it wholly in memory.
+Where *immediate* feedback through events is wanted, calling method
+:meth:`XMLPullParser.flush` can help reduce delay;
+please make sure to study the related security notes.
+
+
Finding interesting elements
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Feeds data to the parser. *data* is encoded data.
+
+ .. method:: flush()
+
+ Triggers parsing of any previously fed unparsed data, which can be
+ used to ensure more immediate feedback, in particular with Expat >=2.6.0.
+ The implementation of :meth:`flush` temporarily disables reparse deferral
+ with Expat (if currently enabled) and triggers a reparse.
+ Disabling reparse deferral has security consequences; please see
+ :meth:`xml.parsers.expat.xmlparser.SetReparseDeferralEnabled` for details.
+
+ Note that :meth:`flush` has been backported to some prior releases of
+ CPython as a security fix. Check for availability of :meth:`flush`
+ using :func:`hasattr` if used in code running across a variety of Python
+ versions.
+
+ .. versionadded:: 3.11.9
+
+
:meth:`XMLParser.feed` calls *target*\'s ``start(tag, attrs_dict)`` method
for each opening tag, its ``end(tag)`` method for each closing tag, and data
is processed by method ``data(data)``. For further supported callback
Feed the given bytes data to the parser.
+ .. method:: flush()
+
+ Triggers parsing of any previously fed unparsed data, which can be
+ used to ensure more immediate feedback, in particular with Expat >=2.6.0.
+ The implementation of :meth:`flush` temporarily disables reparse deferral
+ with Expat (if currently enabled) and triggers a reparse.
+ Disabling reparse deferral has security consequences; please see
+ :meth:`xml.parsers.expat.xmlparser.SetReparseDeferralEnabled` for details.
+
+ Note that :meth:`flush` has been backported to some prior releases of
+ CPython as a security fix. Check for availability of :meth:`flush`
+ using :func:`hasattr` if used in code running across a variety of Python
+ versions.
+
+ .. versionadded:: 3.11.9
+
.. method:: close()
Signal the parser that the data stream is terminated. Unlike
enum XML_Status (*SetEncoding)(XML_Parser parser, const XML_Char *encoding);
int (*DefaultUnknownEncodingHandler)(
void *encodingHandlerData, const XML_Char *name, XML_Encoding *info);
- /* might be none for expat < 2.1.0 */
+ /* might be NULL for expat < 2.1.0 */
int (*SetHashSalt)(XML_Parser parser, unsigned long hash_salt);
+ /* might be NULL for expat < 2.6.0 */
+ XML_Bool (*SetReparseDeferralEnabled)(XML_Parser parser, XML_Bool enabled);
/* always add new stuff to the end! */
};
self.assertEqual(handler_call_args, [("bar", "baz")])
+class ReparseDeferralTest(unittest.TestCase):
+ def test_getter_setter_round_trip(self):
+ parser = expat.ParserCreate()
+ enabled = (expat.version_info >= (2, 6, 0))
+
+ self.assertIs(parser.GetReparseDeferralEnabled(), enabled)
+ parser.SetReparseDeferralEnabled(False)
+ self.assertIs(parser.GetReparseDeferralEnabled(), False)
+ parser.SetReparseDeferralEnabled(True)
+ self.assertIs(parser.GetReparseDeferralEnabled(), enabled)
+
+ def test_reparse_deferral_enabled(self):
+ if expat.version_info < (2, 6, 0):
+ self.skipTest(f'Expat {expat.version_info} does not '
+ 'support reparse deferral')
+
+ started = []
+
+ def start_element(name, _):
+ started.append(name)
+
+ parser = expat.ParserCreate()
+ parser.StartElementHandler = start_element
+ self.assertTrue(parser.GetReparseDeferralEnabled())
+
+ for chunk in (b'<doc', b'/>'):
+ parser.Parse(chunk, False)
+
+ # The key test: Have handlers already fired? Expecting: no.
+ self.assertEqual(started, [])
+
+ parser.Parse(b'', True)
+
+ self.assertEqual(started, ['doc'])
+
+ def test_reparse_deferral_disabled(self):
+ started = []
+
+ def start_element(name, _):
+ started.append(name)
+
+ parser = expat.ParserCreate()
+ parser.StartElementHandler = start_element
+ if expat.version_info >= (2, 6, 0):
+ parser.SetReparseDeferralEnabled(False)
+ self.assertFalse(parser.GetReparseDeferralEnabled())
+
+ for chunk in (b'<doc', b'/>'):
+ parser.Parse(chunk, False)
+
+ # The key test: Have handlers already fired? Expecting: yes.
+ self.assertEqual(started, ['doc'])
+
+
if __name__ == "__main__":
unittest.main()
from io import BytesIO, StringIO
import codecs
import os.path
+import pyexpat
import shutil
import sys
from urllib.error import URLError
self.assertEqual(result.getvalue(), start + b"<doc>text</doc>")
+ def test_flush_reparse_deferral_enabled(self):
+ if pyexpat.version_info < (2, 6, 0):
+ self.skipTest(f'Expat {pyexpat.version_info} does not support reparse deferral')
+
+ result = BytesIO()
+ xmlgen = XMLGenerator(result)
+ parser = create_parser()
+ parser.setContentHandler(xmlgen)
+
+ for chunk in ("<doc", ">"):
+ parser.feed(chunk)
+
+ self.assertEqual(result.getvalue(), start) # i.e. no elements started
+ self.assertTrue(parser._parser.GetReparseDeferralEnabled())
+
+ parser.flush()
+
+ self.assertTrue(parser._parser.GetReparseDeferralEnabled())
+ self.assertEqual(result.getvalue(), start + b"<doc>")
+
+ parser.feed("</doc>")
+ parser.close()
+
+ self.assertEqual(result.getvalue(), start + b"<doc></doc>")
+
+ def test_flush_reparse_deferral_disabled(self):
+ result = BytesIO()
+ xmlgen = XMLGenerator(result)
+ parser = create_parser()
+ parser.setContentHandler(xmlgen)
+
+ for chunk in ("<doc", ">"):
+ parser.feed(chunk)
+
+ if pyexpat.version_info >= (2, 6, 0):
+ parser._parser.SetReparseDeferralEnabled(False)
+
+ self.assertEqual(result.getvalue(), start) # i.e. no elements started
+ self.assertFalse(parser._parser.GetReparseDeferralEnabled())
+
+ parser.flush()
+
+ self.assertFalse(parser._parser.GetReparseDeferralEnabled())
+ self.assertEqual(result.getvalue(), start + b"<doc>")
+
+ parser.feed("</doc>")
+ parser.close()
+
+ self.assertEqual(result.getvalue(), start + b"<doc></doc>")
+
# ===== Locator support
def test_expat_locator_noinfo(self):
</foo>
"""
-fails_with_expat_2_6_0 = (unittest.expectedFailure
- if pyexpat.version_info >= (2, 6, 0) else
- lambda test: test)
-
def checkwarnings(*filters, quiet=False):
def decorator(test):
def newtest(*args, **kwargs):
class XMLPullParserTest(unittest.TestCase):
- def _feed(self, parser, data, chunk_size=None):
+ def _feed(self, parser, data, chunk_size=None, flush=False):
if chunk_size is None:
parser.feed(data)
else:
for i in range(0, len(data), chunk_size):
parser.feed(data[i:i+chunk_size])
+ if flush:
+ parser.flush()
def assert_events(self, parser, expected, max_events=None):
self.assertEqual(
self.assertEqual([(action, elem.tag) for action, elem in events],
expected)
- def test_simple_xml(self, chunk_size=None):
+ def test_simple_xml(self, chunk_size=None, flush=False):
parser = ET.XMLPullParser()
self.assert_event_tags(parser, [])
- self._feed(parser, "<!-- comment -->\n", chunk_size)
+ self._feed(parser, "<!-- comment -->\n", chunk_size, flush)
self.assert_event_tags(parser, [])
self._feed(parser,
"<root>\n <element key='value'>text</element",
- chunk_size)
+ chunk_size, flush)
self.assert_event_tags(parser, [])
- self._feed(parser, ">\n", chunk_size)
+ self._feed(parser, ">\n", chunk_size, flush)
self.assert_event_tags(parser, [('end', 'element')])
- self._feed(parser, "<element>text</element>tail\n", chunk_size)
- self._feed(parser, "<empty-element/>\n", chunk_size)
+ self._feed(parser, "<element>text</element>tail\n", chunk_size, flush)
+ self._feed(parser, "<empty-element/>\n", chunk_size, flush)
self.assert_event_tags(parser, [
('end', 'element'),
('end', 'empty-element'),
])
- self._feed(parser, "</root>\n", chunk_size)
+ self._feed(parser, "</root>\n", chunk_size, flush)
self.assert_event_tags(parser, [('end', 'root')])
self.assertIsNone(parser.close())
- @fails_with_expat_2_6_0
def test_simple_xml_chunk_1(self):
- self.test_simple_xml(chunk_size=1)
+ self.test_simple_xml(chunk_size=1, flush=True)
- @fails_with_expat_2_6_0
def test_simple_xml_chunk_5(self):
- self.test_simple_xml(chunk_size=5)
+ self.test_simple_xml(chunk_size=5, flush=True)
def test_simple_xml_chunk_22(self):
self.test_simple_xml(chunk_size=22)
with self.assertRaises(ValueError):
ET.XMLPullParser(events=('start', 'end', 'bogus'))
+ def test_flush_reparse_deferral_enabled(self):
+ if pyexpat.version_info < (2, 6, 0):
+ self.skipTest(f'Expat {pyexpat.version_info} does not '
+ 'support reparse deferral')
+
+ parser = ET.XMLPullParser(events=('start', 'end'))
+
+ for chunk in ("<doc", ">"):
+ parser.feed(chunk)
+
+ self.assert_event_tags(parser, []) # i.e. no elements started
+ if ET is pyET:
+ self.assertTrue(parser._parser._parser.GetReparseDeferralEnabled())
+
+ parser.flush()
+
+ self.assert_event_tags(parser, [('start', 'doc')])
+ if ET is pyET:
+ self.assertTrue(parser._parser._parser.GetReparseDeferralEnabled())
+
+ parser.feed("</doc>")
+ parser.close()
+
+ self.assert_event_tags(parser, [('end', 'doc')])
+
+ def test_flush_reparse_deferral_disabled(self):
+ parser = ET.XMLPullParser(events=('start', 'end'))
+
+ for chunk in ("<doc", ">"):
+ parser.feed(chunk)
+
+ if pyexpat.version_info >= (2, 6, 0):
+ if not ET is pyET:
+ self.skipTest(f'XMLParser.(Get|Set)ReparseDeferralEnabled '
+ 'methods not available in C')
+ parser._parser._parser.SetReparseDeferralEnabled(False)
+
+ self.assert_event_tags(parser, []) # i.e. no elements started
+ if ET is pyET:
+ self.assertFalse(parser._parser._parser.GetReparseDeferralEnabled())
+
+ parser.flush()
+
+ self.assert_event_tags(parser, [('start', 'doc')])
+ if ET is pyET:
+ self.assertFalse(parser._parser._parser.GetReparseDeferralEnabled())
+
+ parser.feed("</doc>")
+ parser.close()
+
+ self.assert_event_tags(parser, [('end', 'doc')])
#
# xinclude tests (samples from appendix C of the xinclude specification)
else:
yield event
+ def flush(self):
+ if self._parser is None:
+ raise ValueError("flush() called after end of stream")
+ self._parser.flush()
+
def XML(text, parser=None):
"""Parse XML document from string constant.
del self.parser, self._parser
del self.target, self._target
+ def flush(self):
+ was_enabled = self.parser.GetReparseDeferralEnabled()
+ try:
+ self.parser.SetReparseDeferralEnabled(False)
+ self.parser.Parse(b"", False)
+ except self._error as v:
+ self._raiseerror(v)
+ finally:
+ self.parser.SetReparseDeferralEnabled(was_enabled)
# --------------------------------------------------------------------
# C14N 2.0
# FIXME: when to invoke error()?
self._err_handler.fatalError(exc)
+ def flush(self):
+ if self._parser is None:
+ return
+
+ was_enabled = self._parser.GetReparseDeferralEnabled()
+ try:
+ self._parser.SetReparseDeferralEnabled(False)
+ self._parser.Parse(b"", False)
+ except expat.error as e:
+ exc = SAXParseException(expat.ErrorString(e.code), e, self)
+ self._err_handler.fatalError(exc)
+ finally:
+ self._parser.SetReparseDeferralEnabled(was_enabled)
+
def _close_source(self):
source = self._source
try:
--- /dev/null
+Allow controlling Expat >=2.6.0 reparse deferral (CVE-2023-52425) by adding
+five new methods:
+
+* :meth:`xml.etree.ElementTree.XMLParser.flush`
+* :meth:`xml.etree.ElementTree.XMLPullParser.flush`
+* :meth:`xml.parsers.expat.xmlparser.GetReparseDeferralEnabled`
+* :meth:`xml.parsers.expat.xmlparser.SetReparseDeferralEnabled`
+* :meth:`xml.sax.expatreader.ExpatParser.flush`
}
}
+/*[clinic input]
+_elementtree.XMLParser.flush
+
+[clinic start generated code]*/
+
+static PyObject *
+_elementtree_XMLParser_flush_impl(XMLParserObject *self)
+/*[clinic end generated code: output=42fdb8795ca24509 input=effbecdb28715949]*/
+{
+ if (!_check_xmlparser(self)) {
+ return NULL;
+ }
+
+ if (EXPAT(SetReparseDeferralEnabled) == NULL) {
+ Py_RETURN_NONE;
+ }
+
+ // NOTE: The Expat parser in the C implementation of ElementTree is not
+ // exposed to the outside; as a result we known that reparse deferral
+ // is currently enabled, or we would not even have access to function
+ // XML_SetReparseDeferralEnabled in the first place (which we checked
+ // for, a few lines up).
+
+ EXPAT(SetReparseDeferralEnabled)(self->parser, XML_FALSE);
+
+ PyObject *res = expat_parse(self, "", 0, XML_FALSE);
+
+ EXPAT(SetReparseDeferralEnabled)(self->parser, XML_TRUE);
+
+ return res;
+}
+
/*[clinic input]
_elementtree.XMLParser.feed
static PyMethodDef xmlparser_methods[] = {
_ELEMENTTREE_XMLPARSER_FEED_METHODDEF
_ELEMENTTREE_XMLPARSER_CLOSE_METHODDEF
+ _ELEMENTTREE_XMLPARSER_FLUSH_METHODDEF
_ELEMENTTREE_XMLPARSER__PARSE_WHOLE_METHODDEF
_ELEMENTTREE_XMLPARSER__SETEVENTS_METHODDEF
{NULL, NULL}
return _elementtree_XMLParser_close_impl(self);
}
+PyDoc_STRVAR(_elementtree_XMLParser_flush__doc__,
+"flush($self, /)\n"
+"--\n"
+"\n");
+
+#define _ELEMENTTREE_XMLPARSER_FLUSH_METHODDEF \
+ {"flush", (PyCFunction)_elementtree_XMLParser_flush, METH_NOARGS, _elementtree_XMLParser_flush__doc__},
+
+static PyObject *
+_elementtree_XMLParser_flush_impl(XMLParserObject *self);
+
+static PyObject *
+_elementtree_XMLParser_flush(XMLParserObject *self, PyObject *Py_UNUSED(ignored))
+{
+ return _elementtree_XMLParser_flush_impl(self);
+}
+
PyDoc_STRVAR(_elementtree_XMLParser_feed__doc__,
"feed($self, data, /)\n"
"--\n"
exit:
return return_value;
}
-/*[clinic end generated code: output=3fd6fa2ce1aeca76 input=a9049054013a1b77]*/
+/*[clinic end generated code: output=88a3c8b7164a6474 input=a9049054013a1b77]*/
preserve
[clinic start generated code]*/
+PyDoc_STRVAR(pyexpat_xmlparser_SetReparseDeferralEnabled__doc__,
+"SetReparseDeferralEnabled($self, enabled, /)\n"
+"--\n"
+"\n"
+"Enable/Disable reparse deferral; enabled by default with Expat >=2.6.0.");
+
+#define PYEXPAT_XMLPARSER_SETREPARSEDEFERRALENABLED_METHODDEF \
+ {"SetReparseDeferralEnabled", (PyCFunction)pyexpat_xmlparser_SetReparseDeferralEnabled, METH_O, pyexpat_xmlparser_SetReparseDeferralEnabled__doc__},
+
+static PyObject *
+pyexpat_xmlparser_SetReparseDeferralEnabled_impl(xmlparseobject *self,
+ int enabled);
+
+static PyObject *
+pyexpat_xmlparser_SetReparseDeferralEnabled(xmlparseobject *self, PyObject *arg)
+{
+ PyObject *return_value = NULL;
+ int enabled;
+
+ enabled = PyObject_IsTrue(arg);
+ if (enabled < 0) {
+ goto exit;
+ }
+ return_value = pyexpat_xmlparser_SetReparseDeferralEnabled_impl(self, enabled);
+
+exit:
+ return return_value;
+}
+
+PyDoc_STRVAR(pyexpat_xmlparser_GetReparseDeferralEnabled__doc__,
+"GetReparseDeferralEnabled($self, /)\n"
+"--\n"
+"\n"
+"Retrieve reparse deferral enabled status; always returns false with Expat <2.6.0.");
+
+#define PYEXPAT_XMLPARSER_GETREPARSEDEFERRALENABLED_METHODDEF \
+ {"GetReparseDeferralEnabled", (PyCFunction)pyexpat_xmlparser_GetReparseDeferralEnabled, METH_NOARGS, pyexpat_xmlparser_GetReparseDeferralEnabled__doc__},
+
+static PyObject *
+pyexpat_xmlparser_GetReparseDeferralEnabled_impl(xmlparseobject *self);
+
+static PyObject *
+pyexpat_xmlparser_GetReparseDeferralEnabled(xmlparseobject *self, PyObject *Py_UNUSED(ignored))
+{
+ return pyexpat_xmlparser_GetReparseDeferralEnabled_impl(self);
+}
+
PyDoc_STRVAR(pyexpat_xmlparser_Parse__doc__,
"Parse($self, data, isfinal=False, /)\n"
"--\n"
#ifndef PYEXPAT_XMLPARSER_USEFOREIGNDTD_METHODDEF
#define PYEXPAT_XMLPARSER_USEFOREIGNDTD_METHODDEF
#endif /* !defined(PYEXPAT_XMLPARSER_USEFOREIGNDTD_METHODDEF) */
-/*[clinic end generated code: output=3e333b89da3aa58c input=a9049054013a1b77]*/
+/*[clinic end generated code: output=8d544b917e35add6 input=a9049054013a1b77]*/
#define XML_SetNotStandaloneHandler PyExpat_XML_SetNotStandaloneHandler
#define XML_SetParamEntityParsing PyExpat_XML_SetParamEntityParsing
#define XML_SetProcessingInstructionHandler PyExpat_XML_SetProcessingInstructionHandler
+#define XML_SetReparseDeferralEnabled PyExpat_XML_SetReparseDeferralEnabled
#define XML_SetReturnNSTriplet PyExpat_XML_SetReturnNSTriplet
#define XML_SetSkippedEntityHandler PyExpat_XML_SetSkippedEntityHandler
#define XML_SetStartCdataSectionHandler PyExpat_XML_SetStartCdataSectionHandler
#include "Python.h"
#include <ctype.h>
+#include <stdbool.h>
#include "structmember.h" // PyMemberDef
#include "expat.h"
/* NULL if not enabled */
int buffer_size; /* Size of buffer, in XML_Char units */
int buffer_used; /* Buffer units in use */
+ bool reparse_deferral_enabled; /* Whether to defer reparsing of
+ unfinished XML tokens; a de-facto cache of
+ what Expat has the authority on, for lack
+ of a getter API function
+ "XML_GetReparseDeferralEnabled" in Expat
+ 2.6.0 */
PyObject *intern; /* Dictionary to intern strings */
PyObject **handlers;
} xmlparseobject;
#define MAX_CHUNK_SIZE (1 << 20)
+/*[clinic input]
+pyexpat.xmlparser.SetReparseDeferralEnabled
+
+ enabled: bool
+ /
+
+Enable/Disable reparse deferral; enabled by default with Expat >=2.6.0.
+[clinic start generated code]*/
+
+static PyObject *
+pyexpat_xmlparser_SetReparseDeferralEnabled_impl(xmlparseobject *self,
+ int enabled)
+/*[clinic end generated code: output=5ec539e3b63c8c49 input=021eb9e0bafc32c5]*/
+{
+#if XML_COMBINED_VERSION >= 20600
+ XML_SetReparseDeferralEnabled(self->itself, enabled ? XML_TRUE : XML_FALSE);
+ self->reparse_deferral_enabled = (bool)enabled;
+#endif
+ Py_RETURN_NONE;
+}
+
+/*[clinic input]
+pyexpat.xmlparser.GetReparseDeferralEnabled
+
+Retrieve reparse deferral enabled status; always returns false with Expat <2.6.0.
+[clinic start generated code]*/
+
+static PyObject *
+pyexpat_xmlparser_GetReparseDeferralEnabled_impl(xmlparseobject *self)
+/*[clinic end generated code: output=4e91312e88a595a8 input=54b5f11d32b20f3e]*/
+{
+ return PyBool_FromLong(self->reparse_deferral_enabled);
+}
+
/*[clinic input]
pyexpat.xmlparser.Parse
#if XML_COMBINED_VERSION >= 19505
PYEXPAT_XMLPARSER_USEFOREIGNDTD_METHODDEF
#endif
+ PYEXPAT_XMLPARSER_SETREPARSEDEFERRALENABLED_METHODDEF
+ PYEXPAT_XMLPARSER_GETREPARSEDEFERRALENABLED_METHODDEF
{NULL, NULL} /* sentinel */
};
self->handlers = NULL;
self->intern = intern;
Py_XINCREF(self->intern);
+#if XML_COMBINED_VERSION >= 20600
+ self->reparse_deferral_enabled = true;
+#else
+ self->reparse_deferral_enabled = false;
+#endif
/* namespace_separator is either NULL or contains one char + \0 */
self->itself = XML_ParserCreate_MM(encoding, &ExpatMemoryHandler,
#else
capi.SetHashSalt = NULL;
#endif
+#if XML_COMBINED_VERSION >= 20600
+ capi.SetReparseDeferralEnabled = XML_SetReparseDeferralEnabled;
+#else
+ capi.SetReparseDeferralEnabled = NULL;
+#endif
/* export using capsule */
PyObject *capi_object = PyCapsule_New(&capi, PyExpat_CAPSULE_NAME, NULL);
projects / thirdparty / Python / cpython.git / commitdiff
