Timotej Lazar
29598ef4bb
Allow running playbooks without NetBox access. Mainly to bootstrap NetBox itself. Would prefer not to access network from filter plugins, so maybe do that at some point also.
50 lines
2 KiB
Python
50 lines
2 KiB
Python
#!/usr/bin/python
|
|
|
|
import os
|
|
import pynetbox
|
|
|
|
class FilterModule(object):
|
|
'''Various utilities for manipulating NetBox data'''
|
|
|
|
def __init__(self):
|
|
if 'NETBOX_API' in os.environ and 'NETBOX_TOKEN' in os.environ:
|
|
self.nb = pynetbox.api(os.getenv('NETBOX_API'), os.getenv('NETBOX_TOKEN'))
|
|
|
|
def filters(self):
|
|
return {
|
|
'device_address': self.device_address,
|
|
'compact_numlist': self.compact_numlist,
|
|
'allowed_prefixes': self.allowed_prefixes
|
|
}
|
|
|
|
def device_address(self, device):
|
|
'''Return loopback IP addresses for an L3 attached device'''
|
|
for iface in device['interfaces']:
|
|
for addr in iface['ip_addresses']:
|
|
if addr.get('role') and addr['role'].get('value') == 'loopback':
|
|
yield addr
|
|
|
|
def compact_numlist(self, nums, delimiter=',', range_delimiter='-'):
|
|
'''Transform [1,2,3,5,7,8,9] into "1-3,5,7-9"'''
|
|
i = 0
|
|
spans = []
|
|
while i < len(nums):
|
|
j = i + 1
|
|
while j < len(nums) and nums[j]-nums[i] == j-i:
|
|
j += 1
|
|
spans += [f'{nums[i]}{range_delimiter}{nums[j-1]}' if j > i+1 else f'{nums[i]}']
|
|
i = j
|
|
return delimiter.join(spans)
|
|
|
|
def allowed_prefixes(self, service):
|
|
'''Return a list of allowed IP prefixes for the given service'''
|
|
if 'custom_fields' in service:
|
|
service = service['custom_fields']
|
|
if prefixes := service.get('allowed_prefixes'):
|
|
yield from self.nb.ipam.prefixes.filter(id=[prefix['id'] for prefix in prefixes])
|
|
if vlans := service.get('allowed_vlans'):
|
|
yield from self.nb.ipam.prefixes.filter(vlan_id=[vlan['id'] for vlan in vlans])
|
|
if clusters := service.get('allowed_clusters'):
|
|
for device in self.nb.dcim.devices.filter(cluster_id=[cluster['id'] for cluster in clusters]):
|
|
yield from self.nb.ipam.ip_addresses.filter(role='loopback', device_id=device.id)
|