Set up automated network topology discovery with SNMP and LLDP for infrastructure mapping

Update system packages

Start by updating your package manager to ensure you have the latest security patches and package versions.

sudo apt update && sudo apt upgrade -y

sudo dnf update -y

Install SNMP tools and LLDP utilities

Install the core SNMP utilities, LLDP daemon, and network scanning tools needed for device discovery and protocol communication.

sudo apt install -y snmp snmp-mibs-downloader lldpd nmap python3-pip python3-venv git

sudo dnf install -y net-snmp net-snmp-utils lldpd nmap python3-pip python3-virtualenv git

Configure SNMP daemon

Set up the SNMP daemon with proper community strings and access controls for network device communication.

# Community string configuration
rocommunity public 127.0.0.1
rocommunity netdiscovery 10.0.0.0/8
rocommunity netdiscovery 172.16.0.0/12
rocommunity netdiscovery 192.168.0.0/16

# System information
sysLocation "Network Operations Center"
sysContact "admin@example.com"
sysServices 72

# Access control
com2sec readonly default netdiscovery
group MyROGroup v1 readonly
group MyROGroup v2c readonly

# OID access restrictions
view all included .1.3.6.1.2.1.1
view all included .1.3.6.1.2.1.2
view all included .1.3.6.1.2.1.4
view all included .1.3.6.1.4.1

access MyROGroup "" any noauth exact all none none

# Disable SNMP v1/v2c write access
rwcommunity private 127.0.0.1

Configure LLDP daemon

Enable LLDP on network interfaces to discover directly connected devices and their capabilities.

# Configure LLDP daemon
configure lldp tx-interval 30
configure lldp tx-hold 4
configure system hostname discovery-server
configure system description "Network Discovery Server"
configure system platform "Linux"

# Enable CDP compatibility for Cisco devices
configure lldp custom-tlv oui 00,00,0c subtype 1

# Interface configuration
configure ports eth0 lldp portidsubtype local eth0
configure med fast-start enable

Enable and start services

Start the SNMP and LLDP daemons and enable them to start automatically on boot.

sudo systemctl enable --now snmpd
sudo systemctl enable --now lldpd
sudo systemctl status snmpd lldpd

Create Python virtual environment

Set up an isolated Python environment for the network discovery scripts and required libraries.

sudo mkdir -p /opt/network-discovery
sudo chown $(whoami):$(whoami) /opt/network-discovery
cd /opt/network-discovery
python3 -m venv venv
source venv/bin/activate
pip install pysnmp pysnmp-mibs netaddr netifaces graphviz networkx matplotlib

Create network scanner script

Build the core network discovery script that performs SNMP walks and LLDP neighbor detection across your network ranges.

#!/usr/bin/env python3
import json
import subprocess
import socket
from ipaddress import IPv4Network, AddressValueError
from pysnmp.hlapi import *
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class NetworkScanner:
    def __init__(self, community='public', timeout=5, retries=1):
        self.community = community
        self.timeout = timeout
        self.retries = retries
        self.discovered_devices = {}
        
    def snmp_get(self, target, oid):
        """Perform SNMP GET operation"""
        try:
            for (errorIndication, errorStatus, errorIndex, varBinds) in nextCmd(
                SnmpEngine(),
                CommunityData(self.community),
                UdpTransportTarget((target, 161), timeout=self.timeout, retries=self.retries),
                ContextData(),
                ObjectType(ObjectIdentity(oid)),
                lexicographicMode=False,
                maxRows=50):
                
                if errorIndication:
                    break
                elif errorStatus:
                    break
                else:
                    return [str(varBind[1]) for varBind in varBinds]
            return []
        except Exception as e:
            logger.debug(f"SNMP error for {target}: {e}")
            return []
    
    def get_device_info(self, ip):
        """Extract device information via SNMP"""
        device_info = {
            'ip': ip,
            'hostname': '',
            'system_desc': '',
            'uptime': '',
            'interfaces': [],
            'neighbors': [],
            'vendor': '',
            'model': ''
        }
        
        # System information OIDs
        hostname_result = self.snmp_get(ip, '1.3.6.1.2.1.1.5.0')  # sysName
        if hostname_result:
            device_info['hostname'] = hostname_result[0]
        
        desc_result = self.snmp_get(ip, '1.3.6.1.2.1.1.1.0')  # sysDescr
        if desc_result:
            device_info['system_desc'] = desc_result[0]
            # Parse vendor info from description
            desc_lower = desc_result[0].lower()
            if 'cisco' in desc_lower:
                device_info['vendor'] = 'Cisco'
            elif 'juniper' in desc_lower:
                device_info['vendor'] = 'Juniper'
            elif 'hp' in desc_lower or 'hewlett' in desc_lower:
                device_info['vendor'] = 'HP'
            elif 'dell' in desc_lower:
                device_info['vendor'] = 'Dell'
        
        uptime_result = self.snmp_get(ip, '1.3.6.1.2.1.1.3.0')  # sysUpTime
        if uptime_result:
            device_info['uptime'] = uptime_result[0]
        
        # Interface information
        interfaces = self.snmp_get(ip, '1.3.6.1.2.1.2.2.1.2')  # ifDescr
        if interfaces:
            device_info['interfaces'] = interfaces[:10]  # Limit to first 10
        
        # LLDP neighbors (if supported)
        lldp_neighbors = self.snmp_get(ip, '1.0.8802.1.1.2.1.4.1.1.9')  # lldpRemSysName
        if lldp_neighbors:
            device_info['neighbors'] = lldp_neighbors
        
        return device_info
    
    def scan_network_range(self, network_range, max_workers=50):
        """Scan a network range for SNMP-enabled devices"""
        logger.info(f"Scanning network range: {network_range}")
        
        try:
            network = IPv4Network(network_range, strict=False)
        except AddressValueError:
            logger.error(f"Invalid network range: {network_range}")
            return
        
        # Skip network and broadcast addresses for /24 and smaller
        if network.prefixlen >= 24:
            hosts = list(network.hosts())
        else:
            hosts = list(network)
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_ip = {executor.submit(self.check_snmp_device, str(ip)): ip for ip in hosts}
            
            for future in as_completed(future_to_ip, timeout=300):
                ip = future_to_ip[future]
                try:
                    result = future.result()
                    if result:
                        self.discovered_devices[str(ip)] = result
                        logger.info(f"Discovered device: {result['hostname']} ({ip})")
                except Exception as e:
                    logger.debug(f"Error scanning {ip}: {e}")
    
    def check_snmp_device(self, ip):
        """Check if device responds to SNMP and gather info"""
        # Quick port check first
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            sock.settimeout(2)
            sock.connect((ip, 161))
            sock.close()
        except:
            return None
        
        # Try SNMP
        device_info = self.get_device_info(ip)
        if device_info['hostname'] or device_info['system_desc']:
            return device_info
        return None
    
    def get_local_lldp_neighbors(self):
        """Get LLDP neighbors from local daemon"""
        try:
            result = subprocess.run(['lldpctl', '-f', 'json'], 
                                  capture_output=True, text=True, timeout=10)
            if result.returncode == 0:
                return json.loads(result.stdout)
        except Exception as e:
            logger.error(f"Error getting LLDP neighbors: {e}")
        return {}
    
    def generate_topology_map(self):
        """Generate network topology relationships"""
        topology = {
            'devices': self.discovered_devices,
            'connections': [],
            'generated_at': time.strftime('%Y-%m-%d %H:%M:%S')
        }
        
        # Add LLDP-discovered connections
        for device_ip, device_info in self.discovered_devices.items():
            for neighbor in device_info.get('neighbors', []):
                # Find neighbor device by hostname
                for neighbor_ip, neighbor_info in self.discovered_devices.items():
                    if neighbor_info.get('hostname', '').lower() == neighbor.lower():
                        connection = {
                            'source': device_ip,
                            'target': neighbor_ip,
                            'type': 'lldp',
                            'source_name': device_info.get('hostname', device_ip),
                            'target_name': neighbor_info.get('hostname', neighbor_ip)
                        }
                        if connection not in topology['connections']:
                            topology['connections'].append(connection)
        
        return topology
    
    def save_results(self, filename='network_topology.json'):
        """Save discovery results to JSON file"""
        topology = self.generate_topology_map()
        with open(filename, 'w') as f:
            json.dump(topology, f, indent=2)
        logger.info(f"Results saved to {filename}")
        return topology

if __name__ == "__main__":
    scanner = NetworkScanner(community='netdiscovery')
    
    # Define networks to scan
    networks = [
        '192.168.1.0/24',
        '10.0.1.0/24'
    ]
    
    for network in networks:
        scanner.scan_network_range(network)
    
    # Save results
    topology = scanner.save_results('/opt/network-discovery/topology.json')
    print(f"Discovered {len(topology['devices'])} devices")
    print(f"Found {len(topology['connections'])} connections")

Create topology visualizer

Build a script that creates visual network diagrams from the discovered topology data using NetworkX and Matplotlib.

#!/usr/bin/env python3
import json
import networkx as nx
import matplotlib.pyplot as plt
import matplotlib.patches as patches
from datetime import datetime
import argparse

class TopologyVisualizer:
    def __init__(self):
        self.graph = nx.Graph()
        self.vendor_colors = {
            'Cisco': '#1BA0D7',
            'Juniper': '#84BD00', 
            'HP': '#0096D6',
            'Dell': '#007DB8',
            'Unknown': '#666666'
        }
    
    def load_topology(self, filename):
        """Load topology data from JSON file"""
        try:
            with open(filename, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            print(f"Error: Topology file {filename} not found")
            return None
        except json.JSONDecodeError:
            print(f"Error: Invalid JSON in {filename}")
            return None
    
    def build_graph(self, topology_data):
        """Build NetworkX graph from topology data"""
        devices = topology_data.get('devices', {})
        connections = topology_data.get('connections', [])
        
        # Add nodes
        for ip, device_info in devices.items():
            hostname = device_info.get('hostname', ip)
            vendor = device_info.get('vendor', 'Unknown')
            interfaces = len(device_info.get('interfaces', []))
            
            self.graph.add_node(ip, 
                              hostname=hostname,
                              vendor=vendor,
                              interfaces=interfaces,
                              system_desc=device_info.get('system_desc', '')[:50])
        
        # Add edges
        for connection in connections:
            source = connection['source']
            target = connection['target']
            if source in self.graph and target in self.graph:
                self.graph.add_edge(source, target, 
                                  connection_type=connection.get('type', 'unknown'))
    
    def create_visualization(self, output_file='network_topology.png', layout='spring'):
        """Create network topology visualization"""
        if len(self.graph.nodes()) == 0:
            print("No devices to visualize")
            return
        
        # Set up the plot
        plt.figure(figsize=(16, 12))
        ax = plt.gca()
        
        # Choose layout algorithm
        if layout == 'spring':
            pos = nx.spring_layout(self.graph, k=3, iterations=50)
        elif layout == 'circular':
            pos = nx.circular_layout(self.graph)
        elif layout == 'kamada_kawai':
            pos = nx.kamada_kawai_layout(self.graph)
        else:
            pos = nx.spring_layout(self.graph)
        
        # Draw edges
        nx.draw_networkx_edges(self.graph, pos, 
                              edge_color='#CCCCCC', 
                              width=2, 
                              alpha=0.7)
        
        # Draw nodes by vendor
        for vendor, color in self.vendor_colors.items():
            vendor_nodes = [n for n, d in self.graph.nodes(data=True) 
                           if d.get('vendor', 'Unknown') == vendor]
            if vendor_nodes:
                nx.draw_networkx_nodes(self.graph, pos,
                                     nodelist=vendor_nodes,
                                     node_color=color,
                                     node_size=800,
                                     alpha=0.8)
        
        # Add labels
        labels = {}
        for node, data in self.graph.nodes(data=True):
            hostname = data.get('hostname', node)
            if len(hostname) > 15:
                hostname = hostname[:12] + '...'
            labels[node] = f"{hostname}\n({node})"
        
        nx.draw_networkx_labels(self.graph, pos, labels, font_size=8, font_weight='bold')
        
        # Create legend
        legend_elements = []
        for vendor, color in self.vendor_colors.items():
            vendor_nodes = [n for n, d in self.graph.nodes(data=True) 
                           if d.get('vendor', 'Unknown') == vendor]
            if vendor_nodes:
                legend_elements.append(patches.Patch(color=color, label=f'{vendor} ({len(vendor_nodes)})'))
        
        plt.legend(handles=legend_elements, loc='upper left', bbox_to_anchor=(0, 1))
        
        # Add title and info
        plt.title(f"Network Topology Discovery\nGenerated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
                 f"Devices: {len(self.graph.nodes())} | Connections: {len(self.graph.edges())}", 
                 fontsize=14, pad=20)
        
        # Remove axis
        ax.set_axis_off()
        
        # Save the plot
        plt.tight_layout()
        plt.savefig(output_file, dpi=300, bbox_inches='tight', 
                   facecolor='white', edgecolor='none')
        plt.close()
        
        print(f"Network topology visualization saved to {output_file}")
    
    def generate_report(self, topology_data, output_file='network_report.txt'):
        """Generate text report of discovered network"""
        devices = topology_data.get('devices', {})
        connections = topology_data.get('connections', [])
        
        report = []
        report.append("NETWORK TOPOLOGY DISCOVERY REPORT")
        report.append("=" * 50)
        report.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append(f"Total Devices: {len(devices)}")
        report.append(f"Total Connections: {len(connections)}")
        report.append("")
        
        # Device summary by vendor
        vendor_count = {}
        for device_info in devices.values():
            vendor = device_info.get('vendor', 'Unknown')
            vendor_count[vendor] = vendor_count.get(vendor, 0) + 1
        
        report.append("DEVICES BY VENDOR:")
        report.append("-" * 20)
        for vendor, count in sorted(vendor_count.items()):
            report.append(f"{vendor}: {count}")
        report.append("")
        
        # Detailed