通过 Amazon API Gateway 和 Amazon Lambda 实现基于 Restful API 的 CloudFront Distribution 复制/克隆功能

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "acm:ListCertificates",
                "cloudfront:CreateDistribution",
                "cloudfront:GetDistributionConfig"
            ],
            "Resource": "*"
        },
        {
            "Sid": "VisualEditor1",
            "Effect": "Allow",
            "Action": [
                "s3:PutBucketAcl",
                "s3:GetBucketAcl",
                "logs:CreateLogGroup"
            ],
            "Resource": [
                "arn:aws:s3:::<S3_bucket_name >",
                "arn:aws:logs:us-east-1:<account_id>:*"
            ]
        },
        {
            "Sid": "VisualEditor2",
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "arn:aws:logs:us-east-1: :<account_id>:log-group:/aws/lambda/cf_distribution_clone:*"
        }
    ]
}

import boto3
from botocore.config import Config
import botocore.exceptions
from datetime import datetime, timezone
import logging
import json
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()
 
cf_client = boto3.client('cloudfront')
 
my_config = Config(
   region_name = 'us-east-1'
)
acm_client = boto3.client('acm', config=my_config)
 
def lambda_handler(event, context):
    conf_domain = event['queryStringParameters']['domain']
    conf_origin = event['queryStringParameters']['origin']
    conf_ref_dist = event['queryStringParameters']['ref_dist']
 
    # 1) 获取参考域名的配置信息做为基准配置
    ref_config = get_reference_config(conf_ref_dist)
    # 2) 获取账号下ACM证书与域名的对应列表
    certs = get_certificate_mapping()
    # 3) 获取该CNAME对应的ACM证书
    certArn = get_certificate_arn(certs, conf_domain)
    # 4) 构造新Distribution的配置信息
    new_config = set_config_based_on_ref(ref_config, conf_domain, conf_origin, certArn)
    # 5) 创建新Distribution
    distribution = create_distribution(new_config)
   
    response_body = {}
    response_body['requestId'] = context.aws_request_id
    response_body['distributionId'] = distribution['Distribution']['Id']
    responseObject = {}
    responseObject['statusCode'] = 200
    responseObject['body'] = json.dumps(response_body)
    return(responseObject)
 
def get_reference_config(ref_dist):
    try:
        return cf_client.get_distribution_config(Id=ref_dist)
    except botocore.exceptions.ClientError as error:
        logger.exception(f"{format(error)}")
        raise error
 
def get_certificate_mapping():
    try:
        response = acm_client.list_certificates(
            CertificateStatuses=[
                'ISSUED'
            ],
            MaxItems=1000
        )
        certs = response['CertificateSummaryList']
        while "NextToken" in response:
            response = acm_client.list_certificates(
                    CertificateStatuses=[
                    'ISSUED'
                ],
                MaxItems=1000,
                NextToken= response['NextToken']
            )
            certs.extend(response["CertificateSummaryList"])
        cert_dict = {}
        for cert in    certs:
            cert_dict[cert['DomainName']] = cert['CertificateArn']
        return(cert_dict)
    except botocore.exceptions.ClientError as error:
        logger.exception(f"{format(error)}")
        raise error
 
def get_certificate_arn(certs, domain):
    if domain in certs:
        cert = certs[domain]
    else:
        cert_domain = '*.' + domain.split(".", 1)[-1]
        if cert_domain in certs:
            cert = certs[cert_domain]
        else:
            logger.info(f"No certificate for domain - {format(domain)} in ACM. Please create or import one.")
            exit(1)
    logger.info(f"Use ACM certificate for domain \'{format(domain)}\': {format(cert)}.")
    return cert
 
 
def set_config_based_on_ref(ref_config, conf_domain, conf_origin, certArn):
    ref_config['DistributionConfig']['Aliases'] = {
        'Quantity': 1,
         'Items': [
                conf_domain
         ]
    }
    new_config = ref_config['DistributionConfig']
    new_config['CallerReference'] = str(datetime.now(tz=None).timestamp())
    new_config['Origins']['Items'][0]['Id'] = conf_origin
    new_config['Origins']['Items'][0]['DomainName'] = conf_origin
    new_config['DefaultCacheBehavior']['TargetOriginId'] = conf_origin
    new_config['Comment'] = conf_domain
    new_config['ViewerCertificate']['ACMCertificateArn'] = certArn
    return new_config
 
def create_distribution(config):
    try:
        distribution = cf_client.create_distribution(DistributionConfig=config)
        logger.info(f"Done! Created distribution {format(distribution['Distribution']['Id'])}.")
    except botocore.exceptions.ClientError as error:
        logger.exception(f"{format(error)}")
        raise error
    return(distribution)