Convert tinydns zone files to bind

bindify.py

#!/usr/bin/env python
import argparse
import binascii
import collections
import logging
from os import path
import re
from time import time

logger = logging.getLogger(__name__)
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.INFO)

# record types
A = collections.namedtuple('A', ['address', 'hostname', 'ttl'])
AAAA = collections.namedtuple('AAAA', ['address', 'hostname', 'ttl'])
CNAME = collections.namedtuple('CNAME', ['alias', 'canonical'])
COMMENT = collections.namedtuple('COMMENT', ['value'])
MX = collections.namedtuple('MX', ['hostname', 'priority', 'mailhost'])
NS = collections.namedtuple('NS', ['hostname'])
SOA = collections.namedtuple('SOA', ['ns', 'hostmaster', 'serial', 'refresh', 'retry', 'expire', 'minimum', 'ttl'])
SOA.__new__.__defaults__ = (None, None, int(time()), 16384, 2048, 1048576, 2560, 3600)
PTR = collections.namedtuple('PTR', ['address', 'fqdn'])
TXT = collections.namedtuple('TXT', ['hostname', 'value'])
ZONE = collections.namedtuple('ZONE', ['filename', 'type'])

SOA_TEMPLATE = """
$TTL {7}
$ORIGIN {8}.
@  IN  SOA {0}. {1}.     (
            {2}   ; serial
            {3}   ; refresh
            {4}   ; retry
            {5}   ; expire
            {6}   ; minimum
)
"""


class Domain(object):
    name = None


def decode_payload(orig):
    try:
        return orig.strip().decode('string-escape')
    except:
        return orig.strip().encode('utf-8').decode('unicode-escape')


def parse_tinydns_data(infile):
    zones = {}
    comments = []
    last_domain = Domain()

    def new_domain(domain):
        if domain:
            zones.setdefault(domain, [])
            last_domain.name = re.sub(r'^\.', '', domain)
        if comments:
            while comments[::1]:
                zones[last_domain.name].append(COMMENT(comments.pop()))

    with open(infile) as handle:
        for line in handle:
            if len(line):
                parts = line[1:].split(':')
                parts = [decode_payload(p) for p in parts]
                if re.match(r"#\s*[Z=\.\+C\^@&\-:']", line):
                    continue
                elif line[0] == 'Z':
                    new_domain(parts[0])
                    zones[last_domain.name].append(SOA(*parts[1:]))
                elif line[0] == '.':
                    new_domain(parts[0])
                    parts[2] += '.ns.' + last_domain.name if '.' not in parts[2] else ''
                    zones[last_domain.name].append(SOA(parts[2], 'hostmaster.' + last_domain.name))
                    if parts[2].endswith(last_domain.name):  # skip if DNS outside domain
                        zones[last_domain.name].append(NS(parts[2]))
                        zones[last_domain.name].append(A(*parts[1:3]))
                elif line[0] == '&':
                    new_domain(parts[0])
                    parts[2] += '.ns.' + last_domain.name if '.' not in parts[2] else ''
                    zones[last_domain.name].append(NS(parts[2]))
                    zones[last_domain.name].append(A(*parts[1:3]))
                elif line[0] == '#':
                    comments.append(line[1:].strip())
                elif line[0] == '=':
                    new_domain(parts[0])
                    hostname = '.'.join(parts[0].split('.')[0:-2]) or '@'
                    try:
                        ttl = parts[2]
                    except IndexError:
                        ttl = ''
                    zones[last_domain.name].append(A(parts[1], hostname, ttl))
                    zones[last_domain.name].append(PTR(parts[1], parts[0]))
                elif line[0] == '+':
                    new_domain(None)
                    if parts[0] in zones.keys():
                        hostname = '@'
                    else:
                        hostname = parts[0].split('.')[0] or '@'
                    try:
                        ttl = parts[2]
                    except IndexError:
                        ttl = ''
                    zones[last_domain.name].append(A(parts[1], hostname, ttl))
                elif line[0] == '^':
                    new_domain(parts[0])
                    zones[last_domain.name].append(PTR(parts[1], parts[0]))
                elif line[0] == 'C':
                    new_domain(None)
                    hostname = '.'.join(parts[0].split('.')[0:-2])
                    zones[last_domain.name].append(CNAME(hostname, parts[1]))
                elif line[0] == '@':
                    new_domain(None)
                    hostname = parts[2]
                    hostname += '.mx' if '.' not in parts[2] else ''
                    if parts[2] and parts[1]:
                        zones[last_domain.name].append(A(parts[1], hostname, ''))
                    host = re.sub('.?' + last_domain.name + r'\.?\s*$', '', parts[0])
                    host = host or '@'
                    zones[last_domain.name].append(MX(host, parts[3] or 10, hostname))
                elif line[0] == "'":
                    new_domain(None)
                    zones[last_domain.name].append(TXT(parts[0], parts[1]))
                elif line[0] == ':':
                    host = re.sub('.?' + last_domain.name + r'\.?\s*$', '', parts[0])
                    host = host or '@'
                    new_domain(None)
                    if int(parts[1]) == 16:
                        zones[last_domain.name].append(TXT(host, parts[2][1:]))
                    elif int(parts[1]) == 28:
                        hostname = '.'.join(parts[0].split('.')[0:-2])
                        try:
                            ttl = parts[3]
                        except IndexError:
                            ttl = ''
                        ipv6l = [binascii.hexlify(ord(c).to_bytes(1, 'big')) for c in parts[2]]
                        ipv6 = ''
                        for i in range(int(len(ipv6l)/2)):
                            ipv6 += ''.join([b.decode('ascii') for b in ipv6l[i*2:i*2+2]]) + ':'
                        ipv6 = re.sub(r':0+', ':', ipv6)
                        ipv6 = re.sub(r'::+', '::', ipv6)
                        zones[last_domain.name].append(AAAA(ipv6.rstrip(':'), hostname, ttl))
                    else:
                        logger.warning('DNS record type {n} unknown. Please add new record.'.format(n=parts[1]))
            elif last_domain.name:  # Preserve Whitespace
                zones[last_domain.name].append(None)
    return zones


def write_zone_files(zonefile_path, zones, ns):
    # Write out the main zone file
    soa = []
    for zone, records in zones.items():
        filename = path.join(zonefile_path, 'db.' + zone)
        with open(filename, 'w') as handle:
            skip_record = False
            for offset, record in enumerate(records):

                # Preserve Whitespace
                if record is None:
                    handle.write('\n')
                    continue

                if skip_record:
                    skip_record = False
                    continue

                record_type = record.__class__.__name__
                if record_type == 'COMMENT':
                    handle.write('; %s\n' % record.value)
                elif record_type == 'SOA' and zone not in soa:
                    soa.append(zone)
                    handle.write(SOA_TEMPLATE.format(*record, zone))
                    if ns:
                        for s in ns:
                            handle.write('{0:<45} IN  NS    {1}.\n'.format('@', s))
                elif record_type in ['A', 'AAAA']:
                    handle.write('{0:<45} {1} IN  {2}     {3}\n'.format(record.hostname, record.ttl, record_type, record.address))
                elif record_type == 'CNAME':
                    handle.write('{0:<45} IN  CNAME {1}.\n'.format(*record))
                elif record_type == 'MX':
                    handle.write('{0:<45} IN  MX    {1:<5} {2}.\n'.format(*record))
                elif record_type == 'NS':
                    handle.write('{0:<45} IN  NS    {1}.\n'.format('@', record[0]))
                elif record_type == 'PTR':
                    octets = record.address.split('.')
                    rev_address = '{}.in-addr.arpa'.format('.'.join(octets[::1]))
                    handle.write('{0:<45} IN  TXT   "{1}"\n'.format(rev_address, record.fqdn))
                elif record_type == 'TXT':
                    handle.write('{0:<45} IN  TXT   "{1}"\n'.format(*record))


if __name__ == '__main__':
    p = argparse.ArgumentParser(prog='bindify.py', description='Converts tinydns data files to BIND zone files')

    p.add_argument('datafile', default='data', help='tinydns data file')
    p.add_argument('--bind-zonefile-path', '-o', default='/etc/bind')
    p.add_argument('--ns', '-n', action='append')

    cmdl = p.parse_args()

    zones = parse_tinydns_data(infile=cmdl.datafile)
    write_zone_files(cmdl.bind_zonefile_path, zones, cmdl.ns)

If I recall correctly, that was the reason I forked.

Can I buy you a coffee/beer?

All good. Plenty of people I have never gotten a beer or coffee for have helped me out in the same way before. :)