Skip to content
Snippets Groups Projects
Unverified Commit 377c0db3 authored by Max Adamo's avatar Max Adamo
Browse files

refactor: remove cert2json script and update provider handling in combine.py

parent f7cdc185
Branches
Tags
No related merge requests found
#!/usr/bin/python3
#
"""Certbot list 2 json
Usage:
cert2json --provider <PROVIDER>
cert2json (-h | --help)
Options:
-h --help Show this screen
-p PROVIDER --provider=PROVIDER Provider [check /root/.acme.ini for valid providers, or use all]
"""
import json
import os
from re import findall
from glob import glob
from multiprocessing import Pool
from datetime import datetime, timezone
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from docopt import docopt
import wile_coyote.tools
PROCESSES = 6
WEB_BASE = wile_coyote.tools.WEB_BASE
ACME_PROVIDERS = wile_coyote.tools.ACME_PROVIDERS
ALL_PROVIDERS = ACME_PROVIDERS.append("all")
ARGS = docopt(__doc__)
PROVIDER = ARGS['--provider']
if PROVIDER not in ALL_PROVIDERS:
print(f"valid providers are: {', '.join(ALL_PROVIDERS)}")
os.sys.exit()
elif PROVIDER == 'all':
PROVIDERS = ACME_PROVIDERS
else:
PROVIDERS = [PROVIDER]
def inspect_certificate(cert):
""" extract data from the certificate """
with open(os.path.join(cert, 'fullchain.pem'), "rb") as pem_file:
pem_data = pem_file.read()
cert = x509.load_pem_x509_certificate(pem_data, default_backend())
cert_name = cert.subject.get_attributes_for_oid(x509.OID_COMMON_NAME)[0].value.strip()
serial_raw = f'{cert.serial_number:x}'.upper()
if len(serial_raw) % 2:
serial_raw = f'0{serial_raw}'
serial = ':'.join(serial_raw[i:i+2] for i in range(0, len(serial_raw), 2))
san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
san_names = san.value.get_values_for_type(x509.DNSName)
days_left = (cert.not_valid_after_utc - datetime.now(timezone.utc)).days
status = "VALID" if days_left > 0 else "EXPIRED"
valid = f"{status}: {days_left} DAYS"
dict_data = {
"certname": cert_name,
"serial_number": serial,
"domains": san_names,
"expiry_date": valid
}
return dict_data
if __name__ == "__main__":
for acme_provider in PROVIDERS:
_certs = glob(f'/etc/{acme_provider}/live/*')
certs = [cert for cert in _certs if not cert.endswith('README')]
provider_dir = os.path.join(WEB_BASE, acme_provider)
json_file = f"{provider_dir}/{acme_provider}.json"
json_expired = f"{provider_dir}/{acme_provider}_expired.json"
pool = Pool(processes=PROCESSES)
cert_list = pool.map(inspect_certificate, certs)
pool.close()
pool.join()
sorted_certname = sorted(cert_list, key=lambda k: k['certname'])
sorted_expired = sorted(
cert_list,
key=lambda k: int(findall(r'\d+', k['expiry_date'])[0])
)
with open(json_file, "w", encoding="utf8") as json_out, \
open(json_expired, "w", encoding="utf8") as json_exp_out:
json_out.write(json.dumps(sorted_certname))
json_exp_out.write(json.dumps(sorted_expired))
json_out.close()
json_exp_out.close()
# fix permissions
os.system(f"find {WEB_BASE} -type d -exec chmod 755 '{{}}' +")
os.system(f"find {WEB_BASE} -type f -exec chmod 644 '{{}}' +")
......@@ -14,8 +14,8 @@ def keys(certpath, provider, keypath, outpath):
raise NotImplementedError("OS not supported")
# providers and CAs are mapped in acme.ini
providers_ca = wile_coyote.tools.PROVIDERS_CA
capath = os.path.join(ssl_dir, providers_ca[provider])
acme_providers = wile_coyote.tools.ACME_PROVIDERS
capath = os.path.join(ssl_dir, acme_providers[provider])
filenames = [certpath, capath, keypath]
with open(outpath, "w", encoding="utf-8") as outfile:
......
......@@ -22,7 +22,6 @@ for loader, module_name, is_pkg in pkgutil.walk_packages(__path__):
CONSUL_TOKEN = config.get('acme', 'consul_token')
WEB_BASE = config.get('acme', 'web_base')
ACME_PROVIDERS = l_eval(config.get('acme', 'acme_providers'))
PROVIDERS_CA = config["providers_ca"]
# these parameters only work in test
if 'unit-test' not in config.sections():
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment