feat: Add NetworkInterfaceSource

octodns/source/networkinterface.py

@@ -0,0 +1,110 @@
+import logging
+from ipaddress import ip_address
+
+import ifaddr
+
+from ..record import Record
+from .base import BaseSource
+
+
+class NetworkInterfaceSource(BaseSource):
+    SUPPORTS_GEO = False
+    SUPPORTS_DYNAMIC = False
+    SUPPORTS = {'A', 'AAAA'}
+
+    DEFAULT_TTL = 60
+
+    def __init__(
+        self,
+        id,
+        name,
+        ttl=DEFAULT_TTL,
+        is_global=True,
+        is_link_local=False,
+        is_loopback=False,
+        is_multicast=False,
+        is_private=False,
+        is_reserved=False,
+    ):
+        klass = self.__class__.__name__
+        self.log = logging.getLogger(f'{klass}[{id}]')
+        self.log.setLevel(logging.DEBUG)
+        self.log.debug(
+            '__init__: id=%s, name=%s, ttl=%d, is_global=%s, is_link_local=%s, is_loopback=%s, is_multicast=%s, is_private=%s, is_reserved=%s',
+            id,
+            name,
+            ttl,
+            is_global,
+            is_link_local,
+            is_loopback,
+            is_multicast,
+            is_private,
+            is_reserved,
+        )
+        super().__init__(id)
+        self.name = name
+        self.ttl = ttl
+        self.is_global = is_global
+        self.is_link_local = is_link_local
+        self.is_loopback = is_loopback
+        self.is_multicast = is_multicast
+        self.is_private = is_private
+        self.is_reserved = is_reserved
+
+    @staticmethod
+    def _get_ips():  # pragma: no cover
+        # The method can not be covered in tests as it has to always get mocked
+        ips = []
+        for adapter in ifaddr.get_adapters():
+            for ip in adapter.ips:
+                ips.append(ip)
+        return ips
+
+    def populate(self, zone, target=False, lenient=False):
+        self.log.debug(
+            'populate: name=%s, target=%s, lenient=%s',
+            zone.name,
+            target,
+            lenient,
+        )
+
+        before = len(zone.records)
+
+        for ip in self._get_ips():
+            value = ip.ip
+            record_type = 'A'
+            if isinstance(value, tuple):
+                value = value[0]
+                record_type = 'AAAA'
+
+            parsed_ip = ip_address(value)
+            add = False
+            for prop in [
+                'is_global',
+                'is_link_local',
+                'is_loopback',
+                'is_multicast',
+                'is_private',
+                'is_reserved',
+            ]:
+                if getattr(parsed_ip, prop) and getattr(self, prop):
+                    add = True
+                    break
+            if not add:  # pragma: no cover
+                continue
+
+            zone.add_record(
+                Record.new(
+                    zone,
+                    self.name,
+                    {'ttl': self.ttl, 'type': record_type, 'values': [value]},
+                    source=self,
+                    lenient=lenient,
+                ),
+                lenient=lenient,
+            )
+
+        self.log.info(
+            'populate: found %s records, exists=False',
+            len(zone.records) - before,
+        )

requirements.txt

@@ -6,3 +6,4 @@ idna==3.10
 natsort==8.4.0
 python-dateutil==2.9.0.post0
 six==1.16.0
+ifaddr==0.2.0

tests/test_octodns_source_networkinterface.py

@@ -0,0 +1,38 @@
+from unittest import TestCase
+from unittest.mock import MagicMock
+
+from ifaddr import IP
+
+from octodns.source.networkinterface import NetworkInterfaceSource
+from octodns.zone import Zone
+
+
+class TestNetworkInterfaceSource(TestCase):
+    def test_populate(self):
+        name = 'testrecord'
+        source = NetworkInterfaceSource('testid', name, is_loopback=True)
+        source._get_ips = MagicMock(
+            return_value=[IP('127.0.0.1', 0, ''), IP(('::1', 0, 0), 0, '')]
+        )
+
+        zone_name = 'unit.tests.'
+        zone = Zone(zone_name, [])
+        source.populate(zone)
+
+        self.assertEqual(2, len(zone.records))
+
+        a_record = list(
+            filter(lambda record: record._type == 'A', zone.records)
+        )[0]
+        self.assertEqual(name, a_record.name)
+        self.assertEqual(f'{name}.{zone_name}', a_record.fqdn)
+        self.assertEqual(1, len(a_record.values))
+        self.assertEqual('127.0.0.1', a_record.values[0])
+
+        aaaa_record = list(
+            filter(lambda record: record._type == 'AAAA', zone.records)
+        )[0]
+        self.assertEqual(name, aaaa_record.name)
+        self.assertEqual(f'{name}.{zone_name}', aaaa_record.fqdn)
+        self.assertEqual(1, len(aaaa_record.values))
+        self.assertEqual('::1', aaaa_record.values[0])