risotto/funcs.py

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, ZONES_SERVER
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_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 ZONES_SERVER['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]


def get_internal_info_in_zone(zone: str,
                              auto: bool,
                              type: str,
                              index: int=None,
                              ) -> List[str]:
    if not auto:
        return
    for domain_name, domain in DOMAINS.items():
        if zone == domain_name:
            if type == 'host':
                return list(domain[0])
            else:
                return domain[1][index]

# =============================================================