ansible-infra/roles/acme/files/acme_renew_cert.py

184 lines
7.3 KiB
Python
Executable File

#!/usr/bin/python3
import logging
import os
import sys
from datetime import datetime
import josepy as jose
from OpenSSL import crypto
from acme import client, messages, challenges, crypto_util
SSL_CONFIG_DIR="/etc/ssl"
OPENSSL_DATE_FORMAT = '%Y%m%d%H%M%SZ'
USER_AGENT="python-acme"
def select_http01_challenge(order):
for auth in order.authorizations:
for challenge in auth.body.challenges:
if isinstance(challenge.chall, challenges.HTTP01):
return challenge
raise Exception("HTTP-01 challenge was not offered by the CA server.")
def perform_http01_challenge(client_acme, challenge, order, challenge_dir):
response, validation = challenge.response_and_validation(client_acme.net.key)
validation_filename, validation_content = validation.split('.')
challenge_path = os.path.join(challenge_dir, validation_filename)
with open(challenge_path, 'w') as f_out:
f_out.write(validation_content)
client_acme.answer_challenge(challenge, response)
fullchain_pem = client_acme.poll_and_finalize(order).fullchain_pem
os.remove(challenge_path)
return fullchain_pem
def renew_cert(domain,
account_key_path,
privkey_path,
csr_path,
certs_dir,
challenge_dir,
directory_url,
days_before_renewal,
force = False):
logging.info(f"Checking {domain} certificate's expiration date…")
now = datetime.now()
cert_expiration_date = now
fullchain_path = os.path.join(certs_dir, "fullchain.pem")
try:
with open(fullchain_path, 'rb') as pem_in:
fullchain_pem = crypto.load_certificate(crypto.FILETYPE_PEM, pem_in.read())
cert_expiration_date = datetime.strptime(fullchain_pem.get_notAfter().decode(), OPENSSL_DATE_FORMAT)
except IOError as e:
logging.warning(f"Unable to load {domain} certificate: {e}")
cert_expiration_days = (cert_expiration_date - now).days
if not force and cert_expiration_days >= days_before_renewal:
logging.info(f"Certificate expires in {cert_expiration_days} days. Nothing to do.")
return
logging.info(f"Certificate expires in {cert_expiration_days} days! Renewing certificate…")
logging.info(f"Loading account key from {account_key_path}")
with open(account_key_path, 'rb') as pem_in:
account_pem = crypto.load_privatekey(crypto.FILETYPE_PEM, pem_in.read())
account_jwk = jose.JWKRSA(key=account_pem.to_cryptography_key())
account_jwk_pub = account_jwk.public_key()
logging.info(f"Loading {domain} CSR file from {csr_path}")
try:
with open(csr_path, 'r') as pem_in:
csr_pem = pem_in.read()
except IOError as e:
logging.warning(f"Unable to load CSR file: {e}")
logging.info(f"Loading {domain} private key from {privkey_path}")
with open(privkey_path, 'r') as pem_in:
privkey_pem = pem_in.read()
csr_pem = crypto_util.make_csr(privkey_pem, [domain])
client_network = client.ClientNetwork(account_jwk, user_agent=USER_AGENT)
directory = messages.Directory.from_json(client_network.get(directory_url).json())
client_acme = client.ClientV2(directory, net=client_network)
logging.info("Registering with ACME account…")
# Here we assume we already have an account
registration_message = messages.NewRegistration(
key=account_jwk_pub,
only_return_existing=True)
registration_response = client_acme._post(directory["newAccount"], registration_message)
account = client_acme._regr_from_response(registration_response)
client_acme.net.account = account
logging.info("Ordering ACME challenge…")
order = client_acme.new_order(csr_pem)
logging.info("Selecting HTTP-01 ACME challenge…")
challenge = select_http01_challenge(order)
logging.info("Performing HTTP-01 ACME challenge…")
fullchain_pem = perform_http01_challenge(client_acme, challenge, order, challenge_dir)
logging.info(f"Writing {domain} certificates into {certs_dir}")
certs_pem = []
for line in fullchain_pem.split('\n'):
if 'BEGIN CERTIFICATE' in line:
cert_pem = ''
cert_pem += line + '\n'
if 'END CERTIFICATE' in line:
certs_pem.append(cert_pem)
cert_path = os.path.join(certs_dir, "cert.pem")
with open(cert_path, 'w') as pem_out:
pem_out.write(certs_pem[0])
chain_path = os.path.join(certs_dir, "chain.pem")
with open(chain_path, 'w') as pem_out:
pem_out.write(''.join(certs_pem[1:]))
with open(fullchain_path, 'w') as pem_out:
pem_out.write(fullchain_pem)
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(
description="Renew ACME certificates for a list of domains. This script assumes you already have an account on the CA server and a private key for each certificate.",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("domains", nargs='+',
help="List of domain names for which to renew the certificate.")
parser.add_argument("--account_key", "-a",
default=os.path.join(SSL_CONFIG_DIR, "accounts", "acme_account.key"),
help="Path to the account key.")
parser.add_argument("--privkey", "-p",
default=os.path.join(SSL_CONFIG_DIR, "private", "{domain}.pem"),
help="Path to the private certificate.")
parser.add_argument("--csr", "-r",
default=os.path.join(SSL_CONFIG_DIR, "csr", "{domain}.pem"),
help="Path to the CSR file. If the file doesn't exist, it will be generated from the private key and the domain name.")
parser.add_argument("--certs", "-o",
default=os.path.join(SSL_CONFIG_DIR, "certs", "{domain}.d"),
help="Path to the certificates directory.")
parser.add_argument("--challenge", "-c",
default="/var/www/html/.well-known/acme-challenge",
help="Path to the challenge directory.")
parser.add_argument("--directory_url", "-u",
default="https://acme-v02.api.letsencrypt.org/directory",
help="Directory URL on which performing ACME challenges. Only ACME v2 is supported.")
parser.add_argument("--days", "-d", type=int,
default=30,
help="Days before attempting to renew the certificates.")
parser.add_argument("--quiet", "-q",
action="store_true",
help="Quiet mode.")
parser.add_argument("--force", "-f",
action="store_true",
help="Force certificates renewal without checking their expiration date.")
args = parser.parse_args()
log_level = logging.WARN if args.quiet else logging.INFO
logging.basicConfig(stream=sys.stdout, level=log_level, format="%(levelname)s:%(message)s")
for domain in args.domains:
account_key = args.account_key.format(domain=domain)
privkey = args.privkey.format(domain=domain)
csr = args.csr.format(domain=domain)
certs_dir = args.certs.format(domain=domain)
challenge_dir = args.challenge.format(domain=domain)
renew_cert(domain,
account_key_path=account_key,
privkey_path=privkey,
csr_path=csr,
certs_dir=certs_dir,
challenge_dir=challenge_dir,
directory_url=args.directory_url,
days_before_renewal=args.days,
force=args.force)