Chapter 07: Managing IAM Resources with CDK
Learn how to create and manage IAM users, roles, and policies with the AWS CDK Python version through a complete example.
Haiyue
38min
Chapter 7: Managing IAM Resources with CDK
Learning Objectives
- Master the basic usage of the AWS CDK Python version.
- Learn to create and manage IAM users, roles, and policies using CDK.
- Implement Infrastructure as Code for IAM resources.
- Master best practices and patterns for IAM in CDK.
- Implement deployment and management of CDK projects.
CDK IAM Architecture Diagram
🔄 正在渲染 Mermaid 图表...
7.1 CDK Basic Setup
7.1.1 CDK Project Initialization
# cdk_project/app.py
#!/usr/bin/env python3
import aws_cdk as cdk
from iam_stack import IamStack
app = cdk.App()
IamStack(app, "IamStack",
env=cdk.Environment(
account="123456789012", # Replace with your account ID
region="us-east-1"
)
)
app.synth()# cdk_project/iam_stack.py
from aws_cdk import (
Stack,
aws_iam as iam,
aws_s3 as s3,
Tags,
Duration
)
from constructs import Construct
import json
class IamStack(Stack):
"""
IAM Resource Management Stack
"""
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# Create IAM resources
self.create_iam_policies()
self.create_iam_roles()
self.create_iam_users()
self.create_iam_groups()
self.create_service_roles()
# Add tags
self.add_tags()
def create_iam_policies(self):
"""Create IAM policies"""
# 1. Create a developer permissions boundary policy
self.developer_boundary_policy = iam.ManagedPolicy(
self, "DeveloperPermissionsBoundary",
managed_policy_name="DeveloperPermissionsBoundary",
description="Permissions boundary policy to limit developer's maximum permissions",
statements=[
# Allowed services
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"s3:*",
"dynamodb:*",
"lambda:*",
"apigateway:*",
"cloudformation:*",
"logs:*",
"cloudwatch:*",
"ec2:Describe*",
"ec2:RunInstances",
"ec2:StartInstances",
"ec2:StopInstances"
],
resources=["*"]
),
# Deny high-risk operations
iam.PolicyStatement(
effect=iam.Effect.DENY,
actions=[
"iam:CreateRole",
"iam:DeleteRole",
"iam:AttachRolePolicy",
"iam:DetachRolePolicy",
"organizations:*",
"account:*",
"billing:*"
],
resources=["*"]
),
# Restrict instance types
iam.PolicyStatement(
effect=iam.Effect.DENY,
actions=["ec2:RunInstances"],
resources=["arn:aws:ec2:*:*:instance/*"],
conditions={
"ForAnyValue:StringNotEquals": {
"ec2:InstanceType": [
"t2.micro", "t2.small", "t2.medium",
"t3.micro", "t3.small", "t3.medium"
]
}
}
)
]
)
# 2. Create an S3 read-only access policy
self.s3_readonly_policy = iam.ManagedPolicy(
self, "S3ReadOnlyPolicy",
managed_policy_name="S3ReadOnlyAccess-Custom",
description="Custom S3 read-only access policy",
statements=[
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"s3:GetObject",
"s3:GetObjectVersion",
"s3:ListBucket",
"s3:GetBucketLocation"
],
resources=[
"arn:aws:s3:::company-data-*",
"arn:aws:s3:::company-data-*/*"
]
)
]
)
# 3. Create a Lambda execution policy
self.lambda_execution_policy = iam.ManagedPolicy(
self, "LambdaExecutionPolicy",
managed_policy_name="LambdaCustomExecutionPolicy",
description="Custom execution policy for Lambda functions",
statements=[
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
],
resources=["arn:aws:logs:*:*:*"]
),
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"s3:GetObject",
"s3:PutObject"
],
resources=["arn:aws:s3:::lambda-data-bucket/*"]
),
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"dynamodb:GetItem",
"dynamodb:PutItem",
"dynamodb:UpdateItem",
"dynamodb:DeleteItem"
],
resources=["arn:aws:dynamodb:*:*:table/lambda-*"]
)
]
)
def create_iam_roles(self):
"""Create IAM roles"""
# 1. Create an EC2 instance role
self.ec2_role = iam.Role(
self, "EC2InstanceRole",
role_name="EC2-WebServer-Role",
description="EC2 Web Server Role",
assumed_by=iam.ServicePrincipal("ec2.amazonaws.com"),
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("AmazonS3ReadOnlyAccess"),
iam.ManagedPolicy.from_aws_managed_policy_name("CloudWatchAgentServerPolicy")
],
inline_policies={
"CustomWebServerPolicy": iam.PolicyDocument(
statements=[
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"s3:PutObject",
"s3:PutObjectAcl"
],
resources=["arn:aws:s3:::web-server-logs/*"]
),
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"ssm:GetParameter",
"ssm:GetParameters",
"ssm:GetParametersByPath"
],
resources=["arn:aws:ssm:*:*:parameter/webserver/*"]
)
]
)
},
max_session_duration=Duration.hours(4)
)
# Create an instance profile
self.ec2_instance_profile = iam.InstanceProfile(
self, "EC2InstanceProfile",
instance_profile_name="EC2-WebServer-Profile",
role=self.ec2_role
)
# 2. Create a Lambda execution role
self.lambda_role = iam.Role(
self, "LambdaExecutionRole",
role_name="Lambda-DataProcessor-Role",
description="Role for Lambda data processing function",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
managed_policies=[
self.lambda_execution_policy
]
)
# 3. Create a cross-account access role
self.cross_account_role = iam.Role(
self, "CrossAccountRole",
role_name="CrossAccount-DataAccess-Role",
description="Cross-account data access role",
assumed_by=iam.AccountPrincipal("111122223333"), # Trusted account ID
external_ids=["unique-external-id-12345"],
managed_policies=[
self.s3_readonly_policy
],
inline_policies={
"CrossAccountDataPolicy": iam.PolicyDocument(
statements=[
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"dynamodb:GetItem",
"dynamodb:Query"
],
resources=["arn:aws:dynamodb:*:*:table/shared-data"]
)
]
)
}
)
# 4. Create a developer role with a permissions boundary
self.developer_role = iam.Role(
self, "DeveloperRole",
role_name="Developer-AssumeRole",
description="Developer assumed role (with permissions boundary)",
assumed_by=iam.AccountPrincipal(self.account),
permissions_boundary=self.developer_boundary_policy,
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("PowerUserAccess")
]
)
def create_iam_users(self):
"""Create IAM users"""
# 1. Create a developer user
self.developer_user = iam.User(
self, "DeveloperUser",
user_name="john.developer",
permissions_boundary=self.developer_boundary_policy,
managed_policies=[
self.s3_readonly_policy
]
)
# 2. Create an analyst user
self.analyst_user = iam.User(
self, "AnalystUser",
user_name="jane.analyst",
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("ReadOnlyAccess")
]
)
# Create access keys for the user (only when necessary)
self.developer_access_key = iam.AccessKey(
self, "DeveloperAccessKey",
user=self.developer_user
)
def create_iam_groups(self):
"""Create IAM groups"""
# 1. Create a developers group
self.developers_group = iam.Group(
self, "DevelopersGroup",
group_name="Developers",
managed_policies=[
self.s3_readonly_policy,
iam.ManagedPolicy.from_aws_managed_policy_name("AWSCodeCommitReadOnly")
]
)
# Add a user to the group
self.developers_group.add_user(self.developer_user)
# 2. Create an administrators group
self.admins_group = iam.Group(
self, "AdminsGroup",
group_name="Administrators",
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("AdministratorAccess")
]
)
# 3. Create a read-only group
self.readonly_group = iam.Group(
self, "ReadOnlyGroup",
group_name="ReadOnlyUsers",
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("ReadOnlyAccess")
]
)
self.readonly_group.add_user(self.analyst_user)
def create_service_roles(self):
"""Create service-specific roles"""
# 1. Create an ECS task role
self.ecs_task_role = iam.Role(
self, "ECSTaskRole",
role_name="ECS-TaskExecution-Role",
description="ECS Task Execution Role",
assumed_by=iam.ServicePrincipal("ecs-tasks.amazonaws.com"),
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AmazonECSTaskExecutionRolePolicy")
],
inline_policies={
"ECSTaskPolicy": iam.PolicyDocument(
statements=[
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"ssm:GetParameter",
"kms:Decrypt"
],
resources=[
"arn:aws:ssm:*:*:parameter/ecs/*",
"arn:aws:kms:*:*:key/*"
]
)
]
)
}
)
# 2. Create an API Gateway execution role
self.apigateway_role = iam.Role(
self, "APIGatewayRole",
role_name="APIGateway-Execution-Role",
description="API Gateway Execution Role",
assumed_by=iam.ServicePrincipal("apigateway.amazonaws.com"),
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AmazonAPIGatewayPushToCloudWatchLogs")
]
)
# 3. Create a CloudFormation execution role
self.cloudformation_role = iam.Role(
self, "CloudFormationRole",
role_name="CloudFormation-Execution-Role",
description="CloudFormation Stack Execution Role",
assumed_by=iam.ServicePrincipal("cloudformation.amazonaws.com"),
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("PowerUserAccess")
]
)
def add_tags(self):
"""Add tags"""
tags_config = {
"Environment": "Development",
"Project": "IAM-Management",
"ManagedBy": "CDK",
"Owner": "DevOps-Team"
}
for key, value in tags_config.items():
Tags.of(self).add(key, value)# cdk_project/requirements.txt
aws-cdk-lib==2.100.0
constructs>=10.0.0,<11.0.0# cdk_project/cdk.json
{
"app": "python3 app.py",
"watch": {
"include": [
"**"
],
"exclude": [
"README.md",
"cdk*.json",
"requirements*.txt",
"source.bat",
"**/__pycache__",
"**/.venv"
]
},
"context": {
"@aws-cdk/aws-lambda:recognizeLayerVersion": true,
"@aws-cdk/core:checkSecretUsage": true,
"@aws-cdk/core:target-partitions": [
"aws",
"aws-cn"
],
"@aws-cdk-containers/ecs-service-extensions:enableDefaultLogDriver": true,
"@aws-cdk/aws-ec2:uniqueImdsv2TemplateName": true,
"@aws-cdk/aws-ecs:arnFormatIncludesClusterName": true,
"@aws-cdk/core:validateSnapshotRemovalPolicy": true,
"@aws-cdk/aws-codepipeline:crossAccountKeyAliasStackSafeResourceName": true,
"@aws-cdk/aws-s3:createDefaultLoggingPolicy": true,
"@aws-cdk/aws-sns-subscriptions:restrictSqsDescryption": true
}
}7.2 Advanced CDK Patterns
7.2.1 Reusable IAM Constructs
# constructs/secure_role.py
from aws_cdk import (
aws_iam as iam,
Duration
)
from constructs import Construct
from typing import List, Dict, Optional
class SecureRole(Construct):
"""
A secure IAM role construct with best practices
"""
def __init__(
self,
scope: Construct,
construct_id: str,
role_name: str,
assumed_by: iam.IPrincipal,
description: str = "",
managed_policies: Optional[List[iam.IManagedPolicy]] = None,
inline_policies: Optional[Dict[str, iam.PolicyDocument]] = None,
permissions_boundary: Optional[iam.IManagedPolicy]] = None,
max_session_duration: Duration = Duration.hours(1),
external_ids: Optional[List[str]] = None,
conditions: Optional[Dict] = None,
require_mfa: bool = False,
**kwargs
):
super().__init__(scope, construct_id, **kwargs)
self.role_name = role_name
self.require_mfa = require_mfa
# Build trust policy conditions
trust_conditions = self._build_trust_conditions(conditions, require_mfa)
# Create the role
self.role = iam.Role(
self, "Role",
role_name=role_name,
description=description or f"Secure role: {role_name}",
assumed_by=assumed_by,
managed_policies=managed_policies or [],
inline_policies=inline_policies or {},
permissions_boundary=permissions_boundary,
max_session_duration=max_session_duration,
external_ids=external_ids
)
# Add conditions to the trust policy if MFA is required
if trust_conditions:
self._add_trust_policy_conditions(trust_conditions)
# Add security tags
self._add_security_tags()
def _build_trust_conditions(self, conditions: Optional[Dict], require_mfa: bool) -> Dict:
"""Build trust policy conditions"""
trust_conditions = conditions or {}
if require_mfa:
trust_conditions.update({
"Bool": {
"aws:MultiFactorAuthPresent": "true"
},
"NumericLessThan": {
"aws:MultiFactorAuthAge": "3600" # MFA is valid for 1 hour
}
})
return trust_conditions
def _add_trust_policy_conditions(self, conditions: Dict):
"""Add conditions to the trust policy"""
# Due to CDK limitations, we need to use low-level CloudFormation resources here
cfn_role = self.role.node.default_child
# Get the existing trust policy
assume_role_policy = cfn_role.assume_role_policy_document
# Add conditions to the first statement
if assume_role_policy and "Statement" in assume_role_policy:
for statement in assume_role_policy["Statement"]:
if "Condition" not in statement:
statement["Condition"] = {}
statement["Condition"].update(conditions)
def _add_security_tags(self):
"""Add security tags"""
security_tags = {
"SecurityLevel": "High" if self.require_mfa else "Medium",
"MFARequired": str(self.require_mfa),
"CreatedBy": "SecureRoleConstruct",
"LastReviewed": "2024-12-11"
}
for key, value in security_tags.items():
self.role.tags.set_tag(key, value)
def add_to_principals_policy(self, statement: iam.PolicyStatement):
"""Add a policy statement to the role"""
self.role.add_to_principals_policy(statement)
def grant_assume_role(self, grantee: iam.IPrincipal):
"""Grant another principal permission to assume this role"""
return self.role.grant_assume_role(grantee)
class ServiceRole(SecureRole):
"""
A service role construct, specifically for AWS services
"""
def __init__(
self,
scope: Construct,
construct_id: str,
service_name: str,
role_name: Optional[str] = None,
**kwargs
):
if not role_name:
role_name = f"{service_name.title()}ServiceRole"
super().__init__(
scope,
construct_id,
role_name=role_name,
assumed_by=iam.ServicePrincipal(f"{service_name}.amazonaws.com"),
description=f"Service role for {service_name}",
**kwargs
)
class CrossAccountRole(SecureRole):
"""
A cross-account access role construct
"""
def __init__(
self,
scope: Construct,
construct_id: str,
trusted_account_id: str,
external_id: str,
role_name: Optional[str] = None,
**kwargs
):
if not role_name:
role_name = f"CrossAccount-{trusted_account_id}-Role"
super().__init__(
scope,
construct_id,
role_name=role_name,
assumed_by=iam.AccountPrincipal(trusted_account_id),
external_ids=[external_id],
description=f"Cross-account role for account {trusted_account_id}",
require_mfa=True, # Cross-account roles require MFA by default
**kwargs
)7.2.2 Environment-Specific IAM Configuration
# stacks/environment_iam_stack.py
from aws_cdk import (
Stack,
aws_iam as iam,
Environment
)
from constructs import Construct
from constructs.secure_role import SecureRole, ServiceRole, CrossAccountRole
from typing import Dict, Any
class EnvironmentIamStack(Stack):
"""
Environment-specific IAM stack
"""
def __init__(
self,
scope: Construct,
construct_id: str,
environment_config: Dict[str, Any],
**kwargs
) -> None:
super().__init__(scope, construct_id, **kwargs)
self.environment_name = environment_config["name"]
self.is_production = environment_config.get("is_production", False)
# Create different resources based on the environment
self.create_environment_policies(environment_config)
self.create_environment_roles(environment_config)
self.create_monitoring_resources()
def create_environment_policies(self, config: Dict[str, Any]):
"""Create environment-specific policies"""
# Base permissions policy
base_permissions = config.get("base_permissions", [])
self.environment_policy = iam.ManagedPolicy(
self, "EnvironmentPolicy",
managed_policy_name=f"{self.environment_name}-BasePolicy",
description=f"Base policy for {self.environment_name} environment",
statements=[
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=base_permissions,
resources=["*"],
conditions=self._get_environment_conditions()
)
]
)
# Additional restrictions for the production environment
if self.is_production:
self.production_restrictions = iam.ManagedPolicy(
self, "ProductionRestrictions",
managed_policy_name=f"{self.environment_name}-Restrictions",
description="Production environment restrictions",
statements=[
iam.PolicyStatement(
effect=iam.Effect.DENY,
actions=[
"*:Delete*",
"*:Terminate*"
],
resources=["*"],
conditions={
"StringEquals": {
"aws:ResourceTag/Environment": "Production",
"aws:ResourceTag/Protected": "true"
}
}
)
]
)
def create_environment_roles(self, config: Dict[str, Any]):
"""Create environment-specific roles"""
# Application roles
self.app_roles = {}
for app_name, app_config in config.get("applications", {}).items():
role = ServiceRole(
self, f"{app_name}Role",
service_name=app_config.get("service", "lambda"),
role_name=f"{self.environment_name}-{app_name}-Role",
managed_policies=[self.environment_policy]
)
# Add application-specific permissions
if "permissions" in app_config:
for permission in app_config["permissions"]:
role.add_to_principals_policy(
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=permission["actions"],
resources=permission["resources"]
)
)
self.app_roles[app_name] = role
# Operations role
if config.get("create_ops_role", False):
ops_permissions = [
iam.ManagedPolicy.from_aws_managed_policy_name("ReadOnlyAccess")
]
if not self.is_production:
ops_permissions.append(
iam.ManagedPolicy.from_aws_managed_policy_name("PowerUserAccess")
)
self.ops_role = SecureRole(
self, "OpsRole",
role_name=f"{self.environment_name}-Operations-Role",
assumed_by=iam.AccountPrincipal(self.account),
managed_policies=ops_permissions,
require_mfa=True,
max_session_duration=Duration.hours(4 if self.is_production else 8)
)
def create_monitoring_resources(self):
"""Create monitoring-related resources"""
# CloudWatch role
self.cloudwatch_role = ServiceRole(
self, "CloudWatchRole",
service_name="events",
role_name=f"{self.environment_name}-CloudWatch-Role"
)
self.cloudwatch_role.add_to_principals_policy(
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
actions=[
"sns:Publish",
"lambda:InvokeFunction"
],
resources=[
f"arn:aws:sns:*:{self.account}:{self.environment_name}-*",
f"arn:aws:lambda:*:{self.account}:function:{self.environment_name}-*"
]
)
)
def _get_environment_conditions(self) -> Dict:
"""Get environment-specific conditions"""
conditions = {
"StringEquals": {
"aws:ResourceTag/Environment": self.environment_name
}
}
if self.is_production:
conditions["Bool"] = {
"aws:SecureTransport": "true"
}
return conditions
# Environment configurations
ENVIRONMENT_CONFIGS = {
"development": {
"name": "Development",
"is_production": False,
"base_permissions": [
"s3:*",
"dynamodb:*",
"lambda:*",
"logs:*",
"cloudwatch:*"
],
"applications": {
"web-api": {
"service": "lambda",
"permissions": [
{
"actions": ["dynamodb:*"],
"resources": ["arn:aws:dynamodb:*:*:table/dev-*"]
}
]
},
"data-processor": {
"service": "ecs-tasks",
"permissions": [
{
"actions": ["s3:*"],
"resources": ["arn:aws:s3:::dev-data-*/*"]
}
]
}
},
"create_ops_role": True
},
"production": {
"name": "Production",
"is_production": True,
"base_permissions": [
"s3:GetObject",
"s3:PutObject",
"dynamodb:GetItem",
"dynamodb:PutItem",
"dynamodb:UpdateItem",
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"applications": {
"web-api": {
"service": "lambda",
"permissions": [
{
"actions": [
"dynamodb:GetItem",
"dynamodb:PutItem",
"dynamodb:UpdateItem"
],
"resources": ["arn:aws:dynamodb:*:*:table/prod-*"]
}
]
}
},
"create_ops_role": True
}
}7.3 CDK Deployment and Management
7.3.1 Deployment Scripts and Best Practices
# scripts/deploy.py
#!/usr/bin/env python3
"""
CDK Deployment Script
"""
import subprocess
import sys
import json
import boto3
from typing import List, Dict
class CDKDeployManager:
"""CDK Deployment Manager"""
def __init__(self, app_path: str = "app.py"):
self.app_path = app_path
self.sts = boto3.client('sts')
def validate_prerequisites(self) -> bool:
"""Validate deployment prerequisites"""
print("🔍 Validating deployment prerequisites...")
# Check AWS credentials
try:
identity = self.sts.get_caller_identity()
print(f"✅ AWS Account: {identity['Account']}")
print(f"✅ IAM User/Role: {identity['Arn']}")
except Exception as e:
print(f"❌ AWS credential validation failed: {e}")
return False
# Check CDK version
try:
result = subprocess.run(['cdk', '--version'], capture_output=True, text=True)
if result.returncode == 0:
print(f"✅ CDK Version: {result.stdout.strip()}")
else:
print("❌ CDK not installed or unavailable")
return False
except Exception as e:
print(f"❌ CDK check failed: {e}")
return False
# Check Python dependencies
try:
import aws_cdk as cdk
print(f"✅ AWS CDK Library Version: {cdk.__version__}")
except ImportError:
print("❌ AWS CDK Python library not installed")
return False
return True
def bootstrap_environment(self, environments: List[str]) -> bool:
"""Bootstrap CDK environment"""
print("🚀 Bootstrapping CDK environment...")
for env in environments:
try:
cmd = ['cdk', 'bootstrap', env]
result = subprocess.run(cmd, check=True, capture_output=True, text=True)
print(f"✅ Environment {env} bootstrapped successfully")
except subprocess.CalledProcessError as e:
print(f"❌ Environment {env} bootstrapping failed: {e.stderr}")
return False
return True
def synthesize_stacks(self, stack_names: List[str] = None) -> bool:
"""Synthesize CloudFormation templates"""
print("🔧 Synthesizing CloudFormation templates...")
try:
cmd = ['cdk', 'synth']
if stack_names:
cmd.extend(stack_names)
result = subprocess.run(cmd, check=True, capture_output=True, text=True)
print("✅ Template synthesis successful")
# Display synthesized stacks
cmd_list = ['cdk', 'list']
result_list = subprocess.run(cmd_list, capture_output=True, text=True)
if result_list.returncode == 0:
stacks = result_list.stdout.strip().split('\n')
print(f"📋 Discovered stacks: {', '.join(stacks)}")
return True
except subprocess.CalledProcessError as e:
print(f"❌ Template synthesis failed: {e.stderr}")
return False
def deploy_stacks(
self,
stack_names: List[str] = None,
require_approval: bool = True,
rollback: bool = True,
parameters: Dict[str, str] = None
) -> bool:
"""Deploy stacks"""
print("🚀 Starting stack deployment...")
try:
cmd = ['cdk', 'deploy']
if stack_names:
cmd.extend(stack_names)
else:
cmd.append('--all')
if not require_approval:
cmd.append('--require-approval=never')
if not rollback:
cmd.append('--no-rollback')
if parameters:
for key, value in parameters.items():
cmd.extend(['--parameters', f'{key}={value}'])
# Add output format
cmd.extend(['--outputs-file', 'cdk-outputs.json'])
result = subprocess.run(cmd, check=True)
print("✅ Stack deployment successful")
# Read and display outputs
self.display_outputs()
return True
except subprocess.CalledProcessError as e:
print(f"❌ Stack deployment failed")
return False
def display_outputs(self):
"""Display stack outputs"""
try:
with open('cdk-outputs.json', 'r') as f:
outputs = json.load(f)
print("\n📄 Stack Outputs:")
for stack_name, stack_outputs in outputs.items():
print(f"\n{stack_name}:")
for key, value in stack_outputs.items():
print(f" {key}: {value}")
except FileNotFoundError:
print("📄 No stack outputs file found")
except Exception as e:
print(f"❌ Failed to read outputs: {e}")
def run_security_scan(self) -> bool:
"""Run a security scan"""
print("🔒 Running security scan...")
try:
# Use cdk-nag for security scanning (if installed)
cmd = ['cdk', 'synth', '--quiet']
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
print("✅ Security scan passed")
return True
else:
print(f"⚠️ Security scan found issues: {result.stderr}")
return False
except Exception as e:
print(f"⚠️ Security scan skipped: {e}")
return True
def cleanup_resources(self, stack_names: List[str] = None) -> bool:
"""Clean up resources"""
print("🧹 Cleaning up stack resources...")
try:
cmd = ['cdk', 'destroy']
if stack_names:
cmd.extend(stack_names)
else:
cmd.append('--all')
cmd.append('--force') # Skip confirmation
result = subprocess.run(cmd, check=True)
print("✅ Resource cleanup successful")
return True
except subprocess.CalledProcessError as e:
print(f"❌ Resource cleanup failed")
return False
def main():
"""Main deployment flow"""
import argparse
parser = argparse.ArgumentParser(description='CDK IAM Deployment Management')
parser.add_argument('--action', choices=['deploy', 'destroy', 'synth', 'bootstrap'],
default='deploy', help='Action to perform')
parser.add_argument('--stacks', nargs='+', help='Specify stack names')
parser.add_argument('--environment', help='Target environment')
parser.add_argument('--skip-approval', action='store_true', help='Skip deployment confirmation')
args = parser.parse_args()
# Create a deployment manager
deploy_manager = CDKDeployManager()
# Validate prerequisites
if not deploy_manager.validate_prerequisites():
print("❌ Prerequisite validation failed, aborting deployment")
sys.exit(1)
# Execute the corresponding flow based on the action
if args.action == 'bootstrap':
environments = [args.environment] if args.environment else ['aws://123456789012/us-east-1']
success = deploy_manager.bootstrap_environment(environments)
elif args.action == 'synth':
success = deploy_manager.synthesize_stacks(args.stacks)
elif args.action == 'deploy':
# First, synthesize
if not deploy_manager.synthesize_stacks(args.stacks):
sys.exit(1)
# Run security scan
deploy_manager.run_security_scan()
# Deploy
success = deploy_manager.deploy_stacks(
stack_names=args.stacks,
require_approval=not args.skip_approval
)
elif args.action == 'destroy':
success = deploy_manager.cleanup_resources(args.stacks)
if success:
print(f"✅ {args.action} operation completed")
sys.exit(0)
else:
print(f"❌ {args.action} operation failed")
sys.exit(1)
if __name__ == '__main__':
main()Summary
This chapter covered the complete method of managing IAM resources using AWS CDK:
- CDK Basics: Learned the CDK project structure and how to create IAM resources.
- Advanced Patterns: Created reusable IAM constructs and environment-specific configurations.
- Deployment Management: Implemented best practices for CDK deployment, validation, and management.
- Automation: Built a complete deployment pipeline and security scanning process.
With CDK, you can achieve:
- IAM management as Infrastructure as Code
- Reusable and modular IAM components
- Environment-specific permission configurations
- Automated deployment and validation processes
In the next chapter, we will learn about integrating IAM with other AWS services.