Fix swarm tests

Author: mesudip

nginx/Url.py

@@ -36,16 +36,20 @@ def parse(entry_string: str, default_scheme=None, default_port=None, default_loc
         return Url(scheme, host if host else None, port, location)
 
     @staticmethod
-    def is_valid_hostname(hostname: str) -> bool:
+    def is_valid_hostname(hostname: str, allow_wildcard: bool = False, max_length: int = 253) -> bool:
         """
         https://stackoverflow.com/a/33214423/2804342
         :return: True if for valid hostname False otherwise
         """
+        if not hostname:
+            return False
         if hostname[-1] == ".":
             # strip exactly one dot from the right, if present
             hostname = hostname[:-1]
-        if len(hostname) > 253:
+        if len(hostname) > max_length:
             return False
+        if allow_wildcard and hostname.startswith("*."):
+            hostname = hostname[2:]
 
         labels = hostname.split(".")
 

nginx_proxy/BackendTarget.py

@@ -133,3 +133,10 @@ def __init__(self, network_names: List[str], backend_type="container"):
 class NoHostConfiguration(UnconfiguredBackend):
     def __init__(self, backend_type="container"):
         super().__init__(backend_type)
+
+
+class InvalidHostConfiguration(UnconfiguredBackend):
+    def __init__(self, hostname: str, reason: str, backend_type="container"):
+        super().__init__(backend_type)
+        self.hostname = hostname
+        self.reason = reason

nginx_proxy/pre_processors/redirect_processor.py

@@ -8,6 +8,10 @@
 from nginx_proxy.Host import Host
 
 
+def _is_certificate_redirect_target(target: Url):
+    return "https" in target.scheme or "wss" in target.scheme or int(target.port or 80) == 443
+
+
 def process_redirection(backend: BackendTarget, environments: map, vhost_map: Dict[str, Dict[int, Host]]):
     redirect_env = [e[1] for e in environments.items() if e[0].startswith("PROXY_FULL_REDIRECT")]
     hosts = []
@@ -39,8 +43,19 @@ def process_redirection(backend: BackendTarget, environments: map, vhost_map: Di
                     target.port = 443 if "https" in target.scheme or "wss" in target.scheme else 80
                 if not target.scheme:
                     target.scheme = {"https"} if target.port == 443 else {"http"}
+                if not Url.is_valid_hostname(target.hostname, allow_wildcard=True):
+                    print("Invalid PROXY_FULL_REDIRECT target hostname: " + target.hostname)
+                    continue
+                if _is_certificate_redirect_target(target) and not Url.is_valid_hostname(
+                    target.hostname, allow_wildcard=True, max_length=64
+                ):
+                    print("Invalid PROXY_FULL_REDIRECT target certificate hostname: " + target.hostname)
+                    continue
                 for source in sources:
                     if source.hostname is not None:
+                        if not Url.is_valid_hostname(source.hostname, allow_wildcard=True):
+                            print("Invalid PROXY_FULL_REDIRECT source hostname: " + source.hostname)
+                            continue
                         port = 80 if source.port is None else int(source.port)
                         if source.hostname not in vhost_map:
                             host = Host(source.hostname, port)

nginx_proxy/pre_processors/virtual_host_processor.py

@@ -1,7 +1,8 @@
 import re
 
+from nginx import Url
 from nginx_proxy import Host, ProxyConfigData
-from nginx_proxy.BackendTarget import BackendTarget, NoHostConfiguration, UnreachableNetwork
+from nginx_proxy.BackendTarget import BackendTarget, InvalidHostConfiguration, NoHostConfiguration, UnreachableNetwork
 from nginx_proxy.utils import split_url
 
 
@@ -20,6 +21,17 @@ def _default_external_port(schemes):
     return 443 if has_secure_scheme and not has_insecure_scheme else 80
 
 
+def _requires_certificate(host: Host) -> bool:
+    return "https" in host.scheme or "wss" in host.scheme or int(host.port or 80) == 443
+
+
+def _validate_external_host(host: Host):
+    if not Url.is_valid_hostname(host.hostname, allow_wildcard=True):
+        raise InvalidHostConfiguration(host.hostname, "invalid hostname")
+    if _requires_certificate(host) and not Url.is_valid_hostname(host.hostname, allow_wildcard=True, max_length=64):
+        raise InvalidHostConfiguration(host.hostname, "certificate hostnames must be 64 characters or fewer")
+
+
 def _parse_extra_directive(raw_directive: str):
     directive = raw_directive.strip()
     if not directive:
@@ -96,6 +108,14 @@ def process_virtual_hosts(backend: BackendTarget, known_networks: set) -> ProxyC
             "networks: " + ", ".join(list(e.network_names)),
             sep="\t",
         )
+    except InvalidHostConfiguration as e:
+        print(
+            "Invalid VIRTUAL_HOST  ",
+            f"{backend.type:>9}".title() + " Id: " + backend.id[:12],
+            backend.name,
+            f"{e.hostname}: {e.reason}",
+            sep="\t",
+        )
     return hosts
 
 
@@ -176,6 +196,7 @@ def host_generator(backend: BackendTarget, known_networks: set = {}):
 
     for host_config in static_hosts:
         host, location, container_data, extras = _parse_host_entry(host_config)
+        _validate_external_host(host)
 
         if container_data.address is None:
             print(
@@ -208,6 +229,7 @@ def host_generator(backend: BackendTarget, known_networks: set = {}):
 
     for host_config in virtual_hosts:
         host, location, container_data, extras = _parse_host_entry(host_config)
+        _validate_external_host(host)
         # Protect double / in urls.
         if location and not location.endswith("/") and container_data.path and container_data.path.endswith("/"):
             location = location + "/"

tests/integration/test_nginx_proxy.py

@@ -38,6 +38,10 @@ def get_request_url(virtual_host, request_path, scheme="http"):
     return f"{scheme}://{hostname}{request_path}"
 
 
+def _hostname_mode_token(swarm_mode):
+    return {"prefer-local": "pl"}.get(swarm_mode, swarm_mode)
+
+
 def _has_proxy_server(config_str, server_name):
     config = HttpBlock.parse(config_str)
     for server in config.servers:
@@ -230,9 +234,9 @@ def test_proxy_full_redirect_to_https_target_response(
     if not is_reachable(swarm_mode, backend_type):
         pytest.skip("Backend discovery not available for this swarm mode/backend type combination.")
 
-    suffix = f"{backend_type}.{swarm_mode}.{datetime.now(timezone.utc).strftime('%H%M%S%f')}"
-    target_host = f"{suffix}.full-redirect-target.example.com"
-    source_host = f"{suffix}.full-redirect-source.example.com"
+    suffix = f"{backend_type[:3]}.{_hostname_mode_token(swarm_mode)}.{datetime.now(timezone.utc).strftime('%H%M%S%f')}"
+    target_host = f"{suffix}.frt.example.com"
+    source_host = f"{suffix}.frs.example.com"
     env = {
         "VIRTUAL_HOST": f"https://{target_host} -> :8080",
         "VIRTUAL_PORT": "8080",

tests/unit/test_backend_target.py

@@ -228,6 +228,78 @@ def test_process_virtual_hosts_uses_next_network_when_first_ip_blank(self):
         backend = location.backends[0]
         assert backend.address == "10.0.0.12"
 
+    def test_https_virtual_host_rejects_certificate_hostname_longer_than_64_chars(self):
+        long_hostname = f"{'a' * 55}.example.com"
+        assert len(long_hostname) > 64
+        bt = BackendTarget(
+            id="long-https-id",
+            name="long-https-test",
+            env={"VIRTUAL_HOST": f"https://{long_hostname}"},
+            network_settings={"my-net": {"NetworkID": "my-net-id", "IPAddress": "10.0.0.12"}},
+        )
+
+        config_data = process_virtual_hosts(bt, {"my-net-id"})
+
+        assert len(list(config_data.host_list())) == 0
+
+    def test_http_virtual_host_allows_dns_valid_hostname_longer_than_64_chars(self):
+        long_hostname = f"{'a' * 55}.example.com"
+        assert len(long_hostname) > 64
+        bt = BackendTarget(
+            id="long-http-id",
+            name="long-http-test",
+            env={"VIRTUAL_HOST": long_hostname},
+            network_settings={"my-net": {"NetworkID": "my-net-id", "IPAddress": "10.0.0.12"}},
+        )
+
+        config_data = process_virtual_hosts(bt, {"my-net-id"})
+
+        hosts = list(config_data.host_list())
+        assert len(hosts) == 1
+        assert hosts[0].hostname == long_hostname
+
+    def test_https_virtual_host_allows_certificate_hostname_at_64_chars(self):
+        hostname = f"{'a' * 52}.example.com"
+        assert len(hostname) == 64
+        bt = BackendTarget(
+            id="max-https-id",
+            name="max-https-test",
+            env={"VIRTUAL_HOST": f"https://{hostname}"},
+            network_settings={"my-net": {"NetworkID": "my-net-id", "IPAddress": "10.0.0.12"}},
+        )
+
+        config_data = process_virtual_hosts(bt, {"my-net-id"})
+
+        hosts = list(config_data.host_list())
+        assert len(hosts) == 1
+        assert hosts[0].hostname == hostname
+
+    def test_virtual_host_rejects_invalid_hostname(self):
+        bt = BackendTarget(
+            id="invalid-host-id",
+            name="invalid-host-test",
+            env={"VIRTUAL_HOST": "bad_host.example.com"},
+            network_settings={"my-net": {"NetworkID": "my-net-id", "IPAddress": "10.0.0.12"}},
+        )
+
+        config_data = process_virtual_hosts(bt, {"my-net-id"})
+
+        assert len(list(config_data.host_list())) == 0
+
+    def test_virtual_host_allows_wildcard_hostname(self):
+        bt = BackendTarget(
+            id="wildcard-host-id",
+            name="wildcard-host-test",
+            env={"VIRTUAL_HOST": "https://*.example.com"},
+            network_settings={"my-net": {"NetworkID": "my-net-id", "IPAddress": "10.0.0.12"}},
+        )
+
+        config_data = process_virtual_hosts(bt, {"my-net-id"})
+
+        hosts = list(config_data.host_list())
+        assert len(hosts) == 1
+        assert hosts[0].hostname == "*.example.com"
+
     def test_parse_host_entry_simple(self):
         h, loc, c, extras = _parse_host_entry("example.com")
         assert h.hostname == "example.com"