feat(data_classes): Add SecretsManagerRotationEvent

[michaelbrewer]

Issue #, if available:

• feat(data-classes): add AWS Secrets Manager Rotation #985
• Reopening from feat(data-classes): Add support for SecretsManagerRotationEvent #986 (comment)

Description of changes:

Add a simple data class for SecretsManagerRotationEvent and enum SecretsManagerRotationEventStep

Example usage (in the future, check_if_staged_correctly and finish_secret could be included. Benefits noted in this example is the usage for structured logging, usage of Python 3.6+ features, reuse of the fetched metadata, use of enums and constants.

import os from typing import Dict, Any, Optional import boto3 from aws_lambda_powertools import Logger from aws_lambda_powertools.utilities.data_classes import event_source from aws_lambda_powertools.utilities.data_classes.secret_manager import ( SecretsManagerRotationEvent, SecretsManagerRotationEventStep, ) logger = Logger() session = boto3.Session() service_client = session.client("secretsmanager", endpoint_url=os.environ["SECRETS_MANAGER_ENDPOINT"]) AWS_CURRENT = "AWSCURRENT" AWS_PENDING = "AWSPENDING" @event_source(data_class=SecretsManagerRotationEvent) def handler(event: SecretsManagerRotationEvent, _): arn: str = event.secret_id token: str = event.client_request_token metadata = check_if_staged_correctly(arn, token) if not metadata: return step: SecretsManagerRotationEventStep = event.step if step == SecretsManagerRotationEventStep.CREATE_SECRET: create_secret(arn, token) elif step == SecretsManagerRotationEventStep.SET_SECRET: set_secret(arn, token) elif step == SecretsManagerRotationEventStep.TEST_SECRET: test_secret(arn, token) elif step == SecretsManagerRotationEventStep.FINISH_SECRET: finish_secret(metadata, arn, token) else: raise ValueError("Invalid step parameter") def check_if_staged_correctly(arn: str, token: str) -> Optional[Dict[str, Any]]: """Make sure the version is staged correctly""" metadata = service_client.describe_secret(SecretId=arn) if not metadata["RotationEnabled"]: logger.error(f"Secret {arn} is not enabled for rotation") raise ValueError(f"Secret {arn} is not enabled for rotation") versions = metadata["VersionIdsToStages"] if token not in versions: logger.error(f"Secret version {token} has no stage for rotation of secret {arn}.") raise ValueError(f"Secret version {token} has no stage for rotation of secret {arn}.") if AWS_CURRENT in versions[token]: logger.info(f"Secret version {token} already set as {AWS_CURRENT} for secret {arn}.") return None elif AWS_PENDING not in versions[token]: logger.error(f"Secret version {token} not set as {AWS_PENDING} for rotation of secret {arn}.") raise ValueError(f"Secret version {token} not set as {AWS_PENDING} for rotation of secret {arn}.") return metadata def create_secret(arn: str, token: str): ... def set_secret(arn: str, token: str): ... def test_secret(arn: str, token: str): ... def finish_secret(metadata: Dict[str, Any], arn: str, token: str): """Finish the secret""" current_version = None for version in metadata["VersionIdsToStages"]: if AWS_CURRENT in metadata["VersionIdsToStages"][version]: if version == token: # The correct version is already marked as current, return logger.info(f"finishSecret: Version {version} already marked as {AWS_CURRENT} for {arn}") return current_version = version break # Finalize by staging the secret version current service_client.update_secret_version_stage( SecretId=arn, VersionStage=AWS_CURRENT, MoveToVersionId=token, RemoveFromVersionId=current_version, ) logger.info(f"finishSecret: Successfully set {AWS_CURRENT} stage to version {token} for secret {arn}.")

Checklist

• Meet tenets criteria
• Update tests
• Update docs
• PR title follows conventional commit semantics

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[codecov-commenter]

Codecov Report

Merging #1031 (177c53a) into develop (22f8b9a) will increase coverage by 0.00%.
The diff coverage is 100.00%.

@@ Coverage Diff @@ ## develop #1031 +/- ## ======================================== Coverage 99.96% 99.96% ======================================== Files 119 120 +1 Lines 5365 5388 +23 Branches 612 616 +4 ======================================== + Hits 5363 5386 +23  Partials 2 2 

Impacted Files	Coverage Δ
...ities/data_classes/api_gateway_authorizer_event.py	100.00% <100.00%> (ø)
...owertools/utilities/data_classes/secret_manager.py	100.00% <100.00%> (ø)

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 22f8b9a...177c53a. Read the comment docs.

[michaelbrewer]

(closing for now and will reopening when capacity is available for this)