def handler()

in src/release_artifacts_resources/ios/cdk/cdk/credential_rotation/lambda_functions/src/handler.py [0:0]
28 lines of code
11 McCabe index (conditional complexity)

def handler(event, context, *, iam=None, sts=None, secretsmanager=None):
    """
    Invoked with the following event structure:
    ```
    {
      "sources": [
        {
          "type": "aws_session_cred",
          "description": "Temporary AWS Credentials to upload the release artifacts to S3 and invalidate Cloudfront",
          "configuration": {
            "user_env_variable": "IAM_USERNAME",
            "iam_role_env_variable": "IAM_ROLE"
          },
          "destination": {
            "specifier": "aws-sdk-ios-cci",
            "mapping_to_destination": [
              {
                "destination_key_name": "XCF_ACCESS_KEY_ID",
                "result_value_key": "access_key_id"
              },
              {
                "destination_key_name": "XCF_SECRET_ACCESS_KEY",
                "result_value_key": "secret_access_key"
              },
              {
                "destination_key_name": "XCF_SESSION_TOKEN",
                "result_value_key": "session_token"
              }
            ]
          }
        },
        {
          "type": "secrets_manager",
          "description": "",
          "configuration": {
            "secret_key_env_variable": "GITHUB_CREDENTIALS_SECRET"
          },
          "destination": {
            "specifier": "aws-sdk-ios-cci",
            "mapping_to_destination": [
              {
                "destination_key_name": "GITHUB_SPM_TOKEN",
                "result_value_key": "GITHUB_SPM_TOKEN"
              },
              {
                "destination_key_name": "GITHUB_SPM_USER",
                "result_value_key": "GITHUB_SPM_USER"
              }
            ]
          }
        },
        {
          "type": "lambda_env_variables",
          "description": "",
          "configuration": {
            "lambda_env_var_key": "SPM_S3_BUCKET_NAME"
          },
          "destination": {
            "specifier": "aws-sdk-ios-cci",
            "mapping_to_destination": [
              {
                "destination_key_name": "XCF_S3_BUCKET_NAME"
              }
            ]
          }
        }
      ],
      "destinations": {
        "aws-sdk-ios-cci": {
          "type": "cci_env_variable",
          "description": "Circle CI environment variable for AWS SDK iOS repo",
          "github_path": "aws-amplify/aws-sdk-ios",
          "circleci_api_token_secret_id_lambda_env_var_key": "CIRCLE_CI_IOS_SDK_API_TOKEN"
        }
      }
    }
    ```
    """

    sources = event["sources"]
    destinations = event["destinations"]

    destination_values_map = {}
    for source in sources:
        source_type = source["type"]
        destination_specifier = source["destination"]["specifier"]
        destination_mapping = source["destination"]["mapping_to_destination"]
        configuration = source["configuration"]

        source_map = {}

        if source_type == SourceType.AWS_SESSION_CREDENTIALS:
            source_map = aws_session_credential_source.generate_session_credentials(configuration)

        elif source_type == SourceType.SECRETS_MANAGER:
            source_map = secrets_data_source.retrieve_secrets(configuration)

        elif source_type == SourceType.LAMBDA_ENVIRONMENT_VARIABLE:
            source_map = lambda_env_var_data_source.retrieve_lambda_env_var_value(configuration)

        mapped_result = {}
        for item in destination_mapping:
            destination_key_name = item["destination_key_name"]
            result_value_key = item.get("result_value_key", "result")
            if result_value_key in source_map:
                mapped_result[destination_key_name] = source_map[result_value_key]

        destination_values_map.setdefault(destination_specifier, {}).update(mapped_result)

    for name, destination_configuration in destinations.items():

        destination_type = destination_configuration["type"]
        mapped_result = destination_values_map.get(name, {})

        if destination_type == DestinationType.CIRCLECI_ENVIRONMENT_VARIABLE:
            circleci.update_environment_variables(mapped_result, destination_configuration)