Add validation checks for config option amazon_web_services.eks_kms_a…

…rn to ensure KMS-key ARN available

src/_nebari/provider/cloud/amazon_web_services.py

@@ -121,6 +121,30 @@ def instances(region: str) -> Dict[str, str]:
     return {t: t for t in instance_types}
 
 
+@functools.lru_cache()
+def kms_key_arns(region: str) -> Dict[str, dict]:
+    """Return dict of available/enabled KMS key IDs and associated KeyMetadata for the AWS region."""
+    session = aws_session(region=region)
+    client = session.client("kms")
+    paginator = client.get_paginator("list_keys")
+    schema = [
+        "Arn",
+        "KeyUsage",
+        "KeyState",
+        "Origin",
+        "KeyManager",
+        "KeySpec",
+        "EncryptionAlgorithms",
+        "MultiRegion",
+    ]
+    kms_keys = [
+        client.describe_key(KeyId=j["KeyId"]).get("KeyMetadata")
+        for i in paginator.paginate()
+        for j in i["Keys"]
+    ]
+    return {i["KeyId"]: {k: i[k] for k in schema} for i in kms_keys if i["Enabled"]}
+
+
 def aws_get_vpc_id(name: str, namespace: str, region: str) -> Optional[str]:
     """Return VPC ID for the EKS cluster namedd `{name}-{namespace}`."""
     cluster_name = f"{name}-{namespace}"

src/_nebari/stages/infrastructure/__init__.py

@@ -562,6 +562,25 @@ def _check_input(cls, data: Any) -> Any:
                         f"Amazon Web Services instance {node_group.instance} not one of available instance types={available_instances}"
                     )
 
+        # check if kms key is valid
+        available_kms_keys = amazon_web_services.kms_key_arns(data["region"])
+        if "eks_kms_arn" in data:
+            key_id = [id for id in available_kms_keys.keys() if id in data["eks_kms_arn"]]
+            if len(key_id) == 1 and available_kms_keys[key_id[0]]["Arn"] == data["eks_kms_arn"]:
+                key_id = key_id[0]
+                if available_kms_keys[key_id]["KeyUsage"] != "ENCRYPT_DECRYPT":
+                    raise ValueError(
+                        f"Amazon Web Services KMS Key with ID {key_id} does not have KeyUsage configured to encrypt and decrypt data"
+                    )
+                if available_kms_keys[key_id]["KeySpec"] != "SYMMETRIC_DEFAULT":
+                    raise ValueError(
+                        f"Amazon Web Services KMS Key with ID {key_id} is not a Symmetric key"
+                    )
+            else:
+                raise ValueError(
+                    f"Amazon Web Services KMS Key with ARN {data['eks_kms_arn']} not one of available/enabled keys={[v['Arn'] for v in available_kms_keys.values()]}"
+                )
+
         return data