NeighSnapshotter

Design Rationale

Why Another Neighbor Table Tool?

Linux provides several ways to query the neighbor table (ARP/NDP cache), but each has significant limitations:

Core Design Principles

1. Direct Kernel Communication
   • Uses RTNetlink sockets (RTM_GETNEIGH) for direct kernel queries
   • No text parsing, no external commands, no proc/sys filesystem
   • Single syscall retrieves all neighbor entries
2. Complete Protocol Coverage
   • IPv4 ARP cache entries
   • IPv6 Neighbor Discovery (NDP) cache
   • Bridge Forwarding Database (FDB)
   • All protocol families with unified interface
3. Full State Information
   • All NUD (Neighbor Unreachability Detection) states
   • Flags (PROXY, ROUTER, OFFLOADED, etc.)
   • Cache timestamps (confirmed, used, updated)
   • Hardware type decoding
4. Type Safety & Validation
   • C structures ensure correct data types
   • Python provides clean, JSON-serializable output
   • Unknown attributes tracked for future-proofing
5. Performance First
   • Compiled C code for netlink parsing
   • Minimal Python overhead
   • Bulk queries (all neighbors in one call)

Use Cases

NeighSnapshotter is designed for:

• Network Monitoring: Track ARP/NDP cache health and changes
• Security Auditing: Detect ARP spoofing, rogue neighbors
• Container Networking: Inspect bridge FDB for container MAC addresses
• Troubleshooting: Debug IPv6 NDP issues, stale entries
• SDN/OpenFlow: Monitor learned MAC addresses in bridge FDB
• Performance Analysis: Identify neighbor resolution delays
• Automation: Integrate neighbor table state into configuration management

Protocol Deep Dive

IPv4 ARP (Address Resolution Protocol)

IPv6 NDP (Neighbor Discovery Protocol)

Bridge FDB (Forwarding Database)

Architecture

Component Overview

Data Flow

1. Python initiates query: ntq.get_neighbors(family='ipv4')
2. CFFI calls C function: nl_send_getneigh(sock, &seq, AF_INET)
3. Kernel receives RTM_GETNEIGH: Triggers neighbor table dump
4. Kernel sends RTM_NEWNEIGH messages: One per neighbor entry with NDA_* attributes
5. C code buffers responses: nl_recv_response() collects all messages
6. C code parses messages: nl_parse_neighbors() extracts data into structures
7. Python receives CFFI structures: Converts to dictionaries
8. Python enriches data: Resolves interface names, decodes addresses
9. Returns JSON-serializable result: Complete neighbor table as Python list

Implementation Details

Key Data Structures

C Structures

// Neighbor cache information
typedef struct {
    unsigned char cacheinfo_confirmed;  // Last confirmed (jiffies)
    unsigned char cacheinfo_used;       // Last used (jiffies)
    unsigned char cacheinfo_updated;    // Last updated (jiffies)
    unsigned char cacheinfo_refcnt;     // Reference count
    int has_cacheinfo_*;                // Presence flags
} neigh_cacheinfo_t;

// Neighbor entry information
typedef struct {
    int ifindex;                        // Interface index
    unsigned char family;               // AF_INET, AF_INET6, AF_BRIDGE
    unsigned char state;                // NUD_* state bits
    unsigned char flags;                // NTF_* flag bits
    unsigned char type;                 // ARPHRD_* hardware type
    unsigned char dst_addr[16];         // Destination IP/MAC
    unsigned char lladdr[32];           // Link-layer address
    int has_dst_addr;                   // Address present
    int has_lladdr;                     // Link-layer address present
    int lladdr_len;                     // Link-layer address length
    unsigned int probes;                // Number of probes sent
    int has_probes;                     // Probes field present
    unsigned short vlan;                // VLAN ID (bridge only)
    int has_vlan;                       // VLAN present
    unsigned int master;                // Master interface (bridge)
    int has_master;                     // Master present
    neigh_cacheinfo_t cacheinfo;        // Cache timing info
    unsigned short unknown_nda_attrs[64]; // Unknown attributes
    int unknown_nda_attrs_count;        // Count of unknown
} neigh_entry_t;

Python Output Structure

{
  "ifindex": 2,
  "family": "ipv4",
  "state": 2,
  "state_name": "REACHABLE",
  "flags": 0,
  "flag_names": [],
  "type": 1,
  "type_name": "ETHER",
  "dst": "192.168.1.1",
  "dst_canonical": "192.168.1.1",
  "lladdr": "aa:bb:cc:dd:ee:ff",
  "cacheinfo": {
    "confirmed": 12345,
    "used": 12340,
    "updated": 12345,
    "refcnt": 1
  }
}

State Machine

Neighbor Unreachability Detection (NUD)

The Linux kernel maintains a state machine for each neighbor entry:

State Transitions

Attribute Parsing

RTNetlink Attribute Format

Netlink attributes use a Type-Length-Value (TLV) format:

struct rtattr {
    unsigned short rta_len;    // Length including header
    unsigned short rta_type;   // Attribute type (NDA_*)
    // Followed by attribute data
};

NDA_* Attribute Types

Hardware Type Decoding

The type field uses ARPHRD_* constants from if_arp.h:

Usage Guide

Command-Line Interface

Basic Usage

# Full JSON output (all neighbor types)
sudo python3 neighsnapshotter.py

# Human-readable summary
sudo python3 neighsnapshotter.py --summary

Protocol-Specific Queries

# IPv4 ARP only
sudo python3 neighsnapshotter.py --arp --summary

# IPv6 NDP only
sudo python3 neighsnapshotter.py --ndp --summary

# Bridge FDB only
sudo python3 neighsnapshotter.py --bridge --summary

Interface Filtering

# All neighbors on eth0
sudo python3 neighsnapshotter.py --interface eth0 --summary

# IPv4 ARP on specific interface
sudo python3 neighsnapshotter.py --arp --interface eth0 --summary

Python API

Basic Query

from neighsnapshotter import NeighborTableQuery

# Query all neighbors
with NeighborTableQuery() as ntq:
    neighbors = ntq.get_neighbors()

# Print all IPv4 ARP entries
for entry in neighbors:
    if entry['family'] == 'ipv4':
        print(f"{entry['dst']} -> {entry.get('lladdr', 'N/A')}")
        print(f"  State: {entry['state_name']}")
        print(f"  Interface: {entry['ifindex']}")

Protocol-Specific Queries

# Query only IPv4 ARP
with NeighborTableQuery() as ntq:
    arp_entries = ntq.get_neighbors(family='ipv4')

# Query only IPv6 NDP
with NeighborTableQuery() as ntq:
    ndp_entries = ntq.get_neighbors(family='ipv6')

# Query only bridge FDB
with NeighborTableQuery() as ntq:
    fdb_entries = ntq.get_neighbors(family='bridge')

State Analysis

# Find all failed entries
with NeighborTableQuery() as ntq:
    neighbors = ntq.get_neighbors()

failed = [n for n in neighbors if 'FAILED' in n['state_name']]
print(f"Found {len(failed)} failed neighbor entries:")
for entry in failed:
    print(f"  {entry.get('dst', 'unknown')} on interface {entry['ifindex']}")

Router Detection

# Find all routers (typically from IPv6 NDP)
with NeighborTableQuery() as ntq:
    neighbors = ntq.get_neighbors(family='ipv6')

routers = [n for n in neighbors if 'ROUTER' in n['flag_names']]
print(f"Found {len(routers)} routers:")
for router in routers:
    print(f"  {router['dst']}")
    print(f"    MAC: {router.get('lladdr', 'N/A')}")
    print(f"    State: {router['state_name']}")

Bridge FDB Analysis

import socket

# Analyze bridge FDB entries
with NeighborTableQuery() as ntq:
    fdb = ntq.get_neighbors(family='bridge')

# Group by interface
by_interface = {}
for entry in fdb:
    ifindex = entry['ifindex']
    try:
        ifname = socket.if_indextoname(ifindex)
    except:
        ifname = f"if{ifindex}"
    
    if ifname not in by_interface:
        by_interface[ifname] = []
    by_interface[ifname].append(entry)

# Print statistics
for ifname, entries in sorted(by_interface.items()):
    print(f"{ifname}: {len(entries)} MAC addresses")
    
    # Count by state
    states = {}
    for entry in entries:
        state = entry['state_name']
        states[state] = states.get(state, 0) + 1
    
    for state, count in sorted(states.items()):
        print(f"  {state}: {count}")

Cache Health Monitoring

import time

# Monitor cache timing
with NeighborTableQuery() as ntq:
    neighbors = ntq.get_neighbors(family='ipv4')

current_time = int(time.time())

for entry in neighbors:
    if 'cacheinfo' in entry:
        cache = entry['cacheinfo']
        
        # Convert jiffies to seconds (approximate)
        # Note: Actual conversion depends on kernel HZ value
        last_used_sec = cache.get('used', 0) / 100.0
        
        if last_used_sec < 60:
            status = "recently active"
        elif last_used_sec < 300:
            status = "idle"
        else:
            status = "stale"
        
        print(f"{entry['dst']}: {status} (last used ~{last_used_sec:.0f}s ago)")

API Reference

NeighborTableQuery Class

Constructor

NeighborTableQuery(capture_unknown_attrs=True)

Parameters:

• capture_unknown_attrs (bool): Track unknown netlink attributes (default: True)

get_neighbors()

get_neighbors(family: Optional[str] = None) -> List[Dict[str, Any]]

Parameters:

• family (str|None): Protocol family filter
  • 'ipv4': IPv4 ARP only
  • 'ipv6': IPv6 NDP only
  • 'bridge': Bridge FDB only
  • None: All neighbor types (default)

Returns: List of neighbor entry dictionaries

Raises:

• ValueError: Invalid family parameter
• RuntimeError: Failed to create socket or query kernel

Neighbor Entry Dictionary

Common Fields

Optional Fields

Cache Info Dictionary

Troubleshooting

Common Issues

"Failed to create netlink socket"

"Python 3.12+ requires setuptools"

Empty Neighbor Table

IPv6 Shows NOARP Instead of NDP States

Debugging

Enable Verbose Output

import logging
logging.basicConfig(level=logging.DEBUG)

with NeighborTableQuery() as ntq:
    neighbors = ntq.get_neighbors()

Check Unknown Attributes

# See what attributes the library doesn't recognize
with NeighborTableQuery() as ntq:
    neighbors = ntq.get_neighbors()

for entry in neighbors:
    if 'unknown_nda_attrs_decoded' in entry:
        print(f"Unknown attrs: {entry['unknown_nda_attrs_decoded']}")

Validate Against ip Command

# Compare with system tools
import subprocess
import json

# Get data from neighsnapshotter
with NeighborTableQuery() as ntq:
    our_data = ntq.get_neighbors(family='ipv4')

# Get data from ip command
result = subprocess.run(['ip', '-j', 'neigh', 'show'], 
                       capture_output=True, text=True)
system_data = json.loads(result.stdout)

print(f"NeighSnapshotter: {len(our_data)} entries")
print(f"ip command: {len(system_data)} entries")

Performance Tuning

Disable Unknown Attribute Tracking

# For production use, disable tracking to reduce overhead
with NeighborTableQuery(capture_unknown_attrs=False) as ntq:
    neighbors = ntq.get_neighbors()

Filter Early

# Query only what you need
# More efficient than filtering in Python
with NeighborTableQuery() as ntq:
    # Only get IPv4 ARP from kernel
    arp_only = ntq.get_neighbors(family='ipv4')

Common Patterns

Periodic Monitoring

import time

while True:
    with NeighborTableQuery() as ntq:
        neighbors = ntq.get_neighbors(family='ipv4')
    
    failed = [n for n in neighbors if 'FAILED' in n['state_name']]
    
    if failed:
        print(f"WARNING: {len(failed)} failed ARP entries")
        for entry in failed:
            print(f"  {entry.get('dst', 'unknown')}")
    
    time.sleep(60)  # Check every minute

Export to JSON File

import json
from datetime import datetime

with NeighborTableQuery() as ntq:
    neighbors = ntq.get_neighbors()

output = {
    'timestamp': datetime.now().isoformat(),
    'count': len(neighbors),
    'entries': neighbors
}

with open('neighbor_table.json', 'w') as f:
    json.dump(output, f, indent=2)

Contributing

Code Style

• C code: Linux kernel style (tabs, descriptive names)
• Python code: PEP 8 compliant
• Comments: Explain "why", not "what"

Testing

1. Test with real neighbor entries (generate with ping)
2. Test all protocol families (IPv4, IPv6, bridge)
3. Test edge cases (FAILED states, no entries, etc.)
4. Test on multiple kernel versions if possible

Adding New Attributes

To add support for new NDA_* attributes:

1. Add to known_nda_attrs array in C code
2. Add case in parsing switch statement
3. Add field to neigh_entry_t structure
4. Update CFFI definitions
5. Add Python decoding logic
6. Update documentation

References

• RTNetlink man page
• ip-neighbour(8) man page
• RFC 826: ARP
• RFC 4861: IPv6 Neighbor Discovery
• Linux Kernel: neighbour.h
• Linux Kernel: if_arp.h
• CFFI Documentation