Building a serverless connected BBQ as SaaS - Part 4 - AuthZ

AWSTemplateFormatVersion: "2010-09-09"
Transform: "AWS::Serverless-2016-10-31"
Description: Connected BBQ Application Tenant Service
Parameters:
ApplicationName:
Type: String
Description: Name of owning application
Default: bbq-iot
UserPoolStackName:
Type: String
Description: The name of the Stack with the Cognito User Pool

Globals:
Function:
Timeout: 30
MemorySize: 2048
Runtime: python3.12

Resources:
LambdaPDPFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: Lambda/AuthZ
Handler: authz.handler
Environment:
Variables:
JWKS_URL:
Fn::ImportValue: !Sub ${UserPoolStackName}:jwks-url
AUDIENCE:
Fn::ImportValue: !Sub ${UserPoolStackName}:app-audience
Events:
AuthZ:
Type: Api
Properties:
Path: /authz
Method: post
RestApiId: !Ref PDPApi
Auth:
AuthorizationType: AWS_IAM

PDPApi:
Type: AWS::Serverless::Api
Properties:
Name: !Sub ${ApplicationName}-pdp-api
StageName: prod
EndpointConfiguration: REGIONAL
Cors:
AllowMethods: "'GET,PUT,POST,DELETE,OPTIONS'"
AllowHeaders: "'Content-Type,Authorization,X-Amz-Date,X-Api-Key,X-Amz-Security-Token'"
AllowOrigin: "'*'"
Auth:
DefaultAuthorizer: AWS_IAM

Outputs:
PDPApiID:
Value: !Ref PDPApi
Description: The ID of the PDP API
Export:
Name: !Sub ${AWS::StackName}:pdp-api-id

import os
import json
import jwt
from jwt import PyJWKClient

def handler(event, context):
data = json.loads(event["body"])
jwt_token = data["jwt_token"]

try:
jwks_url = os.environ["JWKS_URL"]

jwks_client = PyJWKClient(jwks_url)
signing_key = jwks_client.get_signing_key_from_jwt(jwt_token)

decoded_token = jwt.decode(
jwt_token,
signing_key.key,
algorithms=["RS256"],
audience=os.environ["AUDIENCE"],
)

# Check resource and make an AuthZ decision
if data["resource"] == "tenant":
tenant_id = data["tenant_id"]
token_tenant_id = decoded_token.get("tenant")

if token_tenant_id == tenant_id:
response_body = generate_access(
"Allow", data["action"], data["resource"]
)
return {
"statusCode": 200,
"body": json.dumps(response_body),
"headers": {"Content-Type": "application/json"},
}
elif data["resource"] == "tenantUser":
user_id = data["user_id"]
token_user_id = decoded_token.get("cognito:username")

if token_user_id == user_id:
response_body = generate_access(
"Allow", data["action"], data["resource"]
)
return {
"statusCode": 200,
"body": json.dumps(response_body),
"headers": {"Content-Type": "application/json"},
}

except Exception as e:
print(f"Authorization error: {str(e)}")

# Generate a default response that deny access
response_body = generate_access("Deny", data["action"], data["resource"])

return {
"statusCode": 403,
"body": json.dumps(response_body),
"headers": {"Content-Type": "application/json"},
}

def generate_access(effect, action, resource):
auth_response = {
"effect": effect,
"action": action,
"resource": resource,
}
return auth_response

api_endpoint = os.environ.get("PDP_AUTHZ_API_ENDPOINT")

def handler(event, context):
token = event["headers"].get("authorization", "")

if not token:
raise Exception("Unauthorized")

token = token.replace("Bearer ", "")

try:

......

path_tenant_id = event["pathParameters"]["tenantId"]

session = boto3.Session()
credentials = session.get_credentials().get_frozen_credentials()

region = os.environ["AWS_REGION"]
auth = AWS4Auth(
credentials.access_key,
credentials.secret_key,
region,
"execute-api",
session_token=credentials.token,
)

data = {
"jwt_token": token,
"resource": "tenant",
"action": "read",
"resource_path": event["path"],
"tenant_id": event["pathParameters"]["tenantId"],
}
headers = {"Content-type": "application/json"}
response = requests.post(
api_endpoint + "/authz", data=json.dumps(data), headers=headers, auth=auth
)

......

except Exception as e:
print(f"Authorization error: {str(e)}")