Add MultiHostTCPSite

Author: agners

matter_server/server/helpers/custom_web_runner.py

@@ -0,0 +1,49 @@
+"""MultiHost capable aiohttp Site."""
+from __future__ import annotations
+
+import asyncio
+from ssl import SSLContext
+
+from aiohttp import web
+from yarl import URL
+
+
+class MultiHostTCPSite(web.TCPSite):
+    """MultiHost capable aiohttp Site.
+
+    Vanilla TCPSite accepts only str as host. However, the underlying asyncio's
+    create_server() implementation does take a list of strings to bind to multiple
+    host IP's. To support multiple server_host entries (e.g. to enable dual-stack
+    explicitly), we would like to pass an array of strings.
+    """
+
+    __slots__ = ("_host", "_port", "_reuse_address", "_reuse_port", "_hosturl")
+
+    def __init__(
+        self,
+        runner: web.BaseRunner,
+        host: None | str | list[str],
+        port: int,
+        *,
+        ssl_context: SSLContext | None = None,
+        backlog: int = 128,
+        reuse_address: bool | None = None,
+        reuse_port: bool | None = None,
+    ) -> None:
+        """Initialize HomeAssistantTCPSite."""
+        super().__init__(
+            runner,
+            ssl_context=ssl_context,
+            backlog=backlog,
+        )
+        self._host = host
+        self._port = port
+        self._reuse_address = reuse_address
+        self._reuse_port = reuse_port
+
+    @property
+    def name(self) -> str:
+        """Return server URL."""
+        scheme = "https" if self._ssl_context else "http"
+        host = self._host[0] if isinstance(self._host, list) else "0.0.0.0"
+        return str(URL.build(scheme=scheme, host=host, port=self._port))

matter_server/server/server.py

@@ -9,6 +9,8 @@
 
 from aiohttp import web
 
+from matter_server.server.helpers.custom_web_runner import MultiHostTCPSite
+
 from ..common.const import SCHEMA_VERSION
 from ..common.errors import VersionMismatch
 from ..common.helpers.api import APICommandHandler, api_command
@@ -104,7 +106,7 @@ async def start(self) -> None:
         self.app.router.add_route("GET", "/", self._handle_info)
         self._runner = web.AppRunner(self.app, access_log=None)
         await self._runner.setup()
-        self._http = web.TCPSite(
+        self._http = MultiHostTCPSite(
             self._runner, host=self.listen_addresses, port=self.port
         )
         await self._http.start()