diff --git a/.changelog/2809d288040441ccb8e6633f514b09b0.md b/.changelog/2809d288040441ccb8e6633f514b09b0.md new file mode 100644 index 0000000..91ca407 --- /dev/null +++ b/.changelog/2809d288040441ccb8e6633f514b09b0.md @@ -0,0 +1,5 @@ +--- +type: minor +--- + +Add processor for clamping TTLs diff --git a/octodns/processor/clamp.py b/octodns/processor/clamp.py new file mode 100644 index 0000000..68a3fe2 --- /dev/null +++ b/octodns/processor/clamp.py @@ -0,0 +1,73 @@ +from logging import getLogger + +from .base import BaseProcessor, ProcessorException + + +class TTLArgumentException(ProcessorException): + pass + + +class TtlClampProcessor(BaseProcessor): + """ + Processor that clamps TTL values to a specified range. + + Configuration: + min_ttl: Minimum TTL value (default: 300 seconds / 5 minutes) + max_ttl: Maximum TTL value (default: 86400 seconds / 24 hours) + + Example config.yaml: + processors: + clamp: + class: octodns.processor.clamp.TtlClampProcessor + min_ttl: 300 + max_ttl: 3600 + + zones: + example.com.: + sources: + - config + processors: + - clamp + targets: + - route53 + """ + + def __init__(self, id, min_ttl=300, max_ttl=86400): + super().__init__(id) + self.log = getLogger(self.__class__.__name__) + if not min_ttl <= max_ttl: + raise TTLArgumentException( + f'Min TTL {min_ttl} is not lower than max TTL {max_ttl}' + ) + self.min_ttl = min_ttl + self.max_ttl = max_ttl + self.log.info('__init__: min=%ds, max=%ds', self.min_ttl, self.max_ttl) + + def process_source_zone(self, desired, sources): + """ + Process records from source zone(s). + + Args: + desired: Zone object containing the desired records + sources: List of source names + + Returns: + The modified zone + """ + self.log.debug('process_source_zone: desired=%s', desired.name) + + for record in desired.records: + original_ttl = record.ttl + clamped_ttl = max(self.min_ttl, min(self.max_ttl, original_ttl)) + + if clamped_ttl != original_ttl: + self.log.info( + 'process_source_zone: clamping TTL for %s (%s) %s -> %s', + record.fqdn, + record._type, + original_ttl, + clamped_ttl, + ) + record.ttl = clamped_ttl + + return desired diff --git a/tests/test_octodns_processor_clamp.py b/tests/test_octodns_processor_clamp.py new file mode 100644 index 0000000..63c2de1 --- /dev/null +++ b/tests/test_octodns_processor_clamp.py @@ -0,0 +1,120 @@ +from unittest import TestCase + +from octodns.processor.clamp import TTLArgumentException, TtlClampProcessor +from octodns.record.base import Record +from octodns.zone import Zone + + +class TestClampProcessor(TestCase): + + def test_processor_min(self): + "Test the processor for clamping to the minimum" + min_ttl = 42 + processor = TtlClampProcessor('test', min_ttl=min_ttl) + + too_low_ttl = 23 + self.assertLess(too_low_ttl, min_ttl) + + zone = Zone('unit.tests.', []) + zone.add_record( + Record.new( + zone, '', {'type': 'TXT', 'ttl': too_low_ttl, 'value': 'foo'} + ) + ) + + processed_zone = processor.process_source_zone(zone.copy(), None) + self.assertNotEqual(zone, processed_zone) + + self.assertEqual(len(processed_zone.records), len(zone.records)) + self.assertEqual(len(processed_zone.records), 1) + self.assertEqual(processed_zone.records.pop().ttl, min_ttl) + + def test_processor_max(self): + "Test the processor for clamping to the maximum" + max_ttl = 4711 + processor = TtlClampProcessor('test', max_ttl=max_ttl) + + too_high_ttl = max_ttl + 1 + self.assertLess(max_ttl, too_high_ttl) + + zone = Zone('unit.tests.', []) + zone.add_record( + Record.new( + zone, '', {'type': 'TXT', 'ttl': too_high_ttl, 'value': 'foo'} + ) + ) + + processed_zone = processor.process_source_zone(zone.copy(), None) + self.assertNotEqual(zone, processed_zone) + + self.assertEqual(len(processed_zone.records), len(zone.records)) + self.assertEqual(len(processed_zone.records), 1) + self.assertEqual(processed_zone.records.pop().ttl, max_ttl) + + def test_processor_maxmin(self): + "Test the processor for unlogical arguments" + min_ttl = 42 + max_ttl = 23 + self.assertRaises( + TTLArgumentException, + TtlClampProcessor, + 'test', + min_ttl=min_ttl, + max_ttl=max_ttl, + ) + + def test_processor_minmax(self): + "Test the processor for clamping both min and max values" + min_ttl = 42 + max_ttl = 4711 + processor = TtlClampProcessor('test', min_ttl=min_ttl, max_ttl=max_ttl) + + too_low_ttl = min_ttl - 1 + too_high_ttl = max_ttl + 1 + self.assertLess(too_low_ttl, min_ttl) + self.assertLess(too_low_ttl, min_ttl) + self.assertLess(max_ttl, too_high_ttl) + + zone = Zone('unit.tests.', []) + zone.add_record( + Record.new( + zone, + 'high', + {'type': 'TXT', 'ttl': too_high_ttl, 'value': 'high'}, + ) + ) + zone.add_record( + Record.new( + zone, 'low', {'type': 'TXT', 'ttl': too_low_ttl, 'value': 'low'} + ) + ) + + processed_zone = processor.process_source_zone(zone.copy(), None) + self.assertNotEqual(zone, processed_zone) + + processed_records = sorted( + list(processed_zone.records), key=lambda r: r.ttl + ) + self.assertEqual(len(processed_records), 2) + + self.assertEqual(processed_records[0].ttl, min_ttl) + self.assertEqual(processed_records[1].ttl, max_ttl) + + def test_processor_noclamp(self): + "Test the processor for working with TTLs not requiring any clamping" + min_ttl = 23 + max_ttl = 4711 + processor = TtlClampProcessor('test', min_ttl=min_ttl, max_ttl=max_ttl) + + ttl = 42 + + self.assertLess(min_ttl, ttl) + self.assertLess(ttl, max_ttl) + + zone = Zone('unit.tests.', []) + zone.add_record( + Record.new(zone, '', {'type': 'TXT', 'ttl': ttl, 'value': 'foo'}) + ) + + processed_zone = processor.process_source_zone(zone.copy(), None) + self.assertEqual(processed_zone.records.pop().ttl, ttl)