From dd9b2b1332d14e70a696e7c80d80c60043116a14 Mon Sep 17 00:00:00 2001 From: Jacob Tomlinson Date: Fri, 1 Nov 2024 16:33:55 +0000 Subject: [PATCH] Add support for IPv6 in service account lookups (#512) --- kr8s/_auth.py | 16 +++++++++++++--- kr8s/tests/test_auth.py | 14 ++++++++++++++ 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/kr8s/_auth.py b/kr8s/_auth.py index 0ad6190c..9941c6aa 100644 --- a/kr8s/_auth.py +++ b/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}" diff --git a/kr8s/tests/test_auth.py b/kr8s/tests/test_auth.py index f70998b0..29893b2c 100644 --- a/kr8s/tests/test_auth.py +++ b/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