def lambda_handler()

in serverless-rest-api/python-http-cdk/src/api/authorizer.py [0:0]

• 30 lines of code
• 5 McCabe index (conditional complexity)

def lambda_handler(event, context):
    global admin_group_name
    print(event)
    # print("Client token: " + event['authorizationToken'])
    # print("Method ARN: " + event['methodArn'])
    tmp = event['methodArn'].split(':')
    api_gateway_arn_tmp = tmp[5].split('/')
    region = tmp[3]
    aws_account_id = tmp[4]
    # validate the incoming token
    validated_decoded_token = validate_token(event['authorizationToken'], region)
    if not validated_decoded_token:
        raise Exception('Unauthorized')
    principal_id = validated_decoded_token['sub']
    # initialize the policy
    policy = AuthPolicy(principal_id, aws_account_id)
    policy.restApiId = api_gateway_arn_tmp[0]
    policy.region = region
    policy.stage = api_gateway_arn_tmp[1]
    # allow all public resources/methods explicitly
    policy.allow_method(HttpVerb.GET, "locations")
    policy.allow_method(HttpVerb.GET, "locations/*")
    policy.allow_method(HttpVerb.GET, "locations/*/resources")
    policy.allow_method(HttpVerb.GET, "locations/*/resources/*/bookings")
    # add user specific resources/methods
    policy.allow_method(HttpVerb.GET, f"/users/{principal_id}/bookings")
    policy.allow_method(HttpVerb.GET, f"/users/{principal_id}/bookings/*")
    policy.allow_method(HttpVerb.PUT, f"/users/{principal_id}/bookings")
    policy.allow_method(HttpVerb.DELETE, f"/users/{principal_id}/bookings/*")
    # Check the Cognito group entry for Admin.
    # Assuming here that the Admin group has always higher /precedence
    if 'cognito:groups' in validated_decoded_token and validated_decoded_token['cognito:groups'][0] == admin_group_name:
        # add administrative privileges
        policy.allow_method(HttpVerb.DELETE, "locations")
        policy.allow_method(HttpVerb.DELETE, "locations/*")
        policy.allow_method(HttpVerb.PUT, "locations")
        policy.allow_method(HttpVerb.PUT, "locations/*")
    # Finally, build the policy
    auth_response = policy.build()
    return auth_response