Spinning Up Infrastructure: Boto3 Clients vs AWS CDK

import boto3
import json
import time
from botocore.exceptions import ClientError

s3_client = boto3.client('s3', region_name='us-east-1')
dynamodb_client = boto3.client('dynamodb', region_name='us-east-1')
iam_client = boto3.client('iam')

bucket_name = "corp-data-lake-2026-prod"
table_name = "AgentAuditLogs"
role_name = "AgentExecutionRole"

try:
    # 1. Create S3 Bucket with Encryption
    print("Creating S3 Bucket...")
    s3_client.create_bucket(Bucket=bucket_name)
    s3_client.put_bucket_encryption(
        Bucket=bucket_name,
        ServerSideEncryptionConfiguration={
            'Rules': [{'ApplyServerSideEncryptionByDefault': {'SSEAlgorithm': 'AES256'}}]
        }
    )
    s3_client.put_public_access_block(
        Bucket=bucket_name,
        PublicAccessBlockConfiguration={
            'BlockPublicAcls': True, 'IgnorePublicAcls': True,
            'BlockPublicPolicy': True, 'RestrictPublicBuckets': True
        }
    )

    # 2. Create DynamoDB Table & Poll until Active
    print("Creating DynamoDB Table...")
    dynamodb_client.create_table(
        TableName=table_name,
        KeySchema=[{'AttributeName': 'SessionId', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'SessionId', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )
    waiter = dynamodb_client.get_waiter('table_exists')
    waiter.wait(TableName=table_name)

    # 3. Create IAM Role with Manual JSON Trust Policy
    print("Creating IAM Execution Role...")
    trust_policy = {
        "Version": "2012-10-17",
        "Statement": [{"Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "sts:AssumeRole"}]
    }
    role_resp = iam_client.create_role(
        RoleName=role_name,
        AssumeRolePolicyDocument=json.dumps(trust_policy)
    )

    # 4. Attach Least-Privilege Inline Policy (Hand-crafted JSON strings)
    policy_doc = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Action": ["s3:GetObject", "s3:PutObject"],
                "Resource": f"arn:aws:s3:::{bucket_name}/*"
            },
            {
                "Effect": "Allow",
                "Action": ["dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:UpdateItem"],
                "Resource": f"arn:aws:dynamodb:us-east-1:123456789012:table/{table_name}"
            }
        ]
    }
    iam_client.put_role_policy(
        RoleName=role_name,
        PolicyName="AgentAccessPolicy",
        PolicyDocument=json.dumps(policy_doc)
    )
    print("✅ Provisioning Complete. Note: No state file generated.")

except ClientError as e:
    print(f"❌ Provisioning Failed: {e}")
    # ROLLBACK CODE DOES NOT EXIST! S3 bucket and DynamoDB table remain active and billing.
    raise

from aws_cdk import (
    Stack,
    RemovalPolicy,
    aws_s3 as s3,
    aws_dynamodb as dynamodb,
    aws_iam as iam
)
from constructs import Construct

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

        # 1. Declarative S3 Bucket (Enforces Encryption & Public Access Block by default)
        bucket = s3.Bucket(
            self, "DataLakeBucket",
            encryption=s3.BucketEncryption.S3_MANAGED,
            block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
            enforce_ssl=True,
            removal_policy=RemovalPolicy.RETAIN
        )

        # 2. Serverless DynamoDB Table (Pay-per-request)
        table = dynamodb.Table(
            self, "AuditLogsTable",
            partition_key=dynamodb.Attribute(name="SessionId", type=dynamodb.AttributeType.STRING),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            point_in_time_recovery=True
        )

        # 3. IAM Role with Automatic Least-Privilege Grant Methods
        agent_role = iam.Role(
            self, "AgentRole",
            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
        )

        # 4. Semantic Permissions Synthesis (Zero manual JSON ARNs needed!)
        bucket.grant_read_write(agent_role)
        table.grant_read_write_data(agent_role)