lib-v2: handle validation of ports for both v6 and v4 (#1332)

* better handling of ipv6 and wildcard addresses

* update version

* cleanup tests

* greatly improve readability

* update hashes

* better check

* flake

* regen hashes

library/2.1.8/container.py → library/2.1.9/container.py

@@ -235,10 +235,13 @@ def add_port(self, port_config: dict | None = None, dev_config: dict | None = No
         host_port = config.get("port_number", 0)
         container_port = config.get("container_port", 0) or host_port
         protocol = config.get("protocol", "tcp")
-        host_ip = config.get("host_ip", "0.0.0.0")
+        host_ips = config.get("host_ips", ["0.0.0.0", "::"])
+        if not isinstance(host_ips, list):
+            raise RenderError(f"Expected [host_ips] to be a list, got [{host_ips}]")
 
         if bind_mode == "published":
-            self.ports.add_port(host_port, container_port, {"protocol": protocol, "host_ip": host_ip})
+            for host_ip in host_ips:
+                self.ports.add_port(host_port, container_port, {"protocol": protocol, "host_ip": host_ip})
         elif bind_mode == "exposed":
             self.expose.add_port(container_port, protocol)
 

library/2.1.9/ports.py

@@ -0,0 +1,153 @@
+import ipaddress
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from render import Render
+
+try:
+    from .error import RenderError
+    from .validations import (
+        valid_ip_or_raise,
+        valid_port_mode_or_raise,
+        valid_port_or_raise,
+        valid_port_protocol_or_raise,
+    )
+except ImportError:
+    from error import RenderError
+    from validations import (
+        valid_ip_or_raise,
+        valid_port_mode_or_raise,
+        valid_port_or_raise,
+        valid_port_protocol_or_raise,
+    )
+
+
+class Ports:
+    def __init__(self, render_instance: "Render"):
+        self._render_instance = render_instance
+        self._ports: dict[str, dict] = {}
+
+    def _gen_port_key(self, host_port: int, host_ip: str, proto: str, ip_family: int) -> str:
+        return f"{host_port}_{host_ip}_{proto}_{ip_family}"
+
+    def _is_wildcard_ip(self, ip: str) -> bool:
+        return ip in ["0.0.0.0", "::"]
+
+    def _get_opposite_wildcard(self, ip: str) -> str:
+        return "0.0.0.0" if ip == "::" else "::"
+
+    def _get_sort_key(self, p: dict) -> str:
+        return f"{p['published']}_{p['target']}_{p['protocol']}_{p.get('host_ip', '_')}"
+
+    def _is_ports_same(self, port1: dict, port2: dict) -> bool:
+        return (
+            port1["published"] == port2["published"]
+            and port1["target"] == port2["target"]
+            and port1["protocol"] == port2["protocol"]
+            and port1.get("host_ip", "_") == port2.get("host_ip", "_")
+        )
+
+    def _has_opposite_family_port(self, port_config: dict, wildcard_ports: dict) -> bool:
+        comparison_port = port_config.copy()
+        comparison_port["host_ip"] = self._get_opposite_wildcard(port_config["host_ip"])
+        for p in wildcard_ports.values():
+            if self._is_ports_same(comparison_port, p):
+                return True
+        return False
+
+    def _check_port_conflicts(self, port_config: dict, ip_family: int) -> None:
+        host_port = port_config["published"]
+        host_ip = port_config["host_ip"]
+        proto = port_config["protocol"]
+
+        key = self._gen_port_key(host_port, host_ip, proto, ip_family)
+
+        if key in self._ports.keys():
+            raise RenderError(f"Port [{host_port}/{proto}/ipv{ip_family}] already added for [{host_ip}]")
+
+        wildcard_ip = "0.0.0.0" if ip_family == 4 else "::"
+        if host_ip != wildcard_ip:
+            # Check if there is a port with same details but with wildcard IP of the same family
+            wildcard_key = self._gen_port_key(host_port, wildcard_ip, proto, ip_family)
+            if wildcard_key in self._ports.keys():
+                raise RenderError(
+                    f"Cannot bind port [{host_port}/{proto}/ipv{ip_family}] to [{host_ip}], "
+                    f"already bound to [{wildcard_ip}]"
+                )
+        else:
+            # We are adding a port with wildcard IP
+            # Check if there is a port with same details but with specific IP of the same family
+            for p in self._ports.values():
+                # Skip if the port is not for the same family
+                if ip_family != ipaddress.ip_address(p["host_ip"]).version:
+                    continue
+
+                # Make a copy of the port config
+                search_port = p.copy()
+                # Replace the host IP with wildcard IP
+                search_port["host_ip"] = wildcard_ip
+                # If the ports match, means that a port for specific IP is already added
+                # and we are trying to add it again with wildcard IP. Raise an error
+                if self._is_ports_same(search_port, port_config):
+                    raise RenderError(
+                        f"Cannot bind port [{host_port}/{proto}/ipv{ip_family}] to [{host_ip}], "
+                        f"already bound to [{p['host_ip']}]"
+                    )
+
+    def add_port(self, host_port: int, container_port: int, config: dict | None = None):
+        config = config or {}
+        host_port = valid_port_or_raise(host_port)
+        container_port = valid_port_or_raise(container_port)
+        proto = valid_port_protocol_or_raise(config.get("protocol", "tcp"))
+        mode = valid_port_mode_or_raise(config.get("mode", "ingress"))
+
+        # TODO: Once all apps stop using this function directly, (ie using the container.add_port function)
+        # Remove this, and let container.add_port call this for each host_ip
+        host_ip = config.get("host_ip", None)
+        if host_ip is None:
+            self.add_port(host_port, container_port, config | {"host_ip": "0.0.0.0"})
+            self.add_port(host_port, container_port, config | {"host_ip": "::"})
+            return
+
+        host_ip = valid_ip_or_raise(config.get("host_ip", None))
+        ip = ipaddress.ip_address(host_ip)
+
+        port_config = {
+            "published": host_port,
+            "target": container_port,
+            "protocol": proto,
+            "mode": mode,
+            "host_ip": host_ip,
+        }
+        self._check_port_conflicts(port_config, ip.version)
+
+        key = self._gen_port_key(host_port, host_ip, proto, ip.version)
+        self._ports[key] = port_config
+
+    def has_ports(self):
+        return len(self._ports) > 0
+
+    def render(self):
+        specific_ports = []
+        wildcard_ports = {}
+
+        for port_config in self._ports.values():
+            if self._is_wildcard_ip(port_config["host_ip"]):
+                wildcard_ports[id(port_config)] = port_config.copy()
+            else:
+                specific_ports.append(port_config.copy())
+
+        processed_ports = specific_ports.copy()
+        for wild_port in wildcard_ports.values():
+            processed_port = wild_port.copy()
+
+            # Check if there's a matching wildcard port for the opposite IP family
+            has_opposite_family = self._has_opposite_family_port(wild_port, wildcard_ports)
+
+            if has_opposite_family:
+                processed_port.pop("host_ip")
+
+            if processed_port not in processed_ports:
+                processed_ports.append(processed_port)
+
+        return sorted(processed_ports, key=self._get_sort_key)

library/2.1.8/tests/test_container.py → library/2.1.9/tests/test_container.py

@@ -354,8 +354,16 @@ def test_add_ports(mock_values):
     )
     output = render.render()
     assert output["services"]["test_container"]["ports"] == [
-        {"published": 8081, "target": 8080, "protocol": "tcp", "mode": "ingress", "host_ip": "0.0.0.0"},
-        {"published": 8082, "target": 8080, "protocol": "udp", "mode": "ingress", "host_ip": "0.0.0.0"},
-        {"published": 9091, "target": 9092, "protocol": "udp", "mode": "ingress", "host_ip": "0.0.0.0"},
+        {"published": 8081, "target": 8080, "protocol": "tcp", "mode": "ingress"},
+        {"published": 8082, "target": 8080, "protocol": "udp", "mode": "ingress"},
+        {"published": 9091, "target": 9092, "protocol": "udp", "mode": "ingress"},
     ]
     assert output["services"]["test_container"]["expose"] == ["8080/tcp"]
+
+
+def test_add_ports_with_invalid_host_ips(mock_values):
+    render = Render(mock_values)
+    c1 = render.add_container("test_container", "test_image")
+    c1.healthcheck.disable()
+    with pytest.raises(Exception):
+        c1.add_port({"port_number": 8081, "container_port": 8080, "bind_mode": "published", "host_ips": "invalid"})