Initial commit, squashed

filter_plugins/cumulus.py

@@ -0,0 +1,56 @@
+#!/usr/bin/python
+
+import ipaddress
+import itertools
+import re
+
+class FilterModule(object):
+    '''Filters for Cumulus Linux switches'''
+    def __init__(self):
+        self.iface_regex = re.compile(r'swp([0-9]+)(?:s([0-9]+))?')
+
+    def filters(self):
+        return {
+            'cl_iface_index': self.cl_iface_index,
+            'cl_clag_id': self.cl_clag_id,
+            'cl_clag_sys_mac': self.cl_clag_sys_mac
+        }
+
+    def cl_iface_index(self, interfaces):
+        '''
+        Return interface index from its name, eg. swp3s2 → 17
+
+        Interfaces may be broken out into up to four subinterfaces. Indexing
+        starts at 5 and proceeds as:
+
+            swp1 swp1s0 swp1s1 swp1s2 swp1s3 swp2 swp2s0 …
+        '''
+
+        for interface in interfaces:
+            if m := self.iface_regex.fullmatch(interface.get('name', '')):
+                yield 5 * (int(m.group(1))) + (int(m.group(2))+1 if m.group(2) else 0)
+
+    def cl_clag_id(self, interfaces):
+        '''
+        Generate a clag-id from a list of interfaces making up a MLAG bond
+
+        The clag-id for a bond must be between 1 and 65535. Generate it by
+        combining the indexes of the lowest-numbered interface on each switch.
+        This avoids manual ID assignment while keeping them mostly stable.
+
+        This clag-id does not change unless the lowest-numbered interface for
+        the bond on any switch changes. IDs of other bonds are not affected.
+        '''
+        clag_id = 0
+        key = lambda i: i.get('device', {}).get('name')
+        for device, ifaces in itertools.groupby(sorted(interfaces, key=key), key):
+            clag_id = 256 * clag_id + min(self.cl_iface_index(ifaces))
+        if 1 <= clag_id <= 65535: # sanity checking
+            return clag_id
+
+    def cl_clag_sys_mac(self, address):
+        '''Generate a system MAC address for a MLAG with the given address'''
+
+        index = int(ipaddress.ip_address(address)) % 2**16
+        suffix = f'{index:04x}'
+        return f'44:38:39:ff:{suffix[:2]}:{suffix[2:]}'

filter_plugins/netbox.py

@@ -0,0 +1,39 @@
+#!/usr/bin/python
+
+class FilterModule(object):
+    '''Various utilities for manipulating NetBox data'''
+    def __init__(self):
+        self.virtual_iface_types = ('bridge', 'lag', 'virtual')
+
+    def filters(self):
+        return {
+            'iface_real': self.iface_real,
+            'iface_peer': self.iface_peer,
+            'iface_vlans': self.iface_vlans
+        }
+
+    def iface_real(self, interfaces):
+        '''Return only non-virtual interfaces'''
+        for iface in interfaces:
+            if iface.get('type', {}).get('value') not in self.virtual_iface_types:
+                yield iface
+
+    def iface_peer(self, interfaces):
+        '''Return the name of the device connected to this interface'''
+        for iface in interfaces:
+            endpoint = (iface.get('connected_endpoints') or [{}])[0]
+            if device := endpoint.get('device', {}).get('name'):
+                yield device
+            else:
+                yield None
+
+    def iface_vlans(self, interfaces):
+        '''Returns a list of allowed VLANs for each interface'''
+        for iface in interfaces:
+            match iface.get('mode', {}).get('value'):
+                case 'access':
+                    yield [iface.get('untagged_vlan')] if 'untagged_vlan' in iface else []
+                case 'tagged':
+                    yield iface.get('tagged_vlans', [])
+                case _:
+                    yield []