auto generate ssl certs for dns managed in godaddy/aliyun/route53

auto generate ssl certs

# script to auto generate ssl certs and upload to s3 bucket
import os
import boto3
import logging
import subprocess
import sys
import time
from multiprocessing import Pool

# set logging to stdout
logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler(sys.stdout))

s3_client = boto3.client('s3')
s3_bucket = 'ssl-certs-for-uat'

# generate domain txt file to import into dns provider for dns batch purchasing
def generate_domain_list(root_domain, cert_num):
    """_summary_

    Args:
        root_domain (_type_): _description_
        cert_num (_type_): _description_

    Returns:
        _type_: _description_
    """
    domain_list = []
    for i in range(int(cert_num)):
        # add random suffix to domain name
        domain_list.append('{}{}{}'.format(str(root_domain), i, '.cn'))
    with open('domain_list.txt', 'w') as f:
        for domain in domain_list:
            f.write(domain + '\n')
    logger.info('Generated domain list: {}'.format(domain_list))

# using certbot godaddy plugin to generate ssl certs
def generate_ssl_certs_by_aliyun(domain_name):
    """_summary_

    Args:
        domain_name (_type_): _description_
    """    
    logger.info('Generating ssl certs for domain: %s', domain_name)
    try:
        resp = subprocess.run(['certbot', 'certonly', '--authenticator', 'certbot-dns-aliyun:dns-aliyun', '--certbot-dns-aliyun:dns-aliyun-credentials', './aliyun.ini', '-d', domain_name])
        logger.info('Certs generated: %s', resp)
        upload_ssl_certs(domain_name)
    except Exception as e:
        logger.error('error validating certificate: %s', e)
    logger.info('Generated ssl certs for domain: %s', domain_name)

# using certbot godaddy plugin to generate ssl certs
def generate_ssl_certs_by_godaddy(domain_name):
    logger.info('Generating ssl certs for domain: %s', domain_name)
    try:
        resp = subprocess.run(['certbot', 'certonly', '--authenticator', 'dns-godaddy', '--dns-godaddy-credentials', './godaddy.ini', '--dns-godaddy-propagation-seconds', '60', '--keep-until-expiring', '--non-interactive', '--force-renewal', '--expand', '--server', 'https://acme-v02.api.letsencrypt.org/directory', '-d', domain_name])
        logger.info('Certs generated: %s', resp)
        upload_ssl_certs(domain_name)
    except Exception as e:
        logger.error('error validating certificate: %s', e)
    logger.info('Generated ssl certs for domain: %s', domain_name)

# using certbot route53 plugin to generate ssl certs
def generate_ssl_certs_by_route53(domain_name):
    logger.info('Generating ssl certs for domain: %s', domain_name)
    try:
        resp = subprocess.run(['certbot', 'certonly', '--dns-route53', '--dns-route53-propagation-seconds', '30', '--non-interactive', '--force-renewal', '-d', domain_name])
        logger.info('Certs generated: %s', resp)
        upload_ssl_certs(domain_name)
    except Exception as e:
        logger.error('error validating certificate: %s', e)
    logger.info('Generated ssl certs for domain: %s', domain_name)

# upload ssl certs to s3 bucket
def upload_ssl_certs(domain_name):
    logger.info('Uploading ssl certs for domain: %s', domain_name)
    ssl_cert_path = '/etc/letsencrypt/live/{}/cert.pem'.format(domain_name)
    ssl_chain_path = '/etc/letsencrypt/live/{}/chain.pem'.format(domain_name)
    ssl_fullchain_path = '/etc/letsencrypt/live/{}/fullchain.pem'.format(domain_name)
    ssl_privkey_path = '/etc/letsencrypt/live/{}/privkey.pem'.format(domain_name)
    ssl_cert_content = open(ssl_cert_path, 'rb').read()
    ssl_chain_content = open(ssl_chain_path, 'rb').read()
    ssl_fullchain_content = open(ssl_fullchain_path, 'rb').read()
    ssl_privkey_content = open(ssl_privkey_path, 'rb').read()
    # check if s3 bucket exists and create if not
    try:
        s3_client.head_bucket(Bucket=s3_bucket)
    except Exception as e:
        logger.info('Creating ssl bucket: %s', domain_name)
        # create s3 bucket with acl public-read
        s3_client.create_bucket(Bucket=s3_bucket, CreateBucketConfiguration={'LocationConstraint': 'us-west-2'})
    # upload ssl certs to s3 bucket, include cert.pem, chain.pem, fullchain.pem, privkey.pem
    s3_client.put_object(Body=ssl_cert_content, Bucket=s3_bucket, Key='{}/cert.pem'.format(domain_name))
    s3_client.put_object(Body=ssl_chain_content, Bucket=s3_bucket, Key='{}/chain.pem'.format(domain_name))
    s3_client.put_object(Body=ssl_fullchain_content, Bucket=s3_bucket, Key='{}/fullchain.pem'.format(domain_name))
    s3_client.put_object(Body=ssl_privkey_content, Bucket=s3_bucket, Key='{}/privkey.pem'.format(domain_name))

    logger.info('Uploaded ssl certs for domain: %s', domain_name)

def main():
    if len(sys.argv) <= 3:
        logger.info('Usage: ssl_generate.py <root_domain> <cert_num> <plugin_name>')
        sys.exit(1)

    # parse the option
    logger.info("Generating ssl certs for domain: {}".format(sys.argv[1]))

    # use multiprocessing to generate ssl certs and upload to s3 bucket
    pool = Pool(processes=int(sys.argv[2]))
    for i in range(int(sys.argv[2])):
        domain_name = '{}.{}'.format(i, sys.argv[1])
        if sys.argv[3] == 'godaddy':
            pool.apply(generate_ssl_certs_by_godaddy, (domain_name,))
        elif sys.argv[3] == 'route53':
            pool.apply(generate_ssl_certs_by_route53, (domain_name,))
        elif sys.argv[3] == 'aliyun':
            pool.apply(generate_ssl_certs_by_aliyun, (domain_name,))

    logger.info("Generated ssl certs for domain: {}".format(sys.argv[1]))

if __name__ == '__main__':
    main()
    sys.exit(0)