Coverage for cli / cluster_tunnel.py: 100.00%
166 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Shared helpers for reaching a cluster's (possibly private) EKS API endpoint.
3GCO clusters default to a PRIVATE EKS API endpoint, so ``kubectl`` cannot reach
4the API server from outside the VPC. Two CLI commands need the same machinery to
5bridge that gap:
7* ``gco monitoring open`` — port-forward Grafana/Prometheus over the API server;
8* ``gco cluster tunnel`` — a general SSM tunnel so plain ``kubectl`` works.
10Rather than duplicate "detect endpoint posture → (optionally provision an
11ephemeral bastion) → open an SSM tunnel → tear everything down", both commands
12share this module:
14* :class:`TunnelPlan` + :func:`resolve_tunnel_plan` — a pure, request/response
15 friendly description of how to reach the endpoint (the ``aws ssm start-session``
16 command and the ``kubectl`` flags). This is what ``--print`` and the MCP
17 ``cluster_tunnel_command`` tool return.
18* :func:`provision_bastion` / :func:`teardown_bastion` — the ephemeral-bastion
19 lifecycle with a confirmation prompt and an orphan-check hint on failure.
20* :func:`open_api_server_tunnel` — a context manager that ties it together and
21 guarantees teardown (including when setup fails before yielding).
23The route rule is the same everywhere: an explicit ``--via-ssm`` (an instance
24id, or ``auto``) means an SSM tunnel to the private endpoint whenever private
25access is enabled, ``PUBLIC_AND_PRIVATE`` included, and an explicit request the
26cluster cannot honor fails with an explanation. Without ``--via-ssm`` a public
27endpoint is used directly.
29``describe_cluster_access`` and ``start_api_tunnel`` are reached via their
30modules (not ``from ... import``) so tests can monkeypatch them at the source.
31"""
33from __future__ import annotations
35import subprocess
36from collections.abc import Iterator
37from contextlib import contextmanager
38from dataclasses import dataclass, field
39from typing import Any
41from . import ephemeral_bastion, kubectl_helpers, ssm_tunnel
42from .output import confirm
44# Sentinel for --via-ssm: provision (and later destroy) an ephemeral bastion
45# instead of tunnelling through a caller-supplied instance id.
46AUTO_BASTION = "auto"
48# Local port kubectl uses to reach the API server through the SSM tunnel.
49DEFAULT_API_LOCAL_PORT = 8443
51# The routes open_api_server_tunnel selects (TunnelSession.route). The first two
52# match TunnelPlan.as_dict()'s "reachable" values.
53ROUTE_SSM = "ssm-tunnel" # an SSM tunnel to the private endpoint
54ROUTE_DIRECT = "direct" # the public endpoint, directly
55# kubectl's own kubeconfig server, untried: a private-only endpoint without
56# --via-ssm, or an endpoint whose access mode could not be read.
57ROUTE_UNVERIFIED = "unverified"
59# A syntactically valid placeholder instance id, used only to render a copy-paste
60# command *template* when the caller hasn't supplied a real instance id.
61_PLACEHOLDER_INSTANCE = "i-0123456789abcdef0"
64@dataclass(frozen=True)
65class TunnelPlan:
66 """How to reach a cluster's EKS API endpoint from a laptop.
68 Pure data + pure builders — no side effects — so it can back a ``--print``
69 CLI mode and an MCP tool that just *describe* the connection.
70 """
72 cluster: str
73 region: str
74 endpoint: str
75 public: bool
76 private: bool
77 local_port: int = DEFAULT_API_LOCAL_PORT
79 @property
80 def endpoint_host(self) -> str:
81 """The bare endpoint hostname (kubectl ``--tls-server-name``), or ``""``."""
82 try:
83 return ssm_tunnel.endpoint_host(self.endpoint)
84 except ValueError:
85 return ""
87 @property
88 def access_mode(self) -> str:
89 """The access mode in cdk.json's vocabulary, plus ``PUBLIC`` for a public-only one.
91 EKS never turns both modes off, so an endpoint without public access is
92 private even when the lookup did not report the private flag.
93 """
94 if not self.public:
95 return "PRIVATE"
96 return "PUBLIC_AND_PRIVATE" if self.private else "PUBLIC"
98 @property
99 def private_access(self) -> bool:
100 """Whether the private endpoint, the one an SSM tunnel reaches, is enabled."""
101 return self.access_mode != "PUBLIC"
103 def ssm_command(self, instance_id: str) -> list[str]:
104 """The ``aws ssm start-session`` argv that tunnels to the API endpoint."""
105 return ssm_tunnel.build_remote_host_port_forward_command(
106 instance_id, self.endpoint_host, self.local_port, self.region
107 )
109 def kubectl_flags(self) -> list[str]:
110 """The ``kubectl`` flags that point at the tunnel with the right TLS SNI."""
111 return [
112 "--server",
113 f"https://{ssm_tunnel.LOCAL_TUNNEL_HOST}:{self.local_port}",
114 "--tls-server-name",
115 self.endpoint_host,
116 ]
118 def update_kubeconfig_command(self) -> list[str]:
119 """The ``aws eks update-kubeconfig`` argv for this cluster."""
120 return [
121 "aws",
122 "eks",
123 "update-kubeconfig",
124 "--name",
125 self.cluster,
126 "--region",
127 self.region,
128 ]
130 def as_dict(self, via_ssm: str | None = None) -> dict[str, Any]:
131 """A JSON-friendly connection plan (what ``--print`` / the MCP tool emit).
133 ``via_ssm`` is the ``--via-ssm`` value (an instance id, or ``auto``). It
134 selects the SSM tunnel whenever private access is enabled, as
135 :func:`open_api_server_tunnel` does; without it a public endpoint is
136 reached directly. Callers reject an explicit ``via_ssm`` for an endpoint
137 without private access first (:func:`require_private_access`).
138 """
139 data: dict[str, Any] = {
140 "cluster": self.cluster,
141 "region": self.region,
142 "endpoint": self.endpoint,
143 "endpoint_host": self.endpoint_host,
144 "access_mode": self.access_mode,
145 "public": self.public,
146 "private": self.private,
147 "local_port": self.local_port,
148 "update_kubeconfig": self.update_kubeconfig_command(),
149 }
150 if not self.private_access or (self.public and not via_ssm):
151 data["reachable"] = ROUTE_DIRECT
152 note = (
153 "The API endpoint is publicly reachable — run `aws eks update-kubeconfig` "
154 "then use kubectl directly; no SSM tunnel is required."
155 )
156 if self.private_access:
157 note += (
158 " Private access is also enabled: if the public endpoint times out from "
159 "your network (for example, an egress IP outside its CIDR allowlist), "
160 "pass `--via-ssm <instance-id>` or `--via-ssm auto` to tunnel to the "
161 "private endpoint instead."
162 )
163 data["note"] = note
164 return data
166 data["reachable"] = ROUTE_SSM
167 data["kubectl_flags"] = self.kubectl_flags()
168 data["kubectl_example"] = "kubectl " + " ".join(self.kubectl_flags()) + " get nodes"
169 instance_id = via_ssm if via_ssm != AUTO_BASTION else None
170 if instance_id:
171 cmd = self.ssm_command(instance_id)
172 data["ssm_command"] = cmd
173 data["ssm_command_str"] = " ".join(cmd)
174 data["note"] = (
175 "Run ssm_command_str in one shell to open the tunnel (keep it open), then "
176 "use kubectl_flags (or kubectl_example) in another shell."
177 )
178 else:
179 template = " ".join(self.ssm_command(_PLACEHOLDER_INSTANCE)).replace(
180 _PLACEHOLDER_INSTANCE, "<INSTANCE_ID>"
181 )
182 data["ssm_command_template"] = template
183 data["note"] = (
184 "Private endpoint. Replace <INSTANCE_ID> in ssm_command_template with an "
185 "SSM-managed instance in the cluster VPC and run it to open the tunnel "
186 "(or run `gco cluster tunnel --via-ssm auto` to auto-provision one), then "
187 "use kubectl_flags in another shell."
188 )
189 return data
192def resolve_tunnel_plan(
193 cluster: str, region: str, *, local_port: int = DEFAULT_API_LOCAL_PORT
194) -> TunnelPlan:
195 """Describe how to reach ``cluster``'s API endpoint (raises on lookup failure)."""
196 access = kubectl_helpers.describe_cluster_access(cluster, region)
197 return TunnelPlan(
198 cluster=cluster,
199 region=region,
200 endpoint=str(access.get("endpoint") or ""),
201 public=bool(access.get("public")),
202 private=bool(access.get("private")),
203 local_port=local_port,
204 )
207def require_private_access(plan: TunnelPlan, via_ssm: str) -> None:
208 """Raise ``RuntimeError`` when an explicit ``--via-ssm`` has no private endpoint to reach."""
209 if plan.private_access:
210 return
211 raise RuntimeError(
212 f"--via-ssm {via_ssm} tunnels to the private API endpoint, but {plan.cluster} "
213 "reports private endpoint access disabled (endpointPrivateAccess=false), so there "
214 "is no private endpoint to reach. Re-run without --via-ssm to use its public "
215 "endpoint directly, or turn private access back on: GCO deploys PRIVATE or "
216 f"PUBLIC_AND_PRIVATE, and `gco stacks deploy {plan.cluster} -y` restores the "
217 "mode configured in cdk.json."
218 )
221def resolve_region(config: Any, region: str | None) -> str:
222 """Pick the target region: explicit flag, else first cdk.json regional, else default."""
223 if region:
224 return str(region)
225 from .config import _load_cdk_json
227 cdk = _load_cdk_json()
228 if isinstance(cdk, dict) and cdk.get("regional"):
229 return str(cdk["regional"][0])
230 return str(config.default_region or "us-east-1")
233def _project_name_from_cluster(cluster: str, region: str) -> str:
234 """Recover the project key from a ``{project_name}-{region}`` cluster name.
236 Cluster names are built as ``f"{config.project_name}-{region}"`` (cli/config.py),
237 so stripping the ``-{region}`` suffix yields the project key the bastion's IAM
238 role / instance profile are scoped to — matching the deployment's other
239 project-scoped resources without threading a second config value down here.
240 """
241 suffix = f"-{region}"
242 return cluster[: -len(suffix)] if cluster.endswith(suffix) else cluster
245def private_endpoint_guidance(cluster: str, region: str) -> str:
246 """Actionable message when the API endpoint is private and no tunnel is set up."""
247 return (
248 f"Cluster {cluster!r} has a PRIVATE API endpoint, so kubectl cannot reach it from "
249 "outside the VPC. Either:\n"
250 " - re-run with `--via-ssm <instance-id>` to tunnel through an SSM-managed instance "
251 "in the VPC,\n"
252 " - re-run with `--via-ssm auto` to provision a self-terminating ephemeral bastion,\n"
253 " - connect over a VPN / bastion / AWS SSM session first (see docs/MONITORING.md), or\n"
254 f" - set eks_cluster.endpoint_access to PUBLIC_AND_PRIVATE and redeploy {region}.\n"
255 "Attempting to continue in case you already have VPC connectivity."
256 )
259def _ssm_route_message(plan: TunnelPlan, instance_id: str) -> str:
260 """The route announcement for an SSM tunnel through ``instance_id``."""
261 precedence = " (--via-ssm takes precedence over its public endpoint)" if plan.public else ""
262 return (
263 f"Route: SSM tunnel via {instance_id} to {plan.cluster}'s private API endpoint"
264 f"{precedence}; opening it on {ssm_tunnel.LOCAL_TUNNEL_HOST}:{plan.local_port}..."
265 )
268def _direct_route_message(plan: TunnelPlan, via_ssm: str | None) -> str:
269 """The route announcement for the public endpoint, reached directly."""
270 message = f"Route: {plan.cluster}'s public API endpoint, directly"
271 if via_ssm:
272 # Only with allow_public_fallback: the explicit request had nowhere to go.
273 return f"{message} (it has no private endpoint for --via-ssm {via_ssm} to reach)."
274 if plan.private_access:
275 return (
276 f"{message}. Its private endpoint is also enabled: if the public one times out "
277 "from this network (for example, an egress IP outside its CIDR allowlist), "
278 "re-run with --via-ssm <instance-id> or --via-ssm auto."
279 )
280 return f"{message}."
283def provision_bastion(
284 formatter: Any,
285 cluster: str,
286 region: str,
287 ttl_minutes: int,
288 assume_yes: bool,
289 project_name: str = ephemeral_bastion.DEFAULT_PROJECT_NAME,
290) -> str:
291 """Provision an ephemeral SSM bastion for the tunnel; return its instance id.
293 Prompts for confirmation (it launches a billable EC2 instance) unless
294 ``assume_yes``. The instance self-terminates after ``ttl_minutes`` and is
295 torn down by :func:`teardown_bastion` when the tunnel closes. Its IAM role /
296 instance profile are named for ``project_name``.
297 """
298 if not assume_yes:
299 formatter.print_warning(
300 f"--via-ssm auto will launch a {ephemeral_bastion.BASTION_INSTANCE_TYPE} ephemeral "
301 f"bastion in the {cluster} VPC to reach the private API endpoint. It requires no "
302 f"inbound ports (SSM is outbound-only), self-terminates after {ttl_minutes} minutes, "
303 "and is torn down automatically when the tunnel closes."
304 )
305 confirm("Provision the ephemeral bastion?", abort=True)
307 formatter.print_info(
308 f"Provisioning ephemeral SSM bastion in {region} "
309 "(waiting for it to register with SSM; this takes a minute)..."
310 )
311 instance_id = ephemeral_bastion.create_ephemeral_bastion(
312 cluster, region, project_name=project_name, ttl_minutes=ttl_minutes
313 )
314 formatter.print_success(f"Ephemeral bastion {instance_id} is online.")
315 return instance_id
318def teardown_bastion(
319 formatter: Any,
320 instance_id: str,
321 region: str,
322 project_name: str = ephemeral_bastion.DEFAULT_PROJECT_NAME,
323) -> None:
324 """Tear down an ephemeral bastion; on failure, print the orphan-check command."""
325 try:
326 formatter.print_info(f"Tearing down ephemeral bastion {instance_id}...")
327 ephemeral_bastion.destroy_ephemeral_bastion(instance_id, region, project_name=project_name)
328 formatter.print_success(f"Ephemeral bastion {instance_id} terminated.")
329 except Exception as exc: # never crash teardown; guide the operator
330 formatter.print_error(
331 f"Failed to tear down bastion {instance_id}: {exc}\n"
332 "Check for and terminate any orphan with:\n"
333 f" aws ec2 describe-instances --region {region} "
334 "--filters Name=tag:gco:ephemeral,Values=true "
335 "Name=instance-state-name,Values=running,pending "
336 "--query 'Reservations[].Instances[].InstanceId' --output text"
337 )
340@dataclass(frozen=True)
341class TunnelSession:
342 """What :func:`open_api_server_tunnel` yields to the command body."""
344 # kubectl --server / --tls-server-name overrides, or None when the endpoint
345 # is reachable directly (public) or no tunnel could be established.
346 server: str | None
347 tls_server_name: str | None
348 plan: TunnelPlan
349 # True when an SSM tunnel process is running for this session.
350 active: bool
351 # Process handle lets readiness gates detect a tunnel that exits after its
352 # local listener first appears. It is not part of session equality/repr.
353 process: subprocess.Popen[bytes] | None = field(default=None, compare=False, repr=False)
354 # The SSM target the tunnel runs through (the ephemeral bastion under
355 # --via-ssm auto), so a long-lived caller can reopen a stalled tunnel on
356 # the same local port through the same instance. None without a tunnel.
357 instance_id: str | None = None
358 # The route selected for kubectl: ROUTE_SSM, ROUTE_DIRECT or ROUTE_UNVERIFIED.
359 route: str = ROUTE_UNVERIFIED
362@contextmanager
363def open_api_server_tunnel(
364 formatter: Any,
365 *,
366 cluster: str,
367 region: str,
368 via_ssm: str | None,
369 local_port: int = DEFAULT_API_LOCAL_PORT,
370 bastion_ttl_minutes: int | None = None,
371 assume_yes: bool = False,
372 allow_public_fallback: bool = False,
373) -> Iterator[TunnelSession]:
374 """Select the route to ``cluster``'s API endpoint and open it for the ``with`` body.
376 ``via_ssm`` (an SSM-managed instance id, or ``auto`` to provision an ephemeral
377 bastion) selects an SSM tunnel to the private endpoint whenever private access
378 is enabled, ``PUBLIC_AND_PRIVATE`` included: a CIDR-restricted public endpoint
379 can be unreachable from the caller's egress IP while the private one works.
380 When the endpoint has no private access, or its access mode cannot be read, an
381 explicit ``via_ssm`` fails with an explanation before anything is provisioned.
382 ``allow_public_fallback`` uses the public endpoint instead (the validation
383 harness, which always passes ``auto``). Without ``via_ssm`` a public endpoint
384 is used directly, and a private-only one gets guidance plus a direct attempt
385 for callers already inside the VPC.
387 Every route is announced through ``formatter``. Yields a
388 :class:`TunnelSession`, and tears the tunnel and any bastion it provisioned
389 down on exit — including when setup fails before the ``yield`` (so a failed
390 tunnel never leaks a bastion).
391 """
392 if bastion_ttl_minutes is None:
393 bastion_ttl_minutes = ephemeral_bastion.DEFAULT_TTL_MINUTES
394 # The bastion's IAM role/profile are scoped to the deployment's project key,
395 # recovered from the cluster name (which is f"{project_name}-{region}").
396 project_name = _project_name_from_cluster(cluster, region)
398 posture_known = True
399 try:
400 plan = resolve_tunnel_plan(cluster, region, local_port=local_port)
401 except (RuntimeError, ValueError) as exc:
402 if via_ssm and not allow_public_fallback:
403 raise RuntimeError(
404 f"--via-ssm {via_ssm} needs {cluster}'s API endpoint, but its access mode "
405 f"could not be read: {exc}"
406 ) from exc
407 # Endpoint posture unknown — fall back to a direct attempt.
408 formatter.print_warning(
409 f"Could not determine endpoint access mode: {exc}. Trying the endpoint in "
410 "your kubeconfig directly."
411 )
412 posture_known = False
413 plan = TunnelPlan(
414 cluster=cluster,
415 region=region,
416 endpoint="",
417 public=True,
418 private=False,
419 local_port=local_port,
420 )
421 if via_ssm and not allow_public_fallback:
422 require_private_access(plan, via_ssm)
424 server: str | None = None
425 tls_server_name: str | None = None
426 tunnel = None
427 tunnel_target: str | None = None
428 created_bastion: str | None = None
429 route = ROUTE_UNVERIFIED
430 try:
431 if via_ssm and plan.private_access:
432 if not plan.endpoint_host:
433 raise RuntimeError(
434 f"EKS reported no usable API endpoint for {cluster} ({plan.endpoint!r}), "
435 "so there is nothing for an SSM tunnel to reach."
436 )
437 instance_id = via_ssm
438 if via_ssm == AUTO_BASTION:
439 created_bastion = provision_bastion(
440 formatter, cluster, region, bastion_ttl_minutes, assume_yes, project_name
441 )
442 instance_id = created_bastion
444 formatter.print_info(_ssm_route_message(plan, instance_id))
445 tunnel = ssm_tunnel.start_api_tunnel(instance_id, plan.endpoint, local_port, region)
446 tunnel_target = instance_id
447 server = f"https://{ssm_tunnel.LOCAL_TUNNEL_HOST}:{local_port}"
448 tls_server_name = plan.endpoint_host
449 route = ROUTE_SSM
450 elif not posture_known:
451 pass # the warning above already named the direct attempt
452 elif plan.public:
453 formatter.print_info(_direct_route_message(plan, via_ssm))
454 route = ROUTE_DIRECT
455 else:
456 formatter.print_warning(private_endpoint_guidance(cluster, region))
458 yield TunnelSession(
459 server=server,
460 tls_server_name=tls_server_name,
461 plan=plan,
462 active=tunnel is not None,
463 process=tunnel,
464 instance_id=tunnel_target,
465 route=route,
466 )
467 finally:
468 try:
469 if tunnel is not None:
470 ssm_tunnel.stop_api_tunnel(tunnel)
471 finally:
472 if created_bastion is not None:
473 teardown_bastion(formatter, created_bastion, region, project_name)