Fix CVE-2017-11610 by disabling object traversal in XML-RPC dispatch

Author: mnaberez

supervisor/tests/test_xmlrpc.py

@@ -269,28 +269,89 @@ def test_continue_request_500_if_xmlrpc_dumps_raises(self):
         self.assertEqual(request._error, 500)
 
 class TraverseTests(unittest.TestCase):
-    def test_underscore(self):
+    def test_security_disallows_underscore_methods(self):
         from supervisor import xmlrpc
-        self.assertRaises(xmlrpc.RPCError, xmlrpc.traverse, None, '_', None)
+        class Root:
+            pass
+        class A:
+            def _danger(self):
+                return True
+        root = Root()
+        root.a = A()
+        self.assertRaises(xmlrpc.RPCError, xmlrpc.traverse,
+            root, 'a._danger', [])
+
+    def test_security_disallows_object_traversal(self):
+        from supervisor import xmlrpc
+        class Root:
+            pass
+        class A:
+            pass
+        class B:
+            def danger(self):
+                return True
+        root = Root()
+        root.a = A()
+        root.a.b = B()
+        self.assertRaises(xmlrpc.RPCError, xmlrpc.traverse,
+            root, 'a.b.danger', [])
+
+    def test_namespace_name_not_found(self):
+        from supervisor import xmlrpc
+        class Root:
+            pass
+        root = Root()
+        self.assertRaises(xmlrpc.RPCError, xmlrpc.traverse,
+            root, 'notfound.hello', None)
 
-    def test_notfound(self):
+    def test_method_name_not_found(self):
         from supervisor import xmlrpc
-        self.assertRaises(xmlrpc.RPCError, xmlrpc.traverse, None, 'foo', None)
+        class Root:
+            pass
+        class A:
+            pass
+        root = Root()
+        root.a = A()
+        self.assertRaises(xmlrpc.RPCError, xmlrpc.traverse,
+            root, 'a.notfound', [])
 
-    def test_badparams(self):
+    def test_method_name_exists_but_is_not_a_method(self):
         from supervisor import xmlrpc
-        self.assertRaises(xmlrpc.RPCError, xmlrpc.traverse, self,
-                          'test_badparams', (1, 2, 3))
+        class Root:
+            pass
+        class A:
+            pass
+        class B:
+            pass
+        root = Root()
+        root.a = A()
+        root.a.b = B()
+        self.assertRaises(xmlrpc.RPCError, xmlrpc.traverse,
+            root, 'a.b', [])  # b is not a method
+
+    def test_bad_params(self):
+        from supervisor import xmlrpc
+        class Root:
+            pass
+        class A:
+            def hello(self, name):
+                return "Hello %s" % name
+        root = Root()
+        root.a = A()
+        self.assertRaises(xmlrpc.RPCError, xmlrpc.traverse,
+            root, 'a.hello', ["there", "extra"])  # too many params
 
     def test_success(self):
         from supervisor import xmlrpc
-        L = []
-        class Dummy:
-            def foo(self, a):
-                L.append(a)
-        dummy = Dummy()
-        xmlrpc.traverse(dummy, 'foo', [1])
-        self.assertEqual(L, [1])
+        class Root:
+            pass
+        class A:
+            def hello(self, name):
+                return "Hello %s" % name
+        root = Root()
+        root.a = A()
+        result = xmlrpc.traverse(root, 'a.hello', ["there"])
+        self.assertEqual(result, "Hello there")
 
 class SupervisorTransportTests(unittest.TestCase):
     def _getTargetClass(self):

supervisor/xmlrpc.py

@@ -428,18 +428,27 @@ def call(self, method, params):
         return traverse(self.rpcinterface, method, params)
 
 def traverse(ob, method, params):
-    path = method.split('.')
-    for name in path:
-        if name.startswith('_'):
-            # security (don't allow things that start with an underscore to
-            # be called remotely)
-            raise RPCError(Faults.UNKNOWN_METHOD)
-        ob = getattr(ob, name, None)
-        if ob is None:
-            raise RPCError(Faults.UNKNOWN_METHOD)
+    dotted_parts = method.split('.')
+    # security (CVE-2017-11610, don't allow object traversal)
+    if len(dotted_parts) != 2:
+        raise RPCError(Faults.UNKNOWN_METHOD)
+    namespace, method = dotted_parts
+
+    # security (don't allow methods that start with an underscore to
+    # be called remotely)
+    if method.startswith('_'):
+        raise RPCError(Faults.UNKNOWN_METHOD)
+
+    rpcinterface = getattr(ob, namespace, None)
+    if rpcinterface is None:
+        raise RPCError(Faults.UNKNOWN_METHOD)
+
+    func = getattr(rpcinterface, method, None)
+    if not isinstance(func, types.MethodType):
+        raise RPCError(Faults.UNKNOWN_METHOD)
 
     try:
-        return ob(*params)
+        return func(*params)
     except TypeError:
         raise RPCError(Faults.INCORRECT_PARAMETERS)