-
Notifications
You must be signed in to change notification settings - Fork 91
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add config option to enable the encryption of AWS EKS secrets #2788
base: main
Are you sure you want to change the base?
Conversation
…y ARN to encrypt EKS cluster secrets
@viniciusdc Also in follow-up to your ask, it appears that re-deploying to set KMS encryption on an existing Nebari EKS Cluster, without previous encryption set, does succeed. However, attempting thereafter to re-deploy to remove the previously set EKS secrets encryption will fail as terraform attempts to delete and rebuild the EKS cluster but cannot due to existing node groups. |
session = aws_session(region=region) | ||
client = session.client("kms") | ||
# https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/kms/client/list_keys.html | ||
paginator = client.get_paginator("list_keys") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we actually need a paginator here? It doesn't seem like we are using it. It seems like lines 137-141 could be replaced by
kms_keys = [i["KeyId"] for i in client.list_keys().get("Keys")]
Which seems easier to read to me since it avoids nested loops.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
or if you want it more explicit
key_id_list = client.list_keys().get("Keys")
kms_keys = [i.get("KeyId") for i in key_id_list]
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
updated
for i in paginator.paginate() | ||
for j in i["Keys"] | ||
] | ||
return {i["KeyId"]: {k: i[k] for k in fields} for i in kms_keys if i["Enabled"]} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't feel like this line is super readable. I would suggest extracting it into a separate function.
class Kms_key:
Arn: str
KeyUsage: str
KeySpec: str
def check_kms_keys(key_ids: list[str], client: boto3.Client) -> dict[str, Kms_key]:
keys = []
for id in key_ids:
key = client.describe_key(KeyId=id).get("KeyMetadata")
if key.get("Enabled"):
keys.append(kms_key(
Arn=key.get("Arn"),
KeyUsage=key.get("KeyUsage"),
KeySpec=key.get("KeySpec"),
)
)
return keys
Would be way easier to follow and accomplish the same thing
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I do prefer this as well!
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I've made the function more readable but I did not separate into two functions.
Note: the filter on the dictionary serves no other purpose other than to avoid passing more data than necessary for validating keys for EKS encryption. It's possible this function could be utilized in the future for collecting metadata on kms keys employed for services other than EKS encryption.
This function could just as easily be written:
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")
kms_keys = {}
for key in client.list_keys().get("Keys"):
key_id = key["KeyId"]
key_data = client.describe_key(KeyId=key_id).get("KeyMetadata")
if key_data.get("Enabled"):
kms_keys[key_id] = key_data
return kms_keys
# check if kms key is valid | ||
available_kms_keys = amazon_web_services.kms_key_arns(data["region"]) | ||
if "eks_kms_arn" in data and data["eks_kms_arn"] is not None: | ||
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] | ||
# Symmetric KMS keys with Encrypt and decrypt key-usage have the SYMMETRIC_DEFAULT key-spec | ||
# EKS cluster encryption requires a Symmetric key that is set to encrypt and decrypt data | ||
if available_kms_keys[key_id]["KeySpec"] != "SYMMETRIC_DEFAULT": | ||
if available_kms_keys[key_id]["KeyUsage"] == "GENERATE_VERIFY_MAC": | ||
raise ValueError( | ||
f"Amazon Web Services KMS Key with ID {key_id} does not have KeyUsage set to 'Encrypt and decrypt' data" | ||
) | ||
elif available_kms_keys[key_id]["KeyUsage"] != "ENCRYPT_DECRYPT": | ||
raise ValueError( | ||
f"Amazon Web Services KMS Key with ID {key_id} is not of type Symmetric, and KeyUsage not set to 'Encrypt and decrypt' data" | ||
) | ||
else: | ||
raise ValueError( | ||
f"Amazon Web Services KMS Key with ID {key_id} is not of type Symmetric" | ||
) | ||
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()]}" | ||
) | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You could avoid a lot of nesting here by flipping the logic on these if statements. Something like:
available_kms_keys = amazon_web_services.kms_key_arns(data["region"])
# don't check if eks_kms_arn is not set
if "eks_kms_arn" not in data or data["eks_kms_arn"] is None:
return data
key_id = [
id for id in available_kms_keys.keys() if id in data["eks_kms_arn"]
]
# Raise error if key_id is not found in available_kms_keys
if (
len(key_id) != 1
or available_kms_keys[key_id[0]]["Arn"] != data["eks_kms_arn"]
):
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()]}"
)
key_id = key_id[0]
# EKS cluster encryption requires a Symmetric key that is set 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 of type Symmetric"
)
if available_kms_keys[key_id]["KeyUsage"] != "ENCRYPT_DECRYPT":
raise ValueError(
f"Amazon Web Services KMS Key with ID {key_id} KeyUsage not set to 'Encrypt and decrypt' data"
) available_kms_keys = amazon_web_services.kms_key_arns(data["region"])
# don't check if eks_kms_arn is not set
if "eks_kms_arn" not in data or data["eks_kms_arn"] is None:
return data
key_id = [
id for id in available_kms_keys.keys() if id in data["eks_kms_arn"]
]
# Raise error if key_id is not found in available_kms_keys
if (
len(key_id) != 1
or available_kms_keys[key_id[0]]["Arn"] != data["eks_kms_arn"]
):
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()]}"
)
key_id = key_id[0]
# EKS cluster encryption requires a Symmetric key that is set 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 of type Symmetric"
)
if available_kms_keys[key_id]["KeyUsage"] != "ENCRYPT_DECRYPT":
raise ValueError(
f"Amazon Web Services KMS Key with ID {key_id} KeyUsage not set to 'Encrypt and decrypt' data"
)
Has less deeply nested logic and is therefore easier to read and test.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@dcmcand
my hesitation to invoke return data
in numerous places in this section is that could cause issues in the future if additional validators are appended after the KMS-specific validator lines under _check_input
and validators are potentially skipped prematurely for aws services.
Hi @joneszc, thanks for checking that out! I was already expecting it to fail, but I had another thing in mind: they might be connected. Can you post a sanitized output of the terraform error and any error messages you might encounter in the CloudTrail history? I suspect you will find something related to the KMS key in there. the main reason for this request is to validate if it will be beneficial to have this as an immutable field or, depending on the error, we can add manual steps to the user in our docs to disable it. |
Reference Issues or PRs
Fixes #2681
Fixes #2746
Modifies PR#2723 (Failing Tests / Pytest)
Modifies PR#2752 (Failing Tests / Pytest)
What does this implement/fix?
Put a
x
in the boxes that applyTesting
How to test this PR?
Any other comments?
Allows user to set EKS encryption of secrets by specifying a KMS key ARN in nebari-config.yaml
The KMS key must meet the following conditions: