RTNL classes

Arguments to the constructors:

port: Optional[int]

An integer to be used together with pid in the bind() call, epid = pid + (port << 22).

pid: Optional[int]

Value to be used as the base in while calling bind()

fileno: Optional[int]

An open file descriptor to construct the socket from.

sndbuf: int

Send buffer limit in bytes.

rcvbuf: int

Receive buffer limit in bytes.

rcvsize: int

Maximum recieve packet size.

all_ns: bool

Turns on NETLINK_LISTEN_ALL_NSID on the socket.

async_qsize

Deprecated.

nlm_generator

Deprecated.

target: str

Target field (string) to be provided in the header. Useful when working with sockets in multiple network namespaces.

ext_ack: bool

Extended ACK controls reporting additional error or warning info in NLMSG_ERROR and NLMSG_DONE messages.

strict_check: bool

Controls strict input field checking. By default kernel does not validate the input fields, silently ignoring possible issues, that may lead to regressions in the future.

groups: int (default: pyroute2.netlink.rtnl.RTMGRP_DEFAULTS)

Groups to subscribe when calling bind(), see pyroute2.netlink.rtnl

nlm_echo: bool

Return the request fields in the response.

use_socket: Optional[socket.socket]

An existing socket object to run the protocol on.

netns: str

Network namespace to use.

flags: int (default: os.O_CREAT)

Flags to use when calling netns.create(). By default the library will create netns if it doesn't exist, and reuse if it does. In order to fail when the network namespace already exists, you should provide flags=0.

libc: Optional[ctypes.CDLL]

If you want the socket to use specific libc object when managing network namespaces, you can use this argument.

use_event_loop: Optional[asyncio.AbstractEventLoop]

Use an existing asyncio event loop.

RTNL classes:

class pyroute2.AsyncIPRSocket(port=None, pid=None, fileno=None, sndbuf=1048576, rcvbuf=1048576, rcvsize=16384, all_ns=False, async_qsize=None, nlm_generator=None, target='localhost', ext_ack=False, strict_check=False, groups=67372509, nlm_echo=False, netns=None, netns_path=None, flags=64, libc=None, use_socket=None, use_event_loop=None, telemetry=None)

A low-level class to provide RTNL socket.

This is a low-level class designed to provide an RTNL asyncio-controlled socket. It does not include high-level methods like those found in AsyncIPRoute. Instead, it provides only common netlink methods such as get() and put(). For more details, refer to the AsyncNetlinkSocket documentation.

Since the underlying socket is controlled by asyncio, it is not possible to use it in poll/select loops. If you want such API, consider using synchronous IPRSocket.

Warning

Your code must process incoming messages quickly enough to prevent the RCVBUF from overflowing. If the RCVBUF overflows, all subsequent socket operations will raise an OSError:

>>> [ x async for x in ipr.get() ]
Traceback (most recent call last):
  File ".../python3.13/futures/_base.py", line 456, in result
    return self.__get_result()
           ~~~~~~~~~~~~~~~~~^^
  ...
OSError: [Errno 105] No buffer space available

If this exception occurs, the only solution is to close the socket and create a new one.

This class does not handle protocol-level error propagation; it only provides socket-level error handling. It is the user's responsibility to catch and manage protocol-level errors:

if msg.get(('header', 'type')) == NLMSG_ERROR:
    # prints error code and the request that
    # triggered the error
    print(
        msg.get('error'),
        msg.get('msg'),
    )

class pyroute2.IPRSocket(port=None, pid=None, fileno=None, sndbuf=1048576, rcvbuf=1048576, rcvsize=16384, all_ns=False, async_qsize=None, nlm_generator=None, target='localhost', ext_ack=False, strict_check=False, groups=67372509, nlm_echo=False, netns=None, netns_path=None, flags=64, libc=None, use_socket=None, use_event_loop=None)

Synchronous select-compatible netlink socket.

IPRSocket is the synchronous counterpart to AsyncIPRSocket. A key feature of IPRSocket is that the underlying netlink socket operates out of asyncio control, allowing it to be used in poll/select loops.

Warning

Your code must process incoming messages quickly enough to prevent the RCVBUF from overflowing. If the RCVBUF overflows, all subsequent socket operations will raise an OSError:

>>> iprsock.get()
Traceback (most recent call last):
  File "<python-input-12>", line 1, in <module>
    iprsock.get()
    ~~~~~~~~^^
  File ".../pyroute2/netlink/rtnl/iprsocket.py", line 276, in get
    data = self.socket.recv(16384)
OSError: [Errno 105] No buffer space available
>>>

If this exception occurs, the only solution is to close the socket and create a new one.

Some usage examples:

import select

from pyroute2 import IPRSocket
from pyroute2.netlink import NLM_F_DUMP, NLM_F_REQUEST
from pyroute2.netlink.rtnl import RTM_GETLINK
from pyroute2.netlink.rtnl.ifinfmsg import ifinfmsg

with IPRSocket() as iprsock:
    iprsock.put(
        ifinfmsg(),
        msg_type=RTM_GETLINK,
        msg_flags=NLM_F_REQUEST | NLM_F_DUMP
    )

    ret = []

    while True:
        rl, wl, xl = select.select([iprsock], [], [], 0)
        if not len(rl):
            break
        ret.extend(iprsock.get())

    for link in ret:
        if link.get('event') == 'RTM_NEWLINK':
            print(
                link.get('ifname'),
                link.get('state'),
                link.get('address'),
            )

lo up 00:00:00:00:00:00
eth0 up 52:54:00:72:58:b2

Threadless RT netlink monitoring with blocking I/O calls:

>>> from pyroute2 import IPRSocket
>>> from pprint import pprint
>>> s = IPRSocket()
>>> s.bind()
>>> pprint(s.get())
[{'attrs': [('RTA_TABLE', 254),
            ('RTA_OIF', 2),
            ('RTA_GATEWAY', '192.168.122.1')],
  'dst_len': 0,
  'event': 'RTM_NEWROUTE',
  'family': 2,
  'flags': 0,
  'header': {'error': None,
             'flags': 2,
             'length': 52,
             'pid': 325359,
             'sequence_number': 255,
             'type': 24},
  'proto': 2,
  'scope': 0,
  'src_len': 0,
  'table': 254,
  'tos': 0,
  'type': 2}]
>>>

Like AsyncIPRSocket, it does not perform response reassembly, protocol-level error propagation, or packet buffering.

class pyroute2.AsyncIPRoute(port=None, pid=None, fileno=None, sndbuf=1048576, rcvbuf=1048576, rcvsize=16384, all_ns=False, async_qsize=None, nlm_generator=None, target='localhost', ext_ack=False, strict_check=False, groups=67372509, nlm_echo=False, netns=None, netns_path=None, flags=64, libc=None, use_socket=None, use_event_loop=None, telemetry=None)

Regular ordinary async utility class, provides RTNL API using AsyncIPRSocket as the transport level.

Warning

The project core is currently undergoing refactoring, so some methods may still use the old synchronous API. This will be addressed in future updates.

The main RTNL API class is built on an asyncio core. All methods that send netlink requests are asynchronous and return awaitables. Dump requests return asynchronous generators, while other requests return iterables, such as tuples or lists.

This design choice addresses the fact that RTNL dumps, such as routes or neighbors, can return an extremely large number of objects. Buffering the entire response in memory could lead to performance issues.

import asyncio

from pyroute2 import AsyncIPRoute


async def main():
    async with AsyncIPRoute() as ipr:
        # create a link: immediate evaluation
        await ipr.link("add", ifname="test0", kind="dummy")

        # dump links: lazy evaluation
        async for link in await ipr.link("dump"):
            print(link.get("ifname"))

asyncio.run(main())

lo
eth0
test0

class pyroute2.IPRoute(port=None, pid=None, fileno=None, sndbuf=1048576, rcvbuf=1048576, rcvsize=16384, all_ns=False, async_qsize=None, nlm_generator=None, target='localhost', ext_ack=False, strict_check=False, groups=67372509, nlm_echo=False, netns=None, flags=64, libc=None, use_socket=None, use_event_loop=None, telemetry=None)

A synchronous version of AsyncIPRoute. All the same API, but sync. Provides a legacy API for the old code that is not using asyncio.

This API is designed to be compatible with the old synchronous IPRoute from version 0.8.x and earlier:

from pyroute2 import IPRoute

with IPRoute() as ipr:
    for msg in ipr.addr("dump"):
        addr = msg.get("address")
        mask = msg.get("prefixlen")
        print(f"{addr}/{mask}")

127.0.0.1/8
192.168.122.28/24

from pyroute2 import IPRoute

with IPRoute() as ipr:

    # this request returns one match, one interface index
    eth0 = ipr.link_lookup(ifname="eth0")
    assert len(eth0) == 1  # 1 if exists else 0

    # this requests uses a lambda to filter interfaces
    # and returns all interfaces that are up
    nics_up = set(ipr.link_lookup(lambda x: x.get("flags") & 1))
    assert len(nics_up) == 2
    assert nics_up == {1, 2}

class pyroute2.NetNS(netns=None, flags=64, target='localhost', libc=None, groups=67372509)

The NetNS class, prior to version 0.9.1, was used to run the RTNL API in a network namespace. Starting with pyroute2 version 0.9.1, the network namespace functionality has been integrated into the library core. To run an IPRoute or AsyncIPRoute instance in a network namespace, simply use the netns argument:

from pyroute2 import IPRoute

with IPRoute(netns="test") as ipr:
    assert ipr.status["netns"] == "test"

After initialization, the netns name is available as .status["netns"].

The old synchronous NetNS class is still available for compatibility but now serves as a wrapper around IPRoute.

from pyroute2 import NetNS

with NetNS("test") as ns:
    assert ns.status["netns"] == "test"