From ce77ccddb51c545e1b076d387f57cfea87455d3a Mon Sep 17 00:00:00 2001 From: Ross McFarland Date: Thu, 10 Jul 2025 13:22:42 -0700 Subject: [PATCH] implementation of the URI record type --- .../db3856b033d0439cbfba28ab79b86c50.md | 4 + octodns/record/uri.py | 138 +++++++ tests/test_octodns_record_uri.py | 364 ++++++++++++++++++ 3 files changed, 506 insertions(+) create mode 100644 .changelog/db3856b033d0439cbfba28ab79b86c50.md create mode 100644 octodns/record/uri.py create mode 100644 tests/test_octodns_record_uri.py diff --git a/.changelog/db3856b033d0439cbfba28ab79b86c50.md b/.changelog/db3856b033d0439cbfba28ab79b86c50.md new file mode 100644 index 0000000..6022af8 --- /dev/null +++ b/.changelog/db3856b033d0439cbfba28ab79b86c50.md @@ -0,0 +1,4 @@ +--- +type: minor +--- +Add support for the URI record type \ No newline at end of file diff --git a/octodns/record/uri.py b/octodns/record/uri.py new file mode 100644 index 0000000..2e2fa4e --- /dev/null +++ b/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) diff --git a/tests/test_octodns_record_uri.py b/tests/test_octodns_record_uri.py new file mode 100644 index 0000000..c91d8aa --- /dev/null +++ b/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)