Coverage for cli / kubectl_helpers.py: 100.00%
97 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Shared kubectl helper utilities for GCO CLI.
4Provides common kubectl operations used across multiple CLI modules
5to reduce code duplication and ensure consistent error handling.
6"""
8import logging
9import os
10import re
11import subprocess
12from pathlib import Path
13from urllib.parse import urlsplit
15logger = logging.getLogger(__name__)
17# EKS cluster names: 1-100 chars, alphanumeric and hyphens only.
18# AWS region names: e.g. us-east-1, ap-southeast-2, eu-central-1.
19_CLUSTER_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9\-]{0,99}$")
20_REGION_RE = re.compile(r"^[a-z]{2,4}(?:-[a-z0-9]+)+-[0-9]+$")
23def _validate_cluster_name(cluster_name: str) -> None:
24 """Raise ValueError if cluster_name contains characters outside the EKS naming rules."""
25 if not _CLUSTER_NAME_RE.match(cluster_name):
26 raise ValueError(
27 f"Invalid cluster name {cluster_name!r}: must be 1-100 alphanumeric/hyphen characters"
28 )
31def _validate_region(region: str) -> None:
32 """Raise ValueError if region does not match the standard AWS region pattern."""
33 if not _REGION_RE.match(region):
34 raise ValueError(f"Invalid AWS region {region!r}: expected format like 'us-east-1'")
37#: Hostnames that identify a local API-server tunnel in ``cluster.server``.
38_LOCAL_TUNNEL_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})
41def _kubeconfig_file() -> Path:
42 """The kubeconfig file ``aws eks update-kubeconfig`` would write."""
43 kubeconfig_env = os.environ.get("KUBECONFIG", "")
44 if kubeconfig_env:
45 first = kubeconfig_env.split(os.pathsep)[0]
46 if first:
47 return Path(first)
48 return Path.home() / ".kube" / "config"
51def _tunnel_pinned_server(cluster_name: str) -> str | None:
52 """Return the pinned local-tunnel server for *cluster_name*, if any.
54 ``gco cluster tunnel`` (and callers of its machinery, e.g. the example-job
55 validation harness) rewrite the cluster's kubeconfig entry to point at a
56 localhost tunnel with ``tls-server-name`` pinned to the real endpoint
57 host. Any entry matching that shape is a deliberate operator choice.
58 """
59 import yaml
61 path = _kubeconfig_file()
62 try:
63 config = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
64 except OSError, yaml.YAMLError:
65 return None
66 expected_suffix = f"cluster/{cluster_name}"
67 for entry in config.get("clusters", []) or []:
68 name = str(entry.get("name", ""))
69 if name != cluster_name and not name.endswith(expected_suffix):
70 continue
71 cluster = entry.get("cluster") or {}
72 server = str(cluster.get("server", ""))
73 host = urlsplit(server).hostname or ""
74 if host in _LOCAL_TUNNEL_HOSTS and cluster.get("tls-server-name"):
75 return server
76 return None
79def update_kubeconfig(cluster_name: str, region: str) -> None:
80 """Update kubeconfig for an EKS cluster, preserving an active tunnel pin.
82 When the cluster's kubeconfig entry already points at a localhost tunnel
83 (``gco cluster tunnel`` rewrites ``cluster.server`` to the tunnel and pins
84 ``tls-server-name`` to the real endpoint host), refreshing it with
85 ``aws eks update-kubeconfig`` would silently rewrite the server back to
86 the private endpoint — unreachable from outside the VPC — and break every
87 kubectl-wrapping command mid-session. Caught live by example-job
88 validation run ex241-66c02e71, where ``gco jobs submit-direct`` clobbered
89 the harness's tunnel and every subsequent kubectl call timed out against
90 the private endpoint. A tunnel-shaped entry is preserved untouched; a
91 dead tunnel fails loudly at connect time, which beats a silent rewrite.
93 Args:
94 cluster_name: Name of the EKS cluster
95 region: AWS region where the cluster is located
97 Raises:
98 ValueError: If cluster_name or region contain unexpected characters
99 RuntimeError: If the kubeconfig update fails
100 FileNotFoundError: If the AWS CLI is not installed
101 """
102 _validate_cluster_name(cluster_name)
103 _validate_region(region)
105 pinned = _tunnel_pinned_server(cluster_name)
106 if pinned is not None:
107 logger.info(
108 "kubeconfig for %s points at local tunnel %s; preserving it",
109 cluster_name,
110 pinned,
111 )
112 return
114 cmd = [
115 "aws",
116 "eks",
117 "update-kubeconfig",
118 "--name",
119 cluster_name,
120 "--region",
121 region,
122 ]
124 try:
125 result = subprocess.run(
126 cmd,
127 capture_output=True,
128 text=True,
129 check=False,
130 ) # nosemgrep: dangerous-subprocess-use-audit - inputs validated above; list form, no shell=True
131 if result.returncode != 0:
132 raise RuntimeError(f"Failed to update kubeconfig: {result.stderr}")
133 except subprocess.CalledProcessError as e:
134 raise RuntimeError(f"Failed to update kubeconfig: {e.stderr}") from e
135 except FileNotFoundError as e:
136 raise RuntimeError(
137 "AWS CLI not found. Please install the AWS CLI and ensure it's in your PATH."
138 ) from e
141# ---------------------------------------------------------------------------
142# Port-forward + endpoint helpers (used by `gco monitoring open`)
143# ---------------------------------------------------------------------------
145# svc/name | service/name | pod/name | deploy/name | deployment/name, where the
146# resource name follows the RFC 1123 rules kubectl accepts.
147_PF_TARGET_RE = re.compile(
148 r"^(svc|service|pod|deploy|deployment)/[a-z0-9]([a-z0-9.\-]{0,251}[a-z0-9])?$"
149)
150_NAMESPACE_RE = re.compile(r"^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?$")
153def _validate_port(port: int | str, *, what: str = "port") -> int:
154 """Return the port as an int in 1..65535 or raise ValueError."""
155 try:
156 value = int(port)
157 except (TypeError, ValueError) as exc:
158 raise ValueError(f"Invalid {what} {port!r}: must be an integer") from exc
159 if not 1 <= value <= 65535:
160 raise ValueError(f"Invalid {what} {value}: must be between 1 and 65535")
161 return value
164def build_port_forward_command(
165 namespace: str,
166 target: str,
167 local_port: int | str,
168 remote_port: int | str,
169 *,
170 server: str | None = None,
171 tls_server_name: str | None = None,
172) -> list[str]:
173 """Build a validated ``kubectl port-forward`` argv (list form, never a shell string).
175 ``target`` is a ``kind/name`` reference (e.g. ``svc/kube-prometheus-stack-grafana``).
176 ``server`` / ``tls_server_name`` override the API endpoint and its TLS SNI —
177 used when tunnelling to a private endpoint through an SSM local port, where
178 kubectl talks to ``https://127.0.0.1:<port>`` but must present the real EKS
179 hostname for certificate validation.
180 """
181 if not _NAMESPACE_RE.match(namespace):
182 raise ValueError(f"Invalid namespace {namespace!r}")
183 if not _PF_TARGET_RE.match(target):
184 raise ValueError(
185 f"Invalid port-forward target {target!r}: expected kind/name "
186 "(svc|service|pod|deploy|deployment)"
187 )
188 local = _validate_port(local_port, what="local port")
189 remote = _validate_port(remote_port, what="remote port")
191 cmd = ["kubectl", "port-forward", "-n", namespace, target, f"{local}:{remote}"]
192 if server is not None:
193 if not server.startswith("https://"):
194 raise ValueError(f"Invalid --server {server!r}: must start with https://")
195 cmd += ["--server", server]
196 if tls_server_name is not None:
197 if not re.match(r"^[a-zA-Z0-9.\-]{1,255}$", tls_server_name):
198 raise ValueError(f"Invalid --tls-server-name {tls_server_name!r}")
199 cmd += ["--tls-server-name", tls_server_name]
200 return cmd
203def describe_cluster_access(cluster_name: str, region: str) -> dict[str, object]:
204 """Return the EKS API endpoint and its public/private access posture.
206 Returns a dict with keys ``endpoint`` (str), ``public`` (bool),
207 ``private`` (bool), and ``public_cidrs`` (list[str]). Used by
208 ``gco monitoring open`` to decide whether a plain ``kubectl port-forward``
209 can reach the API server or whether an SSM/VPN/bastion path is required.
210 """
211 _validate_cluster_name(cluster_name)
212 _validate_region(region)
214 cmd = [
215 "aws",
216 "eks",
217 "describe-cluster",
218 "--name",
219 cluster_name,
220 "--region",
221 region,
222 "--query",
223 (
224 "cluster.{endpoint:endpoint,"
225 "public:resourcesVpcConfig.endpointPublicAccess,"
226 "private:resourcesVpcConfig.endpointPrivateAccess,"
227 "publicCidrs:resourcesVpcConfig.publicAccessCidrs}"
228 ),
229 "--output",
230 "json",
231 ]
232 try:
233 result = subprocess.run(
234 cmd,
235 capture_output=True,
236 text=True,
237 check=False,
238 ) # nosemgrep: dangerous-subprocess-use-audit - inputs validated above; list form, no shell=True
239 except FileNotFoundError as exc:
240 raise RuntimeError(
241 "AWS CLI not found. Please install the AWS CLI and ensure it's in your PATH."
242 ) from exc
243 if result.returncode != 0:
244 raise RuntimeError(f"Failed to describe cluster {cluster_name}: {result.stderr}")
246 import json
248 data = json.loads(result.stdout or "{}")
249 return {
250 "endpoint": data.get("endpoint") or "",
251 "public": bool(data.get("public")),
252 "private": bool(data.get("private")),
253 "public_cidrs": data.get("publicCidrs") or [],
254 }