[tor-commits] [stem/master] Listener exceptions and malformed events broke event loop

[atagar at torproject.org]

commit ed6460df4a6fa8446a1e2923bddde652f5d91ed7
Author: Damian Johnson <atagar at torproject.org>
Date:   Fri Aug 10 03:30:47 2018 -0700

    Listener exceptions and malformed events broke event loop
    
    When a listener raised an uncaught exception or tor provides a malformed event
    we terminated our event loop, preventing any further events from being handled.
    I haven't seen this nail us in practice but definitely both oversights.
    
      https://trac.torproject.org/projects/tor/ticket/27053
---
 docs/change_log.rst             |  2 ++
 stem/control.py                 | 25 +++++++++++++++++++++----
 test/unit/control/controller.py | 18 +++++++++---------
 3 files changed, 32 insertions(+), 13 deletions(-)

diff --git a/docs/change_log.rst b/docs/change_log.rst
index 362e4e10..75a25847 100644
--- a/docs/change_log.rst
+++ b/docs/change_log.rst
@@ -47,6 +47,8 @@ The following are only available within Stem's `git repository
  * **Controller**
 
   * Documented v3 hidden service support (:trac:`25124`, :spec:`6bd0a69`)
+  * Listener exceptions and malformed events no longer break further event processing (:trac:`27053`)
+  * Added the stem.control.MALFORMED_EVENTS event listener constant
   * Added support for limiting the maximum number of streams to :func:`~stem.control.Controller.create_ephemeral_hidden_service` (:spec:`2fcb1c2`)
   * Added a timeout argument to :class:`~stem.control.Controller` methods that could await a response (:trac:`26056`)
   * Stacktrace if :func:`stem.connection.connect` had a string port argument
diff --git a/stem/control.py b/stem/control.py
index 7e1d1286..05cf44ce 100644
--- a/stem/control.py
+++ b/stem/control.py
@@ -286,6 +286,8 @@ from stem.util import log
 
 EVENTS_LISTENING_TIMEOUT = 0.1
 
+MALFORMED_EVENTS = 'MALFORMED_EVENTS'
+
 # state changes a control socket can have
 
 State = stem.util.enum.Enum('INIT', 'RESET', 'CLOSED')
@@ -3057,6 +3059,13 @@ class Controller(BaseController):
     If a new control connection is initialized then this listener will be
     reattached.
 
+    If tor emits a malformed event it can be received by listening for the
+    stem.control.MALFORMED_EVENTS constant.
+
+    .. versionchanged:: 1.7.0
+       Listener exceptions and malformed events no longer break further event
+       processing. Added the **MALFORMED_EVENTS** constant.
+
     :param functor listener: function to be called when an event is received
     :param stem.control.EventType events: event types to be listened for
 
@@ -3874,13 +3883,21 @@ class Controller(BaseController):
         log.warn('We were unable assert ownership of tor through TAKEOWNERSHIP, despite being configured to be the owning process through __OwningControllerProcess. (%s)' % response)
 
   def _handle_event(self, event_message):
-    stem.response.convert('EVENT', event_message, arrived_at = time.time())
+    try:
+      stem.response.convert('EVENT', event_message, arrived_at = time.time())
+      event_type = event_message.type
+    except stem.ProtocolError as exc:
+      log.error('Tor sent a malformed event (%s): %s' % (exc, event_message))
+      event_type = MALFORMED_EVENTS
 
     with self._event_listeners_lock:
-      for event_type, event_listeners in list(self._event_listeners.items()):
-        if event_type == event_message.type:
+      for listener_type, event_listeners in list(self._event_listeners.items()):
+        if listener_type == event_type:
           for listener in event_listeners:
-            listener(event_message)
+            try:
+              listener(event_message)
+            except Exception as exc:
+              log.warn('Event listener raised an uncaught exception (%s): %s' % (exc, event_message))
 
   def _attach_listeners(self):
     """
diff --git a/test/unit/control/controller.py b/test/unit/control/controller.py
index fa934f34..c5809192 100644
--- a/test/unit/control/controller.py
+++ b/test/unit/control/controller.py
@@ -15,7 +15,7 @@ import stem.util.system
 import stem.version
 
 from stem import ControllerError, DescriptorUnavailable, InvalidArguments, InvalidRequest, ProtocolError, UnsatisfiableRequest
-from stem.control import _parse_circ_path, Listener, Controller, EventType
+from stem.control import MALFORMED_EVENTS, _parse_circ_path, Listener, Controller, EventType
 from stem.response import ControlMessage
 from stem.exit_policy import ExitPolicy
 
@@ -49,6 +49,9 @@ class TestControl(unittest.TestCase):
       self.bw_listener = Mock()
       self.controller.add_event_listener(self.bw_listener, EventType.BW)
 
+      self.malformed_listener = Mock()
+      self.controller.add_event_listener(self.malformed_listener, MALFORMED_EVENTS)
+
   def test_event_description(self):
     self.assertEqual("Logging at the debug runlevel. This is low level, high volume information about tor's internals that generally isn't useful to users.", stem.control.event_description('DEBUG'))
     self.assertEqual('Event emitted every second with the bytes sent and received by tor.', stem.control.event_description('BW'))
@@ -632,6 +635,7 @@ class TestControl(unittest.TestCase):
     self._emit_event(CIRC_EVENT)
     self.circ_listener.assert_called_once_with(CIRC_EVENT)
     self.bw_listener.assert_not_called()
+    self.malformed_listener.assert_not_called()
 
     self._emit_event(BW_EVENT)
     self.bw_listener.assert_called_once_with(BW_EVENT)
@@ -644,9 +648,6 @@ class TestControl(unittest.TestCase):
     event thread.
     """
 
-    self.skipTest('BUG: stem not working in this regard, about to fix')
-    return
-
     self.circ_listener.side_effect = ValueError('boom')
 
     self.controller._launch_threads()
@@ -654,6 +655,7 @@ class TestControl(unittest.TestCase):
     self._emit_event(CIRC_EVENT)
     self.circ_listener.assert_called_once_with(CIRC_EVENT)
     self.bw_listener.assert_not_called()
+    self.malformed_listener.assert_not_called()
 
     self._emit_event(BW_EVENT)
     self.bw_listener.assert_called_once_with(BW_EVENT)
@@ -666,14 +668,12 @@ class TestControl(unittest.TestCase):
     doesn't break our event thread.
     """
 
-    self.skipTest('BUG: stem not working in this regard, about to fix')
-    return
-
     self.controller._launch_threads()
 
     self._emit_event(BAD_EVENT)
-    self.circ_listener.assert_called_once_with(CIRC_EVENT)
+    self.circ_listener.assert_not_called()
     self.bw_listener.assert_not_called()
+    self.malformed_listener.assert_called_once()
 
     self._emit_event(BW_EVENT)
     self.bw_listener.assert_called_once_with(BW_EVENT)
@@ -814,4 +814,4 @@ class TestControl(unittest.TestCase):
     # doesn't seem necessary in practice, but since event processing is
     # asynchronous giving it a tiny bit of time to get handled
 
-    time.sleep(0.005)
+    time.sleep(0.03)