def lambda_handler()

in tools/rotate-secrets-manager-credentials/docker_hub_change_password.py [0:0]

• 31 lines of code
• 10 McCabe index (conditional complexity)

def lambda_handler(event, context):
    """
    Main lambda handler
    """
    logging.basicConfig(level=logging.INFO)
    logging.getLogger().setLevel(logging.INFO)
    arn = event['SecretId']
    token = event['ClientRequestToken']
    step = event['Step']
    logging.info('Step: ' + step)

    # Setup the client
    service_client = boto3.client('secretsmanager', endpoint_url=os.environ['SECRET_ENDPOINT_URL'])

    # Make sure the version is staged correctly
    metadata = service_client.describe_secret(SecretId=arn)
    if "RotationEnabled" in metadata and not metadata['RotationEnabled']:
        logging.error("Secret %s is not enabled for rotation" % arn)
        raise ValueError("Secret %s is not enabled for rotation" % arn)
    versions = metadata['VersionIdsToStages']
    if token not in versions:
        logging.error("Secret version %s has no stage for rotation of secret %s." % (token, arn))
        raise ValueError("Secret version %s has no stage for rotation of secret %s." % (token, arn))
    if "AWSCURRENT" in versions[token]:
        logging.info("Secret version %s already set as AWSCURRENT for secret %s." % (token, arn))
        return
    elif "AWSPENDING" not in versions[token]:
        logging.error("Secret version %s not set as AWSPENDING for rotation of secret %s." % (token, arn))
        raise ValueError("Secret version %s not set as AWSPENDING for rotation of secret %s." % (token, arn))


    if step == 'createSecret':
        return create_secret(service_client, arn, token)
    elif step == 'setSecret':
        return set_secret(service_client, arn, token)
    elif step == 'testSecret':
        return test_secret(service_client, arn, token)
    elif step == 'finishSecret':
        return finish_secret(service_client, arn, token)

    raise Exception('Unknown Step: ' + step)