Trong quá trình xây dựng các hệ thống backend hiện đại, AWS API Gateway đóng vai trò như một cổng vào bảo mật, mở rộng dễ dàng, và mạnh mẽ để kết nối frontend với backend. Tuy nhiên, việc cấu hình thủ công API Gateway là một quá trình tốn thời gian và dễ sai sót.

Trong bài viết này, mình sẽ chia sẻ cách tự động triển khai toàn bộ API Gateway bằng Python (boto3), tích hợp với ALB (Application Load Balancer) ở phía sau và đặc biệt là Cognito User Pool Authorizer để quản lý xác thực. Đồng thời, bài viết cũng giải thích vì sao khi dùng Cognito bạn không thể để CORS header là *, mà bắt buộc phải chỉ định Origin cụ thể.

📦 Kiến trúc tổng quan

Chúng ta sẽ triển khai API Gateway với các đặc điểm sau:

• Tự động tạo Resource, Method, Integration, Responses dựa trên file JSON mô tả API.

• Hỗ trợ route có path parameter như /api/v1/users/{user_id}/profile-picture.

• Tích hợp Cognito User Pool cho các API yêu cầu xác thực.

• Forward request về phía backend thông qua ALB.

🧠 Tại sao nên dùng Cognito với API Gateway?

Cognito User Pool Authorizer là cơ chế mạnh mẽ để xác thực người dùng sử dụng JWT token (thường lấy được sau khi đăng nhập). Khi tích hợp:

• Bạn không cần triển khai xác thực thủ công trong backend.

• API Gateway sẽ verify JWT và tự động từ chối request không hợp lệ.

Tuy nhiên, có một điểm quan trọng:

⚠️ Khi dùng Cognito và bật CORS, bạn không được dùng Access-Control-Allow-Origin: *!

Vì sao?

JWT token là dữ liệu nhạy cảm. Nếu cho phép * trong CORS, mọi domain đều có thể truy cập token của người dùng → gây rủi ro bảo mật.

Vì vậy, bạn phải:

• Dùng $input.params("Origin") để lấy giá trị header Origin gửi từ trình duyệt.

• So sánh nó với danh sách domain cho phép ($stageVariables.allowedOrigins).

• Chỉ set Access-Control-Allow-Origin nếu domain nằm trong danh sách hợp lệ.

📁 Cấu trúc thư mục và file JSON API

deployment/
└── api-gateway/
    └── authen/
        └── api_list.json

Ví dụ nội dung api_list.json:

[
    {
        "path": "/api/v1/authen/email-authen",
        "method": "POST",
        "authorization": false
    },
    {
        "path": "/api/v1/authen/logout",
        "method": "POST",
        "authorization": true
    },
    {
        "path": "/api/v1/admin/users/{user_id}/profile-picture",
        "method": "GET",
        "authorization": true
    }
]

⚙️ Script Python tự động deploy

Chúng ta sử dụng thư viện boto3 để tương tác với AWS. Script sẽ thực hiện:

1. Tạo Cognito Authorizer nếu chưa có.

2. Duyệt qua tất cả các file api_list.json.

3. Tạo resource + method + integration + response cho từng API.

4. Xử lý chính xác cả path parameter như {user_id}.

5. Tích hợp ALB làm backend.

6. Tự động deploy stage.

✨ Những điểm đáng chú ý

1. Xử lý path parameter

Script sẽ dò các phần tử như {user_id} trong path và cấu hình:

integration_options["requestParameters"][
    f"integration.request.querystring.{param_name}"
] = f"method.request.path.{param_name}"

Điều này giúp API Gateway lấy giá trị từ URL và forward chính xác tới backend ALB.

2. Tùy biến CORS cho Cognito

Trong phần Integration Response, bạn cần thêm logic sau:

#set($origin = $input.params("Origin"))
#set($allowedOrigins = $stageVariables.allowedOrigins.split(","))

#if($origin == "null")
#set($context.responseOverride.header.Access-Control-Allow-Origin = "*")
#elseif($allowedOrigins.contains($origin))
#set($context.responseOverride.header.Access-Control-Allow-Origin = $origin)
#end​

Và nhớ khai báo biến allowedOrigins trong Stage Variables, ví dụ:

allowedOrigins: "https://myfrontend.com,https://admin.myfrontend.com"​

3. Phân quyền theo nhu cầu

Tùy từng API, bạn có thể bật/tắt Cognito Authorization chỉ bằng một flag "authorization": true.

✅ Kết luận

Với cách làm này:

• Bạn có thể quản lý API dễ dàng dưới dạng file JSON.

• Triển khai toàn bộ hệ thống chỉ bằng một lệnh.

• Áp dụng bảo mật theo chuẩn AWS với Cognito.

• Tránh sai sót và tiết kiệm thời gian khi scale API.

Nếu bạn đang dùng API Gateway + Cognito + ALB trong dự án của mình, hãy thử áp dụng ngay cách này để tối ưu hóa pipeline CI/CD nhé!

Full code:

import boto3
import json
import os

API_ID = os.getenv("API_ID")
REGION = os.getenv("AWS_REGION", "ap-northeast-1")
ALB_DNS = os.getenv("ALB_DNS")

session = boto3.Session()
client = session.client("apigateway", region_name=REGION)

# Create Cognito Authorizer
cognito_client = session.client("cognito-idp", region_name=REGION)
auth_client = session.client("apigateway", region_name=REGION)

USER_POOL_ID = os.getenv("COGNITO_USER_POOL_ID")

# Get Cognito Issuer URL
cognito_issuer = f"https://cognito-idp.{REGION}.amazonaws.com/{USER_POOL_ID}"

# Check if the authorizer already exists
existing_authorizers = auth_client.get_authorizers(restApiId=API_ID)["items"]

authorizer_name = "CognitoAuthorizer"
authorizer_id = None

for auth in existing_authorizers:
    if auth["name"] == authorizer_name:
        authorizer_id = auth["id"]
        print(f"✅ Authorizer already exists: {authorizer_name} (ID: {authorizer_id})")
        break

# Create authorizer only if it doesn't exist
if not authorizer_id:
    authorizer_response = auth_client.create_authorizer(
        restApiId=API_ID,
        name=authorizer_name,
        type="COGNITO_USER_POOLS",
        providerARNs=[f"arn:aws:cognito-idp:{REGION}:{os.getenv('AWS_ACCOUNT_ID')}:userpool/{USER_POOL_ID}"],
        identitySource="method.request.header.Authorization"
    )
    authorizer_id = authorizer_response["id"]
    print(f"✅ Cognito Authorizer created: {authorizer_name} (ID: {authorizer_id})")

AUTHORIZE_ID = authorizer_id
print(f"✅ Cognito Authorizer created: {authorizer_name} (ID: {AUTHORIZE_ID})")

# Retrieve the list of existing resources
resources = client.get_resources(restApiId=API_ID)
existing_resources = {item["path"]: item["id"] for item in resources["items"]}

# Find the root resource ID
root_resource_id = None
for resource in resources["items"]:
    if resource["path"] == "/":
        root_resource_id = resource["id"]
        break

if not root_resource_id:
    raise ValueError("Root Resource not found")

# Read API list from all subfolders in `liveapp_api/deployment/api-getway`
script_dir = os.path.dirname(os.path.abspath(__file__))
api_list = []

for subdir in os.listdir(script_dir):
    subdir_path = os.path.join(script_dir, subdir)
    if os.path.isdir(subdir_path):  # Get only subfolders
        json_path = os.path.join(subdir_path, "api_list.json")
        if os.path.exists(json_path):
            with open(json_path) as f:
                try:
                    api_list.extend(json.load(f))  # Merge API list
                except json.JSONDecodeError:
                    print(f"⚠️ Error reading {json_path}, please check valid JSON.")


def create_resource(client, rest_api_id, parent_id, path_part):
    """Create a resource if it does not exist"""
    try:
        response = client.create_resource(
            restApiId=rest_api_id, parentId=parent_id, pathPart=path_part
        )
        return response["id"]
    except client.exceptions.ConflictException:
        # If the resource already exists, get the resource list to find the ID.
        resources = client.get_resources(restApiId=rest_api_id)
        for resource in resources["items"]:
            if resource.get("pathPart") == path_part and resource.get("parentId") == parent_id:
                return resource["id"]
        raise ValueError(f"Resource {path_part} not found")

resource_cache = {}
# Process each API in the merged list
for api in api_list:
    path = api["path"].lstrip("/")  # Remove leading `/`
    method = api["method"]
    requires_auth = api.get("authorization", False)

    print(f"Processing API: {path} [{method}] - Auth: {requires_auth}")

    # Create resource hierarchy
    parent_id = root_resource_id
    parts = path.split("/")  # Split path into segments
    for part in parts:
        full_path = f"{parent_id}/{part}"  # Unique key to cache
        if full_path in resource_cache:
            parent_id = resource_cache[full_path]
        else:
            parent_id = create_resource(client, API_ID, parent_id, part)
            resource_cache[full_path] = parent_id

    resource_id = parent_id  # ID of the last resource in the path

    auth_type = "COGNITO_USER_POOLS" if requires_auth else "NONE"
    auth_params = {"authorizationType": auth_type}

    if requires_auth:
        auth_params["authorizerId"] = AUTHORIZE_ID

    # Create HTTP method
    try:
        client.get_method(
            restApiId=API_ID,
            resourceId=resource_id,
            httpMethod=method,
        )
        print(f"⚠️ Method {method} already exists for {path}, skipping...")
    except client.exceptions.NotFoundException:
        client.put_method(
            restApiId=API_ID,
            resourceId=resource_id,
            httpMethod=method,
            **auth_params,
        )
        print(f"✅ {('Cognito Authorization' if requires_auth else 'No Auth')} added for {path} [{method}]")
    except Exception as e:
        print(f"⚠️ Error adding method {method} for {path}: {e}")

    # Add Method Response
    try:
        client.put_method_response(
            restApiId=API_ID,
            resourceId=resource_id,
            httpMethod=method,
            statusCode="200",
            responseModels={"application/json": "Empty"},
            responseParameters={
                "method.response.header.Content-Type": True,
                "method.response.header.Cross-Origin-Opener-Policy": True,
                "method.response.header.Referrer-Policy": True,
                "method.response.header.X-Content-Type-Options": True,
                "method.response.header.X-Frame-Options": True,
                "method.response.header.Access-Control-Allow-Origin": True,
                "method.response.header.Access-Control-Allow-Credentials": True,
                "method.response.header.Access-Control-Allow-Methods": True,
                "method.response.header.Access-Control-Allow-Headers": True,
            },
        )
    except Exception as e:
        print(f"⚠️ Method Response already exists or encountered an error: {e}")

    # Create Integration with ALB
    try:
        client.get_integration(
            restApiId=API_ID,
            resourceId=resource_id,
            httpMethod=method,
        )
        print(f"⚠️ Integration already exists for {path} [{method}], skipping...")
    except client.exceptions.NotFoundException:
        integration_options = {
            "restApiId": API_ID,
            "resourceId": resource_id,
            "httpMethod": method,
            "type": "HTTP",
            "integrationHttpMethod": method,
            "uri": f"{ALB_DNS}/{path}",
            "requestParameters": {}
        }

        if "{" in path and "}" in path:
            parts = path.split("/")
            for part in parts:
                if part.startswith("{") and part.endswith("}"):
                    param_name = part[1:-1]
                    integration_options["requestParameters"][f"integration.request.querystring.{param_name}"] = f"method.request.path.{param_name}"

        client.put_integration(**integration_options)
        print(f"✅ Integration {path} [{method}] created successfully!")
    except Exception as e:
        print(f"⚠️ Error creating integration {path} [{method}]: {e}")


    response_params = {
        "method.response.header.Content-Type": "'application/json'",
        "method.response.header.Cross-Origin-Opener-Policy": "'same-origin'",
        "method.response.header.Referrer-Policy": "'strict-origin-when-cross-origin'",
        "method.response.header.X-Content-Type-Options": "'nosniff'",
        "method.response.header.X-Frame-Options": "'same-origin'",
        "method.response.header.Access-Control-Allow-Methods": "'OPTIONS, GET, POST, PUT, DELETE'",
        "method.response.header.Access-Control-Allow-Headers": "'*'",
    }

    response_templates = None

    if requires_auth:
        response_params["method.response.header.Access-Control-Allow-Credentials"] = "'true'"

        response_templates = {
            "application/json": """#set($origin = $input.params("Origin"))
            #set($allowedOrigins = $stageVariables.allowedOrigins.split(","))

            #if($origin == "null")
            #set($context.responseOverride.header.Access-Control-Allow-Origin = "*")
            #elseif($allowedOrigins.contains($origin))
            #set($context.responseOverride.header.Access-Control-Allow-Origin = $origin)
            #end"""
        }

    else:
        response_params["method.response.header.Access-Control-Allow-Origin"] = "'*'"

    integration_response_payload = {
        "restApiId": API_ID,
        "resourceId": resource_id,
        "httpMethod": method,
        "statusCode": "200",
        "responseParameters": response_params,
    }

    if response_templates:
        integration_response_payload["responseTemplates"] = response_templates

    # Add Integration Response
    try:
        client.put_integration_response(**integration_response_payload)
        print(f"✅ Integration Response {path} [{method}] updated successfully!")
    except Exception as e:
        print(f"⚠️ Error creating Integration Response {path} [{method}]: {e}")

# Deploy API
client.create_deployment(restApiId=API_ID, stageName=os.getenv("API_GETWAY_STAGE"))
print("🚀 API Gateway successfully updated!")