CVE-2019-15149: 7PK - Security Features (CWE-254)
value: core.py in Mitogen before 0.2.8 has a typo that drops the unidirectional-routing protection mechanism in the case of a child that is initiated by another child. The Ansible extension is unaffected. NOTE: the vendor disputes this issue because it is exploitable only in conjunction with hypothetical other factors, i.e., an affected use case within a library caller, and a bug in the message receiver policy code that led to reliance on this extra protection mechanism
CWE description: Software security is not security software. Here we're concerned with topics like authentication, access control, confidentiality, cryptography, and privilege management.
CVSS v3 base score: 9.8
Published: 2019-08-18T20:15Z
Repository: https://github.com/dw/mitogen
Language: Python
Files changed: mitogen/core.py, tests/router_test.py
Commit date: 2019-08-08 23:03:43 +0100
Commit message: [security] core: undirectional routing wasn't respected in some cases

When creating a context using Router.method(via=somechild),
unidirectional mode was set on the new child correctly, however if the
child were to call Router.method(), due to a typing mistake the new
child would start without it.

This doesn't impact the Ansible extension, as only forked tasks are
started directly by children, and they are not responsible for routing
messages.

Add test so it can't happen again.

Diff: diff --git a/mitogen/core.py b/mitogen/core.py
index 8428a47..c228079 100644
--- a/mitogen/core.py
+++ b/mitogen/core.py
@@ -3621,11 +3621,11 @@ class ExternalContext(object):
         if self.config['profiling']:
             enable_profiling()
         self.broker = Broker(activate_compat=False)
         self.router = Router(self.broker)
         self.router.debug = self.config.get('debug', False)
-        self.router.undirectional = self.config['unidirectional']
+        self.router.unidirectional = self.config['unidirectional']
         self.router.add_handler(
             fn=self._on_shutdown_msg,
             handle=SHUTDOWN,
             policy=has_parent_authority,
         )
diff --git a/tests/router_test.py b/tests/router_test.py
index 1bd6c26..2b4a9d7 100644
--- a/tests/router_test.py
+++ b/tests/router_test.py
@@ -1,11 +1,13 @@
+import sys
 import time
 import zlib
 
 import unittest2
 
 import testlib
+import mitogen.core
 import mitogen.master
 import mitogen.parent
 import mitogen.utils
 
 try:
@@ -339,26 +341,46 @@ class NoRouteTest(testlib.RouterMixin, testlib.TestCase):
             l1.context_id,
             mitogen.context_id,
         ))
 
 
+def test_siblings_cant_talk(router):
+    l1 = router.local()
+    l2 = router.local()
+    logs = testlib.LogCapturer()
+    logs.start()
+
+    try:
+        l2.call(ping_context, l1)
+    except mitogen.core.CallError:
+        e = sys.exc_info()[1]
+
+    msg = mitogen.core.Router.unidirectional_msg % (
+        l2.context_id,
+        l1.context_id,
+    )
+    assert msg in str(e)
+    assert 'routing mode prevents forward of ' in logs.stop()
+
+
+@mitogen.core.takes_econtext
+def test_siblings_cant_talk_remote(econtext):
+    mitogen.parent.upgrade_router(econtext)
+    test_siblings_cant_talk(econtext.router)
+
+
 class UnidirectionalTest(testlib.RouterMixin, testlib.TestCase):
-    def test_siblings_cant_talk(self):
+    def test_siblings_cant_talk_master(self):
         self.router.unidirectional = True
-        l1 = self.router.local()
-        l2 = self.router.local()
-        logs = testlib.LogCapturer()
-        logs.start()
-        e = self.assertRaises(mitogen.core.CallError,
-                              lambda: l2.call(ping_context, l1))
+        test_siblings_cant_talk(self.router)
 
-        msg = self.router.unidirectional_msg % (
-            l2.context_id,
-            l1.context_id,
-        )
-        self.assertTrue(msg in str(e))
-        self.assertTrue('routing mode prevents forward of ' in logs.stop())
+    def test_siblings_cant_talk_parent(self):
+        # ensure 'unidirectional' attribute is respected for contexts started
+        # by children.
+        self.router.unidirectional = True
+        parent = self.router.local()
+        parent.call(test_siblings_cant_talk_remote)
 
     def test_auth_id_can_talk(self):
         self.router.unidirectional = True
         # One stream has auth_id stamped to that of the master, so it should be
         # treated like a parent.

Vulnerable code:         self.router.undirectional = self.config['unidirectional']
    def test_siblings_cant_talk(self):
        l1 = self.router.local()
        l2 = self.router.local()
        logs = testlib.LogCapturer()
        logs.start()
        e = self.assertRaises(mitogen.core.CallError,
                              lambda: l2.call(ping_context, l1))
        msg = self.router.unidirectional_msg % (
            l2.context_id,
            l1.context_id,
        )
        self.assertTrue(msg in str(e))
        self.assertTrue('routing mode prevents forward of ' in logs.stop())
Fixed code:         self.router.unidirectional = self.config['unidirectional']
import sys
import mitogen.core
def test_siblings_cant_talk(router):
    l1 = router.local()
    l2 = router.local()
    logs = testlib.LogCapturer()
    logs.start()

    try:
        l2.call(ping_context, l1)
    except mitogen.core.CallError:
        e = sys.exc_info()[1]

    msg = mitogen.core.Router.unidirectional_msg % (
        l2.context_id,
        l1.context_id,
    )
    assert msg in str(e)
    assert 'routing mode prevents forward of ' in logs.stop()


@mitogen.core.takes_econtext
def test_siblings_cant_talk_remote(econtext):
    mitogen.parent.upgrade_router(econtext)
    test_siblings_cant_talk(econtext.router)


    def test_siblings_cant_talk_master(self):
        test_siblings_cant_talk(self.router)
    def test_siblings_cant_talk_parent(self):
        # ensure 'unidirectional' attribute is respected for contexts started
        # by children.
        self.router.unidirectional = True
        parent = self.router.local()
        parent.call(test_siblings_cant_talk_remote)