Refactoring in preparation of dtls-server

None of this should alter working behavior, but several error cases are
now handled more cleanly or at all.

Author: chrysn

aiocoap/defaults.py

@@ -46,11 +46,7 @@ def get_default_clienttransports(*, loop=None, use_env=True):
     if not oscore_missing_modules():
         yield 'oscore'
 
-    try:
-        from DTLSSocket import dtls # noqa: F401
-    except ImportError:
-        pass
-    else:
+    if not dtls_missing_modules():
         yield 'tinydtls'
 
     yield 'tcpclient'

aiocoap/messagemanager.py

@@ -124,6 +124,13 @@ def dispatch_message(self, message):
             self.log.warning("Received a message with code %s and type %s (those don't fit) from %s, ignoring it.", message.code, message.mtype, message.remote)
 
     def dispatch_error(self, error, remote):
+        if self._active_exchanges is None:
+            # Not entirely sure where it is so far; better just raise a warning
+            # than an exception later, nothing terminally bad should come of
+            # this error.
+            self.log.warning("Internal shutdown sequence msismatch: error dispatched through messagemanager after shutown")
+            return
+
         self.log.debug("Incoming error %s from %r", error, remote)
 
         # cancel requests first, and then exchanges: cancelling the pending

aiocoap/tokenmanager.py

@@ -67,7 +67,13 @@ def next_token(self):
     #
 
     def dispatch_error(self, exception, remote):
-        keys_for_removal = []
+        if self.outgoing_requests is None:
+            # Not entirely sure where it is so far; better just raise a warning
+            # than an exception later, nothing terminally bad should come of
+            # this error.
+            self.log.warning("Internal shutdown sequence msismatch: error dispatched through tokenmanager after shutown")
+            return
+
         # NetworkError is what we promise users to raise from request etc; if
         # it's already a NetworkError and possibly more descriptive (eg. a
         # TimeoutError), we'll just let it through (and thus allow

aiocoap/transports/generic_udp.py

@@ -12,27 +12,24 @@
 class GenericMessageInterface(interfaces.MessageInterface):
     """GenericMessageInterface is not a standalone implementation of a
     message inteface. It does implement everything between the MessageInterface
-    and a not yet fully specified interface of "bound UDP sockets"."""
+    and a not yet fully specified interface of "bound UDP sockets".
 
-    def __init__(self, ctx: interfaces.MessageManager, log, loop):
-        self._ctx = ctx
-        self._log = log
-        self._loop = loop
+    It delegates sending through the address objects (which persist through
+    some time, given this is some kind of bound-socket scenario).
 
-    async def determine_remote(self, request):
-        if request.requested_scheme not in ('coap', None):
-            return None
+    The user must:
+    * set up a ._pool after construction with a shutdown and a connect method
+    * provide their addresses with a send(bytes) method
+    * pass incoming data to the _received_datagram and _received_exception methods
+    """
 
-        if request.unresolved_remote is not None:
-            host, port = util.hostportsplit(request.unresolved_remote)
-            port = port or COAP_PORT
-        elif request.opt.uri_host:
-            host = request.opt.uri_host
-            port = request.opt.uri_port or COAP_PORT
-        else:
-            raise ValueError("No location found to send message to (neither in .opt.uri_host nor in .remote)")
+    def __init__(self, mman: interfaces.MessageManager, log, loop):
+        self._mman = mman
+        self._log = log
+        self._loop = loop
 
-        return await self._pool.connect((host, port))
+    # Callbacks to be hooked up by the user of the class; feed data on to the
+    # message manager
 
     def _received_datagram(self, address, datagram):
         try:
@@ -41,17 +38,34 @@ def _received_datagram(self, address, datagram):
             self._log.warning("Ignoring unparsable message from %s", address)
             return
 
-        self._ctx.dispatch_message(message)
+        self._mman.dispatch_message(message)
 
     def _received_exception(self, address, exception):
-        self._ctx.dispatch_error(exception, address)
+        self._mman.dispatch_error(exception, address)
+
+    # Implementations of MessageInterface
 
     def send(self, message):
-        if self._ctx is None:
+        if self._mman is None:
             self._log.info("Not sending message %r: transport is already shutting down.", message)
         else:
             message.remote.send(message.encode())
 
     async def shutdown(self):
         await self._pool.shutdown()
-        self._ctx = None
+        self._mman = None
+
+    async def determine_remote(self, request):
+        if request.requested_scheme not in ('coap', None):
+            return None
+
+        if request.unresolved_remote is not None:
+            host, port = util.hostportsplit(request.unresolved_remote)
+            port = port or COAP_PORT
+        elif request.opt.uri_host:
+            host = request.opt.uri_host
+            port = request.opt.uri_port or COAP_PORT
+        else:
+            raise ValueError("No location found to send message to (neither in .opt.uri_host nor in .remote)")
+
+        return await self._pool.connect((host, port))

aiocoap/transports/simple6.py

@@ -50,10 +50,9 @@
 from .generic_udp import GenericMessageInterface
 
 class _Connection(asyncio.DatagramProtocol, interfaces.EndpointAddress):
-    def __init__(self, ready_callback, new_message_callback, new_error_callback, stored_sockaddr):
+    def __init__(self, ready_callback, message_interface: "GenericMessageInterface", stored_sockaddr):
         self._ready_callback = ready_callback
-        self._new_message_callback = new_message_callback
-        self._new_error_callback = new_error_callback
+        self._message_interface = message_interface
 
         # This gets stored in the _Connection because not all implementations
         # of datagram transports will expose the get_extra_info('socket')
@@ -128,10 +127,10 @@ def connection_made(self, transport):
         del self._ready_callback
 
     def datagram_received(self, data, address):
-        self._new_message_callback(self, data)
+        self._message_interface._received_datagram(self, data)
 
     def error_received(self, exception):
-        self._new_error_callback(self, exception)
+        self._message_interface._received_exception(self, exception)
 
     def connection_lost(self, exception):
         if exception is None:
@@ -141,14 +140,14 @@ def connection_lost(self, exception):
 
     # whatever it is _DatagramClientSocketpoolSimple6 expects
 
+    # ... because generic_udp expects it from _DatagramClientSocketpoolSimple6
     def send(self, data):
         self._transport.sendto(data, None)
 
     async def shutdown(self):
         self._stage = "shutting down"
         self._transport.abort()
-        del self._new_message_callback
-        del self._new_error_callback
+        del self._message_interface
         self._stage = "destroyed"
 
 class _DatagramClientSocketpoolSimple6:
@@ -174,13 +173,12 @@ class _DatagramClientSocketpoolSimple6:
     # which MessageInterface to talk to for sending), or we move the
     # MessageInterface out completely and have that object be the Protocol,
     # and the Protocol can even send new packages via the address
-    def __init__(self, loop, new_message_callback, new_error_callback):
+    def __init__(self, loop, mi: "GenericMessageInterface"):
         # using an OrderedDict to implement an LRU cache as it's suitable for that purpose according to its documentation
         self._sockets = OrderedDict()
 
         self._loop = loop
-        self._new_message_callback = new_message_callback
-        self._new_error_callback = new_error_callback
+        self._message_interface = mi
 
     async def _maybe_purge_sockets(self):
         while len(self._sockets) >= self.max_sockets: # more of an if
@@ -209,7 +207,7 @@ async def connect(self, sockaddr):
 
         ready = asyncio.get_running_loop().create_future()
         transport, protocol = await self._loop.create_datagram_endpoint(
-                lambda: _Connection(lambda: ready.set_result(None), self._new_message_callback, self._new_error_callback, sockaddr),
+                lambda: _Connection(lambda: ready.set_result(None), self._message_interface, sockaddr),
                 remote_addr=sockaddr)
         await ready
 
@@ -225,6 +223,10 @@ async def connect(self, sockaddr):
         return protocol
 
     async def shutdown(self):
+        # preventing the creation of new sockets early on, and generally
+        # breaking cycles
+        del self._message_interface
+
         if self._sockets:
             done, pending = await asyncio.wait([
                 asyncio.create_task(s.shutdown())
@@ -239,7 +241,8 @@ class MessageInterfaceSimple6(GenericMessageInterface):
     async def create_client_transport_endpoint(cls, ctx, log, loop):
         self = cls(ctx, log, loop)
 
-        self._pool = _DatagramClientSocketpoolSimple6(self._loop, self._received_datagram, self._received_exception)
+        # Cyclic reference broken during shutdown
+        self._pool = _DatagramClientSocketpoolSimple6(self._loop, self)
         return self
 
     async def recognize_remote(self, remote):

aiocoap/transports/simplesocketserver.py

@@ -40,6 +40,7 @@
 import asyncio
 from collections import namedtuple
 
+from .. import error
 from ..numbers import COAP_PORT
 from .. import interfaces
 from .generic_udp import GenericMessageInterface
@@ -87,7 +88,7 @@ class _DatagramServerSocketSimple(asyncio.DatagramProtocol):
     _Address = _Address
 
     @classmethod
-    async def create(cls, bind, log, loop, new_message_callback, new_error_callback):
+    async def create(cls, bind, log, loop, message_interface: "GenericMessageInterface"):
         if bind is None or bind[0] in ('::', '0.0.0.0', '', None):
             # If you feel tempted to remove this check, think about what
             # happens if two configured addresses can both route to a
@@ -100,7 +101,7 @@ async def create(cls, bind, log, loop, new_message_callback, new_error_callback)
         ready = asyncio.get_running_loop().create_future()
 
         transport, protocol = await loop.create_datagram_endpoint(
-                lambda: cls(ready.set_result, new_message_callback, new_error_callback, log),
+                lambda: cls(ready.set_result, message_interface, log),
                 local_addr=bind,
                 reuse_port=defaults.has_reuse_port(),
                 )
@@ -110,12 +111,13 @@ async def create(cls, bind, log, loop, new_message_callback, new_error_callback)
         # hostinfo), and can thus store the local hostinfo without distinction
         protocol.hostinfo_local = hostportjoin(bind[0], bind[1] if bind[1] != COAP_PORT else None)
 
-        return await ready
+        self = await ready
+        self._loop = loop
+        return self
 
-    def __init__(self, ready_callback, new_message_callback, new_error_callback, log):
+    def __init__(self, ready_callback, message_interface: "GenericMessageInterface", log):
         self._ready_callback = ready_callback
-        self._new_message_callback = new_message_callback
-        self._new_error_callback = new_error_callback
+        self._message_interface = message_interface
         self.log = log
 
     async def shutdown(self):
@@ -124,10 +126,26 @@ async def shutdown(self):
     # interface like _DatagramClientSocketpoolSimple6
 
     async def connect(self, sockaddr):
-        # FIXME it might be necessary to resolve the address now to get a
-        # canonical form that can be recognized later when a package comes back
+        # FIXME this is not regularly tested either
+
         self.log.warning("Sending initial messages via a server socket is not recommended")
-        return self._Address(self, sockaddr)
+        # A legitimate case is when something stores return addresses as
+        # URI(part)s and not as remotes. (In similar transports this'd also be
+        # the case if the address's connection is dropped from the pool, but
+        # that doesn't happen here since there is no pooling as there is no
+        # per-connection state).
+
+        # getaddrinfo is not only to needed to resolve any host names (which
+        # would not be recognized otherwise), but also to get a complete (host,
+        # port, zoneinfo, whatwasthefourth) tuple from what is passed in as a
+        # (host, port) tuple.
+        addresses = await self._loop.getaddrinfo(*sockaddr, family=self._transport.get_extra_info('socket').family)
+        if not addresses:
+            raise error.NetworkError("No addresses found for %s" % sockaddr[0])
+        # FIXME could do happy eyebals
+        address = addresses[0][4]
+        address = self._Address(self, address)
+        return address
 
     # datagram protocol interface
 
@@ -137,19 +155,23 @@ def connection_made(self, transport):
         del self._ready_callback
 
     def datagram_received(self, data, sockaddr):
-        self._new_message_callback(self._Address(self, sockaddr), data)
+        self._message_interface._received_datagram(self._Address(self, sockaddr), data)
 
     def error_received(self, exception):
         # This is why this whole implementation is a bad idea (but still the best we got on some platforms)
         self.log.warning("Ignoring error because it can not be mapped to any connection: %s", exception)
 
     def connection_lost(self, exception):
         if exception is None:
-            pass
+            pass # regular shutdown
         else:
             self.log.error("Received unexpected connection loss: %s", exception)
 
 class MessageInterfaceSimpleServer(GenericMessageInterface):
+    # for alteration by tinydtls_server
+    _default_port = COAP_PORT
+    _serversocket = _DatagramServerSocketSimple
+
     @classmethod
     async def create_server(cls, bind, ctx: interfaces.MessageManager, log, loop):
         self = cls(ctx, log, loop)
@@ -158,11 +180,17 @@ async def create_server(cls, bind, ctx: interfaces.MessageManager, log, loop):
         # servers that want a random port (eg. when the service URLs are
         # advertised out-of-band anyway). LwM2M clients should use simple6
         # instead as outlined there.
-        bind = (bind[0], COAP_PORT if bind[1] is None else bind[1])
+        bind = (bind[0], self._default_port if bind[1] is None else bind[1] + (self._default_port - COAP_PORT))
 
-        self._pool = await _DatagramServerSocketSimple.create(bind, log, self._loop, self._received_datagram, self._received_exception)
+        # Cyclic reference broken during shutdown
+        self._pool = await self._serversocket.create(bind, log, self._loop, self)
 
         return self
 
     async def recognize_remote(self, remote):
-        return isinstance(remote, _Address) and remote in remote.serversocket is self._pool
+        # FIXME: This is never tested (as is the connect method) because all
+        # tests create client contexts client-side (which don't build a
+        # simplesocketserver), and because even when a server context is
+        # created, there's a simple6 that grabs such addresses before a request
+        # is sent out
+        return isinstance(remote, _Address) and remote.serversocket is self._pool

aiocoap/transports/tinydtls.py

@@ -151,18 +151,18 @@ def send(self, message):
 
     log = property(lambda self: self.coaptransport.log)
 
-    def _build_accessor(self, method):
+    def _build_accessor(self, method, deadvalue):
         """Think self._build_accessor('_write')() == self._write(), just that
         it's returning a weak wrapper that allows refcounting-based GC to
         happen when the remote falls out of use"""
         weakself = weakref.ref(self)
-        def wrapper(*args, __weakself=weakself, __method=method):
+        def wrapper(*args, __weakself=weakself, __method=method, __deadvalue=deadvalue):
             self = __weakself()
             if self is None:
                 warnings.warn("DTLS module did not shut down the DTLSSocket "
                         "perfectly; it still tried to call %s in vain" %
                         __method)
-                return
+                return __deadvalue
             return getattr(self, __method)(*args)
         wrapper.__name__ = "_build_accessor(%s)" % method
         return wrapper
@@ -181,9 +181,9 @@ async def _start(self):
                     )
 
             self._dtls_socket = dtls.DTLS(
-                    read=self._build_accessor("_read"),
-                    write=self._build_accessor("_write"),
-                    event=self._build_accessor("_event"),
+                    read=self._build_accessor("_read", 0),
+                    write=self._build_accessor("_write", 0),
+                    event=self._build_accessor("_event", 0),
                     pskId=self._pskId,
                     pskStore={self._pskId: self._psk},
                     )