Mini CA using Python `cryptography` module

minica.py

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.x509.oid import NameOID
import datetime
from base64 import b64encode

from typing import Tuple, List

from ipaddress import ip_address

keyUsage = {
    "digital_signature": False,
    "content_commitment": False,
    "key_encipherment": False,
    "data_encipherment": False,
    "key_agreement": False,
    "key_cert_sign": False,
    "crl_sign": False,
    "encipher_only": False,
    "decipher_only": False,
}


def generateRootCA(name: str) -> Tuple[str, str]:
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    rootca_key = key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,
        encryption_algorithm=serialization.NoEncryption(),
    ).decode()

    subject = issuer = x509.Name(
        [
            x509.NameAttribute(NameOID.COUNTRY_NAME, u"IN"),
            x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, u"KA"),
            x509.NameAttribute(NameOID.LOCALITY_NAME, u"Mys"),
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, u"Adyanth HomeLab"),
            x509.NameAttribute(NameOID.COMMON_NAME, name),
        ]
    )

    cert = (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(issuer)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(datetime.datetime.utcnow())
        .not_valid_after(
            # Our certificate will be valid for 10 years
            datetime.datetime.utcnow()
            + datetime.timedelta(days=3650)
        )
        .add_extension(
            x509.BasicConstraints(ca=True, path_length=None),
            critical=False,
        )
        .add_extension(
            x509.KeyUsage(**{**keyUsage, "key_cert_sign": True}), critical=False
        )
        # Sign our certificate with our private key
        .sign(key, hashes.SHA256())
    )

    rootca_cert = cert.public_bytes(serialization.Encoding.PEM).decode()

    return rootca_cert, rootca_key


def generateSignedKeyPair(
    rootCAKey: str, rootCACert: str, sans: List[str], ips: List[str] = []
) -> Tuple[str, str]:
    rootCAKey: rsa.RSAPrivateKey = serialization.load_pem_private_key(
        rootCAKey.encode(), None
    )
    rootCACert: x509.Certificate = x509.load_pem_x509_certificate(rootCACert.encode())

    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    new_key = key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,
        encryption_algorithm=serialization.NoEncryption(),
    ).decode()

    if not sans:
        sans = ["app.domain.com"]

    subject = x509.Name(
        [
            x509.NameAttribute(NameOID.COUNTRY_NAME, u"IN"),
            x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, u"KA"),
            x509.NameAttribute(NameOID.LOCALITY_NAME, u"Mys"),
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, u"Adyanth HomeLab"),
            x509.NameAttribute(NameOID.COMMON_NAME, sans[0]),
        ]
    )

    cert = (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(rootCACert.issuer)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(datetime.datetime.utcnow())
        .not_valid_after(
            # Our certificate will be valid for 1 year
            datetime.datetime.utcnow()
            + datetime.timedelta(days=365)
        )
        .add_extension(
            x509.SubjectAlternativeName(
                list(map(lambda x: x509.DNSName(x), sans))
                + list(map(lambda x: x509.IPAddress(ip_address(x)), ips))
            ),
            critical=False,
        )
        .add_extension(
            x509.KeyUsage(**{**keyUsage, "content_commitment": True}), critical=False
        )
        # Sign our certificate with our private key
        .sign(rootCAKey, hashes.SHA256())
    )

    new_cert = cert.public_bytes(serialization.Encoding.PEM).decode()

    return new_key, new_cert


def generatePKIXBundle(secret: str, key: str, cert: str, certs: List[str] = []) -> str:
    key: rsa.RSAPrivateKey = serialization.load_pem_private_key(key.encode(), None)
    cert: x509.Certificate = x509.load_pem_x509_certificate(cert.encode())
    certs: List[x509.Certificate] = list(
        map(lambda x: x509.load_pem_x509_certificate(x.encode()), certs)
    )
    pkcs12bytes = serialization.pkcs12.serialize_key_and_certificates(
        name=b"pkcs12bundle",
        key=key,
        cert=cert,
        cas=certs,
        encryption_algorithm=serialization.BestAvailableEncryption(secret.encode()),
    )
    return b64encode(pkcs12bytes).decode()


def generateCSRFromKey(key: str, sans: List[str], ips: List[str] = []) -> str:
    key: rsa.RSAPrivateKey = serialization.load_pem_private_key(
        key.encode(), None)

    subject = x509.Name(
        [
            x509.NameAttribute(NameOID.COUNTRY_NAME, u"IN"),
            x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, u"KA"),
            x509.NameAttribute(NameOID.LOCALITY_NAME, u"Mys"),
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, u"Adyanth HomeLab"),
            x509.NameAttribute(NameOID.COMMON_NAME, sans[0]),
        ]
    )

    csr = (
        x509.CertificateSigningRequestBuilder()
        .subject_name(subject)
        .add_extension(
            x509.SubjectAlternativeName(
                list(map(lambda x: x509.DNSName(x), sans))
                + list(map(lambda x: x509.IPAddress(ip_address(x)), ips))
            ),
            critical=False,
        )
        .add_extension(
            x509.KeyUsage(**{**keyUsage, "content_commitment": True}), critical=False
        )
        # Sign our certificate with our private key
        .sign(key, hashes.SHA256())
    )

    return csr.public_bytes(serialization.Encoding.PEM).decode()


def generateKeyAndCSR(sans: List[str], ips: List[str] = []) -> Tuple[str, str]:
    key: rsa.RSAPrivateKey = rsa.generate_private_key(
        public_exponent=65537, key_size=2048)
    new_key = key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,
        encryption_algorithm=serialization.NoEncryption(),
    ).decode()

    return new_key, generateCSRFromKey(new_key, sans, ips)


def signCSR(rootCAKey: str, rootCACert: str, csr: str) -> str:
    rootCAKey: rsa.RSAPrivateKey = serialization.load_pem_private_key(
        rootCAKey.encode(), None
    )
    rootCACert: x509.Certificate = x509.load_pem_x509_certificate(
        rootCACert.encode())

    csr: x509.CertificateSigningRequest = x509.load_pem_x509_csr(csr.encode())

    certBuilder = (
        x509.CertificateBuilder().subject_name(
            csr.subject
        ).issuer_name(
            rootCACert.subject
        ).public_key(
            csr.public_key()
        ).serial_number(
            x509.random_serial_number()
        ).not_valid_before(
            datetime.datetime.utcnow()
        )
        # Our certificate will be valid for 1 year
        .not_valid_after(
            datetime.datetime.utcnow() + datetime.timedelta(days=365)
        )
    )

    # Copy extensions from CSR to the certificate
    for ext in csr.extensions:
        certBuilder = certBuilder.add_extension(
            ext.value, critical=ext.critical)

    # Sign our certificate with our private key
    cert = certBuilder.sign(rootCAKey, hashes.SHA256())

    return cert.public_bytes(serialization.Encoding.PEM).decode()