Skip to content

Instantly share code, notes, and snippets.

@piyushjaware
Last active May 4, 2020 15:49
Show Gist options
  • Save piyushjaware/be3fd9ffc42a353e4ed0a3025bd14079 to your computer and use it in GitHub Desktop.
Save piyushjaware/be3fd9ffc42a353e4ed0a3025bd14079 to your computer and use it in GitHub Desktop.
CFN Lambda backed Custom Resource to inject dynamic values
AWSTemplateFormatVersion: "2010-09-09"
Description: Custom Ressource Tester Template
Parameters:
ApiName:
Type: String
Description: Enter the appsync api name
Resources:
ApiHelperResource:
Type: "Custom::ApiHelper"
Properties:
Description: |
For a given appsync api, this resource returns `ApiId`, `ApiUrl` and `ApiKey` to be used anywhere in the template.
This also validates that the apiKey is active for at least the next 30 days; fails deplyment otherwise.
Input params: api_name
ServiceToken: !GetAtt
- ApiHelperFunction
- Arn
api_name: !Ref ApiName
ApiHelperFunction:
Type: "AWS::Lambda::Function"
Properties:
FunctionName: ApiHelperFunction
Code:
ZipFile: |
from __future__ import print_function
import json
import boto3
import cfnresponse
import datetime
SUCCESS = "SUCCESS"
FAILED = "FAILED"
print('Loading function')
appsync = boto3.client('appsync')
def lambda_handler(event, context):
print("Received event: " + json.dumps(event, indent=2))
responseData={}
try:
if event['RequestType'] == 'Create' or event['RequestType'] == 'Update':
print("Request Type:",event['RequestType'])
data = get_appsync_api_details(event['ResourceProperties']['api_name'])
responseData=data
elif event['RequestType'] == 'Delete':
print("Delete Request. No change needed.")
responseStatus = 'SUCCESS'
except Exception as e:
print('Failed to process:', e)
responseStatus = FAILED
responseData = {'Failure': f'Something went wrong. {str(e)}'}
cfnresponse.send(event, context, responseStatus, responseData)
def get_appsync_api_details(api_name):
api_list = appsync.list_graphql_apis()['graphqlApis']
filtered_apis = list(filter(lambda api: api['name'] == api_name, api_list))
if not filtered_apis:
raise Exception(f'{api_name} not found.')
api = filtered_apis[0]
api_keys = appsync.list_api_keys(apiId=api['apiId'])['apiKeys']
print(api_keys)
api_key= api_keys[0]
validate_apikey(api_key['expires'])
return {
'ApiId': api['apiId'],
'ApiUrl': api['uris']['GRAPHQL'],
'ApiKey': api_key['id']
}
def validate_apikey(expires_utc):
expires = datetime.datetime.fromtimestamp(expires_utc)
current_time = datetime.datetime.utcnow()
diff = (expires - current_time).days
if diff < 30:
raise Exception(f'Api key is going to expire in 30 days. Please renew it.')
Handler: index.lambda_handler
Runtime: python3.6
Timeout: 50
Role: !GetAtt
- LambdaExecutionRole
- Arn
LambdaExecutionRole:
Type: "AWS::IAM::Role"
Properties:
AssumeRolePolicyDocument:
Version: 2012-10-17
Statement:
- Effect: Allow
Principal:
Service:
- lambda.amazonaws.com
Action:
- "sts:AssumeRole"
Policies:
- PolicyName: lambdaPolicy
PolicyDocument:
Version: 2012-10-17
Statement:
- Effect: Allow
Action:
- "logs:CreateLogGroup"
- "logs:CreateLogStream"
- "logs:PutLogEvents"
Resource: "arn:aws:logs:*:*:*"
- PolicyName: LambdaAppsyncPolicy
PolicyDocument:
Version: "2012-10-17"
Statement:
- Effect: Allow
Action:
- appsync:*
Resource: "*"
Outputs:
ApiId:
Description: ApiId value
Value: !GetAtt ApiHelperResource.ApiId # Read output of the custom resource
ApiUrl:
Description: ApiUrl value
Value: !GetAtt ApiHelperResource.ApiUrl
ApiKey:
Description: ApiKey value
Value: !GetAtt ApiHelperResource.ApiKey
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment