bundlewrap/bundles/routeros/metadata.py

import re
from json import load
from os.path import join


with open(join(repo.path, 'configs', 'netbox', f'{node.name}.json')) as f:
    netbox = load(f)

ips = {}
ports = {}
vlans = {
    v['name']: {
        'id': v['vid'],
        'delete': False,
        'tagged': set(),
        'untagged': set(),
    }
    for v in netbox['vlans']
}

for port, conf in netbox['interfaces'].items():
    for ip in conf['ips']:
        ips[ip] = {'interface': port}

    if conf['type'].lower() == 'virtual':
        # these are VLAN interfaces (for management IPs)
        if conf['ips']:
            # this makes management services available in the VLAN
            try:
                vlans[port]['tagged'].add('bridge')
            except KeyError:
                raise ValueError(
                    f'name of virtual interface "{port}" on {node.name} '
                    f'matches none of the known VLANs: {list(vlans.keys())} '
                    '(you probably need to rename the interface in Netbox '
                    'and/or run netbox-dump)'
                )
        # We do not create the actual VLAN interface here, that
        # happens automatically in items.py.
        continue
    elif not conf['enabled'] or not conf['mode']:
        # disable unconfigured ports
        ports[port] = {
            'disabled': True,
            'description': conf.get('description', ''),
        }
        # dont add vlans for this port
        continue
    else:
        ports[port] = {
            'disabled': False,
            'description': conf.get('description', ''),
        }
        if conf.get('ips', []):
            ports[port]['ips'] = set(conf['ips'])
        if conf['type'] in (
            '1000base-t',
            '10gbase-x-sfpp',
            'A_1000BASE_T',
            'A_10GBASE_X_SFPP',
        ):
            ports[port]['hw'] = True

    if conf['untagged_vlan']:
        vlans[conf['untagged_vlan']]['untagged'].add(port)
        if conf['ips']:
            # this makes management services available in the VLAN
            vlans[conf['untagged_vlan']]['tagged'].add('bridge')

    # tagged

    if conf['mode'] in ('TAGGED_ALL', 'tagged-all'):
        tagged = set(vlans.keys()) - {conf['untagged_vlan']}
    else:
        tagged = conf['tagged_vlans']

    for vlan in tagged:
        vlans[vlan]['tagged'].add(port)

        # this makes management services available in the VLAN
        if conf['ips']:
            vlans[vlan]['tagged'].add('bridge')

defaults = {
    'icinga2_api': {
        'routeros': {
            'services': {
                'TEMPERATURE': {
                    'check_command': 'snmp',
                    'vars.snmp_oid': '1.3.6.1.4.1.14988.1.1.3.11.0',
                    'vars.snmp_version': '2c',
                    'vars.snmp_community': 'public',
                    'vars.warn': '@750:799',  # 1/10 °C
                    'vars.crit': '@800:9999',
                },
            },
        },
    },
    'routeros': {
        'ips': ips,
        'ports': ports,
        'vlans': vlans,
    },
}


@metadata_reactor.provides('routeros/gateway')
def gateway(metadata):
    ip_pattern = re.compile(r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.)\d{1,3}')
    gateway = ip_pattern.match(node.hostname).group(1) + '1'

    return {
        'routeros': {
            'gateway': gateway,
        },
    }