### Setup Development Environment Source: https://github.com/svinota/pyroute2/blob/master/README.contribute.rst Installs necessary tools, clones the repository, and runs the test suite to ensure the environment is correctly set up. ```bash # Make sure you have the following installed: # bash # git # python # GNU make, sed, awk # Clone the repository git clone ${pyroute2_git_url} cd pyroute2 # Run the test suite sudo make test ``` -------------------------------- ### Initialize NDB Object Source: https://github.com/svinota/pyroute2/blob/master/docs/ndb_init.rst The simplest way to start an NDB instance. No specific setup is required beyond importing the NDB class. ```python ndb = NDB() ``` -------------------------------- ### Install pyroute2 from Source Source: https://github.com/svinota/pyroute2/blob/master/README.rst Install pyroute2 from source after cloning the repository. Requires make. ```bash git clone https://github.com/svinota/pyroute2.git cd pyroute2 make install ``` -------------------------------- ### Install pyroute2 using pip Source: https://github.com/svinota/pyroute2/blob/master/README.rst Install the pyroute2 library from PyPI using pip. ```bash pip install pyroute2 ``` -------------------------------- ### Install pyroute2 from Git Source: https://github.com/svinota/pyroute2/blob/master/README.rst Install the pyroute2 library directly from its Git repository using pip. ```bash pip install git+https://github.com/svinota/pyroute2.git ``` -------------------------------- ### Example Test Session Configuration Source: https://github.com/svinota/pyroute2/blob/master/README.contribute.rst Demonstrates how to configure a test session with various options like exitfirst, verbosity, timeout, and pdb debugger. ```bash make test \ session=core-python3.14 \ noxconfig='{ "exitfirst": false, "verbose": false, "timeout": 10, "pdb": true, "sub": "test_plan9/test_basic.py" }' ``` -------------------------------- ### Start pyroute2-cli Server Mode Source: https://github.com/svinota/pyroute2/blob/master/docs/pyroute2-cli.rst Launch pyroute2-cli in server mode, listening on a specified IP address and port. Debug logging is enabled. ```bash pyroute2-cli -l debug -m S -a 127.0.0.1 -p 8080 ``` -------------------------------- ### Run DHCP Client with Specific Options Source: https://github.com/svinota/pyroute2/blob/master/docs/pyroute2-dhcp-client.rst Example of running the DHCP client with disabled hooks, no release on exit, and a 5-second timeout for acquiring a lease on the eth0 interface. ```bash # pyroute2-dhcp-client --disable-hooks -R -x 5 eth0 ``` -------------------------------- ### List Devlink-Enabled Hardware Devices and Ports Source: https://context7.com/svinota/pyroute2/llms.txt This example demonstrates how to use the DL (Devlink) class to list devlink-enabled hardware devices and their associated ports. The DL instance is managed using a context manager. ```python from pyroute2 import DL with DL() as dl: # List all devlink-enabled devices devices = dl.list() for dev in devices: bus = dev.get_attr("DEVLINK_ATTR_BUS_NAME") name = dev.get_attr("DEVLINK_ATTR_DEV_NAME") print(f"Device: {bus}/{name}") # List all devlink ports ports = dl.port_list() for port in ports: port_idx = port.get_attr("DEVLINK_ATTR_PORT_INDEX") port_type = port.get_attr("DEVLINK_ATTR_PORT_TYPE") print(f" Port {port_idx}: type={port_type}") ``` -------------------------------- ### Asynchronous Network Interface Management with AsyncIPRoute Source: https://github.com/svinota/pyroute2/blob/master/README.rst An asynchronous version of the IPRoute example, demonstrating how to manage network interfaces and namespaces using asyncio. ```python import asyncio from pyroute2 import AsyncIPRoute async def main(): # get access to the netlink socket ipr = AsyncIPRoute() # print interfaces async for link in await ipr.get_links(): print(link) # create VETH pair and move v0p1 to netns 'test' await ipr.link('add', ifname='v0p0', peer='v0p1', kind='veth') # wait for the devices: peer, veth = await ipr.poll( ipr.link, 'dump', timeout=5, ifname=lambda x: x in ('v0p0', 'v0p1') ) await ipr.link('set', index=peer['index'], net_ns_fd='test') ... ipr.close() asyncio.run(main()) ``` -------------------------------- ### Sync API: Dump Network Interfaces Source: https://github.com/svinota/pyroute2/blob/master/docs/usage.rst Demonstrates how to use the synchronous API to iterate over and print information about network interfaces. Ensure the pyroute2 library is installed. ```python from pyroute2 import IPRoute def main(): ipr = IPRoute() for link in ipr.link("dump"): print(link.get("ifname"), link.get("state"), link.get("address")) ipr.close() main() ``` -------------------------------- ### DHCP Server Detection Example Source: https://github.com/svinota/pyroute2/blob/master/docs/dhcp-server-detector.rst Detects DHCP servers on the 'wlp61s0' interface and exits after the first offer is received. The output is in JSON format. ```bash # dhcp-server-detector -1 wlp61s0 { "interface": "wlp61s0", "message": { "dhcp": { "op": 2, "htype": 1, "hlen": 6, "hops": 0, "xid": 2900208454, "secs": 0, "flags": 32768, "ciaddr": "0.0.0.0", "yiaddr": "192.168.94.166", "siaddr": "0.0.0.0", "giaddr": "0.0.0.0", "chaddr": "a0:a4:c5:93:ac:60", "sname": "", "file": "", "cookie": "63:82:53:63", "options": { "message_type": 2, "server_id": "192.168.94.254", "lease_time": 43200, "subnet_mask": "255.255.255.0", "router": [ "192.168.94.254" ], "name_server": [ "192.168.94.254" ], "broadcast_address": "192.168.94.255" } }, "eth_src": "14:0c:76:62:51:64", "eth_dst": "ff:ff:ff:ff:ff:ff", "ip_src": "192.168.94.254", "ip_dst": "255.255.255.255", "sport": 67, "dport": 68 } } ``` -------------------------------- ### Importing NetNS from pyroute2.netns.nslink Source: https://github.com/svinota/pyroute2/blob/master/docs/usage.rst Example of importing a class directly from its specific module. This import signature may change in future versions. ```python # Import a pyroute2 class directly. In the next versions # the import signature can be changed, e.g., NetNS from # pyroute2.netns.nslink it can be moved somewhere else. # from pyroute2.iproute.linux import NetNS ns = NetNS('test') ``` -------------------------------- ### Importing NetNS from root module Source: https://github.com/svinota/pyroute2/blob/master/docs/usage.rst Example of importing a class from the root pyroute2 module. This import signature is stable and will not change between versions. ```python # Import the same class from root module. This signature # will stay the same, any layout change is reflected in # the root module. # from pyroute2 import NetNS ns = NetNS('test') ``` -------------------------------- ### IPRoute.route() Source: https://context7.com/svinota/pyroute2/llms.txt Manages the kernel routing table. Supports 'add', 'del', 'replace', 'change', 'append', 'get', and 'dump' commands. ```APIDOC ## IPRoute.route() ### Description Manages the kernel routing table. Supported commands are `add`, `del`, `replace`, `change`, `append`, `get`, and `dump`. Supports multipath, encapsulation (MPLS, SEG6, SEG6local), and policy routing. ### Method Signature `route(cmd, **kw)` ### Parameters - `cmd` (string): The command to execute ('add', 'del', 'replace', 'change', 'append', 'get', 'dump'). - `dst` (string): Destination prefix (e.g., '10.0.0.0/24'). - `gateway` (string): Gateway address. - `table` (int): Routing table ID. - `type` (string): Route type (e.g., 'blackhole'). - `metrics` (dict): Route metrics like 'mtu', 'hoplimit'. - `multipath` (list): List of dictionaries for multipath routes. - `oif` (int): Output interface index. - `encap` (dict): Encapsulation details (e.g., `{'type': 'mpls', 'labels': '200/300'}`). ### Example ```python from pyroute2 import IPRoute with IPRoute() as ipr: # Simple unicast route ipr.route("add", dst="10.0.0.0/24", gateway="192.168.1.1") # Route with metrics (MTU, hop limit) ipr.route("add", dst="172.16.0.0/24", gateway="10.0.0.10", metrics={"mtu": 1400, "hoplimit": 16}) # Dump routes in main table (254) for route in ipr.route("dump", table=254): print(route.get("dst"), "via", route.get("gateway")) ``` ``` -------------------------------- ### IPSet - Netfilter IP Sets Management Source: https://context7.com/svinota/pyroute2/llms.txt Provides examples for managing Netfilter IP Sets using the `IPSet` class. This includes creating different types of sets (hash:ip, bitmap:port, hash:net,port), adding entries, testing for membership, listing sets, swapping sets atomically, and flushing/destroying sets. ```APIDOC ## IPSet — Netfilter IP Sets `IPSet` (and `AsyncIPSet`) manages netfilter ipsets for efficient packet matching. Supports `hash:ip`, `hash:net`, `hash:net,port`, `bitmap:port`, `list:set`, and other set types. ```python import socket from pyroute2 import IPSet from pyroute2.ipset import PortEntry, PortRange with IPSet() as ipset: # Create a hash:ip set ipset.create("blocked_ips", stype="hash:ip") ipset.add("blocked_ips", "198.51.100.1", etype="ip") ipset.add("blocked_ips", "198.51.100.2", etype="ip") print(ipset.test("blocked_ips", "198.51.100.1")) # True print(ipset.test("blocked_ips", "1.2.3.4")) # False # Create a bitmap:port set for port ranges ipset.create("allowed_ports", stype="bitmap:port", bitmap_ports_range=(1000, 9000)) ipset.add("allowed_ports", 8080, etype="port") ipset.add("allowed_ports", PortRange(8443, 8450), etype="port") # Create a hash:net,port set for ACL-style rules ipset.create("acl", stype="hash:net,port") tcp = socket.getprotobyname("tcp") ipset.add("acl", ("10.0.0.0/8", PortEntry(443, protocol=tcp)), etype="net,port") # List all sets for s in ipset.list(): print(s.get_attr("IPSET_ATTR_SETNAME"), s.get_attr("IPSET_ATTR_TYPENAME")) # Swap two sets atomically (for atomic rule updates) ipset.create("blocked_ips_new", stype="hash:ip") ipset.add("blocked_ips_new", "203.0.113.5", etype="ip") ipset.swap("blocked_ips", "blocked_ips_new") # Flush and destroy ipset.flush("blocked_ips") ipset.destroy("blocked_ips") ipset.destroy("blocked_ips_new") ipset.destroy("allowed_ports") ipset.destroy("acl") ``` ``` -------------------------------- ### Conntrack - Connection Tracking Source: https://context7.com/svinota/pyroute2/llms.txt Demonstrates how to use the `Conntrack` class to interact with the Netfilter connection tracking subsystem. Examples include retrieving connection statistics, counting active connections, dumping all connection entries, and filtering entries by protocol or destination address. ```APIDOC ## Conntrack — Connection Tracking `Conntrack` provides access to netfilter's connection tracking subsystem for monitoring and manipulating stateful connections. ```python import socket from pyroute2 import Conntrack from pyroute2.netlink.nfnetlink.nfctsocket import NFCTAttrTuple ct = Conntrack() # Get global conntrack statistics (per-CPU) stats = ct.stat() for cpu_stat in stats: print(cpu_stat) # Get current active connection count print(f"Active connections: {ct.count()}") print(f"Max table size: {ct.conntrack_max_size()}") # Dump all connections for entry in ct.dump_entries(): print(entry) # Filter by protocol — only TCP connections for entry in ct.dump_entries( tuple_orig=NFCTAttrTuple(proto=socket.IPPROTO_TCP) ): orig = entry.get_attr("CTA_TUPLE_ORIG") if orig: ip = orig.get_attr("CTA_TUPLE_IP") print(f"TCP: {ip.get_attr('CTA_IP_V4_SRC')} -> " f"{ip.get_attr('CTA_IP_V4_DST')}") # Filter by destination address for entry in ct.dump_entries( tuple_orig=NFCTAttrTuple(proto=socket.IPPROTO_ICMP, daddr="8.8.8.8") ): print(entry) ``` ``` -------------------------------- ### DHCP Lease JSON Output Source: https://github.com/svinota/pyroute2/blob/master/docs/pyroute2-dhcp-client.rst Example of the JSON output for a DHCP lease obtained by the client, including IP configuration, lease times, and network parameters. ```json { "ack": { "op": 2, "htype": 1, "hlen": 6, "hops": 0, "xid": 391458644, "secs": 0, "flags": 32768, "ciaddr": "0.0.0.0", "yiaddr": "192.168.124.180", "siaddr": "192.168.124.1", "giaddr": "0.0.0.0", "chaddr": "aa:80:fa:2c:49:a2", "sname": "", "file": "", "cookie": "63:82:53:63", "options": { "message_type": 5, "server_id": "192.168.124.1", "lease_time": 3600, "renewal_time": 1800, "rebinding_time": 3150, "subnet_mask": "255.255.255.0", "broadcast_address": "192.168.124.255", "router": [ "192.168.124.1" ], "name_server": [ "192.168.124.1" ] } }, "interface": "eth0", "server_mac": "52:54:00:e9:d3:1d", "obtained": 1740562827.7350075 } ``` -------------------------------- ### Interface Management with IPRoute.link() Source: https://context7.com/svinota/pyroute2/llms.txt Manage network interfaces using the `link()` method with commands like `add`, `set`, `del`, `dump`, and `get`. Supports creating various virtual interface types and configuring their properties. ```python from pyroute2 import IPRoute with IPRoute() as ipr: # Create a bridge ipr.link("add", ifname="br0", kind="bridge") br_idx = ipr.link_lookup(ifname="br0")[0] # Create a VLAN interface on top of eth0 eth0_idx = ipr.link_lookup(ifname="eth0")[0] ipr.link("add", ifname="eth0.100", kind="vlan", link=eth0_idx, vlan_id=100) # Create a VXLAN tunnel interface ipr.link("add", ifname="vxlan0", kind="vxlan", vxlan_id=42, vxlan_group="239.1.1.1", vxlan_port=4789) # Rename interface ipr.link("set", index=br_idx, ifname="mybr0") # Set MAC address and MTU ipr.link("set", index=br_idx, address="00:11:22:33:44:55", mtu=9000) # Add port to bridge eth1_idx = ipr.link_lookup(ifname="eth1")[0] ipr.link("set", index=eth1_idx, master=br_idx) # Bring bridge up ipr.link("set", index=br_idx, state="up") # Delete interface ipr.link("del", index=br_idx) ``` -------------------------------- ### Initialize AsyncIPRoute in a Network Namespace Source: https://github.com/svinota/pyroute2/blob/master/docs/iproute_netns.rst Use the `netns` argument with `AsyncIPRoute` to initialize a socket within a specified network namespace. This example demonstrates how to check the current netns. ```python import asyncio from pyroute2 import AsyncIPRoute async def main(): async with AsyncIPRoute(netns="test") as ipr: print(f"current netns: {ipr.status['netns']}") asyncio.run(main()) ``` -------------------------------- ### Directly Call Async get() in Running Loop Source: https://context7.com/svinota/pyroute2/llms.txt Shows how to directly call the async get() method of AsyncCoreSocket within a running asyncio event loop. Requires an active loop. ```python async def raw_dump(): async with AsyncCoreSocket() as sock: print(type(sock)) # AsyncCoreSocket ``` -------------------------------- ### NDB Initialization and Basic Usage Source: https://context7.com/svinota/pyroute2/llms.txt Demonstrates basic initialization of the NDB with a local RTNL source and how to list interfaces. It also shows multi-source initialization and accessing interfaces from specific sources, along with persistent SQLite database usage. ```APIDOC ## NDB — High-Level RTNL Database API `NDB` provides a high-level interface backed by an in-memory SQLite database. It tracks all network state, supports transactional changes with commit/rollback, and can manage multiple RTNL sources including remote hosts and network namespaces simultaneously. ```python from pyroute2 import NDB # Basic initialization — starts with local RTNL source with NDB() as ndb: # List all interfaces for key in ndb.interfaces: nic = ndb.interfaces[key] print(nic["ifname"], nic["state"], nic["address"]) # Multi-source NDB — manage multiple hosts/namespaces at once with NDB( sources=[ {"target": "localhost", "kind": "local"}, {"target": "worker1", "kind": "local"}, # simulated here {"netns": "testns"}, ], log="on", ) as ndb: # Get interface from a specific source wrk1_eth0 = ndb.interfaces[{"target": "worker1", "ifname": "eth0"}] print(wrk1_eth0["mtu"]) # Persistent SQLite database (useful for debugging) with NDB(db_provider="sqlite3", db_spec="/tmp/netdb.db") as ndb: pass # Data persists in /tmp/netdb.db after exit ``` ``` -------------------------------- ### IPRoute.neigh() Source: https://context7.com/svinota/pyroute2/llms.txt Manages ARP (IPv4) and NDP (IPv6) neighbor cache entries. Supported commands are 'add', 'set', 'del', 'dump', and 'get'. ```APIDOC ## IPRoute.neigh() ### Description Manages ARP (IPv4) and NDP (IPv6) neighbor cache entries. Supported commands are `add`, `set`, `del`, `dump`, and `get`. ### Method Signature `neigh(cmd, **kw)` ### Parameters - `cmd` (string): The command to execute ('add', 'set', 'del', 'dump', 'get'). - `dst` (string): Destination IP address. - `lladdr` (string): Link-layer address (MAC address). - `ifindex` (int): Interface index. - `state` (int): Neighbor state (e.g., `ndmsg.states["permanent"]`). ### Example ```python from socket import AF_INET, AF_BRIDGE from pyroute2 import IPRoute from pyroute2.netlink.rtnl import ndmsg with IPRoute() as ipr: idx = ipr.link_lookup(ifname="eth0")[0] # Add a permanent ARP entry ipr.neigh("add", dst="172.16.45.1", lladdr="00:11:22:33:44:55", ifindex=idx, state=ndmsg.states["permanent"]) # Dump all neighbors on a specific interface for entry in ipr.get_neighbours(ifindex=idx): print(entry.get("dst"), entry.get("lladdr")) # Delete a neighbor entry ipr.neigh("del", dst="172.16.45.1", ifindex=idx) ``` ``` -------------------------------- ### Manage Traffic Control (QoS) with IPRoute.tc() Source: https://context7.com/svinota/pyroute2/llms.txt Demonstrates adding, configuring, and dumping traffic control elements like qdiscs, classes, and filters on a network interface. Ensure the interface (e.g., 'eth0') exists and has appropriate permissions to modify traffic control settings. ```python from pyroute2 import IPRoute with IPRoute() as ipr: idx = ipr.link_lookup(ifname="eth0")[0] # Add a root HTB qdisc ipr.tc("add", "htb", index=idx, handle="1:", default=10) # Add an HTB class with rate limiting (1Mbit guaranteed, 2Mbit max) ipr.tc("add-class", "htb", index=idx, handle="1:10", parent="1:", rate="1mbit", ceil="2mbit") # Add a u32 filter to match and classify traffic ipr.tc("add-filter", "u32", index=idx, parent="1:0", prio=10, protocol=0x0800, target="1:10", keys=["0x0/0x0+12"]) # Add SFQ (Stochastic Fair Queuing) as leaf ipr.tc("add", "sfq", index=idx, handle="10:", parent="1:10", perturb=10) # Dump all qdiscs for qdisc in ipr.get_qdiscs(): print(qdisc.get("kind"), "on ifindex", qdisc.get("index")) # Remove root qdisc (removes all children too) ipr.tc("del", index=idx, handle="1:", parent="FFFF:") ``` -------------------------------- ### Create and Configure WireGuard VPN Interface Source: https://context7.com/svinota/pyroute2/llms.txt This snippet demonstrates creating a WireGuard interface using NDB, setting its IP address and state, and then configuring its private key, listen port, and peers using the WireGuard class. It also shows how to query configuration and remove a peer. ```python from pyroute2 import NDB, WireGuard IFNAME = "wg0" # Step 1: Create the WireGuard network interface with NDB() as ndb: (ndb.interfaces .create(kind="wireguard", ifname=IFNAME) .add_ip("10.200.0.1/24") .set("state", "up") .commit()) # Step 2: Configure keys, listen port and peers wg = WireGuard() # Set device private key and listen port peer = { "public_key": "TGFHcm9zc2VCaWNoZV9DJ2VzdExhUGx1c0JlbGxlPDM=", "endpoint_addr": "203.0.113.10", "endpoint_port": 51820, "persistent_keepalive": 25, "allowed_ips": ["10.200.0.2/32", "192.168.50.0/24"], } wg.set(IFNAME, private_key="RCdhcHJlc0JpY2hlLEplU2VyYWlzTGFQbHVzQm9ubmU=", listen_port=51820, fwmark=0x1337, peer=peer) # Add a second peer with a preshared key peer2 = { "public_key": "RCdBcHJlc0JpY2hlLFZpdmVMZXNQcm9iaW90aXF1ZXM=", "preshared_key": "Pz8/V2FudFRvVHJ5TXlBZXJvR3Jvc3NlQmljaGU/Pz8=", "endpoint_addr": "198.51.100.5", "endpoint_port": 51820, "persistent_keepalive": 25, "allowed_ips": ["::/0"], } wg.set(IFNAME, peer=peer2) # Query current configuration info = wg.info(IFNAME) print(info[0].get("WGDEVICE_A_LISTEN_PORT")) # Remove a peer wg.set(IFNAME, peer={"public_key": peer2["public_key"], "remove": True}) wg.close() ``` -------------------------------- ### Initialize and List Interfaces with NDB Source: https://context7.com/svinota/pyroute2/llms.txt Demonstrates basic NDB initialization with the local RTNL source and how to list and access interface details. For multi-source NDB, specify sources in the constructor. Persistent SQLite databases can be enabled via `db_provider` and `db_spec`. ```python from pyroute2 import NDB # Basic initialization — starts with local RTNL source with NDB() as ndb: # List all interfaces for key in ndb.interfaces: nic = ndb.interfaces[key] print(nic["ifname"], nic["state"], nic["address"]) ``` ```python # Multi-source NDB — manage multiple hosts/namespaces at once with NDB( sources=[ {"target": "localhost", "kind": "local"}, {"target": "worker1", "kind": "local"}, # simulated here {"netns": "testns"}, ], log="on", ) as ndb: # Get interface from a specific source wrk1_eth0 = ndb.interfaces[{"target": "worker1", "ifname": "eth0"}] print(wrk1_eth0["mtu"]) ``` ```python # Persistent SQLite database (useful for debugging) with NDB(db_provider="sqlite3", db_spec="/tmp/netdb.db") as ndb: pass # Data persists in /tmp/netdb.db after exit ``` -------------------------------- ### Manage Network Interfaces and Addresses with IPRoute Source: https://github.com/svinota/pyroute2/blob/master/README.rst Demonstrates creating a VETH pair, moving one interface to a network namespace, bringing it up, and adding an IP address using IPRoute. ```python from socket import AF_INET from pyroute2 import IPRoute # get access to the netlink socket ipr = IPRoute() # no monitoring here -- thus no bind() # print interfaces for link in ipr.get_links(): print(link) # create VETH pair and move v0p1 to netns 'test' ipr.link('add', ifname='v0p0', peer='v0p1', kind='veth') # wait for the devices: peer, veth = ipr.poll( ipr.link, 'dump', timeout=5, ifname=lambda x: x in ('v0p0', 'v0p1') ) ipr.link('set', index=peer['index'], net_ns_fd='test') # bring v0p0 up and add an address ipr.link('set', index=veth['index'], state='up') ipr.addr('add', index=veth['index'], address='10.0.0.1', prefixlen=24) # release Netlink socket ipr.close() ``` -------------------------------- ### Manage Kernel Routing Table Source: https://context7.com/svinota/pyroute2/llms.txt Shows how to add various types of routes, including unicast, routes in specific tables, blackhole routes, routes with metrics, multipath routes, and MPLS encapsulation. Also demonstrates dumping and flushing routes. ```python from pyroute2 import IPRoute with IPRoute() as ipr: # Simple unicast route ipr.route("add", dst="10.0.0.0/24", gateway="192.168.1.1") # Route in a specific table ipr.route("add", dst="10.0.0.0/24", gateway="192.168.1.1", table=100) # Blackhole route ipr.route("add", dst="192.168.100.0/24", type="blackhole") # Route with metrics (MTU, hop limit) ipr.route("add", dst="172.16.0.0/24", gateway="10.0.0.10", metrics={"mtu": 1400, "hoplimit": 16}) # Multipath (ECMP) route ipr.route("add", dst="10.1.0.0/24", multipath=[ {"gateway": "192.168.1.1", "hops": 2}, {"gateway": "192.168.1.2", "hops": 1}, ]) # MPLS encapsulation oif = ipr.link_lookup(ifname="eth0")[0] ipr.route("add", dst="10.0.0.0/24", oif=oif, encap={"type": "mpls", "labels": "200/300"}) # Dump routes in main table (254) for route in ipr.route("dump", table=254): print(route.get("dst"), "via", route.get("gateway")) # Flush all routes matching a prefix ipr.flush_routes(dst="10.0.0.0", dst_len=24) ``` -------------------------------- ### Get Default Routes with Custom Parser (Python) Source: https://github.com/svinota/pyroute2/blob/master/docs/parser.rst Assigns a specific parser to optimize the retrieval of default route records, parsing only matched records with the standard routine. ```python from pyroute2.iproute.linux import IPRoute class RTNL_API(IPRoute): def get_default_routes(self): return self.get(IPRoute.GETLINK) IPRoute.get_default_routes = RTNL_API.get_default_routes ``` -------------------------------- ### Using IPRoute with a Custom asyncio Event Loop Source: https://github.com/svinota/pyroute2/blob/master/docs/asyncio.rst Demonstrates how to instantiate IPRoute with a custom asyncio event loop, which can be useful for integrating with specific loop implementations like uvloop. ```python selector = selectors.SelectSelector() loop = asyncio.SelectorEventLoop(selector) with IPRoute(use_event_loop=loop) as ipr: pass ``` -------------------------------- ### Initialize IPRoute with Flags (Fail if Namespace Not Found) Source: https://github.com/svinota/pyroute2/blob/master/docs/iproute_netns.rst When initializing `IPRoute`, setting `flags=0` prevents the creation of a new network namespace if it does not already exist, causing a `FileNotFoundError` in such cases. ```python from pyroute2 import IPRoute try: ipr = IPRoute(netns="foo", flags=0) except FileNotFoundError: print("netns doesn't exist, refuse to start") ``` -------------------------------- ### Manage Policy Routing Rules Source: https://context7.com/svinota/pyroute2/llms.txt Illustrates adding policy routing rules based on source/destination IP, firewall marks, and incoming interface names. Also shows how to dump and delete rules. ```python from pyroute2 import IPRoute with IPRoute() as ipr: # Route all traffic from a source subnet via table 100 ipr.rule("add", table=100, priority=32000, src="10.64.75.0", src_len=24) # Route traffic to a destination via table 200 ipr.rule("add", table=200, priority=32001, dst="10.64.75.141", dst_len=32) # Match fwmark (useful with iptables MARK target) ipr.rule("add", table=150, priority=32002, fwmark=0x10, fwmask=0xFF) # Match incoming interface ipr.rule("add", table=50, priority=32003, iifname="eth1") # List all rules for rule in ipr.rule("dump"): print(f"priority={rule.get('priority')} " f"table={rule.get('table')} " f"src={rule.get('src')}") # Remove a rule ipr.rule("del", table=100, priority=32000, src="10.64.75.0", src_len=24) ``` -------------------------------- ### Get NetlinkRequest Response Source: https://github.com/svinota/pyroute2/blob/master/docs/parser.rst Use this method to retrieve the response for a NetlinkRequest. It handles buffering of messages to ensure the correct response is returned, even with complex sequence number interactions. ```python response = request.response() ``` -------------------------------- ### Enable RTNL Debugging in NDB Source: https://github.com/svinota/pyroute2/blob/master/docs/ndb_debug.rst Start NDB with the `rtnl_debug` option enabled to store all loaded RTNL events in the DB. This feature should not be used in production as it can consume significant memory. ```python ndb = NDB(rtnl_debug=True) ``` -------------------------------- ### Run pyroute2-cli Command Source: https://github.com/svinota/pyroute2/blob/master/docs/pyroute2-cli.rst Execute a network configuration command directly using the -c option. Debug logging can be enabled with -l. ```bash pyroute2-cli -l debug -c "interfaces eth0 set { state up } => add_ip { 10.0.0.2/24 } => commit" ``` -------------------------------- ### IPRoute.link() - Interface Management Source: https://context7.com/svinota/pyroute2/llms.txt The link() method manages network interfaces, supporting operations like add, set, del, dump, and get. It allows creation of various virtual interface types. ```APIDOC ## IPRoute.link() ### Description The `link()` method handles all network interface operations. Supported commands are `add`, `set`, `del`, `dump`, and `get`. It supports creating virtual interfaces of many kinds: `dummy`, `bridge`, `bond`, `vlan`, `veth`, `vxlan`, `wireguard`, `macvlan`, etc. ### Usage Example ```python from pyroute2 import IPRoute with IPRoute() as ipr: # Create a bridge ipr.link("add", ifname="br0", kind="bridge") br_idx = ipr.link_lookup(ifname="br0")[0] # Create a VLAN interface on top of eth0 eth0_idx = ipr.link_lookup(ifname="eth0")[0] ipr.link("add", ifname="eth0.100", kind="vlan", link=eth0_idx, vlan_id=100) # Create a VXLAN tunnel interface ipr.link("add", ifname="vxlan0", kind="vxlan", vxlan_id=42, vxlan_group="239.1.1.1", vxlan_port=4789) # Rename interface ipr.link("set", index=br_idx, ifname="mybr0") # Set MAC address and MTU ipr.link("set", index=br_idx, address="00:11:22:33:44:55", mtu=9000) # Add port to bridge eth1_idx = ipr.link_lookup(ifname="eth1")[0] ipr.link("set", index=eth1_idx, master=br_idx) # Bring bridge up ipr.link("set", index=br_idx, state="up") # Delete interface ipr.link("del", index=br_idx) ``` ``` -------------------------------- ### Run pyroute2-cli Script File Source: https://github.com/svinota/pyroute2/blob/master/docs/pyroute2-cli.rst Execute a sequence of network configuration commands stored in a script file. The file content is interpreted by pyroute2-cli. ```bash interfaces eth0 set { state up } add_ip { 10.0.0.2/24 } commit pyroute2-cli -l debug script.pr2 ``` -------------------------------- ### Netlink Socket: Subscribe and Receive Source: https://github.com/svinota/pyroute2/blob/master/docs/usage.rst Illustrates the basic usage of a Netlink socket with pyroute2, including binding to broadcast messages and receiving data. Note that direct recv() is not supported; use get() instead. ```python from pyroute2 import IPRoute # create RTNL socket ipr = IPRoute() # subscribe to broadcast messages ipr.bind() # wait for parsed data data = ipr.get() ``` -------------------------------- ### Run Nox Directly with Configuration Source: https://github.com/svinota/pyroute2/blob/master/README.contribute.rst Executes a nox session directly using the 'nox' command, passing the environment and configuration separately. ```bash nox \ -e core-python3.14 -- \ '{ "exitfirst": false, "verbose": false, "timeout": 10, "pdb": true, "sub": "test_plan9/test_basic.py" }' ``` -------------------------------- ### AsyncCoreSocket Component Verification Source: https://github.com/svinota/pyroute2/blob/master/docs/asyncio.rst Verifies the components and functionalities of the AsyncCoreSocket, including its socket, transport, protocol, message queue, enqueue, get methods, and marshal object. Ensures compatibility and correct asynchronous behavior. ```python import asyncio import inspect from pyroute2 import IPRoute from pyroute2.netlink.core import CoreMessageQueue from pyroute2.netlink.marshal import Marshal with IPRoute() as ipr: # AsyncCoreSocket.socket, compatibility, management assert callable(ipr.asyncore.socket.recv) assert callable(ipr.asyncore.socket.send) assert callable(ipr.asyncore.socket.recvmsg) assert callable(ipr.asyncore.socket.sendmsg) assert callable(ipr.asyncore.socket.bind) assert ipr.asyncore.transport._sock == ipr.asyncore.socket # AsyncCoreSocket.endpoint assert isinstance(ipr.asyncore.transport, asyncio.Transport) assert isinstance(ipr.asyncore.protocol, asyncio.Protocol) # msg_queue assert isinstance(ipr.asyncore.msg_queue, CoreMessageQueue) # enqueue() e_flags = ipr.asyncore.enqueue.__code__.co_flags assert callable(ipr.asyncore.enqueue) assert not e_flags & inspect.CO_ASYNC_GENERATOR # get() g_flags = ipr.asyncore.get.__code__.co_flags assert callable(ipr.asyncore.get) assert g_flags & inspect.CO_ASYNC_GENERATOR # marshal assert isinstance(ipr.asyncore.marshal, Marshal) ``` -------------------------------- ### NDB - Multi-Source and Namespace Management Source: https://context7.com/svinota/pyroute2/llms.txt Illustrates how to use NDB to manage network state across multiple sources, including different network namespaces and remote hosts. It covers connecting to multiple sources at initialization, adding sources dynamically, and creating network interfaces within specific namespaces. ```APIDOC ## NDB — Multi-Source and Namespace Management NDB can aggregate network state from multiple sources — local namespaces, multiple netns, or remote hosts — into a single queryable database. ```python from pyroute2 import NDB # Connect multiple netns sources at init time with NDB(sources=[ {"target": "localhost"}, {"netns": "ns01"}, {"netns": "ns02"}, ]) as ndb: # View interfaces from all sources for key, nic in ndb.interfaces.items(): print(f"[{nic['target']}] {nic['ifname']} {nic['state']}") # Create an interface in a specific namespace (ndb.interfaces .create(target="ns01", ifname="veth0", kind="veth", peer={"ifname": "veth1", "net_ns_fd": "ns02"}) .commit()) # Add sources dynamically after init with NDB() as ndb: ndb.sources.add(netns="newns") # List all active sources for src in ndb.sources.summary().format("json"): print(src) # Access interface in dynamically added source ns_lo = ndb.interfaces[{"target": "newns", "ifname": "lo"}] print(ns_lo["state"]) ``` ``` -------------------------------- ### Marshal: Get and Run Parsers Source: https://github.com/svinota/pyroute2/blob/master/docs/parser.rst The Marshal class determines the appropriate parser for a Netlink message based on the message type, flags, and sequence number. It supports custom key formats and allows overriding the get_parser method. ```python from pyroute2.netlink.nlsocket import Marshal # Conceptual example of custom key format: # marshal = Marshal() # marshal.key_format = ' <-- here you can be sure that the interface is up & renamed ``` -------------------------------- ### Run pyroute2-cli via Stdin Source: https://github.com/svinota/pyroute2/blob/master/docs/pyroute2-cli.rst Provide multi-line commands to pyroute2-cli through standard input using a here-document. This is useful for complex configurations. ```bash cat < interfaces eth0 > set { state up } > add_ip { 10.0.0.2/24 } > commit > EOF ``` -------------------------------- ### IPRoute: Dump IP Addresses Source: https://github.com/svinota/pyroute2/blob/master/docs/usage.rst Demonstrates how to use the IPRoute object within a context manager to retrieve and print IP addresses and their subnet masks. This ensures resources are properly released. ```python from pyroute2 import IPRoute # RTNL interface with IPRoute() as ipr: # get IP addresses for msg in ipr.addr("dump"): addr = msg.get("address") mask = msg.get("prefixlen") print(f"{addr}/{mask}") ``` -------------------------------- ### Script to Detect DHCP Servers on All Interfaces Source: https://github.com/svinota/pyroute2/blob/master/docs/dhcp-server-detector.rst This script iterates through all network interfaces, checks for DHCP servers on each with a 1-second detection duration and immediate exit on first offer, and prints a message if a server is found. It uses 'ip --json l' and 'jq' to get interface names. ```bash # prints a line for every interface on which a DHCP server is detected. # waits 1 second for each interface. for ifname in $(ip --json l | jq -r '.[].ifname'); do if dhcp-server-detector -d 1 -1 $ifname > /dev/null; then echo "DHCP server found on $ifname" fi done ``` -------------------------------- ### Integrate pyroute2 with a Custom Asyncio Event Loop Source: https://context7.com/svinota/pyroute2/llms.txt This snippet shows how to use pyroute2's AsyncCoreSocket foundation by integrating it with a custom asyncio event loop that uses a specific selector. It then demonstrates dumping network links using IPRoute within this custom loop. ```python import asyncio import selectors from pyroute2 import IPRoute from pyroute2.netlink.core import AsyncCoreSocket, CoreSocket # Use a custom event loop (e.g., with specific selector) selector = selectors.SelectSelector() loop = asyncio.SelectorEventLoop(selector) with IPRoute(use_event_loop=loop) as ipr: for link in ipr.link("dump"): print(link.get("ifname")) ``` -------------------------------- ### Manage Routes with NDB Source: https://context7.com/svinota/pyroute2/llms.txt Illustrates NDB's route management capabilities, including creating, modifying, and deleting routes. Supports specifying non-default routing tables and retrieving routes using dictionary selectors. The `summary()` method can list all routing tables in use. ```python from pyroute2 import NDB with NDB() as ndb: # Create a simple route ndb.routes.create( dst="10.0.0.0/24", gateway="192.168.1.1" ).commit() ``` ```python # Modify a route with ndb.routes["10.0.0.0/24"] as route: route.set("gateway", "192.168.1.10") ``` ```python # Route in a non-default table ndb.routes.create( dst="10.0.0.0/24", gateway="192.168.1.1", table=101 ).commit() ``` ```python # Retrieve route from a specific table with ndb.routes[{"table": 101, "dst": "10.0.0.0/24"}] as route: print(route["gateway"]) route.set("gateway", "192.168.1.99") ``` ```python # List all routing tables in use print(set(x.table for x in ndb.routes.summary())) # {101, 254, 255} ``` ```python # Remove a route with ndb.routes["10.0.0.0/24"] as route: route.remove() ``` -------------------------------- ### Run Specific Test Modules Source: https://github.com/svinota/pyroute2/blob/master/README.contribute.rst Executes specific test modules or sessions using the 'make test' command with session arguments. ```bash make test session=unit # run only tests/test_unit make test session=core-python3.14 # run tests/test_core with Python 3.14 ``` -------------------------------- ### Initialize NDB with Sources Source: https://github.com/svinota/pyroute2/blob/master/docs/ndb_init.rst Initializes NDB with specified RTNL sources. This is useful for connecting to different network namespaces or targets. ```python sources = [{'netns': 'test01'}, {'netns': 'test02'}, {'target': 'localhost', 'kind': 'local'}] ndb = NDB(log='on', sources=sources) ```