Fix bind unspecified address controller will raise TimeoutError

The controller throws a TimeoutError when binding a unspecified address (e.g. 0.0.0.0)

Author: wevsty

aiosmtpd/controller.py

@@ -79,6 +79,24 @@ def get_localhost() -> Literal["::1", "127.0.0.1"]:
         raise
 
 
+@public
+def is_unspecified_address(address: str) -> bool:
+    unspecified_address_list = ["", "0.0.0.0", "::"]
+    return address in unspecified_address_list
+
+
+@public
+def convert_unspecified_address_to_localhost(
+        address: str
+) -> Literal["::1", "127.0.0.1"]:
+    address_dict = {
+        "": get_localhost(),
+        "0.0.0.0": "127.0.0.1",
+        "::": "::1",
+    }
+    return address_dict.get(address, get_localhost())
+
+
 class _FakeServer(asyncio.StreamReaderProtocol):
     """
     Returned by _factory_invoker() in lieu of an SMTP instance in case
@@ -396,8 +414,15 @@ def __init__(
             loop,
             **kwargs,
         )
-        self._localhost = get_localhost()
-        self.hostname = self._localhost if hostname is None else hostname
+        if is_unspecified_address(hostname) is True:
+            self._localhost = convert_unspecified_address_to_localhost(hostname)
+            self.hostname = hostname
+        elif hostname is None:
+            self._localhost = get_localhost()
+            self.hostname = self._localhost
+        else:
+            self._localhost = get_localhost()
+            self.hostname = hostname
         self.port = port
 
     def _create_server(self) -> Coroutine:
@@ -421,7 +446,7 @@ def _trigger_server(self):
         """
         # At this point, if self.hostname is Falsy, it most likely is "" (bind to all
         # addresses). In such case, it should be safe to connect to localhost)
-        hostname = self.hostname or self._localhost
+        hostname = self._localhost if is_unspecified_address(self.hostname) is True else self.hostname
         with ExitStack() as stk:
             s = stk.enter_context(create_connection((hostname, self.port), 1.0))
             if self.ssl_context:

aiosmtpd/docs/NEWS.rst

@@ -14,7 +14,7 @@ Fixed/Improved
 --------------
 * All Controllers now have more rationale design, as they are now composited from a Base + a Mixin
 * A whole bunch of annotations
-
+* Fix bind unspecified address controller will raise TimeoutError
 
 1.4.2 (2021-03-08)
 =====================

aiosmtpd/tests/test_server.py

@@ -28,6 +28,8 @@
     UnixSocketUnthreadedController,
     _FakeServer,
     get_localhost,
+    is_unspecified_address,
+    convert_unspecified_address_to_localhost,
 )
 from aiosmtpd.handlers import Sink
 from aiosmtpd.smtp import SMTP as Server
@@ -293,6 +295,20 @@ def test_hostname_none(self):
         finally:
             cont.stop()
 
+    def test_hostname_unspecified_ipv4(self):
+        cont = Controller(Sink(), hostname="0.0.0.0")
+        try:
+            cont.start()
+        finally:
+            cont.stop()
+
+    def test_hostname_unspecified_ipv6(self):
+        cont = Controller(Sink(), hostname="::")
+        try:
+            cont.start()
+        finally:
+            cont.stop()
+
     def test_testconn_raises(self, mocker: MockFixture):
         mocker.patch("socket.socket.recv", side_effect=RuntimeError("MockError"))
         cont = Controller(Sink(), hostname="")
@@ -347,6 +363,18 @@ def test_getlocalhost_error(self, mocker):
         assert exc.value.errno == errno.EFAULT
         mock_makesock.assert_called_with(socket.AF_INET6, socket.SOCK_STREAM)
 
+    def test_is_unspecified_address(self):
+        assert is_unspecified_address("127.0.0.1") is False
+        assert is_unspecified_address("0.0.0.0") is True
+        assert is_unspecified_address("::") is True
+        assert is_unspecified_address("") is True
+
+    def test_convert_unspecified_address_to_localhost(self):
+        assert convert_unspecified_address_to_localhost("") == get_localhost()
+        assert convert_unspecified_address_to_localhost("0.0.0.0") == "127.0.0.1"
+        assert convert_unspecified_address_to_localhost("::") == "::1"
+        assert convert_unspecified_address_to_localhost("0.0.0.1") == get_localhost()
+
     def test_stop_default(self):
         controller = Controller(Sink())
         with pytest.raises(AssertionError, match="SMTP daemon not running"):