Implement TtlRestrictionFilter w/tests

CHANGELOG.md

@@ -11,6 +11,7 @@
 * Now that it's used as it needed to be YamlProvider overrides
   Provider.supports and just always says Yes so that any dynamically registered
   types will be supported.
+* Add TtlRestrictionFilter processor for adding ttl restriction/checking
 
 ## v0.9.18 - 2022-08-14 - Subzone handling
 

octodns/processor/base.py

@@ -10,6 +10,10 @@ from __future__ import (
 )
 
 
+class ProcessorException(Exception):
+    pass
+
+
 class BaseProcessor(object):
     def __init__(self, name):
         self.name = name

octodns/processor/restrict.py

@@ -0,0 +1,60 @@
+#
+#
+#
+
+from __future__ import (
+    absolute_import,
+    division,
+    print_function,
+    unicode_literals,
+)
+
+from .base import BaseProcessor, ProcessorException
+
+
+class RestrictionException(ProcessorException):
+    pass
+
+
+class TtlRestrictionFilter(BaseProcessor):
+    '''
+    Ensure that configured TTLs are between a configured minimum and maximum.
+    The default minimum is 1 (the behavior of 0 is undefined spec-wise) and the
+    default maximum is 604800 (seven days.)
+
+    Example usage:
+
+    processors:
+      min-max-ttl:
+        class: octodns.processor.restrict.TtlRestrictionFilter
+        min_ttl: 60
+        max_ttl: 3600
+
+    zones:
+      exxampled.com.:
+        sources:
+          - config
+        processors:
+          - min-max-ttl
+        targets:
+          - azure
+    '''
+
+    SEVEN_DAYS = 60 * 60 * 24 * 7
+
+    def __init__(self, name, min_ttl=1, max_ttl=SEVEN_DAYS):
+        super().__init__(name)
+        self.min_ttl = min_ttl
+        self.max_ttl = max_ttl
+
+    def process_source_zone(self, zone, *args, **kwargs):
+        for record in zone.records:
+            if record.ttl < self.min_ttl:
+                raise RestrictionException(
+                    f'{record.fqdn} ttl={record.ttl} too low, min_ttl={self.min_ttl}'
+                )
+            elif record.ttl > self.max_ttl:
+                raise RestrictionException(
+                    f'{record.fqdn} ttl={record.ttl} too high, max_ttl={self.max_ttl}'
+                )
+        return zone

tests/test_octodns_processor_restrict.py

@@ -0,0 +1,89 @@
+from unittest import TestCase
+
+from octodns.processor.restrict import (
+    RestrictionException,
+    TtlRestrictionFilter,
+)
+from octodns.record import Record
+from octodns.zone import Zone
+
+
+class TestTtlRestrictionFilter(TestCase):
+    zone = Zone('unit.tests.', [])
+    matches = Record.new(
+        zone, 'matches', {'type': 'A', 'ttl': 42, 'value': '1.2.3.4'}
+    )
+    zone.add_record(matches)
+    doesnt = Record.new(
+        zone, 'doesnt', {'type': 'A', 'ttl': 42, 'value': '2.3.4.5'}
+    )
+    zone.add_record(doesnt)
+    matchable1 = Record.new(
+        zone, 'start-f43ad96-end', {'type': 'A', 'ttl': 42, 'value': '3.4.5.6'}
+    )
+    zone.add_record(matchable1)
+    matchable2 = Record.new(
+        zone, 'start-a3b444c-end', {'type': 'A', 'ttl': 42, 'value': '4.5.6.7'}
+    )
+    zone.add_record(matchable2)
+
+    def test_restrict_ttl(self):
+        # configured values
+        restrictor = TtlRestrictionFilter('test', min_ttl=32, max_ttl=1024)
+
+        zone = Zone('unit.tests.', [])
+        good = Record.new(
+            zone, 'good', {'type': 'A', 'ttl': 42, 'value': '1.2.3.4'}
+        )
+        zone.add_record(good)
+
+        restricted = restrictor.process_source_zone(zone)
+        self.assertEqual(zone.records, restricted.records)
+
+        low = Record.new(
+            zone, 'low', {'type': 'A', 'ttl': 16, 'value': '1.2.3.4'}
+        )
+        copy = zone.copy()
+        copy.add_record(low)
+        with self.assertRaises(RestrictionException) as ctx:
+            restrictor.process_source_zone(copy)
+        self.assertEqual(
+            'low.unit.tests. ttl=16 too low, min_ttl=32', str(ctx.exception)
+        )
+
+        high = Record.new(
+            zone, 'high', {'type': 'A', 'ttl': 2048, 'value': '1.2.3.4'}
+        )
+        copy = zone.copy()
+        copy.add_record(high)
+        with self.assertRaises(RestrictionException) as ctx:
+            restrictor.process_source_zone(copy)
+        self.assertEqual(
+            'high.unit.tests. ttl=2048 too high, max_ttl=1024',
+            str(ctx.exception),
+        )
+
+        # defaults
+        restrictor = TtlRestrictionFilter('test')
+        low = Record.new(
+            zone, 'low', {'type': 'A', 'ttl': 0, 'value': '1.2.3.4'}
+        )
+        copy = zone.copy()
+        copy.add_record(low)
+        with self.assertRaises(RestrictionException) as ctx:
+            restrictor.process_source_zone(copy)
+        self.assertEqual(
+            'low.unit.tests. ttl=0 too low, min_ttl=1', str(ctx.exception)
+        )
+
+        high = Record.new(
+            zone, 'high', {'type': 'A', 'ttl': 999999, 'value': '1.2.3.4'}
+        )
+        copy = zone.copy()
+        copy.add_record(high)
+        with self.assertRaises(RestrictionException) as ctx:
+            restrictor.process_source_zone(copy)
+        self.assertEqual(
+            'high.unit.tests. ttl=999999 too high, max_ttl=604800',
+            str(ctx.exception),
+        )