localstack · simonrw · Oct 31, 2024 · Oct 2, 2024 · Oct 18, 2024
@@ -1,11 +1,14 @@
+import ipaddress
 import logging
 import os
 import platform
+import re
 import socket
 import subprocess
 import tempfile
 import time
 import warnings
+from collections import defaultdict
 from typing import Any, Dict, List, Mapping, Optional, Tuple, TypeVar, Union
 
 from localstack import constants
@@ -500,6 +503,21 @@ def is_trace_logging_enabled():
     )
 
 
+def is_ipv6_address(host: str) -> bool:
+    &
8000
quot;""
+    Returns True if the given host is an IPv6 address.
+    """
+
+    if not host:
+        return False
+
+    try:
+        ipaddress.IPv6Address(host)
+        return True
+    except ipaddress.AddressValueError:
+        return False
+
+
 class HostAndPort:
     """
     Definition of an address for a server to listen to.
@@ -528,16 +546,36 @@ def parse(
             - 0.0.0.0:4566 -> host=0.0.0.0, port=4566
             - 0.0.0.0      -> host=0.0.0.0, port=`default_port`
             - :4566        -> host=`default_host`, port=4566
+            - [::]:4566    -> host=[::], port=4566
+            - [::1]        -> host=[::1], port=`default_port`
         """
         host, port = default_host, default_port
-        if ":" in input:
+
+        # recognize IPv6 addresses (+ port)
+        if input.startswith("["):
+            ipv6_pattern = re.compile(r"^\[(?P<host>[^]]+)\](:(?P<port>\d+))?$")
+            match = ipv6_pattern.match(input)
+
+            if match:
+                host = match.group("host")
+                if not is_ipv6_address(host):
+                    raise ValueError(
+                        f"input looks like an IPv6 address (is enclosed in square brackets), but is not valid: {host}"
+                    )
+                port_s = match.group("port")
+                if port_s:
+                    port = cls._validate_port(port_s)
+            else:
+                raise ValueError(
+                    f'input looks like an IPv6 address, but is invalid. Should be formatted "[ip]:port": {input}'
+                )
+
+        # recognize IPv4 address + port
+        elif ":" in input:
             hostname, port_s = input.split(":", 1)
             if hostname.strip():
                 host = hostname.strip()
            try:
-                port = int(port_s)
-            except ValueError as e:
-                raise ValueError(f"specified port {port_s} not a number") from e
+            port = cls._validate_port(port_s)
         else:
             if input.strip():
                 host = input.strip()
@@ -548,6 +586,15 @@ def parse(
 
         return cls(host=host, port=port)
 
+    @classmethod
+    def _validate_port(cls, port_s: str) -> int:
+        try:
+            port = int(port_s)
+        except ValueError as e:
+            raise ValueError(f"specified port {port_s} not a number") from e
+
+        return port
+
     def _get_unprivileged_port_range_start(self) -> int:
         try:
             with open(
@@ -562,7 +609,8 @@ def is_unprivileged(self) -> bool:
         return self.port >= self._get_unprivileged_port_range_start()
 
     def host_and_port(self):
-        return f"{self.host}:{self.port}" if self.port is not None else self.host
+        formatted_host = f"[{self.host}]" if is_ipv6_address(self.host) else self.host
+        return f"{formatted_host}:{self.port}" if self.port is not None else formatted_host
 
     def __hash__(self) -> int:
         return hash((self.host, self.port))
@@ -587,40 +635,57 @@ class UniqueHostAndPortList(List[HostAndPort]):
     """
     Container type that ensures that ports added to the list are unique based
     on these rules:
-        - 0.0.0.0 "trumps" any other binding, i.e. adding 127.0.0.1:4566 to
-          [0.0.0.0:4566] is a no-op
-        - adding identical hosts and ports is a no-op
-        - adding `0.0.0.0:4566` to [`127.0.0.1:4566`] "upgrades" the binding to
-          create [`0.0.0.0:4566`]
+        - :: "trumps" any other binding on the same port, including both IPv6 and IPv4
+          addresses. All other bindings for this port are removed, since :: already
+          covers all interfaces. For example, adding 127.0.0.1:4566, [::1]:4566,
+8000
          and [::]:4566 would result in only [::]:4566 being preserved.
+        - 0.0.0.0 "trumps" any other binding on IPv4 addresses only. IPv6 addresses
+          are not removed.
+        - Identical identical hosts and ports are de-duped
     """
 
-    def __init__(self, iterable=None):
-        super().__init__()
-        for item in iterable or []:
-            self.append(item)
+    def __init__(self, iterable: Union[List[HostAndPort], None] = None):
+        super().__init__(iterable or [])
+        self._ensure_unique()
 
-    def append(self, value: HostAndPort):
-        # no exact duplicates
-        if value in self:
+    def _ensure_unique(self):
+        """
+        Ensure that all bindings on the same port are de-duped.
+        """
+        if len(self) <= 1:
             return
 
-        # if 0.0.0.0:<port> already exists in the list, then do not add the new
-        # item
+        unique: List[HostAndPort] = list()
+
+        # Build a dictionary of hosts by port
+        hosts_by_port: Dict[int, List[str]] = defaultdict(list)
         for item in self:
-            if item.host == "0.0.0.0" and item.port == value.port:
-                return
-
-        # if we add 0.0.0.0:<port> and already contain *:<port> then bind on
-        # 0.0.0.0
-        contained_ports = {every.port for every in self}
-        if value.host == "0.0.0.0" and value.port in contained_ports:
-            for item in self:
-                if item.port == value.port:
-                    item.host = value.host
-            return
+            hosts_by_port[item.port].append(item.host)
+
+        # For any given port, dedupe the hosts
+        for port, hosts in hosts_by_port.items():
+            deduped_hosts = set(hosts)
+
+            # IPv6 all interfaces: this is the most general binding.
+            # Any others should be removed.
+            if "::" in deduped_hosts:
+                unique.append(HostAndPort(host="::", port=port))
+                continue
+            # IPv4 all interfaces: this is the next most general binding.
+            # Any others should be removed.
+            if "0.0.0.0" in deduped_hosts:
+                unique.append(HostAndPort(host="0.0.0.0", port=port))
+                continue
 
-        # append the item
+            # All other bindings just need to be unique
+            unique.extend([HostAndPort(host=host, port=port) for host in deduped_hosts])
+
+        self.clear()
+        self.extend(unique)
+
+    def append(self, value: HostAndPort):
         super().append(value)
+        self._ensure_unique()
 
 
 def populate_edge_configuration(

@@ -33,8 +33,13 @@ def register(
 
         # add endpoint for each host/port combination
         for host_and_port in listen:
-            # TODO: interface = host?
-            endpoint = endpoints.TCP4ServerEndpoint(reactor, host_and_port.port)
+            if config.is_ipv6_address(host_and_port.host):
+                endpoint = endpoints.TCP6ServerEndpoint(
+                    reactor, host_and_port.port, interface=host_and_port.host
+                )
+            else:
+                # TODO: interface = host?
+                endpoint = endpoints.TCP4ServerEndpoint(reactor, host_and_port.port)
             endpoint.listen(protocol_factory)
 
     def run(self):

diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py
@@ -203,6 +203,26 @@ def test_add_all_interfaces_value(self):
             HostAndPort("0.0.0.0", 42),
         ]
 
+    def test_add_all_interfaces_value_ipv6(self):
+        ports = config.UniqueHostAndPortList()
+        ports.append(HostAndPort("::", 42))
+        ports.append(HostAndPort("::1", 42))
+
+        assert ports == [
+            HostAndPort("::", 42),
+        ]
+
+    def test_add_all_interfaces_value_mixed_ipv6_wins(self):
+        ports = config.UniqueHostAndPortList()
+        ports.append(HostAndPort("0.0.0.0", 42))
+        ports.append(HostAndPort("::", 42))
+        ports.append(HostAndPort("127.0.0.1", 42))
+        ports.append(HostAndPort("::1", 42))
+
+        assert ports == [
+            HostAndPort("::", 42),
+        ]
+
     def test_add_all_interfaces_value_after(self):
         ports = config.UniqueHostAndPortList()
         ports.append(HostAndPort("127.0.0.1", 42))
@@ -212,6 +232,24 @@ def test_add_all_interfaces_value_after(self):
             HostAndPort("0.0.0.0", 42),
         ]
 
+    def test_add_all_interfaces_value_after_ipv6(self):
+        ports = config.UniqueHostAndPortList()
+        ports.append(HostAndPort("::1", 42))
+        ports.append(HostAndPort("::", 42))
+
+        assert ports == [
+            HostAndPort("::", 42),
+        ]
+
+    def test_add_all_interfaces_value_after_mixed_ipv6_wins(self):
+        ports = config.UniqueHostAndPortList()
+        ports.append(HostAndPort("::1", 42))
+        ports.append(HostAndPort("127.0.0.1", 42))
+        ports.append(HostAndPort("::", 42))
+        ports.append(HostAndPort("0.0.0.0", 42))
+
+        assert ports == [HostAndPort("::", 42)]
+
     def test_index_access(self):
         ports = config.UniqueHostAndPortList(
             [
@@ -260,6 +298,26 @@ def test_invalid_port(self):
 
         assert "specified port not-a-port not a number" in str(exc_info)
 
+    def test_parsing_ipv6_with_port(self):
+        h = config.HostAndPort.parse(
+            "[5601:f95d:0:10:4978::2]:1000", default_host="", default_port=9876
+        )
+        assert h == HostAndPort(host="5601:f95d:0:10:4978::2", port=1000)
+
+    def test_parsing_ipv6_with_default_port(self):
+        h = config.HostAndPort.parse("[5601:f95d:0:10:4978::2]", default_host="", default_port=9876)
+        assert h == HostAndPort(host="5601:f95d:0:10:4978::2", port=9876)
+
+    def test_parsing_ipv6_all_interfaces_with_default_port(self):
+        h = config.HostAndPort.parse("[::]", default_host="", default_port=9876)
+        assert h == HostAndPort(host="::", port=9876)
+
+    def test_parsing_ipv6_with_invalid_address(self):
+        with pytest.raises(ValueError) as exc_info:
+            config.HostAndPort.parse("[i-am-invalid]", default_host="", default_port=9876)
+
+        assert "input looks like an IPv6 address" in str(exc_info)
+
     @pytest.mark.parametrize("port", [-1000, -1, 2**16, 100_000])
     def test_port_out_of_range(self, port):
         with pytest.raises(ValueError) as exc_info: