Coverage for cli / cluster_doctor.py: 100.00%
186 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Diagnosis for the three independent layers of EKS cluster access.
3Reaching a GCO cluster's API from a laptop needs three things to be right at
4once, and each fails with a misleading symptom:
6* **Reachability** — the endpoint mode (PRIVATE needs an SSM tunnel, VPN, or
7 bastion; PUBLIC_AND_PRIVATE may carry a CIDR allowlist your egress IP is
8 not in). Failure symptom: kubectl timeouts.
9* **Authentication** — the cluster authenticates through EKS access entries,
10 and by default only platform Lambda roles have one. Failure symptom:
11 ``Unauthorized`` even though the network path is fine.
12* **Authorization** — an access entry with no associated access policy
13 authenticates but can do nothing. Failure symptom: RBAC ``Forbidden``.
15Plus the kubeconfig context itself, which is where the two most-confused
16failures live: a stale context pointing at a **destroyed** cluster produces
17the same kubectl symptom (``no such host``) as a private-only endpoint, and
18the two remedies are completely different. ``gco cluster doctor`` names each
19layer's state separately and the remedy per case.
21The module is split into subprocess probes (thin, monkeypatchable seams that
22mirror :mod:`cli.kubectl_helpers`' aws-CLI usage) and a pure
23:func:`diagnose` over their results, so the decision table is directly
24testable without any AWS access.
25"""
27from __future__ import annotations
29import json
30import re
31import subprocess
32from dataclasses import dataclass
33from typing import Any
34from urllib.parse import urlsplit
36from . import kubectl_helpers
38_ASSUMED_ROLE_RE = re.compile(r":assumed-role/([^/]+)/")
40# Symptom string kubectl prints for a DNS-dead endpoint; used in findings so
41# an operator can match what they saw to the diagnosis.
42NO_SUCH_HOST = "no such host"
45@dataclass(frozen=True)
46class DoctorCheck:
47 """One layer's diagnosis: what was found and what fixes it."""
49 layer: str # "cluster" | "reachability" | "authentication" | "authorization" | "kubeconfig"
50 status: str # "ok" | "warn" | "fail" | "unknown"
51 finding: str
52 remedy: str | None = None
54 def as_dict(self) -> dict[str, Any]:
55 payload: dict[str, Any] = {
56 "layer": self.layer,
57 "status": self.status,
58 "finding": self.finding,
59 }
60 if self.remedy:
61 payload["remedy"] = self.remedy
62 return payload
65@dataclass(frozen=True)
66class ClusterProbe:
67 """Everything the probes could learn about one cluster's access layers."""
69 cluster: str
70 region: str
71 exists: bool
72 describe_error: str | None
73 endpoint: str
74 public: bool
75 private: bool
76 public_cidrs: list[str]
77 caller_arn: str | None
78 access_entries: list[str] | None
79 associated_policies: list[str] | None
80 kubeconfig_server: str | None
81 kubeconfig_tunnel_pinned: bool
84def _run_aws(args: list[str]) -> subprocess.CompletedProcess[str]:
85 """Run one aws-CLI command (list form, never a shell string)."""
86 return subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - fixed argv head, validated inputs, list form, no shell=True
87 ["aws", *args], capture_output=True, text=True
88 )
91def caller_principal_arn() -> str | None:
92 """The caller's IAM principal, with assumed-role ARNs normalized.
94 EKS access entries are created for the base role
95 (``arn:...:role/Name``), while STS reports an assumed-role session
96 (``arn:...:assumed-role/Name/session``); comparing the raw session ARN
97 against the entry list would report a false "no access entry". The same
98 normalization ``gco stacks access`` applies when creating the entry.
99 """
100 try:
101 result = _run_aws(["sts", "get-caller-identity", "--output", "json"])
102 except FileNotFoundError:
103 return None
104 if result.returncode != 0:
105 return None
106 try:
107 payload = json.loads(result.stdout or "{}")
108 except json.JSONDecodeError:
109 return None
110 arn = str(payload.get("Arn") or "")
111 if not arn:
112 return None
113 assumed = _ASSUMED_ROLE_RE.search(arn)
114 if assumed:
115 partition = arn.split(":")[1] if arn.count(":") >= 2 else "aws"
116 account = str(payload.get("Account") or "")
117 return f"arn:{partition}:iam::{account}:role/{assumed.group(1)}"
118 return arn
121def list_access_entries(cluster: str, region: str) -> list[str] | None:
122 """Principal ARNs holding an EKS access entry, or ``None`` when unknowable."""
123 try:
124 result = _run_aws(
125 [
126 "eks",
127 "list-access-entries",
128 "--cluster-name",
129 cluster,
130 "--region",
131 region,
132 "--output",
133 "json",
134 ]
135 )
136 except FileNotFoundError:
137 return None
138 if result.returncode != 0:
139 return None
140 try:
141 payload = json.loads(result.stdout or "{}")
142 except json.JSONDecodeError:
143 return None
144 return [str(entry) for entry in payload.get("accessEntries", [])]
147def list_associated_access_policies(
148 cluster: str, region: str, principal_arn: str
149) -> list[str] | None:
150 """Access-policy ARNs associated with one principal, or ``None`` when unknowable."""
151 try:
152 result = _run_aws(
153 [
154 "eks",
155 "list-associated-access-policies",
156 "--cluster-name",
157 cluster,
158 "--region",
159 region,
160 "--principal-arn",
161 principal_arn,
162 "--output",
163 "json",
164 ]
165 )
166 except FileNotFoundError:
167 return None
168 if result.returncode != 0:
169 return None
170 try:
171 payload = json.loads(result.stdout or "{}")
172 except json.JSONDecodeError:
173 return None
174 policies = payload.get("associatedAccessPolicies", [])
175 return [
176 str(policy.get("policyArn", ""))
177 for policy in policies
178 if isinstance(policy, dict) and policy.get("policyArn")
179 ]
182def kubeconfig_cluster_entry(cluster_name: str) -> tuple[str, bool] | None:
183 """The kubeconfig ``(server, tunnel_pinned)`` recorded for this cluster.
185 Matches entries the same way :func:`cli.kubectl_helpers._tunnel_pinned_server`
186 does (exact name or the ARN-shaped ``…cluster/<name>`` suffix), but
187 returns the server for ANY matching entry — the doctor needs to see a
188 stale non-tunnel server too, not just tunnel pins.
189 """
190 import yaml
192 path = kubectl_helpers._kubeconfig_file()
193 try:
194 config = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
195 except OSError, yaml.YAMLError:
196 return None
197 expected_suffix = f"cluster/{cluster_name}"
198 for entry in config.get("clusters", []) or []:
199 name = str(entry.get("name", ""))
200 if name != cluster_name and not name.endswith(expected_suffix):
201 continue
202 cluster = entry.get("cluster") or {}
203 server = str(cluster.get("server", ""))
204 host = urlsplit(server).hostname or ""
205 pinned = host in kubectl_helpers._LOCAL_TUNNEL_HOSTS and bool(
206 cluster.get("tls-server-name")
207 )
208 return (server, pinned)
209 return None
212def probe_cluster(cluster: str, region: str) -> ClusterProbe:
213 """Collect every layer's raw state for :func:`diagnose`."""
214 exists = True
215 describe_error: str | None = None
216 access: dict[str, Any] = {"endpoint": "", "public": False, "private": False, "public_cidrs": []}
217 try:
218 access = kubectl_helpers.describe_cluster_access(cluster, region)
219 except (RuntimeError, ValueError) as exc:
220 exists = False
221 describe_error = str(exc)
223 caller = caller_principal_arn()
224 entries = list_access_entries(cluster, region) if exists else None
225 policies: list[str] | None = None
226 if exists and caller and entries is not None and caller in entries:
227 policies = list_associated_access_policies(cluster, region, caller)
229 kubeconfig = kubeconfig_cluster_entry(cluster)
230 return ClusterProbe(
231 cluster=cluster,
232 region=region,
233 exists=exists,
234 describe_error=describe_error,
235 endpoint=str(access.get("endpoint") or ""),
236 public=bool(access.get("public")),
237 private=bool(access.get("private")),
238 public_cidrs=[str(cidr) for cidr in access.get("public_cidrs") or []],
239 caller_arn=caller,
240 access_entries=entries,
241 associated_policies=policies,
242 kubeconfig_server=kubeconfig[0] if kubeconfig else None,
243 kubeconfig_tunnel_pinned=kubeconfig[1] if kubeconfig else False,
244 )
247def _host(url: str) -> str:
248 return (urlsplit(url).hostname or "").lower()
251def _diagnose_missing_cluster(probe: ClusterProbe) -> list[DoctorCheck]:
252 """The cluster cannot be described: destroyed, renamed, or unqueryable."""
253 checks: list[DoctorCheck] = []
254 not_found = "ResourceNotFoundException" in (probe.describe_error or "")
255 if not not_found:
256 checks.append(
257 DoctorCheck(
258 layer="cluster",
259 status="unknown",
260 finding=(
261 f"could not describe cluster {probe.cluster!r} in {probe.region}: "
262 f"{probe.describe_error}"
263 ),
264 remedy=("Check AWS credentials/region (aws sts get-caller-identity) and retry."),
265 )
266 )
267 return checks
269 checks.append(
270 DoctorCheck(
271 layer="cluster",
272 status="fail",
273 finding=f"cluster {probe.cluster!r} does not exist in {probe.region}",
274 remedy=(
275 f"Deploy it (gco stacks deploy {probe.cluster} -y) or check "
276 "`gco stacks list` for where this deployment actually runs."
277 ),
278 )
279 )
280 if probe.kubeconfig_server and not probe.kubeconfig_tunnel_pinned:
281 checks.append(
282 DoctorCheck(
283 layer="kubeconfig",
284 status="fail",
285 finding=(
286 f"kubeconfig still has a context for {probe.cluster!r} pointing at "
287 f"{_host(probe.kubeconfig_server)} — kubectl against it fails with "
288 f"'{NO_SUCH_HOST}'. That is a stale context for a destroyed cluster, "
289 "NOT a private-endpoint problem."
290 ),
291 remedy=(
292 "Remove the stale context (kubectl config delete-context) or refresh "
293 "it after redeploying (aws eks update-kubeconfig "
294 f"--name {probe.cluster} --region {probe.region})."
295 ),
296 )
297 )
298 return checks
301def _diagnose_reachability(probe: ClusterProbe) -> DoctorCheck:
302 if probe.public:
303 if probe.public_cidrs and "0.0.0.0/0" not in probe.public_cidrs:
304 return DoctorCheck(
305 layer="reachability",
306 status="ok",
307 finding=(
308 f"public endpoint restricted to CIDR allowlist: {', '.join(probe.public_cidrs)}"
309 ),
310 remedy=(
311 "If kubectl times out from this machine, confirm your egress IP is "
312 "inside the allowlist; adjust with gco stacks eks endpoint set "
313 "PUBLIC_AND_PRIVATE --cidr <your-ip>/32 and redeploy."
314 ),
315 )
316 return DoctorCheck(
317 layer="reachability",
318 status="ok",
319 finding="public endpoint reachable from the internet (0.0.0.0/0; IAM still gates use)",
320 remedy=(
321 "Consider restricting it: gco stacks eks endpoint set PUBLIC_AND_PRIVATE "
322 "--cidr <your-ip>/32, or PRIVATE with gco cluster tunnel."
323 ),
324 )
325 if probe.kubeconfig_tunnel_pinned:
326 return DoctorCheck(
327 layer="reachability",
328 status="ok",
329 finding=(
330 "PRIVATE endpoint with an SSM tunnel pin in kubeconfig "
331 f"({probe.kubeconfig_server}) — kubectl works while that tunnel is open"
332 ),
333 remedy="If kubectl fails, reopen the tunnel: gco cluster tunnel --via-ssm auto.",
334 )
335 return DoctorCheck(
336 layer="reachability",
337 status="warn",
338 finding=(
339 "PRIVATE endpoint — kubectl from outside the VPC cannot connect "
340 "(connection timeouts, NOT an authentication problem)"
341 ),
342 remedy=(
343 "Open a tunnel: gco cluster tunnel --via-ssm auto (self-terminating bastion), "
344 "or use a VPN/bastion. Note the tunnel does not replace an access entry."
345 ),
346 )
349def _diagnose_authentication(probe: ClusterProbe) -> DoctorCheck:
350 if not probe.caller_arn:
351 return DoctorCheck(
352 layer="authentication",
353 status="unknown",
354 finding="could not resolve the caller's IAM principal",
355 remedy="Configure AWS credentials (aws sts get-caller-identity must succeed).",
356 )
357 if probe.access_entries is None:
358 return DoctorCheck(
359 layer="authentication",
360 status="unknown",
361 finding="could not list the cluster's EKS access entries",
362 remedy=(
363 "The caller needs eks:ListAccessEntries to run this check; "
364 "try gco stacks access -r " + probe.region + " which also creates the entry."
365 ),
366 )
367 if probe.caller_arn in probe.access_entries:
368 return DoctorCheck(
369 layer="authentication",
370 status="ok",
371 finding=f"access entry exists for {probe.caller_arn}",
372 )
373 return DoctorCheck(
374 layer="authentication",
375 status="fail",
376 finding=(
377 f"no EKS access entry for {probe.caller_arn} — kubectl fails with "
378 "'Unauthorized' even over a working tunnel or public endpoint"
379 ),
380 remedy=(
381 f"Run gco stacks access -r {probe.region} (one-shot entry + admin policy), "
382 "or declare the principal in cdk.json eks_cluster.developer_access and "
383 "redeploy for a namespace-scoped grant."
384 ),
385 )
388def _diagnose_authorization(probe: ClusterProbe) -> DoctorCheck | None:
389 if not probe.caller_arn or probe.access_entries is None:
390 return None
391 if probe.caller_arn not in probe.access_entries:
392 return None
393 if probe.associated_policies is None:
394 return DoctorCheck(
395 layer="authorization",
396 status="unknown",
397 finding="could not list associated access policies for the caller's entry",
398 remedy="The caller needs eks:ListAssociatedAccessPolicies to run this check.",
399 )
400 if not probe.associated_policies:
401 return DoctorCheck(
402 layer="authorization",
403 status="fail",
404 finding=(
405 "the access entry has no associated access policies — kubectl "
406 "authenticates but every verb is Forbidden"
407 ),
408 remedy=(
409 f"Run gco stacks access -r {probe.region} to associate "
410 "AmazonEKSClusterAdminPolicy, or set a namespace-scoped policy via "
411 "cdk.json eks_cluster.developer_access."
412 ),
413 )
414 names = ", ".join(arn.rsplit("/", 1)[-1] for arn in probe.associated_policies)
415 return DoctorCheck(
416 layer="authorization",
417 status="ok",
418 finding=f"associated access policies: {names}",
419 )
422def _diagnose_kubeconfig(probe: ClusterProbe) -> DoctorCheck:
423 if probe.kubeconfig_server is None:
424 return DoctorCheck(
425 layer="kubeconfig",
426 status="warn",
427 finding=f"no kubeconfig context for {probe.cluster!r}",
428 remedy=(
429 f"aws eks update-kubeconfig --name {probe.cluster} --region {probe.region} "
430 f"(gco stacks access -r {probe.region} also does this)."
431 ),
432 )
433 if probe.kubeconfig_tunnel_pinned:
434 return DoctorCheck(
435 layer="kubeconfig",
436 status="ok",
437 finding=f"context pinned to a local SSM tunnel ({probe.kubeconfig_server})",
438 remedy="Deliberate tunnel pin; gco commands preserve it while the tunnel is open.",
439 )
440 if _host(probe.kubeconfig_server) == _host(probe.endpoint):
441 return DoctorCheck(
442 layer="kubeconfig",
443 status="ok",
444 finding="context points at the live cluster endpoint",
445 )
446 return DoctorCheck(
447 layer="kubeconfig",
448 status="fail",
449 finding=(
450 f"context points at {_host(probe.kubeconfig_server)} but the live endpoint is "
451 f"{_host(probe.endpoint)} — kubectl errors like '{NO_SUCH_HOST}' come from this "
452 "stale entry, not from the endpoint access mode"
453 ),
454 remedy=(
455 f"aws eks update-kubeconfig --name {probe.cluster} --region {probe.region} "
456 "to repoint the context."
457 ),
458 )
461def diagnose(probe: ClusterProbe) -> list[DoctorCheck]:
462 """Turn one probe into per-layer findings with remedies (pure)."""
463 if not probe.exists:
464 return _diagnose_missing_cluster(probe)
465 checks = [
466 _diagnose_reachability(probe),
467 _diagnose_authentication(probe),
468 ]
469 authorization = _diagnose_authorization(probe)
470 if authorization is not None:
471 checks.append(authorization)
472 checks.append(_diagnose_kubeconfig(probe))
473 return checks
476def endpoint_drift(
477 configured_mode: str,
478 configured_cidrs: list[str],
479 live: dict[str, Any],
480) -> str | None:
481 """Describe cdk.json-vs-live endpoint drift, or ``None`` when converged.
483 Used by ``gco stacks status`` so an endpoint flip that was configured
484 (``gco stacks eks endpoint set``) but not yet deployed — or applied
485 out-of-band and never written back — shows up as explicit drift.
486 """
487 live_public = bool(live.get("public"))
488 live_cidrs = sorted(str(cidr) for cidr in live.get("public_cidrs") or [])
489 configured_public = configured_mode == "PUBLIC_AND_PRIVATE"
490 if configured_public != live_public:
491 live_mode = "PUBLIC_AND_PRIVATE" if live_public else "PRIVATE"
492 return (
493 f"cdk.json eks_cluster.endpoint_access={configured_mode} but the live "
494 f"endpoint is {live_mode}"
495 )
496 if not configured_public:
497 return None
498 expected = sorted(str(cidr) for cidr in configured_cidrs) or ["0.0.0.0/0"]
499 if expected != live_cidrs:
500 return (
501 "cdk.json eks_cluster.public_access_cidrs="
502 f"[{', '.join(expected)}] but the live allowlist is "
503 f"[{', '.join(live_cidrs) or '0.0.0.0/0'}]"
504 )
505 return None