Self Signed PKI

self-signed.py

#!/usr/bin/env python3
# Title: POOR MAN'S PKI
# Author: James Veitch
# Date: 2018
# =================
# Description:
#   Designed to quickly create a Root CA and signed bundle for
#   a defined host. NOT INTENDED FOR PRODUCTION.
#   Assumes you have OpenSSL installed and available in $PATH.


import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
import sys
import argparse
import logging
import re
import os


format = '%(asctime)s - %(levelname)s - %(message)s'
logging.basicConfig(format=format)
log = logging.getLogger(__name__)
log.setLevel(logging.INFO)

MY_DOMAIN: str
HOST: str
ROOT_DIRECTORY: Path = Path(os.getcwd())


@dataclass
class HostnameDetails:
    host: str
    domain: str
    fqdn: str


@dataclass
class Certificate:
    hostname: HostnameDetails
    san: Optional[str] = None
    root_ca: Optional['Certificate'] = None
    certificate: Optional[str] = None
    csr: Optional[str] = None
    extFile: Optional[str] = None
    private_key: Optional[str] = None
    bundle: Optional[str] = None


def _extract_components(hostname: str) -> HostnameDetails:
    if len(hostname.split('.')) > 2:
        log.debug('FQDN with subdomain detected')
        host = '.'.join(hostname.split('.')[:-2])
        domain = '.'.join(hostname.split('.')[-2:])
    elif len(hostname.split('.')) == 2:
        log.debug('TLD detected')
        host = '.'.join(hostname.split('.')[:-1])
        domain = '.'.join(hostname.split('.')[-1:])
    elif len(hostname.split('.')) == 1:
        log.debug('Single host detected')
        hostname = f'{hostname}.local'
        host = '.'.join(hostname.split('.')[:-1])
        domain = '.'.join(hostname.split('.')[-1:])
    else:
        log.error('Unable to extract components from {hostname}')
        return

    log.info(f'Host: {host}')
    log.info(f'Domain: {domain}')
    log.info(f'FQDN: {hostname}')

    return HostnameDetails(host=host, domain=domain, fqdn=hostname)


def _is_valid_hostname(hostname: str) -> HostnameDetails:
    if hostname[-1] == ".":
        # strip exactly one dot from the right, if present
        hostname = hostname[:-1]
    if len(hostname) > 253:
        return False

    labels = hostname.split(".")

    # the TLD must be not all-numeric
    if re.match(r"[0-9]+$", labels[-1]):
        return False

    allowed = re.compile(r"(?!-)[a-z0-9-]{1,63}(?<!-)$", re.IGNORECASE)
    valid = all(allowed.match(label) for label in labels)

    if not valid:
        log.error(f'{hostname} is not a valid host')
        return
    log.debug(f'Valid hostname: {hostname}')
    log.debug(f'Associated labels: {labels}')

    return _extract_components(hostname)


def extract_host_components(hostname: str) -> HostnameDetails:
    return _is_valid_hostname(hostname)


def _create_directory_structure(hostname: HostnameDetails) -> tuple[str, str]:
    # Create root
    domain_certs_root = Path(os.path.join(ROOT_DIRECTORY, hostname.domain))
    os.makedirs(domain_certs_root, exist_ok=True)
    # Create host
    host_certs_root = os.path.join(domain_certs_root, hostname.fqdn)
    os.makedirs(host_certs_root, exist_ok=True)

    return domain_certs_root, host_certs_root


def create_certificate(hostname: HostnameDetails, keysize: int = 4096,
                             days: int = 3650, is_ca: bool = False) -> Certificate:  # noqa
    """
    Creates an individual certificate for a subdomain/host.
    """
    domain_certs_root, host_certs_root = _create_directory_structure(hostname)

    cert = Certificate(hostname=hostname)

    # Check if we have a RootCA already
    if (not cert.root_ca or not os.path.exists(cert.root_ca.certificate)) and not is_ca:  # noqa
        log.info(f'No Root CA detected for {hostname.domain}')
        cert.root_ca = create_certificate(
            HostnameDetails(None, hostname.domain, fqdn=hostname.domain),
            is_ca=True)

    # Create certificate using OpenSSL
    cert.certificate = os.path.join(host_certs_root, 'cert.pem')
    cert.csr = os.path.join(host_certs_root, 'csr.pem')
    cert.extFile = os.path.join(host_certs_root, 'extfile.cnf')
    cert.private_key = os.path.join(host_certs_root, 'privkey.pem')
    cert.bundle = os.path.join(host_certs_root, 'fullchain.pem')

    if is_ca:
        log.debug(f'Creating Root CA for {hostname.domain}')
        cmd = ['openssl', 'req', '-x509', '-newkey', f'rsa:{keysize}',
               '-keyout', cert.private_key, '-out', cert.certificate,
               '-days', str(days), '-sha512', '-nodes',
               '-subj', f'/CN=*.{cert.hostname.domain}']
        log.debug(f'Running OpenSSL with command: {" ".join(cmd)}')
        result = subprocess.run(cmd, stdout=subprocess.PIPE)
        log.debug(result)
        log.info(f'Created Root CA: {cert.certificate}')
    else:
        log.debug(f'Creating Certificate for {hostname.fqdn}')
        # Generate private key for host
        log.debug(f'Creating Private Key for {hostname.fqdn}')
        cmd = ['openssl', 'genrsa', '-out', cert.private_key, str(keysize)]  # noqa cSpell:ignore genrsa
        result = subprocess.run(cmd, stdout=subprocess.PIPE)
        log.debug(result)
        log.info(f'Created Private Key: {cert.private_key}')
        # Generate CSR
        log.debug(f'Creating Certificate Signing Request (CSR) for {hostname.fqdn}')  # noqa
        cert.san = f'DNS:{hostname.fqdn},IP:127.0.0.1'
        with open(cert.extFile, 'w+') as file:
            file.write(f'subjectAltName = {cert.san}')
        cmd = ['openssl', 'req', '-new', '-key', cert.private_key, '-out',
               cert.csr, '-subj', f'/CN={hostname.fqdn}']
        result = subprocess.run(cmd, stdout=subprocess.PIPE)
        # Sign the CSR with our Root CA and create Certificate
        log.debug(f'Signing CSR with {cert.root_ca}')
        cmd = ['openssl', 'x509', '-CA', cert.root_ca.certificate,
               '-CAkey', cert.root_ca.private_key, '-CAcreateserial',
               '-req', '-in', cert.csr, '-out', cert.certificate,
               '-days', str(days), '-extfile', cert.extFile]
        result = subprocess.run(cmd, stdout=subprocess.PIPE)
        log.info(f'Created signed Certificate: {cert.certificate}')
        # Create bundle
        log.debug(f'Creating bundle for {hostname.fqdn}')
        cmd = ['cat', cert.certificate, cert.root_ca.certificate]
        result = subprocess.run(cmd, stdout=open(cert.bundle, 'w+'))
        log.info(f'Created Certificate bundle: {cert.bundle}')
        # Verify!
        log.debug(f'Verifying bundle for {hostname.fqdn}')
        cmd = ['openssl', 'verify', '-CAfile', cert.root_ca.certificate,
               cert.bundle]
        result = subprocess.run(cmd, stdout=subprocess.PIPE)
        log.info(f'Verified bundle: {cert.bundle} with Root CA {cert.root_ca}')

    return cert


if __name__ == '__main__':

    parser = argparse.ArgumentParser(
        description="""
            Poor Man's Self Signed Certificates.
            Create a quick Public Key Infrastructure for Testing and
            Development purposes.
        """,
        epilog="... Yes that's right. Testing and Development ONLY ...")
    group = parser.add_mutually_exclusive_group()
    parser.add_argument(
        '--host', type=str, nargs='+',
        help='Hostnames we want to create certificates for. Use a FQDN.')
    group.add_argument(
        '-v', '--verbose', action="store_true",
        help='Increase script logging verbosity.')
    group.add_argument(
        '-q', '--quiet', action="store_true",
        help='Run script without any output.')
    parser.add_argument_group(group)
    args = parser.parse_args()

    # Logging
    if args.verbose:
        log.setLevel(logging.DEBUG)
        log.info("Set logging level to DEBUG")
    if args.quiet:
        log.setLevel(logging.NOTSET)

    # Hosts
    if not args.host:
        log.critical("""
            No hosts found. Please use -h or --host to pass in
            host(s) to generate certificates for.""")
        sys.exit(1)

    for h in args.host:
        log.debug(f"Parsed --host {h}")
        details = extract_host_components(h)
        log.info(details)
        create_certificate(details)