path: root/src/conf_mode/lldp.py

#!/usr/bin/env python3
#
# Copyright (C) 2017-2019 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import re
import sys
import os
import jinja2

from copy import deepcopy
from vyos.config import Config
from vyos.validate import is_addr_assigned,is_loopback_addr
from vyos import ConfigError

# Please be careful if you edit the template.
config_file = "/etc/default/lldpd"
lldp_tmpl = """
### Autogenerated by lldp.py ###
DAEMON_ARGS="-M 4{% if options.snmp %} -x{% endif %}{% if options.cdp %} -c{% endif %}{% if options.edp %} -e{% endif %}{% if options.fdp %} -f{% endif %}{% if options.sonmp %} -s{% endif %}"

"""

vyos_config_file = "/etc/lldpd.d/01-vyos.conf"
vyos_tmpl = """
### Autogenerated by lldp.py ###

configure system platform VyOS
configure system description "VyOS {{ options.description }}"
{% if options.listen_on -%}
configure system interface pattern "{{ ( options.listen_on | select('equalto','all') | map('replace','all','*') | list + options.listen_on | select('equalto','!all') | map('replace','!all','!*') | list + options.listen_on | reject('equalto','all') | reject('equalto','!all') | list ) | unique | join(",") }}"
{%- endif %}
{% if options.mgmt_addr -%}
configure system ip management pattern {{ options.mgmt_addr | join(",") }}
{%- endif %}
{%- for loc in location -%}
{%- if loc.elin %}
configure ports {{ loc.name }} med location elin "{{ loc.elin }}"
{%- endif %}
{%- if loc.coordinate_based %}
configure ports {{ loc.name }} med location coordinate {% if loc.coordinate_based.latitude %}latitude {{ loc.coordinate_based.latitude }}{% endif %} {% if loc.coordinate_based.longitude %}longitude {{ loc.coordinate_based.longitude }}{% endif %} {% if loc.coordinate_based.altitude %}altitude {{ loc.coordinate_based.altitude }} m{% endif %} {% if loc.coordinate_based.datum %}datum {{ loc.coordinate_based.datum }}{% endif %}
{%- endif %}


{% endfor %}
"""

default_config_data = {
    "options": '',
    "interface_list": '',
    "location": ''
}

def get_options(config):
    options = {}
    config.set_level('service lldp')

    options['listen_vlan'] = config.exists('listen-vlan')
    options['mgmt_addr'] = []
    for addr in config.return_values('management-address'):
        if is_addr_assigned(addr) and not is_loopback_addr(addr):
            options['mgmt_addr'].append(addr)
        else:
            message = 'WARNING: LLDP management address {0} invalid - '.format(addr)
            if is_loopback_addr(addr):
                message += '(loopback address).'
            else:
                message += 'address not found.'
            print(message)

    snmp = config.exists('snmp enable')
    options["snmp"] = snmp
    if snmp:
        config.set_level('')
        options["sys_snmp"] = config.exists('service snmp')
        config.set_level('service lldp')

    config.set_level('service lldp legacy-protocols')
    options['cdp'] = config.exists('cdp')
    options['edp'] = config.exists('edp')
    options['fdp'] = config.exists('fdp')
    options['sonmp'] = config.exists('sonmp')

    # start with an unknown version information
    options['description'] = 'unknown'
    options['listen_on'] = []

    return options

def get_interface_list(config):
    config.set_level('service lldp')
    intfs_names = config.list_nodes('interface')
    if len(intfs_names) < 0:
        return 0

    interface_list = []
    for name in intfs_names:
        config.set_level('service lldp interface {0}'.format(name))
        disable = config.exists('disable')
        intf = {
            'name': name,
            'disable': disable
        }
        interface_list.append(intf)
    return interface_list


def get_location_intf(config, name):
    path = 'service lldp interface {0}'.format(name)
    config.set_level(path)

    config.set_level('{} location'.format(path))
    elin = ''
    coordinate_based = {}

    if config.exists('elin'):
        elin = config.return_value('elin')

    if config.exists('coordinate-based'):
        config.set_level('{} location coordinate-based'.format(path))

        coordinate_based['latitude'] = config.return_value('latitude')
        coordinate_based['longitude'] = config.return_value('longitude')

        coordinate_based['altitude'] = '0'
        if config.exists('altitude'):
            coordinate_based['altitude'] = config.return_value('altitude')

        coordinate_based['datum'] = 'WGS84'
        if config.exists('datum'):
            coordinate_based['datum'] = config.return_value('datum')

    intf = {
        'name': name,
        'elin': elin,
        'coordinate_based': coordinate_based

    }
    return intf


def get_location(config):
    config.set_level('service lldp')
    intfs_names = config.list_nodes('interface')
    if len(intfs_names) < 0:
        return 0

    if config.exists('disable'):
        return 0

    intfs_location = []
    for name in intfs_names:
        intf = get_location_intf(config, name)
        intfs_location.append(intf)

    return intfs_location


def get_config():
    lldp = deepcopy(default_config_data)
    conf = Config()
    if not conf.exists('service lldp'):
        return None
    else:
        lldp['options'] = get_options(conf)
        lldp['interface_list'] = get_interface_list(conf)
        lldp['location'] = get_location(conf)

        return lldp


def verify(lldp):
    # bail out early - looks like removal from running config
    if lldp is None:
        return

    # check location
    for location in lldp['location']:
        # check coordinate-based
        if len(location['coordinate_based']) > 0:
            # check longitude and latitude
            if not location['coordinate_based']['longitude']:
                raise ConfigError('Must define longitude for interface {0}'.format(location['name']))

            if not location['coordinate_based']['latitude']:
                raise ConfigError('Must define latitude for interface {0}'.format(location['name']))

            if not re.match(r'^(\d+)(\.\d+)?[nNsS]$', location['coordinate_based']['latitude']):
                raise ConfigError('Invalid location for interface {0}:\n' \
                                  'latitude should be a number followed by S or N'.format(location['name']))

            if not re.match(r'^(\d+)(\.\d+)?[eEwW]$', location['coordinate_based']['longitude']):
                raise ConfigError('Invalid location for interface {0}:\n' \
                                  'longitude should be a number followed by E or W'.format(location['name']))

            # check altitude and datum if exist
            if location['coordinate_based']['altitude']:
                if not re.match(r'^[-+0-9\.]+$', location['coordinate_based']['altitude']):
                    raise ConfigError('Invalid location for interface {0}:\n' \
                                      'altitude should be a positive or negative number'.format(location['name']))

            if location['coordinate_based']['datum']:
                if not re.match(r'^(WGS84|NAD83|MLLW)$', location['coordinate_based']['datum']):
                    raise ConfigError("Invalid location for interface {0}:\n' \
                                      'datum should be WGS84, NAD83, or MLLW".format(location['name']))

        # check elin
        elif location['elin']:
            if not re.match(r'^[0-9]{10,25}$', location['elin']):
                raise ConfigError('Invalid location for interface {0}:\n' \
                                  'ELIN number must be between 10-25 numbers'.format(location['name']))

    # check options
    if lldp['options']['snmp']:
        if not lldp['options']['sys_snmp']:
            raise ConfigError('SNMP must be configured to enable LLDP SNMP')


def generate(lldp):
    # bail out early - looks like removal from running config
    if lldp is None:
        return

    with open('/opt/vyatta/etc/version', 'r') as f:
        tmp = f.read()
        lldp['options']['description'] = tmp.split()[1]


    # generate listen on interfaces
    for intf in lldp['interface_list']:
        tmp = ''
        # add exclamation mark if interface is disabled
        if intf['disable']:
            tmp = '!'

        tmp += intf['name']
        lldp['options']['listen_on'].append(tmp)

    # generate /etc/default/lldpd
    tmpl = jinja2.Template(lldp_tmpl)
    config_text = tmpl.render(lldp)
    with open(config_file, 'w') as f:
        f.write(config_text)

    # generate /etc/lldpd.d/01-vyos.conf
    tmpl = jinja2.Template(vyos_tmpl)
    config_text = tmpl.render(lldp)
    with open(vyos_config_file, 'w') as f:
        f.write(config_text)


def apply(lldp):
    if lldp:
        # start/restart lldp service
        os.system('sudo systemctl restart lldpd.service')
    else:
        # LLDP service has been terminated
        os.system('sudo systemctl stop lldpd.service')
        os.unlink(config_file)
        os.unlink(vyos_config_file)

if __name__ == '__main__':
    try:
        c = get_config()
        verify(c)
        generate(c)
        apply(c)
    except ConfigError as e:
        print(e)
        sys.exit(1)