Custom resource to automatically associate domain to App Runner service

cdk.py

from aws_cdk import (
    Duration,
    CustomResource,
    aws_apprunner_alpha as apprunner,
    aws_lambda as lambda_,
    aws_route53 as route53,
    aws_logs as logs,
    custom_resources,
)

app = apprunner.Service(...)
domain = "svc.myapprunner.com"
hosted_zone = route53.HostedZone(...)

domain_assoc_provider = custom_resources.Provider(
    self, "Domain Association Provider",
    on_event_handler=lambda_.Function(
        self, "Domain Association Function",
        runtime=lambda_.Runtime.PYTHON_3_8,
        handler="index.on_event",
        code=lambda_.Code.from_asset("domain-assoc-cr"),
        log_retention=logs.RetentionDays.ONE_MONTH,
        initial_policy=[iam.PolicyStatement(
            actions=[
                "apprunner:AssociateCustomDomain",
                "apprunner:DescribeCustomDomains",
                "apprunner:DisassociateCustomDomain",
            ],
            resources=["*"],
        )],
        timeout=Duration.minutes(10),
    ),
    log_retention=logs.RetentionDays.ONE_MONTH,
)
domain_assoc = CustomResource(
    self, "Domain Association",  # change the name when any parameter changes as we don't really support updates
    service_token=domain_assoc_provider.service_token,
    properties={
        "ServiceArn": app.service_arn,
        "DomainName": domain,
        "HostedZoneName": hosted_zone.zone_name,
    },
)
for i in range(1, 4):
    route53.CnameRecord(
        self, f"Certificate Validation {i}",
        zone=hosted_zone,
        record_name=domain_assoc.get_att_string(f"Name{i}"),
        domain_name=domain_assoc.get_att_string(f"Value{i}"),
    )

domain-assoc-cr slash index.py

import boto3
import json
import time

apprunner = boto3.client("apprunner")


def on_event(event, context):
    print(event)
    request_type = event["RequestType"]
    if request_type == "Create":
        return on_create(event)
    if request_type == "Update":
        return on_update(event)
    if request_type == "Delete":
        return on_delete(event)
    raise Exception("Invalid request type: %s" % request_type)


def on_create(event):
    props = event["ResourceProperties"]
    print("create new resource with props %s" % props)

    service_arn = props["ServiceArn"]
    domain_name = props["DomainName"]
    hosted_zone_name = props["HostedZoneName"]

    apprunner.associate_custom_domain(
        ServiceArn=service_arn,
        DomainName=domain_name,
    )

    while True:
        status = apprunner.describe_custom_domains(ServiceArn=service_arn)
        for domain_status in status["CustomDomains"]:
            if domain_status["DomainName"] == domain_name:
                if domain_status["Status"].lower() == "pending_certificate_dns_validation":
                    if len(domain_status["CertificateValidationRecords"]) != 3:
                        raise Exception(f"DNS validation records not in length of 3: {domain_status}")

                    return {
                        "PhysicalResourceId": domain_name,
                        "Data": {
                            "Name1": domain_status["CertificateValidationRecords"][0]["Name"].replace(f".{hosted_zone_name}.", ""),
                            "Value1": domain_status["CertificateValidationRecords"][0]["Value"],
                            "Name2": domain_status["CertificateValidationRecords"][1]["Name"].replace(f".{hosted_zone_name}.", ""),
                            "Value2": domain_status["CertificateValidationRecords"][1]["Value"],
                            "Name3": domain_status["CertificateValidationRecords"][2]["Name"].replace(f".{hosted_zone_name}.", ""),
                            "Value3": domain_status["CertificateValidationRecords"][2]["Value"],
                        },
                    }

        time.sleep(5)


def on_update(event):
    physical_id = event["PhysicalResourceId"]
    props = event["ResourceProperties"]
    print("(nop) update resource %s with props %s" % (physical_id, props))


def on_delete(event):
    physical_id = event["PhysicalResourceId"]
    props = event["ResourceProperties"]
    print("delete resource %s" % physical_id)

    service_arn = props["ServiceArn"]
    domain_name = props["DomainName"]
    
    try:
        apprunner.disassociate_custom_domain(
            ServiceArn=service_arn,
            DomainName=domain_name,
        )
    except Exception as e:
        print("Failed, but ignoring:", e)