Coverage for cli/kubectl_helpers.py: 93.02%

1""" 

2Shared kubectl helper utilities for GCO CLI. 

3 

4Provides common kubectl operations used across multiple CLI modules 

5to reduce code duplication and ensure consistent error handling. 

6""" 

7 

8import logging 

9import re 

10import subprocess 

11 

12logger = logging.getLogger(__name__) 

13 

14# EKS cluster names: 1-100 chars, alphanumeric and hyphens only. 

15# AWS region names: e.g. us-east-1, ap-southeast-2, eu-central-1. 

16_CLUSTER_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9\-]{0,99}$") 

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

18 

19 

20def _validate_cluster_name(cluster_name: str) -> None: 

21 """Raise ValueError if cluster_name contains characters outside the EKS naming rules.""" 

22 if not _CLUSTER_NAME_RE.match(cluster_name): 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

23 raise ValueError( 

24 f"Invalid cluster name {cluster_name!r}: must be 1-100 alphanumeric/hyphen characters" 

25 ) 

26 

27 

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

29 """Raise ValueError if region does not match the standard AWS region pattern.""" 

30 if not _REGION_RE.match(region): 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true

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

32 

33 

34def update_kubeconfig(cluster_name: str, region: str) -> None: 

35 """Update kubeconfig for an EKS cluster. 

36 

37 Args: 

38 cluster_name: Name of the EKS cluster 

39 region: AWS region where the cluster is located 

40 

41 Raises: 

42 ValueError: If cluster_name or region contain unexpected characters 

43 RuntimeError: If the kubeconfig update fails 

44 FileNotFoundError: If the AWS CLI is not installed 

45 """ 

46 _validate_cluster_name(cluster_name) 

47 _validate_region(region) 

48 

49 cmd = [ 

50 "aws", 

51 "eks", 

52 "update-kubeconfig", 

53 "--name", 

54 cluster_name, 

55 "--region", 

56 region, 

57 ] 

58 

59 try: 

60 result = subprocess.run( 

61 cmd, capture_output=True, text=True 

62 ) # nosemgrep: dangerous-subprocess-use-audit - inputs validated above; list form, no shell=True 

63 if result.returncode != 0: 

64 raise RuntimeError(f"Failed to update kubeconfig: {result.stderr}") 

65 except subprocess.CalledProcessError as e: 

66 raise RuntimeError(f"Failed to update kubeconfig: {e.stderr}") from e 

67 except FileNotFoundError as e: 

68 raise RuntimeError( 

69 "AWS CLI not found. Please install the AWS CLI and ensure it's in your PATH." 

70 ) from e 

71 

72 

73# --------------------------------------------------------------------------- 

74# Port-forward + endpoint helpers (used by `gco monitoring open`) 

75# --------------------------------------------------------------------------- 

76 

77# svc/name | service/name | pod/name | deploy/name | deployment/name, where the 

78# resource name follows the RFC 1123 rules kubectl accepts. 

79_PF_TARGET_RE = re.compile( 

80 r"^(svc|service|pod|deploy|deployment)/[a-z0-9]([a-z0-9.\-]{0,251}[a-z0-9])?$" 

81) 

82_NAMESPACE_RE = re.compile(r"^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?$") 

83 

84 

85def _validate_port(port: int | str, *, what: str = "port") -> int: 

86 """Return the port as an int in 1..65535 or raise ValueError.""" 

87 try: 

88 value = int(port) 

89 except (TypeError, ValueError) as exc: 

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

91 if not 1 <= value <= 65535: 

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

93 return value 

94 

95 

96def build_port_forward_command( 

97 namespace: str, 

98 target: str, 

99 local_port: int | str, 

100 remote_port: int | str, 

101 *, 

102 server: str | None = None, 

103 tls_server_name: str | None = None, 

104) -> list[str]: 

105 """Build a validated ``kubectl port-forward`` argv (list form, never a shell string). 

106 

107 ``target`` is a ``kind/name`` reference (e.g. ``svc/kube-prometheus-stack-grafana``). 

108 ``server`` / ``tls_server_name`` override the API endpoint and its TLS SNI — 

109 used when tunnelling to a private endpoint through an SSM local port, where 

110 kubectl talks to ``https://localhost:<port>`` but must present the real EKS 

111 hostname for certificate validation. 

112 """ 

113 if not _NAMESPACE_RE.match(namespace): 

114 raise ValueError(f"Invalid namespace {namespace!r}") 

115 if not _PF_TARGET_RE.match(target): 

116 raise ValueError( 

117 f"Invalid port-forward target {target!r}: expected kind/name " 

118 "(svc|service|pod|deploy|deployment)" 

119 ) 

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

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

122 

123 cmd = ["kubectl", "port-forward", "-n", namespace, target, f"{local}:{remote}"] 

124 if server is not None: 

125 if not server.startswith("https://"): 

126 raise ValueError(f"Invalid --server {server!r}: must start with https://") 

127 cmd += ["--server", server] 

128 if tls_server_name is not None: 

129 if not re.match(r"^[a-zA-Z0-9.\-]{1,255}$", tls_server_name): 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true

130 raise ValueError(f"Invalid --tls-server-name {tls_server_name!r}") 

131 cmd += ["--tls-server-name", tls_server_name] 

132 return cmd 

133 

134 

135def describe_cluster_access(cluster_name: str, region: str) -> dict[str, object]: 

136 """Return the EKS API endpoint and its public/private access posture. 

137 

138 Returns a dict with keys ``endpoint`` (str), ``public`` (bool), 

139 ``private`` (bool), and ``public_cidrs`` (list[str]). Used by 

140 ``gco monitoring open`` to decide whether a plain ``kubectl port-forward`` 

141 can reach the API server or whether an SSM/VPN/bastion path is required. 

142 """ 

143 _validate_cluster_name(cluster_name) 

144 _validate_region(region) 

145 

146 cmd = [ 

147 "aws", 

148 "eks", 

149 "describe-cluster", 

150 "--name", 

151 cluster_name, 

152 "--region", 

153 region, 

154 "--query", 

155 ( 

156 "cluster.{endpoint:endpoint," 

157 "public:resourcesVpcConfig.endpointPublicAccess," 

158 "private:resourcesVpcConfig.endpointPrivateAccess," 

159 "publicCidrs:resourcesVpcConfig.publicAccessCidrs}" 

160 ), 

161 "--output", 

162 "json", 

163 ] 

164 try: 

165 result = subprocess.run( 

166 cmd, capture_output=True, text=True 

167 ) # nosemgrep: dangerous-subprocess-use-audit - inputs validated above; list form, no shell=True 

168 except FileNotFoundError as exc: 

169 raise RuntimeError( 

170 "AWS CLI not found. Please install the AWS CLI and ensure it's in your PATH." 

171 ) from exc 

172 if result.returncode != 0: 

173 raise RuntimeError(f"Failed to describe cluster {cluster_name}: {result.stderr}") 

174 

175 import json 

176 

177 data = json.loads(result.stdout or "{}") 

178 return { 

179 "endpoint": data.get("endpoint") or "", 

180 "public": bool(data.get("public")), 

181 "private": bool(data.get("private")), 

182 "public_cidrs": data.get("publicCidrs") or [], 

183 }