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://localhost:<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. 

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 re 

22import subprocess 

23import time 

24from urllib.parse import urlparse 

25 

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

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

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

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

30 

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

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

33REMOTE_HOST_DOCUMENT = "AWS-StartPortForwardingSessionToRemoteHost" 

34 

35 

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

37 if not _INSTANCE_RE.match(instance_id): 

38 raise ValueError( 

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

40 ) 

41 

42 

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

44 if not _HOST_RE.match(host): 

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

46 

47 

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

49 if not _REGION_RE.match(region): 

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

51 

52 

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

54 try: 

55 value = int(port) 

56 except (TypeError, ValueError) as exc: 

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

58 if not 1 <= value <= 65535: 

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

60 return value 

61 

62 

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

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

65 

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

67 

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

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

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

71 """ 

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

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

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

75 if not host: 

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

77 return host 

78 

79 

80def build_remote_host_port_forward_command( 

81 instance_id: str, 

82 remote_host: str, 

83 local_port: int | str, 

84 region: str, 

85 remote_port: int | str = 443, 

86) -> list[str]: 

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

88 

89 Forwards ``localhost:<local_port>`` through ``instance_id`` to 

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

91 """ 

92 _validate_instance_id(instance_id) 

93 _validate_host(remote_host) 

94 _validate_region(region) 

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

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

97 

98 parameters = json.dumps( 

99 { 

100 "host": [remote_host], 

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

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

103 } 

104 ) 

105 return [ 

106 "aws", 

107 "ssm", 

108 "start-session", 

109 "--target", 

110 instance_id, 

111 "--region", 

112 region, 

113 "--document-name", 

114 REMOTE_HOST_DOCUMENT, 

115 "--parameters", 

116 parameters, 

117 ] 

118 

119 

120def start_api_tunnel( 

121 instance_id: str, 

122 endpoint: str, 

123 local_port: int, 

124 region: str, 

125 *, 

126 ready_wait_seconds: float = 3.0, 

127) -> subprocess.Popen[bytes]: 

128 """Launch a background SSM tunnel to the cluster API endpoint. 

129 

130 Returns the :class:`subprocess.Popen` handle so the caller can terminate it 

131 when the port-forward session ends. Raises ``RuntimeError`` if the session 

132 process exits immediately (e.g. missing Session Manager plugin or IAM). 

133 """ 

134 host = endpoint_host(endpoint) 

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

136 try: 

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

138 cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE 

139 ) 

140 except FileNotFoundError as exc: 

141 raise RuntimeError( 

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

143 ) from exc 

144 # Give the session a moment to establish; if it died immediately, surface it. 

145 time.sleep(ready_wait_seconds) 

146 if proc.poll() is not None: 

147 _, stderr = proc.communicate() 

148 detail = stderr.decode("utf-8", "replace").strip() if stderr else "(no output)" 

149 raise RuntimeError( 

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

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

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

153 ) 

154 return proc