Skip to content

Instantly share code, notes, and snippets.

@kichik
Created February 3, 2023 22:35
Show Gist options
  • Save kichik/961b26d8360c4bbda4f7a33a79ec7af3 to your computer and use it in GitHub Desktop.
Save kichik/961b26d8360c4bbda4f7a33a79ec7af3 to your computer and use it in GitHub Desktop.
Custom resource to automatically associate domain to App Runner service
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}"),
)
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment