Merge b6ebfd8 into 2789348

CHANGES.rst

@@ -11,7 +11,13 @@ https://github.com/zopefoundation/Zope/blob/4.0a6/CHANGES.rst
 4.0.1 (unreleased)
 ------------------
 
-- Nothing changed yet.
+Features
+++++++++
+
+- Optionally control the use of Zope's built-in XML-RPC support for
+  POST requests with Content-Type ``text/xml`` via the
+  registration of a ``ZPublisher.interfaces.IXmlrpcChecker` utility
+  (`#620 <https://github.com/zopefoundation/Zope/issues/620>`_).
 
 
 4.0 (2019-05-10)

src/ZPublisher/HTTPRequest.py

@@ -32,6 +32,7 @@
 
 from AccessControl.tainted import should_be_tainted
 from AccessControl.tainted import taint_string
+from zope.component import queryUtility
 from zope.i18n.interfaces import IUserPreferredLanguages
 from zope.i18n.locales import LoadLocaleError
 from zope.i18n.locales import locales
@@ -44,6 +45,7 @@
 from ZPublisher.BaseRequest import BaseRequest
 from ZPublisher.BaseRequest import quote
 from ZPublisher.Converters import get_converter
+from ZPublisher.interfaces import IXmlrpcChecker
 from ZPublisher.utils import basic_auth_decode
 
 
@@ -523,7 +525,8 @@ def processInputs(
                 # Stash XML request for interpretation by a SOAP-aware view
                 other['SOAPXML'] = fs.value
             elif (method == 'POST'
-                  and 'text/xml' in fs.headers.get('content-type', '')):
+                  and 'text/xml' in fs.headers.get('content-type', '')
+                  and use_builtin_xmlrpc(self)):
                 # Ye haaa, XML-RPC!
                 meth, self.args = xmlrpc.parse_input(fs.value)
                 response = xmlrpc.response(response)
@@ -1837,3 +1840,8 @@ def _decode(value, charset):
     elif isinstance(value, binary_type):
         return text_type(value, charset, 'replace')
     return value
+
+
+def use_builtin_xmlrpc(request):
+    checker = queryUtility(IXmlrpcChecker)
+    return checker is None or checker(request)

src/ZPublisher/interfaces.py

@@ -77,3 +77,24 @@ class UseTraversalDefault(Exception):
     indicate that it has no special casing for the given name and that standard
     traversal logic should be applied.
     """
+
+
+###############################################################################
+# XML-RPC control
+
+class IXmlrpcChecker(Interface):
+    """Utility interface to control Zope's built-in XML-RPC support."""
+    def __call__(request):
+        """return true, when Zope's internal XML-RPC support should be used.
+
+        Only called for a non-SOAP POST request whose `Content-Type`
+        contains `text/xml` (any other request automatically does not
+        use Zope's buildin XML-RPC).
+
+        Note: this is called very early during request handling when most
+        typical attributes of *request* are not yet set up -- e.g. it
+        cannot rely on information in `form` or `other`.
+        Usually, it will look up information in `request.environ`
+        which at this time is garanteed (only) to contain the
+        typical CGI information, such as `PATH_INFO` and `QUERY_STRING`.
+        """

src/ZPublisher/tests/testHTTPRequest.py

@@ -13,21 +13,25 @@
 
 import sys
 import unittest
+from contextlib import contextmanager
 from io import BytesIO
 
 from six import PY2
 
 from AccessControl.tainted import should_be_tainted
 from zExceptions import NotFound
+from zope.component import getGlobalSiteManager
 from zope.component import provideAdapter
 from zope.i18n.interfaces import IUserPreferredLanguages
 from zope.i18n.interfaces.locales import ILocale
 from zope.publisher.browser import BrowserLanguages
 from zope.publisher.interfaces.http import IHTTPRequest
 from zope.testing.cleanup import cleanUp
 from ZPublisher.HTTPRequest import search_type
+from ZPublisher.interfaces import IXmlrpcChecker
 from ZPublisher.tests.testBaseRequest import TestRequestViewsBase
 from ZPublisher.utils import basic_auth_encode
+from ZPublisher.xmlrpc import is_xmlrpc_response
 
 
 if sys.version_info >= (3, ):
@@ -1213,6 +1217,48 @@ def test_text__password_field(self):
         self.assertNotIn('secret', req.text())
         self.assertIn('password obscured', req.text())
 
+    _xmlrpc_call = b"""<?xml version="1.0"?>
+    <methodCall>
+      <methodName>examples.getStateName</methodName>
+      <params>
+         <param>
+            <value><i4>41</i4></value>
+            </param>
+         </params>
+      </methodCall>
+    """
+
+    def test_processInputs_xmlrpc_with_args(self):
+        req = self._makeOne(
+            stdin=BytesIO(self._xmlrpc_call),
+            environ=dict(REQUEST_METHOD="POST", CONTENT_TYPE="text/xml"))
+        req.processInputs()
+        self.assertTrue(is_xmlrpc_response(req.response))
+        self.assertEqual(req.args, (41,))
+        self.assertEqual(req.other["PATH_INFO"], "/examples/getStateName")
+
+    def test_processInputs_xmlrpc_controlled_allowed(self):
+        req = self._makeOne(
+            stdin=BytesIO(self._xmlrpc_call),
+            environ=dict(REQUEST_METHOD="POST", CONTENT_TYPE="text/xml"))
+        with self._xmlrpc_control(lambda request: True):
+            req.processInputs()
+        self.assertTrue(is_xmlrpc_response(req.response))
+
+    def test_processInputs_xmlrpc_controlled_disallowed(self):
+        req = self._makeOne(
+            environ=dict(REQUEST_METHOD="POST", CONTENT_TYPE="text/xml"))
+        with self._xmlrpc_control(lambda request: False):
+            req.processInputs()
+        self.assertFalse(is_xmlrpc_response(req.response))
+
+    @contextmanager
+    def _xmlrpc_control(self, allow):
+        gsm = getGlobalSiteManager()
+        gsm.registerUtility(allow, IXmlrpcChecker)
+        yield
+        gsm.unregisterUtility(allow, IXmlrpcChecker)
+
 
 class TestHTTPRequestZope3Views(TestRequestViewsBase):