Coverage for cli / cluster_ui.py: 100.00%
171 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Shared plumbing for the in-cluster web UIs GCO reaches through kubectl.
3``gco gitops`` (Argo CD) and ``gco crossplane`` (the Crossview dashboard) sit
4on the same three steps ``gco monitoring open`` takes for Grafana — reach the
5(private by default) EKS API through :func:`cli.cluster_tunnel.open_api_server_tunnel`,
6``kubectl port-forward`` a ClusterIP Service to localhost, and point a browser
7or a headless Playwright page at it — plus two reads they need around it: a
8key out of a Kubernetes Secret (the Argo CD admin password) and the pinned
9chart entry from ``lambda/helm-installer/charts.yaml`` (for ``status``).
10:func:`exec_port_forward` is the foreground forward all four ``open``-style
11commands (``monitoring open``, ``costs dashboard``, ``gitops open``,
12``crossplane open``) share, so none announces a URL before kubectl listens.
14Everything here is list-form subprocess calls and pure helpers, so the command
15modules stay thin and the tests can drive them with fakes. Playwright is
16imported lazily so the CLI loads without the rendering extra installed.
17"""
19from __future__ import annotations
21import base64
22import json
23import re
24import socket
25import subprocess
26import threading
27import time
28from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
29from contextlib import contextmanager
30from pathlib import Path
31from typing import IO, Any, cast
33import click
34import yaml
36#: Viewport for UI captures (matches the monitoring screenshots).
37VIEWPORT_WIDTH = 1600
38VIEWPORT_HEIGHT = 900
40#: How long a capture waits after navigation for single-page apps that poll
41#: (network-idle never comes), mirroring the Grafana captures.
42RENDER_WAIT_MS = 5000
44#: How long a background port-forward may take to accept connections.
45PORT_FORWARD_READY_TIMEOUT_SECONDS = 30.0
47#: How long a foreground port-forward may take to report its listener. kubectl
48#: gives up on an unreachable API server by itself well inside this.
49FOREGROUND_PORT_FORWARD_READY_TIMEOUT_SECONDS = 120.0
51#: kubectl prints one of these per local address it binds, once the API server
52#: has accepted the port-forward (client-go tools/portforward).
53_FORWARDING_LINE = re.compile(r"^Forwarding from \S+:(?P<port>\d+) -> \d+")
55#: How often the foreground readiness wait re-checks kubectl (it stays
56#: responsive to Ctrl-C between checks).
57_READY_POLL_SECONDS = 0.2
59#: How long the output relay may take to drain once kubectl has exited.
60_RELAY_DRAIN_SECONDS = 2.0
62#: Default self-terminate backstop for an ``--via-ssm auto`` bastion (mirrors
63#: ``cli.ephemeral_bastion.DEFAULT_TTL_MINUTES``; kept literal so importing
64#: this module stays cheap).
65DEFAULT_BASTION_TTL_MINUTES = 120
67_INSTALL_HINT = (
68 "Playwright is not installed. Install the rendering extra and its browser: "
69 "pip install 'gco[diagrams]' && playwright install chromium"
70)
72_NAME_RE = re.compile(r"^[a-z0-9]([-a-z0-9.]*[a-z0-9])?$")
74#: charts.yaml as shipped next to this package in a source checkout.
75CHARTS_YAML = Path(__file__).resolve().parents[1] / "lambda" / "helm-installer" / "charts.yaml"
77#: The kubectl-applier manifest directory in a source checkout.
78MANIFESTS_DIR = (
79 Path(__file__).resolve().parents[1] / "lambda" / "kubectl-applier-simple" / "manifests"
80)
83def load_cdk_context(cdk_json_path: Path | None = None) -> dict[str, Any]:
84 """The ``context`` object of cdk.json (the working directory's by default).
86 Raises ``RuntimeError`` when cdk.json is missing or is not JSON; a
87 document without a ``context`` object reads as ``{}``.
88 """
89 path = cdk_json_path or Path.cwd() / "cdk.json"
90 if not path.exists():
91 raise RuntimeError(f"cdk.json not found at {path}")
92 try:
93 document = json.loads(path.read_text(encoding="utf-8"))
94 except json.JSONDecodeError as exc:
95 raise RuntimeError(f"cdk.json at {path} is not valid JSON: {exc}") from exc
96 context = document.get("context") if isinstance(document, dict) else None
97 return context if isinstance(context, dict) else {}
100def regional_regions(context: Mapping[str, Any]) -> list[str]:
101 """``deployment_regions.regional`` from a cdk.json context (``[]`` when absent)."""
102 regions = context.get("deployment_regions")
103 regional = regions.get("regional") if isinstance(regions, Mapping) else None
104 return [str(region) for region in regional] if isinstance(regional, list) else []
107def chart_entry(name: str, charts_yaml: Path | None = None) -> dict[str, Any] | None:
108 """The pinned ``charts.yaml`` entry (chart, version, namespace), or ``None``.
110 ``None`` when the file is not there (an installed CLI without the source
111 tree) or has no such chart; status commands then just omit the pin.
112 """
113 path = charts_yaml or CHARTS_YAML
114 try:
115 document = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
116 except OSError, yaml.YAMLError:
117 return None
118 charts = document.get("charts") if isinstance(document, dict) else None
119 entry = charts.get(name) if isinstance(charts, dict) else None
120 if not isinstance(entry, dict):
121 return None
122 return {
123 "name": name,
124 "chart": str(entry.get("chart") or name),
125 "version": str(entry.get("version") or ""),
126 "namespace": str(entry.get("namespace") or ""),
127 "repo_url": str(entry.get("repo_url") or ""),
128 }
131def manifest_documents(filename: str, manifests_dir: Path | None = None) -> list[dict[str, Any]]:
132 """Parse one shipped kubectl-applier manifest (``[]`` when it is not there).
134 ``{{TOKEN}}`` placeholders are left in place; callers read only fields
135 that never carry one (kinds, names, pinned package references).
136 """
137 path = (manifests_dir or MANIFESTS_DIR) / filename
138 try:
139 text = path.read_text(encoding="utf-8")
140 except OSError:
141 return []
142 # Structural tokens are bare (not quoted) and are not YAML on their own.
143 text = re.sub(r"(?<![\"'])\{\{[A-Z0-9_]+\}\}", "null", text)
144 return [doc for doc in yaml.safe_load_all(text) if isinstance(doc, dict)]
147def kubectl_server_flags(server: str | None, tls_server_name: str | None) -> list[str]:
148 """``--server`` / ``--tls-server-name`` for a tunnelled API endpoint (or nothing)."""
149 flags: list[str] = []
150 if server is not None:
151 if not server.startswith("https://"):
152 raise ValueError(f"Invalid --server {server!r}: must start with https://")
153 flags += ["--server", server]
154 if tls_server_name is not None:
155 if not re.fullmatch(r"[a-zA-Z0-9.\-]{1,255}", tls_server_name):
156 raise ValueError(f"Invalid --tls-server-name {tls_server_name!r}")
157 flags += ["--tls-server-name", tls_server_name]
158 return flags
161def read_secret_key(
162 namespace: str,
163 name: str,
164 key: str,
165 *,
166 server: str | None = None,
167 tls_server_name: str | None = None,
168) -> str:
169 """Decode one key of a Kubernetes Secret through kubectl.
171 Raises ``RuntimeError`` naming the Secret when kubectl is missing, the
172 read fails, or the key is absent (the message never carries the value).
173 """
174 for label, value in (("namespace", namespace), ("Secret name", name)):
175 if not _NAME_RE.match(value):
176 raise ValueError(f"Invalid {label} {value!r}")
177 cmd = [
178 "kubectl",
179 "get",
180 "secret",
181 name,
182 "-n",
183 namespace,
184 "-o",
185 "json",
186 *kubectl_server_flags(server, tls_server_name),
187 ]
188 try:
189 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - validated argv, list form, no shell=True
190 cmd, capture_output=True, text=True, check=False, timeout=60
191 )
192 except FileNotFoundError as exc:
193 raise RuntimeError(
194 "kubectl not found. Install kubectl and ensure it's on your PATH."
195 ) from exc
196 except subprocess.TimeoutExpired as exc:
197 raise RuntimeError(f"Timed out reading Secret {namespace}/{name}") from exc
198 if result.returncode != 0:
199 detail = (result.stderr or "").strip()
200 raise RuntimeError(f"Failed to read Secret {namespace}/{name}: {detail}")
201 data = (json.loads(result.stdout or "{}").get("data")) or {}
202 if key not in data:
203 raise RuntimeError(f"Secret {namespace}/{name} has no {key!r} key")
204 return base64.b64decode(data[key]).decode("utf-8")
207def wait_for_local_port(
208 port: int,
209 process: subprocess.Popen[bytes] | None = None,
210 *,
211 timeout: float = PORT_FORWARD_READY_TIMEOUT_SECONDS,
212) -> None:
213 """Block until 127.0.0.1:``port`` accepts a TCP connection.
215 Raises ``RuntimeError`` when ``process`` (the port-forward) exits first or
216 the deadline passes.
217 """
218 deadline = time.monotonic() + timeout
219 while True:
220 if process is not None and process.poll() is not None:
221 stderr = process.stderr.read().decode("utf-8", "replace") if process.stderr else ""
222 raise RuntimeError(
223 f"kubectl port-forward exited before localhost:{port} was ready: "
224 f"{stderr.strip() or f'exit code {process.returncode}'}"
225 )
226 try:
227 with socket.create_connection(("127.0.0.1", port), timeout=1.0):
228 return
229 except OSError:
230 if time.monotonic() >= deadline:
231 raise RuntimeError(
232 f"localhost:{port} did not accept connections within {int(timeout)}s"
233 ) from None
234 time.sleep(0.5)
237@contextmanager
238def background_port_forward(
239 cmd: Sequence[str],
240 local_port: int,
241 *,
242 timeout: float = PORT_FORWARD_READY_TIMEOUT_SECONDS,
243) -> Iterator[subprocess.Popen[bytes]]:
244 """Run a validated ``kubectl port-forward`` argv in the background until exit.
246 Yields once the local port accepts connections; always terminates the
247 process (kill after a short grace period) on the way out.
248 """
249 process = subprocess.Popen( # nosemgrep: dangerous-subprocess-use-audit - argv built by build_port_forward_command; list form, no shell=True
250 list(cmd), stdout=subprocess.DEVNULL, stderr=subprocess.PIPE
251 )
252 try:
253 wait_for_local_port(local_port, process, timeout=timeout)
254 yield process
255 finally:
256 _stop_process(process)
259def _stop_process(process: subprocess.Popen[bytes]) -> None:
260 """Terminate ``process`` if it still runs, killing it after a short grace period."""
261 if process.poll() is None:
262 process.terminate()
263 try:
264 process.wait(timeout=5)
265 except subprocess.TimeoutExpired:
266 process.kill()
267 process.wait(timeout=5)
270def capture_page(
271 url: str,
272 output: Path,
273 *,
274 cookies: Sequence[Mapping[str, Any]] = (),
275 headless: bool = True,
276 render_wait_ms: int = RENDER_WAIT_MS,
277) -> Path:
278 """Screenshot ``url`` into ``output`` (full page) with Playwright's Chromium.
280 ``cookies`` are added to the browser context before navigation (the Argo
281 CD session token rides this way). Parent directories are created.
282 """
283 try:
284 from playwright.sync_api import sync_playwright
285 except ImportError as exc:
286 raise RuntimeError(_INSTALL_HINT) from exc
288 output = Path(output)
289 output.parent.mkdir(parents=True, exist_ok=True)
290 with sync_playwright() as playwright:
291 browser = playwright.chromium.launch(headless=headless)
292 try:
293 context = browser.new_context(
294 viewport={"width": VIEWPORT_WIDTH, "height": VIEWPORT_HEIGHT}
295 )
296 if cookies:
297 # Playwright types the list as its SetCookieParam TypedDict; the
298 # callers pass plain name/value/url mappings of that shape.
299 context.add_cookies(cast(Any, [dict(cookie) for cookie in cookies]))
300 page = context.new_page()
301 page.goto(url, wait_until="load")
302 page.wait_for_timeout(render_wait_ms)
303 page.screenshot(path=str(output), full_page=True)
304 finally:
305 browser.close()
306 return output
309def tunnel_options(func: Any) -> Any:
310 """The --region / --via-ssm / --bastion-ttl-minutes / --yes options UI commands share."""
311 func = click.option(
312 "--yes",
313 "-y",
314 "assume_yes",
315 is_flag=True,
316 help="Skip the confirmation prompt when provisioning an `--via-ssm auto` bastion.",
317 )(func)
318 func = click.option(
319 "--bastion-ttl-minutes",
320 type=int,
321 default=DEFAULT_BASTION_TTL_MINUTES,
322 show_default=True,
323 help="Self-terminate backstop (minutes) for an `--via-ssm auto` bastion.",
324 )(func)
325 func = click.option(
326 "--via-ssm",
327 "via_ssm",
328 metavar="INSTANCE_ID|auto",
329 help=(
330 "Tunnel to the private API endpoint through an SSM-managed instance. "
331 "Pass an instance id to use an existing one, or 'auto' to provision a "
332 "self-terminating ephemeral bastion and tear it down afterwards. Used "
333 "whenever private endpoint access is on, PUBLIC_AND_PRIVATE included."
334 ),
335 )(func)
336 func = click.option(
337 "--region", "-r", help="Cluster region (defaults to the first cdk.json regional entry)."
338 )(func)
339 return func
342def _relay_port_forward_output(
343 stream: Iterable[bytes], local_port: int, ready: threading.Event
344) -> None:
345 """Echo kubectl's stdout lines, setting ``ready`` at its listener on ``local_port``."""
346 for raw in stream:
347 line = raw.decode("utf-8", "replace").rstrip("\r\n")
348 click.echo(line)
349 match = _FORWARDING_LINE.match(line)
350 if match is not None and int(match.group("port")) == local_port:
351 ready.set()
354def exec_port_forward(
355 cmd: Sequence[str],
356 local_port: int,
357 *,
358 on_ready: Callable[[], object] | None = None,
359 ready_timeout: float = FOREGROUND_PORT_FORWARD_READY_TIMEOUT_SECONDS,
360) -> None:
361 """Run the (validated) kubectl port-forward argv in the foreground until it exits.
363 kubectl's output is relayed line by line, and ``on_ready`` runs once kubectl
364 reports a listener on ``local_port`` (``Forwarding from 127.0.0.1:PORT ->
365 ...``). kubectl prints that only after the API server accepted the forward,
366 so a caller's "forwarding" message never precedes a working forward.
368 Raises ``RuntimeError`` when kubectl exits, or ``ready_timeout`` passes,
369 before that line, and when kubectl exits non-zero after it. kubectl's own
370 error reaches the terminal on its inherited stderr. The process is stopped
371 on the way out, Ctrl-C included.
372 """
373 process = subprocess.Popen( # nosemgrep: dangerous-subprocess-use-audit - argv built by build_port_forward_command; list form, no shell=True
374 list(cmd), stdout=subprocess.PIPE
375 )
376 ready = threading.Event()
377 relay = threading.Thread(
378 target=_relay_port_forward_output,
379 args=(cast(IO[bytes], process.stdout), local_port, ready),
380 name="kubectl-port-forward-output",
381 daemon=True,
382 )
383 relay.start()
384 try:
385 deadline = time.monotonic() + ready_timeout
386 while not ready.wait(_READY_POLL_SECONDS):
387 if process.poll() is not None:
388 # The listener line may still be in flight from the relay.
389 relay.join(_RELAY_DRAIN_SECONDS)
390 if ready.is_set():
391 break
392 raise RuntimeError(
393 f"kubectl port-forward exited with code {process.returncode} before "
394 f"localhost:{local_port} was listening; its error is above."
395 )
396 if time.monotonic() >= deadline:
397 raise RuntimeError(
398 f"kubectl port-forward did not open localhost:{local_port} within "
399 f"{ready_timeout:g}s."
400 )
401 if on_ready is not None:
402 on_ready()
403 returncode = process.wait()
404 if returncode != 0:
405 raise RuntimeError(f"kubectl port-forward exited with code {returncode}.")
406 finally:
407 _stop_process(process)
408 relay.join(_RELAY_DRAIN_SECONDS)