certtool.py

#!/usr/bin/env python3

import argparse
import os
import re
import shutil
import sys
from pathlib import Path
from subprocess import PIPE, STDOUT, Popen

from symbolchain.core.Network import NetworkLocator
from symbolchain.core.PrivateKeyStorage import PrivateKeyStorage
from symbolchain.core.symbol.KeyPair import KeyPair
from symbolchain.core.symbol.Network import Network
from zenlog import log


def run_openssl(args, show_output=True):
    command_line = ['openssl'] + args
    all_lines = []
    with Popen(command_line, stdout=PIPE, stderr=STDOUT) as process:
        for line_bin in iter(process.stdout.readline, b''):
            line = line_bin.decode('ascii')
            all_lines.append(line)

            if show_output:
                sys.stdout.write(line)
                sys.stdout.flush()

        process.wait()

    return all_lines


def check_openssl_version():
    version_output = ''.join(run_openssl(['version', '-v'], False))
    match = re.match(r'^OpenSSL +([^ ]*) ', version_output)
    if not match or not match.group(1).startswith('1.1.1'):
        raise RuntimeError(f'{__file__} requires openssl version >=1.1.1')


def get_common_name(default_value, prompt):
    if default_value:
        return default_value

    return input(f'Enter {prompt}: ').strip()


def prepare_ca_config(ca_pem_path, ca_cn):
    with open('ca.cnf', 'wt', encoding='utf8') as output_file:
        output_file.write(f'''[ca]
default_ca = CA_default

[CA_default]
new_certs_dir = ./new_certs

database = index.txt
serial   = serial.dat
private_key = {ca_pem_path}
certificate = ca.crt.pem
policy = policy_catapult

[policy_catapult]
commonName              = supplied

[req]
prompt = no
distinguished_name = dn

[dn]
CN = {ca_cn}
''')

    os.makedirs('new_certs')
    os.chmod('new_certs', 0o700)

    with open('index.txt', 'wt', encoding='utf8') as output_file:
        output_file.write('')


def prepare_node_config(node_cn):
    with open('node.cnf', 'wt', encoding='utf8') as output_file:
        output_file.write(f'''[req]
prompt = no
distinguished_name = dn
[dn]
CN = {node_cn}
''')


def openssl_prepare_keys(ca_path):
    log.info('creating ca.pubkey.pem')
    run_openssl([
        'pkey',
        '-in', ca_path,
        '-out', 'ca.pubkey.pem',
        '-pubout'
    ])

    log.info('creating random node.key.pem')
    run_openssl([
        'genpkey',
        '-out', 'node.key.pem',
        '-outform', 'PEM',
        '-algorithm', 'ed25519'
    ])


def openssl_prepare_certs(ca_path):
    log.info('creating CA certificate')
    run_openssl([
        'req',
        '-config', 'ca.cnf',
        '-keyform', 'PEM',
        '-key', ca_path,
        '-new', '-x509',
        '-days', '7300',
        '-out', 'ca.crt.pem'
    ])

    log.info('preparing node CSR')
    run_openssl([
        'req',
        '-config', 'node.cnf',
        '-key', 'node.key.pem',
        '-new',
        '-out', 'node.csr.pem'
    ])

    log.info('signing node certificate')
    run_openssl([
        'rand',
        '-out', './serial.dat',
        '-hex',
        '19'
    ])
    run_openssl([
        'ca',
        '-config', 'ca.cnf',
        '-days', '375',
        '-notext',
        '-batch',
        '-in', 'node.csr.pem',
        '-out', 'node.crt.pem'
    ])


def prepare_directory(directory_path, force):
    filepath = Path(directory_path)
    if filepath.exists():
        if not force:
            raise FileExistsError(f'output directory ({filepath}) already exists, use --force to overwrite')

        shutil.rmtree(filepath)

    os.makedirs(filepath)
    return filepath


def main():
    parser = argparse.ArgumentParser(description='Cert generation tool', formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument('--working', help='certificates working directory (all files will be created inside)', default='cert')
    parser.add_argument('--package', help='certificates package directory (all required server files will be copied inside)')
    parser.add_argument('--package-mode', help='contents of the package', choices=('all', 'node'), default='all')
    parser.add_argument('--ca', help='path to key PEM file that will be used as a CA key', default='ca.key.pem')
    parser.add_argument('--name-ca', help='use provided name as CA CN (common name) - suggested: account name')
    parser.add_argument('--name-node', help='use provided name as node CN (common name) - suggested: node name, host or ip')
    parser.add_argument('--network', help='network to use when using autogenerated ca names', default='mainnet')
    parser.add_argument('--force', help='overwrite output directory if it already exists', action='store_true')
    args = parser.parse_args()

    check_openssl_version()

    # obtain full paths prior to switching directory
    ca_path = Path(args.ca).absolute()

    if args.package:
        package_path = Path(args.package).absolute()
        prepare_directory(args.package, args.force)

    os.chdir(prepare_directory(args.working, args.force))

    log.info('preparing configuration files')
    if args.name_ca:
        ca_cn = args.name_ca
    elif '.pem' == ca_path.suffix:
        main_private_key = PrivateKeyStorage(ca_path.parent).load(ca_path.stem)
        main_public_key = KeyPair(main_private_key).public_key

        network = NetworkLocator.find_by_name(Network.NETWORKS, args.network)
        main_address = network.public_key_to_address(main_public_key)
        ca_cn = str(main_address)
    else:
        ca_cn = get_common_name(args.name_ca, 'CA common name')

    node_cn = get_common_name(args.name_node, 'node common name')

    log.info(f' *   CA common name: {ca_cn}')
    log.info(f' * Node common name: {node_cn}')

    prepare_ca_config(ca_path, ca_cn)
    prepare_node_config(node_cn)

    openssl_prepare_keys(ca_path)
    openssl_prepare_certs(ca_path)

    log.info(f'certificates generated in {args.working} directory')

    if args.package:
        package_filenames = ['node.crt.pem', 'node.key.pem']
        if 'all' == args.package_mode:
            package_filenames += ['ca.pubkey.pem', 'ca.crt.pem']

        for filename in package_filenames:
            destination_path = package_path / filename
            shutil.copyfile(filename, destination_path)
            os.chmod(destination_path, 0o400)

        log.info(f'certificates packaged in {args.package} directory')


if __name__ == '__main__':
    main()