Add support for IPv6 in service account lookups (#512)

Author: jacobtomlinson

kr8s/_auth.py

@@ -1,6 +1,7 @@
 # SPDX-FileCopyrightText: Copyright (c) 2023-2024, Kr8s Developers (See LICENSE for list)
 # SPDX-License-Identifier: BSD 3-Clause License
 import base64
+import ipaddress
 import json
 import os
 import pathlib
@@ -256,9 +257,10 @@ async def _load_service_account(self) -> None:
         self._serviceaccount = os.path.expanduser(self._serviceaccount)
         if not os.path.isdir(self._serviceaccount):
             return
-        host = os.environ["KUBERNETES_SERVICE_HOST"]
-        port = os.environ["KUBERNETES_SERVICE_PORT"]
-        self.server = f"https://{host}:{port}"
+        self.server = self._format_server_address(
+            os.environ["KUBERNETES_SERVICE_HOST"],
+            os.environ["KUBERNETES_SERVICE_PORT"],
+        )
         async with await anyio.open_file(
             os.path.join(self._serviceaccount, "token")
         ) as f:
@@ -269,3 +271,11 @@ async def _load_service_account(self) -> None:
                 os.path.join(self._serviceaccount, "namespace")
             ) as f:
                 self.namespace = await f.read()
+
+    @staticmethod
+    def _format_server_address(host, port):
+        try:
+            ipaddress.IPv6Address(host)
+            return f"https://[{host}]:{port}"
+        except ipaddress.AddressValueError:
+            return f"https://{host}:{port}"

kr8s/tests/test_auth.py

@@ -10,6 +10,7 @@
 import yaml
 
 import kr8s
+from kr8s._auth import KubeAuth
 from kr8s._config import KubeConfig
 from kr8s._testutils import set_env
 
@@ -317,3 +318,16 @@ async def test_certs_not_encoded(kubeconfig_with_decoded_certs):
 async def test_certs_with_encoded_line_breaks(kubeconfig_with_line_breaks_in_certs):
     api = await kr8s.asyncio.api(kubeconfig=kubeconfig_with_line_breaks_in_certs)
     assert await api.get("pods", namespace=kr8s.ALL)
+
+
+@pytest.mark.parametrize(
+    "host,port,expected",
+    [
+        ("localhost", "8080", "https://localhost:8080"),
+        ("9.9.9.9", "1234", "https://9.9.9.9:1234"),
+        ("fd97:3495:4300::1", "443", "https://[fd97:3495:4300::1]:443"),
+        ("kubernetes.default.svc", "8080", "https://kubernetes.default.svc:8080"),
+    ],
+)
+def test_url_formatting(host, port, expected):
+    assert KubeAuth._format_server_address(host, port) == expected