openstack
/
nova
Code Issues Proposed changes

Makes safe xml data calls raise 400 http error instead of 500 

When we parse incoming XML safely, if there was an error raised it would
be an expat.Expat() error, which would bubble up to the api and turn
into a HTTP 500 (Internal Error)

It turns out that all the places we use the safe_xml parsing are in
Deserializers, close to the API, so in this patch we just change the
error it raises straight to nova.exception.MalformedRequest().

This causes the api to fail with the proper 400 (Malformed Request) when
it encounters corrupt XML. This is caught at
nova.api.openstack.wsgi._process_stack and __call__.

We also take the opportunity to move the new safe parser from nova.utils
to nova.api.openstack.xmlutil as the openstack api is the only thing
that uses it.

Fixes: bug #1133111
Change-Id: Ifa2ed7ee128241cfe8dbcdc5bd75194d96b6cdb5
This commit is contained in:
Matthew Sherborne 2013-02-26 10:35:42 +10:00 committed by Gerrit Code Review
parent 2830ef14ec
commit 3478f1e121
19 changed files with 233 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
nova/api/openstack/common.py
View File

@ -31,7 +31,6 @@ from nova.compute import vm_states
from nova import exception
from nova.openstack.common import log as logging
from nova import quota
from nova import utils
osapi_opts = [
cfg.IntOpt('osapi_max_limit',
@ -356,7 +355,7 @@ def raise_http_conflict_for_instance_invalid_state(exc, action):
class MetadataDeserializer(wsgi.MetadataXMLDeserializer):
def deserialize(self, text):
dom = utils.safe_minidom_parse_string(text)
dom = xmlutil.safe_minidom_parse_string(text)
metadata_node = self.find_first_child_named(dom, "metadata")
metadata = self.extract_metadata(metadata_node)
return {'body': {'metadata': metadata}}
@ -364,7 +363,7 @@ class MetadataDeserializer(wsgi.MetadataXMLDeserializer):
class MetaItemDeserializer(wsgi.MetadataXMLDeserializer):
def deserialize(self, text):
dom = utils.safe_minidom_parse_string(text)
dom = xmlutil.safe_minidom_parse_string(text)
metadata_item = self.extract_metadata(dom)
return {'body': {'meta': metadata_item}}
@ -382,7 +381,7 @@ class MetadataXMLDeserializer(wsgi.XMLDeserializer):
return metadata
def _extract_metadata_container(self, datastring):
dom = utils.safe_minidom_parse_string(datastring)
dom = xmlutil.safe_minidom_parse_string(datastring)
metadata_node = self.find_first_child_named(dom, "metadata")
metadata = self.extract_metadata(metadata_node)
return {'body': {'metadata': metadata}}
@ -394,7 +393,7 @@ class MetadataXMLDeserializer(wsgi.XMLDeserializer):
return self._extract_metadata_container(datastring)
def update(self, datastring):
dom = utils.safe_minidom_parse_string(datastring)
dom = xmlutil.safe_minidom_parse_string(datastring)
metadata_item = self.extract_metadata(dom)
return {'body': {'meta': metadata_item}}

8
nova/api/openstack/compute/contrib/cells.py
View File

@ -19,7 +19,6 @@
from oslo.config import cfg
from webob import exc
from xml.parsers import expat
from nova.api.openstack import common
from nova.api.openstack import extensions
@ -31,7 +30,6 @@ from nova import db
from nova import exception
from nova.openstack.common import log as logging
from nova.openstack.common import timeutils
from nova import utils
LOG = logging.getLogger(__name__)
@ -98,11 +96,7 @@ class CellDeserializer(wsgi.XMLDeserializer):
def default(self, string):
"""Deserialize an xml-formatted cell create request."""
try:
node = utils.safe_minidom_parse_string(string)
except expat.ExpatError:
msg = _("cannot understand XML")
raise exception.MalformedRequestBody(reason=msg)
node = xmlutil.safe_minidom_parse_string(string)
return {'body': {'cell': self._extract_cell(node)}}

8
nova/api/openstack/compute/contrib/hosts.py
View File

@ -16,7 +16,6 @@
"""The hosts admin extension."""
import webob.exc
from xml.parsers import expat
from nova.api.openstack import extensions
from nova.api.openstack import wsgi
@ -24,7 +23,6 @@ from nova.api.openstack import xmlutil
from nova import compute
from nova import exception
from nova.openstack.common import log as logging
from nova import utils
LOG = logging.getLogger(__name__)
authorize = extensions.extension_authorizer('compute', 'hosts')
@ -71,11 +69,7 @@ class HostShowTemplate(xmlutil.TemplateBuilder):
class HostUpdateDeserializer(wsgi.XMLDeserializer):
def default(self, string):
try:
node = utils.safe_minidom_parse_string(string)
except expat.ExpatError:
msg = _("cannot understand XML")
raise exception.MalformedRequestBody(reason=msg)
node = xmlutil.safe_minidom_parse_string(string)
updates = {}
updates_node = self.find_first_child_named(node, 'updates')

5
nova/api/openstack/compute/contrib/security_groups.py
View File

@ -32,7 +32,6 @@ from nova import exception
from nova.network.security_group import openstack_driver
from nova.network.security_group import quantum_driver
from nova.openstack.common import log as logging
from nova import utils
from nova.virt import netutils
@ -113,7 +112,7 @@ class SecurityGroupXMLDeserializer(wsgi.MetadataXMLDeserializer):
"""
def default(self, string):
"""Deserialize an xml-formatted security group create request."""
dom = utils.safe_minidom_parse_string(string)
dom = xmlutil.safe_minidom_parse_string(string)
security_group = {}
sg_node = self.find_first_child_named(dom,
'security_group')
@ -134,7 +133,7 @@ class SecurityGroupRulesXMLDeserializer(wsgi.MetadataXMLDeserializer):
def default(self, string):
"""Deserialize an xml-formatted security group create request."""
dom = utils.safe_minidom_parse_string(string)
dom = xmlutil.safe_minidom_parse_string(string)
security_group_rule = self._extract_security_group_rule(dom)
return {'body': {'security_group_rule': security_group_rule}}

2
nova/api/openstack/compute/contrib/volumes.py
View File

@ -154,7 +154,7 @@ class CreateDeserializer(CommonDeserializer):
def default(self, string):
"""Deserialize an xml-formatted volume create request."""
dom = utils.safe_minidom_parse_string(string)
dom = xmlutil.safe_minidom_parse_string(string)
vol = self._extract_volume(dom)
return {'body': {'volume': vol}}

4
nova/api/openstack/compute/servers.py
View File

@ -317,7 +317,7 @@ class ActionDeserializer(CommonDeserializer):
"""
def default(self, string):
dom = utils.safe_minidom_parse_string(string)
dom = xmlutil.safe_minidom_parse_string(string)
action_node = dom.childNodes[0]
action_name = action_node.tagName
@ -424,7 +424,7 @@ class CreateDeserializer(CommonDeserializer):
def default(self, string):
"""Deserialize an xml-formatted server create request."""
dom = utils.safe_minidom_parse_string(string)
dom = xmlutil.safe_minidom_parse_string(string)
server = self._extract_server(dom)
return {'body': {'server': server}}

27
nova/api/openstack/wsgi.py
View File

@ -19,15 +19,14 @@ import inspect
import math
import time
from xml.dom import minidom
from xml.parsers import expat
from lxml import etree
import webob
from nova.api.openstack import xmlutil
from nova import exception
from nova.openstack.common import jsonutils
from nova.openstack.common import log as logging
from nova import utils
from nova import wsgi
@ -216,13 +215,8 @@ class XMLDeserializer(TextDeserializer):
def _from_xml(self, datastring):
plurals = set(self.metadata.get('plurals', {}))
try:
node = utils.safe_minidom_parse_string(datastring).childNodes[0]
return {node.nodeName: self._from_xml_node(node, plurals)}
except expat.ExpatError:
msg = _("cannot understand XML")
raise exception.MalformedRequestBody(reason=msg)
node = xmlutil.safe_minidom_parse_string(datastring).childNodes[0]
return {node.nodeName: self._from_xml_node(node, plurals)}
def _from_xml_node(self, node, listnames):
"""Convert a minidom node to a simple Python type.
@ -634,7 +628,7 @@ def action_peek_json(body):
def action_peek_xml(body):
"""Determine action to invoke."""
dom = utils.safe_minidom_parse_string(body)
dom = xmlutil.safe_minidom_parse_string(body)
action_node = dom.childNodes[0]
return action_node.tagName
@ -890,17 +884,8 @@ class Resource(wsgi.Application):
# function. If we try to audit __call__(), we can
# run into troubles due to the @webob.dec.wsgify()
# decorator.
try:
return self._process_stack(request, action, action_args,
content_type, body, accept)
except expat.ExpatError:
msg = _("Invalid XML in request body")
return Fault(webob.exc.HTTPBadRequest(explanation=msg))
except LookupError as e:
#NOTE(Vijaya Erukala): XML input such as
# <?xml version="1.0" encoding="TF-8"?>
# raises LookupError: unknown encoding: TF-8
return Fault(webob.exc.HTTPBadRequest(explanation=unicode(e)))
return self._process_stack(request, action, action_args,
content_type, body, accept)
def _process_stack(self, request, action, action_args,
content_type, body, accept):

61
nova/api/openstack/xmlutil.py
View File

@ -18,7 +18,12 @@
import os.path
from lxml import etree
from xml.dom import minidom
from xml.parsers import expat
from xml import sax
from xml.sax import expatreader
from nova import exception
from nova import utils
@ -905,3 +910,59 @@ def make_flat_dict(name, selector=None, subselector=None, ns=None):
# Return the template
return root
class ProtectedExpatParser(expatreader.ExpatParser):
"""An expat parser which disables DTD's and entities by default."""
def __init__(self, forbid_dtd=True, forbid_entities=True,
*args, **kwargs):
# Python 2.x old style class
expatreader.ExpatParser.__init__(self, *args, **kwargs)
self.forbid_dtd = forbid_dtd
self.forbid_entities = forbid_entities
def start_doctype_decl(self, name, sysid, pubid, has_internal_subset):
raise ValueError("Inline DTD forbidden")
def entity_decl(self, entityName, is_parameter_entity, value, base,
systemId, publicId, notationName):
raise ValueError("<!ENTITY> entity declaration forbidden")
def unparsed_entity_decl(self, name, base, sysid, pubid, notation_name):
# expat 1.2
raise ValueError("<!ENTITY> unparsed entity forbidden")
def external_entity_ref(self, context, base, systemId, publicId):
raise ValueError("<!ENTITY> external entity forbidden")
def notation_decl(self, name, base, sysid, pubid):
raise ValueError("<!ENTITY> notation forbidden")
def reset(self):
expatreader.ExpatParser.reset(self)
if self.forbid_dtd:
self._parser.StartDoctypeDeclHandler = self.start_doctype_decl
self._parser.EndDoctypeDeclHandler = None
if self.forbid_entities:
self._parser.EntityDeclHandler = self.entity_decl
self._parser.UnparsedEntityDeclHandler = self.unparsed_entity_decl
self._parser.ExternalEntityRefHandler = self.external_entity_ref
self._parser.NotationDeclHandler = self.notation_decl
try:
self._parser.SkippedEntityHandler = None
except AttributeError:
# some pyexpat versions do not support SkippedEntity
pass
def safe_minidom_parse_string(xml_string):
"""Parse an XML string using minidom safely."""
try:
return minidom.parseString(xml_string, parser=ProtectedExpatParser())
except (sax.SAXParseException, ValueError,
expat.ExpatError, LookupError) as e:
#NOTE(Vijaya Erukala): XML input such as
# <?xml version="1.0" encoding="TF-8"?>
# raises LookupError: unknown encoding: TF-8
raise exception.MalformedRequestBody(reason=str(e))

8
nova/tests/api/openstack/compute/contrib/test_cells.py
View File

@ -27,6 +27,7 @@ from nova import exception
from nova.openstack.common import timeutils
from nova import test
from nova.tests.api.openstack import fakes
from nova.tests import utils
FAKE_CELLS = [
@ -394,3 +395,10 @@ class TestCellsXMLDeserializer(test.TestCase):
deserializer = cells_ext.CellDeserializer()
result = deserializer.deserialize(intext)
self.assertEqual(dict(body=expected), result)
def test_with_corrupt_xml(self):
deserializer = cells_ext.CellDeserializer()
self.assertRaises(
exception.MalformedRequestBody,
deserializer.deserialize,
utils.killer_xml_body())

8
nova/tests/api/openstack/compute/contrib/test_hosts.py
View File

@ -21,9 +21,11 @@ from nova.compute import power_state
from nova.compute import vm_states
from nova import context as context_maker
from nova import db
from nova import exception
from nova.openstack.common import log as logging
from nova import test
from nova.tests import fake_hosts
from nova.tests import utils
LOG = logging.getLogger(__name__)
@ -390,3 +392,9 @@ class HostSerializerTest(test.TestCase):
result = self.deserializer.deserialize(intext)
self.assertEqual(dict(body=exemplar), result)
def test_corrupt_xml(self):
self.assertRaises(
exception.MalformedRequestBody,
self.deserializer.deserialize,
utils.killer_xml_body())

22
nova/tests/api/openstack/compute/contrib/test_security_groups.py
View File

@ -30,6 +30,7 @@ from nova.openstack.common import jsonutils
from nova import quota
from nova import test
from nova.tests.api.openstack import fakes
from nova.tests import utils
CONF = cfg.CONF
FAKE_UUID = 'a47ae74e-ab08-447f-8eee-ffd43fc46c16'
@ -727,13 +728,6 @@ class TestSecurityGroupRules(test.TestCase):
self.assertEquals(security_group_rule['to_port'], 81)
def test_create_by_invalid_cidr_json(self):
rules = {
"security_group_rule": {
"ip_protocol": "tcp",
"from_port": "22",
"to_port": "22",
"parent_group_id": self.sg2['id'],
"cidr": "10.2.3.124/2433"}}
rule = security_group_rule_template(
ip_protocol="tcp",
from_port=22,
@ -1146,6 +1140,13 @@ class TestSecurityGroupRulesXMLDeserializer(test.TestCase):
}
self.assertEquals(request['body'], expected)
def test_corrupt_xml(self):
"""Should throw a 400 error on corrupt xml."""
self.assertRaises(
exception.MalformedRequestBody,
self.deserializer.deserialize,
utils.killer_xml_body())
class TestSecurityGroupXMLDeserializer(test.TestCase):
@ -1192,6 +1193,13 @@ class TestSecurityGroupXMLDeserializer(test.TestCase):
}
self.assertEquals(request['body'], expected)
def test_corrupt_xml(self):
"""Should throw a 400 error on corrupt xml."""
self.assertRaises(
exception.MalformedRequestBody,
self.deserializer.deserialize,
utils.killer_xml_body())
class TestSecurityGroupXMLSerializer(test.TestCase):
def setUp(self):

8
nova/tests/api/openstack/compute/test_server_actions.py
View File

@ -32,6 +32,7 @@ from nova import test
from nova.tests.api.openstack import fakes
from nova.tests.image import fake
from nova.tests import matchers
from nova.tests import utils
CONF = cfg.CONF
CONF.import_opt('password_length', 'nova.utils')
@ -1146,3 +1147,10 @@ class TestServerActionXMLDeserializer(test.TestCase):
self.deserializer.deserialize,
serial_request,
'action')
def test_corrupt_xml(self):
"""Should throw a 400 error on corrupt xml."""
self.assertRaises(
exception.MalformedRequestBody,
self.deserializer.deserialize,
utils.killer_xml_body())

8
nova/tests/api/openstack/compute/test_servers.py
View File

@ -51,6 +51,7 @@ from nova.tests.api.openstack import fakes
from nova.tests import fake_network
from nova.tests.image import fake
from nova.tests import matchers
from nova.tests import utils
CONF = cfg.CONF
CONF.import_opt('password_length', 'nova.utils')
@ -3859,6 +3860,13 @@ class TestServerCreateRequestXMLDeserializer(test.TestCase):
}
self.assertEquals(request['body'], expected)
def test_corrupt_xml(self):
"""Should throw a 400 error on corrupt xml."""
self.assertRaises(
exception.MalformedRequestBody,
self.deserializer.deserialize,
utils.killer_xml_body())
class TestAddressesXMLSerialization(test.TestCase):

14
nova/tests/api/openstack/test_common.py
View File

@ -28,7 +28,7 @@ from nova.api.openstack import common
from nova.api.openstack import xmlutil
from nova import exception
from nova import test
from nova.tests import utils as test_utils
from nova.tests import utils
NS = "{http://docs.openstack.org/compute/api/v1.1}"
@ -297,7 +297,7 @@ class MiscFunctionsTest(test.TestCase):
self.fail("webob.exc.HTTPConflict was not raised")
def test_check_img_metadata_properties_quota_valid_metadata(self):
ctxt = test_utils.get_test_admin_context()
ctxt = utils.get_test_admin_context()
metadata1 = {"key": "value"}
actual = common.check_img_metadata_properties_quota(ctxt, metadata1)
self.assertEqual(actual, None)
@ -311,7 +311,7 @@ class MiscFunctionsTest(test.TestCase):
self.assertEqual(actual, None)
def test_check_img_metadata_properties_quota_inv_metadata(self):
ctxt = test_utils.get_test_admin_context()
ctxt = utils.get_test_admin_context()
metadata1 = {"a" * 260: "value"}
self.assertRaises(webob.exc.HTTPBadRequest,
common.check_img_metadata_properties_quota, ctxt, metadata1)
@ -512,3 +512,11 @@ class MetadataXMLSerializationTest(test.TestCase):
""".replace(" ", "").replace("\n", ""))
self.assertEqual(expected.toxml(), actual.toxml())
def test_metadata_deserializer(self):
"""Should throw a 400 error on corrupt xml."""
deserializer = common.MetadataXMLDeserializer()
self.assertRaises(
exception.MalformedRequestBody,
deserializer.deserialize,
utils.killer_xml_body())

16
nova/tests/api/openstack/test_wsgi.py
View File

@ -7,6 +7,7 @@ from nova.api.openstack import wsgi
from nova import exception
from nova import test
from nova.tests.api.openstack import fakes
from nova.tests import utils
class RequestTest(test.TestCase):
@ -272,6 +273,21 @@ class ResourceTest(test.TestCase):
'<fooAction>true</fooAction>')
self.assertEqual(controller._action_foo, method)
def test_get_method_action_corrupt_xml(self):
class Controller(wsgi.Controller):
@wsgi.action('fooAction')
def _action_foo(self, req, id, body):
return body
controller = Controller()
resource = wsgi.Resource(controller)
self.assertRaises(
exception.MalformedRequestBody,
resource.get_method,
None, 'action',
'application/xml',
utils.killer_xml_body())
def test_get_method_action_bad_body(self):
class Controller(wsgi.Controller):
@wsgi.action('fooAction')

64
nova/tests/api/openstack/test_xmlutil.py
View File

@ -16,9 +16,12 @@
# under the License.
from lxml import etree
from xml.dom import minidom
from nova.api.openstack import xmlutil
from nova import exception
from nova import test
from nova.tests import utils as tests_utils
class SelectorTest(test.TestCase):
@ -720,3 +723,64 @@ class MiscellaneousXMLUtilTests(test.TestCase):
tmpl = xmlutil.MasterTemplate(root, 1)
result = tmpl.serialize(dict(wrapper=dict(a='foo', b='bar')))
self.assertEqual(result, expected_xml)
def test_safe_parse_xml(self):
normal_body = ("""
<?xml version="1.0" ?><foo>
<bar>
<v1>hey</v1>
<v2>there</v2>
</bar>
</foo>""").strip()
dom = xmlutil.safe_minidom_parse_string(normal_body)
self.assertEqual(normal_body, str(dom.toxml()))
self.assertRaises(exception.MalformedRequestBody,
xmlutil.safe_minidom_parse_string,
tests_utils.killer_xml_body())
class SafeParserTestCase(test.TestCase):
def test_external_dtd(self):
xml_string = ("""<?xml version="1.0" encoding="utf-8"?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN"
"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html>
<head/>
<body>html with dtd</body>
</html>""")
parser = xmlutil.ProtectedExpatParser(forbid_dtd=False,
forbid_entities=True)
self.assertRaises(ValueError,
minidom.parseString,
xml_string, parser)
def test_external_file(self):
xml_string = """<!DOCTYPE external [
<!ENTITY ee SYSTEM "file:///PATH/TO/root.xml">
]>
<root>&ee;</root>"""
parser = xmlutil.ProtectedExpatParser(forbid_dtd=False,
forbid_entities=True)
self.assertRaises(ValueError,
minidom.parseString,
xml_string, parser)
def test_notation(self):
xml_string = """<?xml version="1.0" standalone="no"?>
<!-- comment data -->
<!DOCTYPE x [
<!NOTATION notation SYSTEM "notation.jpeg">
]>
<root attr1="value1">
</root>"""
parser = xmlutil.ProtectedExpatParser(forbid_dtd=False,
forbid_entities=True)
self.assertRaises(ValueError,
minidom.parseString,
xml_string, parser)

78
nova/tests/test_utils.py
View File

@ -23,7 +23,6 @@ import os
import os.path
import StringIO
import tempfile
from xml.dom import minidom
import mox
import netaddr
@ -450,39 +449,6 @@ class GenericUtilsTestCase(test.TestCase):
self.assertEqual(fake_execute.uid, 2)
self.assertEqual(fake_execute.uid, os.getuid())
def test_safe_parse_xml(self):
normal_body = ("""
<?xml version="1.0" ?><foo>
<bar>
<v1>hey</v1>
<v2>there</v2>
</bar>
</foo>""").strip()
def killer_body():
return (("""<!DOCTYPE x [
<!ENTITY a "%(a)s">
<!ENTITY b "%(b)s">
<!ENTITY c "%(c)s">]>
<foo>
<bar>
<v1>%(d)s</v1>
</bar>
</foo>""") % {
'a': 'A' * 10,
'b': '&a;' * 10,
'c': '&b;' * 10,
'd': '&c;' * 9999,
}).strip()
dom = utils.safe_minidom_parse_string(normal_body)
self.assertEqual(normal_body, str(dom.toxml()))
self.assertRaises(ValueError,
utils.safe_minidom_parse_string,
killer_body())
def test_xhtml_escape(self):
self.assertEqual('&quot;foo&quot;', utils.xhtml_escape('"foo"'))
self.assertEqual('&apos;foo&apos;', utils.xhtml_escape("'foo'"))
@ -992,47 +958,3 @@ class StringLengthTestCase(test.TestCase):
self.assertRaises(exception.InvalidInput,
utils.check_string_length,
'a' * 256, 'name', max_length=255)
class SafeParserTestCase(test.TestCase):
def test_external_dtd(self):
xml_string = ("""<?xml version="1.0" encoding="utf-8"?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN"
"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html>
<head/>
<body>html with dtd</body>
</html>""")
parser = utils.ProtectedExpatParser(forbid_dtd=False,
forbid_entities=True)
self.assertRaises(ValueError,
minidom.parseString,
xml_string, parser)
def test_external_file(self):
xml_string = """<!DOCTYPE external [
<!ENTITY ee SYSTEM "file:///PATH/TO/root.xml">
]>
<root>&ee;</root>"""
parser = utils.ProtectedExpatParser(forbid_dtd=False,
forbid_entities=True)
self.assertRaises(ValueError,
minidom.parseString,
xml_string, parser)
def test_notation(self):
xml_string = """<?xml version="1.0" standalone="no"?>
<!-- comment data -->
<!DOCTYPE x [
<!NOTATION notation SYSTEM "notation.jpeg">
]>
<root attr1="value1">
</root>"""
parser = utils.ProtectedExpatParser(forbid_dtd=False,
forbid_entities=True)
self.assertRaises(ValueError,
minidom.parseString,
xml_string, parser)

17
nova/tests/utils.py
View File

@ -183,3 +183,20 @@ def cleanup_dns_managers():
for manager in test_dns_managers:
manager.delete_dns_file()
test_dns_managers = []
def killer_xml_body():
return (("""<!DOCTYPE x [
<!ENTITY a "%(a)s">
<!ENTITY b "%(b)s">
<!ENTITY c "%(c)s">]>
<foo>
<bar>
<v1>%(d)s</v1>
</bar>
</foo>""") % {
'a': 'A' * 10,
'b': '&a;' * 10,
'c': '&b;' * 10,
'd': '&c;' * 9999,
}).strip()

58
nova/utils.py
View File

@ -36,10 +36,6 @@ import struct
import sys
import tempfile
import time
from xml.dom import minidom
from xml.parsers import expat
from xml import sax
from xml.sax import expatreader
from xml.sax import saxutils
from eventlet import event
@ -657,60 +653,6 @@ class DynamicLoopingCall(LoopingCallBase):
return self.done
class ProtectedExpatParser(expatreader.ExpatParser):
"""An expat parser which disables DTD's and entities by default."""
def __init__(self, forbid_dtd=True, forbid_entities=True,
*args, **kwargs):
# Python 2.x old style class
expatreader.ExpatParser.__init__(self, *args, **kwargs)
self.forbid_dtd = forbid_dtd
self.forbid_entities = forbid_entities
def start_doctype_decl(self, name, sysid, pubid, has_internal_subset):
raise ValueError("Inline DTD forbidden")
def entity_decl(self, entityName, is_parameter_entity, value, base,
systemId, publicId, notationName):
raise ValueError("<!ENTITY> entity declaration forbidden")
def unparsed_entity_decl(self, name, base, sysid, pubid, notation_name):
# expat 1.2
raise ValueError("<!ENTITY> unparsed entity forbidden")
def external_entity_ref(self, context, base, systemId, publicId):
raise ValueError("<!ENTITY> external entity forbidden")
def notation_decl(self, name, base, sysid, pubid):
raise ValueError("<!ENTITY> notation forbidden")
def reset(self):
expatreader.ExpatParser.reset(self)
if self.forbid_dtd:
self._parser.StartDoctypeDeclHandler = self.start_doctype_decl
self._parser.EndDoctypeDeclHandler = None
if self.forbid_entities:
self._parser.EntityDeclHandler = self.entity_decl
self._parser.UnparsedEntityDeclHandler = self.unparsed_entity_decl
self._parser.ExternalEntityRefHandler = self.external_entity_ref
self._parser.NotationDeclHandler = self.notation_decl
try:
self._parser.SkippedEntityHandler = None
except AttributeError:
# some pyexpat versions do not support SkippedEntity
pass
def safe_minidom_parse_string(xml_string):
"""Parse an XML string using minidom safely.
"""
try:
return minidom.parseString(xml_string, parser=ProtectedExpatParser())
except sax.SAXParseException as se:
raise expat.ExpatError()
def xhtml_escape(value):
"""Escapes a string so it is valid within XML or XHTML.