X-Git-Url: https://git.ralfj.de/zonemaker.git/blobdiff_plain/8afd5eb407397e88da8f7d2151088d14abbe51e4..d1f1548719487b0107be8befe090483401abbe52:/zonemaker/zone.py?ds=sidebyside diff --git a/zonemaker/zone.py b/zonemaker/zone.py index 5e83854..e5e6641 100644 --- a/zonemaker/zone.py +++ b/zonemaker/zone.py @@ -1,14 +1,41 @@ -import ipaddress +import re +from ipaddress import IPv4Address, IPv6Address +from typing import List, Dict, Any, Iterator + second = 1 minute = 60*second hour = 60*minute day = 24*hour +week = 7*day + +def check_hostname(name: str) -> str: + # check hostname for validity + label = r'[a-zA-Z90-9]([a-zA-Z90-9-]{0,61}[a-zA-Z90-9])?' # must not start or end with hyphen + pattern = r'^{0}(\.{0})*\.?'.format(label) + if re.match(pattern, name): + return name + raise Exception(name+" is not a valid hostname") + +def time(time: int) -> str: + if time == 0: + return "0" + elif time % week == 0: + return str(time//week)+"w" + elif time % day == 0: + return str(time//day)+"d" + elif time % hour == 0: + return str(time//hour)+"h" + elif time % minute == 0: + return str(time//minute)+"m" + else: + return str(time) class Address: - def __init__(self, IPv4 = None, IPv6 = None): - self._IPv4 = None if IPv4 is None else ipaddress.IPv4Address(IPv4) - self._IPv6 = None if IPv6 is None else ipaddress.IPv6Address(IPv6) + # mypy does not know about the ipaddress types, so leave this class unannotated for now + def __init__(self, IPv4 = None, IPv6 = None) -> None: + self._IPv4 = None if IPv4 is None else IPv4Address(IPv4) + self._IPv6 = None if IPv6 is None else IPv6Address(IPv6) def IPv4(self): return Address(IPv4 = self._IPv4) @@ -17,31 +44,104 @@ class Address: return Address(IPv6 = self._IPv6) class Name: - def __init__(self, address = None, MX = None, TCP = None, UDP = None): + def __init__(self, address: Address = None, MX: List = None, + TCP: Dict[int, Any] = None, UDP: Dict[int, Any] = None) -> None: self._address = address class Service: - def __init__(self, SRV = None, TLSA=None): - self._SRV = SRV + def __init__(self, SRV: str = None, TLSA: str=None) -> None: + self._SRV = None if SRV is None else check_hostname(SRV) self._TLSA = TLSA class CName: - def __init__(self, name): - self._name = name + def __init__(self, name: str) -> None: + self._name = check_hostname(name) class Delegation(): - def __init__(self, NS, DS = None): + def __init__(self, NS: str, DS: str = None) -> None: pass class Zone: - def __init__(self, name, mail, NS, - secondary_refresh, secondary_retry, secondary_discard, - NX_TTL = None, A_TTL = None, other_TTL = None, - domains = []): - assert other_TTL is not None - self._NX_TTL = other_TTL if NX_TTL is None else NX_TTL - self._A_TTL = other_TTL if A_TTL is None else A_TTL + def __init__(self, name: str, serialfile: str, dbfile: str, mail: str, NS: List[str], + secondary_refresh: int, secondary_retry: int, secondary_expire: int, + NX_TTL: int = None, A_TTL: int = None, other_TTL: int = None, + domains: Dict[str, Any] = {}) -> None: + self._serialfile = serialfile + self._dbfile = dbfile + + if not name.endswith('.'): raise Exception("Expected an absolute hostname") + self._name = check_hostname(name) + if not mail.endswith('.'): raise Exception("Mail must be absolute, end with a dot") + atpos = mail.find('@') + if atpos < 0 or atpos > mail.find('.'): raise Exception("Mail must contain an @ before the first dot") + self._mail = check_hostname(mail.replace('@', '.', 1)) + self._NS = list(map(check_hostname, NS)) + + self._refresh = secondary_refresh + self._retry = secondary_retry + self._expire = secondary_expire + + if other_TTL is None: raise Exception("Must give other_TTL") + self._NX_TTL = NX_TTL + self._A_TTL = A_TTL self._other_TTL = other_TTL + + self._domains = domains + + def RR(self, owner: str, recordType: str, data: str) -> str: + '''generate given RR, in textual representation''' + assert re.match(r'^[A-Z]+$', recordType), "got invalid record type" + # figure out TTL + attrname = "_"+recordType+"_TTL" + TTL = None # type: int + if hasattr(self, attrname): + TTL = getattr(self, attrname) + if TTL is None: + TTL = self._other_TTL + # be done + return "{0}\t{1}\t{2}\t{3}".format(self.abs_hostname(owner), TTL, recordType, data) + + def abs_hostname(self, name): + if name.endswith('.'): + return name + return name+"."+self._name + + def inc_serial(self) -> int: + # get serial + cur_serial = 0 + try: + with open(self._serialfile) as f: + cur_serial = int(f.read()) + except FileNotFoundError: + pass + # increment serial + cur_serial += 1 + # save serial + with open(self._serialfile, 'w') as f: + f.write(str(cur_serial)) + # be done + return cur_serial + + def generate_rrs(self) -> Iterator: + # SOA record + serial = self.inc_serial() + yield self.RR(self._name, 'SOA', + ('{NS} {mail} ({serial} {refresh} {retry} {expire} {NX_TTL}) ; '+ + '(serial refresh retry expire NX_TTL)').format( + NS=self.abs_hostname(self._NS[0]), mail=self._mail, serial=serial, + refresh=time(self._refresh), retry=time(self._retry), expire=time(self._expire), + NX_TTL=time(self._NX_TTL)) + ) + # NS records + for ns in self._NS: + yield self.RR(self._name, 'NS', self.abs_hostname(ns)) + + # all the rest + for name in sorted(self._domains.keys(), key=lambda s: list(reversed(s.split('.')))): + print(name) + #for rr in self._domains[name].generate_rrs(self): + #yield rr - def write(self, file): - raise NotImplementedError() + def write(self) -> None: + for rr in self.generate_rrs(): + print(rr)