Coverage for gco / eks_capabilities_config.py: 100.00%

1"""Schema, defaults and validation for the ``eks_capabilities`` cdk.json block. 

2 

3EKS Capabilities are AWS-managed installations of AWS Controllers for 

4Kubernetes (ACK) and kro that run in an AWS-owned account and attach to a 

5cluster as ``AWS::EKS::Capability`` resources. GCO exposes them as opt-in 

6per-type knobs: every type is **off by default**, and each enabled type 

7synthesizes one capability IAM role and one capability per selected regional 

8cluster. Argo CD is deliberately not offered here: GCO installs and runs it 

9in the cluster itself (``helm.argocd``, see ``gco.argocd_config``). 

10 

11This module owns the block's shape so the config loader, the regional stack, 

12the CLI and the live-validation harness all agree on names and defaults. Like 

13``gco.enablement_overrides`` and ``gco.inference_proxy_config`` it lives at 

14the package top level and imports nothing from ``aws_cdk``: the ``gco`` CLI 

15runs without the CDK toolchain installed, so it must be able to read the 

16block without pulling ``gco.config`` (whose package import loads the CDK 

17app). It deliberately raises :class:`EksCapabilitiesConfigError` (a 

18``ValueError``) rather than the loader's ``ConfigValidationError``; the 

19loader re-raises with its own exception type. 

20""" 

21 

22from __future__ import annotations 

23 

24import copy 

25import json 

26import re 

27from collections.abc import Iterable, Mapping 

28from typing import Any 

29 

30#: cdk.json context key. 

31EKS_CAPABILITIES_CONTEXT_KEY = "eks_capabilities" 

32 

33#: Run-scoped CDK context key carrying a JSON object deep-merged over the 

34#: cdk.json block for one deploy (``cdk deploy --context 

35#: eks_capabilities_overrides='{"kro": {"enabled": true}}'``). The 

36#: capabilities sibling of ``helm_enabled_overrides`` / ``feature_enabled_overrides``: 

37#: the live-validation and example harnesses, whose preflight requires a clean 

38#: worktree, enable capabilities for one run this way. 

39EKS_CAPABILITIES_OVERRIDES_CONTEXT_KEY = "eks_capabilities_overrides" 

40 

41#: Capability types GCO can attach, keyed the way cdk.json spells them. 

42EKS_CAPABILITY_TYPES: tuple[str, ...] = ("ack", "kro") 

43 

44#: cdk.json spelling -> ``AWS::EKS::Capability`` ``Type`` value. 

45CAPABILITY_TYPE_API_NAMES: dict[str, str] = {"ack": "ACK", "kro": "KRO"} 

46 

47#: Session name the EKS kro capability sets when it assumes its role. The 

48#: capability's Kubernetes user is therefore 

49#: ``arn:<partition>:sts::<account>:assumed-role/<role name>/KRO`` — the 

50#: subject GCO's tenant RBAC for kro binds (AWS docs, "Security 

51#: considerations for EKS Capabilities"). 

52KRO_SESSION_NAME = "KRO" 

53 

54#: A managed IAM policy ARN: an AWS managed policy (``iam::aws:policy/...``) 

55#: or a customer managed one in a 12-digit account. 

56_POLICY_ARN_RE = re.compile(r"arn:aws[a-z-]*:iam::(aws|\d{12}):policy/[\w+=,.@/-]+") 

57 

58#: The full block with every knob at its default. ``regions: []`` means every 

59#: regional deployment region; a non-empty list restricts the type to that 

60#: subset of clusters. 

61EKS_CAPABILITIES_DEFAULTS: dict[str, Any] = { 

62 "ack": { 

63 "enabled": False, 

64 "regions": [], 

65 # ACK service controllers to leave out (ACK's disabledServices). 

66 "disabled_services": [], 

67 "enable_cross_namespace": False, 

68 # IAM roles the capability role may assume (ACK IAM Role Selectors, 

69 # the documented least-privilege model). Empty grants nothing. 

70 "assume_role_arns": [], 

71 # Managed IAM policies attached to the capability role itself (the 

72 # documented "simple permission setup"). Empty grants nothing: ACK 

73 # can call no AWS API until one of these two lists says otherwise. 

74 "iam_policy_arns": [], 

75 }, 

76 "kro": { 

77 "enabled": False, 

78 "regions": [], 

79 }, 

80} 

81 

82 

83class EksCapabilitiesConfigError(ValueError): 

84 """Raised when the ``eks_capabilities`` block is malformed.""" 

85 

86 

87def _deep_merge(defaults: Mapping[str, Any], overrides: Mapping[str, Any]) -> dict[str, Any]: 

88 merged: dict[str, Any] = copy.deepcopy(dict(defaults)) 

89 for key, value in overrides.items(): 

90 if isinstance(value, Mapping) and isinstance(merged.get(key), Mapping): 

91 merged[key] = _deep_merge(merged[key], value) 

92 else: 

93 merged[key] = copy.deepcopy(value) 

94 return merged 

95 

96 

97def parse_eks_capabilities_overrides(raw: object) -> dict[str, Any]: 

98 """Parse the ``eks_capabilities_overrides`` context value into a mapping. 

99 

100 Accepts a JSON object string (the only shape ``cdk --context`` can carry) 

101 or a mapping (cdk.json-style); ``None`` and ``""`` mean no overrides. The 

102 result is validated later as part of the merged block, so this only 

103 checks the shape. 

104 """ 

105 if raw is None: 

106 return {} 

107 if isinstance(raw, Mapping): 

108 return copy.deepcopy(dict(raw)) 

109 if isinstance(raw, str): 

110 if not raw.strip(): 

111 return {} 

112 try: 

113 parsed = json.loads(raw) 

114 except json.JSONDecodeError as exc: 

115 raise EksCapabilitiesConfigError( 

116 f"{EKS_CAPABILITIES_OVERRIDES_CONTEXT_KEY} must be a JSON object: {exc}" 

117 ) from exc 

118 if not isinstance(parsed, Mapping): 

119 raise EksCapabilitiesConfigError( 

120 f"{EKS_CAPABILITIES_OVERRIDES_CONTEXT_KEY} must be a JSON object, got " 

121 f"{type(parsed).__name__}" 

122 ) 

123 return dict(parsed) 

124 raise EksCapabilitiesConfigError( 

125 f"{EKS_CAPABILITIES_OVERRIDES_CONTEXT_KEY} must be a JSON object string or an object, " 

126 f"got {type(raw).__name__}" 

127 ) 

128 

129 

130def merge_eks_capabilities_overrides(raw: object, overrides: Mapping[str, Any]) -> object: 

131 """Deep-merge run-scoped ``overrides`` over the raw cdk.json block. 

132 

133 Returns ``raw`` untouched when there is nothing to merge (so an absent 

134 block stays absent), the overrides alone when the block is absent, and 

135 the merged mapping otherwise. A non-mapping block is returned as-is so 

136 validation reports it by name. 

137 """ 

138 if not overrides: 

139 return raw 

140 if raw is None: 

141 return copy.deepcopy(dict(overrides)) 

142 if not isinstance(raw, Mapping): 

143 return raw 

144 return _deep_merge(raw, overrides) 

145 

146 

147def normalize_eks_capabilities_config(raw: object) -> dict[str, Any]: 

148 """Return the block with defaults filled in (does not validate). 

149 

150 ``None`` and ``{}`` both mean "everything off". Unknown keys are kept so 

151 :func:`validate_eks_capabilities_config` can reject them by name. 

152 """ 

153 if raw is None: 

154 return copy.deepcopy(EKS_CAPABILITIES_DEFAULTS) 

155 if not isinstance(raw, Mapping): 

156 raise EksCapabilitiesConfigError( 

157 f"{EKS_CAPABILITIES_CONTEXT_KEY} must be an object, got {type(raw).__name__}" 

158 ) 

159 return _deep_merge(EKS_CAPABILITIES_DEFAULTS, raw) 

160 

161 

162def _require_bool(value: object, path: str) -> None: 

163 if type(value) is not bool: 

164 raise EksCapabilitiesConfigError(f"{path} must be a boolean, got {value!r}") 

165 

166 

167def _require_string_list(value: object, path: str) -> list[str]: 

168 if not isinstance(value, list) or not all(isinstance(item, str) for item in value): 

169 raise EksCapabilitiesConfigError(f"{path} must be a list of strings, got {value!r}") 

170 if any(not item.strip() for item in value): 

171 raise EksCapabilitiesConfigError(f"{path} must not contain empty strings") 

172 if len(set(value)) != len(value): 

173 raise EksCapabilitiesConfigError(f"{path} lists a value twice") 

174 return value 

175 

176 

177def _require_arn(value: object, path: str, *, prefix: str = "arn:") -> None: 

178 if not isinstance(value, str) or not value.startswith(prefix): 

179 raise EksCapabilitiesConfigError( 

180 f"{path} must be an ARN starting with {prefix!r}, got {value!r}" 

181 ) 

182 

183 

184def _reject_unknown_keys(block: Mapping[str, Any], allowed: Iterable[str], path: str) -> None: 

185 unknown = sorted(str(key) for key in block if key not in set(allowed)) 

186 if unknown: 

187 raise EksCapabilitiesConfigError( 

188 f"{path} contains unknown key(s): {', '.join(unknown)}; " 

189 f"allowed keys: {', '.join(sorted(allowed))}" 

190 ) 

191 

192 

193def _validate_regions(value: object, path: str, deployment_regions: Iterable[str] | None) -> None: 

194 regions = _require_string_list(value, path) 

195 if deployment_regions is None: 

196 return 

197 unknown = sorted(set(regions) - set(deployment_regions)) 

198 if unknown: 

199 raise EksCapabilitiesConfigError( 

200 f"{path} names region(s) that are not regional deployment regions: {', '.join(unknown)}" 

201 ) 

202 

203 

204def validate_eks_capabilities_config( 

205 raw: object, 

206 deployment_regions: Iterable[str] | None = None, 

207) -> dict[str, Any]: 

208 """Validate the raw cdk.json block and return it normalized. 

209 

210 ``deployment_regions`` (the regional deployment regions) lets the 

211 per-type ``regions`` subsets be checked; pass ``None`` to skip that check 

212 (e.g. when the caller has no region list yet). 

213 """ 

214 if raw is None: 

215 return copy.deepcopy(EKS_CAPABILITIES_DEFAULTS) 

216 if not isinstance(raw, Mapping): 

217 raise EksCapabilitiesConfigError( 

218 f"{EKS_CAPABILITIES_CONTEXT_KEY} must be an object, got {type(raw).__name__}" 

219 ) 

220 _reject_unknown_keys(raw, EKS_CAPABILITY_TYPES, EKS_CAPABILITIES_CONTEXT_KEY) 

221 for type_name in EKS_CAPABILITY_TYPES: 

222 block = raw.get(type_name) 

223 if block is not None and not isinstance(block, Mapping): 

224 raise EksCapabilitiesConfigError( 

225 f"{EKS_CAPABILITIES_CONTEXT_KEY}.{type_name} must be an object, got {block!r}" 

226 ) 

227 if isinstance(block, Mapping): 

228 _reject_unknown_keys( 

229 block, 

230 EKS_CAPABILITIES_DEFAULTS[type_name].keys(), 

231 f"{EKS_CAPABILITIES_CONTEXT_KEY}.{type_name}", 

232 ) 

233 config = _deep_merge(EKS_CAPABILITIES_DEFAULTS, raw) 

234 regions_list = list(deployment_regions) if deployment_regions is not None else None 

235 

236 for type_name in EKS_CAPABILITY_TYPES: 

237 block = config[type_name] 

238 path = f"{EKS_CAPABILITIES_CONTEXT_KEY}.{type_name}" 

239 _require_bool(block["enabled"], f"{path}.enabled") 

240 _validate_regions(block["regions"], f"{path}.regions", regions_list) 

241 

242 ack = config["ack"] 

243 _require_string_list(ack["disabled_services"], "eks_capabilities.ack.disabled_services") 

244 _require_bool(ack["enable_cross_namespace"], "eks_capabilities.ack.enable_cross_namespace") 

245 for index, arn in enumerate( 

246 _require_string_list(ack["assume_role_arns"], "eks_capabilities.ack.assume_role_arns") 

247 ): 

248 _require_arn(arn, f"eks_capabilities.ack.assume_role_arns[{index}]") 

249 for index, arn in enumerate( 

250 _require_string_list(ack["iam_policy_arns"], "eks_capabilities.ack.iam_policy_arns") 

251 ): 

252 if not _POLICY_ARN_RE.fullmatch(arn): 

253 raise EksCapabilitiesConfigError( 

254 f"eks_capabilities.ack.iam_policy_arns[{index}] must be a managed IAM policy " 

255 f"ARN (arn:<partition>:iam::aws:policy/... or arn:<partition>:iam::" 

256 f"<account>:policy/...), got {arn!r}" 

257 ) 

258 

259 return config 

260 

261 

262def capability_enabled_in_region(config: Mapping[str, Any], type_name: str, region: str) -> bool: 

263 """True when ``type_name`` is on and applies to ``region`` (empty regions = all).""" 

264 block = config.get(type_name) 

265 if not isinstance(block, Mapping) or block.get("enabled") is not True: 

266 return False 

267 regions = block.get("regions") or [] 

268 return not regions or region in regions 

269 

270 

271def enabled_capability_types(config: Mapping[str, Any], region: str) -> list[str]: 

272 """The cdk.json type names enabled for ``region``, in canonical order.""" 

273 return [ 

274 name for name in EKS_CAPABILITY_TYPES if capability_enabled_in_region(config, name, region) 

275 ] 

276 

277 

278def kro_kubernetes_username(*, partition: str, account: str, role_name: str) -> str: 

279 """The Kubernetes user the EKS kro capability acts as. 

280 

281 Every argument may be a CloudFormation token (the stack passes its 

282 partition, account and the generated role name); they are only ever 

283 concatenated. 

284 """ 

285 return f"arn:{partition}:sts::{account}:assumed-role/{role_name}/{KRO_SESSION_NAME}" 

286 

287 

288#: kubectl-applier tokens rendered for the capability manifests. 

289#: ``07-kro-tenant-access.yaml`` is gated on the kro user name. 

290EKS_CAPABILITIES_MANIFEST_TOKENS: tuple[str, ...] = ("{{KRO_CAPABILITY_USERNAME}}",) 

291 

292 

293def compute_eks_capabilities_replacements(*, kro_username: str | None) -> dict[str, str]: 

294 """Build the kubectl-applier replacements for the capability manifests. 

295 

296 Shared by the regional stack (deploy) and the tests so both render 

297 ``07-kro-tenant-access.yaml`` identically. The file resolves only when 

298 ``kro_username`` is given, i.e. the kro capability is enabled for this 

299 region. An absent key is the off switch: the applier gates the manifest 

300 out of the plan while its token stays unresolved and prunes the inventory 

301 registered for that token, exactly like the FSx, Valkey and observability 

302 gates. 

303 """ 

304 if kro_username is None: 

305 return {} 

306 return {"{{KRO_CAPABILITY_USERNAME}}": kro_username}