7
\$\begingroup\$

INTRO

About SNMP

SNMP stands for simple network management protocol. It is a way that servers can share information about their current state, and also a channel through which an administer can modify pre-defined values. SNMP is a protocol that is implemented on the application layer of the networking stack (click here to learn about networking layers). The protocol was created as a way of gathering information from very different systems in a consistent manner.

You can read more about SNMP, OIDs and SNMP methods in the above link. As a summary, this script uses:

  • snmp version 2c
  • snmpwalk as the main snmp method to get the data
  • the data fetched from a device is fetched from a specific OID (more LLDP OIDs can be found here)

About LLDP

The Link Layer Discovery Protocol (LLDP) is a vendor neutral layer 2 protocol that can be used by a station attached to a specific LAN segment to advertise its identity and capabilities and to also receive same from a physically adjacent layer 2 peer.


Combining SNMP and LLDP using python

The purpose of my program is, by using Python3.6 and provided a file of switches data (community string, snmp port and switch ip), to return the neighbours data (local and remote port + name of the neighbours) for all the switches in the file.

Example config file:

community_string1, snmp_port1, ip1
community_string2, snmp_port2, ip2
community_string3, snmp_port3, ip3

Example output:

[
    {
        "name1": {
            "ip": "ip1",
            "neighbours": [
                {
                    "neighbour_name1": "neighbour_name1",
                    "local_port1": "local_port1",
                    "remote_port1": "remote_port1"
                },
                {
                    "neighbour_name2": "neighbour_name2",
                    "local_port2": "local_port2",
                    "remote_port2": "remote_port2"
                },
                {
                    "neighbour_name3": "neighbour_name3",
                    "local_port3": "local_port3",
                    "remote_port3": "remote_port3"
                },
            ]
        },
        "name2":  {data here},
        "name3":  {data here},
    }
]

Explaining the output

  • name1 represents the name of the switch from the first line of the config file (which is retrieved by doing a snmp walk for PARENT_NAME_OID)
  • ip1 represents the ip of the switch from the first line of the config file (this is taken as is from the config file)
  • the neighbours are all retrieved via snmp using specific OIDs (see code below).

I thought this JSON output format is the most relevant but if you have better ideas, I'd like to hear.


The code

Now, the code is a bit messy but it does its job using the pysnmp library which can be easily installed via pip. It receives the config file as a CLI argument, parses it and the processes the info in it.

"""
Parse a file which contains switches information (community, snmp_port, ip)
and query those devices (neighbours information) via LLDP. Return the data
as a JSON object.
"""

import argparse
import itertools
import pprint
import os
import re

from pysnmp.hlapi import *


NEIGHBOUR_PORT_OID = '1.0.8802.1.1.2.1.4.1.1.8.0'
NEIGHBOUR_NAME_OID = '1.0.8802.1.1.2.1.4.1.1.9'
PARENT_NAME_OID = '1.0.8802.1.1.2.1.3.3'


class MissingOidParameter(Exception):
    """
    Custom exception used when the OID is missing.
    """
    pass


def is_file_valid(filepath):
    """
    Check if a file exists or not.

    Args:
        filepath (str): Path to the switches file
    Returns:
        filepath or raise exception if invalid
    """

    if not os.path.exists(filepath):
        raise ValueError('Invalid filepath')
    return filepath


def get_cli_arguments():
    """
    Simple command line parser function.
    """

    parser = argparse.ArgumentParser(description="")
    parser.add_argument(
        '-f',
        '--file',
        type=is_file_valid,
        help='Path to the switches file'
    )
    return parser


def get_switches_from_file():
    """Return data as a list from a file.

    The file format is the following:

    community_string1, snmp_port1, ip1
    community_string2, snmp_port2, ip2
    community_string3, snmp_port3, ip3

    The output:

    [
        {"community": "community_string1", "snmp_port": "snmp_port1", "ip": "ip1"},
        {"community": "community_string2", "snmp_port": "snmp_port2", "ip": "ip2"},
        {"community": "community_string3", "snmp_port": "snmp_port3", "ip": "ip3"},
    ]
    """

    args = get_cli_arguments().parse_args()
    switches_info = []
    with open(args.file) as switches_info_fp:
        for line in switches_info_fp:
            line = line.rstrip().split(',')
            switches_info.append({
                'community': line[0].strip(),
                'snmp_port': line[1].strip(),
                'ip': line[2].strip(),
            })
    return switches_info


def parse_neighbours_ports_result(result):
    """
    One line of result looks like this:

    result = iso.0.8802.1.1.2.1.4.1.1.8.0.2.3 = 2

    Where the last "2" from the OID is the local port and the value
    after '=' is the remote port (2)
    """
    if not result:
        raise MissingOidParameter('No OID provided.')

    value = result.split(' = ')
    if not value:
        return 'Missing entire value for OID={}'.format(result)
    else:
        oid, port = value
        local_port = re.search(r'{}\.(\d+)'.format(NEIGHBOUR_PORT_OID[2:]), oid).group(1)

        if port:
            remote_port = re.search(r'(\d+)', port).group(1)
        else:
            remote_port = 'Unknown'

    return 'local_port', local_port, 'remote_port', remote_port


def parse_parent_name(result):
    """
    One line of result looks like this:

    result = iso.0.8802.1.1.2.1.3.3.0 = Switch01

    The name of the parent is "Switch01"
    """

    if not result:
        raise MissingOidParameter('No OID provided.')

    value = result.split(' = ')
    if not value:
        return 'Missing entire value for OID={}'.format(result)
    else:
        return 'Unknown' if not value[-1] else value[-1]


def parse_neighbour_names_results(result):
    """
    One line of result looks like this:

    result = iso.0.8802.1.1.2.1.4.1.1.9.0.2.3 = HP-2920-24G

    The name of the parent is "Switch01"
    """

    if not result:
        raise MissingOidParameter('No OID provided.')

    value = result.split(' = ')
    if not value:
        return 'Missing entire value for OID={}'.format(result)
    else:
        return ('name', 'Unknown') if not value[-1] else ('name', value[-1])


def main():
    data = {}
    switches_filedata = get_switches_from_file()

    for switch in switches_filedata:
        community = switch.get('community')
        snmp_port = switch.get('snmp_port')
        ip = switch.get('ip')

        name = ''
        for (error_indication, error_status, error_index, var_binds) in nextCmd(
                SnmpEngine(),
                CommunityData(community),
                UdpTransportTarget((ip, snmp_port)),
                ContextData(),
                ObjectType(ObjectIdentity(PARENT_NAME_OID)),
                lexicographicMode=False
        ):
            # this should always return one result
            name = parse_parent_name(str(var_binds[0]))

        if not name:
            print('Could not retrieve name of switch. Moving to the next one...')
            continue

        neighbour_names = []
        neighbour_local_remote_ports = []

        for (error_indication, error_status, error_index, var_binds) in nextCmd(
                SnmpEngine(),
                CommunityData(community),
                UdpTransportTarget((ip, snmp_port)),
                ContextData(),
                ObjectType(ObjectIdentity(NEIGHBOUR_NAME_OID)),
                lexicographicMode=False
        ):
            for var_bind in var_binds:
                neighbour_names.append(
                    parse_neighbour_names_results(str(var_bind))
                )

        for (error_indication, error_status, error_index, var_binds) in nextCmd(
                SnmpEngine(),
                CommunityData(community),
                UdpTransportTarget((ip, snmp_port)),
                ContextData(),
                ObjectType(ObjectIdentity(NEIGHBOUR_PORT_OID)),
                lexicographicMode=False
        ):
            for var_bind in var_binds:
                neighbour_local_remote_ports.append(
                    parse_neighbours_ports_result(str(var_bind))
                )

        neighbours = []
        for a, b in itertools.zip_longest(
            neighbour_names,
            neighbour_local_remote_ports,
            fillvalue='unknown'
        ):
            neighbours.append({
                a[0]: a[1],
                b[0]: b[1],
                b[2]: b[3]
            })

        data[name] = {
            'ip': ip,
            'neighbors': neighbours
        }

    return data


if __name__ == '__main__':
    all_data = main()
    pprint.pprint(all_data, indent=4)

What I'm especially looking after:

  • better / more performant ways of using pysnmp's functionality (perhaps I can do only one SNMP walk to store all the data and then from there get the needed data for all the OIDs) - kinda like we're doing when parsing lxmls html tree.
  • better ways of structuring my code
  • improvements regarding the names of the functions/names
  • I tried to stick to PEP8 but I didn't really focused on it. I'm familiar with almost everything related to it and so I would like you guys not to focus on this too much.
\$\endgroup\$
  • \$\begingroup\$ NOTE: Unless you have access to some switches within a network this can hardly be tested and I'm sorry that I can't provide a test environment where you can play with it. I hope you guys will still have the pleasure to review this ^^ \$\endgroup\$ – Grajdeanu Alex. Mar 12 at 16:45
5
+300
\$\begingroup\$

Type hints

PEP484 type hints will help; an example:

def is_file_valid(filepath: str) -> bool:

Function contracts

is_file_valid isn't actually what's happening. You're using this function to do two things:

  • Cast an input string to whatever the program needs through the argparse system
  • Verify that the input string is correct

The documentation shows how this should actually be done:

parser.add_argument('bar', type=open)

This will call open and give you back a file object, or fail if the file doesn't exist.

Parsing

The get_switches_from_file can use tuple unpacking:

community, port, ip = (t.strip() for t in line.split(','))

This has the added advantage that irregular lines with more than three parts will cause an error rather than being silently ignored.

Better yet, delegate this to a class:

class Switch:
    def __init__(self, line: str):
        self.community, self.port, self.ip = (
            field.strip() for field in line.split(',')
        )

Avoid iterative concatenation

Rather than maintaining switches_info, simply yield each dictionary from the inner loop. This will cause your method to go from O(n) memory to O(1) memory, at a probable slight cost to runtime.

If you use the Switch class above, this could look like

    args = get_cli_arguments().parse_args()
    with args.file as switches_info_fp:
        for line in switches_info_fp:
            yield Switch(line)

Separation of concerns

parse_neighbours_ports_result has an odd return format. It's not clear that the first and third string are useful. Either return a 2-tuple with the actual port values, or return a named tuple or class instance.

Implicit tuple unpacking

for (error_indication, error_status, error_index, var_binds)

can lose the parens.

Imports

To abbreviate your code, do some from x import y:

from argparse import ArgumentParser
from itertools import zip_longest
| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Those are some nice suggestions but the is_file_valid has to return something since it's used in one of parser.add_arguments params. I'd also love to hear some other suggestions regarding the structure of the whole project :) \$\endgroup\$ – Grajdeanu Alex. Mar 15 at 17:11
  • \$\begingroup\$ is_file_valid has to return something - Sure; I've edited my answer \$\endgroup\$ – Reinderien Mar 15 at 19:46
  • \$\begingroup\$ I'd also love to hear some other suggestions regarding the structure of the whole project - use a helper class for named parse elements; edited \$\endgroup\$ – Reinderien Mar 15 at 20:13

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.