CDK constructs

To be used with the AWS CDK for Python

Core

DNS/Edge services

CrossAccountDomainDelegation

Creates delegated domains

from ca_cdk_constructs import CrossAccountDomainDelegation, R53ParentZoneConfig

zone = PublicHostedZone(self, "Zone", zone_name="my-subdomain.acme.org")

# creates delegation records in for my-subdomain.acme.org in acme.org
CrossAccountDomainDelegation(
    subdomain_stack,
    "MySubdomainDnsDelegation",
    parent_zone_config=R53ParentZoneConfig(
        account_id="1234566789012",
        zone_name="acme.org",
        role_name="R53UpdateRole" # existing role in the parent zone account
    ),
    hosted_zone=zone
)

Edge services

WAF Rule Templates

TODO TODO TODO

Kubernetes / AWS EKS

ca_cdk_constructs includes a vendored cdk8s library that is compatible with the currently supported CA Kubernetes platform version.

You need to update the version of ca_cdk_constructs in your project if both of these are correct:

• your project deploys k8s resources using the vendored cdk8s library
• the CA platform was updated to a new version

To vendor the libraries for the K8s version currently supported by the CA Kubernetes platform:

• update the k8s / crd versions in cdk8s.yaml
• run just import-cdk8s

The library can be then used as follows:

import ca_cdk_constructs.eks.imports.k8s as k8s

k8s.KubeDeployment(.....)

EksClusterIntegration

Makes it possible to deploy to imported EKS clusters.

from ca_cdk_constructs.eks import EksClusterIntegration

# in an existing stack
eks_integration = EksClusterIntegration(self, "EksIntegration", vpc=vpc, cluster_name="mycluster")

# for imported clusters the kubectl role must be manually added to aws-auth
# The role ARN will also be available in the K8sAuthRoleArn output
eks_integration.role
# the EKS cluster
eks_integration.cluster

ExternalAwsSecretsChart

cdk8s Chart to deploy External Secrets referencing one or more AWS SecretsManager or ParameterStore secrets.

See the tests

Storage

ModifyDBClusterPassword

Modifies the password of an Aurora cluster

modify_cluster_password = ModifyDBClusterPassword(self, "ModifyClusterPassword", cluster_id=cluster_id, secret=db_secret)
modify_cluster_password.trigger_on_stack_create_update()
# access the udnerlaying lambda to e.g. add it to a state machine
modify_cluster_password.lambda_funct

AuroraFastClone

Clones an Aurora cluster.

from ca_cdk_constructs.storage.aurora_clone_refresh import AuroraCloneRefresh

source_cluster = DatabaseCluster(self, "AuroraCluster", ....) # or lookup one
vpc = source_cluster.vpc # or look it up

cluster_pg = CfnDBClusterParameterGroup(
    self,
    "DBClusterParameterGroup",
    description=f"Cluster parameter group for test clone",
    family=source_cluster.engine.parameter_group_family,
    parameters={"log_hostname": 1},
)
cluster_instance_pg = rds.CfnDBParameterGroup(
    self,
    "DBParameterGroup",
    description=f"DB parameter group for test clone instance",
    family=source_cluster.engine.parameter_group_family,
    parameters={"log_hostname": 1},
)

# periodically clone the source cluster
cloned_cluster = AuroraCloneRefresh(self, "TestClone",
                              source_cluster=source_cluster,
                              source_cluster_vpc=vpc,
                              source_cluster_master_username=username,
                              db_instance_class="db.t3.medium",
                              cluster_parameter_group=cluster_pg,
                              instance_parameter_group=cluster_instance_pg,
                                    tags={
                                        "Tag": "Value"
                                    },
                              clone_schedule=Schedule.cron(minute="0", hour="8"),
                              notifications_topic=topic)

# allow access to the clone from certain ranges
cloned_cluster.allow_from(ec2.Peer.ipv4(vpc.vpc_cidr_block))
# or
clone.cluster_sg.allow_....

# access the cloned cluster credentials
cloned_cluster.clone_secret # DatabaseSecret
# the clone SNS topic
cloned_cluster.notifications_topic # Topic

# the event rule
cloned_cluster.event_rule

Contributing

We welcome contributions.

This project uses just for running tasks and uv for project management.

Install and run just to see the list of available commands.

Tests

just pytest

To run tests against a particular version of python (e.g. 3.10), run uv venv --python 3.10 --clear.

Releases

This project uses semantic versioning and versions must be bumped in accordance with SemVer rules.

The Justfile contains a helper command for making releases, just draft-release. To use it:

1. Determine the release type (patch, minor, major)
2. Run just draft-release <patch | minor | major>
3. Approve the version bump if acceptable
4. Check the staged changes and approve if acceptable
5. Follow the link to the draft release and edit it to add details if requied
6. Publish the release