From f79e486371ec29468c2e1a5a86a8b246eedd984a Mon Sep 17 00:00:00 2001 From: Emmanuel Garette Date: Wed, 21 Dec 2022 16:25:38 +0100 Subject: [PATCH] remove old scripts --- bootstrap.py | 107 ------------------ funcs.py | 163 ---------------------------- src/risotto/x509.py | 258 -------------------------------------------- 3 files changed, 528 deletions(-) delete mode 100755 bootstrap.py delete mode 100644 funcs.py delete mode 100644 src/risotto/x509.py diff --git a/bootstrap.py b/bootstrap.py deleted file mode 100755 index b31d6d3..0000000 --- a/bootstrap.py +++ /dev/null @@ -1,107 +0,0 @@ -#!/usr/bin/env python3 - -from asyncio import run -from os import listdir, link, makedirs -from os.path import isdir, join -from shutil import rmtree -from copy import copy - -from risotto.utils import RISOTTO_CONFIG, SERVERS -#from risotto.image import load -from risotto.machine import templates, load, ROUGAIL_NAMESPACE -from rougail.utils import normalize_family - - -INSTALL_DIR = RISOTTO_CONFIG['directories']['dest'] -CONFIG_DEST_DIR = 'configurations' -CONFIG_DIFF_DIR = 'diff' -CONFIG_ORI_DIR = 'ori' -SRV_DEST_DIR = 'srv' - - -async def main(): - if isdir(INSTALL_DIR): - rmtree(INSTALL_DIR) - makedirs(INSTALL_DIR) - try: - module_infos, rougailconfig, config = await load('a.py', - 'n.py', - clean_directories=True, - copy_manual_dir=True, - copy_tests=True, - ) - except Exception as err: -# import traceback -# traceback.print_exc() - print(err) - exit(1) - modules_done = [] - for server_name in SERVERS: - module_name = SERVERS[server_name]['module'] - module_info = module_infos[module_name] - subconfig = config.option(normalize_family(server_name)) - try: - add_srv = await subconfig.option('machine.add_srv').value.get() - except AttributeError: - add_srv = False - rougailconfig['tmp_dir'] = 'tmp' - rougailconfig['destinations_dir'] = join(INSTALL_DIR, module_name, CONFIG_DEST_DIR, server_name) - rougailconfig['templates_dir'] = module_info['infos'].templates_dir - if module_name == 'host': - tmpfile = await subconfig.option(f'{ROUGAIL_NAMESPACE}.host_install_dir').value.get() - rougailconfig['tmpfile_dest_dir'] = f'{tmpfile}/host/configurations/{server_name}' - rougailconfig['default_systemd_directory'] = '/usr/local/lib/systemd' - else: - rougailconfig['tmpfile_dest_dir'] = '/usr/local/lib' - rougailconfig['default_systemd_directory'] = '/systemd' -# cfg['templates_dir'] = module_info['infos'].templates_dir - if isdir('tmp'): - rmtree('tmp') - makedirs(rougailconfig['tmp_dir']) - makedirs(rougailconfig['destinations_dir']) - if add_srv: - srv = join(INSTALL_DIR, SRV_DEST_DIR, server_name) - else: - srv = None - await templates(server_name, - subconfig, - rougailconfig, - srv=srv, - ) - # - await config.property.read_write() - try: - await subconfig.option('general.hide_secret').value.set(True) - except AttributeError as err: - # print(err) - pass - await config.property.read_only() - rougailconfig['destinations_dir'] = join(INSTALL_DIR, module_name, CONFIG_DIFF_DIR, server_name) - rmtree('tmp') - makedirs(rougailconfig['tmp_dir']) - makedirs(rougailconfig['destinations_dir']) - await templates(server_name, - subconfig, - rougailconfig, - ) - await config.property.read_write() - try: - await subconfig.option('general.hide_secret').value.set(False) - except AttributeError as err: - pass - await config.property.read_only() - # - if module_name not in modules_done: - rougailconfig['destinations_dir'] = join(INSTALL_DIR, module_name, CONFIG_ORI_DIR) - rmtree('tmp') - makedirs(rougailconfig['tmp_dir']) - makedirs(rougailconfig['destinations_dir']) - await templates(server_name, - subconfig, - rougailconfig, - just_copy=True, - ) - modules_done.append(module_name) - with open(join(INSTALL_DIR, module_name, 'install_machines'), 'w') as fh: - fh.write(f'./install_machine {module_name} {server_name}\n') -run(main()) diff --git a/funcs.py b/funcs.py deleted file mode 100644 index 97dc833..0000000 --- a/funcs.py +++ /dev/null @@ -1,163 +0,0 @@ -from tiramisu import valid_network_netmask, valid_ip_netmask, valid_broadcast, valid_in_network, valid_not_equal as valid_differ, valid_not_equal, calc_value -from os.path import dirname, abspath, join as _join, isdir as _isdir, isfile as _isfile -from typing import List -from secrets import token_urlsafe as _token_urlsafe - -from rougail.utils import normalize_family - -from risotto.utils import multi_function, DOMAINS, ZONES, load_zones, load_zones_server, load_domains, SERVERS_JSON -from risotto.x509 import gen_cert as _x509_gen_cert, gen_ca as _x509_gen_ca, gen_pub as _x509_gen_pub, has_pub as _x509_has_pub - - -HERE = dirname(abspath(__file__)) - - -@multi_function -def get_chain(authority_cn: str, - authority_name: str, - hide: bool, - ): - if hide: - return "XXXXX" - if not authority_cn or not authority_name or authority_name is None: - if isinstance(authority_name, list): - return [] - return - if not isinstance(authority_cn, list): - is_list = False - authority_cn = [authority_cn] - else: - is_list = True - authorities = [] - - for auth_cn in authority_cn: - ret = _x509_gen_ca(auth_cn, - authority_name, - HERE, - ) - if not is_list: - return ret - authorities.append(ret) - return authorities - - -@multi_function -def get_certificate(cn, - authority_name: str, - hide: bool, - authority_cn: str=None, - extra_domainnames: list=[], - type: str='server', - ): - if hide: - return "XXXXX" - if isinstance(cn, list) and extra_domainnames: - raise Exception('cn cannot be a list with extra_domainnames set') - if not cn or authority_name is None: - if isinstance(cn, list): - return [] - return - return _x509_gen_cert(cn, - extra_domainnames, - authority_cn, - authority_name, - type, - 'crt', - HERE, - ) - - -@multi_function -def get_private_key(cn: str, - hide: bool, - authority_name: str=None, - authority_cn: str=None, - type: str='server', - ): - if hide: - return "XXXXX" - if not cn: - if isinstance(cn, list): - return [] - return - if authority_name is None: - if _x509_has_pub(cn, HERE): - return _x509_gen_pub(cn, - 'key', - HERE, - ) - if isinstance(cn, list): - return [] - return - return _x509_gen_cert(cn, - [], - authority_cn, - authority_name, - type, - 'key', - HERE, - ) - - -def get_public_key(cn: str, - hide: bool, - ): - if hide: - return "XXXXX" - if not cn: - return - return _x509_gen_pub(cn, - 'pub', - HERE, - ) - - -def zone_information(zone_name: str, - type: str, - multi: bool=False, - index: int=None, - ) -> str: - if not zone_name: - return - if type == 'gateway' and index != 0: - return - load_zones() - if zone_name not in ZONES: - raise ValueError(f"cannot get zone informations in unknown zone '{zone_name}'") - zone = ZONES[zone_name] - if type not in zone: - raise ValueError(f"unknown type '{type}' in zone '{zone_name}'") - value = zone[type] - if multi: - value = [value] - return value - - -def get_internal_zones() -> List[str]: - load_domains() - return list(DOMAINS.keys()) - - -@multi_function -def get_zones_info(type: str) -> str: - load_zones_server() - ret = [] - for data in SERVERS_JSON['zones'].values(): - ret.append(data[type]) - return ret - -@multi_function -def get_internal_zone_names() -> List[str]: - load_zones() - return list(ZONES.keys()) - - -def get_internal_zone_information(zone: str, - info: str, - ) -> str: - load_domains() - if info == 'cidr': - return ZONES[zone]['gateway'] + '/' + ZONES[zone]['network'].split('/')[-1] - return ZONES[zone][info] - -# ============================================================= diff --git a/src/risotto/x509.py b/src/risotto/x509.py deleted file mode 100644 index 4311c7b..0000000 --- a/src/risotto/x509.py +++ /dev/null @@ -1,258 +0,0 @@ -from OpenSSL.crypto import load_certificate, load_privatekey, dump_certificate, dump_privatekey, dump_publickey, PKey, X509, X509Extension, TYPE_RSA, FILETYPE_PEM -from os import makedirs, symlink -from os.path import join, isdir, isfile, exists -#from shutil import rmtree -from datetime import datetime - - -PKI_DIR = 'pki/x509' -#FIXME -EMAIL = 'gnunux@gnunux.info' -COUNTRY = 'FR' -LOCALITY = 'Dijon' -STATE = 'France' -ORG_NAME = 'Cadoles' -ORG_UNIT_NAME = 'CSS' - - -def _gen_key_pair(): - key = PKey() - key.generate_key(TYPE_RSA, 4096) - return key - - -def _gen_cert(is_ca, - common_names, - serial_number, - validity_end_in_seconds, - key_file, - cert_file, - type=None, - ca_cert=None, - ca_key=None, - email_address=None, - country_name=None, - locality_name=None, - state_or_province_name=None, - organization_name=None, - organization_unit_name=None, - ): - #can look at generated file using openssl: - #openssl x509 -inform pem -in selfsigned.crt -noout -text - # create a key pair - if isfile(key_file): - with open(key_file) as fh: - filecontent = bytes(fh.read(), 'utf-8') - key = load_privatekey(FILETYPE_PEM, filecontent) - else: - key = _gen_key_pair() - cert = X509() - cert.set_version(2) - cert.get_subject().C = country_name - cert.get_subject().ST = state_or_province_name - cert.get_subject().L = locality_name - cert.get_subject().O = organization_name - cert.get_subject().OU = organization_unit_name - cert.get_subject().CN = common_names[0] - cert.get_subject().emailAddress = email_address - cert_ext = [] - if not is_ca: - cert_ext.append(X509Extension(b'basicConstraints', False, b'CA:FALSE')) - cert_ext.append(X509Extension(b'keyUsage', True, b'digitalSignature, keyEncipherment')) - cert_ext.append(X509Extension(b'subjectAltName', False, ", ".join([f'DNS:{common_name}' for common_name in common_names]).encode('ascii'))) - if type == 'server': - cert_ext.append(X509Extension(b'extendedKeyUsage', True, b'serverAuth')) - else: - cert_ext.append(X509Extension(b'extendedKeyUsage', True, b'clientAuth')) - else: - cert_ext.append(X509Extension(b'basicConstraints', False, b'CA:TRUE')) - cert_ext.append(X509Extension(b"keyUsage", True, b'keyCertSign, cRLSign')) - cert_ext.append(X509Extension(b'subjectAltName', False, f'email:{email_address}'.encode())) - cert_ext.append(X509Extension(b'subjectKeyIdentifier', False, b"hash", subject=cert)) - cert.add_extensions(cert_ext) - cert.set_serial_number(serial_number) - cert.gmtime_adj_notBefore(0) - cert.gmtime_adj_notAfter(validity_end_in_seconds) - if is_ca: - ca_cert = cert - ca_key = key - else: - with open(ca_cert) as fh: - filecontent = bytes(fh.read(), 'utf-8') - ca_cert = load_certificate(FILETYPE_PEM, filecontent) - with open(ca_key) as fh: - filecontent = bytes(fh.read(), 'utf-8') - ca_key = load_privatekey(FILETYPE_PEM, filecontent) - cert.set_issuer(ca_cert.get_subject()) - cert.add_extensions([X509Extension(b"authorityKeyIdentifier", False, b'keyid:always', issuer=ca_cert)]) - cert.set_pubkey(key) - cert.sign(ca_key, "sha512") - - with open(cert_file, "wt") as f: - f.write(dump_certificate(FILETYPE_PEM, cert).decode("utf-8")) - if not is_ca: - f.write(dump_certificate(FILETYPE_PEM, ca_cert).decode("utf-8")) - with open(key_file, "wt") as f: - f.write(dump_privatekey(FILETYPE_PEM, key).decode("utf-8")) - - -def gen_ca(authority_dns, - authority_name, - base_dir, - ): - authority_cn = authority_name + '+' + authority_dns - week_number = datetime.now().isocalendar().week - root_dir_name = join(base_dir, PKI_DIR, authority_cn) - ca_dir_name = join(root_dir_name, 'ca') - sn_ca_name = join(ca_dir_name, 'serial_number') - key_ca_name = join(ca_dir_name, 'private.key') - cert_ca_name = join(ca_dir_name, f'certificate_{week_number}.crt') - if not isfile(cert_ca_name): - if not isdir(ca_dir_name): - # rmtree(ca_dir_name) - makedirs(ca_dir_name) - if isfile(sn_ca_name): - with open(sn_ca_name, 'r') as fh: - serial_number = int(fh.read().strip()) + 1 - else: - serial_number = 0 - _gen_cert(True, - [authority_cn], - serial_number, - 10*24*60*60, - key_ca_name, - cert_ca_name, - email_address=EMAIL, - country_name=COUNTRY, - locality_name=LOCALITY, - state_or_province_name=STATE, - organization_name=ORG_NAME, - organization_unit_name=ORG_UNIT_NAME, - ) - with open(sn_ca_name, 'w') as fh: - fh.write(str(serial_number)) - with open(cert_ca_name, 'r') as fh: - return fh.read().strip() - - -def gen_cert_iter(cn, - extra_domainnames, - authority_cn, - authority_name, - type, - base_dir, - dir_name, - ): - week_number = datetime.now().isocalendar().week - root_dir_name = join(base_dir, PKI_DIR, authority_cn) - ca_dir_name = join(root_dir_name, 'ca') - key_ca_name = join(ca_dir_name, 'private.key') - cert_ca_name = join(ca_dir_name, f'certificate_{week_number}.crt') - sn_name = join(dir_name, f'serial_number') - key_name = join(dir_name, f'private.key') - cert_name = join(dir_name, f'certificate_{week_number}.crt') - if not isfile(cert_ca_name): - raise Exception(f'cannot find CA file "{cert_ca_name}"') - if not isfile(cert_name): - if not isfile(key_ca_name): - raise Exception(f"cannot find CA private key (\"{authority_cn}\") to sign certificat for \"{cn}\" ({key_ca_name}), is it a Let's Encrypt certification?") - if not isdir(dir_name): - makedirs(dir_name) - if isfile(sn_name): - with open(sn_name, 'r') as fh: - serial_number = int(fh.read().strip()) + 1 - else: - serial_number = 0 - common_names = [cn] - common_names.extend(extra_domainnames) - _gen_cert(False, - common_names, - serial_number, - 10*24*60*60, - key_name, - cert_name, - ca_cert=cert_ca_name, - ca_key=key_ca_name, - type=type, - email_address=EMAIL, - country_name=COUNTRY, - locality_name=LOCALITY, - state_or_province_name=STATE, - organization_name=ORG_NAME, - organization_unit_name=ORG_UNIT_NAME, - ) - with open(sn_name, 'w') as fh: - fh.write(str(serial_number)) - for extra in extra_domainnames: - extra_dir_name = join(base_dir, PKI_DIR, authority_name + '+' + extra) - if not exists(extra_dir_name): - symlink(root_dir_name, extra_dir_name) - for extra in extra_domainnames: - extra_dir_name = join(base_dir, PKI_DIR, authority_name + '+' + extra) - if not exists(extra_dir_name): - raise Exception(f'file {extra_dir_name} not already exists that means subjectAltName is not set in certificat, please remove {cert_name}') - return cert_name - - -def gen_cert(cn, - extra_domainnames, - authority_cn, - authority_name, - type, - file_type, - base_dir, - ): - if '.' in authority_name: - raise Exception(f'dot is not allowed in authority_name "{authority_name}"') - if type == 'server' and authority_cn is None: - authority_cn = cn - if authority_cn is None: - raise Exception(f'authority_cn is mandatory when authority type is client') - if extra_domainnames is None: - extra_domainnames = [] - auth_cn = authority_name + '+' + authority_cn - dir_name = join(base_dir, PKI_DIR, auth_cn, 'certificats', cn, type) - if file_type == 'crt': - filename = gen_cert_iter(cn, - extra_domainnames, - auth_cn, - authority_name, - type, - base_dir, - dir_name, - ) - else: - filename = join(dir_name, f'private.key') - with open(filename, 'r') as fh: - return fh.read().strip() - - -def has_pub(cn, - base_dir, - ): - dir_name = join(base_dir, PKI_DIR, 'public', cn) - cert_name = join(dir_name, f'public.pub') - return isfile(cert_name) - - -def gen_pub(cn, - file_type, - base_dir, - ): - dir_name = join(base_dir, PKI_DIR, 'public', cn) - key_name = join(dir_name, f'private.key') - if file_type == 'pub': - pub_name = join(dir_name, f'public.pub') - if not isfile(pub_name): - if not isdir(dir_name): - makedirs(dir_name) - key = _gen_key_pair() - with open(pub_name, "wt") as f: - f.write(dump_publickey(FILETYPE_PEM, key).decode("utf-8")) - with open(key_name, "wt") as f: - f.write(dump_privatekey(FILETYPE_PEM, key).decode("utf-8")) - filename = pub_name - else: - filename = key_name - with open(filename, 'r') as fh: - return fh.read().strip()