Sky Cone

There are four ways you can authorize applications to interact with your AWS AppSync GraphQL API, including API_KEY, AWS_IAM, OPENID_CONNECT, and AMAZON_COGNITO_USER_POOLS.

When using API key for authentication, we normally store the key in the Secrets Manager service. Then an application code could retrieve the key via API calls, and we can manage key rotation policy in the Secrets Manager as well. The AppSync service is not able to rotated the key automatically.

To automatically rotate the API key, we need to write code to achieve the goal.

To illustrate the process, I'm going to host the code in the Lambda function, and let the Secrets Manger to invoke the Lambda function with a pre-defined rotation interval.

Prerequisites
Make sure you have below resources in place before move to the rest of this post.

• An AppSync API
• A secret in Secrets Manager with a secret configured
• A VPC endpoint to access Secrets Manager without going out of VPC

Configure IAM role

First, we create an IAM role to be attached to a Lambda function, which we will create in the next step and use it for rotating the AppSyc API keys. This role should have below policy document associated to it:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "appsync:CreateApiKey",
                "appsync:DeleteApiKey",
                "secretsmanager:GetRandomPassword",
                "secretsmanager:GetSecretValue",
                "secretsmanager:DescribeSecret",
                "secretsmanager:PutSecretValue",
                "secretsmanager:UpdateSecretVersionStage"
            ],
            "Resource": "*"
        }
    ]
}

Also, you need to associate the manage policy "AWSLambdaVPCAccessExecutionRole", to make sure logs could be written into the CloudWatch Logs. This policy is also necessary for Lambda function, if it is deployed to a VPC environment.


Create Lambda Function

Second, create a Lambda function. Associate the IAM role created in the first step with this Lambda function. In this example, we use runtime of Python 3.8. You can use any language that is convenient for you. The Lambda is deployed into VPC.


Don't worry if you could not access above block.


Create two environment variables in the Lambda function. One uses a key called "AppSyncApiId", and its value is the API ID of the AppSync API. The other uses a key called "SECRETS_MANAGER_ENDPOINT", and its value is the DNS name of the VPC endpoint of the Secrets Manager.

The Python code:

import boto3
import json
import logging
import os


logger = logging.getLogger()
logger.setLevel(logging.INFO)

AppSyncApiId = os.environ['AppSyncApiId']

service_client = boto3.client(
    'secretsmanager', 
    endpoint_url=os.environ['SECRETS_MANAGER_ENDPOINT']
)
appsync_client = boto3.client('appsync')


def lambda_handler(event, context):
    """Secrets Manager Rotation Template
    This is a template for creating an AWS Secrets Manager rotation lambda
    Args:
        event (dict): Lambda dictionary of event parameters. These keys 
        must include the following:
            - SecretId: The secret ARN or identifier
            - ClientRequestToken: The ClientRequestToken of the secret version
            - Step: The rotation step (one of createSecret, setSecret, 
            testSecret, or finishSecret)
        context (LambdaContext): The Lambda runtime information
    Raises:
        ResourceNotFoundException: If the secret with the specified arn 
        and stage does not exist
        ValueError: If the secret is not properly configured for rotation
        KeyError: If the event parameters do not contain the expected keys
    """
    
    arn = event['SecretId']
    token = event['ClientRequestToken']
    step = event['Step']

    # Make sure the version is staged correctly
    metadata = service_client.describe_secret(SecretId=arn)
    if not metadata['RotationEnabled']:
        logger.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:
        logger.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]:
        logger.info("Secret version %s already set as AWSCURRENT for secret %s." % (token, arn))
        return
    elif "AWSPENDING" not in versions[token]:
        logger.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":
        create_secret(service_client, arn, token)

    elif step == "setSecret":
        set_secret(service_client, arn, token)

    elif step == "testSecret":
        test_secret(service_client, arn, token)

    elif step == "finishSecret":
        finish_secret(service_client, arn, token)

    else:
        raise ValueError("Invalid step parameter")


def create_secret(service_client, arn, token):
    """Create the secret
    This method first checks for the existence of a secret for the passed 
    in token. If one does not exist, it will generate a
    new secret and put it with the passed in token.
    Args:
        service_client (client): The secrets manager service client
        arn (string): The secret ARN or other identifier
        token (string): The ClientRequestToken associated with the secret version
    Raises:
        ResourceNotFoundException: If the secret with the specified arn and stage does not exist
    """
    # Make sure the current secret exists
    service_client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")

    # Now try to get the secret version, if that fails, put a new secret
    try:
        service_client.get_secret_value(SecretId=arn, VersionId=token, VersionStage="AWSPENDING")
        logger.info("createSecret: Successfully retrieved secret for %s." % arn)
    except service_client.exceptions.ResourceNotFoundException:
        # Get exclude characters from environment variable
        exclude_characters = os.environ['EXCLUDE_CHARACTERS'] if 'EXCLUDE_CHARACTERS' in os.environ else '/@"\'\\'
        # Generate a random password
        passwd = service_client.get_random_password(ExcludeCharacters=exclude_characters)

        # Put the secret
        service_client.put_secret_value(SecretId=arn, ClientRequestToken=token, SecretString=passwd['RandomPassword'], VersionStages=['AWSPENDING'])
        logger.info("createSecret: Successfully put secret for ARN %s and version %s." % (arn, token))


def set_secret(service_client, arn, token):
    """
    This method should set the AWSPENDING secret in the service that the 
    secret belongs to. For example, if the secret is a database
    credential, this method should take the value of the AWSPENDING 
    secret and set the user's password to this value in the database.
    Args:
        service_client (client): The secrets manager service client
        arn (string): The secret ARN or other identifier
        token (string): The ClientRequestToken associated with the secret 
        version
    """
    # This is where the secret should be set in the service
    
    try:
        secret_value = service_client.get_secret_value(
            SecretId=arn
        )
        apikey = json.loads(secret_value['SecretString'])["APIKEY"]
        
        # Generate new API key in AppSync service
        new_api_key = appsync_client.create_api_key(
            apiId=AppSyncApiId
        )['apiKey']['id']
        
        # Store the API key in Secrets Manager
        service_client.put_secret_value(
            SecretId=arn,
            SecretString=json.dumps({"APIKEY": new_api_key}),
            VersionStages=[
                "AWSCURRENT",
            ]
        )
        logger.info("setSecret: Successfully set secret for %s in Secrets Manager." % arn)
        
        # Delete the old API key in AppSync service, after successfully 
        # create the API key
        appsync_client.delete_api_key(
            apiId=AppSyncApiId,
            id=apikey
        )
        
    except Exception as e:
        print(e)
        raise NotImplementedError


def test_secret(service_client, arn, token):
    """Test the secret
    This method should validate that the AWSPENDING secret works in the 
    service that the secret belongs to. For example, if the secret
    is a database credential, this method should validate that the user can 
    login with the password in AWSPENDING and that the user has
    all of the expected permissions against the database.
    Args:
        service_client (client): The secrets manager service client
        arn (string): The secret ARN or other identifier
        token (string): The ClientRequestToken associated with the secret version
    """
    logger.info("Entered function: test_secret")
    # This is where the secret should be tested against the service
    raise NotImplementedError


def finish_secret(service_client, arn, token):
    """Finish the secret
    This method finalizes the rotation process by marking the secret 
    version passed in as the AWSCURRENT secret.
    Args:
        service_client (client): The secrets manager service client
        arn (string): The secret ARN or other identifier
        token (string): The ClientRequestToken associated with the 
        secret version
    Raises:
        ResourceNotFoundException: If the secret with the specified arn 
        does not exist
    """
    # First describe the secret to get the current version
    metadata = service_client.describe_secret(SecretId=arn)
    current_version = None
    for version in metadata["VersionIdsToStages"]:
        if "AWSCURRENT" in metadata["VersionIdsToStages"][version]:
            if version == token:
                # The correct version is already marked as current, return
                logger.info("finishSecret: Version %s already marked as AWSCURRENT for %s" % (version, arn))
                return
            current_version = version
            break

    # Finalize by staging the secret version current
    service_client.update_secret_version_stage(SecretId=arn, VersionStage="AWSCURRENT", MoveToVersionId=token, RemoveFromVersionId=current_version)
    logger.info("finishSecret: Successfully set AWSCURRENT stage to version %s for secret %s." % (token, arn))

Grant Permission

The following command grants Secrets Manager permission to call the Lambda function.
~]$ aws lambda add-permission --function-name [Your Lambda Function Name] --action lambda:InvokeFunction --statement-id secretsmanager --principal secretsmanager.amazonaws.com

{
    "Statement": "{\"Sid\":\"secretsmanager\",\"Effect\":\"Allow\",\"Principal\":{\"Service\":\"secretsmanager.amazonaws.com\"},\"Action\":\"lambda:InvokeFunction\",\"Resource\":\"arn:aws:lambda:us-west-2:123456789012:function:[LambdaFunctionName]\"}"
}

Configure Secret Rotation



If encount below error, make sure the permission is granted. Refer to the third step for guidance.



Test



Check AppSync, see if the API key has been actually rotated.


Check the Secrets Manager, see if the secret key has been updated there.


We noticed that both the AppSync API key and the secret in Secrets Manager have been successfully updated.


Wrap Up
To wrap up, we built a powerful solution to rotate the API key of the AppSync API. We accomplished it from several perspects:

• Security: The security is reinforced by using a VPC endpoint to avoid sensitive traffic flow out of the VPC environment. We followed the least privilege principle and defined the IAM policy to only necessary API permissions. The logs that Lambda write to the CloudWatch Logs do not include any sensitive data, i.e. API keys and etc.. The security group around the VPC endpoint for Secrets Manager service only allows traffic from Lambda subnets to its HTTPS 443 port.
• Simplicity: We used managed services to achieve the goal, including Secrets Manager and Lambda function. We developed the workflow using Python code and built the solution very fast.

References

GitHub - aws-samples / aws-secrets-manager-rotation-lambdas

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/appsync.html

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/secretsmanager.html

AppSync Developer Guide - Security

Rotating AWS Secrets Manager secrets for other databases or services

AWS service endpoints

Using resource-based policies for AWS Lambda

Overview of the Lambda rotation function