implementation of the URI record type

.changelog/db3856b033d0439cbfba28ab79b86c50.md

@@ -0,0 +1,4 @@
+---
+type: minor
+---
+Add support for the URI record type

octodns/record/uri.py

@@ -0,0 +1,138 @@
+#
+#
+#
+
+import re
+
+from ..equality import EqualityTupleMixin
+from ..idna import idna_encode
+from .base import Record, ValuesMixin, unquote
+from .rr import RrParseError
+
+
+class UriValue(EqualityTupleMixin, dict):
+    @classmethod
+    def parse_rdata_text(self, value):
+        try:
+            priority, weight, target = value.split(' ')
+        except ValueError:
+            raise RrParseError()
+        try:
+            priority = int(priority)
+        except ValueError:
+            pass
+        try:
+            weight = int(weight)
+        except ValueError:
+            pass
+        target = unquote(target)
+        return {'priority': priority, 'weight': weight, 'target': target}
+
+    @classmethod
+    def validate(cls, data, _type):
+        reasons = []
+        for value in data:
+            # TODO: validate algorithm and fingerprint_type values
+            try:
+                int(value['priority'])
+            except KeyError:
+                reasons.append('missing priority')
+            except ValueError:
+                reasons.append(f'invalid priority "{value["priority"]}"')
+            try:
+                int(value['weight'])
+            except KeyError:
+                reasons.append('missing weight')
+            except ValueError:
+                reasons.append(f'invalid weight "{value["weight"]}"')
+            try:
+                target = value['target']
+                if not target:
+                    reasons.append('missing target')
+                    continue
+                # actual validation of the target is non-trivial and specific
+                # to the details of the schema etc. rfc3986 has support for
+                # validation, but we don't currently require the module and
+                # this seems too esoteric a use case to add it
+            except KeyError:
+                reasons.append('missing target')
+        return reasons
+
+    @classmethod
+    def process(cls, values):
+        return [cls(v) for v in values]
+
+    def __init__(self, value):
+        super().__init__(
+            {
+                'priority': int(value['priority']),
+                'weight': int(value['weight']),
+                'target': idna_encode(value['target']),
+            }
+        )
+
+    @property
+    def priority(self):
+        return self['priority']
+
+    @priority.setter
+    def priority(self, value):
+        self['priority'] = value
+
+    @property
+    def weight(self):
+        return self['weight']
+
+    @weight.setter
+    def weight(self, value):
+        self['weight'] = value
+
+    @property
+    def target(self):
+        return self['target']
+
+    @target.setter
+    def target(self, value):
+        self['target'] = value
+
+    @property
+    def data(self):
+        return self
+
+    @property
+    def rdata_text(self):
+        return f'{self.priority} {self.weight} "{self.target}"'
+
+    def template(self, params):
+        if '{' not in self.target:
+            return self
+        new = self.__class__(self)
+        new.target = new.target.format(**params)
+        return new
+
+    def __hash__(self):
+        return hash(self.__repr__())
+
+    def _equality_tuple(self):
+        return (self.priority, self.weight, self.target)
+
+    def __repr__(self):
+        return f"'{self.priority} {self.weight} \"{self.target}\"'"
+
+
+# https://datatracker.ietf.org/doc/html/rfc7553
+class UriRecord(ValuesMixin, Record):
+    _type = 'URI'
+    _value_type = UriValue
+    _name_re = re.compile(r'^(\*|_[^\.]+)\.[^\.]+')
+
+    @classmethod
+    def validate(cls, name, fqdn, data):
+        reasons = []
+        if not cls._name_re.match(name):
+            reasons.append('invalid name for URI record')
+        reasons.extend(super().validate(name, fqdn, data))
+        return reasons
+
+
+Record.register_type(UriRecord)

tests/test_octodns_record_uri.py

@@ -0,0 +1,364 @@
+#
+#
+#
+
+from unittest import TestCase
+
+from helpers import SimpleProvider
+
+from octodns.record import Record
+from octodns.record.exception import ValidationError
+from octodns.record.rr import RrParseError
+from octodns.record.uri import UriRecord, UriValue
+from octodns.zone import Zone
+
+
+class TestRecordUri(TestCase):
+    zone = Zone('unit.tests.', [])
+
+    def test_uri(self):
+        a_values = [
+            UriValue(
+                {
+                    'priority': 10,
+                    'weight': 11,
+                    'target': 'https://server1/foo/bar',
+                }
+            ),
+            UriValue(
+                {
+                    'priority': 20,
+                    'weight': 21,
+                    'target': 'https://server2/foo/bar',
+                }
+            ),
+        ]
+        a_data = {'ttl': 30, 'values': a_values}
+        a = UriRecord(self.zone, '_a._tcp', a_data)
+        self.assertEqual('_a._tcp', a.name)
+        self.assertEqual('_a._tcp.unit.tests.', a.fqdn)
+        self.assertEqual(30, a.ttl)
+        self.assertEqual(a_values[0]['priority'], a.values[0].priority)
+        self.assertEqual(a_values[0]['weight'], a.values[0].weight)
+        self.assertEqual(a_values[0]['target'], a.values[0].target)
+        self.assertEqual(a_data, a.data)
+
+        b_value = UriValue(
+            {'priority': 30, 'weight': 31, 'target': 'ftp://server3/here'}
+        )
+        b_data = {'ttl': 30, 'value': b_value}
+        b = UriRecord(self.zone, '_b._tcp', b_data)
+        self.assertEqual(b_value['priority'], b.values[0].priority)
+        self.assertEqual(b_value['weight'], b.values[0].weight)
+        self.assertEqual(b_value['target'], b.values[0].target)
+        self.assertEqual(b_data, b.data)
+
+        target = SimpleProvider()
+        # No changes with self
+        self.assertFalse(a.changes(a, target))
+        # Diff in priority causes change
+        other = UriRecord(
+            self.zone, '_a._icmp', {'ttl': 30, 'values': a_values}
+        )
+        other.values[0].priority = 22
+        change = a.changes(other, target)
+        self.assertEqual(change.existing, a)
+        self.assertEqual(change.new, other)
+        # Diff in weight causes change
+        other.values[0].priority = a.values[0].priority
+        other.values[0].weight = 33
+        change = a.changes(other, target)
+        self.assertEqual(change.existing, a)
+        self.assertEqual(change.new, other)
+        # Diff in target causes change
+        other.values[0].target = 'ftp://serverX/there'
+        change = a.changes(other, target)
+        self.assertEqual(change.existing, a)
+        self.assertEqual(change.new, other)
+
+        # __repr__ doesn't blow up
+        a.__repr__()
+
+    def test_uri_value_rdata_text(self):
+        # empty string won't parse
+        with self.assertRaises(RrParseError):
+            UriValue.parse_rdata_text('')
+
+        # single word won't parse
+        with self.assertRaises(RrParseError):
+            UriValue.parse_rdata_text('nope')
+
+        # 2nd word won't parse
+        with self.assertRaises(RrParseError):
+            UriValue.parse_rdata_text('1 2')
+
+        # 4th word won't parse
+        with self.assertRaises(RrParseError):
+            UriValue.parse_rdata_text('1 2 3 4')
+
+        # priority and weight not ints
+        self.assertEqual(
+            {
+                'priority': 'one',
+                'weight': 'two',
+                'target': 'http://uri.unit.tests./',
+            },
+            UriValue.parse_rdata_text('one two "http://uri.unit.tests./"'),
+        )
+
+        # valid
+        self.assertEqual(
+            {'priority': 1, 'weight': 2, 'target': 'http://uri.unit.tests./'},
+            UriValue.parse_rdata_text('1 2 "http://uri.unit.tests./"'),
+        )
+
+        # quoted
+        self.assertEqual(
+            {
+                'priority': 1,
+                'weight': 2,
+                'target': 'ftp://uri.unit.tests./there',
+            },
+            UriValue.parse_rdata_text('1 2 "ftp://uri.unit.tests./there"'),
+        )
+
+        zone = Zone('unit.tests.', [])
+        a = UriRecord(
+            zone,
+            '_uri._tcp',
+            {
+                'ttl': 32,
+                'value': {
+                    'priority': 1,
+                    'weight': 2,
+                    'target': 'ssh://uri.unit.tests./',
+                },
+            },
+        )
+        self.assertEqual(1, a.values[0].priority)
+        self.assertEqual(2, a.values[0].weight)
+        self.assertEqual('ssh://uri.unit.tests./', a.values[0].target)
+        self.assertEqual('1 2 "ssh://uri.unit.tests./"', a.values[0].rdata_text)
+
+        # both directions should match
+        rdata = '1 2 "https://uri.unit.tests./path/to/it"'
+        record = UriRecord(
+            zone,
+            '_uri._tcp',
+            {'ttl': 32, 'value': UriValue.parse_rdata_text(rdata)},
+        )
+        self.assertEqual(rdata, record.values[0].rdata_text)
+
+    def test_uri_value(self):
+        a = UriValue({'priority': 0, 'weight': 0, 'target': 'tel:123-123-1234'})
+        b = UriValue({'priority': 1, 'weight': 0, 'target': 'tel:123-123-1234'})
+        c = UriValue({'priority': 0, 'weight': 2, 'target': 'tel:123-123-1234'})
+        e = UriValue({'priority': 0, 'weight': 0, 'target': 'news:mmm.blip'})
+
+        self.assertEqual(a, a)
+        self.assertEqual(b, b)
+        self.assertEqual(c, c)
+        self.assertEqual(e, e)
+
+        self.assertNotEqual(a, b)
+        self.assertNotEqual(a, c)
+        self.assertNotEqual(a, e)
+        self.assertNotEqual(b, a)
+        self.assertNotEqual(b, c)
+        self.assertNotEqual(b, e)
+        self.assertNotEqual(c, a)
+        self.assertNotEqual(c, b)
+        self.assertNotEqual(c, e)
+        self.assertNotEqual(e, a)
+        self.assertNotEqual(e, b)
+        self.assertNotEqual(e, c)
+
+        self.assertTrue(a < b)
+        self.assertTrue(a < c)
+
+        self.assertTrue(b > a)
+        self.assertTrue(b > c)
+
+        self.assertTrue(c > a)
+        self.assertTrue(c < b)
+
+        self.assertTrue(a <= b)
+        self.assertTrue(a <= c)
+        self.assertTrue(a <= a)
+        self.assertTrue(a >= a)
+
+        self.assertTrue(b >= a)
+        self.assertTrue(b >= c)
+        self.assertTrue(b >= b)
+        self.assertTrue(b <= b)
+
+        self.assertTrue(c >= a)
+        self.assertTrue(c <= b)
+        self.assertTrue(c >= c)
+        self.assertTrue(c <= c)
+
+        # Hash
+        values = set()
+        values.add(a)
+        self.assertTrue(a in values)
+        self.assertFalse(b in values)
+        values.add(b)
+        self.assertTrue(b in values)
+
+    def test_valiation(self):
+        # doesn't blow up
+        Record.new(
+            self.zone,
+            '_uri._tcp',
+            {
+                'type': 'URI',
+                'ttl': 600,
+                'value': {
+                    'priority': 1,
+                    'weight': 2,
+                    'target': 'telnet://foo.bar.baz.',
+                },
+            },
+        )
+
+        # permit wildcard entries
+        Record.new(
+            self.zone,
+            '*._tcp',
+            {
+                'type': 'URI',
+                'ttl': 600,
+                'value': {
+                    'priority': 1,
+                    'weight': 2,
+                    'target': 'http://food.bar.baz.',
+                },
+            },
+        )
+
+        # invalid name
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                'neup',
+                {
+                    'type': 'URI',
+                    'ttl': 600,
+                    'value': {
+                        'priority': 1,
+                        'weight': 2,
+                        'target': 'http://foo.bar.baz.',
+                    },
+                },
+            )
+        self.assertEqual(['invalid name for URI record'], ctx.exception.reasons)
+
+        # missing priority
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                '_uri._tcp',
+                {
+                    'type': 'URI',
+                    'ttl': 600,
+                    'value': {'weight': 2, 'target': 'blip://foo.bar.baz.'},
+                },
+            )
+        self.assertEqual(['missing priority'], ctx.exception.reasons)
+
+        # invalid priority
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                '_uri._tcp',
+                {
+                    'type': 'URI',
+                    'ttl': 600,
+                    'value': {
+                        'priority': 'foo',
+                        'weight': 2,
+                        'target': 'http://foo.bar.baz.',
+                    },
+                },
+            )
+        self.assertEqual(['invalid priority "foo"'], ctx.exception.reasons)
+
+        # missing weight
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                '_uri._tcp',
+                {
+                    'type': 'URI',
+                    'ttl': 600,
+                    'value': {'priority': 1, 'target': 'telnet://foo.bar.baz.'},
+                },
+            )
+        self.assertEqual(['missing weight'], ctx.exception.reasons)
+        # invalid weight
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                '_uri._tcp',
+                {
+                    'type': 'URI',
+                    'ttl': 600,
+                    'value': {
+                        'priority': 1,
+                        'weight': 'foo',
+                        'target': 'http://foo.bar.baz.',
+                    },
+                },
+            )
+        self.assertEqual(['invalid weight "foo"'], ctx.exception.reasons)
+
+        # missing target
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                '_uri._tcp',
+                {
+                    'type': 'URI',
+                    'ttl': 600,
+                    'value': {'priority': 1, 'weight': 2},
+                },
+            )
+        self.assertEqual(['missing target'], ctx.exception.reasons)
+        # invalid target
+        # pretty much anything is valid in the general case for a URI
+
+        # falsey target
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                '_uri._tcp',
+                {
+                    'type': 'URI',
+                    'ttl': 600,
+                    'value': {'priority': 1, 'weight': 2, 'target': ''},
+                },
+            )
+        self.assertEqual(['missing target'], ctx.exception.reasons)
+
+        # target must be a valid URI
+
+
+class TestUriValue(TestCase):
+
+    def test_template(self):
+        value = UriValue(
+            {'priority': 10, 'weight': 11, 'target': 'no_placeholders'}
+        )
+        got = value.template({'needle': 42})
+        self.assertIs(value, got)
+
+        value = UriValue(
+            {
+                'priority': 10,
+                'weight': 11,
+                'target': 'http://has_{needle}_placeholder/some/path',
+            }
+        )
+        got = value.template({'needle': 42})
+        self.assertIsNot(value, got)
+        self.assertEqual('http://has_42_placeholder/some/path', got.target)