Coverage for cli / kubectl_helpers.py: 100.00%
97 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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, capture_output=True, text=True
127 ) # nosemgrep: dangerous-subprocess-use-audit - inputs validated above; list form, no shell=True
128 if result.returncode != 0:
129 raise RuntimeError(f"Failed to update kubeconfig: {result.stderr}")
130 except subprocess.CalledProcessError as e:
131 raise RuntimeError(f"Failed to update kubeconfig: {e.stderr}") from e
132 except FileNotFoundError as e:
133 raise RuntimeError(
134 "AWS CLI not found. Please install the AWS CLI and ensure it's in your PATH."
135 ) from e
138# ---------------------------------------------------------------------------
139# Port-forward + endpoint helpers (used by `gco monitoring open`)
140# ---------------------------------------------------------------------------
142# svc/name | service/name | pod/name | deploy/name | deployment/name, where the
143# resource name follows the RFC 1123 rules kubectl accepts.
144_PF_TARGET_RE = re.compile(
145 r"^(svc|service|pod|deploy|deployment)/[a-z0-9]([a-z0-9.\-]{0,251}[a-z0-9])?$"
146)
147_NAMESPACE_RE = re.compile(r"^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?$")
150def _validate_port(port: int | str, *, what: str = "port") -> int:
151 """Return the port as an int in 1..65535 or raise ValueError."""
152 try:
153 value = int(port)
154 except (TypeError, ValueError) as exc:
155 raise ValueError(f"Invalid {what} {port!r}: must be an integer") from exc
156 if not 1 <= value <= 65535:
157 raise ValueError(f"Invalid {what} {value}: must be between 1 and 65535")
158 return value
161def build_port_forward_command(
162 namespace: str,
163 target: str,
164 local_port: int | str,
165 remote_port: int | str,
166 *,
167 server: str | None = None,
168 tls_server_name: str | None = None,
169) -> list[str]:
170 """Build a validated ``kubectl port-forward`` argv (list form, never a shell string).
172 ``target`` is a ``kind/name`` reference (e.g. ``svc/kube-prometheus-stack-grafana``).
173 ``server`` / ``tls_server_name`` override the API endpoint and its TLS SNI —
174 used when tunnelling to a private endpoint through an SSM local port, where
175 kubectl talks to ``https://127.0.0.1:<port>`` but must present the real EKS
176 hostname for certificate validation.
177 """
178 if not _NAMESPACE_RE.match(namespace):
179 raise ValueError(f"Invalid namespace {namespace!r}")
180 if not _PF_TARGET_RE.match(target):
181 raise ValueError(
182 f"Invalid port-forward target {target!r}: expected kind/name "
183 "(svc|service|pod|deploy|deployment)"
184 )
185 local = _validate_port(local_port, what="local port")
186 remote = _validate_port(remote_port, what="remote port")
188 cmd = ["kubectl", "port-forward", "-n", namespace, target, f"{local}:{remote}"]
189 if server is not None:
190 if not server.startswith("https://"):
191 raise ValueError(f"Invalid --server {server!r}: must start with https://")
192 cmd += ["--server", server]
193 if tls_server_name is not None:
194 if not re.match(r"^[a-zA-Z0-9.\-]{1,255}$", tls_server_name):
195 raise ValueError(f"Invalid --tls-server-name {tls_server_name!r}")
196 cmd += ["--tls-server-name", tls_server_name]
197 return cmd
200def describe_cluster_access(cluster_name: str, region: str) -> dict[str, object]:
201 """Return the EKS API endpoint and its public/private access posture.
203 Returns a dict with keys ``endpoint`` (str), ``public`` (bool),
204 ``private`` (bool), and ``public_cidrs`` (list[str]). Used by
205 ``gco monitoring open`` to decide whether a plain ``kubectl port-forward``
206 can reach the API server or whether an SSM/VPN/bastion path is required.
207 """
208 _validate_cluster_name(cluster_name)
209 _validate_region(region)
211 cmd = [
212 "aws",
213 "eks",
214 "describe-cluster",
215 "--name",
216 cluster_name,
217 "--region",
218 region,
219 "--query",
220 (
221 "cluster.{endpoint:endpoint,"
222 "public:resourcesVpcConfig.endpointPublicAccess,"
223 "private:resourcesVpcConfig.endpointPrivateAccess,"
224 "publicCidrs:resourcesVpcConfig.publicAccessCidrs}"
225 ),
226 "--output",
227 "json",
228 ]
229 try:
230 result = subprocess.run(
231 cmd, capture_output=True, text=True
232 ) # nosemgrep: dangerous-subprocess-use-audit - inputs validated above; list form, no shell=True
233 except FileNotFoundError as exc:
234 raise RuntimeError(
235 "AWS CLI not found. Please install the AWS CLI and ensure it's in your PATH."
236 ) from exc
237 if result.returncode != 0:
238 raise RuntimeError(f"Failed to describe cluster {cluster_name}: {result.stderr}")
240 import json
242 data = json.loads(result.stdout or "{}")
243 return {
244 "endpoint": data.get("endpoint") or "",
245 "public": bool(data.get("public")),
246 "private": bool(data.get("private")),
247 "public_cidrs": data.get("publicCidrs") or [],
248 }