Coverage for cli/cluster_tunnel.py: 100.00%
116 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +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).
23``describe_cluster_access`` and ``start_api_tunnel`` are reached via their
24modules (not ``from ... import``) so tests can monkeypatch them at the source.
25"""
27from __future__ import annotations
29from collections.abc import Iterator
30from contextlib import contextmanager
31from dataclasses import dataclass
32from typing import Any
34import click
36from . import ephemeral_bastion, kubectl_helpers, ssm_tunnel
38# Sentinel for --via-ssm: provision (and later destroy) an ephemeral bastion
39# instead of tunnelling through a caller-supplied instance id.
40AUTO_BASTION = "auto"
42# Local port kubectl uses to reach the API server through the SSM tunnel.
43DEFAULT_API_LOCAL_PORT = 8443
45# A syntactically valid placeholder instance id, used only to render a copy-paste
46# command *template* when the caller hasn't supplied a real instance id.
47_PLACEHOLDER_INSTANCE = "i-0123456789abcdef0"
50@dataclass(frozen=True)
51class TunnelPlan:
52 """How to reach a cluster's EKS API endpoint from a laptop.
54 Pure data + pure builders — no side effects — so it can back a ``--print``
55 CLI mode and an MCP tool that just *describe* the connection.
56 """
58 cluster: str
59 region: str
60 endpoint: str
61 public: bool
62 private: bool
63 local_port: int = DEFAULT_API_LOCAL_PORT
65 @property
66 def endpoint_host(self) -> str:
67 """The bare endpoint hostname (kubectl ``--tls-server-name``), or ``""``."""
68 try:
69 return ssm_tunnel.endpoint_host(self.endpoint)
70 except ValueError:
71 return ""
73 def ssm_command(self, instance_id: str) -> list[str]:
74 """The ``aws ssm start-session`` argv that tunnels to the API endpoint."""
75 return ssm_tunnel.build_remote_host_port_forward_command(
76 instance_id, self.endpoint_host, self.local_port, self.region
77 )
79 def kubectl_flags(self) -> list[str]:
80 """The ``kubectl`` flags that point at the tunnel with the right TLS SNI."""
81 return [
82 "--server",
83 f"https://localhost:{self.local_port}",
84 "--tls-server-name",
85 self.endpoint_host,
86 ]
88 def update_kubeconfig_command(self) -> list[str]:
89 """The ``aws eks update-kubeconfig`` argv for this cluster."""
90 return [
91 "aws",
92 "eks",
93 "update-kubeconfig",
94 "--name",
95 self.cluster,
96 "--region",
97 self.region,
98 ]
100 def as_dict(self, instance_id: str | None = None) -> dict[str, Any]:
101 """A JSON-friendly connection plan (what ``--print`` / the MCP tool emit)."""
102 data: dict[str, Any] = {
103 "cluster": self.cluster,
104 "region": self.region,
105 "endpoint": self.endpoint,
106 "endpoint_host": self.endpoint_host,
107 "public": self.public,
108 "private": self.private,
109 "local_port": self.local_port,
110 "update_kubeconfig": self.update_kubeconfig_command(),
111 }
112 if not self.private:
113 data["reachable"] = "direct"
114 data["note"] = (
115 "The API endpoint is publicly reachable — run `aws eks update-kubeconfig` "
116 "then use kubectl directly; no SSM tunnel is required."
117 )
118 return data
120 data["reachable"] = "ssm-tunnel"
121 data["kubectl_flags"] = self.kubectl_flags()
122 data["kubectl_example"] = "kubectl " + " ".join(self.kubectl_flags()) + " get nodes"
123 if instance_id:
124 cmd = self.ssm_command(instance_id)
125 data["ssm_command"] = cmd
126 data["ssm_command_str"] = " ".join(cmd)
127 data["note"] = (
128 "Run ssm_command_str in one shell to open the tunnel (keep it open), then "
129 "use kubectl_flags (or kubectl_example) in another shell."
130 )
131 else:
132 template = " ".join(self.ssm_command(_PLACEHOLDER_INSTANCE)).replace(
133 _PLACEHOLDER_INSTANCE, "<INSTANCE_ID>"
134 )
135 data["ssm_command_template"] = template
136 data["note"] = (
137 "Private endpoint. Replace <INSTANCE_ID> in ssm_command_template with an "
138 "SSM-managed instance in the cluster VPC and run it to open the tunnel "
139 "(or run `gco cluster tunnel --via-ssm auto` to auto-provision one), then "
140 "use kubectl_flags in another shell."
141 )
142 return data
145def resolve_tunnel_plan(
146 cluster: str, region: str, *, local_port: int = DEFAULT_API_LOCAL_PORT
147) -> TunnelPlan:
148 """Describe how to reach ``cluster``'s API endpoint (raises on lookup failure)."""
149 access = kubectl_helpers.describe_cluster_access(cluster, region)
150 return TunnelPlan(
151 cluster=cluster,
152 region=region,
153 endpoint=str(access.get("endpoint") or ""),
154 public=bool(access.get("public")),
155 private=bool(access.get("private")),
156 local_port=local_port,
157 )
160def resolve_region(config: Any, region: str | None) -> str:
161 """Pick the target region: explicit flag, else first cdk.json regional, else default."""
162 if region:
163 return str(region)
164 from .config import _load_cdk_json
166 cdk = _load_cdk_json()
167 if isinstance(cdk, dict) and cdk.get("regional"):
168 return str(cdk["regional"][0])
169 return str(config.default_region or "us-east-1")
172def _project_name_from_cluster(cluster: str, region: str) -> str:
173 """Recover the project key from a ``{project_name}-{region}`` cluster name.
175 Cluster names are built as ``f"{config.project_name}-{region}"`` (cli/config.py),
176 so stripping the ``-{region}`` suffix yields the project key the bastion's IAM
177 role / instance profile are scoped to — matching the deployment's other
178 project-scoped resources without threading a second config value down here.
179 """
180 suffix = f"-{region}"
181 return cluster[: -len(suffix)] if cluster.endswith(suffix) else cluster
184def private_endpoint_guidance(cluster: str, region: str) -> str:
185 """Actionable message when the API endpoint is private and no tunnel is set up."""
186 return (
187 f"Cluster {cluster!r} has a PRIVATE API endpoint, so kubectl cannot reach it from "
188 "outside the VPC. Either:\n"
189 " - re-run with `--via-ssm <instance-id>` to tunnel through an SSM-managed instance "
190 "in the VPC,\n"
191 " - re-run with `--via-ssm auto` to provision a self-terminating ephemeral bastion,\n"
192 " - connect over a VPN / bastion / AWS SSM session first (see docs/MONITORING.md), or\n"
193 f" - set eks_cluster.endpoint_access to PUBLIC_AND_PRIVATE and redeploy {region}.\n"
194 "Attempting to continue in case you already have VPC connectivity."
195 )
198def provision_bastion(
199 formatter: Any,
200 cluster: str,
201 region: str,
202 ttl_minutes: int,
203 assume_yes: bool,
204 project_name: str = ephemeral_bastion.DEFAULT_PROJECT_NAME,
205) -> str:
206 """Provision an ephemeral SSM bastion for the tunnel; return its instance id.
208 Prompts for confirmation (it launches a billable EC2 instance) unless
209 ``assume_yes``. The instance self-terminates after ``ttl_minutes`` and is
210 torn down by :func:`teardown_bastion` when the tunnel closes. Its IAM role /
211 instance profile are named for ``project_name``.
212 """
213 if not assume_yes:
214 formatter.print_warning(
215 f"--via-ssm auto will launch a {ephemeral_bastion.BASTION_INSTANCE_TYPE} ephemeral "
216 f"bastion in the {cluster} VPC to reach the private API endpoint. It requires no "
217 f"inbound ports (SSM is outbound-only), self-terminates after {ttl_minutes} minutes, "
218 "and is torn down automatically when the tunnel closes."
219 )
220 click.confirm("Provision the ephemeral bastion?", abort=True)
222 formatter.print_info(
223 f"Provisioning ephemeral SSM bastion in {region} "
224 "(waiting for it to register with SSM; this takes a minute)..."
225 )
226 instance_id = ephemeral_bastion.create_ephemeral_bastion(
227 cluster, region, project_name=project_name, ttl_minutes=ttl_minutes
228 )
229 formatter.print_success(f"Ephemeral bastion {instance_id} is online.")
230 return instance_id
233def teardown_bastion(
234 formatter: Any,
235 instance_id: str,
236 region: str,
237 project_name: str = ephemeral_bastion.DEFAULT_PROJECT_NAME,
238) -> None:
239 """Tear down an ephemeral bastion; on failure, print the orphan-check command."""
240 try:
241 formatter.print_info(f"Tearing down ephemeral bastion {instance_id}...")
242 ephemeral_bastion.destroy_ephemeral_bastion(instance_id, region, project_name=project_name)
243 formatter.print_success(f"Ephemeral bastion {instance_id} terminated.")
244 except Exception as exc: # noqa: BLE001 — never crash teardown; guide the operator
245 formatter.print_error(
246 f"Failed to tear down bastion {instance_id}: {exc}\n"
247 "Check for and terminate any orphan with:\n"
248 f" aws ec2 describe-instances --region {region} "
249 "--filters Name=tag:gco:ephemeral,Values=true "
250 "Name=instance-state-name,Values=running,pending "
251 "--query 'Reservations[].Instances[].InstanceId' --output text"
252 )
255@dataclass(frozen=True)
256class TunnelSession:
257 """What :func:`open_api_server_tunnel` yields to the command body."""
259 # kubectl --server / --tls-server-name overrides, or None when the endpoint
260 # is reachable directly (public) or no tunnel could be established.
261 server: str | None
262 tls_server_name: str | None
263 plan: TunnelPlan
264 # True when an SSM tunnel process is running for this session.
265 active: bool
268@contextmanager
269def open_api_server_tunnel(
270 formatter: Any,
271 *,
272 cluster: str,
273 region: str,
274 via_ssm: str | None,
275 local_port: int = DEFAULT_API_LOCAL_PORT,
276 bastion_ttl_minutes: int | None = None,
277 assume_yes: bool = False,
278) -> Iterator[TunnelSession]:
279 """Resolve endpoint posture and, when private + ``via_ssm``, open an SSM tunnel.
281 Yields a :class:`TunnelSession`. Manages the ephemeral-bastion (``--via-ssm
282 auto``) and tunnel lifecycle and tears both down on exit — including when
283 setup fails before the ``yield`` (so a failed tunnel never leaks a bastion).
284 """
285 if bastion_ttl_minutes is None:
286 bastion_ttl_minutes = ephemeral_bastion.DEFAULT_TTL_MINUTES
287 # The bastion's IAM role/profile are scoped to the deployment's project key,
288 # recovered from the cluster name (which is f"{project_name}-{region}").
289 project_name = _project_name_from_cluster(cluster, region)
291 try:
292 plan = resolve_tunnel_plan(cluster, region, local_port=local_port)
293 except (RuntimeError, ValueError) as exc:
294 # Endpoint posture unknown — fall back to a direct attempt.
295 formatter.print_warning(f"Could not determine endpoint access mode: {exc}")
296 plan = TunnelPlan(
297 cluster=cluster,
298 region=region,
299 endpoint="",
300 public=True,
301 private=False,
302 local_port=local_port,
303 )
305 server: str | None = None
306 tls_server_name: str | None = None
307 tunnel = None
308 created_bastion: str | None = None
309 try:
310 if not plan.public:
311 instance_id = via_ssm
312 if via_ssm == AUTO_BASTION:
313 created_bastion = provision_bastion(
314 formatter, cluster, region, bastion_ttl_minutes, assume_yes, project_name
315 )
316 instance_id = created_bastion
318 if instance_id:
319 formatter.print_info(
320 f"Opening SSM tunnel to the private API endpoint via {instance_id}..."
321 )
322 tunnel = ssm_tunnel.start_api_tunnel(instance_id, plan.endpoint, local_port, region)
323 server = f"https://localhost:{local_port}"
324 tls_server_name = plan.endpoint_host
325 else:
326 formatter.print_warning(private_endpoint_guidance(cluster, region))
328 yield TunnelSession(
329 server=server,
330 tls_server_name=tls_server_name,
331 plan=plan,
332 active=tunnel is not None,
333 )
334 finally:
335 if tunnel is not None:
336 tunnel.terminate()
337 if created_bastion is not None:
338 teardown_bastion(formatter, created_bastion, region, project_name)