forked from stove/risotto
163 lines
4.7 KiB
Python
163 lines
4.7 KiB
Python
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]
|
|
|
|
# =============================================================
|