import os
import hmac
import hashlib
import base64
import twython
import crcmod.predefined
CONSUMER_SECRET = os.environ['TWITTER_CONSUMER_SECRET']
def lambda_handler(event, context):
if event['httpMethod'] == 'POST':
body = str(event['body'])
crc_token = None
for key, value in event['headers'].items():
if key.lower() == 'x-twitter-webhooks-signature':
crc_token = value
break
if crc_token is not None:
crc_token = crc_token.replace('sha256=', '')
crc_digest = hmac.new(
key=CONSUMER_SECRET.encode(),
msg=body.encode(),
digestmod=hashlib.sha256
).digest()
crc_digest_b64 = base64.b64encode(crc_digest).decode()
if crc_token == crc_digest_b64:
return {
'statusCode': 200,
'body': 'CRC token validation successfully completed.'
}
return {
'statusCode': 400,
'body': 'Invalid request.'
}
以下是一个示例API Gateway映射模板:
{
"body": $input.json('$'),
"headers": {
#foreach($header in $input.params().header.keySet())
"$header": "$input.params().header.get($header)"
#if($foreach.hasNext),#end
#end
},
"httpMethod": "$context.httpMethod",
"path": "$context.resourcePath",