Creating Python Lambda@Edge to access a secret in Secrets Manager

Question

I wrote a python lambda that does basic HTTP auth using a user name and password. I created a simple static website in S3 and the python auth lambda is to be invoked by CloudFront viewer request to get a user name password from Secrets Manager to do authorization against. I deployed my solution to CloudFront and I'm getting a 503 error. From my research, it looks like I'm running into the multi-thread limits in Python. I tried using async but I'm running into issues with the output returning as a coroutine. I've seen similar examples to this done in NodeJS, I figured I would give it a try in Python. Is what I'm doing even possible or worth doing in Python or should try to do the same solution in NodeJS which I'm not as familiar with? See code below. Thanks.

import requests
import boto3
import base64
from base64 import b64encode
from botocore.exceptions import ClientError
import json
import asyncio


def secrets_manager_client():
    return boto3.client('secretsmanager')


async def get_secret(secret_name):
    try:
       get_secret_value_response = await secrets_manager_client().get_secret_value(SecretId=secret_name)
    except ClientError as e:
        if await e.response['Error']['Code'] == 'DecryptionFailureException':
            raise e from e
        elif await e.response['Error']['Code'] == 'InternalServiceErrorException':
            raise e from e
        elif await e.response['Error']['Code'] == 'InvalidParameterException':
            raise e from e
        elif await e.response['Error']['Code'] == 'InvalidRequestException':
            raise e from e
        elif await e.response['Error']['Code'] == 'ResourceNotFoundException':
            raise e from e
    else:
        return get_secret_value_response['SecretString']
    await processDataInWorker(v)


def lambda_handler(event, context):

   request = event['Records'][0]['cf']['request']
   headers = request['headers']

   secret = json.loads(get_secret(secret_name="mysecret/location"))

   base64credentials = base64.b64encode((secret['username'] + ":" + secret['password']).encode()).decode()
   authstring = 'Basic ' + base64credentials

   if headers['authorization'] == 'undefined' or headers['authorization'][0] != authstring:
        return {
           "body": "Unauthorized",
           "headers": {
            'www-authenticate': [{"key": "WWW-Authenticate", "value": "Basic"}]
           },
           "status": "401",
           "statusDescription": "Unauthorized"
         }
   else:
       return request

Answer 1

Have not seen implementations of this in Pyhon but there is a plethora in javascript e.g. this gist.

You would just need to add the secretmanager retrieval with an async function + await as you mentioned.