Require explicit isolated Kubernetes connections and reject credential plugins (976f6583) · Commits · OOP / code / transformation-function-sdk

README.md

+32 −20

Changes for README.md: 32 added lines, 20 removed lines.
Original line number	Diff line number	Diff line
		@@ -188,8 +188,8 @@ NEF ->> 5GS: QoS session creation
		Apache 2.0 License – see [`LICENSE`](LICENSE) file for details.
		# Kubernetes control-path connections

		Supply the selected control path's connection in the infrastructure/config
		dictionary passed to `build_adapter("kubernetes", config)`. For example:
		Configure each adapter explicitly. File mode requires an absolute mounted file,
		its context and a namespace:

		```python
		adapter = build_adapter("kubernetes", {
		@@ -200,21 +200,33 @@ adapter = build_adapter("kubernetes", {
		})
		```

		Helm lifecycle operations and Kubernetes Service reads use that same connection.
		`api_server` and `tls_verify`, when supplied, override the kubeconfig consistently
		for both clients. Construction performs no file or network access; an invalid
		explicit kubeconfig fails before Helm contacts a backend and never selects
		another cluster. Kubeconfigs are loaded without persisting credential updates.

		Use `in_cluster: True` for the current Pod's service account, without a
		kubeconfig path or context. Helm receives that identity through its child
		environment; no token is placed in command arguments. Operator tooling must
		resolve any `kubeconfig_ref` and provide the mounted `kubeconfig_path`; the SDK
		does not implement a `secret://` resolver. Configuration and credentials stay
		outside topology models and responses.

		For compatibility, omitted connection settings capture KUBECONFIG,
		HELM_KUBECONTEXT or the in-cluster identity environment when constructing the
		adapter. Later environment changes do not redirect it. Explicit per-path
		configuration is required to distinguish clusters. The adapter does not mutate
		the process environment or the Kubernetes client's global default configuration.
		Alternatively use `in_cluster: True` with an explicit namespace for the Pod's
		service account. Missing/mixed selection, `api_server` overrides and unresolved
		`kubeconfig_ref` fail validation. `endpoint_host` addresses the deployed
		application; it is independent of the Kubernetes API server in the kubeconfig.
		`tls_verify`, when supplied, applies consistently to both transports.

		Construction performs no credential/file/network I/O. On first use, the SDK
		selects and pins one context/cluster/user profile. Helm receives a private
		snapshot of that validated profile; Kubernetes observation uses the same
		profile. Source kubeconfig changes require a new instance. Token files can
		rotate without changing the selected backend. Token/token-file and client-
		certificate/key credentials are supported; selected-user `exec`, `auth-provider`,
		basic authentication and kubeconfig proxy modes are rejected before either
		transport runs. Unselected credential providers are never loaded.

		Mount a writable `runtime_directory` (default `/var/run/openop-tfsdk`) for
		private execution files. Helm snapshots are removed after each command.
		The pinned Python client may retain temporary decoded certificate files until
		process exit. Do not put this directory in a persistent application catalog.
		The SDK never mutates process environment or global Kubernetes defaults.

		Helm uses a controlled child environment, the `secret` storage driver and
		isolated config/cache/data/plugin directories. Ambient KUBECONFIG, HELM_KUBE*,
		cloud profiles, proxies and plugin settings are not inherited. An absolute
		`helm_binary` may select the installed tool; the default `helm` uses the fixed
		system PATH. `helm_environment` accepts only absolute `HELM_REGISTRY_CONFIG`,
		`HELM_REPOSITORY_CONFIG` and `HELM_REPOSITORY_CACHE` paths for deliberately
		mounted artifact configuration. Proxy modes require separate qualification.
		Operator tooling owns credential resolution and mounts; no secret resolver
		is implemented in the SDK.

src/openop_tfsdk/contract/adapters/kubernetes.py

+5 −2

Changes for src/openop_tfsdk/contract/adapters/kubernetes.py: 5 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -55,9 +55,12 @@ CONTAINER_CHART = str(Path(__file__).parent / "charts" / "container")

		def _helm(args: list, connection: KubernetesConnection):
		try:
		options, environment = connection.helm_options()
		with connection.helm_options() as (options, environment):
		return subprocess.run(
		["helm", args, options], env=environment, capture_output=True, text=True
		[connection.helm_binary, args, options],
		env=environment,
		capture_output=True,
		text=True,
		)
		except ConfigException as error:
		raise DomainExecutionFailed(

src/openop_tfsdk/contract/adapters/kubernetes_connection.py

+188 −75

Changes for src/openop_tfsdk/contract/adapters/kubernetes_connection.py: 188 added lines, 75 removed lines.
Original line number	Diff line number	Diff line
		"""Connection settings belong to one adapter, shared by Helm and API reads."""
		"""Explicit per-instance connection shared by Helm and Kubernetes observation."""

		import copy
		import os
		import tempfile
		from contextlib import contextmanager
		from pathlib import Path
		from threading import Lock

		import yaml
		from kubernetes import client, config
		@@ -9,103 +13,212 @@ from kubernetes.config.config_exception import ConfigException

		from openop_tfsdk.contract.errors import ValidationError

		# Artifact configuration may be mounted explicitly; backend/provider overrides may not.
		_HELM_FILES = frozenset({"HELM_REGISTRY_CONFIG", "HELM_REPOSITORY_CONFIG", "HELM_REPOSITORY_CACHE"})
		_EXECUTION_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"


		def _nonempty(value):
		return isinstance(value, str) and bool(value.strip())


		def _entry(entries, name):
		matches = [
		entry for entry in entries or [] if isinstance(entry, dict) and entry.get("name") == name
		]
		if not _nonempty(name) or len(matches) != 1:
		raise ConfigException("Kubernetes control-path reference is missing or ambiguous")
		return matches[0]


		class KubernetesConnection:
		def __init__(self, settings: dict):
		# Compatibility defaults are captured once; execution never reads os.environ.
		self._environment = dict(os.environ)
		path = settings.get("kubeconfig_path")
		context = settings.get("kube_context")
		if any(
		value is not None and (not isinstance(value, str) or not value.strip())
		for value in (path, context)
		):
		self.path = settings.get("kubeconfig_path")
		self.context = settings.get("kube_context")
		self.namespace = settings.get("namespace")
		self.in_cluster = settings.get("in_cluster", False)
		if not _nonempty(self.namespace):
		raise ValidationError("Kubernetes namespace must be explicit and nonempty")
		if "api_server" in settings:
		raise ValidationError(
		"Kubernetes configuration path and context must be nonempty strings"
		"api_server overrides are unsupported; configure the selected kubeconfig"
		)
		if settings.get("kubeconfig_ref") and not path:
		if settings.get("kubeconfig_ref"):
		raise ValidationError(
		"Resolve kubeconfig_ref to kubeconfig_path before building the adapter"
		"Resolve kubeconfig_ref to a mounted kubeconfig_path before construction"
		)
		self.in_cluster = settings.get(
		"in_cluster",
		not (path or context or self._environment.get("KUBECONFIG"))
		and bool(self._environment.get("KUBERNETES_SERVICE_HOST")),
		)
		if not isinstance(self.in_cluster, bool) or (self.in_cluster and (path or context)):
		if not isinstance(self.in_cluster, bool):
		raise ValidationError("in_cluster must be a boolean")
		if self.in_cluster:
		if self.path is not None or self.context is not None:
		raise ValidationError("Choose in_cluster or kubeconfig_path/kube_context")
		self._explicit = bool(path or context or settings.get("api_server"))
		self.path = (
		None
		if self.in_cluster
		else (path or self._environment.get("KUBECONFIG") or str(Path.home() / ".kube/config"))
		)
		self.context = (
		None
		if self.in_cluster
		else (context if path else (context or self._environment.get("HELM_KUBECONTEXT")))
		)
		self.server = settings.get("api_server")
		if self.server is not None and (
		not isinstance(self.server, str) or not self.server.startswith(("https://", "http://"))
		elif (
		not _nonempty(self.path)
		or not Path(self.path).is_absolute()
		or os.pathsep in self.path
		or not _nonempty(self.context)
		):
		raise ValidationError("api_server must be an HTTP or HTTPS URL")
		raise ValidationError(
		"Choose an absolute kubeconfig_path and kube_context, or in_cluster=true"
		)
		self.tls_verify = settings.get("tls_verify")
		if self.tls_verify is not None and not isinstance(self.tls_verify, bool):
		raise ValidationError("tls_verify must be a boolean")
		self.runtime_directory = settings.get("runtime_directory", "/var/run/openop-tfsdk")
		if not _nonempty(self.runtime_directory) or not Path(self.runtime_directory).is_absolute():
		raise ValidationError("runtime_directory must be an absolute writable directory")
		self.helm_binary = settings.get("helm_binary", "helm")
		if self.helm_binary != "helm" and (
		not _nonempty(self.helm_binary) or not Path(self.helm_binary).is_absolute()
		):
		raise ValidationError(
		"helm_binary must be helm or an explicit absolute executable path"
		)
		environment = settings.get("helm_environment") or {}
		if not isinstance(environment, dict) or any(
		key not in _HELM_FILES or not _nonempty(value) or not Path(value).is_absolute()
		for key, value in environment.items()
		):
		raise ValidationError(
		"helm_environment accepts only explicit Helm registry/repository file paths"
		)
		self._helm_files = dict(environment)
		# Only explicitly selected in-cluster mode consumes these Pod runtime facts.
		self._service_host = os.environ.get("KUBERNETES_SERVICE_HOST") if self.in_cluster else None
		self._service_port = os.environ.get("KUBERNETES_SERVICE_PORT") if self.in_cluster else None
		self._token_file = config.incluster_config.SERVICE_TOKEN_FILENAME
		self._ca_file = config.incluster_config.SERVICE_CERT_FILENAME
		self._selected = None
		self._lock = Lock()

		def configuration(self):
		configuration = client.Configuration()
		def _profile(self):
		with self._lock:
		if self._selected is None:
		try:
		self._selected = self._load_selected_profile()
		except (OSError, ValueError, TypeError, KeyError, yaml.YAMLError) as error:
		raise ConfigException(
		"Kubernetes control-path configuration could not be loaded"
		) from error
		return copy.deepcopy(self._selected)

		def _load_selected_profile(self):
		if self.in_cluster:
		# The pinned client's loader accepts an explicit environment snapshot.
		config.incluster_config.InClusterConfigLoader(
		token_filename=config.incluster_config.SERVICE_TOKEN_FILENAME,
		cert_filename=config.incluster_config.SERVICE_CERT_FILENAME,
		environ=self._environment,
		).load_and_set(configuration)
		if not _nonempty(self._service_host) or not _nonempty(self._service_port):
		raise ConfigException(
		"Explicit in-cluster mode requires Kubernetes Pod connection settings"
		)
		host = self._service_host
		if ":" in host and not host.startswith("["):
		host = "[" + host + "]"
		cluster = {
		"server": f"https://{host}:{self._service_port}",
		"certificate-authority": self._ca_file,
		}
		user = {"tokenFile": self._token_file}
		context = "in-cluster"
		else:
		config.load_kube_config(
		config_file=self.path,
		context=self.context,
		client_configuration=configuration,
		persist_config=False,
		with open(self.path, encoding="utf-8") as handle:
		source = yaml.safe_load(handle)
		if not isinstance(source, dict):
		raise ConfigException("Kubernetes kubeconfig must be a mapping")
		selected = _entry(source.get("contexts"), self.context)["context"]
		if not isinstance(selected, dict):
		raise ConfigException("Selected Kubernetes context must be a mapping")
		cluster = copy.deepcopy(
		_entry(source.get("clusters"), selected.get("cluster"))["cluster"]
		)
		user_ref = selected.get("user")
		user = copy.deepcopy(_entry(source.get("users"), user_ref)["user"]) if user_ref else {}
		context = self.context
		if not isinstance(cluster, dict) or not isinstance(user, dict):
		raise ConfigException("Selected Kubernetes cluster and user must be mappings")
		if "exec" in user or "auth-provider" in user:
		raise ValidationError(
		"External Kubernetes credential providers are unsupported in this release"
		)
		except (OSError, ValueError, yaml.YAMLError) as error:
		if "proxy-url" in cluster or "username" in user or "password" in user:
		raise ValidationError(
		"Use supported static/file credentials without a kubeconfig proxy"
		)
		base = Path(self.path).parent
		for mapping, keys in (
		(cluster, ("certificate-authority",)),
		(user, ("client-certificate", "client-key", "tokenFile")),
		):
		for key in keys:
		if mapping.get(key):
		mapping[key] = str(base / mapping[key])
		server = cluster.get("server")
		if not isinstance(server, str) or not server.startswith(("https://", "http://")):
		raise ConfigException("Selected Kubernetes cluster requires an HTTP or HTTPS server")
		if self.tls_verify is not None:
		cluster["insecure-skip-tls-verify"] = not self.tls_verify
		return {
		"apiVersion": "v1",
		"kind": "Config",
		"current-context": context,
		"clusters": [{"name": "selected", "cluster": cluster}],
		"users": [{"name": "selected", "user": user}],
		"contexts": [
		{
		"name": context,
		"context": {
		"cluster": "selected",
		"user": "selected",
		"namespace": self.namespace,
		},
		}
		],
		}

		def configuration(self):
		configuration = client.Configuration()
		try:
		config.kube_config.KubeConfigLoader(
		config_dict=self._profile(),
		active_context=self.context if not self.in_cluster else "in-cluster",
		temp_file_path=self.runtime_directory,
		).load_and_set(configuration)
		except (OSError, ValueError, TypeError, yaml.YAMLError) as error:
		raise ConfigException(
		"Kubernetes control-path configuration could not be loaded"
		) from error
		if self.server:
		configuration.host = self.server
		if self.tls_verify is not None:
		configuration.verify_ssl = self.tls_verify
		configuration.retries = 0
		return configuration

		@contextmanager
		def helm_options(self):
		# Load the vetted profile, not the source kubeconfig a second time. Neither
		# client sees unselected users, ambient defaults or executable providers.
		profile = self._profile()
		self.configuration() # Fail locally on missing credential/CA files before Helm.
		try:
		temporary = tempfile.TemporaryDirectory(prefix="helm-", dir=self.runtime_directory)
		except OSError as error:
		raise ConfigException("Kubernetes execution directory could not be prepared") from error
		with temporary as directory:
		root = Path(directory)
		snapshot = root / "kubeconfig"
		try:
		with snapshot.open("x", encoding="utf-8") as handle:
		os.chmod(snapshot, 0o600)
		yaml.safe_dump(profile, handle)
		except OSError as error:
		raise ConfigException("Kubernetes execution files could not be prepared") from error
		environment = {
		key: value
		for key, value in self._environment.items()
		if key != "KUBECONFIG" and not key.startswith("HELM_KUBE")
		"PATH": _EXECUTION_PATH,
		"HOME": directory,
		"HELM_DRIVER": "secret",
		"HELM_CONFIG_HOME": str(root / "config"),
		"HELM_CACHE_HOME": str(root / "cache"),
		"HELM_DATA_HOME": str(root / "data"),
		"HELM_PLUGINS": str(root / "plugins"),
		**self._helm_files,
		}
		args = ["--kubeconfig", self.path or os.devnull]
		if self.path and os.pathsep in self.path:
		# Helm merges KUBECONFIG lists only through its child environment.
		environment["KUBECONFIG"] = self.path
		args = []
		if self.context:
		args += ["--kube-context", self.context]
		# Validate explicit credentials locally before Helm can contact a backend.
		if self.in_cluster or self._explicit:
		configuration = self.configuration()
		if self.in_cluster:
		environment.update(
		HELM_KUBEAPISERVER=configuration.host,
		HELM_KUBECAFILE=configuration.ssl_ca_cert,
		HELM_KUBETOKEN=configuration.api_key["authorization"].split(" ", 1)[-1],
		)
		if self.server:
		args += ["--kube-apiserver", self.server]
		if self.tls_verify is not None:
		args += ["--kube-insecure-skip-tls-verify=" + str(not self.tls_verify).lower()]
		return args, environment
		yield [
		"--kubeconfig",
		str(snapshot),
		"--kube-context",
		profile["current-context"],
		], environment

tests/contract/conftest.py

0 → 100644

+27 −0

Changes for tests/contract/conftest.py: 27 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""Explicit local connection fixtures; tests never use a developer's cluster."""

		import pytest
		import yaml


		@pytest.fixture
		def kubernetes_settings(tmp_path):
		path = tmp_path / "kubeconfig"
		path.write_text(
		yaml.safe_dump(
		{
		"apiVersion": "v1",
		"kind": "Config",
		"current-context": "test",
		"clusters": [{"name": "test", "cluster": {"server": "https://test.invalid"}}],
		"users": [{"name": "test", "user": {"token": "fixture-token"}}],
		"contexts": [{"name": "test", "context": {"cluster": "test", "user": "test"}}],
		}
		)
		)
		return {
		"kubeconfig_path": str(path),
		"kube_context": "test",
		"namespace": "default",
		"runtime_directory": str(tmp_path),
		}

tests/contract/test_kubernetes_container.py

+26 −16

Changes for tests/contract/test_kubernetes_container.py: 26 added lines, 16 removed lines.
Original line number	Diff line number	Diff line
		@@ -63,13 +63,15 @@ def test_bundled_chart_ships_with_the_adapter():
		"visibility, service_type", [("external", "NodePort"), ("internal", "ClusterIP")]
		)
		def test_container_runs_the_image_from_the_bundled_chart(
		helm_calls, action, visibility, service_type
		kubernetes_settings, helm_calls, action, visibility, service_type
		):
		interfaces = [
		_interface(80, visibility=visibility),
		_interface(9000, "udp", visibility=visibility),
		]
		getattr(KubernetesCapabilityAdapter(), action)(_request("container", interfaces=interfaces))
		getattr(KubernetesCapabilityAdapter(kubernetes_settings), action)(
		_request("container", interfaces=interfaces)
		)
		assert helm_calls[0][3] == CONTAINER_CHART
		assert _values(helm_calls[0]) == {
		"image": "nginx:1.25",
		@@ -81,8 +83,8 @@ def test_container_runs_the_image_from_the_bundled_chart(
		}


		def test_container_without_ports_gets_no_external_service(helm_calls):
		KubernetesCapabilityAdapter().activate(_request("container"))
		def test_container_without_ports_gets_no_external_service(kubernetes_settings, helm_calls):
		KubernetesCapabilityAdapter(kubernetes_settings).activate(_request("container"))
		assert _values(helm_calls[0]) == {
		"image": "nginx:1.25",
		"ports": [],
		@@ -90,8 +92,8 @@ def test_container_without_ports_gets_no_external_service(helm_calls):
		}


		def test_helm_unit_still_installs_its_own_chart(helm_calls):
		KubernetesCapabilityAdapter().activate(
		def test_helm_unit_still_installs_its_own_chart(kubernetes_settings, helm_calls):
		KubernetesCapabilityAdapter(kubernetes_settings).activate(
		_request("helm", artifact_ref="oci://registry/video:1.0")
		)
		assert helm_calls[0][3] == "oci://registry/video:1.0"
		@@ -102,17 +104,25 @@ def test_helm_unit_still_installs_its_own_chart(helm_calls):
		"runtime_type, interfaces",
		[("k8s_manifest", None), ("container", [_interface(7000, "custom")])],
		)
		def test_unsupported_unit_is_rejected_before_helm(helm_calls, runtime_type, interfaces):
		def test_unsupported_unit_is_rejected_before_helm(
		kubernetes_settings, helm_calls, runtime_type, interfaces
		):
		with pytest.raises(ValidationError):
		KubernetesCapabilityAdapter().activate(_request(runtime_type, interfaces=interfaces))
		KubernetesCapabilityAdapter(kubernetes_settings).activate(
		_request(runtime_type, interfaces=interfaces)
		)
		assert helm_calls == []


		@pytest.mark.parametrize("action", ["activate", "update"])
		def test_mixed_visibility_is_rejected_before_exposing_internal_ports(helm_calls, action):
		def test_mixed_visibility_is_rejected_before_exposing_internal_ports(
		kubernetes_settings, helm_calls, action
		):
		interfaces = [_interface(80, visibility="external"), _interface(9000, "udp")]
		with pytest.raises(ValidationError, match="mixed internal and external"):
		getattr(KubernetesCapabilityAdapter(), action)(_request("container", interfaces=interfaces))
		getattr(KubernetesCapabilityAdapter(kubernetes_settings), action)(
		_request("container", interfaces=interfaces)
		)
		assert helm_calls == []


		@@ -141,18 +151,18 @@ def _service():
		)


		def test_container_reports_its_node_ports_as_endpoints(monkeypatch):
		def test_container_reports_its_node_ports_as_endpoints(kubernetes_settings, monkeypatch):
		calls = _cluster(monkeypatch, _service)
		request = _request("container", interfaces=[_interface(80, visibility="external")])
		result = KubernetesCapabilityAdapter(HOST).activate(request)
		result = KubernetesCapabilityAdapter({kubernetes_settings, HOST}).activate(request)
		assert calls[1] == ["service", "video-app", "default"]
		assert result.endpoints == [ServiceEndpoint(interface_id="api", fqdn="192.0.2.10", port=31234)]


		def test_no_endpoints_without_a_configured_host(monkeypatch):
		def test_no_endpoints_without_a_configured_host(kubernetes_settings, monkeypatch):
		calls = _cluster(monkeypatch, _service)
		request = _request("container", interfaces=[_interface(80, visibility="external")])
		assert KubernetesCapabilityAdapter().activate(request).endpoints is None
		assert KubernetesCapabilityAdapter(kubernetes_settings).activate(request).endpoints is None
		assert len(calls) == 1


		@@ -168,9 +178,9 @@ def _forbidden_service():
		"observation",
		[_forbidden_service, _unavailable_service],
		)
		def test_unreadable_service_still_activates(monkeypatch, observation):
		def test_unreadable_service_still_activates(kubernetes_settings, monkeypatch, observation):
		_cluster(monkeypatch, observation)
		request = _request("container", interfaces=[_interface(80, visibility="external")])
		result = KubernetesCapabilityAdapter(HOST).activate(request)
		result = KubernetesCapabilityAdapter({kubernetes_settings, HOST}).activate(request)
		assert result.state == LifecycleState.ACTIVE
		assert result.endpoints is None