basic_auth.py

import base64
import boto3


def basicAuth(event, context):
    # Return a policy which allows this user to access to this api
    # this call is cached for all authenticated calls, so we need to give
    # access to the whole api. This could be done by having a policyDocument
    # for each available function, but I don't really care :)
    arn = "%s/*" % "/".join(event["methodArn"].split("/")[0:2])

    # if a basic auth header is set, use that to find the correct user/token
    authorizationHeader = None
    if 'Authorization' in event['headers']:
        authorizationHeader = event["headers"]["Authorization"]
    if 'authorization' in event['headers']:
        authorizationHeader = event["headers"]["authorization"]

    if authorizationHeader:
        b64_token = authorizationHeader.split(" ")[-1]

        # decode the base64 encoded header value
        username, token = base64.b64decode(b64_token).decode("utf-8").split(":")
        # search for the given api key
        client = boto3.client("apigateway")
        response = client.get_api_keys(nameQuery=username, includeValues=True)

        # if no keys found, deny access
        if len(response["items"]) != 1:
            print("Couldn't find key")
            raise Exception("Unauthorized")

        # if the key value does not match, deny access
        if response["items"][0]["value"] != token:
            print("Key value mismatch")
            raise Exception("Unauthorized")

    # check if an x-api-token header is set, if so, take it as-is, api gateway
    # will check the validity
    elif "x-api-key" in event["headers"]:
        print("x-api-key received")
        username = "token"
        token = event["headers"]["x-api-key"]

    # no authentication headers found, deny
    else:
        print("No authentication header found")
        raise Exception("Unauthorized")

    authResponse = {
        "principalId": username,
        "usageIdentifierKey": token,
        "policyDocument": {
            "Version": "2012-10-17",
            "Statement": [
                {"Action": "execute-api:Invoke", "Effect": "Allow", "Resource": arn}
            ],
        },
    }
    print("Authentication response: %s" % authResponse)

    return authResponse