gen_self_signed_cert.py

import os
import getpass
import datetime

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


CA_CERT_FILE = 'cacert.crt'
CA_KEY_FILE  = 'cakey.pem'

NEW_CERT_FILE = 'newcert.crt'
NEW_KEY_FILE  = 'newkey.pem'
    
def get_oids():
    # list of fields to configure the CA cert with, and their
    # respective cryptography module name mapping
    fields = {
        'C'  : 'COUNTRY_NAME',        
        'S'  : 'STATE_OR_PROVINCE_NAME',
        'L'  : 'LOCALITY_NAME',
        'O'  : 'ORGANIZATION_NAME',
        'OU' : 'ORGANIZATIONAL_UNIT_NAME',
        'CN' : 'COMMON_NAME',
    }    
    flen = max(map(len, fields.values()))
    while True:
        entries = []
        for field in fields:
            while True:
                entry = input(f'{fields[field]:>{flen}} ({field}) : ')
                
                # skip blank entries
                if not entry:
                    break
                    
                try:
                    entries.append(
                        x509.NameAttribute(
                            getattr(NameOID, fields[field]),
                            entry,
                    ))
                    break
                except ValueError as e:
                    print(''.join(e.args))
        
        if 'n' not in input('Proceed with these inputs? [y]n > '):
            break
    
    return entries


def generate_CA_cert(*, days=825):
    """Generates a self-signed CA certificate"""
    pw = getpass.getpass('Enter CA private key passphrase:').encode()
    key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=2048,
    )
    
    with open(CA_KEY_FILE, 'xb') as f:
        f.write(key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.BestAvailableEncryption(pw),
        ))
    print('Saved CA key to', CA_KEY_FILE)
    
    # self-signed means the subject is the issuer
    subject = issuer = x509.Name(get_oids())
    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 about 10 years
        datetime.datetime.utcnow() + datetime.timedelta(days=days)
    ).add_extension(
        x509.BasicConstraints(
            ca=True, 
            path_length=None
        ), 
        critical=True,
    ).add_extension(
        x509.KeyUsage(
            digital_signature=True,
            content_commitment=False,
            key_encipherment=False,
            data_encipherment=False,
            key_agreement=False,
            key_cert_sign=True,
            crl_sign=True,
            encipher_only=False,
            decipher_only=False
        ),
        critical=True,
    ).sign(key, hashes.SHA256())
    
    # Write our new certificate out to disk.
    with open(CA_CERT_FILE, 'xb') as f:
        f.write(cert.public_bytes(serialization.Encoding.PEM))
    print('Saved CA cert to', CA_CERT_FILE)
    
    print('Done.')


def generate_signed_cert(*, days=825, authority_key='cakey.pem', authority_cert='cacert.crt'):
    """Generates a certificate signed by a CA certificate. The resultant
    can either be an intermediate CA or a host-level certificate.
    """
    pw = getpass.getpass('Enter new private key passphrase:').encode()
    key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=2048,
    )
    
    with open(NEW_KEY_FILE, 'xb') as f:
        f.write(key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.BestAvailableEncryption(pw),
        ))
    print('Saved to', NEW_KEY_FILE)
    
    subject = x509.Name(get_oids())
    
    # Create a CSR and allow user to save it for other purposes.
    csr = x509.CertificateSigningRequestBuilder().subject_name(subject)
    
    sans = []
    if 'y' in input('Add any Subject Alternative Name (SAN) fields? y/[n]: '):
        print('Enter nothing to end the list')
        while (san := input('Enter a DNS Name: ')):
            sans.append(san)
    
    if sans:
        sans = [x509.DNSName(s) for s in sans]
        csr = csr.add_extension(
            x509.SubjectAlternativeName(sans),
            critical=False
        )
    
    ans = input('Is this an intermediate certificate? (y/[n]) > ')
    if 'y' in ans:
        csr = csr.add_extension(
            x509.BasicConstraints(
                ca=True, 
                path_length=None
            ), 
            critical=True,
        ).add_extension(
            x509.KeyUsage(
                digital_signature=True,
                content_commitment=False,
                key_encipherment=False,
                data_encipherment=False,
                key_agreement=False,
                key_cert_sign=True,
                crl_sign=True,
                encipher_only=False,
                decipher_only=False
            ),
            critical=True
        )
    
    else:
        csr = csr.add_extension(
            x509.BasicConstraints(
                ca=False, 
                path_length=None
            ), 
            critical=True,
        ).add_extension(
            x509.KeyUsage(
                digital_signature=True,
                content_commitment=False,
                key_encipherment=True,
                data_encipherment=False,
                key_agreement=False,
                key_cert_sign=False,
                crl_sign=False,
                encipher_only=False,
                decipher_only=False
            ),
            critical=True,
        )
    
    
    # sign it with the new key
    csr = csr.sign(key, hashes.SHA256())

    print(
        'If you want to save the Certificate Signing '
        'Request, enter a file name:', end=' ')
    if (csr_file := input()):
        if not csr_file.endswith('.csr'):
            csr_file += '.csr'
        with open(csr_file, 'xb') as f:
            f.write(csr.public_bytes(serialization.Encoding.PEM))
        print('Saved to', csr_file)
    else:
        print('CSR not saved.')

    # Next, open the CA files to sign the certificate
    pw = getpass.getpass('Enter password for the authority key: ').encode()
    with open(authority_key, 'rb') as f:
        authority_key = serialization.load_pem_private_key(
            f.read(),
            password=pw,
        )
    
    print('OK')
    
    with open(authority_cert, 'rb') as f:
        authority_cert = x509.load_pem_x509_certificate(f.read())
    
    # Make the actual certificate.
    cert = x509.CertificateBuilder()\
    .subject_name(
        csr.subject
    ).issuer_name(
        authority_cert.subject
    ).public_key(
        csr.public_key()
    ).serial_number(
        x509.random_serial_number()
    ).not_valid_before(
        datetime.datetime.utcnow()
    ).not_valid_after(
        datetime.datetime.utcnow() + datetime.timedelta(days=days)
    )
    
    for ext in csr.extensions._extensions:
        cert = cert.add_extension(ext.value, ext.critical)
        
    # finally, sign the new certificate with the CA cert.
    cert = cert.sign(authority_key, hashes.SHA256())
    
    # Write our new certificate out to disk.
    with open(NEW_CERT_FILE, 'xb') as f:
        f.write(cert.public_bytes(serialization.Encoding.PEM))
    print('Saved certificate to', NEW_CERT_FILE)


def start_ssl_server(cert_chain, keyfile):
    import ssl, asyncio
    
    print('Starting SSL server...')
    
    ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    print('If the server key is password protected, enter the password now:')
    
    # If password kwarg not supplied and key has a password, it will prompt for one
    ctx.load_cert_chain(
        certfile=cert_chain, 
        keyfile=keyfile,
    )
    
    async def serve_webpage(reader, writer):
        try:
            print(
                await asyncio.wait_for(
                    reader.read(1024), 3
            ))
        except:
            writer.close()
            await writer.wait_closed()
            return
        
        html = '<html><body><h1>it works!</h1></html>'.encode()
        http = (
            b'HTTP/1.1 200 OK\r\n'
            b'Connection: close\r\n'
            b'Content-Type: text/html\r\n'
            b'Content-Length:'
        )
        http += str(len(html)).encode() + b'\r\n\r\n' + html
        
        writer.write(http)
        await writer.drain()
        writer.close()
        await writer.wait_closed()
        
    
    async def amain():
        host = 'localhost', 0
        server = await asyncio.start_server(
            serve_webpage,
            *host,
            ssl=ctx,
        )
        print('Serving on:')
        for sock in server.sockets:
            host = sock.getsockname()
            if len(host) == 2:
                url = 'https://{}:{}'.format(*host)
            if len(host) == 4:
                url = 'https://[{}]:{}'.format(*host[:2], host[3])
            print(url)
        print(f'https://localhost:{host[1]}')
        
        await server.serve_forever()
    
    try:
        asyncio.run(amain())
    except KeyboardInterrupt:
        pass
    
def append_cert(base, append, out):
    with open(base, 'rb') as f1:
        with open(append, 'rb') as f2:
            with open(out, 'xb') as w:
                w.write(f1.read())
                w.write(f2.read())


def test():    
    generate_CA_cert(days=99999)
    generate_signed_cert(days=99999)
    append_cert('newcert.crt', 'cacert.crt', 'serverchain.crt')
    os.rename('newcert.crt', 'server.crt')
    os.rename('newkey.pem', 'serverkey.pem')
    start_ssl_server('serverchain.crt', 'serverkey.pem')


if __name__ == '__main__':
    test()