mirror/github-octodns
1
0
Fork 0
mirror of https://github.com/github/octodns.git synced 2024-05-11 05:55:00 +00:00
Code Issues Releases Activity

TindyDNS rework passing tests

Browse Source
This commit is contained in:
Ross McFarland
2023-06-30 14:59:48 -07:00
parent 4a7df31445
commit ca3a4541a0
3 changed files with 114 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

162
octodns/source/tinydns.py
View File

@@ -3,16 +3,14 @@
#
import logging
import re
import textwrap
from collections import defaultdict
from ipaddress import ip_address
from os import listdir
from os.path import join
from pprint import pprint
from ..record import Record
from ..zone import DuplicateRecordException, SubzoneRecordException
from ..zone import SubzoneRecordException
from .base import BaseSource
@@ -28,9 +26,15 @@ class TinyDnsBaseSource(BaseSource):
def _records_for_at(self, zone, name, lines, arpa=False):
# @fqdn:ip:x:dist:ttl:timestamp:lo
# MX (and optional A)
if arpa:
# no arpa
return []
if not zone.owns('MX', name):
# if name doesn't live under our zone there's nothing for us to do
return
# see if we can find a ttl on any of the lines, first one wins
ttl = self.default_ttl
for line in lines:
@@ -46,7 +50,7 @@ class TinyDnsBaseSource(BaseSource):
# if there's a . in the mx we hit a special case and use it as-is
if '.' not in mx:
# otherwise we treat it as the MX hostnam and construct the rest
mx = f'{mx}.ns.{zone.name}'
mx = f'{mx}.mx.{zone.name}'
elif mx[-1] != '.':
mx = f'{mx}.'
@@ -59,8 +63,7 @@ class TinyDnsBaseSource(BaseSource):
# if we have an IP then we need to create an A for the MX
ip = line[1]
if ip:
mx_name = zone.hostname_from_fqdn(mx)
yield 'A', mx_name, ttl, [ip]
yield 'A', mx, ttl, [ip]
values.append({'preference': dist, 'exchange': mx})
@@ -69,9 +72,15 @@ class TinyDnsBaseSource(BaseSource):
def _records_for_C(self, zone, name, lines, arpa=False):
# Cfqdn:p:ttl:timestamp:lo
# CNAME
if arpa:
# no arpa
return []
if not zone.owns('CNAME', name):
# if name doesn't live under our zone there's nothing for us to do
return
value = lines[0][1]
if value[-1] != '.':
value = f'{value}.'
@@ -88,19 +97,35 @@ class TinyDnsBaseSource(BaseSource):
yield 'CNAME', name, ttl, [value]
def _records_for_caret(self, zone, name, lines, arpa=False):
# .fqdn:ip:x:ttl:timestamp:lo
# NS (and optional A)
# ^fqdn:p:ttl:timestamp:lo
# PTR, line may be a A/AAAA or straight PTR
if not arpa:
print('bailing')
# we only operate on arpa
return []
print('here')
values = []
names = defaultdict(list)
for line in lines:
value = line[1]
if line[0].endswith('in-addr.arpa') or line[0].endswith(
'ip6.arpa.'
):
# it's a straight PTR record, already in in-addr.arpa format,
# 2nd item is the name it points to
name = line[0]
value = line[1]
else:
# it's not a PTR we need to build up the PTR data from what
# we're given
value = line[0]
addr = line[1]
if '.' not in addr:
addr = u':'.join(textwrap.wrap(line[1], 4))
addr = ip_address(addr)
name = addr.reverse_pointer
if value[-1] != '.':
value = f'{value}.'
values.append(value)
names[name].append(value)
# see if we can find a ttl on any of the lines, first one wins
ttl = self.default_ttl
@@ -111,23 +136,30 @@ class TinyDnsBaseSource(BaseSource):
except IndexError:
pass
pprint({'caret': values})
yield 'PTR', name, ttl, values
for name, values in names.items():
if zone.owns('PTR', name):
yield 'PTR', name, ttl, values
def _records_for_equal(self, zone, name, lines, arpa=False):
# =fqdn:ip:ttl:timestamp:lo
# A (arpa False) & PTR (arpa True)
print(f'here for {name}: {lines}')
yield from self._records_for_plus(zone, name, lines, arpa)
yield from self._records_for_caret(zone, name, lines, arpa)
if arpa:
yield from self._records_for_caret(zone, name, lines, arpa)
else:
yield from self._records_for_plus(zone, name, lines, arpa)
def _records_for_dot(self, zone, name, lines, arpa=False):
# .fqdn:ip:x:ttl:timestamp:lo
# NS (and optional A)
if arpa:
# no arpa
return []
if not zone.owns('NS', name):
# if name doesn't live under our zone there's nothing for us to do
return
# see if we can find a ttl on any of the lines, first one wins
ttl = self.default_ttl
for line in lines:
@@ -150,8 +182,7 @@ class TinyDnsBaseSource(BaseSource):
# if we have an IP then we need to create an A for the MX
ip = line[1]
if ip:
ns_name = zone.hostname_from_fqdn(ns)
yield 'A', ns_name, ttl, [ip]
yield 'A', ns, ttl, [ip]
values.append(ns)
@@ -162,9 +193,15 @@ class TinyDnsBaseSource(BaseSource):
def _records_for_plus(self, zone, name, lines, arpa=False):
# +fqdn:ip:ttl:timestamp:lo
# A
if arpa:
# no arpa
return []
if not zone.owns('A', name):
# if name doesn't live under our zone there's nothing for us to do
return
# collect our ip(s)
ips = [l[1] for l in lines if l[1] != '0.0.0.0']
@@ -186,9 +223,15 @@ class TinyDnsBaseSource(BaseSource):
def _records_for_quote(self, zone, name, lines, arpa=False):
# 'fqdn:s:ttl:timestamp:lo
# TXT
if arpa:
# no arpa
return []
if not zone.owns('TXT', name):
# if name doesn't live under our zone there's nothing for us to do
return
# collect our ip(s)
values = [
l[1].encode('latin1').decode('unicode-escape').replace(";", "\\;")
@@ -209,9 +252,15 @@ class TinyDnsBaseSource(BaseSource):
def _records_for_three(self, zone, name, lines, arpa=False):
# 3fqdn:ip:ttl:timestamp:lo
# AAAA
if arpa:
# no arpa
return []
if not zone.owns('AAAA', name):
# if name doesn't live under our zone there's nothing for us to do
return
# collect our ip(s)
ips = []
for line in lines:
@@ -234,8 +283,10 @@ class TinyDnsBaseSource(BaseSource):
def _records_for_six(self, zone, name, lines, arpa=False):
# 6fqdn:ip:ttl:timestamp:lo
# AAAA (arpa False) & PTR (arpa True)
yield from self._records_for_three(zone, name, lines, arpa)
yield from self._records_for_caret(zone, name, lines, arpa)
if arpa:
yield from self._records_for_caret(zone, name, lines, arpa)
else:
yield from self._records_for_three(zone, name, lines, arpa)
SYMBOL_MAP = {
'=': _records_for_equal, # A
@@ -256,8 +307,6 @@ class TinyDnsBaseSource(BaseSource):
}
def _process_lines(self, zone, lines):
name_re = re.compile(fr'((?P<name>.+)\.)?{zone.name[:-1]}\.?$')
data = defaultdict(lambda: defaultdict(list))
for line in lines:
symbol = line[0]
@@ -266,15 +315,7 @@ class TinyDnsBaseSource(BaseSource):
line = line[1:].split('#', 1)[0]
# Split on :'s including :: and strip leading/trailing ws
line = [p.strip() for p in line.split(':')]
# make sure the name portion matches the zone we're currently
# working on
name = line[0]
if not name_re.match(name):
self.log.info('skipping name %s, not a match, %s: ', name, line)
continue
# remove the zone name
name = zone.hostname_from_fqdn(name)
data[symbol][name].append(line)
data[symbol][line[0]].append(line)
return data
@@ -294,6 +335,8 @@ class TinyDnsBaseSource(BaseSource):
for _type, name, ttl, values in records_for(
self, zone, name, lines, arpa=arpa
):
# remove the zone name
name = zone.hostname_from_fqdn(name)
types[_type][name].extend(values)
# last one wins
ttls[_type][name] = ttl
@@ -318,7 +361,6 @@ class TinyDnsBaseSource(BaseSource):
# first group lines by their symbol and name
symbols = self._process_lines(zone, self._lines())
pprint({'symbols': symbols})
# then work through those to group values by their _type and name
zone_name = zone.name
@@ -326,7 +368,6 @@ class TinyDnsBaseSource(BaseSource):
'ip6.arpa.'
)
types, ttls = self._process_symbols(zone, symbols, arpa)
pprint({'types': types, 'ttls': ttls})
# now we finally have all the values for each (soon to be) record
# collected together, turn them into their coresponding record and add
@@ -338,9 +379,7 @@ class TinyDnsBaseSource(BaseSource):
data['values'] = values
else:
data['value'] = values[0]
pprint({'name': name, 'data': data})
record = Record.new(zone, name, data, lenient=lenient)
pprint({'lenient': lenient})
try:
zone.add_record(record, lenient=lenient)
except SubzoneRecordException:
@@ -352,53 +391,6 @@ class TinyDnsBaseSource(BaseSource):
'populate: found %s records', len(zone.records) - before
)
def _populate_arpa_arpa(self, zone, lenient):
name_re = re.compile(fr'(?P<name>.+)\.{zone.name[:-1]}\.?$')
for line in self._lines():
_type = line[0]
# We're only interested in = (A+PTR), and ^ (PTR) records
if _type not in ('=', '^', '&'):
continue
# Skip type, remove trailing comments, and omit newline
line = line[1:].split('#', 1)[0]
# Split on :'s including :: and strip leading/trailing ws
line = [p.strip() for p in self.split_re.split(line)]
if line[0].endswith('in-addr.arpa'):
# since it's already in in-addr.arpa format
match = name_re.match(line[0])
value = line[1]
else:
addr = ip_address(line[1])
match = name_re.match(addr.reverse_pointer)
value = line[0]
if match:
try:
ttl = line[2]
except IndexError:
ttl = self.default_ttl
if value[-1] != '.':
value = f'{value}.'
name = match.group('name')
record = Record.new(
zone,
name,
{'ttl': ttl, 'type': 'PTR', 'value': value},
source=self,
lenient=lenient,
)
try:
zone.add_record(record, lenient=lenient)
except DuplicateRecordException:
self.log.warning(
f'Duplicate PTR record for {addr}, skipping'
)
class TinyDnsFileSource(TinyDnsBaseSource):
'''

50
tests/test_octodns_source_tinydns.py
View File

@@ -17,7 +17,7 @@ class TestTinyDnsFileSource(TestCase):
def test_populate_normal(self):
got = Zone('example.com.', [])
self.source.populate(got)
self.assertEqual(17, len(got.records))
self.assertEqual(24, len(got.records))
expected = Zone('example.com.', [])
for name, data in (
@@ -26,8 +26,13 @@ class TestTinyDnsFileSource(TestCase):
'',
{
'type': 'NS',
'ttl': 3600,
'values': ['ns1.ns.com.', 'ns2.ns.com.'],
'ttl': 31,
'values': [
'a.ns.example.com.',
'b.ns.example.com.',
'ns1.ns.com.',
'ns2.ns.com.',
],
},
),
(
@@ -75,11 +80,11 @@ class TestTinyDnsFileSource(TestCase):
'values': [
{
'preference': 30,
'exchange': 'smtp-1-host.example.com.',
'exchange': 'smtp-3-host.mx.example.com.',
},
{
'preference': 40,
'exchange': 'smtp-2-host.example.com.',
'exchange': 'smtp-4-host.mx.example.com.',
},
],
},
@@ -111,6 +116,26 @@ class TestTinyDnsFileSource(TestCase):
'value': 'v=DKIM1\\; k=rsa\\; p=blah',
},
),
('b.ns', {'type': 'A', 'ttl': 31, 'value': '43.44.45.46'}),
('a.ns', {'type': 'A', 'ttl': 3600, 'value': '42.43.44.45'}),
(
'smtp-3-host.mx',
{'type': 'A', 'ttl': 1800, 'value': '21.22.23.24'},
),
(
'smtp-4-host.mx',
{'type': 'A', 'ttl': 1800, 'value': '22.23.24.25'},
),
('ns5.ns', {'type': 'A', 'ttl': 30, 'value': '14.15.16.17'}),
('ns6.ns', {'type': 'A', 'ttl': 30, 'value': '15.16.17.18'}),
(
'other',
{
'type': 'NS',
'ttl': 30,
'values': ['ns5.ns.example.com.', 'ns6.ns.example.com.'],
},
),
):
record = Record.new(expected, name, data)
expected.add_record(record)
@@ -162,7 +187,10 @@ class TestTinyDnsFileSource(TestCase):
{
'type': 'PTR',
'ttl': 3600,
'value': 'has-dup-def123.example.com.',
'values': [
'has-dup-def123.example.com.',
'has-dup-def456.example.com.',
],
},
),
(
@@ -178,18 +206,10 @@ class TestTinyDnsFileSource(TestCase):
expected.add_record(record)
changes = expected.changes(got, SimpleProvider())
from pprint import pprint
pprint(
{
'changes': changes,
'expected': expected.records,
'got': got.records,
}
)
self.assertEqual([], changes)
def test_ignores_subs(self):
got = Zone('example.com.', ['sub'])
self.source.populate(got)
# we don't see one www.sub.example.com. record b/c it's in a sub
self.assertEqual(23, len(got.records))

4
tests/zones/tinydns/example.com
View File

@@ -28,8 +28,8 @@ Ccname.other.foo:www.other.foo
@example.com::smtp-1-host.example.com:10
@example.com.::smtp-2-host.example.com:20
# MX with ttl and ip
@smtp.example.com:21.22.23.24:smtp-1-host:30:1800
@smtp.example.com.:22.23.24.25:smtp-2-host:40:1800
@smtp.example.com:21.22.23.24:smtp-3-host:30:1800
@smtp.example.com.:22.23.24.25:smtp-4-host:40:1800
# NS for sub
.sub.example.com::ns3.ns.com:30