Contents:
An Authorizer in API Gateway is a mechanism that allows you to control authentication and request authorization (based on token or policy) before the request is forwarded to the backend.
def generate_policy(principal_id, effect, resources, context=None):
if not isinstance(resources, list):
resources = [resources]
policy = {
"principalId": principal_id,
"policyDocument": {
"Version": "2012-10-17",
"Statement": [{
"Action": "execute-api:Invoke",
"Effect": effect,
"Resource": resources
}]
}
}
if context:
policy["context"] = context
return policy
user_type_config = {
"normalUser": {
"resource": "Users"
},
"premiumUser": {
"resource": "PremiumUsers"
},
"devs": {
"resource": "Devs"
}
}
def build_allowed_resource(method_arn, resource_pattern):
arn_parts = method_arn.split(":")
region = arn_parts[3]
account_id = arn_parts[4]
api_gateway_arn_parts = arn_parts[5].split("/")
api_id = api_gateway_arn_parts[0]
stage = api_gateway_arn_parts[1]
http_method = api_gateway_arn_parts[2]
resource_base = f"arn:aws:execute-api:{region}:{account_id}:{api_id}/{stage}/{http_method}"
clean_resource_pattern = resource_pattern.rstrip("/*")
return [f"{resource_base}/{clean_resource_pattern}"]
def lambda_handler(event, context):
token = event.get("authorizationToken")
method_arn = event["methodArn"]
if token not in user_type_config:
return generate_policy(
principal_id="unknown",
effect="Deny",
resources=method_arn,
context={
"token": token or "None",
"methodArn": method_arn
}
)
config = user_type_config[token]
allowed_resources = build_allowed_resource(method_arn, config["resource"])
return generate_policy(
principal_id=token,
effect="Allow",
resources=allowed_resources,
context={
"token": token,
"methodArn": method_arn
}
)
"unknown"
)."Allow"
or "Deny"
.def generate_policy(principal_id, effect, resources, context=None):
if not isinstance(resources, list):
resources = [resources]
policy = {
"principalId": principal_id,
"policyDocument": {
"Version": "2012-10-17",
"Statement": [{
"Action": "execute-api:Invoke",
"Effect": effect,
"Resource": resources
}]
}
}
if context:
policy["context"] = context
return policy
"normalUser"
→ can only call /Users
.user_type_config = {
"normalUser": {
"resource": "Users"
},
"premiumUser": {
"resource": "PremiumUsers"
},
"devs": {
"resource": "Devs"
}
}
def build_allowed_resource(method_arn, resource_pattern):
arn_parts = method_arn.split(":")
region = arn_parts[3]
account_id = arn_parts[4]
api_gateway_arn_parts = arn_parts[5].split("/")
api_id = api_gateway_arn_parts[0]
stage = api_gateway_arn_parts[1]
http_method = api_gateway_arn_parts[2]
resource_base = f"arn:aws:execute-api:{region}:{account_id}:{api_id}/{stage}/{http_method}"
clean_resource_pattern = resource_pattern.rstrip("/*")
return [f"{resource_base}/{clean_resource_pattern}"]
user_type_config
→ return Deny policy.user_type_config
→ build allowed resource ARN → return Allow policy, granting access only to the correct resource.def lambda_handler(event, context):
token = event.get("authorizationToken")
method_arn = event["methodArn"]
if token not in user_type_config:
return generate_policy(
principal_id="unknown",
effect="Deny",
resources=method_arn,
context={
"token": token or "None",
"methodArn": method_arn
}
)
config = user_type_config[token]
allowed_resources = build_allowed_resource(method_arn, config["resource"])
return generate_policy(
principal_id=token,
effect="Allow",
resources=allowed_resources,
context={
"token": token,
"methodArn": method_arn
}
)
Token source should be set to Authorization
, since it is a default header in CORS.