Coverage for cli / cluster_doctor.py: 100.00%
189 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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],
88 capture_output=True,
89 text=True,
90 check=False,
91 )
94def caller_principal_arn() -> str | None:
95 """The caller's IAM principal, with assumed-role ARNs normalized.
97 EKS access entries are created for the base role
98 (``arn:...:role/Name``), while STS reports an assumed-role session
99 (``arn:...:assumed-role/Name/session``); comparing the raw session ARN
100 against the entry list would report a false "no access entry". The same
101 normalization ``gco stacks access`` applies when creating the entry.
102 """
103 try:
104 result = _run_aws(["sts", "get-caller-identity", "--output", "json"])
105 except FileNotFoundError:
106 return None
107 if result.returncode != 0:
108 return None
109 try:
110 payload = json.loads(result.stdout or "{}")
111 except json.JSONDecodeError:
112 return None
113 arn = str(payload.get("Arn") or "")
114 if not arn:
115 return None
116 assumed = _ASSUMED_ROLE_RE.search(arn)
117 if assumed:
118 partition = arn.split(":")[1] if arn.count(":") >= 2 else "aws"
119 account = str(payload.get("Account") or "")
120 return f"arn:{partition}:iam::{account}:role/{assumed.group(1)}"
121 return arn
124def list_access_entries(cluster: str, region: str) -> list[str] | None:
125 """Principal ARNs holding an EKS access entry, or ``None`` when unknowable."""
126 try:
127 result = _run_aws(
128 [
129 "eks",
130 "list-access-entries",
131 "--cluster-name",
132 cluster,
133 "--region",
134 region,
135 "--output",
136 "json",
137 ]
138 )
139 except FileNotFoundError:
140 return None
141 if result.returncode != 0:
142 return None
143 try:
144 payload = json.loads(result.stdout or "{}")
145 except json.JSONDecodeError:
146 return None
147 return [str(entry) for entry in payload.get("accessEntries", [])]
150def list_associated_access_policies(
151 cluster: str, region: str, principal_arn: str
152) -> list[str] | None:
153 """Access-policy ARNs associated with one principal, or ``None`` when unknowable."""
154 try:
155 result = _run_aws(
156 [
157 "eks",
158 "list-associated-access-policies",
159 "--cluster-name",
160 cluster,
161 "--region",
162 region,
163 "--principal-arn",
164 principal_arn,
165 "--output",
166 "json",
167 ]
168 )
169 except FileNotFoundError:
170 return None
171 if result.returncode != 0:
172 return None
173 try:
174 payload = json.loads(result.stdout or "{}")
175 except json.JSONDecodeError:
176 return None
177 policies = payload.get("associatedAccessPolicies", [])
178 return [
179 str(policy.get("policyArn", ""))
180 for policy in policies
181 if isinstance(policy, dict) and policy.get("policyArn")
182 ]
185def kubeconfig_cluster_entry(cluster_name: str) -> tuple[str, bool] | None:
186 """The kubeconfig ``(server, tunnel_pinned)`` recorded for this cluster.
188 Matches entries the same way :func:`cli.kubectl_helpers._tunnel_pinned_server`
189 does (exact name or the ARN-shaped ``…cluster/<name>`` suffix), but
190 returns the server for ANY matching entry — the doctor needs to see a
191 stale non-tunnel server too, not just tunnel pins.
192 """
193 import yaml
195 path = kubectl_helpers._kubeconfig_file()
196 try:
197 config = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
198 except OSError, yaml.YAMLError:
199 return None
200 expected_suffix = f"cluster/{cluster_name}"
201 for entry in config.get("clusters", []) or []:
202 name = str(entry.get("name", ""))
203 if name != cluster_name and not name.endswith(expected_suffix):
204 continue
205 cluster = entry.get("cluster") or {}
206 server = str(cluster.get("server", ""))
207 host = urlsplit(server).hostname or ""
208 pinned = host in kubectl_helpers._LOCAL_TUNNEL_HOSTS and bool(
209 cluster.get("tls-server-name")
210 )
211 return (server, pinned)
212 return None
215def probe_cluster(cluster: str, region: str) -> ClusterProbe:
216 """Collect every layer's raw state for :func:`diagnose`."""
217 exists = True
218 describe_error: str | None = None
219 access: dict[str, Any] = {"endpoint": "", "public": False, "private": False, "public_cidrs": []}
220 try:
221 access = kubectl_helpers.describe_cluster_access(cluster, region)
222 except (RuntimeError, ValueError) as exc:
223 exists = False
224 describe_error = str(exc)
226 caller = caller_principal_arn()
227 entries = list_access_entries(cluster, region) if exists else None
228 policies: list[str] | None = None
229 if exists and caller and entries is not None and caller in entries:
230 policies = list_associated_access_policies(cluster, region, caller)
232 kubeconfig = kubeconfig_cluster_entry(cluster)
233 return ClusterProbe(
234 cluster=cluster,
235 region=region,
236 exists=exists,
237 describe_error=describe_error,
238 endpoint=str(access.get("endpoint") or ""),
239 public=bool(access.get("public")),
240 private=bool(access.get("private")),
241 public_cidrs=[str(cidr) for cidr in access.get("public_cidrs") or []],
242 caller_arn=caller,
243 access_entries=entries,
244 associated_policies=policies,
245 kubeconfig_server=kubeconfig[0] if kubeconfig else None,
246 kubeconfig_tunnel_pinned=kubeconfig[1] if kubeconfig else False,
247 )
250def _host(url: str) -> str:
251 return (urlsplit(url).hostname or "").lower()
254def _diagnose_missing_cluster(probe: ClusterProbe) -> list[DoctorCheck]:
255 """The cluster cannot be described: destroyed, renamed, or unqueryable."""
256 checks: list[DoctorCheck] = []
257 not_found = "ResourceNotFoundException" in (probe.describe_error or "")
258 if not not_found:
259 checks.append(
260 DoctorCheck(
261 layer="cluster",
262 status="unknown",
263 finding=(
264 f"could not describe cluster {probe.cluster!r} in {probe.region}: "
265 f"{probe.describe_error}"
266 ),
267 remedy=("Check AWS credentials/region (aws sts get-caller-identity) and retry."),
268 )
269 )
270 return checks
272 checks.append(
273 DoctorCheck(
274 layer="cluster",
275 status="fail",
276 finding=f"cluster {probe.cluster!r} does not exist in {probe.region}",
277 remedy=(
278 f"Deploy it (gco stacks deploy {probe.cluster} -y) or check "
279 "`gco stacks list` for where this deployment actually runs."
280 ),
281 )
282 )
283 if probe.kubeconfig_server and not probe.kubeconfig_tunnel_pinned:
284 checks.append(
285 DoctorCheck(
286 layer="kubeconfig",
287 status="fail",
288 finding=(
289 f"kubeconfig still has a context for {probe.cluster!r} pointing at "
290 f"{_host(probe.kubeconfig_server)} — kubectl against it fails with "
291 f"'{NO_SUCH_HOST}'. That is a stale context for a destroyed cluster, "
292 "NOT a private-endpoint problem."
293 ),
294 remedy=(
295 "Remove the stale context (kubectl config delete-context) or refresh "
296 "it after redeploying (aws eks update-kubeconfig "
297 f"--name {probe.cluster} --region {probe.region})."
298 ),
299 )
300 )
301 return checks
304def _diagnose_reachability(probe: ClusterProbe) -> DoctorCheck:
305 if probe.public:
306 if probe.public_cidrs and "0.0.0.0/0" not in probe.public_cidrs:
307 remedy = (
308 "If kubectl times out from this machine, confirm your egress IP is "
309 "inside the allowlist; adjust with gco stacks eks endpoint set "
310 "PUBLIC_AND_PRIVATE --cidr <your-ip>/32 and redeploy."
311 )
312 if probe.private:
313 # A rotating egress pool can't be allowlisted; the private
314 # endpoint needs no allowlist at all.
315 remedy += (
316 " Or bypass the allowlist: private access is on too, so "
317 "--via-ssm reaches the private endpoint "
318 "(gco cluster tunnel --via-ssm auto)."
319 )
320 return DoctorCheck(
321 layer="reachability",
322 status="ok",
323 finding=(
324 f"public endpoint restricted to CIDR allowlist: {', '.join(probe.public_cidrs)}"
325 ),
326 remedy=remedy,
327 )
328 return DoctorCheck(
329 layer="reachability",
330 status="ok",
331 finding="public endpoint reachable from the internet (0.0.0.0/0; IAM still gates use)",
332 remedy=(
333 "Consider restricting it: gco stacks eks endpoint set PUBLIC_AND_PRIVATE "
334 "--cidr <your-ip>/32, or PRIVATE with gco cluster tunnel."
335 ),
336 )
337 if probe.kubeconfig_tunnel_pinned:
338 return DoctorCheck(
339 layer="reachability",
340 status="ok",
341 finding=(
342 "PRIVATE endpoint with an SSM tunnel pin in kubeconfig "
343 f"({probe.kubeconfig_server}) — kubectl works while that tunnel is open"
344 ),
345 remedy="If kubectl fails, reopen the tunnel: gco cluster tunnel --via-ssm auto.",
346 )
347 return DoctorCheck(
348 layer="reachability",
349 status="warn",
350 finding=(
351 "PRIVATE endpoint — kubectl from outside the VPC cannot connect "
352 "(connection timeouts, NOT an authentication problem)"
353 ),
354 remedy=(
355 "Open a tunnel: gco cluster tunnel --via-ssm auto (self-terminating bastion), "
356 "or use a VPN/bastion. Note the tunnel does not replace an access entry."
357 ),
358 )
361def _diagnose_authentication(probe: ClusterProbe) -> DoctorCheck:
362 if not probe.caller_arn:
363 return DoctorCheck(
364 layer="authentication",
365 status="unknown",
366 finding="could not resolve the caller's IAM principal",
367 remedy="Configure AWS credentials (aws sts get-caller-identity must succeed).",
368 )
369 if probe.access_entries is None:
370 return DoctorCheck(
371 layer="authentication",
372 status="unknown",
373 finding="could not list the cluster's EKS access entries",
374 remedy=(
375 "The caller needs eks:ListAccessEntries to run this check; "
376 "try gco stacks access -r " + probe.region + " which also creates the entry."
377 ),
378 )
379 if probe.caller_arn in probe.access_entries:
380 return DoctorCheck(
381 layer="authentication",
382 status="ok",
383 finding=f"access entry exists for {probe.caller_arn}",
384 )
385 return DoctorCheck(
386 layer="authentication",
387 status="fail",
388 finding=(
389 f"no EKS access entry for {probe.caller_arn} — kubectl fails with "
390 "'Unauthorized' even over a working tunnel or public endpoint"
391 ),
392 remedy=(
393 f"Run gco stacks access -r {probe.region} (one-shot entry + admin policy), "
394 "or declare the principal in cdk.json eks_cluster.developer_access and "
395 "redeploy for a namespace-scoped grant."
396 ),
397 )
400def _diagnose_authorization(probe: ClusterProbe) -> DoctorCheck | None:
401 if not probe.caller_arn or probe.access_entries is None:
402 return None
403 if probe.caller_arn not in probe.access_entries:
404 return None
405 if probe.associated_policies is None:
406 return DoctorCheck(
407 layer="authorization",
408 status="unknown",
409 finding="could not list associated access policies for the caller's entry",
410 remedy="The caller needs eks:ListAssociatedAccessPolicies to run this check.",
411 )
412 if not probe.associated_policies:
413 return DoctorCheck(
414 layer="authorization",
415 status="fail",
416 finding=(
417 "the access entry has no associated access policies — kubectl "
418 "authenticates but every verb is Forbidden"
419 ),
420 remedy=(
421 f"Run gco stacks access -r {probe.region} to associate "
422 "AmazonEKSClusterAdminPolicy, or set a namespace-scoped policy via "
423 "cdk.json eks_cluster.developer_access."
424 ),
425 )
426 names = ", ".join(arn.rsplit("/", 1)[-1] for arn in probe.associated_policies)
427 return DoctorCheck(
428 layer="authorization",
429 status="ok",
430 finding=f"associated access policies: {names}",
431 )
434def _diagnose_kubeconfig(probe: ClusterProbe) -> DoctorCheck:
435 if probe.kubeconfig_server is None:
436 return DoctorCheck(
437 layer="kubeconfig",
438 status="warn",
439 finding=f"no kubeconfig context for {probe.cluster!r}",
440 remedy=(
441 f"aws eks update-kubeconfig --name {probe.cluster} --region {probe.region} "
442 f"(gco stacks access -r {probe.region} also does this)."
443 ),
444 )
445 if probe.kubeconfig_tunnel_pinned:
446 return DoctorCheck(
447 layer="kubeconfig",
448 status="ok",
449 finding=f"context pinned to a local SSM tunnel ({probe.kubeconfig_server})",
450 remedy="Deliberate tunnel pin; gco commands preserve it while the tunnel is open.",
451 )
452 if _host(probe.kubeconfig_server) == _host(probe.endpoint):
453 return DoctorCheck(
454 layer="kubeconfig",
455 status="ok",
456 finding="context points at the live cluster endpoint",
457 )
458 return DoctorCheck(
459 layer="kubeconfig",
460 status="fail",
461 finding=(
462 f"context points at {_host(probe.kubeconfig_server)} but the live endpoint is "
463 f"{_host(probe.endpoint)} — kubectl errors like '{NO_SUCH_HOST}' come from this "
464 "stale entry, not from the endpoint access mode"
465 ),
466 remedy=(
467 f"aws eks update-kubeconfig --name {probe.cluster} --region {probe.region} "
468 "to repoint the context."
469 ),
470 )
473def diagnose(probe: ClusterProbe) -> list[DoctorCheck]:
474 """Turn one probe into per-layer findings with remedies (pure)."""
475 if not probe.exists:
476 return _diagnose_missing_cluster(probe)
477 checks = [
478 _diagnose_reachability(probe),
479 _diagnose_authentication(probe),
480 ]
481 authorization = _diagnose_authorization(probe)
482 if authorization is not None:
483 checks.append(authorization)
484 checks.append(_diagnose_kubeconfig(probe))
485 return checks
488def endpoint_drift(
489 configured_mode: str,
490 configured_cidrs: list[str],
491 live: dict[str, Any],
492) -> str | None:
493 """Describe cdk.json-vs-live endpoint drift, or ``None`` when converged.
495 Used by ``gco stacks status`` so an endpoint flip that was configured
496 (``gco stacks eks endpoint set``) but not yet deployed — or applied
497 out-of-band and never written back — shows up as explicit drift.
498 """
499 live_public = bool(live.get("public"))
500 live_cidrs = sorted(str(cidr) for cidr in live.get("public_cidrs") or [])
501 configured_public = configured_mode == "PUBLIC_AND_PRIVATE"
502 if configured_public != live_public:
503 live_mode = "PUBLIC_AND_PRIVATE" if live_public else "PRIVATE"
504 return (
505 f"cdk.json eks_cluster.endpoint_access={configured_mode} but the live "
506 f"endpoint is {live_mode}"
507 )
508 if not configured_public:
509 return None
510 expected = sorted(str(cidr) for cidr in configured_cidrs) or ["0.0.0.0/0"]
511 if expected != live_cidrs:
512 return (
513 "cdk.json eks_cluster.public_access_cidrs="
514 f"[{', '.join(expected)}] but the live allowlist is "
515 f"[{', '.join(live_cidrs) or '0.0.0.0/0'}]"
516 )
517 return None