Coverage for cli / ssm_tunnel.py: 100.00%

1"""SSM Session Manager tunnel helpers for reaching a private EKS API endpoint. 

2 

3GCO clusters default to a PRIVATE EKS API endpoint (``eks_cluster.endpoint_access 

4= "PRIVATE"``), so ``kubectl`` — and therefore ``kubectl port-forward`` to 

5Grafana — cannot reach the API server from a laptop outside the VPC. This module 

6builds an ``aws ssm start-session`` port-forwarding tunnel through an SSM-managed 

7instance in the VPC to the cluster's API endpoint, giving kubectl a 

8``https://127.0.0.1:<port>`` server to talk to. 

9 

10The command builder is pure and validated (list form, never a shell string) so 

11it is fully unit-testable; :func:`start_api_tunnel` is the thin runtime wrapper 

12that launches it as a background process and waits for its IPv4 listener. 

13 

14Requires the Session Manager plugin on the local machine 

15(https://docs.aws.amazon.com/systems-manager/latest/userguide/session-manager-working-with-install-plugin.html). 

16""" 

17 

18from __future__ import annotations 

19 

20import json 

21import os 

22import re 

23import shutil 

24import signal 

25import socket 

26import subprocess 

27import time 

28from typing import Any 

29from urllib.parse import urlparse 

30 

31# SSM managed-node ids: EC2 (i-...) and hybrid/managed (mi-...), 8 or 17 hex. 

32_INSTANCE_RE = re.compile(r"^(i|mi)-[0-9a-f]{8}([0-9a-f]{9})?$") 

33_HOST_RE = re.compile(r"^[a-zA-Z0-9.\-]{1,255}$") 

34_REGION_RE = re.compile(r"^[a-z]{2,4}(?:-[a-z0-9]+)+-[0-9]+$") 

35 

36# AWS SSM document that forwards a local port to an arbitrary remote host 

37# reachable from the managed node (here, the private EKS API endpoint). 

38REMOTE_HOST_DOCUMENT = "AWS-StartPortForwardingSessionToRemoteHost" 

39LOCAL_TUNNEL_HOST = "127.0.0.1" 

40_DEFAULT_READY_TIMEOUT_SECONDS = 30.0 

41_DEFAULT_READY_POLL_SECONDS = 0.25 

42_DEFAULT_CONNECT_TIMEOUT_SECONDS = 0.5 

43_DEFAULT_STOP_TIMEOUT_SECONDS = 5.0 

44 

45 

46def _validate_instance_id(instance_id: str) -> None: 

47 if not _INSTANCE_RE.match(instance_id): 

48 raise ValueError( 

49 f"Invalid SSM target {instance_id!r}: expected an instance id like i-0123456789abcdef0" 

50 ) 

51 

52 

53def _validate_host(host: str) -> None: 

54 if not _HOST_RE.match(host): 

55 raise ValueError(f"Invalid remote host {host!r}") 

56 

57 

58def _validate_region(region: str) -> None: 

59 if not _REGION_RE.match(region): 

60 raise ValueError(f"Invalid AWS region {region!r}: expected format like 'us-east-1'") 

61 

62 

63def _validate_port(port: int | str, *, what: str) -> int: 

64 try: 

65 value = int(port) 

66 except (TypeError, ValueError) as exc: 

67 raise ValueError(f"Invalid {what} {port!r}: must be an integer") from exc 

68 if not 1 <= value <= 65535: 

69 raise ValueError(f"Invalid {what} {value}: must be between 1 and 65535") 

70 return value 

71 

72 

73def endpoint_host(endpoint: str) -> str: 

74 """Extract the bare hostname from an EKS endpoint URL (or a bare host). 

75 

76 ``https://ABC123.gr7.us-east-1.eks.amazonaws.com`` -> ``ABC123.gr7...``. 

77 

78 Uses ``netloc`` rather than ``urlparse(...).hostname`` because the latter 

79 lowercases the host — EKS endpoint IDs are case-sensitive in the server 

80 certificate SAN, and this value becomes kubectl's ``--tls-server-name``. 

81 """ 

82 netloc = urlparse(endpoint).netloc if "://" in endpoint else endpoint 

83 # Strip any userinfo and :port, preserving original case. 

84 host = netloc.split("@")[-1].split(":")[0] 

85 if not host: 

86 raise ValueError(f"Could not parse a hostname from endpoint {endpoint!r}") 

87 return host 

88 

89 

90def build_remote_host_port_forward_command( 

91 instance_id: str, 

92 remote_host: str, 

93 local_port: int | str, 

94 region: str, 

95 remote_port: int | str = 443, 

96) -> list[str]: 

97 """Build a validated ``aws ssm start-session`` argv for a remote-host tunnel. 

98 

99 Forwards ``127.0.0.1:<local_port>`` through ``instance_id`` to 

100 ``remote_host:<remote_port>`` (default 443, the EKS API port). 

101 """ 

102 _validate_instance_id(instance_id) 

103 _validate_host(remote_host) 

104 _validate_region(region) 

105 local = _validate_port(local_port, what="local port") 

106 remote = _validate_port(remote_port, what="remote port") 

107 

108 parameters = json.dumps( 

109 { 

110 "host": [remote_host], 

111 "portNumber": [str(remote)], 

112 "localPortNumber": [str(local)], 

113 } 

114 ) 

115 return [ 

116 "aws", 

117 "ssm", 

118 "start-session", 

119 "--target", 

120 instance_id, 

121 "--region", 

122 region, 

123 "--document-name", 

124 REMOTE_HOST_DOCUMENT, 

125 "--parameters", 

126 parameters, 

127 ] 

128 

129 

130def _process_output_detail(stdout: bytes | None, stderr: bytes | None) -> str: 

131 output = stderr or stdout 

132 return output.decode("utf-8", "replace").strip()[:2000] if output else "(no output)" 

133 

134 

135def _signal_api_tunnel_tree( 

136 proc: subprocess.Popen[bytes], 

137 *, 

138 force: bool, 

139 wait_seconds: float, 

140) -> None: 

141 """Signal the process group containing the AWS CLI and Session Manager plugin.""" 

142 pid = getattr(proc, "pid", None) 

143 if not isinstance(pid, int): 

144 (proc.kill if force else proc.terminate)() 

145 return 

146 

147 if os.name == "nt": 

148 taskkill = shutil.which("taskkill.exe") or shutil.which("taskkill") 

149 if taskkill is not None: 

150 command = [taskkill, "/PID", str(pid), "/T"] 

151 if force: 

152 command.append("/F") 

153 try: 

154 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - resolved Windows utility and numeric child PID 

155 command, 

156 capture_output=True, 

157 check=False, 

158 timeout=wait_seconds, 

159 creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), 

160 ) 

161 except OSError, subprocess.TimeoutExpired: 

162 pass 

163 else: 

164 if result.returncode == 0: 

165 return 

166 # Preserve the tracked root PID through the graceful wait. If graceful 

167 # tree termination fails, stop_api_tunnel escalates to /T /F before 

168 # any wrapper-only fallback that could orphan the plugin. 

169 if not force: 

170 return 

171 proc.kill() 

172 return 

173 

174 try: 

175 os.killpg(pid, signal.SIGKILL if force else signal.SIGTERM) 

176 except OSError: 

177 if proc.poll() is None: 

178 (proc.kill if force else proc.terminate)() 

179 

180 

181def stop_api_tunnel( 

182 proc: subprocess.Popen[bytes], 

183 *, 

184 wait_seconds: float = _DEFAULT_STOP_TIMEOUT_SECONDS, 

185) -> tuple[bytes, bytes]: 

186 """Terminate and reap the complete tunnel process group within bounded waits.""" 

187 if wait_seconds <= 0: 

188 raise ValueError("wait_seconds must be positive") 

189 if proc.poll() is None: 

190 _signal_api_tunnel_tree(proc, force=False, wait_seconds=wait_seconds) 

191 try: 

192 return proc.communicate(timeout=wait_seconds) 

193 except subprocess.TimeoutExpired: 

194 # The AWS wrapper may have exited while session-manager-plugin still 

195 # owns the listener and inherited pipes, so force the whole group even 

196 # when proc.poll() now reports a wrapper exit. 

197 _signal_api_tunnel_tree(proc, force=True, wait_seconds=wait_seconds) 

198 try: 

199 return proc.communicate(timeout=wait_seconds) 

200 except subprocess.TimeoutExpired as exc: 

201 for stream in (proc.stdout, proc.stderr): 

202 if stream is not None: 

203 stream.close() 

204 if proc.poll() is None: 

205 proc.kill() 

206 try: 

207 proc.wait(timeout=wait_seconds) 

208 except subprocess.TimeoutExpired as wait_exc: 

209 raise RuntimeError( 

210 "SSM tunnel process tree did not exit after forced termination" 

211 ) from wait_exc 

212 raise RuntimeError( 

213 "SSM tunnel process tree retained inherited output pipes after forced termination" 

214 ) from exc 

215 

216 

217def exited_api_tunnel_detail(proc: subprocess.Popen[bytes]) -> str | None: 

218 """Return diagnostics for an exited tunnel, or ``None`` while it is running.""" 

219 returncode = proc.poll() 

220 if returncode is None: 

221 return None 

222 try: 

223 stdout, stderr = stop_api_tunnel(proc) 

224 except RuntimeError as exc: 

225 return f"exit code {returncode}; process-tree cleanup failed: {exc}" 

226 return f"exit code {returncode}; {_process_output_detail(stdout, stderr)}" 

227 

228 

229def start_api_tunnel( 

230 instance_id: str, 

231 endpoint: str, 

232 local_port: int, 

233 region: str, 

234 *, 

235 ready_wait_seconds: float = _DEFAULT_READY_TIMEOUT_SECONDS, 

236 ready_poll_seconds: float = _DEFAULT_READY_POLL_SECONDS, 

237 connect_timeout_seconds: float = _DEFAULT_CONNECT_TIMEOUT_SECONDS, 

238) -> subprocess.Popen[bytes]: 

239 """Launch an SSM tunnel and return only after its IPv4 listener accepts. 

240 

241 ``ready_wait_seconds`` is a compatibility-preserving name for the bounded 

242 listener-readiness timeout. Process exit and timeout paths include captured 

243 Session Manager diagnostics and always reap the complete process group. 

244 """ 

245 if ready_wait_seconds < 0: 

246 raise ValueError("ready_wait_seconds must be non-negative") 

247 if ready_poll_seconds <= 0: 

248 raise ValueError("ready_poll_seconds must be positive") 

249 if connect_timeout_seconds <= 0: 

250 raise ValueError("connect_timeout_seconds must be positive") 

251 

252 host = endpoint_host(endpoint) 

253 cmd = build_remote_host_port_forward_command(instance_id, host, local_port, region) 

254 popen_kwargs: dict[str, Any] = { 

255 "stdout": subprocess.PIPE, 

256 "stderr": subprocess.PIPE, 

257 "start_new_session": os.name == "posix", 

258 } 

259 if os.name == "nt": 

260 popen_kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) 

261 try: 

262 proc = subprocess.Popen( # nosemgrep: dangerous-subprocess-use-audit - argv validated by builder; list form, no shell=True 

263 cmd, 

264 **popen_kwargs, 

265 ) 

266 except FileNotFoundError as exc: 

267 raise RuntimeError( 

268 "AWS CLI not found. Install the AWS CLI and the Session Manager plugin." 

269 ) from exc 

270 

271 deadline = time.monotonic() + ready_wait_seconds 

272 try: 

273 while True: 

274 if detail := exited_api_tunnel_detail(proc): 

275 raise RuntimeError( 

276 "SSM port-forwarding session failed to start. Ensure the Session " 

277 "Manager plugin is installed and the target instance can reach the " 

278 f"cluster endpoint. Details: {detail}" 

279 ) 

280 try: 

281 connection = socket.create_connection( 

282 (LOCAL_TUNNEL_HOST, local_port), 

283 timeout=connect_timeout_seconds, 

284 ) 

285 except OSError as exc: 

286 last_error = f"{type(exc).__name__}: {exc}" 

287 else: 

288 connection.close() 

289 detail = exited_api_tunnel_detail(proc) 

290 if detail is None: 

291 return proc 

292 raise RuntimeError(f"SSM port-forwarding session exited during readiness: {detail}") 

293 

294 remaining = deadline - time.monotonic() 

295 if remaining <= 0: 

296 stdout, stderr = stop_api_tunnel(proc) 

297 raise RuntimeError( 

298 "SSM port-forwarding session did not accept local connections on " 

299 f"{LOCAL_TUNNEL_HOST}:{local_port} within {ready_wait_seconds:.1f}s. " 

300 f"Last error: {last_error}. SSM output: " 

301 f"{_process_output_detail(stdout, stderr)}" 

302 ) 

303 time.sleep(min(ready_poll_seconds, remaining)) 

304 except BaseException as exc: 

305 try: 

306 stop_api_tunnel(proc) 

307 except Exception as cleanup_exc: 

308 raise RuntimeError( 

309 f"SSM tunnel startup failed and process-tree cleanup also failed: {cleanup_exc}" 

310 ) from exc 

311 raise