Add --bind / -b option for multiple socket bindings

docs/settings.md

@@ -46,6 +46,12 @@ uvicorn itself.
 * `--port <int>` - Bind to a socket with this port. If set to 0, an available port will be picked. **Default:** *8000*.
 * `--uds <path>` - Bind to a UNIX domain socket, for example `--uds /tmp/uvicorn.sock`. Useful if you want to run Uvicorn behind a reverse proxy.
 * `--fd <int>` - Bind to socket from this file descriptor. Useful if you want to run Uvicorn within a process manager.
+* `--bind <str>` / `-b <str>` - Bind to one or more addresses. May be specified multiple times to listen on multiple sockets simultaneously. Supported formats: `HOST:PORT` (e.g. `0.0.0.0:8000`), `[HOST]:PORT` for IPv6 (e.g. `[::1]:8000`), `unix:PATH` (e.g. `unix:/tmp/uvicorn.sock`), `fd://NUM` (e.g. `fd://3`). Mutually exclusive with `--host`, `--port`, `--uds`, and `--fd`.
+
+!!! note
+    The `--host`, `--port`, `--uds`, and `--fd` options each bind to a single address of a single type.
+    Use `--bind` when you need to listen on multiple addresses (e.g. dual-stack IPv4 + IPv6) or mix
+    transport types (e.g. a TCP port for internal services and a unix socket behind a reverse proxy).
 
 ## Development
 

tests/test_cli.py

@@ -201,3 +201,75 @@ def test_set_app_via_environment_variable():
             args, _ = mock_run.call_args
             assert result.exit_code == 0
             assert args == (app_path,)
+
+
+def test_cli_bind_option() -> None:
+    runner = CliRunner()
+
+    with mock.patch.object(main, "run") as mock_run:
+        result = runner.invoke(cli, ["tests.test_cli:App", "--bind", "0.0.0.0:8000"])
+
+    assert result.output == ""
+    assert result.exit_code == 0
+    mock_run.assert_called_once()
+    assert mock_run.call_args[1]["bind"] == ["0.0.0.0:8000"]
+
+
+def test_cli_bind_multiple() -> None:
+    runner = CliRunner()
+
+    with mock.patch.object(main, "run") as mock_run:
+        result = runner.invoke(cli, ["tests.test_cli:App", "-b", "127.0.0.1:8000", "-b", "127.0.0.1:9000"])
+
+    assert result.output == ""
+    assert result.exit_code == 0
+    mock_run.assert_called_once()
+    assert mock_run.call_args[1]["bind"] == ["127.0.0.1:8000", "127.0.0.1:9000"]
+
+
+@pytest.mark.parametrize(
+    "extra_args",
+    [
+        ["--host", "0.0.0.0"],
+        ["--port", "9000"],
+        ["--uds", "/tmp/test.sock"],
+        ["--fd", "3"],
+    ],
+    ids=["host", "port", "uds", "fd"],
+)
+def test_cli_bind_mutually_exclusive(extra_args: list[str]) -> None:
+    runner = CliRunner()
+
+    with pytest.raises(ValueError, match="'bind' is mutually exclusive with.*"):
+        runner.invoke(cli, ["tests.test_cli:App", "-b", "127.0.0.1:8000", *extra_args], catch_exceptions=False)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="require unix-like system")
+def test_cli_bind_unix_cleanup() -> None:  # pragma: py-win32
+    sock_path = "/tmp/uvicorn_test_cleanup.sock"
+    runner = CliRunner()
+
+    try:
+        Path(sock_path).touch()
+        with mock.patch.object(Config, "bind_sockets") as mock_bind_sockets:
+            with mock.patch.object(Multiprocess, "run") as mock_run:
+                result = runner.invoke(cli, ["tests.test_cli:App", "--workers=2", "-b", f"unix:{sock_path}"])
+
+        assert result.exit_code == 0
+        mock_bind_sockets.assert_called_once()
+        mock_run.assert_called_once()
+        assert not Path(sock_path).exists()
+    finally:
+        if Path(sock_path).exists():  # pragma: no cover
+            os.remove(sock_path)
+
+
+def test_cli_bind_without_value_passes_none() -> None:
+    runner = CliRunner()
+
+    with mock.patch.object(main, "run") as mock_run:
+        result = runner.invoke(cli, ["tests.test_cli:App"])
+
+    assert result.exit_code in (0, 3)
+    mock_run.assert_called_once()
+    assert mock_run.call_args[1]["bind"] is None

tests/test_config.py

@@ -593,3 +593,88 @@ def test_setup_event_loop_is_removed(caplog: pytest.LogCaptureFixture) -> None:
         AttributeError, match="The `setup_event_loop` method was replaced by `get_loop_factory` in uvicorn 0.36.0."
     ):
         config.setup_event_loop()
+
+
+@pytest.mark.parametrize(
+    "bind_str, expected_family",
+    [
+        ("127.0.0.1:0", socket.AF_INET),
+        ("0.0.0.0:0", socket.AF_INET),
+        ("[::1]:0", socket.AF_INET6),
+        ("[::]:0", socket.AF_INET6),
+        ("localhost:0", socket.AF_INET),
+    ],
+    ids=["ipv4", "ipv4-wildcard", "ipv6", "ipv6-wildcard", "hostname"],
+)
+def test_bind_sockets_address_formats(bind_str: str, expected_family: socket.AddressFamily) -> None:
+    config = Config(app=asgi_app, bind=[bind_str])
+    sockets = config.bind_sockets()
+    assert len(sockets) == 1
+    assert sockets[0].family == expected_family
+    sockets[0].close()
+
+
+def test_bind_sockets_multiple() -> None:
+    config = Config(app=asgi_app, bind=["127.0.0.1:0", "127.0.0.1:0"])
+    sockets = config.bind_sockets()
+    assert len(sockets) == 2
+    for sock in sockets:
+        assert sock.family == socket.AF_INET
+        sock.close()
+
+
+def test_bind_sockets_default_port() -> None:
+    config = Config(app=asgi_app, bind=["127.0.0.1"])
+    sockets = config.bind_sockets()
+    assert len(sockets) == 1
+    assert sockets[0].getsockname()[1] == 8000
+    sockets[0].close()
+
+
+def test_bind_sockets_fallback() -> None:
+    config = Config(app=asgi_app, bind=None)
+    sockets = config.bind_sockets()
+    assert len(sockets) == 1
+    sockets[0].close()
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"host": "0.0.0.0"},
+        {"port": 9000},
+        {"uds": "/tmp/test.sock"},
+        {"fd": 3},
+    ],
+    ids=["host", "port", "uds", "fd"],
+)
+def test_bind_mutually_exclusive_with_other_params(kwargs: dict[str, Any]) -> None:
+    with pytest.raises(ValueError, match="'bind' is mutually exclusive with"):
+        Config(app=asgi_app, bind=["127.0.0.1:0"], **kwargs)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="requires unix sockets")
+def test_bind_sockets_unix() -> None:  # pragma: py-win32
+    sock_path = "/tmp/uvicorn_test_bind.sock"
+    try:
+        config = Config(app=asgi_app, bind=[f"unix:{sock_path}"])
+        sockets = config.bind_sockets()
+        assert len(sockets) == 1
+        assert sockets[0].family == socket.AF_UNIX
+        sockets[0].close()
+    finally:
+        if os.path.exists(sock_path):
+            os.unlink(sock_path)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="requires unix sockets")
+def test_bind_sockets_fd(tmp_path: Path) -> None:  # pragma: py-win32
+    # Create a socket, then bind via its file descriptor.
+    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    listener.bind(("127.0.0.1", 0))
+    fd = listener.fileno()
+    config = Config(app=asgi_app, bind=[f"fd://{fd}"])
+    sockets = config.bind_sockets()
+    assert len(sockets) == 1
+    sockets[0].close()
+    listener.close()

tests/test_main.py

@@ -117,6 +117,20 @@ async def test_exit_on_create_server_with_invalid_host() -> None:
     assert exc_info.value.code == 1
 
 
+async def test_run_with_bind(unused_tcp_port: int) -> None:
+    config = Config(app=app, bind=[f"127.0.0.1:{unused_tcp_port}"], loop="asyncio", limit_max_requests=1)
+    async with run_server(config):
+        async with httpx.AsyncClient() as client:
+            response = await client.get(f"http://127.0.0.1:{unused_tcp_port}")
+    assert response.status_code == 204
+
+
+async def test_run_with_bind_multiple() -> None:
+    config = Config(app=app, bind=["127.0.0.1:0", "127.0.0.1:0"], loop="asyncio", limit_max_requests=1)
+    async with run_server(config):
+        pass  # Startup itself validates multiple sockets work
+
+
 def test_deprecated_server_state_from_main() -> None:
     with pytest.deprecated_call(
         match="uvicorn.main.ServerState is deprecated, use uvicorn.server.ServerState instead."

uvicorn/config.py

@@ -183,6 +183,7 @@ def __init__(
         port: int = 8000,
         uds: str | None = None,
         fd: int | None = None,
+        bind: list[str] | None = None,
         loop: LoopFactoryType | str = "auto",
         http: type[asyncio.Protocol] | HTTPProtocolType | str = "auto",
         ws: type[asyncio.Protocol] | WSProtocolType | str = "auto",
@@ -233,6 +234,7 @@ def __init__(
         self.port = port
         self.uds = uds
         self.fd = fd
+        self.bind = bind
         self.loop = loop
         self.http = http
         self.ws = ws
@@ -342,13 +344,23 @@ def __init__(
         if self.reload and self.workers > 1:
             logger.warning('"workers" flag is ignored when reloading is enabled.')
 
+        if self.bind is not None:
+            # Only flag options explicitly set to non-default values.
+            conflicting: list[str] = []
+            if self.host != "127.0.0.1":  # default host
+                conflicting.append("host")
+            if self.port != 8000:  # default port
+                conflicting.append("port")
+            if self.uds is not None:
+                conflicting.append("uds")
+            if self.fd is not None:
+                conflicting.append("fd")
+            if conflicting:
+                raise ValueError(f"'bind' is mutually exclusive with {', '.join(map(repr, conflicting))}")
+
     @property
     def asgi_version(self) -> Literal["2.0", "3.0"]:
-        mapping: dict[str, Literal["2.0", "3.0"]] = {
-            "asgi2": "2.0",
-            "asgi3": "3.0",
-            "wsgi": "3.0",
-        }
+        mapping: dict[str, Literal["2.0", "3.0"]] = {"asgi2": "2.0", "asgi3": "3.0", "wsgi": "3.0"}
         return mapping[self.interface]
 
     @property
@@ -496,54 +508,83 @@ def get_loop_factory(self) -> Callable[[], asyncio.AbstractEventLoop] | None:
             return None
         return loop_factory(use_subprocess=self.use_subprocess)
 
-    def bind_socket(self) -> socket.socket:
+    def _bind_one(
+        self,
+        *,
+        uds: str | None = None,
+        fd: int | None = None,
+        host: str = "127.0.0.1",
+        port: int = 8000,
+    ) -> socket.socket:
         logger_args: list[str | int]
-        if self.uds:  # pragma: py-win32
-            path = self.uds
+        if uds:  # pragma: py-win32
             sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
             try:
-                sock.bind(path)
-                uds_perms = 0o666
-                os.chmod(self.uds, uds_perms)
+                sock.bind(uds)
+                os.chmod(uds, 0o666)
             except OSError as exc:  # pragma: full coverage
                 logger.error(exc)
                 sys.exit(1)
-
             message = "Uvicorn running on unix socket %s (Press CTRL+C to quit)"
             sock_name_format = "%s"
             color_message = "Uvicorn running on " + click.style(sock_name_format, bold=True) + " (Press CTRL+C to quit)"
-            logger_args = [self.uds]
-        elif self.fd:  # pragma: py-win32
-            sock = socket.fromfd(self.fd, socket.AF_UNIX, socket.SOCK_STREAM)
+            logger_args = [uds]
+        elif fd is not None:  # pragma: py-win32
+            sock = socket.fromfd(fd, socket.AF_UNIX, socket.SOCK_STREAM)
             message = "Uvicorn running on socket %s (Press CTRL+C to quit)"
             fd_name_format = "%s"
             color_message = "Uvicorn running on " + click.style(fd_name_format, bold=True) + " (Press CTRL+C to quit)"
             logger_args = [sock.getsockname()]
         else:
             family = socket.AF_INET
             addr_format = "%s://%s:%d"
-
-            if self.host and ":" in self.host:  # pragma: full coverage
-                # It's an IPv6 address.
+            if host and ":" in host:  # pragma: full coverage
                 family = socket.AF_INET6
                 addr_format = "%s://[%s]:%d"
-
             sock = socket.socket(family=family)
             sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             try:
-                sock.bind((self.host, self.port))
+                sock.bind((host, port))
             except OSError as exc:  # pragma: full coverage
                 logger.error(exc)
                 sys.exit(1)
-
             message = f"Uvicorn running on {addr_format} (Press CTRL+C to quit)"
             color_message = "Uvicorn running on " + click.style(addr_format, bold=True) + " (Press CTRL+C to quit)"
             protocol_name = "https" if self.is_ssl else "http"
-            logger_args = [protocol_name, self.host, sock.getsockname()[1]]
+            logger_args = [protocol_name, host, sock.getsockname()[1]]
         logger.info(message, *logger_args, extra={"color_message": color_message})
         sock.set_inheritable(True)
         return sock
 
+    def bind_socket(self) -> socket.socket:
+        return self._bind_one(uds=self.uds, fd=self.fd, host=self.host, port=self.port)
+
+    def bind_sockets(self) -> list[socket.socket]:
+        if self.bind is None:
+            return [self.bind_socket()]
+
+        sockets: list[socket.socket] = []
+        for bind_str in self.bind:
+            if bind_str.startswith("unix:"):  # pragma: py-win32
+                sock = self._bind_one(uds=bind_str[5:])
+            elif bind_str.startswith("fd://"):  # pragma: py-win32
+                sock = self._bind_one(fd=int(bind_str[5:]))
+            else:
+                # Strip brackets for IPv6, then rsplit on last colon.
+                raw = bind_str.replace("[", "").replace("]", "")
+                try:
+                    host, port_str = raw.rsplit(":", 1)
+                    port = int(port_str)
+                except (ValueError, IndexError):
+                    host, port = raw, 8000
+                sock = self._bind_one(host=host, port=port)
+            sockets.append(sock)
+        return sockets
+
+    @property
+    def bind_unix_paths(self) -> list[str]:  # pragma: py-win32
+        return [b[5:] for b in (self.bind or []) if b.startswith("unix:")]
+
     @property
     def should_reload(self) -> bool:
         return isinstance(self.app, str) and self.reload