Import python-nlmanager mirror copy

Ticket: CM-7360
Reviewed By: CCR-4721
Testing Done: smoke / testifreload / Tested on amd64 platform (by Sam)

For now, we use a mirror copy of nlmanager sources to make sure we don't
depend on it as an external package

Signed-off-by: Julien Fortin <julien@cumulusnetworks.com>

nlmanager/README

@@ -0,0 +1,6 @@
+DO NOT EDIT NLMANAGER SOURCES.
+This is a mirror copy of python-nlmanager sources.
+It was extracted and directly included here to support some usecases where
+user don't have python-nlmanager already installed on their system. So we
+decided to have local copy and build with it. It is the mainter responsability
+to keep an updated version of nlmanager.

nlmanager/__init__.py

@@ -0,0 +1 @@
+

nlmanager/nllistener.py

@@ -0,0 +1,538 @@
+#!/usr/bin/env python
+
+from nlpacket import *
+from nlmanager import NetlinkManager
+from select import select
+from struct import pack, unpack, calcsize
+from threading import Thread, Event, Lock
+from Queue import Queue
+import logging
+import socket
+
+log = logging.getLogger(__name__)
+
+
+class NetlinkListener(Thread):
+
+    def __init__(self, manager, groups):
+        """
+        groups controls what types of messages we are interested in hearing
+        To get everything pass:
+            RTMGRP_LINK | \
+            RTMGRP_IPV4_IFADDR | \
+            RTMGRP_IPV4_ROUTE | \
+            RTMGRP_IPV6_IFADDR | \
+            RTMGRP_IPV6_ROUTE
+        """
+        Thread.__init__(self)
+        self.manager = manager
+        self.shutdown_event = Event()
+        self.groups = groups
+
+    def __str__(self):
+        return 'NetlinkListener'
+
+    def run(self):
+        manager = self.manager
+        header_PACK = 'IHHII'
+        header_LEN = calcsize(header_PACK)
+
+        # The RX socket is used to listen to all netlink messages that fly by
+        # as things change in the kernel. We need a very large SO_RCVBUF here
+        # else we tend to miss messages.
+        self.rx_socket = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, 0)
+        self.rx_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 10000000)
+        self.rx_socket.bind((manager.pid+1, self.groups))
+        self.rx_socket_prev_seq = {}
+
+        if not manager.tx_socket:
+            manager.tx_socket_allocate()
+
+        my_sockets = (manager.tx_socket, self.rx_socket)
+
+        socket_string = {
+            manager.tx_socket: "TX",
+            self.rx_socket: "RX"
+        }
+
+        supported_messages = (RTM_NEWLINK, RTM_DELLINK, RTM_NEWADDR,
+                              RTM_DELADDR, RTM_NEWNEIGH, RTM_DELNEIGH,
+                              RTM_NEWROUTE, RTM_DELROUTE)
+
+        ignore_messages = (RTM_GETLINK, RTM_GETADDR, RTM_GETNEIGH,
+                           RTM_GETROUTE, RTM_GETQDISC, NLMSG_ERROR, NLMSG_DONE)
+
+        while True:
+
+            if self.shutdown_event.is_set():
+                log.info("%s: shutting down" % self)
+                return
+
+            # Only block for 1 second so we can wake up to see
+            try:
+                (readable, writeable, exceptional) = select(my_sockets, [], my_sockets, 1)
+            except Exception as e:
+                log.error('select() error: ' + str(e))
+                continue
+
+            if not readable:
+                continue
+
+            set_alarm = False
+            set_tx_socket_rxed_ack_alarm = False
+            missed_a_packet = False
+
+            for s in readable:
+                data = None
+
+                try:
+                    data = s.recv(4096)
+
+                # If recv() failed, we missed a packet
+                except Exception as e:
+                    log.error('recv() error: ' + str(e))
+                    missed_a_packet = True
+                    continue
+
+                total_length = len(data)
+                while data:
+
+                    # Extract the length, etc from the header
+                    (length, msgtype, flags, seq, pid) = unpack(header_PACK, data[:header_LEN])
+
+                    if manager.debug_listener:
+                        log.info('%s: RXed %s seq %d, pid %d, %d bytes (%d total)' %
+                                 (socket_string[s], NetlinkPacket.type_to_string[msgtype],
+                                  seq, pid, length, total_length))
+
+                    # 99% of the time when we see an ERROR the error code is
+                    # zero which means ACK
+                    possible_ack = False
+
+                    if msgtype == NLMSG_DONE:
+                        possible_ack = True
+
+                    elif msgtype == NLMSG_ERROR:
+                        # TODO - change this > to = ?
+                        error_code = int(unpack('>H', data[header_LEN:header_LEN+2])[0])
+
+                        if error_code:
+                            log.debug("%s: RXed NLMSG_ERROR code %d" % (socket_string[s], error_code))
+                        else:
+                            possible_ack = True
+
+                    if possible_ack and seq == manager.target_seq and pid == manager.target_pid:
+                        if manager.target_seq_pid_debug:
+                            log.debug("%s: Setting RXed ACK alarm for seq %d, pid %d" %
+                                      (socket_string[s], seq, pid))
+                        set_tx_socket_rxed_ack_alarm = True
+
+                    # Put the message on the manager's netlinkq
+                    if msgtype in supported_messages:
+                        set_alarm = True
+                        manager.netlinkq.append((msgtype, length, flags, seq, pid, data[0:length]))
+
+                    # There are certain message types we do not care about
+                    # (RTM_GETs for example)
+                    elif msgtype in ignore_messages:
+                        pass
+
+                    # And there are certain message types we have not added
+                    # support for yet (QDISC). Log an error for these just
+                    # as a reminder to add support for them.
+                    else:
+                        if msgtype in NetlinkPacket.type_to_string:
+                            log.warning('%s: RXed unsupported message %s (type %d)' %
+                                        (socket_string[s], NetlinkPacket.type_to_string[msgtype], msgtype))
+                        else:
+                            log.warning('%s: RXed unknown message type %d' %
+                                        (socket_string[s], msgtype))
+
+                    # Track the previous PID sequence number for RX and TX sockets
+                    if s == self.rx_socket:
+                        prev_seq = self.rx_socket_prev_seq
+                    elif s == manager.tx_socket:
+                        prev_seq = manager.tx_socket_prev_seq
+
+                    if pid in prev_seq and prev_seq[pid] and prev_seq[pid] != seq and (prev_seq[pid]+1 != seq):
+                        log.info('%s: Went from seq %d to %d' % (socket_string[s], prev_seq[pid], seq))
+                        missed_a_packet = True
+                    prev_seq[pid] = seq
+
+                    data = data[length:]
+
+            if set_tx_socket_rxed_ack_alarm:
+                manager.target_lock.acquire()
+                manager.target_seq = None
+                manager.target_pid = None
+                manager.target_lock.release()
+                manager.tx_socket_rxed_ack.set()
+
+            if set_alarm:
+                manager.workq.put(('SERVICE_NETLINK_QUEUE', None))
+                manager.alarm.set()
+
+        self.rx_socket.close()
+
+
+class NetlinkManagerWithListener(NetlinkManager):
+
+    def __init__(self, groups, start_listener=True):
+        NetlinkManager.__init__(self)
+        self.groups = groups
+        self.workq = Queue()
+        self.netlinkq = []
+        self.alarm = Event()
+        self.shutdown_event = Event()
+        self.tx_socket_rxed_ack = Event()
+        self.tx_socket_rxed_ack.clear()
+        self.target_seq = None
+        self.target_pid = None
+        self.target_seq_pid_debug = False
+        self.target_lock = Lock()
+        self.tx_socket_prev_seq = {}
+        self.debug_listener = False
+        self.debug_seq_pid = {}
+        self.ifname_by_index = {}
+        self.blacklist_filter = {}
+        self.whitelist_filter = {}
+
+        # Listen to netlink messages
+        if start_listener:
+            self.listener = NetlinkListener(self, self.groups)
+            self.listener.start()
+        else:
+            self.listener = None
+
+    def __str__(self):
+        return 'NetlinkManagerWithListener'
+
+    def signal_term_handler(self, signal, frame):
+        log.info("NetlinkManagerWithListener: Caught SIGTERM")
+        self.shutdown_flag = True  # For NetlinkManager shutdown
+        self.shutdown_event.set()
+        self.alarm.set()
+
+    def signal_int_handler(self, signal, frame):
+        log.info("NetlinkManagerWithListener: Caught SIGINT")
+        self.shutdown_flag = True  # For NetlinkManager shutdown
+        self.shutdown_event.set()
+        self.alarm.set()
+
+    def tx_nlpacket_ack_on_listener(self, nlpacket):
+        """
+        TX the message and wait for an ack
+        """
+
+        # NetlinkListener looks at the manager's target_seq and target_pid
+        # to know when we've RXed the ack that we want
+        self.target_lock.acquire()
+        self.target_seq = nlpacket.seq
+        self.target_pid = nlpacket.pid
+        self.target_lock.release()
+
+        if not self.tx_socket:
+            self.tx_socket_allocate()
+        self.tx_socket.sendall(nlpacket.message)
+
+        # Wait for NetlinkListener to RX an ACK or DONE for this (seq, pid)
+        self.tx_socket_rxed_ack.wait()
+        self.tx_socket_rxed_ack.clear()
+
+    # These are here to show some basic examples of how one might react to RXing
+    # various netlink message types. Odds are our child class will redefine these
+    # to do more than log a message.
+    def rx_rtm_newlink(self, msg):
+        log.info("RX RTM_NEWLINK for %s, state %s" % (msg.get_attribute_value(msg.IFLA_IFNAME), "up" if msg.is_up() else "down"))
+
+    def rx_rtm_dellink(self, msg):
+        log.info("RX RTM_DELLINK for %s, state %s" % (msg.get_attribute_value(msg.IFLA_IFNAME), "up" if msg.is_up() else "down"))
+
+    def rx_rtm_newaddr(self, msg):
+        log.info("RX RTM_NEWADDR for %s/%d on %s" %
+                 (msg.get_attribute_value(msg.IFA_ADDRESS), msg.prefixlen, self.ifname_by_index.get(msg.ifindex)))
+
+    def rx_rtm_deladdr(self, msg):
+        log.info("RX RTM_DELADDR for %s/%d on %s" %
+                 (msg.get_attribute_value(msg.IFA_ADDRESS), msg.prefixlen, self.ifname_by_index.get(msg.ifindex)))
+
+    def rx_rtm_newneigh(self, msg):
+        log.info("RX RTM_NEWNEIGH for %s on %s" % (msg.get_attribute_value(msg.NDA_DST), self.ifname_by_index.get(msg.ifindex)))
+
+    def rx_rtm_delneigh(self, msg):
+        log.info("RX RTM_DELNEIGH for %s on %s" % (msg.get_attribute_value(msg.NDA_DST), self.ifname_by_index.get(msg.ifindex)))
+
+    def rx_rtm_newroute(self, msg):
+        log.info("RX RTM_NEWROUTE for %s%s" %
+                 (msg.get_prefix_string(), msg.get_nexthops_string(self.ifname_by_index)))
+
+    def rx_rtm_delroute(self, msg):
+        log.info("RX RTM_NEWROUTE for %s%s" %
+                 (msg.get_prefix_string(), msg.get_nexthops_string(self.ifname_by_index)))
+
+    # Note that tx_nlpacket_ack_on_listener will block until NetlinkListener has RXed
+    # an Ack/DONE for the message we TXed
+    def get_all_addresses(self):
+        family = socket.AF_UNSPEC
+        debug = RTM_GETADDR in self.debug
+
+        addr = Address(RTM_GETADDR, debug)
+        addr.flags = NLM_F_REQUEST | NLM_F_DUMP
+        addr.body = pack('Bxxxi', family, 0)
+        addr.build_message(self.sequence.next(), self.pid)
+
+        if debug:
+            self.debug_seq_pid[(addr.seq, addr.pid)] = True
+
+        self.tx_nlpacket_ack_on_listener(addr)
+
+    def get_all_links(self):
+        family = socket.AF_UNSPEC
+        debug = RTM_GETLINK in self.debug
+
+        link = Link(RTM_GETLINK, debug)
+        link.flags = NLM_F_REQUEST | NLM_F_DUMP
+        link.body = pack('Bxxxiii', family, 0, 0, 0)
+        link.build_message(self.sequence.next(), self.pid)
+
+        if debug:
+            self.debug_seq_pid[(link.seq, link.pid)] = True
+
+        self.tx_nlpacket_ack_on_listener(link)
+
+    def get_all_neighbors(self):
+        family = socket.AF_UNSPEC
+        debug = RTM_GETNEIGH in self.debug
+
+        neighbor = Neighbor(RTM_GETNEIGH, debug)
+        neighbor.flags = NLM_F_REQUEST | NLM_F_DUMP
+        neighbor.body = pack('Bxxxii', family, 0, 0)
+        neighbor.build_message(self.sequence.next(), self.pid)
+
+        if debug:
+            self.debug_seq_pid[(neighbor.seq, neighbor.pid)] = True
+
+        self.tx_nlpacket_ack_on_listener(neighbor)
+
+    def get_all_routes(self):
+        family = socket.AF_UNSPEC
+        debug = RTM_GETROUTE in self.debug
+
+        route = Route(RTM_GETROUTE, debug)
+        route.flags = NLM_F_REQUEST | NLM_F_DUMP
+        route.body = pack('Bxxxii', family, 0, 0)
+        route.build_message(self.sequence.next(), self.pid)
+
+        if debug:
+            self.debug_seq_pid[(route.seq, route.pid)] = True
+
+        self.tx_nlpacket_ack_on_listener(route)
+
+    def nested_attributes_match(self, msg, attr_filter):
+        """
+        attr_filter will be a dictionary such as:
+        attr_filter = {
+            Link.IFLA_LINKINFO: {
+                Link.IFLA_INFO_KIND: 'vlan'
+            }
+        }
+        """
+        for (key, value) in attr_filter.items():
+            if type(value) is dict:
+                if not self.nested_attributes_match(msg, value):
+                    return False
+            else:
+                attr_value = msg.get_attribute_value(key)
+                if attr_value != value:
+                    return False
+        return True
+
+    def filter_rule_matches(self, msg, rule):
+        field = rule[0]
+        options = rule[1:]
+
+        if field == 'IFINDEX':
+            ifindex = options[0]
+
+            if msg.ifindex == ifindex:
+                return True
+
+        elif field == 'ATTRIBUTE':
+            (attr_type, target_value) = options[0:2]
+            attr_value = msg.get_attribute_value(attr_type)
+
+            if attr_value == target_value:
+                return True
+
+        elif field == 'NESTED_ATTRIBUTE':
+            if self.nested_attributes_match(msg, options[0]):
+                return True
+
+        elif field == 'FAMILY':
+            family = options[0]
+
+            if msg.family == family:
+                return True
+        else:
+            raise Exception("Add support to filter based on %s" % field)
+
+        return False
+
+    def filter_permit(self, msg):
+        """
+        Return True if our whitelist/blacklist filters permit this netlink msg
+        """
+        if msg.msgtype in self.whitelist_filter:
+            found_it = False
+
+            for rule in self.whitelist_filter[msg.msgtype]:
+                if self.filter_rule_matches(msg, rule):
+                    found_it = True
+                    break
+
+            return found_it
+
+        elif msg.msgtype in self.blacklist_filter:
+            for rule in self.blacklist_filter[msg.msgtype]:
+                if self.filter_rule_matches(msg, rule):
+                    return False
+            return True
+
+        else:
+            return True
+
+    def _filter_update(self, add, filter_type, msgtype, filter_guts):
+        assert filter_type in ('whitelist', 'blacklist'), "whitelist and blacklist are the only supported filter options"
+
+        if add:
+            if filter_type == 'whitelist':
+
+                # Keep things simple, do not allow both whitelist and blacklist
+                if self.blacklist_filter and self.blacklist_filter.get(msgtype):
+                    raise Exception("whitelist and blacklist filters cannot be used at the same time")
+
+                if msgtype not in self.whitelist_filter:
+                    self.whitelist_filter[msgtype] = []
+                self.whitelist_filter[msgtype].append(filter_guts)
+
+            elif filter_type == 'blacklist':
+
+                # Keep things simple, do not allow both whitelist and blacklist
+                if self.whitelist_filter and self.whitelist_filter.get(msgtype):
+                    raise Exception("whitelist and blacklist filters cannot be used at the same time")
+
+                if msgtype not in self.blacklist_filter:
+                    self.blacklist_filter[msgtype] = []
+                self.blacklist_filter[msgtype].append(filter_guts)
+
+        else:
+            if filter_type == 'whitelist':
+                if msgtype in self.whitelist_filter:
+                    self.whitelist_filter[msgtype].remove(filter_guts)
+
+                    if not self.whitelist_filter[msgtype]:
+                        del self.whitelist_filter[msgtype]
+
+            elif filter_type == 'blacklist':
+                if msgtype in self.blacklist_filter:
+                    self.blacklist_filter[msgtype].remove(filter_guts)
+
+                    if not self.blacklist_filter[msgtype]:
+                        del self.blacklist_filter[msgtype]
+
+    def filter_by_address_family(self, add, filter_type, msgtype, family):
+        self._filter_update(add, filter_type, msgtype, ('FAMILY', family))
+
+    def filter_by_ifindex(self, add, filter_type, msgtype, ifindex):
+        self._filter_update(add, filter_type, msgtype, ('IFINDEX', ifindex))
+
+    def filter_by_attribute(self, add, filter_type, msgtype, attribute, attribute_value):
+        self._filter_update(add, filter_type, msgtype, ('ATTRIBUTE', attribute, attribute_value))
+
+    def filter_by_nested_attribute(self, add, filter_type, msgtype, attr_filter):
+        self._filter_update(add, filter_type, msgtype, ('NESTED_ATTRIBUTE', attr_filter))
+
+    def service_netlinkq(self):
+        msg_count = {}
+        processed = 0
+
+        for (msgtype, length, flags, seq, pid, data) in self.netlinkq:
+            processed += 1
+
+            # If this is a reply to a TX message that debugs were enabled for then debug the reply
+            if (seq, pid) in self.debug_seq_pid:
+                debug = True
+            else:
+                debug = self.debug_this_packet(msgtype)
+
+            if msgtype == RTM_NEWLINK or msgtype == RTM_DELLINK:
+                msg = Link(msgtype, debug)
+
+            elif msgtype == RTM_NEWADDR or msgtype == RTM_DELADDR:
+                msg = Address(msgtype, debug)
+
+            elif msgtype == RTM_NEWNEIGH or msgtype == RTM_DELNEIGH:
+                msg = Neighbor(msgtype, debug)
+
+            elif msgtype == RTM_NEWROUTE or msgtype == RTM_DELROUTE:
+                msg = Route(msgtype, debug)
+
+            else:
+                log.warning('RXed unknown netlink message type %s' % msgtype)
+                continue
+
+            msg.decode_packet(length, flags, seq, pid, data)
+
+            if not self.filter_permit(msg):
+                continue
+
+            if debug:
+                msg.dump()
+
+            # Only used for printing debugs about how many we RXed of each type
+            if msg.msgtype not in msg_count:
+                msg_count[msg.msgtype] = 0
+            msg_count[msg.msgtype] += 1
+
+            # Call the appropriate handler method based on the msgtype.  The handler
+            # functions are defined in our child class.
+            if msg.msgtype == RTM_NEWLINK:
+
+                # We will use ifname_by_index to display the interface name in debug output
+                self.ifname_by_index[msg.ifindex] = msg.get_attribute_value(msg.IFLA_IFNAME)
+                self.rx_rtm_newlink(msg)
+
+            elif msg.msgtype == RTM_DELLINK:
+
+                # We will use ifname_by_index to display the interface name in debug output
+                if msg.ifindex in self.ifname_by_index:
+                    del self.ifname_by_index[msg.ifindex]
+                self.rx_rtm_dellink(msg)
+
+            elif msg.msgtype == RTM_NEWADDR:
+                self.rx_rtm_newaddr(msg)
+
+            elif msg.msgtype == RTM_DELADDR:
+                self.rx_rtm_deladdr(msg)
+
+            elif msg.msgtype == RTM_NEWNEIGH:
+                self.rx_rtm_newneigh(msg)
+
+            elif msg.msgtype == RTM_DELNEIGH:
+                self.rx_rtm_delneigh(msg)
+
+            elif msg.msgtype == RTM_NEWROUTE:
+                self.rx_rtm_newroute(msg)
+
+            elif msg.msgtype == RTM_DELROUTE:
+                self.rx_rtm_delroute(msg)
+
+            else:
+                log.warning('RXed unknown netlink message type %s' % msgtype)
+
+        if processed:
+            self.netlinkq = self.netlinkq[processed:]
+
+        # too chatty
+        # for msgtype in msg_count:
+        #     log.debug('RXed %d %s messages' % (msg_count[msgtype], NetlinkPacket.type_to_string[msgtype]))

nlmanager/nlmanager.py

@@ -0,0 +1,591 @@
+#!/usr/bin/env python
+
+from ipaddr import IPv4Address, IPv6Address
+from nlpacket import *
+from select import select
+from struct import pack, unpack
+from tabulate import tabulate
+import logging
+import os
+import socket
+
+log = logging.getLogger(__name__)
+
+
+class NetlinkError(Exception):
+    pass
+
+
+class NetlinkNoAddressError(Exception):
+    pass
+
+
+class InvalidInterfaceNameVlanCombo(Exception):
+    pass
+
+
+class Sequence(object):
+
+    def __init__(self):
+        self._next = 0
+
+    def next(self):
+        self._next += 1
+        return self._next
+
+
+class NetlinkManager(object):
+
+    def __init__(self):
+        self.pid = os.getpid()
+        self.sequence = Sequence()
+        self.shutdown_flag = False
+        self.ifindexmap = {}
+        self.tx_socket = None
+
+        # debugs
+        self.debug = {}
+        self.debug_link(False)
+        self.debug_address(False)
+        self.debug_neighbor(False)
+        self.debug_route(False)
+
+    def __str__(self):
+        return 'NetlinkManager'
+
+    def signal_term_handler(self, signal, frame):
+        log.info("NetlinkManager: Caught SIGTERM")
+        self.shutdown_flag = True
+
+    def signal_int_handler(self, signal, frame):
+        log.info("NetlinkManager: Caught SIGINT")
+        self.shutdown_flag = True
+
+    def shutdown(self):
+        if self.tx_socket:
+            self.tx_socket.close()
+            self.tx_socket = None
+        log.info("NetlinkManager: shutdown complete")
+
+    def _debug_set_clear(self, msg_types, enabled):
+        """
+        Enable or disable debugs for all msgs_types messages
+        """
+
+        for x in msg_types:
+            if enabled:
+                self.debug[x] = True
+            else:
+                if x in self.debug:
+                    del self.debug[x]
+
+    def debug_link(self, enabled):
+        self._debug_set_clear((RTM_NEWLINK, RTM_DELLINK, RTM_GETLINK, RTM_SETLINK), enabled)
+
+    def debug_address(self, enabled):
+        self._debug_set_clear((RTM_NEWADDR, RTM_DELADDR, RTM_GETADDR), enabled)
+
+    def debug_neighbor(self, enabled):
+        self._debug_set_clear((RTM_NEWNEIGH, RTM_DELNEIGH, RTM_GETNEIGH), enabled)
+
+    def debug_route(self, enabled):
+        self._debug_set_clear((RTM_NEWROUTE, RTM_DELROUTE, RTM_GETROUTE), enabled)
+
+    def debug_this_packet(self, mtype):
+        if mtype in self.debug:
+            return True
+        return False
+
+    def tx_socket_allocate(self):
+        """
+        The TX socket is used for install requests, sending RTM_GETXXXX
+        requests, etc
+        """
+        self.tx_socket = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, 0)
+        self.tx_socket.bind((self.pid, 0))
+
+    def tx_nlpacket_raw(self, message):
+        """
+        TX a bunch of concatenated nlpacket.messages....do NOT wait for an ACK
+        """
+        if not self.tx_socket:
+            self.tx_socket_allocate()
+        self.tx_socket.sendall(message)
+
+    def tx_nlpacket(self, nlpacket):
+        """
+        TX a netlink packet but do NOT wait for an ACK
+        """
+        if not nlpacket.message:
+            log.error('You must first call build_message() to create the packet')
+            return
+
+        if not self.tx_socket:
+            self.tx_socket_allocate()
+        self.tx_socket.sendall(nlpacket.message)
+
+    def tx_nlpacket_get_response(self, nlpacket):
+
+        if not nlpacket.message:
+            log.error('You must first call build_message() to create the packet')
+            return
+
+        if not self.tx_socket:
+            self.tx_socket_allocate()
+        self.tx_socket.sendall(nlpacket.message)
+
+        # If debugs are enabled we will print the contents of the
+        # packet via the decode_packet call...so avoid printing
+        # two messages for one packet.
+        if not nlpacket.debug:
+            log.info("TXed %12s, pid %d, seq %d, %d bytes" %
+                     (nlpacket.get_type_string(), nlpacket.pid, nlpacket.seq, nlpacket.length))
+
+        header_PACK = NetlinkPacket.header_PACK
+        header_LEN = NetlinkPacket.header_LEN
+        null_read = 0
+        MAX_NULL_READS = 30
+        msgs = []
+
+        # Now listen to our socket and wait for the reply
+        while True:
+
+            if self.shutdown_flag:
+                log.info('shutdown flag is True, exiting')
+                return msgs
+
+            # Only block for 1 second so we can wake up to see if self.shutdown_flag is True
+            (readable, writeable, exceptional) = select([self.tx_socket, ], [], [self.tx_socket, ], 1)
+
+            if not readable:
+                null_read += 1
+
+                # Safety net to make sure we do not spend too much time in
+                # this while True loop
+                if null_read >= MAX_NULL_READS:
+                    log.warning('Socket was not readable for %d attempts' % null_read)
+                    return msgs
+
+                continue
+
+            for s in readable:
+                data = s.recv(4096)
+
+                if not data:
+                    log.info('RXed zero length data, the socket is closed')
+                    return msgs
+
+                while data:
+
+                    # Extract the length, etc from the header
+                    (length, msgtype, flags, seq, pid) = unpack(header_PACK, data[:header_LEN])
+
+                    debug_str = "RXed %12s, pid %d, seq %d, %d bytes" % (NetlinkPacket.type_to_string[msgtype], pid, seq, length)
+
+                    # This shouldn't happen but it would be nice to be aware of it if it does
+                    if pid != nlpacket.pid:
+                        log.debug(debug_str + '...we are not interested in this pid %s since ours is %s' %
+                                    (pid, nlpacket.pid))
+                        data = data[length:]
+                        continue
+                    if seq != nlpacket.seq:
+                        log.debug(debug_str + '...we are not interested in this seq %s since ours is %s' %
+                                    (seq, nlpacket.seq))
+                        data = data[length:]
+                        continue
+                    # See if we RXed an ACK for our RTM_GETXXXX
+                    if msgtype == NLMSG_DONE:
+                        log.debug(debug_str + '...this is an ACK')
+                        return msgs
+
+                    elif msgtype == NLMSG_ERROR:
+
+                        # The error code is a signed negative number.
+                        error_code = abs(unpack('=i', data[header_LEN:header_LEN+4])[0])
+                        msg = Error(msgtype, nlpacket.debug)
+                        msg.decode_packet(length, flags, seq, pid, data)
+
+                        debug_str += ", error code %s" % msg.error_to_string.get(error_code)
+
+                        # 0 is NLE_SUCCESS...everything else is a true error
+                        if error_code:
+                            if error_code == Error.NLE_NOADDR:
+                                raise NetlinkNoAddressError(debug_str)
+                            else:
+                                raise NetlinkError(debug_str)
+                        else:
+                            log.info(debug_str + '...this is an ACK')
+                            return msgs
+
+                    # No ACK...create a nlpacket object and append it to msgs
+                    else:
+
+                        # If debugs are enabled we will print the contents of the
+                        # packet via the decode_packet call...so avoid printing
+                        # two messages for one packet.
+                        if not nlpacket.debug:
+                            log.info(debug_str)
+
+                        if msgtype == RTM_NEWLINK or msgtype == RTM_DELLINK:
+                            msg = Link(msgtype, nlpacket.debug)
+
+                        elif msgtype == RTM_NEWADDR or msgtype == RTM_DELADDR:
+                            msg = Address(msgtype, nlpacket.debug)
+
+                        elif msgtype == RTM_NEWNEIGH or msgtype == RTM_DELNEIGH:
+                            msg = Neighbor(msgtype, nlpacket.debug)
+
+                        elif msgtype == RTM_NEWROUTE or msgtype == RTM_DELROUTE:
+                            msg = Route(msgtype, nlpacket.debug)
+
+                        else:
+                            raise Exception("RXed unknown netlink message type %s" % msgtype)
+
+                        msg.decode_packet(length, flags, seq, pid, data)
+                        msgs.append(msg)
+
+                    data = data[length:]
+
+    def ip_to_afi(self, ip):
+        type_ip = type(ip)
+
+        if type_ip == IPv4Address:
+            return socket.AF_INET
+        elif type_ip == IPv6Address:
+            return socket.AF_INET6
+        else:
+            raise Exception("%s is an invalid IP type" % type_ip)
+
+    def request_dump(self, rtm_type, family, debug):
+        """
+        Issue a RTM_GETROUTE, etc with the NLM_F_DUMP flag
+        set and return the results
+        """
+
+        if rtm_type == RTM_GETADDR:
+            msg = Address(rtm_type, debug)
+            msg.body = pack('Bxxxi', family, 0)
+
+        elif rtm_type == RTM_GETLINK:
+            msg = Link(rtm_type, debug)
+            msg.body = pack('Bxxxiii', family, 0, 0, 0)
+
+        elif rtm_type == RTM_GETNEIGH:
+            msg = Neighbor(rtm_type, debug)
+            msg.body = pack('Bxxxii', family, 0, 0)
+
+        elif rtm_type == RTM_GETROUTE:
+            msg = Route(rtm_type, debug)
+            msg.body = pack('Bxxxii', family, 0, 0)
+
+        else:
+            log.error("request_dump RTM_GET %s is not supported" % rtm_type)
+            return None
+
+        msg.flags = NLM_F_REQUEST | NLM_F_DUMP
+        msg.attributes = {}
+        msg.build_message(self.sequence.next(), self.pid)
+        return self.tx_nlpacket_get_response(msg)
+
+    # ======
+    # Routes
+    # ======
+    def _routes_add_or_delete(self, add_route, routes, ecmp_routes, table, protocol, route_scope, route_type):
+
+        def tx_or_concat_message(total_message, route):
+            """
+            Adding an ipv4 route only takes 60 bytes, if we are adding thousands
+            of them this can add up to a lot of send calls.  Concat several of
+            them together before TXing.
+            """
+
+            if not total_message:
+                total_message = route.message
+            else:
+                total_message += route.message
+
+            if len(total_message) >= PACKET_CONCAT_SIZE:
+                self.tx_nlpacket_raw(total_message)
+                total_message = None
+
+            return total_message
+
+        if add_route:
+            rtm_command = RTM_NEWROUTE
+        else:
+            rtm_command = RTM_DELROUTE
+
+        total_message = None
+        PACKET_CONCAT_SIZE = 16384
+        debug = rtm_command in self.debug
+
+        if routes:
+            for (afi, ip, mask, nexthop, interface_index) in routes:
+                route = Route(rtm_command, debug)
+                route.flags = NLM_F_REQUEST | NLM_F_CREATE
+                route.body = pack('BBBBBBBBi', afi, mask, 0, 0, table, protocol,
+                                  route_scope, route_type, 0)
+                route.family = afi
+                route.add_attribute(Route.RTA_DST, ip)
+                if nexthop:
+                    route.add_attribute(Route.RTA_GATEWAY, nexthop)
+                route.add_attribute(Route.RTA_OIF, interface_index)
+                route.build_message(self.sequence.next(), self.pid)
+                total_message = tx_or_concat_message(total_message, route)
+
+            if total_message:
+                self.tx_nlpacket_raw(total_message)
+
+        if ecmp_routes:
+
+            for (route_key, value) in ecmp_routes.iteritems():
+                (afi, ip, mask) = route_key
+
+                route = Route(rtm_command, debug)
+                route.flags = NLM_F_REQUEST | NLM_F_CREATE
+                route.body = pack('BBBBBBBBi', afi, mask, 0, 0, table, protocol,
+                                  route_scope, route_type, 0)
+                route.family = afi
+                route.add_attribute(Route.RTA_DST, ip)
+                route.add_attribute(Route.RTA_MULTIPATH, value)
+                route.build_message(self.sequence.next(), self.pid)
+                total_message = tx_or_concat_message(total_message, route)
+
+            if total_message:
+                self.tx_nlpacket_raw(total_message)
+
+    def routes_add(self, routes, ecmp_routes,
+                   table=Route.RT_TABLE_MAIN,
+                   protocol=Route.RT_PROT_XORP,
+                   route_scope=Route.RT_SCOPE_UNIVERSE,
+                   route_type=Route.RTN_UNICAST):
+        self._routes_add_or_delete(True, routes, ecmp_routes, table, protocol, route_scope, route_type)
+
+    def routes_del(self, routes, ecmp_routes,
+                   table=Route.RT_TABLE_MAIN,
+                   protocol=Route.RT_PROT_XORP,
+                   route_scope=Route.RT_SCOPE_UNIVERSE,
+                   route_type=Route.RTN_UNICAST):
+        self._routes_add_or_delete(False, routes, ecmp_routes, table, protocol, route_scope, route_type)
+
+    def route_get(self, ip, debug=False):
+        """
+        ip must be one of the following:
+        - IPv4Address
+        - IPv6Address
+        """
+        # Transmit a RTM_GETROUTE to query for the route we want
+        route = Route(RTM_GETROUTE, debug)
+        route.flags = NLM_F_REQUEST | NLM_F_ACK
+
+        # Set everything in the service header as 0 other than the afi
+        afi = self.ip_to_afi(ip)
+        route.body = pack('Bxxxxxxxi', afi, 0)
+        route.family = afi
+        route.add_attribute(Route.RTA_DST, ip)
+        route.build_message(self.sequence.next(), self.pid)
+        return self.tx_nlpacket_get_response(route)
+
+    def routes_dump(self, family=socket.AF_UNSPEC, debug=True):
+        return self.request_dump(RTM_GETROUTE, family, debug)
+
+    def routes_print(self, routes):
+        """
+        Use tabulate to print a table of 'routes'
+        """
+        header = ['Prefix', 'nexthop', 'ifindex']
+        table = []
+
+        for x in routes:
+            if Route.RTA_DST not in x.attributes:
+                log.warning("Route is missing RTA_DST")
+                continue
+
+            table.append(('%s/%d' % (x.attributes[Route.RTA_DST].value, x.src_len),
+                          str(x.attributes[Route.RTA_GATEWAY].value) if Route.RTA_GATEWAY in x.attributes else None,
+                          x.attributes[Route.RTA_OIF].value))
+
+        print tabulate(table, header, tablefmt='simple') + '\n'
+
+    # =====
+    # Links
+    # =====
+    def _get_iface_by_name(self, ifname):
+        """
+        Return a Link object for ifname
+        """
+        debug = RTM_GETLINK in self.debug
+
+        link = Link(RTM_GETLINK, debug)
+        link.flags = NLM_F_REQUEST | NLM_F_ACK
+        link.body = pack('=Bxxxiii', socket.AF_UNSPEC, 0, 0, 0)
+        link.add_attribute(Link.IFLA_IFNAME, ifname)
+        link.build_message(self.sequence.next(), self.pid)
+
+        try:
+            return self.tx_nlpacket_get_response(link)[0]
+
+        except NetlinkNoAddressError:
+            log.info("Netlink did not find interface %s" % ifname)
+            return None
+
+    def get_iface_index(self, ifname):
+        """
+        Return the interface index for ifname
+        """
+        iface = self._get_iface_by_name(ifname)
+
+        if iface:
+            return iface.ifindex
+        return None
+
+    def _link_add(self, ifindex, ifname, kind, ifla_info_data):
+        """
+        Build and TX a RTM_NEWLINK message to add the desired interface
+        """
+        debug = RTM_NEWLINK in self.debug
+
+        link = Link(RTM_NEWLINK, debug)
+        link.flags = NLM_F_CREATE | NLM_F_REQUEST
+        link.body = pack('Bxxxiii', socket.AF_UNSPEC, 0, 0, 0)
+        link.add_attribute(Link.IFLA_IFNAME, ifname)
+        link.add_attribute(Link.IFLA_LINK, ifindex)
+        link.add_attribute(Link.IFLA_LINKINFO, {
+            Link.IFLA_INFO_KIND: kind,
+            Link.IFLA_INFO_DATA: ifla_info_data
+        })
+        link.build_message(self.sequence.next(), self.pid)
+        return self.tx_nlpacket(link)
+
+    def link_add_vlan(self, ifindex, ifname, vlanid):
+        """
+        ifindex is the index of the parent interface that this sub-interface
+        is being added to
+        """
+
+        '''
+        If you name an interface swp2.17 but assign it to vlan 12, the kernel
+        will return a very misleading NLE_MSG_OVERFLOW error.  It only does
+        this check if the ifname uses dot notation.
+
+        Do this check here so we can provide a more intuitive error
+        '''
+        if '.' in ifname:
+            ifname_vlanid = int(ifname.split('.')[-1])
+
+            if ifname_vlanid != vlanid:
+                raise InvalidInterfaceNameVlanCombo("Interface %s must belong "
+                                                    "to VLAN %d (VLAN %d was requested)" %
+                                                    (ifname, ifname_vlanid, vlanid))
+
+        return self._link_add(ifindex, ifname, 'vlan', {Link.IFLA_VLAN_ID: vlanid})
+
+    def link_add_macvlan(self, ifindex, ifname):
+        """
+        ifindex is the index of the parent interface that this sub-interface
+        is being added to
+        """
+        return self._link_add(ifindex, ifname, 'macvlan', {Link.IFLA_MACVLAN_MODE: Link.MACVLAN_MODE_PRIVATE})
+
+    def _link_bridge_vlan(self, msgtype, ifindex, vlanid, pvid, untagged, master):
+        """
+        Build and TX a RTM_NEWLINK message to add the desired interface
+        """
+
+        if master:
+            flags = 0
+        else:
+            flags = Link.BRIDGE_FLAGS_SELF
+
+        if pvid:
+            vflags = Link.BRIDGE_VLAN_INFO_PVID | Link.BRIDGE_VLAN_INFO_UNTAGGED
+        elif untagged:
+            vflags = Link.BRIDGE_VLAN_INFO_UNTAGGED
+        else:
+            vflags = 0
+
+        debug = msgtype in self.debug
+
+        link = Link(msgtype, debug)
+        link.flags = NLM_F_REQUEST | NLM_F_ACK
+        link.body = pack('Bxxxiii', socket.AF_BRIDGE, ifindex, 0, 0)
+        link.add_attribute(Link.IFLA_AF_SPEC, {
+            Link.IFLA_BRIDGE_FLAGS: flags,
+            Link.IFLA_BRIDGE_VLAN_INFO: (vflags, vlanid)
+        })
+        link.build_message(self.sequence.next(), self.pid)
+        return self.tx_nlpacket(link)
+
+    def link_add_bridge_vlan(self, ifindex, vlanid, pvid=False, untagged=False, master=False):
+        self._link_bridge_vlan(RTM_SETLINK, ifindex, vlanid, pvid, untagged, master)
+
+    def link_del_bridge_vlan(self, ifindex, vlanid, pvid=False, untagged=False, master=False):
+        self._link_bridge_vlan(RTM_DELLINK, ifindex, vlanid, pvid, untagged, master)
+
+    def link_set_updown(self, ifname, state):
+        """
+        Either bring ifname up or take it down
+        """
+
+        if state == 'up':
+            if_flags = Link.IFF_UP
+        elif state == 'down':
+            if_flags = 0
+        else:
+            raise Exception('Unsupported state %s, valid options are "up" and "down"' % state)
+
+        debug = RTM_NEWLINK in self.debug
+        if_change = Link.IFF_UP
+
+        link = Link(RTM_NEWLINK, debug)
+        link.flags = NLM_F_REQUEST
+        link.body = pack('=BxxxiLL', socket.AF_UNSPEC, 0, if_flags, if_change)
+        link.add_attribute(Link.IFLA_IFNAME, ifname)
+        link.build_message(self.sequence.next(), self.pid)
+        return self.tx_nlpacket(link)
+
+    def link_set_protodown(self, ifname, state):
+        """
+        Either bring ifname up or take it down by setting IFLA_PROTO_DOWN on or off
+        """
+        flags = 0
+        protodown = 1 if state == "on" else 0
+
+        debug = RTM_NEWLINK in self.debug
+
+        link = Link(RTM_NEWLINK, debug)
+        link.flags = NLM_F_REQUEST
+        link.body = pack('=BxxxiLL', socket.AF_UNSPEC, 0, 0, 0)
+        link.add_attribute(Link.IFLA_IFNAME, ifname)
+        link.add_attribute(Link.IFLA_PROTO_DOWN, protodown)
+        link.build_message(self.sequence.next(), self.pid)
+        return self.tx_nlpacket(link)
+
+    # =========
+    # Neighbors
+    # =========
+    def neighbor_add(self, afi, ifindex, ip, mac):
+        debug = RTM_NEWNEIGH in self.debug
+        service_hdr_flags = 0
+
+        nbr = Neighbor(RTM_NEWNEIGH, debug)
+        nbr.flags = NLM_F_CREATE | NLM_F_REQUEST
+        nbr.family = afi
+        nbr.body = pack('=BxxxiHBB', afi, ifindex, Neighbor.NUD_REACHABLE, service_hdr_flags, Route.RTN_UNICAST)
+        nbr.add_attribute(Neighbor.NDA_DST, ip)
+        nbr.add_attribute(Neighbor.NDA_LLADDR, mac)
+        nbr.build_message(self.sequence.next(), self.pid)
+        return self.tx_nlpacket(nbr)
+
+    def neighbor_del(self, afi, ifindex, ip, mac):
+        debug = RTM_DELNEIGH in self.debug
+        service_hdr_flags = 0
+
+        nbr = Neighbor(RTM_DELNEIGH, debug)
+        nbr.flags = NLM_F_REQUEST
+        nbr.family = afi
+        nbr.body = pack('=BxxxiHBB', afi, ifindex, Neighbor.NUD_REACHABLE, service_hdr_flags, Route.RTN_UNICAST)
+        nbr.add_attribute(Neighbor.NDA_DST, ip)
+        nbr.add_attribute(Neighbor.NDA_LLADDR, mac)
+        nbr.build_message(self.sequence.next(), self.pid)
+        return self.tx_nlpacket(nbr)

setup.py

@@ -18,6 +18,13 @@ setup(name='ifupdown2',
                       'addons/addressvirtual.py', 'addons/vxlan.py',
                       'addons/link.py', 'addons/vrf.py',
                       'addons/bridgevlan.py']),
+                   ('/usr/share/ifupdown2/nlmanager/',
+                    ['nlmanager/nllistener.py',
+                     'nlmanager/nlmanager.py',
+                     'nlmanager/nlpacket.py',
+                     'nlmanager/__init__.py',
+                     'nlmanager/README']),
+                   ('/etc/network/ifupdown2/', ['config/addons.conf']),
                   ('/etc/network/ifupdown2/', ['config/addons.conf']),
                   ('/var/lib/ifupdown2/policy.d/', []),
                   ('/etc/network/ifupdown2/policy.d/', [])