CDK Escape Hatches, Tokens, and Advanced Patterns
CDK Escape Hatches, Tokens, and Advanced Patterns
CDK’s abstraction layers are powerful until they’re not. When an L2 construct doesn’t expose a property you need, when you need to reference a value that doesn’t exist yet (deploy-time resolution), or when you need to conditionally create resources based on CloudFormation conditions — you need to go below the abstraction layer.
The Token System: Deploy-Time Values
CDK Tokens are placeholders for values that aren’t known until CloudFormation deploy time. When you write bucket.bucket_arn, that’s not a string — it’s a Token that resolves to {"Fn::GetAtt": ["Bucket", "Arn"]} in the CloudFormation template:
from aws_cdk import Token, Fn, CfnOutput, Lazy
import aws_cdk as cdk
class TokenDemoStack(Stack):
def __init__(self, scope, id, **kwargs):
super().__init__(scope, id, **kwargs)
bucket = aws_s3.Bucket(self, 'DataBucket')
# This looks like a string but it's a Token
arn = bucket.bucket_arn
print(f"ARN type: {type(arn)}") # <class 'str'> — but it's encoded
print(f"Is token: {Token.is_unresolved(arn)}") # True
# You CANNOT do string operations on tokens:
# BAD: arn.split(':')[4] — This splits the token encoding, not the ARN
# Instead, use Fn.select with Fn.split for deploy-time string ops:
account_from_arn = Fn.select(4, Fn.split(':', arn))
# Lazy values: Compute at synthesis time (not deploy time)
def compute_name():
return f"processed-{bucket.bucket_name}-output"
# Lazy.string defers evaluation until CDK synthesizes the template
lazy_name = Lazy.string(producer=lambda: compute_name())
# Token.as_string wraps any CloudFormation intrinsic function
conditional_value = Token.as_string(
Fn.condition_if('IsProd', 'production-value', 'dev-value')
)
# Cross-stack reference (automatically creates Export/Import)
CfnOutput(self, 'BucketArn', value=bucket.bucket_arn, export_name='data-bucket-arn')
class ConsumerStack(Stack):
def __init__(self, scope, id, **kwargs):
super().__init__(scope, id, **kwargs)
# Import from other stack
bucket_arn = Fn.import_value('data-bucket-arn')
imported_bucket = aws_s3.Bucket.from_bucket_arn(self, 'ImportedBucket', bucket_arn)
Escape Hatches: Reaching Below L2
When an L2 construct doesn’t expose a property, use node.default_child to access the underlying L1 (Cfn) resource:
class EscapeHatchDemo(Stack):
def __init__(self, scope, id, **kwargs):
super().__init__(scope, id, **kwargs)
# L2 construct: Lambda function
fn = lambda_.Function(self, 'Handler',
runtime=lambda_.Runtime.PYTHON_3_12,
handler='index.handler',
code=lambda_.Code.from_inline('def handler(e,c): return {}')
)
# Problem: L2 doesn't expose RecursiveLoop protection (new feature)
# Solution: Access the L1 resource and set it directly
cfn_function = fn.node.default_child # Type: CfnFunction
cfn_function.add_property_override('RecursiveLoop', 'Terminate')
# Problem: Need to add a DependsOn that L2 doesn't support
# Solution: Use cfn_options
cfn_function.cfn_options.depends_on = [some_other_resource.node.default_child]
# Problem: DynamoDB table needs a property not in L2
table = dynamodb.Table(self, 'Table',
partition_key=dynamodb.Attribute(name='pk', type=dynamodb.AttributeType.STRING)
)
cfn_table = table.node.default_child
# Add resource policy (not exposed in L2 as of CDK 2.x)
cfn_table.add_property_override('ResourcePolicy', {
'PolicyDocument': {
'Version': '2012-10-17',
'Statement': [{
'Effect': 'Deny',
'Principal': '*',
'Action': 'dynamodb:DeleteTable',
'Resource': '*',
'Condition': {
'StringNotEquals': {
'aws:PrincipalArn': 'arn:aws:iam::123456789012:role/Admin'
}
}
}]
}
})
# Remove a property that L2 sets by default
cfn_table.add_property_deletion_override('SSESpecification')
import software.amazon.awscdk.services.lambda.Function;
import software.amazon.awscdk.services.lambda.CfnFunction;
import software.amazon.awscdk.services.dynamodb.Table;
import software.amazon.awscdk.services.dynamodb.CfnTable;
import java.util.Map;
public class EscapeHatchDemo extends Stack {
public EscapeHatchDemo(Construct scope, String id, StackProps props) {
super(scope, id, props);
Function fn = Function.Builder.create(this, "Handler")
.runtime(software.amazon.awscdk.services.lambda.Runtime.PYTHON_3_12)
.handler("index.handler")
.code(software.amazon.awscdk.services.lambda.Code.fromInline(
"def handler(e,c): return {}"))
.build();
// Access L1 escape hatch
CfnFunction cfnFn = (CfnFunction) fn.getNode().getDefaultChild();
cfnFn.addPropertyOverride("RecursiveLoop", "Terminate");
// DynamoDB table with property override
Table table = Table.Builder.create(this, "Table")
.partitionKey(Attribute.builder().name("pk").type(AttributeType.STRING).build())
.build();
CfnTable cfnTable = (CfnTable) table.getNode().getDefaultChild();
cfnTable.addPropertyOverride("ResourcePolicy", Map.of(
"PolicyDocument", Map.of(
"Version", "2012-10-17",
"Statement", java.util.List.of(Map.of(
"Effect", "Deny",
"Principal", "*",
"Action", "dynamodb:DeleteTable",
"Resource", "*"
))
)
));
}
}
Context and Feature Flags
CDK context provides environment-specific configuration without code branches:
# cdk.json
# {
# "context": {
# "environments": {
# "dev": {"account": "111111111111", "region": "us-east-1", "instanceSize": "small"},
# "prod": {"account": "222222222222", "region": "us-east-1", "instanceSize": "xlarge"}
# }
# }
# }
class ConfigurableStack(Stack):
def __init__(self, scope, id, *, env_name: str, **kwargs):
super().__init__(scope, id, **kwargs)
# Read from CDK context
env_config = self.node.try_get_context('environments')[env_name]
instance_size = env_config['instanceSize']
# Feature flags: Enable/disable features per environment
enable_waf = env_name == 'prod'
enable_deletion_protection = env_name == 'prod'
table = dynamodb.Table(self, 'Table',
deletion_protection=enable_deletion_protection,
billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST if env_name == 'prod'
else dynamodb.BillingMode.PROVISIONED
)
if env_name != 'prod':
# Dev: Lower provisioned capacity for cost
table.auto_scale_write_capacity(min_capacity=1, max_capacity=10)
Dynamic References: Secrets at Deploy Time
Never hardcode secrets in CDK code. Use dynamic references to resolve at deploy time:
from aws_cdk import SecretValue, aws_secretsmanager as secretsmanager
class SecureStack(Stack):
def __init__(self, scope, id, **kwargs):
super().__init__(scope, id, **kwargs)
# Reference existing secret (resolved at deploy time, never in template plaintext)
db_secret = secretsmanager.Secret.from_secret_name_v2(
self, 'DbSecret', 'prod/database/credentials'
)
# Use in RDS (SecretValue is never logged or stored in CloudFormation)
rds.DatabaseInstance(self, 'Database',
credentials=rds.Credentials.from_secret(db_secret),
# ...
)
# Lambda environment variable from SSM Parameter Store
fn = lambda_.Function(self, 'Handler',
environment={
# Static value (appears in template — DON'T use for secrets)
'TABLE_NAME': 'orders',
# Dynamic reference (resolved at deploy time)
'API_KEY': secretsmanager.Secret.from_secret_name_v2(
self, 'ApiKey', 'prod/api-key'
).secret_value.unsafe_unwrap() # Only for env vars
}
)
# Better: Grant the Lambda permission to read the secret at runtime
secret = secretsmanager.Secret.from_secret_name_v2(
self, 'RuntimeSecret', 'prod/api-key'
)
secret.grant_read(fn)
fn.add_environment('SECRET_ARN', secret.secret_arn)
# Function reads secret at runtime via SDK, not embedded in env var
Stack Policies: Preventing Accidental Destruction
# Prevent CloudFormation from replacing or deleting critical resources
class ProtectedStack(Stack):
def __init__(self, scope, id, **kwargs):
super().__init__(scope, id, **kwargs)
# DynamoDB table with all safety nets
table = dynamodb.Table(self, 'CriticalTable',
table_name='production-orders',
partition_key=dynamodb.Attribute(name='pk', type=dynamodb.AttributeType.STRING),
removal_policy=RemovalPolicy.RETAIN, # CDK won't delete on stack destroy
deletion_protection=True, # AWS won't allow DeleteTable API
)
# CloudFormation stack policy (prevents resource replacement)
# Applied via AWS CLI after deployment:
# aws cloudformation set-stack-policy --stack-name Production \
# --stack-policy-body file://stack-policy.json
# stack-policy.json:
# {
# "Statement": [{
# "Effect": "Deny",
# "Action": ["Update:Replace", "Update:Delete"],
# "Principal": "*",
# "Resource": "LogicalResourceId/CriticalTable*"
# }, {
# "Effect": "Allow",
# "Action": "Update:*",
# "Principal": "*",
# "Resource": "*"
# }]
# }
Final wisdom on CDK: The infrastructure code IS the documentation. If your CDK code is well-structured with named constructs and clear composition, anyone can read it and understand your architecture. Treat it with the same care as application code: tests, code review, small deployable units, and clear ownership.