Secure AI Agent Gateway on AWS

Introduction

30 Second Summary

Automation becomes dangerous when it acts with more authority than the person who triggered it. Every request needs trustworthy answers to two questions: who sent it, and what are they allowed to do?

In this project, you will build a secure AI agent execution gateway on AWS with AWS CDK. The result includes reproducible security evidence from local tests through a live deployment.

What You'll Build

Your finished demo shows a signed request creating and reading a tenant record while every forged or overreaching request is denied.

By the end of this project, you'll have:

• Signed cloud operations that use SigV4 to create or read tenant-blue records through an API Gateway endpoint. Unsigned calls stop before Lambda runs.
• A server-owned identity binding that gives the verified role only records:read plus records:create. Every unbound principal or unknown action hits default deny.
• A reproducible evidence report built from passing pytest checks, reviewed CloudFormation changes, exact AWS Identity and Access Management actions, and a fixed 100-request benchmark. It shows zero unauthorized tool executions plus zero tenant-isolation violations.
• Secret Mission: Exploit a confused deputy inside a local pytest fixture. Then prove the deployable GatewayService blocks the same forged payload while preserving legitimate access.

Are there any prerequisites?

You need an AWS account with IAM Identity Center access plus a secure temporary-credential workflow.

Your machine also needs Python 3.12, AWS CLI v2, Node.js, and Visual Studio Code.

Before We Start

This checkpoint locks in the intended 12-digit AWS account ID and Region before any cloud work begins. The security boundary trusts API Gateway's verified requestContext.identity.userArn. It applies a server-owned binding from the generated role to tenant-blue with records:read and records:create. Model output and request-body role, principal_id, and tenant_id claims remain untrusted.

Preflight the AWS Security Boundary

A cloud deployment can target the wrong account or rely on credentials that should never be used for project work. This preflight stops the deployment boundary before any write reaches AWS.

You will pin the local toolchain before using read-only checks to verify the selected account, Region, temporary credentials, bootstrap state, and legacy role state. Any missing inspection or deployment permission keeps AWS writes paused for administrator review.

In this step, get ready to:

• Create the local workspace with Python 3.12 and pinned project dependencies.
• Authenticate the AWS CLI through the agent-gateway SSO profile.
• Verify the account boundary without creating or modifying AWS resources.

Create the pinned workspace

A virtual environment keeps this project's Python packages isolated from packages installed elsewhere on your Mac. The pinned files make local tests use the same dependency versions every time.

• Press Cmd+Space to open macOS search.
• Type Terminal and press Enter.
• Check the active Python version by running this command:

python --version

What does this check do?

The command reports which Python interpreter your shell currently resolves. This project uses Python 3.12 so its environment matches the Lambda runtime configured later.

✔️ I see the required version

Your shell reports Python 3.12. That interpreter is ready to create the project environment.

ⓧ I see an older version

Your current interpreter cannot create the pinned Python 3.12 environment.

• Download Python 3.12 from the official Python downloads page.
• Complete the macOS installer.
• Return to Terminal after the installation completes.
• Repeat the version check above.

ⓧ Command not found

Your shell cannot find a Python interpreter yet.

• Download Python 3.12 from the official Python downloads page.
• Complete the macOS installer.
• Return to Terminal after the installation completes.
• Repeat the version check above.

• Move to your Desktop by running this command:

cd ~/Desktop

What does this command do?

The command changes Terminal's current location to your macOS Desktop. The new workspace will be easy to find from Finder.

• Create the secure-agent-gateway workspace in Visual Studio Code with a Python 3.12 environment by running these commands:

mkdir secure-agent-gateway
cd secure-agent-gateway
code .
python3.12 -m venv .venv

What do these commands create?

• The first two commands create the secure-agent-gateway folder on your Desktop and move Terminal into it.
• The code command opens that folder in Visual Studio Code.
• The final command creates an isolated environment in .venv with Python 3.12.

Visual Studio Code opens with secure-agent-gateway at the top of the Explorer sidebar. The .venv folder appears inside the workspace.

Workspace or environment missing?

• Confirm Terminal is inside ~/Desktop/secure-agent-gateway before creating the environment.
• Confirm the Python 3.12 installer completed if the python3.12 executable is unavailable.
• Use help me diagnose why the workspace or virtual environment was not created if the folder or environment is still missing.

The workspace now needs a dependency manifest and a small AWS CDK configuration file. Both files stay pinned so later synthesis and testing remain reproducible.

• Click the New File icon in the Visual Studio Code Explorer sidebar.
• Enter requirements.txt as the file name.
• Add the pinned Python dependencies by pasting this file content:

aws-cdk-lib==2.275.0
constructs==10.8.1
boto3==1.43.126
botocore==1.43.126
pytest==9.2.0

What does this dependency file control?

The file pins the AWS CDK library, its construct layer, the AWS SDK packages, and the test runner. Exact pins prevent a later package release from silently changing the project environment.

• Save requirements.txt.
• Click the New File icon in the Explorer sidebar.
• Enter cdk.json as the file name.
• Add the CDK application settings by pasting this file content:

{
  "app": "python3 app.py",
  "outputsFile": "outputs.json"
}

What does this configuration control?

The app setting identifies the Python entry point that AWS CDK will synthesize later. The outputsFile setting gives deployed stack outputs a deterministic local destination.

• Save cdk.json.
• Confirm the Explorer sidebar lists requirements.txt and cdk.json inside secure-agent-gateway.

✔️ Awesome, I've got everything!

Your two setup files are saved in the workspace. Continue with the dependency installation.

ⓧ I'd like to double check the full code

Your requirements.txt file should match this reference.

aws-cdk-lib==2.275.0
constructs==10.8.1
boto3==1.43.126
botocore==1.43.126
pytest==9.2.0

Your cdk.json file should match this reference.

{
  "app": "python3 app.py",
  "outputsFile": "outputs.json"
}

The first dependency install can take a minute while Python downloads the pinned packages. A stream of package names in Terminal means the installation is progressing.

• Switch back to Terminal.
• Activate the virtual environment and install the pinned tools by running these commands:

source .venv/bin/activate
python -m pip install -r requirements.txt
python --version
npx aws-cdk@2.1148.0 --version

What does this setup prove?

• Activation makes commands in this Terminal session use the interpreter inside .venv.
• The package installation reads every exact pin from requirements.txt.
• The final checks report Python 3.12 and AWS CDK Toolkit 2.1148.0.

The Terminal prompt now begins with (.venv). The version checks report Python 3.12 and CDK Toolkit 2.1148.0.

✔️ I see the pinned CDK version

AWS CDK Toolkit reports 2.1148.0. Your local CDK command is pinned for this project.

ⓧ I see a different version

The project command requests AWS CDK Toolkit 2.1148.0 directly, so a different result means the version check did not finish as expected.

• Confirm the command contains aws-cdk@2.1148.0.
• Repeat the pinned version check from the command block above.

ⓧ The CDK check cannot run

The npx command comes from Node.js, which must be available before the pinned toolkit can run.

• Install Node.js from its official website.
• Return to Terminal after the installation completes.
• Repeat the pinned version check from the command block above.

Dependency installation failed?

• Confirm the prompt begins with (.venv) before installing the packages.
• Confirm every line in requirements.txt matches the pinned reference.
• Use help me troubleshoot the pinned environment installation if the install still fails.

Authenticate the project profile

The preflight trusts the account and Region you selected before authentication. It compares those expectations with a temporary SSO session instead of accepting whichever account happens to be active.

• Check the installed AWS CLI version by running this command:

aws --version

What does this check do?

The command reports the installed AWS CLI release. This project uses AWS CLI v2 because its SSO configuration and login flow provide temporary credentials for the named profile.

✔️ I see AWS CLI v2

AWS CLI v2 is ready to configure the project profile.

ⓧ I see AWS CLI v1

AWS CLI v1 is older than the toolchain used by this project.

• Follow the official AWS CLI installation guide to install or update AWS CLI v2.
• Return to Terminal after the update completes.
• Repeat the version check above.

ⓧ Command not found

Your shell cannot locate the AWS CLI installation.

• Follow the official AWS CLI installation guide to install AWS CLI v2.
• Return to Terminal after the installation completes.
• Repeat the version check above.

The SSO setup wizard records how the CLI reaches your organization's IAM Identity Center. It also lets you select the intended account and permission set before naming the profile.

• Start the SSO configuration wizard by running this command:

aws configure sso

What happens during SSO setup?

The CLI asks for your organization's SSO details before opening a browser authorization flow. It then presents the accounts and permission sets available to your identity.

• Complete the browser authorization when it opens.
• Select the intended AWS account.
• Select the permission set approved for this project.
• Select the intended AWS Region.
• Enter agent-gateway as the CLI profile name.

The login command may open another browser window to refresh your temporary session. That browser flow is expected and keeps long-lived access keys out of the project.

• Authenticate the named profile and capture the selected security boundary by running these commands:

export AWS_PROFILE=agent-gateway
aws sso login --profile "$AWS_PROFILE"
read -r -p "Expected 12-digit AWS account ID: " EXPECTED_ACCOUNT_ID
read -r -p "Expected AWS Region: " EXPECTED_REGION
export EXPECTED_ACCOUNT_ID EXPECTED_REGION
export AWS_REGION="$EXPECTED_REGION"
export TRUSTED_CALLER_ARN="$(aws sts get-caller-identity --profile "$AWS_PROFILE" --query Arn --output text)"

What does this session setup bind?

• The AWS_PROFILE value routes SDK and CLI requests through agent-gateway.
• The expected account and Region come from your explicit choices.
• The AWS_REGION value makes the SDK session use that expected Region.
• The TRUSTED_CALLER_ARN value records the active temporary assumed-role session for the stack's trust policy.

The SSO login completes with a browser success confirmation. Terminal then waits for the account ID and Region you selected before returning to the activated environment prompt.

SSO login or profile unavailable?

• Confirm the profile name entered during setup is agent-gateway.
• Restart the SSO login if the browser authorization expires.
• Use help me troubleshoot the temporary SSO session if the session still cannot authenticate.

Run the read-only preflight

The preflight treats the selected account and Region as invariants. It reads identity, bootstrap, and legacy role state without creating resources or changing IAM.

• Switch back to Visual Studio Code.
• Click the New File icon in the Explorer sidebar.
• Enter preflight.py as the file name.
• Add the imports, read-only resource names, and AWS error helper by pasting this first chunk:

import json
import os

import boto3


LEGACY_ROLE_NAME = "SecureAgentGatewayInvokerRole"
BOOTSTRAP_PARAMETER = "/cdk-bootstrap/hnb659fds/version"


def error_code(error: Exception) -> str:
    return getattr(error, "response", {}).get("Error", {}).get("Code", "Unknown")

What does this first chunk define?

• The imports provide environment access, deterministic JSON output, and AWS SDK clients.
• The bootstrap parameter identifies the modern CDK bootstrap version record in Parameter Store.
• The legacy role name is used only as a conflict indicator.
• The helper extracts AWS error codes without changing the inspected resource.

• Add the expectation, Region, account, and temporary-credential checks directly below error_code() by pasting this chunk:

def main() -> None:
    expected_account = os.environ.get("EXPECTED_ACCOUNT_ID", "")
    expected_region = os.environ.get("EXPECTED_REGION", "")
    if not expected_account.isdigit() or len(expected_account) != 12:
        raise SystemExit("EXPECTED_ACCOUNT_ID must be a 12-digit account ID")
    if not expected_region:
        raise SystemExit("EXPECTED_REGION is required")

    session = boto3.Session()
    if session.region_name != expected_region:
        raise SystemExit(
            f"Region mismatch: profile={session.region_name!r}, expected={expected_region!r}"
        )

    identity = session.client("sts").get_caller_identity()
    caller_arn = identity["Arn"]
    temporary = ":sts::" in caller_arn and ":assumed-role/" in caller_arn
    if identity["Account"] != expected_account:
        raise SystemExit(
            f"Account mismatch: active={identity['Account']}, expected={expected_account}"
        )
    if not temporary:
        raise SystemExit("Active credentials are not an STS assumed-role session")

How does the boundary fail closed?

The function rejects missing expectations before making AWS calls. It also exits when the SDK Region, STS account, or credential type differs from the selected boundary.

• Save preflight.py.
• Confirm the two initial chunks load correctly by switching back to Terminal.
• Run the current script:

python preflight.py

What does this intermediate run prove?

The Terminal prompt returns without a Python traceback. That confirms the imports and function definitions parse before you add the AWS inspection branches.

• Return to preflight.py in Visual Studio Code.
• Add the bootstrap and legacy role inspections below the temporary-credential check by pasting this chunk:

    bootstrap_state = "UNKNOWN"
    bootstrap_detail = "not checked"
    try:
        response = session.client("ssm").get_parameter(Name=BOOTSTRAP_PARAMETER)
        bootstrap_state = "READY"
        bootstrap_detail = response["Parameter"]["Value"]
    except Exception as error:
        code = error_code(error)
        if code == "ParameterNotFound":
            bootstrap_state = "MISSING"
            bootstrap_detail = BOOTSTRAP_PARAMETER
        else:
            bootstrap_detail = code

    legacy_role_state = "UNKNOWN"
    try:
        session.client("iam").get_role(RoleName=LEGACY_ROLE_NAME)
        legacy_role_state = "EXISTS_UNTOUCHED"
    except Exception as error:
        code = error_code(error)
        if code == "NoSuchEntity":
            legacy_role_state = "NOT_FOUND"
        else:
            legacy_role_state = f"UNKNOWN:{code}"

What do these inspections read?

The first branch reads the CDK bootstrap version parameter from AWS Systems Manager Parameter Store. The second asks AWS Identity and Access Management whether the fixed legacy role exists.

Neither branch creates, imports, updates, replaces, or deletes the role. An uninspectable state remains UNKNOWN so the deployment cannot mistake missing permission for a safe result.

• Add the deterministic JSON report and script entry point below the legacy role inspection by pasting this final chunk:

    print(
        json.dumps(
            {
                "account_id": identity["Account"],
                "region": session.region_name,
                "caller_arn": caller_arn,
                "temporary_credentials": temporary,
                "bootstrap_state": bootstrap_state,
                "bootstrap_detail": bootstrap_detail,
                "legacy_role_name": LEGACY_ROLE_NAME,
                "legacy_role_state": legacy_role_state,
                "writes_performed": False,
            },
            indent=2,
            sort_keys=True,
        )
    )


if __name__ == "__main__":
    main()

Why produce deterministic JSON?

The report gathers every preflight decision into a stable evidence format. The explicit writes_performed field records that this script only inspected the boundary.

• Save preflight.py.

✔️ Awesome, I've got everything!

Your complete preflight script is saved. Continue to the live read-only verification.

ⓧ I'd like to double check the full code

Compare preflight.py with this complete reference.

import json
import os

import boto3


LEGACY_ROLE_NAME = "SecureAgentGatewayInvokerRole"
BOOTSTRAP_PARAMETER = "/cdk-bootstrap/hnb659fds/version"


def error_code(error: Exception) -> str:
    return getattr(error, "response", {}).get("Error", {}).get("Code", "Unknown")


def main() -> None:
    expected_account = os.environ.get("EXPECTED_ACCOUNT_ID", "")
    expected_region = os.environ.get("EXPECTED_REGION", "")
    if not expected_account.isdigit() or len(expected_account) != 12:
        raise SystemExit("EXPECTED_ACCOUNT_ID must be a 12-digit account ID")
    if not expected_region:
        raise SystemExit("EXPECTED_REGION is required")

    session = boto3.Session()
    if session.region_name != expected_region:
        raise SystemExit(
            f"Region mismatch: profile={session.region_name!r}, expected={expected_region!r}"
        )

    identity = session.client("sts").get_caller_identity()
    caller_arn = identity["Arn"]
    temporary = ":sts::" in caller_arn and ":assumed-role/" in caller_arn
    if identity["Account"] != expected_account:
        raise SystemExit(
            f"Account mismatch: active={identity['Account']}, expected={expected_account}"
        )
    if not temporary:
        raise SystemExit("Active credentials are not an STS assumed-role session")

    bootstrap_state = "UNKNOWN"
    bootstrap_detail = "not checked"
    try:
        response = session.client("ssm").get_parameter(Name=BOOTSTRAP_PARAMETER)
        bootstrap_state = "READY"
        bootstrap_detail = response["Parameter"]["Value"]
    except Exception as error:
        code = error_code(error)
        if code == "ParameterNotFound":
            bootstrap_state = "MISSING"
            bootstrap_detail = BOOTSTRAP_PARAMETER
        else:
            bootstrap_detail = code

    legacy_role_state = "UNKNOWN"
    try:
        session.client("iam").get_role(RoleName=LEGACY_ROLE_NAME)
        legacy_role_state = "EXISTS_UNTOUCHED"
    except Exception as error:
        code = error_code(error)
        if code == "NoSuchEntity":
            legacy_role_state = "NOT_FOUND"
        else:
            legacy_role_state = f"UNKNOWN:{code}"

    print(
        json.dumps(
            {
                "account_id": identity["Account"],
                "region": session.region_name,
                "caller_arn": caller_arn,
                "temporary_credentials": temporary,
                "bootstrap_state": bootstrap_state,
                "bootstrap_detail": bootstrap_detail,
                "legacy_role_name": LEGACY_ROLE_NAME,
                "legacy_role_state": legacy_role_state,
                "writes_performed": False,
            },
            indent=2,
            sort_keys=True,
        )
    )


if __name__ == "__main__":
    main()

Before you run the completed preflight, do you expect the active session to match every boundary you selected?

• Run the completed preflight from the activated environment:

python preflight.py

What should the report prove?

• The account and Region match the values you entered.
• The caller ARN uses an STS assumed-role session and temporary_credentials is true.
• The bootstrap state is READY, MISSING, or UNKNOWN.
• The legacy role state reports NOT_FOUND, EXISTS_UNTOUCHED, or an UNKNOWN result.
• The final writes_performed value is false.

That is the security boundary confirmed. Your selected account, Region, temporary session, bootstrap visibility, and untouched legacy role state now have deterministic evidence.

✔️ Bootstrap is ready

A READY state means the modern bootstrap version parameter is visible in the selected account and Region. Leave the shared CDKToolkit resources in place.

ⓧ Bootstrap is missing

Preparing bootstrap infrastructure can feel risky because it affects shared account resources. The next command creates a reviewable change set with execution disabled, so it does not apply the bootstrap changes.

• Prepare the bootstrap change set without executing it by running:

npx aws-cdk@2.1148.0 bootstrap \
  "aws://${EXPECTED_ACCOUNT_ID}/${EXPECTED_REGION}" \
  --profile "$AWS_PROFILE" \
  --execute false \
  --termination-protection

What does preparation protect?

The command asks AWS CDK to prepare bootstrap changes with termination protection requested. The --execute false flag keeps the prepared change set from running.

• Export the canonical bootstrap template for administrator review by running:

npx aws-cdk@2.1148.0 bootstrap --show-template > bootstrap-template.yaml

What does the exported template provide?

The command writes the toolkit's canonical bootstrap template to bootstrap-template.yaml without deploying it. An administrator can inspect the exact infrastructure and permission requirements.

• Confirm the Visual Studio Code Explorer lists bootstrap-template.yaml inside secure-agent-gateway.
• Stop AWS writes if preparation reports AccessDenied.
• Provide bootstrap-template.yaml and the later synthesized application template to an administrator.

Bootstrap preparation denied?

Keep the local workspace and read-only evidence. Do not use root credentials, create long-lived keys, weaken application policies, or replace shared bootstrap resources.

Use help me prepare an administrator review package if you need help documenting the blocked path.

ⓧ Bootstrap state is unknown

An UNKNOWN state means the preflight could not safely distinguish an unavailable parameter from missing inspection permission. Keep AWS writes paused.

• Export the canonical bootstrap template for administrator review by running:

npx aws-cdk@2.1148.0 bootstrap --show-template > bootstrap-template.yaml

Why export without preparing?

The generated bootstrap-template.yaml file gives an administrator a reviewable artifact while your session avoids further AWS writes.

• Confirm the Visual Studio Code Explorer lists bootstrap-template.yaml.
• Provide the template with the intended account and Region to an administrator.
• Continue only with local tests while bootstrap access is reviewed.

Preflight stopped on a mismatch?

• Re-enter the intended account ID if the value is not exactly 12 digits.
• Re-authenticate the agent-gateway profile if the caller is missing or no longer temporary.
• Keep AWS writes paused when the active account or Region differs from your selected boundary.
• Use help me diagnose the failed AWS preflight if the mismatch remains unclear.

Your deployment boundary is pinned and reviewable. Next, you will build the local gateway that converts a verified role ARN into narrow tenant permissions with default deny.

Build and Test the Identity-Bound Gateway

Your preflight proved that the workspace points to the intended AWS account and Region. It also confirmed that your active credentials come from a temporary assumed-role session.

An authenticated ARN identifies the caller. Your gateway still needs a server-owned rule that maps this identity to one tenant and a narrow permission set. Every unbound principal or unknown action must reach a default-deny decision.

In this step, get ready to:

• Build the identity normalization and default-deny policy layer.
• Connect the policy layer to the verified request context.
• Prove legitimate access and seven security outcomes with deterministic tests.

Build the server-owned identity boundary

An IAM-authorized request reaches Amazon API Gateway with a verified assumed-role session ARN. The gateway converts that session ARN to its IAM role ARN before comparing it with the trusted role.

The matching role receives the server-configured tenant and permissions. Request-body fields never choose the principal or tenant.

• Select the `secure-agent-gateway` workspace in the VS Code Explorer sidebar.
• Click the New Folder button in the Explorer toolbar.
• Enter gateway as the folder name.
• Select the new gateway folder.
• Click the New File button in the Explorer toolbar.
• Enter __init__.py as the file name.
• Click the New File button again.
• Enter domain.py as the file name.

• Mark gateway/__init__.py as the package entry point by entering this code:

"""Secure agent execution gateway package."""

What does this file do?

The package marker lets Python import code through paths such as gateway.domain. This keeps the policy layer available to the handler and tests.

• Save gateway/__init__.py.
• Confirm that __init__.py appears inside the gateway folder in the Explorer sidebar.

Package file missing?

Check that __init__.py contains two leading underscores and two trailing underscores. Confirm that the file sits inside gateway.

Still stuck? Help me check the Python package structure in my secure-agent-gateway workspace.

The policy module has four responsibilities. It validates identity first. It then validates the request shape before authorizing an action and touching tenant-scoped storage.

• Build gateway/domain.py by entering the complete file from the double-check tab below:

✔️ Awesome, I've got everything!

Keep the identity boundary server-owned. The request may provide an action and record data only.

• Normalize the verified STS session ARN with role_session_to_role_arn().
• Bind the exact trusted role to tenant-blue through verify_identity().
• Reject authority fields through validate_request().
• Apply the permission allowlist through authorize().
• Keep local tests deterministic through MemoryStore.

ⓧ I'd like to double check the full code

from __future__ import annotations

import json
import re
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Callable, Protocol


RECORD_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,64}$")


class ValidationFailure(Exception):
    pass


class IdentityFailure(Exception):
    pass


class DownstreamFailure(Exception):
    pass


class DuplicateRecord(Exception):
    pass


@dataclass(frozen=True)
class Identity:
    principal_arn: str
    tenant_id: str
    permissions: frozenset[str]


@dataclass(frozen=True)
class AgentRequest:
    action: str
    record_id: str
    data: dict[str, Any] | None


@dataclass(frozen=True)
class Decision:
    allowed: bool
    reason: str


class Store(Protocol):
    def read(self, tenant_id: str, record_id: str) -> dict[str, Any] | None: ...
    def create_with_audit(self, tenant_id: str, record_id: str, data: dict[str, Any], audit: dict[str, str]) -> None: ...
    def write_audit(self, audit: dict[str, str]) -> None: ...


def role_session_to_role_arn(caller_arn: str) -> str:
    parts = caller_arn.split(":", 5)
    if len(parts) != 6 or parts[2] != "sts":
        raise IdentityFailure("malformed_principal_arn")
    resource = parts[5].split("/", 2)
    if len(resource) != 3 or resource[0] != "assumed-role" or not resource[2]:
        raise IdentityFailure("malformed_principal_arn")
    return f"arn:{parts[1]}:iam::{parts[4]}:role/{resource[1]}"


def verify_identity(
    caller_arn: str | None,
    trusted_role_arn: str,
    tenant_id: str,
    permissions: frozenset[str],
) -> Identity:
    if not caller_arn:
        raise IdentityFailure("missing_verified_identity")
    if role_session_to_role_arn(caller_arn) != trusted_role_arn:
        raise IdentityFailure("untrusted_principal")
    return Identity(caller_arn, tenant_id, permissions)


def validate_request(payload: Any) -> AgentRequest:
    if not isinstance(payload, dict):
        raise ValidationFailure("body_must_be_object")
    allowed_keys = {"action", "record_id", "data"}
    if set(payload) - allowed_keys:
        raise ValidationFailure("unexpected_fields")
    action = payload.get("action")
    record_id = payload.get("record_id")
    data = payload.get("data")
    if not isinstance(action, str) or not action or len(action) > 64:
        raise ValidationFailure("invalid_action")
    if not isinstance(record_id, str) or not RECORD_ID_PATTERN.fullmatch(record_id):
        raise ValidationFailure("invalid_record_id")
    if action == "records:create":
        if not isinstance(data, dict) or not data:
            raise ValidationFailure("create_requires_data")
        if len(data) > 20 or len(json.dumps(data, separators=(",", ":"))) > 4096:
            raise ValidationFailure("data_too_large")
    elif data is not None:
        raise ValidationFailure("data_not_allowed_for_action")
    return AgentRequest(action, record_id, data)


def authorize(identity: Identity, request: AgentRequest) -> Decision:
    if request.action in identity.permissions:
        return Decision(True, "allowlisted_action")
    return Decision(False, "default_deny")


def _record_key(tenant_id: str, record_id: str) -> str:
    return f"TENANT#{tenant_id}#RECORD#{record_id}"


def _string_item(values: dict[str, str]) -> dict[str, dict[str, str]]:
    return {key: {"S": value} for key, value in values.items()}


class DynamoStore:
    def __init__(self, client: Any, table_name: str) -> None:
        self.client = client
        self.table_name = table_name

    def read(self, tenant_id: str, record_id: str) -> dict[str, Any] | None:
        try:
            response = self.client.get_item(
                TableName=self.table_name,
                Key={"pk": {"S": _record_key(tenant_id, record_id)}},
                ConsistentRead=True,
            )
        except Exception as error:
            raise DownstreamFailure("read_failed") from error
        item = response.get("Item")
        return None if not item else json.loads(item["payload"]["S"])

    def create_with_audit(self, tenant_id: str, record_id: str, data: dict[str, Any], audit: dict[str, str]) -> None:
        record = _string_item({
            "pk": _record_key(tenant_id, record_id),
            "entity_type": "record",
            "tenant_id": tenant_id,
            "record_id": record_id,
            "payload": json.dumps(data, separators=(",", ":"), sort_keys=True),
        })
        audit_item = _string_item({"pk": f"AUDIT#{audit['request_id']}", **audit})
        try:
            self.client.transact_write_items(TransactItems=[
                {"Put": {"TableName": self.table_name, "Item": record, "ConditionExpression": "attribute_not_exists(pk)"}},
                {"Put": {"TableName": self.table_name, "Item": audit_item}},
            ])
        except Exception as error:
            code = getattr(error, "response", {}).get("Error", {}).get("Code")
            if code == "TransactionCanceledException":
                raise DuplicateRecord("record_exists") from error
            raise DownstreamFailure("create_failed") from error

    def write_audit(self, audit: dict[str, str]) -> None:
        try:
            self.client.put_item(
                TableName=self.table_name,
                Item=_string_item({"pk": f"AUDIT#{audit['request_id']}", **audit}),
                ConditionExpression="attribute_not_exists(pk)",
            )
        except Exception as error:
            raise DownstreamFailure("audit_failed") from error


class MemoryStore:
    def __init__(self, records: dict[tuple[str, str], dict[str, Any]] | None = None, *, fail_writes: bool = False) -> None:
        self.records = dict(records or {})
        self.audits: list[dict[str, str]] = []
        self.fail_writes = fail_writes

    def read(self, tenant_id: str, record_id: str) -> dict[str, Any] | None:
        return self.records.get((tenant_id, record_id))

    def create_with_audit(self, tenant_id: str, record_id: str, data: dict[str, Any], audit: dict[str, str]) -> None:
        if self.fail_writes:
            raise DownstreamFailure("create_failed")
        key = (tenant_id, record_id)
        if key in self.records:
            raise DuplicateRecord("record_exists")
        self.records[key] = data
        self.audits.append(audit)

    def write_audit(self, audit: dict[str, str]) -> None:
        if self.fail_writes:
            raise DownstreamFailure("audit_failed")
        self.audits.append(audit)


class GatewayService:
    def __init__(
        self,
        store: Store,
        trusted_role_arn: str,
        tenant_id: str = "tenant-blue",
        permissions: frozenset[str] = frozenset({"records:read", "records:create"}),
        emit_log: Callable[[dict[str, Any]], None] | None = None,
    ) -> None:
        self.store = store
        self.trusted_role_arn = trusted_role_arn
        self.tenant_id = tenant_id
        self.permissions = permissions
        self.emit_log = emit_log or (lambda event: None)

    def handle(self, event: dict[str, Any]) -> dict[str, Any]:
        started = time.perf_counter_ns()
        context = event.get("requestContext") or {}
        request_id = context.get("requestId") or uuid.uuid4().hex
        caller_arn = (context.get("identity") or {}).get("userArn")
        try:
            identity = verify_identity(caller_arn, self.trusted_role_arn, self.tenant_id, self.permissions)
        except IdentityFailure as error:
            audit = self._audit(request_id, caller_arn or "unknown", "unknown", "unknown", "unknown", "DENY", str(error), started)
            self._persist_denial(audit)
            return self._response(401, {"decision": "DENY", "reason": str(error)})
        try:
            raw_body = event.get("body") or "{}"
            request = validate_request(json.loads(raw_body) if isinstance(raw_body, str) else raw_body)
        except (json.JSONDecodeError, ValidationFailure) as error:
            reason = "invalid_json" if isinstance(error, json.JSONDecodeError) else str(error)
            audit = self._audit(request_id, identity.principal_arn, identity.tenant_id, "unknown", "unknown", "DENY", reason, started)
            self._persist_denial(audit)
            return self._response(400, {"decision": "DENY", "reason": reason})
        decision = authorize(identity, request)
        if not decision.allowed:
            audit = self._audit(request_id, identity.principal_arn, identity.tenant_id, request.action, request.record_id, "DENY", decision.reason, started)
            self._persist_denial(audit)
            return self._response(403, {"decision": "DENY", "reason": decision.reason})
        try:
            if request.action == "records:read":
                result = self.store.read(identity.tenant_id, request.record_id)
                audit = self._audit(request_id, identity.principal_arn, identity.tenant_id, request.action, request.record_id, "ALLOW", "read_complete", started)
                self.store.write_audit(audit)
                self.emit_log({"audit": audit, "audit_persisted": True})
                return self._response(404 if result is None else 200, {"decision": "ALLOW", "record": result})
            audit = self._audit(request_id, identity.principal_arn, identity.tenant_id, request.action, request.record_id, "ALLOW", "create_complete", started)
            self.store.create_with_audit(identity.tenant_id, request.record_id, request.data or {}, audit)
            self.emit_log({"audit": audit, "audit_persisted": True})
            return self._response(201, {"decision": "ALLOW", "result": "created"})
        except DuplicateRecord:
            return self._response(409, {"decision": "DENY", "reason": "record_exists"})
        except DownstreamFailure:
            audit = self._audit(request_id, identity.principal_arn, identity.tenant_id, request.action, request.record_id, "DENY", "downstream_failure", started)
            self.emit_log({"audit": audit, "audit_persisted": False})
            return self._response(503, {"decision": "DENY", "reason": "downstream_failure"})

    def _persist_denial(self, audit: dict[str, str]) -> None:
        persisted = True
        try:
            self.store.write_audit(audit)
        except DownstreamFailure:
            persisted = False
        self.emit_log({"audit": audit, "audit_persisted": persisted})

    @staticmethod
    def _audit(request_id: str, principal_arn: str, tenant_id: str, action: str, record_id: str, decision: str, reason: str, started: int) -> dict[str, str]:
        return {
            "request_id": request_id,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "principal_arn": principal_arn,
            "tenant_id": tenant_id,
            "action": action,
            "record_id": record_id,
            "decision": decision,
            "reason": reason,
            "latency_us": str((time.perf_counter_ns() - started) // 1000),
        }

    @staticmethod
    def _response(status_code: int, payload: dict[str, Any]) -> dict[str, Any]:
        return {"statusCode": status_code, "headers": {"content-type": "application/json"}, "body": json.dumps(payload, separators=(",", ":"), sort_keys=True)}

How does the security boundary work?

• The role_session_to_role_arn() function converts the verified session identity into a stable IAM role ARN. Session names cannot change the trusted role comparison.
• The verify_identity() function returns an Identity only for an exact trusted-role match. The tenant and permissions come from server configuration.
• The validate_request() function accepts only action, record_id, and data. Fields such as role, principal_id, and tenant_id become validation failures.
• The authorize() function permits configured actions only. Every other action receives default_deny.
• The DynamoStore implementation prefixes record keys with the tenant. It uses an atomic Amazon DynamoDB transaction for each record and audit pair.

• Save gateway/domain.py.
• Confirm that the VS Code Explorer shows saved versions of gateway/__init__.py and gateway/domain.py.

Seeing Python problems in the policy module?

Check that every class and function remains at the indentation level shown in the full reference. Confirm that from __future__ import annotations is the first line.

Still stuck? Help me compare my identity-bound gateway domain module with the expected structure.

Connect the Lambda handler

The deployable AWS Lambda handler needs a small adapter around GatewayService. It reads the effective caller only from requestContext.identity.userArn.

The trusted role ARN and tenant permissions come from environment configuration. This keeps request-body claims outside the authority path.

• Select the gateway folder in the VS Code Explorer sidebar.
• Click the New File button in the Explorer toolbar.
• Enter handler.py as the file name.
• Configure the Lambda adapter in gateway/handler.py by entering this code:

import json
import os
from typing import Any

import boto3

from gateway.domain import DynamoStore, GatewayService


def emit_structured_log(event: dict[str, Any]) -> None:
    print(json.dumps(event, separators=(",", ":"), sort_keys=True))


_service = GatewayService(
    DynamoStore(boto3.client("dynamodb"), os.environ["TABLE_NAME"]),
    trusted_role_arn=os.environ["TRUSTED_ROLE_ARN"],
    tenant_id=os.environ["TENANT_ID"],
    permissions=frozenset(os.environ["ALLOWED_ACTIONS"].split(",")),
    emit_log=emit_structured_log,
)


def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    del context
    return _service.handle(event)

What does this handler do?

• The module creates one GatewayService backed by DynamoStore.
• The TRUSTED_ROLE_ARN value selects the only role that can receive the configured binding.
• The TENANT_ID value supplies tenant-blue during deployment.
• The ALLOWED_ACTIONS value becomes an immutable permission set containing records:read and records:create.
• The lambda_handler() function delegates the verified event to the service. Structured logs preserve the resulting audit evidence.

• Save gateway/handler.py.
• Confirm that handler.py appears beside domain.py in the gateway folder.

Handler imports not resolving?

Check that the folder contains __init__.py. Confirm that the import uses from gateway.domain import DynamoStore, GatewayService exactly.

Still stuck? Help me troubleshoot the gateway handler imports and environment configuration.

Double-check the complete handler before adding the security tests.

✔️ Awesome, I've got everything!

Great. Save gateway/handler.py before continuing.

ⓧ I'd like to double check the full code

import json
import os
from typing import Any

import boto3

from gateway.domain import DynamoStore, GatewayService


def emit_structured_log(event: dict[str, Any]) -> None:
    print(json.dumps(event, separators=(",", ":"), sort_keys=True))


_service = GatewayService(
    DynamoStore(boto3.client("dynamodb"), os.environ["TABLE_NAME"]),
    trusted_role_arn=os.environ["TRUSTED_ROLE_ARN"],
    tenant_id=os.environ["TENANT_ID"],
    permissions=frozenset(os.environ["ALLOWED_ACTIONS"].split(",")),
    emit_log=emit_structured_log,
)


def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    del context
    return _service.handle(event)

Prove the policy fails closed

A security boundary needs evidence for both allowed and denied requests. These pytest cases use MemoryStore so every outcome stays local and deterministic.

The suite also removes records:create from the server-owned permissions for one test. A correctly authenticated principal still receives 403 with default_deny for that action.

• Select the secure-agent-gateway workspace in the VS Code Explorer sidebar.
• Click the New Folder button in the Explorer toolbar.
• Enter tests as the folder name.
• Select the new tests folder.
• Click the New File button in the Explorer toolbar.
• Enter test_security.py as the file name.
• Build the seven deterministic cases in tests/test_security.py by entering the complete file from the double-check tab below:

✔️ Awesome, I've got everything!

Keep the tests local. The suite must not invoke AWS or depend on a deployed resource.

• Use the generated-role ARN and its corresponding simulated STS session ARN.
• Cover legitimate create and read operations.
• Cover forged authority fields and an untrusted caller.
• Cover a cross-tenant claim and an unknown action.
• Cover a removed permission and a downstream write failure.

ⓧ I'd like to double check the full code

import json

from gateway.domain import GatewayService, MemoryStore


ROLE_ARN = "arn:aws:iam::123456789012:role/generated-agent-role"
SESSION_ARN = "arn:aws:sts::123456789012:assumed-role/generated-agent-role/simulated-agent"


def event(payload: dict, caller_arn: str | None = SESSION_ARN) -> dict:
    identity = {} if caller_arn is None else {"userArn": caller_arn}
    return {"body": json.dumps(payload), "requestContext": {"requestId": "request-1", "identity": identity}}


def service(store: MemoryStore | None = None, permissions: frozenset[str] = frozenset({"records:read", "records:create"})) -> GatewayService:
    return GatewayService(store or MemoryStore(), ROLE_ARN, permissions=permissions)


def test_legitimate_create_and_read() -> None:
    store = MemoryStore()
    gateway = service(store)
    assert gateway.handle(event({"action": "records:create", "record_id": "run-1", "data": {"pace": 8}}))["statusCode"] == 201
    assert gateway.handle(event({"action": "records:read", "record_id": "run-1"}))["statusCode"] == 200


def test_privilege_escalation_is_rejected() -> None:
    response = service().handle(event({"action": "records:read", "record_id": "run-1", "role": "admin", "principal_id": "root"}))
    assert response["statusCode"] == 400


def test_forged_identity_is_rejected() -> None:
    forged = "arn:aws:sts::123456789012:assumed-role/forged-admin/attacker"
    assert service().handle(event({"action": "records:read", "record_id": "run-1"}, forged))["statusCode"] == 401


def test_cross_tenant_access_is_rejected() -> None:
    store = MemoryStore({("tenant-red", "secret"): {"value": "classified"}})
    response = service(store).handle(event({"action": "records:read", "record_id": "secret", "tenant_id": "tenant-red"}))
    assert response["statusCode"] == 400
    assert "classified" not in response["body"]


def test_unknown_action_is_denied() -> None:
    response = service().handle(event({"action": "admin:delete", "record_id": "run-1"}))
    assert response["statusCode"] == 403
    assert json.loads(response["body"])["reason"] == "default_deny"


def test_bound_principal_still_defaults_to_deny_without_action_permission() -> None:
    gateway = service(permissions=frozenset({"records:read"}))
    response = gateway.handle(event({"action": "records:create", "record_id": "run-1", "data": {"pace": 8}}))
    assert response["statusCode"] == 403
    assert json.loads(response["body"])["reason"] == "default_deny"


def test_downstream_failure_fails_closed_without_mutation() -> None:
    store = MemoryStore(fail_writes=True)
    response = service(store).handle(event({"action": "records:create", "record_id": "run-1", "data": {"pace": 8}}))
    assert response["statusCode"] == 503
    assert store.records == {}

What do these tests prove?

• The legitimate test proves that the trusted session can create and read a record inside tenant-blue.
• The privilege-escalation test proves that body fields cannot grant authority.
• The forged-identity test proves that a different role session cannot inherit the trusted binding.
• The cross-tenant test proves that a body claim cannot expose the protected tenant-red record.
• The default-deny tests prove that authenticated identity alone cannot authorize an unknown or removed action.
• The downstream-failure test proves that a failed create leaves MemoryStore unchanged.

• Save tests/test_security.py.

Before you run the suite, which requests do you expect to reach the store after authorization?

• Run the focused security suite in the existing VS Code terminal with this command:

python -m pytest -q tests/test_security.py

You should see seven passing tests. The forged caller case is denied before any protected record can be read.

That is the intended boundary: local request data cannot create a trusted gateway identity. Excellent work. Your service now permits the two tenant-scoped actions while every tested escalation path fails closed.

Tests not passing?

• Confirm that the terminal still uses the .venv interpreter from the previous step.
• Check that gateway and tests sit directly inside secure-agent-gateway.
• Compare the failing test name with the matching function in gateway/domain.py.
• Still stuck? Help me diagnose the failing identity-bound gateway security test.

Your local authorization boundary now has deterministic evidence behind it. Next, you will place AWS_IAM and signed requests in front of the service before any live operation reaches it.

Prepare, Inspect, Deploy, and Smoke-Test AWS

Your local gateway now rejects forged authority while preserving legitimate tenant-scoped access. The next goal is to place that security boundary behind an authenticated AWS endpoint.

A synthesized CloudFormation template shows the intended resources. However, deployability requires a prepared change set that tests the practical permission path before execution. You will review that evidence before allowing AWS CDK to create the stack.

In this step, get ready to:

• Define the generated roles, data store, logs, Lambda function, and IAM-protected REST method.
• Synthesize the application and inspect a non-executed change set.
• Deploy the reviewed stack and prove its security controls with live smoke tests.

Build and inspect the CDK application

The stack keeps authority server-owned from deployment through runtime. Amazon API Gateway verifies signed requests before Lambda runs. CloudFormation generates both role names so the legacy fixed-name role remains untouched.

• In the VS Code Explorer sidebar, click the New File control.
• Enter app.py inside the secure-agent-gateway folder.
• Configure the CDK entry point by adding this code:

import os

from aws_cdk import App

from gateway_stack import SecureAgentGatewayStack


trusted_caller_arn = os.environ.get("TRUSTED_CALLER_ARN")
if not trusted_caller_arn:
    raise ValueError("TRUSTED_CALLER_ARN must come from aws sts get-caller-identity")

app = App()
SecureAgentGatewayStack(
    app,
    "SecureAgentGatewayStack",
    trusted_caller_arn=trusted_caller_arn,
)
app.synth()

What does this entry point do?

• The environment lookup requires the temporary caller identity captured during preflight.
• The stack receives that caller ARN as the only principal allowed to assume the generated invoker role.
• The final call converts the CDK construct tree into a CloudFormation template.

• Save app.py.

Entry point not saving correctly?

Confirm that app.py sits directly inside secure-agent-gateway. Check that the environment lookup uses TRUSTED_CALLER_ARN exactly.

Still stuck? Help me check my CDK entry point.

✔️ Awesome, I've got everything!

Great. Confirm that you saved app.py before building the stack.

ⓧ I'd like to double check the full code

import os

from aws_cdk import App

from gateway_stack import SecureAgentGatewayStack


trusted_caller_arn = os.environ.get("TRUSTED_CALLER_ARN")
if not trusted_caller_arn:
    raise ValueError("TRUSTED_CALLER_ARN must come from aws sts get-caller-identity")

app = App()
SecureAgentGatewayStack(
    app,
    "SecureAgentGatewayStack",
    trusted_caller_arn=trusted_caller_arn,
)
app.synth()

Use this reference to compare the complete entry point with your saved file.

• In the VS Code Explorer sidebar, click the New File control.
• Enter gateway_stack.py inside the secure-agent-gateway folder.
• Start the stack with its imports and trusted-caller validation by adding this code:

from aws_cdk import CfnOutput, Duration, RemovalPolicy, Stack
from aws_cdk import aws_apigateway as apigateway
from aws_cdk import aws_dynamodb as dynamodb
from aws_cdk import aws_iam as iam
from aws_cdk import aws_lambda as lambda_
from aws_cdk import aws_logs as logs
from constructs import Construct


class SecureAgentGatewayStack(Stack):
    def __init__(
        self,
        scope: Construct,
        construct_id: str,
        *,
        trusted_caller_arn: str,
        **kwargs: object,
    ) -> None:
        super().__init__(scope, construct_id, **kwargs)

        if ":sts::" not in trusted_caller_arn or ":assumed-role/" not in trusted_caller_arn:
            raise ValueError("TRUSTED_CALLER_ARN must be an STS assumed-role ARN")

Why validate the caller ARN?

The generated invoker role trusts the temporary assumed-role session established during preflight. The constructor rejects a value that does not match the expected STS session shape.

The stack never imports or modifies SecureAgentGatewayInvokerRole. The new roles receive CloudFormation-generated physical names.

• Save gateway_stack.py.
• Confirm that the initial stack can synthesize by running this command:

npx aws-cdk@2.1148.0 synth SecureAgentGatewayStack

What does synthesis prove?

This command loads app.py and validates the initial construct tree. It produces a CloudFormation template without deploying resources.

You should see synthesized CloudFormation output for SecureAgentGatewayStack. This proves that the entry point can load the stack with your trusted caller ARN.

Synthesis stopped before producing a template?

Confirm that the .venv from the previous step is active. Check that TRUSTED_CALLER_ARN still contains the temporary assumed-role ARN verified by preflight.py.

Still stuck? Help me diagnose the failed initial synthesis.

The gateway needs durable tenant-prefixed records plus an audit trail. Amazon DynamoDB provides the transactional store. Amazon CloudWatch Logs receives the structured runtime logs.

• In gateway_stack.py, place the resource definitions below the trusted-caller validation:

        table = dynamodb.Table(
            self,
            "GatewayTable",
            partition_key=dynamodb.Attribute(
                name="pk",
                type=dynamodb.AttributeType.STRING,
            ),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.DESTROY,
        )
        log_group = logs.LogGroup(
            self,
            "GatewayLogs",
            retention=logs.RetentionDays.ONE_DAY,
            removal_policy=RemovalPolicy.DESTROY,
        )

What do these resources enforce?

• The table uses the pk string key expected by DynamoStore.
• On-demand billing ties table charges to actual requests.
• The destroy policies let the project remove its table and dedicated log group with the application stack.

• Save gateway_stack.py.
• Confirm that the synthesized template now includes the table and log group by running:

npx aws-cdk@2.1148.0 synth SecureAgentGatewayStack

What changed in the template?

The synthesized template now contains the DynamoDB table and one-day log group. No AWS resources are created during this check.

You should see resource definitions corresponding to GatewayTable and GatewayLogs in the synthesized output.

Least privilege begins with separate identities for invocation and execution. The invoker can call one API method. The Lambda execution role can write only the required logs and table items.

• In gateway_stack.py, place the generated role definitions below the log group:

        invoker_role = iam.Role(
            self,
            "AgentInvokerRole",
            assumed_by=iam.ArnPrincipal(trusted_caller_arn),
            description="Generated least-privilege role for the simulated agent",
            max_session_duration=Duration.hours(1),
        )
        execution_role = iam.Role(
            self,
            "GatewayExecutionRole",
            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
            description="Generated least-privilege Lambda execution role",
        )

Why omit fixed role names?

Neither role sets role_name. CloudFormation therefore assigns unique physical names without colliding with the legacy role checked during preflight.

The invoker trust policy accepts only trusted_caller_arn. Lambda receives its own service-assumable execution role.

• Place the exact logging and DynamoDB allowlists below the role definitions:

        execution_role.add_to_policy(
            iam.PolicyStatement(
                actions=["logs:CreateLogStream", "logs:PutLogEvents"],
                resources=[f"{log_group.log_group_arn}:*"],
            )
        )
        execution_role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "dynamodb:GetItem",
                    "dynamodb:PutItem",
                    "dynamodb:TransactWriteItems",
                ],
                resources=[table.table_arn],
            )
        )

How narrow are these policies?

The logging statement grants only stream creation and event writes for the dedicated log group. The table statement grants only the three operations used by DynamoStore.

The deployment verifier later compares the deployed actions with these exact sets. Any extra or missing action fails the check.

• Save gateway_stack.py.
• Confirm that the generated role resources and inline policies synthesize by running:

npx aws-cdk@2.1148.0 synth SecureAgentGatewayStack

What does this synthesis add?

The template now includes two generated IAM roles plus the execution role's resource-scoped policies. The legacy fixed-name role remains outside the template.

The runtime layer packages the tested gateway code into an AWS Lambda function. Server-owned environment values bind the verified role to tenant-blue and the two approved actions.

• Place the Lambda function below the execution role policies:

        gateway_function = lambda_.Function(
            self,
            "GatewayFunction",
            runtime=lambda_.Runtime.PYTHON_3_12,
            handler="gateway.handler.lambda_handler",
            code=lambda_.Code.from_asset(
                ".",
                exclude=[
                    ".git",
                    ".venv",
                    "cdk.out",
                    "tests",
                    "__pycache__",
                    "outputs.json",
                    "README.md",
                    "bootstrap-template.yaml",
                ],
            ),
            role=execution_role,
            environment={
                "TABLE_NAME": table.table_name,
                "TRUSTED_ROLE_ARN": invoker_role.role_arn,
                "TENANT_ID": "tenant-blue",
                "ALLOWED_ACTIONS": "records:read,records:create",
            },
            memory_size=128,
            timeout=Duration.seconds(5),
            log_group=log_group,
        )

What enters the Lambda asset?

The asset includes the deployable gateway package. Local tests, virtual-environment files, generated outputs, reports, and bootstrap review material stay outside the Lambda bundle.

The environment contains only server-owned configuration. Request bodies cannot replace the trusted role, tenant, or action set.

• Place the REST API, protected method, invocation grant, and outputs below the Lambda function:

        api = apigateway.RestApi(
            self,
            "GatewayApi",
            rest_api_name="secure-agent-execution-gateway",
            endpoint_types=[apigateway.EndpointType.REGIONAL],
        )
        execute_method = api.root.add_resource("execute").add_method(
            "POST",
            apigateway.LambdaIntegration(gateway_function),
            authorization_type=apigateway.AuthorizationType.IAM,
        )
        execute_method.grant_execute(invoker_role)

        CfnOutput(self, "ApiUrl", value=api.url_for_path("/execute"))
        CfnOutput(self, "InvokerRoleArn", value=invoker_role.role_arn)
        CfnOutput(self, "InvokerRoleName", value=invoker_role.role_name)
        CfnOutput(self, "ExecutionRoleName", value=execution_role.role_name)
        CfnOutput(self, "TableName", value=table.table_name)
        CfnOutput(self, "LogGroupName", value=log_group.log_group_name)

Where is the outer security gate?

The regional REST API exposes only POST /execute through this stack. Its authorization type is AWS_IAM, so unsigned requests stop at API Gateway.

The method grant gives the generated invoker role only execute-api:Invoke. The outputs provide the identifiers used by deployment verification and signed smoke tests.

• Save gateway_stack.py.

✔️ Awesome, I've got everything!

Your stack now defines the complete deployment boundary. Confirm that gateway_stack.py is saved.

ⓧ I'd like to double check the full code

from aws_cdk import CfnOutput, Duration, RemovalPolicy, Stack
from aws_cdk import aws_apigateway as apigateway
from aws_cdk import aws_dynamodb as dynamodb
from aws_cdk import aws_iam as iam
from aws_cdk import aws_lambda as lambda_
from aws_cdk import aws_logs as logs
from constructs import Construct


class SecureAgentGatewayStack(Stack):
    def __init__(
        self,
        scope: Construct,
        construct_id: str,
        *,
        trusted_caller_arn: str,
        **kwargs: object,
    ) -> None:
        super().__init__(scope, construct_id, **kwargs)

        if ":sts::" not in trusted_caller_arn or ":assumed-role/" not in trusted_caller_arn:
            raise ValueError("TRUSTED_CALLER_ARN must be an STS assumed-role ARN")

        table = dynamodb.Table(
            self,
            "GatewayTable",
            partition_key=dynamodb.Attribute(
                name="pk",
                type=dynamodb.AttributeType.STRING,
            ),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.DESTROY,
        )
        log_group = logs.LogGroup(
            self,
            "GatewayLogs",
            retention=logs.RetentionDays.ONE_DAY,
            removal_policy=RemovalPolicy.DESTROY,
        )

        invoker_role = iam.Role(
            self,
            "AgentInvokerRole",
            assumed_by=iam.ArnPrincipal(trusted_caller_arn),
            description="Generated least-privilege role for the simulated agent",
            max_session_duration=Duration.hours(1),
        )
        execution_role = iam.Role(
            self,
            "GatewayExecutionRole",
            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
            description="Generated least-privilege Lambda execution role",
        )
        execution_role.add_to_policy(
            iam.PolicyStatement(
                actions=["logs:CreateLogStream", "logs:PutLogEvents"],
                resources=[f"{log_group.log_group_arn}:*"],
            )
        )
        execution_role.add_to_policy(
            iam.PolicyStatement(
                actions=[
                    "dynamodb:GetItem",
                    "dynamodb:PutItem",
                    "dynamodb:TransactWriteItems",
                ],
                resources=[table.table_arn],
            )
        )

        gateway_function = lambda_.Function(
            self,
            "GatewayFunction",
            runtime=lambda_.Runtime.PYTHON_3_12,
            handler="gateway.handler.lambda_handler",
            code=lambda_.Code.from_asset(
                ".",
                exclude=[
                    ".git",
                    ".venv",
                    "cdk.out",
                    "tests",
                    "__pycache__",
                    "outputs.json",
                    "README.md",
                    "bootstrap-template.yaml",
                ],
            ),
            role=execution_role,
            environment={
                "TABLE_NAME": table.table_name,
                "TRUSTED_ROLE_ARN": invoker_role.role_arn,
                "TENANT_ID": "tenant-blue",
                "ALLOWED_ACTIONS": "records:read,records:create",
            },
            memory_size=128,
            timeout=Duration.seconds(5),
            log_group=log_group,
        )

        api = apigateway.RestApi(
            self,
            "GatewayApi",
            rest_api_name="secure-agent-execution-gateway",
            endpoint_types=[apigateway.EndpointType.REGIONAL],
        )
        execute_method = api.root.add_resource("execute").add_method(
            "POST",
            apigateway.LambdaIntegration(gateway_function),
            authorization_type=apigateway.AuthorizationType.IAM,
        )
        execute_method.grant_execute(invoker_role)

        CfnOutput(self, "ApiUrl", value=api.url_for_path("/execute"))
        CfnOutput(self, "InvokerRoleArn", value=invoker_role.role_arn)
        CfnOutput(self, "InvokerRoleName", value=invoker_role.role_name)
        CfnOutput(self, "ExecutionRoleName", value=execution_role.role_name)
        CfnOutput(self, "TableName", value=table.table_name)
        CfnOutput(self, "LogGroupName", value=log_group.log_group_name)

Use this complete reference to compare every resource, permission, exclusion, environment value, and output.

The final template should reveal every application resource and IAM change before deployment. The diff gives you a second view focused on changes against the selected AWS environment.

• Synthesize the complete stack and inspect its diff by running these commands:

npx aws-cdk@2.1148.0 synth SecureAgentGatewayStack
npx aws-cdk@2.1148.0 diff SecureAgentGatewayStack --profile "$AWS_PROFILE"

What should you inspect?

• Confirm that the two application roles use generated physical names.
• Confirm that the API method uses AWS_IAM authorization.
• Confirm that the legacy SecureAgentGatewayInvokerRole is absent from the planned resources.
• Confirm that the policies contain only the expected invocation, logging, and DynamoDB actions.

You should see a complete template plus the proposed application changes. No application resources have been deployed at this point.

Diff cannot inspect the selected environment?

Confirm that AWS_PROFILE still identifies the authenticated project profile. Rerun the preflight from the previous step if the temporary session has expired.

Still stuck? Help me troubleshoot CDK synthesis or diff.

Prepare and execute the reviewed change set

A local template cannot prove that your principal can assume the bootstrap deployment role or publish assets. A prepared CloudFormation change set tests that path while leaving the application resources unexecuted.

Preparation may publish the Lambda asset and create the change-set record. It does not execute the proposed application changes.

• Prepare AgentGatewayPreflight without executing it by running:

npx aws-cdk@2.1148.0 deploy SecureAgentGatewayStack \
  --profile "$AWS_PROFILE" \
  --method prepare-change-set \
  --change-set-name AgentGatewayPreflight \
  --require-approval broadening

What does preparation test?

CDK tests bootstrap-role assumption, asset publication, and CloudFormation change-set creation. Security-sensitive IAM changes remain available for review before execution.

Preparation returned an access denial?

Stop AWS writes at this point. Keep the synthesized application template plus bootstrap-template.yaml for administrator review.

Request delegated access to the existing bootstrap deployment path. Do not switch to root credentials, static access keys, imported roles, or broader application policies.

Need help preparing the review package? Help me explain the failed deployment path safely.

Before you inspect the result, do you expect the change set to be executable or already executed?

• Inspect the prepared change set by running:

aws cloudformation describe-change-set \
  --stack-name SecureAgentGatewayStack \
  --change-set-name AgentGatewayPreflight \
  --profile "$AWS_PROFILE" \
  --query '{Status:Status,ExecutionStatus:ExecutionStatus,Changes:Changes[*].ResourceChange.{Action:Action,LogicalId:LogicalResourceId,Type:ResourceType,Replacement:Replacement}}'

How do you read this result?

The status should be CREATE_COMPLETE. The execution status should be AVAILABLE.

The changes list is your final review surface for resource types, actions, and replacements. Application resources remain unexecuted until the next command.

You should see CREATE_COMPLETE with AVAILABLE. That result proves the practical deployment path can prepare the reviewed stack.

Executing the change set creates usage-based AWS resources. Review the proposed changes first. The project removes them in the cleanup section.

• Execute the reviewed change set and capture its outputs by running:

npx aws-cdk@2.1148.0 deploy SecureAgentGatewayStack \
  --profile "$AWS_PROFILE" \
  --method execute-change-set \
  --change-set-name AgentGatewayPreflight \
  --outputs-file outputs.json

What does execution create?

CloudFormation creates the regional REST API, Lambda function, DynamoDB table, dedicated log group, generated invoker role, and generated execution role. CDK writes the resulting identifiers to outputs.json.

The deployment can take a few minutes while CloudFormation creates the resources. A quiet terminal during this period does not mean the process has stopped.

The reviewed gateway is now live. outputs.json should contain ApiUrl, InvokerRoleArn, InvokerRoleName, ExecutionRoleName, TableName, and LogGroupName.

Change-set execution did not complete?

Keep the reviewed policies unchanged. Inspect the CloudFormation event associated with the failed resource to identify the missing deployment permission.

Still stuck? Help me diagnose the failed change-set execution.

Verify IAM and run signed smoke tests

A completed stack proves resource creation. It does not prove that the deployed roles contain only the intended actions. The verification script reads the resulting inline policies and rejects any action-set drift.

• In the VS Code Explorer sidebar, click the New File control.
• Enter verify_deployment.py inside the secure-agent-gateway folder.
• Add the imports, stack name, exact allowlists, and policy-statement helper:

import json
from typing import Any

import boto3


STACK_NAME = "SecureAgentGatewayStack"
EXPECTED_INVOKER_ACTIONS = {"execute-api:Invoke"}
EXPECTED_EXECUTION_ACTIONS = {
    "logs:CreateLogStream",
    "logs:PutLogEvents",
    "dynamodb:GetItem",
    "dynamodb:PutItem",
    "dynamodb:TransactWriteItems",
}


def statements(document: dict[str, Any]) -> list[dict[str, Any]]:
    value = document.get("Statement", [])
    return value if isinstance(value, list) else [value]

What is the verification contract?

The expected sets encode the complete allowlists for both generated roles. Set equality catches missing actions and unexpected privilege broadening.

The helper normalizes an IAM policy document that contains either one statement or a list of statements.

• Save verify_deployment.py.

You should now see verify_deployment.py beside gateway_stack.py in the Explorer sidebar.

• Place the inline-policy reader below statements().

def actions_for_role(iam: Any, role_name: str) -> set[str]:
    actions: set[str] = set()
    response = iam.list_role_policies(RoleName=role_name)
    if response.get("IsTruncated"):
        raise RuntimeError("Unexpected IAM policy pagination")
    for policy_name in response["PolicyNames"]:
        document = iam.get_role_policy(RoleName=role_name, PolicyName=policy_name)["PolicyDocument"]
        for statement in statements(document):
            if statement.get("Effect") == "Allow":
                value = statement.get("Action", [])
                actions.update([value] if isinstance(value, str) else value)
    return actions

How are deployed actions collected?

The function loads every inline policy attached to a generated role. It collects only actions from allow statements into one set.

Unexpected policy pagination fails the verification instead of silently inspecting an incomplete result.

• Place the stack-status and IAM checks below actions_for_role().

def main() -> None:
    with open("outputs.json", encoding="utf-8") as file:
        deployed = json.load(file)[STACK_NAME]
    session = boto3.Session()
    stack = session.client("cloudformation").describe_stacks(StackName=STACK_NAME)["Stacks"][0]
    if stack["StackStatus"] not in {"CREATE_COMPLETE", "UPDATE_COMPLETE"}:
        raise SystemExit(f"Unexpected stack status: {stack['StackStatus']}")
    resources = session.client("cloudformation").list_stack_resources(StackName=STACK_NAME)["StackResourceSummaries"]
    incomplete = [resource for resource in resources if not resource["ResourceStatus"].endswith("_COMPLETE")]
    if incomplete:
        raise SystemExit(f"Incomplete stack resources: {incomplete}")
    iam = session.client("iam")
    invoker_actions = actions_for_role(iam, deployed["InvokerRoleName"])
    execution_actions = actions_for_role(iam, deployed["ExecutionRoleName"])
    if invoker_actions != EXPECTED_INVOKER_ACTIONS:
        raise SystemExit(f"Invoker IAM mismatch: {sorted(invoker_actions)}")
    if execution_actions != EXPECTED_EXECUTION_ACTIONS:
        raise SystemExit(f"Execution IAM mismatch: {sorted(execution_actions)}")
    print(json.dumps({
        "stack_status": stack["StackStatus"],
        "resource_count": len(resources),
        "invoker_actions": sorted(invoker_actions),
        "execution_actions": sorted(execution_actions),
        "least_privilege_verified": True,
    }, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

What causes this script to fail?

The script stops if the stack is incomplete, a resource lacks a completed status, or either role's actions differ from the allowlists. Successful output includes the deployed action sets and least_privilege_verified.

• Save verify_deployment.py.

✔️ Awesome, I've got everything!

Your deployment verifier now checks stack completion, resource completion, and exact IAM action sets.

ⓧ I'd like to double check the full code

import json
from typing import Any

import boto3


STACK_NAME = "SecureAgentGatewayStack"
EXPECTED_INVOKER_ACTIONS = {"execute-api:Invoke"}
EXPECTED_EXECUTION_ACTIONS = {
    "logs:CreateLogStream",
    "logs:PutLogEvents",
    "dynamodb:GetItem",
    "dynamodb:PutItem",
    "dynamodb:TransactWriteItems",
}


def statements(document: dict[str, Any]) -> list[dict[str, Any]]:
    value = document.get("Statement", [])
    return value if isinstance(value, list) else [value]


def actions_for_role(iam: Any, role_name: str) -> set[str]:
    actions: set[str] = set()
    response = iam.list_role_policies(RoleName=role_name)
    if response.get("IsTruncated"):
        raise RuntimeError("Unexpected IAM policy pagination")
    for policy_name in response["PolicyNames"]:
        document = iam.get_role_policy(RoleName=role_name, PolicyName=policy_name)["PolicyDocument"]
        for statement in statements(document):
            if statement.get("Effect") == "Allow":
                value = statement.get("Action", [])
                actions.update([value] if isinstance(value, str) else value)
    return actions


def main() -> None:
    with open("outputs.json", encoding="utf-8") as file:
        deployed = json.load(file)[STACK_NAME]
    session = boto3.Session()
    stack = session.client("cloudformation").describe_stacks(StackName=STACK_NAME)["Stacks"][0]
    if stack["StackStatus"] not in {"CREATE_COMPLETE", "UPDATE_COMPLETE"}:
        raise SystemExit(f"Unexpected stack status: {stack['StackStatus']}")
    resources = session.client("cloudformation").list_stack_resources(StackName=STACK_NAME)["StackResourceSummaries"]
    incomplete = [resource for resource in resources if not resource["ResourceStatus"].endswith("_COMPLETE")]
    if incomplete:
        raise SystemExit(f"Incomplete stack resources: {incomplete}")
    iam = session.client("iam")
    invoker_actions = actions_for_role(iam, deployed["InvokerRoleName"])
    execution_actions = actions_for_role(iam, deployed["ExecutionRoleName"])
    if invoker_actions != EXPECTED_INVOKER_ACTIONS:
        raise SystemExit(f"Invoker IAM mismatch: {sorted(invoker_actions)}")
    if execution_actions != EXPECTED_EXECUTION_ACTIONS:
        raise SystemExit(f"Execution IAM mismatch: {sorted(execution_actions)}")
    print(json.dumps({
        "stack_status": stack["StackStatus"],
        "resource_count": len(resources),
        "invoker_actions": sorted(invoker_actions),
        "execution_actions": sorted(execution_actions),
        "least_privilege_verified": True,
    }, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()

Use this full reference to compare the stack checks and both exact IAM allowlists.

• Verify the deployed stack and role policies by running:

python verify_deployment.py

What should the IAM evidence show?

The invoker action list should contain only execute-api:Invoke. The execution action list should contain the two log writes plus the three DynamoDB operations.

A value of true for least_privilege_verified means both deployed sets match exactly.

That is the least-privilege boundary confirmed: the stack is complete and both generated roles match their reviewed allowlists.

IAM verification reported a mismatch?

Compare the reported actions with EXPECTED_INVOKER_ACTIONS and EXPECTED_EXECUTION_ACTIONS. Do not add permissions merely to make the check pass.

Still stuck? Help me investigate the IAM drift.

The live client now needs to assume the generated invoker role and sign each permitted request with SigV4. An unsigned request follows the same endpoint but should be rejected before Lambda runs.

• In the VS Code Explorer sidebar, click the New File control.
• Enter invoke.py inside the secure-agent-gateway folder.
• Add the output loader and temporary invoker-role session:

import json
import urllib.error
import urllib.request

import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest


STACK_NAME = "SecureAgentGatewayStack"


def outputs() -> dict[str, str]:
    with open("outputs.json", encoding="utf-8") as file:
        return json.load(file)[STACK_NAME]


def agent_session(role_arn: str) -> boto3.Session:
    base = boto3.Session()
    response = base.client("sts").assume_role(RoleArn=role_arn, RoleSessionName="simulated-agent", DurationSeconds=900)
    credentials = response["Credentials"]
    return boto3.Session(
        aws_access_key_id=credentials["AccessKeyId"],
        aws_secret_access_key=credentials["SecretAccessKey"],
        aws_session_token=credentials["SessionToken"],
        region_name=base.region_name,
    )

How does the simulated agent get authority?

The client reads the generated invoker role ARN from outputs.json and requests a temporary simulated-agent session. It never creates or stores long-lived credentials.

The resulting Boto3 session holds the temporary credentials used for signing.

• Save invoke.py.

You should now see invoke.py beside the deployment verifier in the Explorer sidebar.

• Place the response decoder and signed request helper below agent_session().

def decode_response(body: bytes) -> dict:
    text = body.decode("utf-8")
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"raw": text}


def post(url: str, payload: dict, session: boto3.Session | None) -> tuple[int, dict]:
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
    headers = {"content-type": "application/json"}
    if session is not None:
        credentials = session.get_credentials()
        if credentials is None or session.region_name is None:
            raise RuntimeError("Temporary credentials and Region are required")
        request = AWSRequest(method="POST", url=url, data=body, headers=headers)
        SigV4Auth(credentials.get_frozen_credentials(), "execute-api", session.region_name).add_auth(request)
        prepared = request.prepare()
        body, headers = prepared.body, dict(prepared.headers)
    request = urllib.request.Request(url, data=body, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(request, timeout=10) as response:
            return response.status, decode_response(response.read())
    except urllib.error.HTTPError as error:
        return error.code, decode_response(error.read())

How is each request signed?

The helper serializes the request body consistently. When a session is present, SigV4Auth signs the request for the execute-api service in the selected Region.

Passing no session sends the same request unsigned. HTTP denials are returned as testable status codes instead of terminating the script.

• Place the five live cases and failure check below post().

def main() -> None:
    deployed = outputs()
    session = agent_session(deployed["InvokerRoleArn"])
    cases = [
        ("signed_create", {"action": "records:create", "record_id": "agent-demo-record", "data": {"status": "approved"}}, session, {201, 409}),
        ("signed_read", {"action": "records:read", "record_id": "agent-demo-record"}, session, {200}),
        ("admin_action", {"action": "admin:delete", "record_id": "agent-demo-record"}, session, {403}),
        ("forged_tenant", {"action": "records:read", "record_id": "agent-demo-record", "tenant_id": "tenant-red"}, session, {400}),
        ("unsigned", {"action": "records:read", "record_id": "agent-demo-record"}, None, {403}),
    ]
    failures = []
    for name, payload, credentials, expected in cases:
        status, response = post(deployed["ApiUrl"], payload, credentials)
        print(json.dumps({"case": name, "status": status, "response": response}, sort_keys=True))
        if status not in expected:
            failures.append(name)
    if failures:
        raise SystemExit(f"Smoke test failures: {failures}")
    print("Live smoke tests passed with zero unauthorized successful responses")


if __name__ == "__main__":
    main()

What security paths are exercised?

• The signed create accepts 201 or the repeat-safe 409 response.
• The signed read must return 200.
• The administrative action must return 403 from the gateway's default-deny policy.
• The forged tenant field must return 400 because authority fields are outside the request schema.
• The unsigned request must return 403 from API Gateway.

• Save invoke.py.

✔️ Awesome, I've got everything!

Your live client now covers the two permitted operations plus three independent denial paths.

ⓧ I'd like to double check the full code

import json
import urllib.error
import urllib.request

import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest


STACK_NAME = "SecureAgentGatewayStack"


def outputs() -> dict[str, str]:
    with open("outputs.json", encoding="utf-8") as file:
        return json.load(file)[STACK_NAME]


def agent_session(role_arn: str) -> boto3.Session:
    base = boto3.Session()
    response = base.client("sts").assume_role(RoleArn=role_arn, RoleSessionName="simulated-agent", DurationSeconds=900)
    credentials = response["Credentials"]
    return boto3.Session(
        aws_access_key_id=credentials["AccessKeyId"],
        aws_secret_access_key=credentials["SecretAccessKey"],
        aws_session_token=credentials["SessionToken"],
        region_name=base.region_name,
    )


def decode_response(body: bytes) -> dict:
    text = body.decode("utf-8")
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"raw": text}


def post(url: str, payload: dict, session: boto3.Session | None) -> tuple[int, dict]:
    body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
    headers = {"content-type": "application/json"}
    if session is not None:
        credentials = session.get_credentials()
        if credentials is None or session.region_name is None:
            raise RuntimeError("Temporary credentials and Region are required")
        request = AWSRequest(method="POST", url=url, data=body, headers=headers)
        SigV4Auth(credentials.get_frozen_credentials(), "execute-api", session.region_name).add_auth(request)
        prepared = request.prepare()
        body, headers = prepared.body, dict(prepared.headers)
    request = urllib.request.Request(url, data=body, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(request, timeout=10) as response:
            return response.status, decode_response(response.read())
    except urllib.error.HTTPError as error:
        return error.code, decode_response(error.read())


def main() -> None:
    deployed = outputs()
    session = agent_session(deployed["InvokerRoleArn"])
    cases = [
        ("signed_create", {"action": "records:create", "record_id": "agent-demo-record", "data": {"status": "approved"}}, session, {201, 409}),
        ("signed_read", {"action": "records:read", "record_id": "agent-demo-record"}, session, {200}),
        ("admin_action", {"action": "admin:delete", "record_id": "agent-demo-record"}, session, {403}),
        ("forged_tenant", {"action": "records:read", "record_id": "agent-demo-record", "tenant_id": "tenant-red"}, session, {400}),
        ("unsigned", {"action": "records:read", "record_id": "agent-demo-record"}, None, {403}),
    ]
    failures = []
    for name, payload, credentials, expected in cases:
        status, response = post(deployed["ApiUrl"], payload, credentials)
        print(json.dumps({"case": name, "status": status, "response": response}, sort_keys=True))
        if status not in expected:
            failures.append(name)
    if failures:
        raise SystemExit(f"Smoke test failures: {failures}")
    print("Live smoke tests passed with zero unauthorized successful responses")


if __name__ == "__main__":
    main()

Use this full reference to compare the temporary session, signer, five cases, and failure handling.

Before you run the live client, do you expect every denied case to reach Lambda?

• Run the signed and unsigned smoke tests with this command:

python invoke.py

What should the live results prove?

Signed create returns 201 or repeat-safe 409. Signed read returns 200.

The signed administrative request returns 403. The forged tenant request returns 400. The unsigned request returns 403 before Lambda runs.

The final line confirms that no unauthorized case produced a successful response.

You have now proved the complete boundary. API Gateway authenticates the signed generated-role session. Lambda binds that verified identity to one tenant and two actions. Every tested attempt to supply or borrow extra authority is denied.

A live smoke-test status did not match?

Use the printed case name to isolate the failing path. Confirm that outputs.json belongs to the current stack and that your SSO session can assume the generated invoker role.

If only the unsigned case succeeds, recheck the method's AWS_IAM authorization setting. If a forged field succeeds, compare the deployed Lambda asset with the tested gateway package.

Still stuck? Help me diagnose the live smoke-test failure.

Your AWS gateway is deployed with reviewed IAM, verified role allowlists, and live denial evidence. Next, you will benchmark the policy path and turn the measured results into a reproducible security report.

Benchmark and Generate Reproducible Evidence

Your deployed AWS gateway now accepts signed tenant-scoped operations. It also blocks unsigned requests plus forged authority claims.

A successful demonstration provides one moment of evidence. This step turns the security claims into measured results that another engineer can reproduce.

In this step, get ready to:

• Run the complete security suite against a fixed 100-request benchmark.
• Generate README.md from passing security metrics.
• Confirm the evidence meets every acceptance criterion.

Build the fixed security benchmark

The pytest suite proves individual controls. The benchmark repeats a fixed mix of allowed plus denied requests to measure correctness across 100 policy decisions.

The timing covers the local in-process policy path. It excludes API Gateway latency, network latency, Lambda startup time, plus every other live AWS runtime cost.

• Create benchmark.py inside the existing secure-agent-gateway folder by using the VS Code file sidebar.
• Build the deterministic workload by adding this first section:

import json
import statistics
import time
from typing import Any

from gateway.domain import GatewayService, MemoryStore


ROLE_ARN = "arn:aws:iam::123456789012:role/generated-agent-role"
SESSION_ARN = "arn:aws:sts::123456789012:assumed-role/generated-agent-role/benchmark-agent"


def event(payload: dict[str, Any], request_id: str) -> dict[str, Any]:
    return {"body": json.dumps(payload), "requestContext": {"requestId": request_id, "identity": {"userArn": SESSION_ARN}}}


def workload() -> list[tuple[dict[str, Any], set[int], bool, bool]]:
    items: list[tuple[dict[str, Any], set[int], bool, bool]] = []
    items += [({"action": "records:read", "record_id": "seed"}, {200}, False, False) for _ in range(40)]
    items += [({"action": "records:create", "record_id": f"created-{i}", "data": {"index": i}}, {201}, False, False) for i in range(20)]
    items += [({"action": "admin:delete", "record_id": f"admin-{i}"}, {403}, False, True) for i in range(10)]
    items += [({"action": "unknown:tool", "record_id": f"unknown-{i}"}, {403}, False, True) for i in range(10)]
    items += [({"action": "records:read", "record_id": "seed", "role": "admin"}, {400}, False, True) for _ in range(10)]
    items += [({"action": "records:read", "record_id": "secret", "tenant_id": "tenant-red"}, {400}, True, True) for _ in range(10)]
    return items

How Is the Workload Fixed?

• The event() helper gives each request a verified assumed-role session ARN.
• The allowed portion contains 40 reads plus 20 creates.
• The denial portion contains four groups of 10 requests.
• Each workload entry records its expected status plus its security category.

• Complete the benchmark runner by continuing directly below workload() with this section:

def run_benchmark() -> dict[str, float | int]:
    store = MemoryStore({("tenant-blue", "seed"): {"status": "visible"}, ("tenant-red", "secret"): {"status": "protected"}})
    gateway = GatewayService(store, ROLE_ARN)
    latencies = []
    correct = denied = tenant_violations = unauthorized_executions = 0
    for index, (payload, expected, cross_tenant, unauthorized) in enumerate(workload()):
        started = time.perf_counter_ns()
        response = gateway.handle(event(payload, f"benchmark-{index}"))
        latencies.append((time.perf_counter_ns() - started) / 1_000_000)
        status = int(response["statusCode"])
        correct += int(status in expected)
        denied += int(status >= 400)
        tenant_violations += int(cross_tenant and status < 400)
        unauthorized_executions += int(unauthorized and status < 400)
    ordered = sorted(latencies)
    return {
        "requests": 100,
        "authorization_correctness_percent": round(correct, 2),
        "denial_rate_percent": round(denied, 2),
        "median_latency_ms": round(statistics.median(ordered), 4),
        "p95_latency_ms": round(ordered[94], 4),
        "unauthorized_tool_executions": unauthorized_executions,
        "tenant_isolation_violations": tenant_violations,
    }


if __name__ == "__main__":
    print(json.dumps(run_benchmark(), indent=2, sort_keys=True))

What Does the Benchmark Measure?

• The MemoryStore starts with one permitted record plus one protected cross-tenant record.
• Every response status is compared with the expected status set.
• The sorted timing samples produce local median plus p95 policy latency.
• Separate counters expose any unauthorized execution or tenant-isolation violation.

• Save benchmark.py.

Before you run the suite, consider whether all 40 hostile requests should remain denied.

• Run the full security suite plus the fixed benchmark with these commands:

python -m pytest -q
python benchmark.py

What Should You See?

You should see seven passing security tests. The benchmark JSON should report 100 requests with 100.0 percent authorization correctness.

The denial rate should be 40.0 percent. Both violation counters should be 0 while the median plus p95 latency values reflect your machine.

Benchmark Failing or Reporting Violations?

• Confirm benchmark.py is inside the secure-agent-gateway folder beside gateway.
• Check that each workload group uses the exact status set shown above.
• Rerun the seven existing security tests before investigating benchmark metrics.
• Still stuck? Help me diagnose my security benchmark results.

✔️ Awesome, I've got everything!

Your saved benchmark now produces repeatable correctness plus local policy-latency evidence.

ⓧ I'd like to double check the full code

import json
import statistics
import time
from typing import Any

from gateway.domain import GatewayService, MemoryStore


ROLE_ARN = "arn:aws:iam::123456789012:role/generated-agent-role"
SESSION_ARN = "arn:aws:sts::123456789012:assumed-role/generated-agent-role/benchmark-agent"


def event(payload: dict[str, Any], request_id: str) -> dict[str, Any]:
    return {"body": json.dumps(payload), "requestContext": {"requestId": request_id, "identity": {"userArn": SESSION_ARN}}}


def workload() -> list[tuple[dict[str, Any], set[int], bool, bool]]:
    items: list[tuple[dict[str, Any], set[int], bool, bool]] = []
    items += [({"action": "records:read", "record_id": "seed"}, {200}, False, False) for _ in range(40)]
    items += [({"action": "records:create", "record_id": f"created-{i}", "data": {"index": i}}, {201}, False, False) for i in range(20)]
    items += [({"action": "admin:delete", "record_id": f"admin-{i}"}, {403}, False, True) for i in range(10)]
    items += [({"action": "unknown:tool", "record_id": f"unknown-{i}"}, {403}, False, True) for i in range(10)]
    items += [({"action": "records:read", "record_id": "seed", "role": "admin"}, {400}, False, True) for _ in range(10)]
    items += [({"action": "records:read", "record_id": "secret", "tenant_id": "tenant-red"}, {400}, True, True) for _ in range(10)]
    return items


def run_benchmark() -> dict[str, float | int]:
    store = MemoryStore({("tenant-blue", "seed"): {"status": "visible"}, ("tenant-red", "secret"): {"status": "protected"}})
    gateway = GatewayService(store, ROLE_ARN)
    latencies = []
    correct = denied = tenant_violations = unauthorized_executions = 0
    for index, (payload, expected, cross_tenant, unauthorized) in enumerate(workload()):
        started = time.perf_counter_ns()
        response = gateway.handle(event(payload, f"benchmark-{index}"))
        latencies.append((time.perf_counter_ns() - started) / 1_000_000)
        status = int(response["statusCode"])
        correct += int(status in expected)
        denied += int(status >= 400)
        tenant_violations += int(cross_tenant and status < 400)
        unauthorized_executions += int(unauthorized and status < 400)
    ordered = sorted(latencies)
    return {
        "requests": 100,
        "authorization_correctness_percent": round(correct, 2),
        "denial_rate_percent": round(denied, 2),
        "median_latency_ms": round(statistics.median(ordered), 4),
        "p95_latency_ms": round(ordered[94], 4),
        "unauthorized_tool_executions": unauthorized_executions,
        "tenant_isolation_violations": tenant_violations,
    }


if __name__ == "__main__":
    print(json.dumps(run_benchmark(), indent=2, sort_keys=True))

Generate the evidence report

Raw test output is easy to lose. The report generator reruns the checks before writing README.md, so the document only represents passing evidence.

The report also joins local measurements with the deployed outputs from outputs.json. This creates one reviewable record of the gateway's architecture plus security boundary.

• Create report.py inside the existing secure-agent-gateway folder by using the VS Code file sidebar.
• Paste the following four consecutive sections into report.py before saving:

import json
import subprocess
import sys
from pathlib import Path

from benchmark import run_benchmark


STACK_NAME = "SecureAgentGatewayStack"


def main() -> None:
    tests = subprocess.run([sys.executable, "-m", "pytest", "-q"], capture_output=True, text=True)
    if tests.returncode:
        print(tests.stdout)
        print(tests.stderr, file=sys.stderr)
        raise SystemExit(tests.returncode)
    metrics = run_benchmark()
    if metrics["authorization_correctness_percent"] != 100 or metrics["unauthorized_tool_executions"] != 0 or metrics["tenant_isolation_violations"] != 0:
        raise SystemExit("Security acceptance metrics failed")
    with open("outputs.json", encoding="utf-8") as file:
        deployed = json.load(file)[STACK_NAME]

How Does the Evidence Gate Work?

• The report invokes the complete local test suite through the active Python interpreter.
• A failing test stops report generation.
• The benchmark must report complete correctness.
• Both security violation counters must remain at zero.

    readme = f"""# Secure AI Agent Execution Gateway

A production-inspired AWS gateway that treats agent output and body identity claims as untrusted input.

## Architecture

```mermaid
flowchart TD
    A[Deterministic agent] --> B[Generated STS invoker role]
    B --> C[SigV4]
    C --> D[API Gateway AWS_IAM]
    D --> E[Lambda verified userArn]
    E --> F[Server-side principal binding]
    F --> G[Default-deny policy]
    G --> H[DynamoDB tools and audit]
    E --> I[CloudWatch JSON logs]
```

## Principal-to-tenant mapping

The stack passes its generated invoker role ARN to Lambda as trusted configuration. Lambda normalizes API Gateway's verified assumed-role session ARN to that IAM role ARN. Only an exact match receives `tenant-blue` with `records:read` and `records:create`. The request schema has no authority field.

## Deployed evidence

- API URL: `{deployed['ApiUrl']}`
- Generated invoker role: `{deployed['InvokerRoleArn']}`
- Generated execution role: `{deployed['ExecutionRoleName']}`
- DynamoDB table: `{deployed['TableName']}`
- CloudWatch log group: `{deployed['LogGroupName']}`

What Goes Into the Architecture Record?

• The Mermaid diagram traces the signed request through authentication plus authorization.
• The principal mapping records the exact tenant plus permission binding.
• The deployed evidence comes directly from the current stack outputs.

## Security evidence

```text
{tests.stdout.strip()}
```

```json
{json.dumps(metrics, indent=2, sort_keys=True)}
```

The latency values are local in-process policy measurements, not AWS network or Lambda cold-start claims.

## Threat model

| Threat | Control | Evidence |
| --- | --- | --- |
| Unsigned caller | API Gateway `AWS_IAM` | Live unsigned `403` |
| Forged identity | Exact normalized role ARN match | Deterministic forged-identity test |
| Privilege escalation | Authority fields rejected | Local and live denial |
| Cross-tenant access | Tenant derived from server binding | Zero violations |
| Unknown tool | Default deny | Explicit pytest case |
| Partial create without audit | DynamoDB transaction | Downstream failure test |

## Deployment controls

The project verifies the intended account, Region, assumed-role credentials, bootstrap version, legacy-role conflict state, non-executed change set, stack status, resource status, and exact role action sets before accepting the deployment.

## Cost

The under-$1 estimate is conditional and not guaranteed. Actual charges vary by Region, account eligibility, usage, logs, and bootstrap storage. Destroy the application stack promptly.

Why Qualify the Latency Result?

The benchmark executes GatewayService inside the local Python process. The qualification prevents the p95 value from being presented as an API Gateway, Lambda, or network measurement.

The threat model links each risk to one control plus one evidence source. This makes every security claim reviewable.

## Resume bullet

Built a SigV4-authenticated AWS agent execution gateway with Python 3.12 and AWS CDK v2, mapped verified role identity to tenant-scoped permissions, and achieved {metrics['authorization_correctness_percent']}% correctness across 100 deterministic requests with {metrics['unauthorized_tool_executions']} unauthorized executions, {metrics['tenant_isolation_violations']} tenant violations, and {metrics['p95_latency_ms']} ms local p95 policy latency.

## LinkedIn-ready summary

I built a secure AWS execution gateway that never trusts an AI agent's claimed role or tenant. API Gateway validates SigV4, Lambda maps the verified role ARN to server-owned permissions, and the default-deny policy exposes only two tenant-scoped DynamoDB operations. Automated tests, live smoke tests, IAM inspection, and a 100-request benchmark produced reproducible evidence of zero unauthorized executions and zero cross-tenant violations.
"""
    Path("README.md").write_text(readme, encoding="utf-8")
    print("README.md generated from passing security evidence")


if __name__ == "__main__":
    main()

How Does the Report Finish?

• The resume bullet inserts the measured correctness plus p95 latency.
• The professional summary connects the controls to their evidence.
• The final write creates README.md only after every acceptance gate passes.

• Save report.py.
• Confirm report.py now appears beside benchmark.py in the VS Code file sidebar.

✔️ Awesome, I've got everything!

Your report generator now gates documentation on passing tests plus zero measured security violations.

ⓧ I'd like to double check the full code

import json
import subprocess
import sys
from pathlib import Path

from benchmark import run_benchmark


STACK_NAME = "SecureAgentGatewayStack"


def main() -> None:
    tests = subprocess.run([sys.executable, "-m", "pytest", "-q"], capture_output=True, text=True)
    if tests.returncode:
        print(tests.stdout)
        print(tests.stderr, file=sys.stderr)
        raise SystemExit(tests.returncode)
    metrics = run_benchmark()
    if metrics["authorization_correctness_percent"] != 100 or metrics["unauthorized_tool_executions"] != 0 or metrics["tenant_isolation_violations"] != 0:
        raise SystemExit("Security acceptance metrics failed")
    with open("outputs.json", encoding="utf-8") as file:
        deployed = json.load(file)[STACK_NAME]
    readme = f"""# Secure AI Agent Execution Gateway

A production-inspired AWS gateway that treats agent output and body identity claims as untrusted input.

## Architecture

```mermaid
flowchart TD
    A[Deterministic agent] --> B[Generated STS invoker role]
    B --> C[SigV4]
    C --> D[API Gateway AWS_IAM]
    D --> E[Lambda verified userArn]
    E --> F[Server-side principal binding]
    F --> G[Default-deny policy]
    G --> H[DynamoDB tools and audit]
    E --> I[CloudWatch JSON logs]
```

## Principal-to-tenant mapping

The stack passes its generated invoker role ARN to Lambda as trusted configuration. Lambda normalizes API Gateway's verified assumed-role session ARN to that IAM role ARN. Only an exact match receives `tenant-blue` with `records:read` and `records:create`. The request schema has no authority field.

## Deployed evidence

- API URL: `{deployed['ApiUrl']}`
- Generated invoker role: `{deployed['InvokerRoleArn']}`
- Generated execution role: `{deployed['ExecutionRoleName']}`
- DynamoDB table: `{deployed['TableName']}`
- CloudWatch log group: `{deployed['LogGroupName']}`

## Security evidence

```text
{tests.stdout.strip()}
```

```json
{json.dumps(metrics, indent=2, sort_keys=True)}
```

The latency values are local in-process policy measurements, not AWS network or Lambda cold-start claims.

## Threat model

| Threat | Control | Evidence |
| --- | --- | --- |
| Unsigned caller | API Gateway `AWS_IAM` | Live unsigned `403` |
| Forged identity | Exact normalized role ARN match | Deterministic forged-identity test |
| Privilege escalation | Authority fields rejected | Local and live denial |
| Cross-tenant access | Tenant derived from server binding | Zero violations |
| Unknown tool | Default deny | Explicit pytest case |
| Partial create without audit | DynamoDB transaction | Downstream failure test |

## Deployment controls

The project verifies the intended account, Region, assumed-role credentials, bootstrap version, legacy-role conflict state, non-executed change set, stack status, resource status, and exact role action sets before accepting the deployment.

## Cost

The under-$1 estimate is conditional and not guaranteed. Actual charges vary by Region, account eligibility, usage, logs, and bootstrap storage. Destroy the application stack promptly.

## Resume bullet

Built a SigV4-authenticated AWS agent execution gateway with Python 3.12 and AWS CDK v2, mapped verified role identity to tenant-scoped permissions, and achieved {metrics['authorization_correctness_percent']}% correctness across 100 deterministic requests with {metrics['unauthorized_tool_executions']} unauthorized executions, {metrics['tenant_isolation_violations']} tenant violations, and {metrics['p95_latency_ms']} ms local p95 policy latency.

## LinkedIn-ready summary

I built a secure AWS execution gateway that never trusts an AI agent's claimed role or tenant. API Gateway validates SigV4, Lambda maps the verified role ARN to server-owned permissions, and the default-deny policy exposes only two tenant-scoped DynamoDB operations. Automated tests, live smoke tests, IAM inspection, and a 100-request benchmark produced reproducible evidence of zero unauthorized executions and zero cross-tenant violations.
"""
    Path("README.md").write_text(readme, encoding="utf-8")
    print("README.md generated from passing security evidence")


if __name__ == "__main__":
    main()

Record the accepted security evidence

The final report run is the acceptance gate. A failed test or nonzero violation prevents the evidence document from being written.

Before you run the report, consider whether the current tests plus benchmark satisfy every gate.

• Generate the evidence-backed README by running this command:

python report.py

What Should You See?

You should see README.md generated from passing security evidence.

This confirms the tests passed. It also confirms 100 percent correctness plus zero unauthorized executions plus zero tenant-isolation violations.

Report Did Not Generate?

• Read the test output printed by report.py to find the failing security case.
• Check that outputs.json contains the SecureAgentGatewayStack deployment outputs.
• Run the benchmark separately if the report mentions failed acceptance metrics.
• Still stuck? Help me debug my evidence report gate.

• Select README.md in the existing VS Code file sidebar.
• Confirm the architecture diagram traces the generated role through SigV4 plus AWS_IAM.
• Locate the benchmark JSON under the security evidence section.
• Verify authorization_correctness_percent is 100.0.
• Verify unauthorized_tool_executions is 0.
• Verify tenant_isolation_violations is 0.
• Confirm the latency statement identifies the values as local in-process policy measurements.
• Confirm the threat model links each risk to a control plus its evidence.

That completes the evidence chain. Your gateway now has reproducible proof of identity binding, default deny, least-privilege deployment, authorization correctness, plus tenant isolation.

Secret mission

The Agent That Borrowed Someone Else's Authority

A forged request claims an administrative role inside tenant-red. You will expose the resulting data leak in a local fixture. You will then prove that the deployable gateway rejects the same payload.

Clean Up Your Resources

Clean Up Your Resources

A deployed AWS stack can continue generating usage-based charges while it exists. Decide whether to keep it for a short demonstration, pause API calls, or delete it now.

Cost warning

A short run with immediate teardown is designed to stay under $1. This estimate is conditional.

Amazon API Gateway, AWS Lambda, Amazon DynamoDB, and Amazon CloudWatch Logs use usage-based pricing. Retained bootstrap assets can also incur storage charges.

AWS Budgets can update every 8 to 12 hours. It does not cap spending in real time.

Resources you used:

• AWS CloudFormation application stack SecureAgentGatewayStack.
• IAM-protected Amazon API Gateway REST API.
• Python 3.12 AWS Lambda function.
• Two generated AWS Identity and Access Management roles for invocation and execution.
• Amazon DynamoDB table containing tenant-scoped records and audit entries.
• Amazon CloudWatch Logs log group containing structured gateway logs.
• Local secure-agent-gateway workspace containing the code and generated evidence.

Keep everything running

No action needed. Choose this if you are still demonstrating the signed gateway.

• Keep SecureAgentGatewayStack deployed.
• Keep outputs.json with your generated role names and API URL.
• Use temporary IAM Identity Center sessions for future signed requests.
• Set a date to delete the stack because usage-based charges can continue.

The shared CDKToolkit stack remains protected. Its bootstrap assets stay available for other CDK deployments.

Pause - I'll come back to this later

Stop invoking the endpoint to reduce request-driven usage. The deployed stack remains ready for your return.

• Stop sending requests to the deployed ApiUrl.
• Allow the temporary role session to expire.
• Keep outputs.json for the next signed smoke test.
• Keep the local secure-agent-gateway workspace for future changes.

Serverless resources have no project-level pause switch. Storage-related charges can continue while the stack remains deployed.

Delete - I don't want to use this again

Deleting the application stack is a deliberate reset. The stack template removes its managed API, function, roles, table, and log group.

• Keep outputs.json available until the cleanup checks finish.
• Destroy the application stack with the built-in cleanup checks by running:

npx aws-cdk@2.1148.0 destroy SecureAgentGatewayStack --force --profile "$AWS_PROFILE"
aws cloudformation describe-stacks \
  --stack-name SecureAgentGatewayStack \
  --profile "$AWS_PROFILE"
aws iam get-role \
  --role-name "$(python -c 'import json; print(json.load(open("outputs.json"))["SecureAgentGatewayStack"]["InvokerRoleName"])')" \
  --profile "$AWS_PROFILE"
aws sso logout

What do these checks prove?

The CloudFormation check returns ValidationError when SecureAgentGatewayStack no longer exists. That absence proves the stack-managed resources were removed.

The IAM check returns NoSuchEntity when the generated invoker role is gone. The pre-existing SecureAgentGatewayInvokerRole remains untouched.

The shared CDKToolkit stack remains protected. Its bootstrap storage can support other CDK projects in the same account and Region.

The local deletion is permanent. It removes README.md with the rest of the workspace.

macOS

• Return to the terminal from earlier.
• Confirm its prompt is inside secure-agent-gateway.
• Delete the local workspace by running:

cd ..
rm -rf secure-agent-gateway

What does this cleanup do?

The first command moves to the folder containing secure-agent-gateway. The second command permanently removes the workspace.

Your terminal returns to the containing folder. The secure-agent-gateway directory no longer exists.

Windows

• Return to the terminal from earlier.
• Confirm its prompt is inside secure-agent-gateway.
• Delete the local workspace by running:

cd ..
rmdir /s /q secure-agent-gateway

What does this cleanup do?

The first command moves to the folder containing secure-agent-gateway. The second command permanently removes the workspace.

Your terminal returns to the containing folder. The secure-agent-gateway directory no longer exists.

Nice Work!

Nice Work!

Excellent work! You built a secure AI agent execution gateway on AWS with AWS CDK.

You've learned how to:

• Complete a safe AWS deployment preflight before any write occurs. Inspect an AWS CloudFormation change set before executing it. Leave existing IAM roles untouched through generated role names.
• Enforce identity-bound authorization with Amazon API Gateway AWS_IAM. Require SigV4 before AWS Lambda receives a request. Bind the verified role ARN to a server-owned permission set for tenant-blue. Permit records:read plus records:create.
• Generate a GitHub-ready security report from passing evidence. Measure 100 local requests with 100 percent authorization correctness. Report the measured local p95 policy latency. Prove zero unauthorized tool executions. Prove zero tenant-isolation violations.
• Complete the optional Secret Mission by exploiting a confused deputy inside tests/test_confused_deputy.py. Prove GatewayService rejects the forged authority. Confirm legitimate tenant-blue access still succeeds. Keep the vulnerable fixture outside the Lambda asset.

Ready to quiz yourself?