Fix a race condition that leads to path duplication (#114)

This adds a *pending* set of clients to the `Path` instance which
prevents it from being removed while another client is currently in
the handshake process.

Author: lgrahl

saltyrtc/server/protocol.py

@@ -3,6 +3,7 @@
 import struct
 # noinspection PyUnresolvedReferences
 from typing import Dict  # noqa
+from typing import Set  # noqa
 from typing import (
     Any,
     Iterable,
@@ -79,6 +80,7 @@
 
 class Path:
     __slots__ = (
+        '_pending',
         '_initiator',
         '_responders',
         'log',
@@ -93,6 +95,7 @@ def __init__(
             number: int,
             attached: bool = True,
     ) -> None:
+        self._pending = set()  # type: Set[PathClient]
         self._initiator = None  # type: Optional[PathClient]
         self._responders = {}  # type: Dict[ResponderAddress, PathClient]
         self.log = util.get_logger('path.{}'.format(number))
@@ -105,7 +108,24 @@ def empty(self) -> bool:
         """
         Return whether the path is empty.
         """
-        return self._initiator is None and len(self._responders) == 0
+        return (len(self._pending) == 0 and
+                self._initiator is None and
+                len(self._responders) == 0)
+
+    def add_pending(self, client: 'PathClient') -> None:
+        """
+        Add a :class:`PathClient` as pending to the set of pending
+        clients.
+
+        Arguments:
+            - `client`: A :class:`PathClient` instance.
+
+        Raises :exc:`ValueError` in case of a state violation on the
+        :class:`PathClient`.
+        """
+        if not self.attached:
+            raise ValueError('Patch has been detached!')
+        self._pending.add(client)
 
     def has_client(self, client: 'PathClient') -> bool:
         """
@@ -114,7 +134,13 @@ def has_client(self, client: 'PathClient') -> bool:
 
         Arguments:
             - `client`: The :class:`PathClient` instance to look for.
+
+        Raises :exc:`ValueError` in case of a state violation on the
+        :class:`PathClient`.
         """
+        if not self.attached:
+            raise ValueError('Patch has been detached!')
+
         # Note: No need to check for an unassigned ID since the server's ID will never
         #       be available in the slots.
         id_ = client.id
@@ -137,8 +163,13 @@ def get_initiator(self) -> 'PathClient':
         """
         Return the initiator's :class:`PathClient` instance.
 
-        Raises :exc:`KeyError` if there is no initiator.
+        Raises:
+            - :exc:`KeyError` if there is no initiator.
+            - :exc:`ValueError` in case of a state violation on the
+              :class:`PathClient`.
         """
+        if not self.attached:
+            raise ValueError('Patch has been detached!')
         if self._initiator is None:
             raise KeyError('No initiator present')
         return self._initiator
@@ -150,11 +181,21 @@ def set_initiator(self, initiator: 'PathClient') -> Optional['PathClient']:
         Arguments:
             - `initiator`: A :class:`PathClient` instance.
 
-        Raises :exc:`ValueError` in case of a state violation on the
-        :class:`PathClient`.
+        Raises:
+            - :exc:`KeyError` in case the initiator was not in the set
+              of pending clients.
+            - :exc:`ValueError` in case of a state violation on the
+              :class:`PathClient`.
 
         Return the previously set initiator or `None`.
         """
+        if not self.attached:
+            raise ValueError('Patch has been detached!')
+
+        # Remove initiator from 'pending' set
+        self._pending.remove(initiator)
+
+        # Update initiator
         previous_initiator = self._initiator
         self._initiator = initiator
         self.log.debug('Set initiator {}', initiator)
@@ -173,15 +214,25 @@ def get_responder(self, id_: ResponderAddress) -> 'PathClient':
         Arguments:
             - `id_`: The identifier of the responder.
 
-        Raises :exc:`KeyError`: If `id_` cannot be associated to a
-        :class:`PathClient` instance.
+        Raises:
+            - :exc:`KeyError`: If `id_` cannot be associated to a
+              :class:`PathClient` instance.
+            - :exc:`ValueError` in case of a state violation on the
+              :class:`PathClient`.
         """
+        if not self.attached:
+            raise ValueError('Patch has been detached!')
         return self._responders[id_]
 
     def get_responder_ids(self) -> Iterable[ResponderAddress]:
         """
         Return an iterable of responder identifiers (slots).
+
+        Raises :exc:`ValueError` in case of a state violation on the
+        :class:`PathClient`.
         """
+        if not self.attached:
+            raise ValueError('Patch has been detached!')
         return self._responders.keys()
 
     def add_responder(self, responder: 'PathClient') -> ResponderAddress:
@@ -193,18 +244,27 @@ def add_responder(self, responder: 'PathClient') -> ResponderAddress:
 
         Raises:
             - :exc:`SlotsFullError` if no free slot exists on the path.
+            - :exc:`KeyError` in case the responder was not in the set
+              of pending clients.
             - :exc:`ValueError` in case of a state violation on the
-              :class:`PathClient`.
+              :class:`PathClient` or in case the responder was not in
+              the list of pending clients.
 
         Return the assigned slot identifier.
         """
+        if not self.attached:
+            raise ValueError('Patch has been detached!')
+
         # Calculate slot id
         id_ = len(self._responders) + 0x02
         try:
             id_ = ResponderAddress(id_)
         except ValueError as exc:
             raise SlotsFullError('No free slot on path') from exc
 
+        # Remove responder from 'pending' set
+        self._pending.remove(responder)
+
         # Set responder
         self._responders[id_] = responder
         self.log.debug('Added responder {}', responder)
@@ -226,13 +286,31 @@ def remove_client(self, client: 'PathClient') -> None:
         Arguments:
              - `client`: The :class:`PathClient` instance.
 
-        Raises :exc:`KeyError` in case the client provided an
-        invalid slot identifier.
+        Raises:
+            - :exc:`KeyError` in case the client provided an invalid
+              slot identifier, or the client should have been but was
+              not in the set of pending clients.
+            - :exc:`ValueError` in case of a state violation on the
+              :class:`PathClient`.
         """
+        if not self.attached:
+            raise ValueError('Patch has been detached!')
+
         if client.state == ClientState.restricted:
-            # Client has never been authenticated. Nothing to do.
+            # Client has never been authenticated, so it should be in
+            # the 'pending' set
+            self._pending.remove(client)
             return
 
+        # Client should not be in the 'pending' set!
+        try:
+            self._pending.remove(client)
+        except KeyError:
+            pass
+        else:
+            self.log.error(
+                'Client {} in state {} was in the pending set!', client, client.state)
+
         # Remove initiator or responder
         id_ = client.id
         is_initiator = id_ == INITIATOR_ADDRESS
@@ -250,6 +328,24 @@ def remove_client(self, client: 'PathClient') -> None:
             del self._responders[id_]
         self.log.debug('Removed {}', 'initiator' if is_initiator else 'responder')
 
+    def clear(self) -> None:
+        """
+        Clear an empty path.
+        """
+        if len(self._pending) != 0:
+            self.log.error(
+                'Clearing path that has pending clients: {}',
+                self._pending)
+        self._pending.clear()
+        if self._initiator is not None:
+            self.log.error(
+                'Clearing path that has an attached initiator: {}', self._initiator)
+        self._initiator = None
+        if len(self._responders) != 0:
+            self.log.error(
+                'Clearing path that has attached responders: {}', self._responders)
+        self._responders.clear()
+
 
 class PathClient:
     __slots__ = (

saltyrtc/server/server.py

@@ -345,8 +345,16 @@ async def handler(self) -> None:
         else:
             client.log.debug('Job queue completed')
 
-        # Send disconnected message if client was authenticated
-        if client.state == ClientState.authenticated:
+        # Send disconnected message if the path is not empty and client was authenticated
+        if path.empty:
+            description = 'Skipping potential disconnected message as the path has ' \
+                          'already been detached'
+            client.log.debug(description)
+        elif client.state != ClientState.authenticated:
+            client.log.debug(
+                'Skipping potential disconnected message due to {} state',
+                client.state.name)
+        else:
             # Initiator: Send to all responders
             if client.type == AddressType.initiator:
                 responder_ids = path.get_responder_ids()
@@ -385,10 +393,6 @@ async def handler(self) -> None:
                         client.log.exception(description, exc)
             else:
                 client.log.error('Invalid address type: {}', client.type)
-        else:
-            client.log.debug(
-                'Skipping potential disconnected message due to {} state',
-                client.state.name)
 
         # Wait for the connection to be closed
         await close_awaitable
@@ -417,6 +421,10 @@ def close(self, code: CloseCode) -> None:
                 # We can safely ignore this since clients will be removed immediately
                 # from the path in case they are being dropped by another client.
                 pass
+            except ValueError:
+                # We can also safely ignore this since a client may have already removed
+                # itself from the path.
+                pass
 
     def get_path_client(
             self,
@@ -441,6 +449,9 @@ def get_path_client(
         # Create client instance
         client = PathClient(connection, path.number, initiator_key, loop=self._loop)
 
+        # Attach client to path as 'pending'
+        path.add_pending(client)
+
         # Return path and client
         return path, client
 
@@ -888,6 +899,11 @@ def _drop_client(self, client: PathClient, code: CloseCode) -> None:
         Arguments:
             - `client`: The client to be dropped.
             - `close`: The close code.
+
+        Raises:
+            - :exc:`KeyError` in case the client is not attached to the
+              path.
+            - :exc:`ValueError` in case the path has been detached.
         """
         # Drop the client
         client.drop(code)
@@ -914,14 +930,18 @@ def get(self, initiator_key: InitiatorPublicPermanentKey) -> Path:
         return self.paths[initiator_key]
 
     def clean(self, path: Path) -> None:
-        if path.attached and path.empty:
+        if not path.attached:
+            self._log.error('Path {} has already been detached', path.number)
+            return
+        if path.empty:
             path.attached = False
             try:
                 del self.paths[path.initiator_key]
             except KeyError:
                 self._log.error('Path {} has already been removed', path.number)
             else:
                 self._log.debug('Removed empty path: {}', path.number)
+            path.clear()
 
 
 class Server:

tests/test_protocol.py

@@ -1746,6 +1746,8 @@ async def test_path_full_lite(self, initiator_key, server, client_factory):
 
         # Add fake clients to path
         clients = [_FakePathClient() for _ in range(0x02, 0x100)]
+        for client in clients:
+            path.add_pending(client)
         for client in clients:
             path.add_responder(client)