keybridge: per-host ports for kmip

sambacc/commands/satellite/keybridge.py

@@ -214,8 +214,7 @@ def _kmip_scope(
 
     return sambacc.kmip.scope.KMIPScope(
         scope_cfg.subname,
-        hostnames=scope_cfg.hostnames,
-        port=scope_cfg.port,
+        hosts=scope_cfg.host_ports,
         tls_paths=sambacc.kmip.scope.TLSPaths(**tls),
     )
 

sambacc/config.py

@@ -24,6 +24,7 @@
 import json
 import sys
 import typing
+import urllib
 
 from .opener import Opener, FileOpener
 
@@ -614,6 +615,20 @@ def subname(self) -> str:
     def hostnames(self) -> list[str]:
         return list(self._cfg.get("hostnames", []))
 
+    @property
+    def host_ports(self) -> list[tuple[str, int]]:
+        default_port = self.port
+        out: list[tuple[str, int]] = []
+        for host in self.hostnames:
+            host, port = _safe_split_host_port(host)
+            if port <= 0 and default_port <= 0:
+                raise ValueError("no host port and no default port")
+            elif port <= 0:
+                out.append((host, default_port))
+            else:
+                out.append((host, int(port)))
+        return out
+
     @property
     def port(self) -> int:
         return int(self._cfg.get("port", -1))
@@ -772,3 +787,13 @@ def _globals_data(gconfig: GlobalConfig, iconfig: dict) -> list:
     except KeyError:
         return []
     return [gconfig.data["globals"][n] for n in gnames]
+
+
+def _safe_split_host_port(host: str, scheme: str = "https") -> tuple[str, int]:
+    if host.count(":") > 1 and "[" not in host:
+        # assume this is an undecorated ipv6 and pass it thru w/o port
+        return (host, -1)
+    # otherwise let urllib do the work for us
+    parsed = urllib.parse.urlparse(f"{scheme}://{host}")
+    assert parsed.hostname, "invalid hostname"  # should be impossible
+    return parsed.hostname, int(parsed.port or -1)

sambacc/kmip/scope.py

@@ -64,22 +64,19 @@ def __init__(
         self,
         kmip_name: str,
         *,
-        hostnames: typing.Collection[str],
-        port: int,
+        hosts: typing.Collection[tuple[str, int]],
         tls_paths: TLSPaths,
     ) -> None:
         self.kmip_name = kmip_name
-        self.hostnames = hostnames
-        self.port = port
+        self.hosts = hosts
         self.tls_paths = tls_paths
         self._kmip_version = kmip_enums.KMIPVersion.KMIP_1_2
         self._cache_lock = threading.Lock()
         self._kmip_cache: dict[str, _Value] = {}
         _logger.debug(
-            "Created KMIP Scope with name=%r, hostnames=%r, port=%r, tls=%r",
+            "Created KMIP Scope with name=%r, hosts=%r, tls=%r",
             self.kmip_name,
-            self.hostnames,
-            self.port,
+            self.hosts,
             self.tls_paths,
         )
 
@@ -96,11 +93,11 @@ def about(self) -> ScopeInfo:
 
     @contextlib.contextmanager
     def _client(self) -> typing.Iterator[ProxyKmipClient]:
-        for hostname in self.hostnames:
+        for hostname, port in self.hosts:
             try:
                 client = ProxyKmipClient(
                     hostname=hostname,
-                    port=self.port,
+                    port=port,
                     cert=self.tls_paths.cert,
                     key=self.tls_paths.key,
                     ca=self.tls_paths.ca_cert,