diff --git a/pyproject.toml b/pyproject.toml index fc4425c..ac11228 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,9 @@ dependencies = [ "docker>=7.1.0", "fastapi>=0.128.0", "httpx>=0.28.1", - "kubernetes>=34.1.0", + # <36: kubernetes-client 36.0.0 stops picking up kubelet-rotated bound SA + # tokens in-cluster, causing permanent 401s after ~1h on AKS (issue #72). + "kubernetes>=34.1.0,<36", "minio>=7.2.20", "pyasn1>=0.6.3", "pydantic>=2.12.5", diff --git a/src/services/kubernetes/client.py b/src/services/kubernetes/client.py index a6628e1..94439d4 100644 --- a/src/services/kubernetes/client.py +++ b/src/services/kubernetes/client.py @@ -6,6 +6,7 @@ import json import os +import time from functools import lru_cache from typing import Optional, Tuple @@ -123,6 +124,13 @@ def build_custom_labels( _initialized: bool = False _init_error: str | None = None +# Throttle for 401-triggered client resets (seconds). A reset re-runs config +# loading, which re-reads the mounted ServiceAccount token from disk. The +# throttle keeps a persistent auth failure (e.g. RBAC misconfiguration) from +# turning every API call into a config reload. +_UNAUTHORIZED_RESET_INTERVAL_SECONDS = 30.0 +_last_unauthorized_reset: float = 0.0 + def _load_config() -> bool: """Load Kubernetes configuration. @@ -203,6 +211,50 @@ def get_kubernetes_client() -> tuple[CoreV1Api | None, BatchV1Api | None]: return _core_api, _batch_api +def reset_clients() -> None: + """Drop the cached API clients so the next call rebuilds them. + + Rebuilding re-runs config loading, which re-reads the mounted + ServiceAccount token from disk. Old client objects are left for GC so + in-flight requests on them can complete. + """ + global _core_api, _batch_api, _initialized, _init_error + _core_api = None + _batch_api = None + _initialized = False + _init_error = None + + +def handle_unauthorized(status: int | None) -> bool: + """Recover from a 401 by resetting the cached clients (throttled). + + The API server returning 401 means the token we presented is no longer + valid — most commonly a stale bound ServiceAccount token after kubelet + rotation (issue #72). Resetting forces the next ``get_core_api()`` / + ``get_batch_api()`` to reload config and pick up the fresh on-disk token. + + Call this from ``except ApiException`` handlers with ``e.status``. + + Returns: + True when a reset was performed. + """ + global _last_unauthorized_reset + + if status != 401: + return False + + now = time.monotonic() + if now - _last_unauthorized_reset < _UNAUTHORIZED_RESET_INTERVAL_SECONDS: + return False + _last_unauthorized_reset = now + + logger.warning( + "Kubernetes API returned 401 Unauthorized — resetting cached clients to reload credentials", + ) + reset_clients() + return True + + def get_core_api() -> CoreV1Api | None: """Get the Core V1 API client for pod operations.""" core, _ = get_kubernetes_client() diff --git a/src/services/kubernetes/job_executor.py b/src/services/kubernetes/job_executor.py index ac51000..ce8ea62 100644 --- a/src/services/kubernetes/job_executor.py +++ b/src/services/kubernetes/job_executor.py @@ -19,6 +19,7 @@ get_batch_api, get_core_api, get_current_namespace, + handle_unauthorized, ) from .models import ( ExecutionResult, @@ -155,6 +156,7 @@ async def create_job( ) except ApiException as e: + handle_unauthorized(e.status) logger.error( "Failed to create job", job_name=job_name, @@ -225,6 +227,7 @@ async def wait_for_pod_ready( return False except ApiException as e: + handle_unauthorized(e.status) logger.warning( "Error checking pod status", job_name=job.name, @@ -397,6 +400,7 @@ async def delete_job(self, job: JobHandle): logger.debug("Deleted job", job_name=job.name) except ApiException as e: + handle_unauthorized(e.status) if e.status != 404: logger.warning( "Failed to delete job", diff --git a/src/services/kubernetes/pool.py b/src/services/kubernetes/pool.py index 831d52a..0e235e8 100644 --- a/src/services/kubernetes/pool.py +++ b/src/services/kubernetes/pool.py @@ -22,6 +22,7 @@ get_core_api, get_current_namespace, get_initialization_error, + handle_unauthorized, ) from .models import ( ExecutionResult, @@ -414,6 +415,7 @@ async def _create_warm_pod(self) -> PooledPod | None: return pooled_pod except ApiException as e: + handle_unauthorized(e.status) logger.error( "Failed to create warm pod (Kubernetes API error)", pod_name=pod_name, @@ -559,6 +561,7 @@ async def _delete_pod(self, handle: PodHandle): logger.debug("Deleted pod", pod_name=handle.name) except ApiException as e: + handle_unauthorized(e.status) if e.status != 404: logger.warning( "Failed to delete pod", @@ -936,6 +939,7 @@ async def _inspect_pod_failure(self, handle: PodHandle) -> str | None: lambda: core_api.read_namespaced_pod(handle.name, handle.namespace), ) except ApiException as e: + handle_unauthorized(e.status) if e.status == 404: return "pod not found (deleted or restarted)" return None diff --git a/tests/unit/test_kubernetes_client_auth.py b/tests/unit/test_kubernetes_client_auth.py new file mode 100644 index 0000000..d1d6e08 --- /dev/null +++ b/tests/unit/test_kubernetes_client_auth.py @@ -0,0 +1,123 @@ +"""Tests for Kubernetes client auth resilience (issue #72). + +Two layers of protection are covered here: + +1. A regression test against the installed ``kubernetes`` client library: + ``load_incluster_config``'s loader must actually wire the ServiceAccount + token into ``Configuration.auth_settings()`` and serve a *fresh* token + after kubelet rotation. kubernetes-client 36.0.0 broke this (the + regenerated ``auth_settings()`` reads ``api_key['BearerToken']`` while the + in-cluster loader writes ``api_key['authorization']``, so requests went + out with no Authorization header at all). This test fails on any future + dependency bump that reintroduces that class of regression. + +2. Unit tests for the throttled 401 recovery in ``client.py``: + ``handle_unauthorized`` resets the cached API clients so the next call + reloads config (and the rotated on-disk token) instead of failing forever. +""" + +import datetime + +import pytest + +from src.services.kubernetes import client as k8s_client + + +class TestInClusterAuthRegression: + """Guard against kubernetes-client in-cluster auth regressions.""" + + @pytest.fixture + def incluster_config(self, tmp_path): + from kubernetes.client import Configuration + from kubernetes.config import incluster_config + + token_file = tmp_path / "token" + cert_file = tmp_path / "ca.crt" + token_file.write_text("TOKEN_V1") + cert_file.write_text("dummy-ca") + + loader = incluster_config.InClusterConfigLoader( + token_filename=str(token_file), + cert_filename=str(cert_file), + try_refresh_token=True, + environ={ + "KUBERNETES_SERVICE_HOST": "1.2.3.4", + "KUBERNETES_SERVICE_PORT": "443", + }, + ) + configuration = Configuration() + loader.load_and_set(configuration) + return loader, configuration, token_file + + def test_incluster_loader_populates_auth_settings(self, incluster_config): + """auth_settings() must expose the SA token as an Authorization header. + + On kubernetes-client 36.0.0 this returns {} — every in-cluster request + is sent unauthenticated and the API server answers 401. + """ + _, configuration, _ = incluster_config + auth = configuration.auth_settings() + + assert "BearerToken" in auth, ( + "In-cluster config produced no BearerToken auth setting — the " + "kubernetes client would send unauthenticated requests (issue #72). " + "Check the pinned kubernetes-client version." + ) + assert auth["BearerToken"]["in"] == "header" + assert auth["BearerToken"]["key"] == "authorization" + assert auth["BearerToken"]["value"] == "bearer TOKEN_V1" + + def test_incluster_token_refreshes_after_rotation(self, incluster_config): + """A kubelet-rotated token must be picked up once the cached one expires.""" + loader, configuration, token_file = incluster_config + + token_file.write_text("TOKEN_V2") + loader.token_expires_at = datetime.datetime.now() - datetime.timedelta(seconds=1) + + auth = configuration.auth_settings() + assert auth["BearerToken"]["value"] == "bearer TOKEN_V2" + + +class TestHandleUnauthorized: + """Throttled client reset on 401 responses.""" + + @pytest.fixture(autouse=True) + def _reset_module_state(self, monkeypatch): + monkeypatch.setattr(k8s_client, "_core_api", object()) + monkeypatch.setattr(k8s_client, "_batch_api", object()) + monkeypatch.setattr(k8s_client, "_initialized", True) + monkeypatch.setattr(k8s_client, "_last_unauthorized_reset", 0.0) + + def test_401_resets_cached_clients(self): + assert k8s_client.handle_unauthorized(401) is True + assert k8s_client._core_api is None + assert k8s_client._batch_api is None + assert k8s_client._initialized is False + + def test_non_401_does_not_reset(self): + for status in (None, 403, 404, 500): + assert k8s_client.handle_unauthorized(status) is False + assert k8s_client._initialized is True + assert k8s_client._core_api is not None + + def test_reset_is_throttled(self, monkeypatch): + assert k8s_client.handle_unauthorized(401) is True + + # A second 401 right after the reset must not thrash config loading. + monkeypatch.setattr(k8s_client, "_core_api", object()) + monkeypatch.setattr(k8s_client, "_initialized", True) + assert k8s_client.handle_unauthorized(401) is False + assert k8s_client._initialized is True + + def test_reset_allowed_after_throttle_window(self, monkeypatch): + assert k8s_client.handle_unauthorized(401) is True + + monkeypatch.setattr(k8s_client, "_core_api", object()) + monkeypatch.setattr(k8s_client, "_initialized", True) + monkeypatch.setattr( + k8s_client, + "_last_unauthorized_reset", + k8s_client._last_unauthorized_reset - k8s_client._UNAUTHORIZED_RESET_INTERVAL_SECONDS - 1, + ) + assert k8s_client.handle_unauthorized(401) is True + assert k8s_client._initialized is False diff --git a/uv.lock b/uv.lock index 8e9f8a0..7cfa41c 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 2 requires-python = ">=3.13" resolution-markers = [ "python_full_version >= '3.14'", @@ -903,7 +903,7 @@ requires-dist = [ { name = "docker", specifier = ">=7.1.0" }, { name = "fastapi", specifier = ">=0.128.0" }, { name = "httpx", specifier = ">=0.28.1" }, - { name = "kubernetes", specifier = ">=34.1.0" }, + { name = "kubernetes", specifier = ">=34.1.0,<36" }, { name = "minio", specifier = ">=7.2.20" }, { name = "pyasn1", specifier = ">=0.6.3" }, { name = "pydantic", specifier = ">=2.12.5" }, @@ -937,10 +937,9 @@ dev = [ [[package]] name = "kubernetes" -version = "36.0.0" +version = "35.0.0" source = { registry = "https://pypi.org/simple/" } dependencies = [ - { name = "aiohttp" }, { name = "certifi" }, { name = "durationpy" }, { name = "python-dateutil" }, @@ -951,9 +950,9 @@ dependencies = [ { name = "urllib3" }, { name = "websocket-client" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/bf/59/dc635e4e9afb3884bc5c57f14fe23783e4c04601aa20b835ac75c41d1625/kubernetes-36.0.0.tar.gz", hash = "sha256:027b606bb8032e6c6464a53236bdd9bd9a94c237e1063bc45a303c25b304ced9", size = 2346728, upload-time = "2026-05-20T20:44:24.28Z" } +sdist = { url = "https://files.pythonhosted.org/packages/2c/8f/85bf51ad4150f64e8c665daf0d9dfe9787ae92005efb9a4d1cba592bd79d/kubernetes-35.0.0.tar.gz", hash = "sha256:3d00d344944239821458b9efd484d6df9f011da367ecb155dadf9513f05f09ee", size = 1094642, upload-time = "2026-01-16T01:05:27.76Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/cd/d2/6f99ca9c7eb961dfdd45b9643101399a8ee20922c662c362c91e9cc7e832/kubernetes-36.0.0-py2.py3-none-any.whl", hash = "sha256:a766433357ec9f90db7565cccf52e28e7fca40b0ef366c80a6022adbc0ac0425", size = 4660469, upload-time = "2026-05-20T20:44:20.893Z" }, + { url = "https://files.pythonhosted.org/packages/0c/70/05b685ea2dffcb2adbf3cdcea5d8865b7bc66f67249084cf845012a0ff13/kubernetes-35.0.0-py2.py3-none-any.whl", hash = "sha256:39e2b33b46e5834ef6c3985ebfe2047ab39135d41de51ce7641a7ca5b372a13d", size = 2017602, upload-time = "2026-01-16T01:05:25.991Z" }, ] [[package]]