path: root/python/netlink/route/link.py

#
# Copyright (c) 2011 Thomas Graf <tgraf@suug.ch>
#

"""Module providing access to network links

This module provides an interface to view configured network links,
modify them and to add and delete virtual network links.

The following is a basic example:
    import netlink.core as netlink
    import netlink.route.link as link

    sock = netlink.Socket()
    sock.connect(netlink.NETLINK_ROUTE)

    cache = link.LinkCache()	# create new empty link cache
    cache.refill(sock)		# fill cache with all configured links
    eth0 = cache['eth0']		# lookup link "eth0"
    print eth0			# print basic configuration

The module contains the following public classes:

  - Link -- Represents a network link. Instances can be created directly
        via the constructor (empty link objects) or via the refill()
        method of a LinkCache.
  - LinkCache -- Derived from netlink.Cache, holds any number of
         network links (Link instances). Main purpose is to keep
         a local list of all network links configured in the
         kernel.

The following public functions exist:
  - get_from_kernel(socket, name)

"""

from __future__ import absolute_import

__version__ = '0.1'
__all__ = [
    'LinkCache',
    'Link',
]

import socket
from .. import core as netlink
from .. import capi as core_capi
from .  import capi as capi
from .links  import inet as inet
from .. import util as util

# Link statistics definitions
RX_PACKETS = 0
TX_PACKETS = 1
RX_BYTES = 2
TX_BYTES = 3
RX_ERRORS = 4
TX_ERRORS = 5
RX_DROPPED = 6
TX_DROPPED = 7
RX_COMPRESSED = 8
TX_COMPRESSED = 9
RX_FIFO_ERR = 10
TX_FIFO_ERR = 11
RX_LEN_ERR = 12
RX_OVER_ERR = 13
RX_CRC_ERR = 14
RX_FRAME_ERR = 15
RX_MISSED_ERR = 16
TX_ABORT_ERR = 17
TX_CARRIER_ERR = 18
TX_HBEAT_ERR = 19
TX_WIN_ERR = 20
COLLISIONS = 21
MULTICAST = 22
IP6_INPKTS = 23
IP6_INHDRERRORS = 24
IP6_INTOOBIGERRORS = 25
IP6_INNOROUTES = 26
IP6_INADDRERRORS = 27
IP6_INUNKNOWNPROTOS = 28
IP6_INTRUNCATEDPKTS = 29
IP6_INDISCARDS = 30
IP6_INDELIVERS = 31
IP6_OUTFORWDATAGRAMS = 32
IP6_OUTPKTS = 33
IP6_OUTDISCARDS = 34
IP6_OUTNOROUTES = 35
IP6_REASMTIMEOUT = 36
IP6_REASMREQDS = 37
IP6_REASMOKS = 38
IP6_REASMFAILS = 39
IP6_FRAGOKS = 40
IP6_FRAGFAILS = 41
IP6_FRAGCREATES = 42
IP6_INMCASTPKTS = 43
IP6_OUTMCASTPKTS = 44
IP6_INBCASTPKTS = 45
IP6_OUTBCASTPKTS = 46
IP6_INOCTETS = 47
IP6_OUTOCTETS = 48
IP6_INMCASTOCTETS = 49
IP6_OUTMCASTOCTETS = 50
IP6_INBCASTOCTETS = 51
IP6_OUTBCASTOCTETS = 52
ICMP6_INMSGS = 53
ICMP6_INERRORS = 54
ICMP6_OUTMSGS = 55
ICMP6_OUTERRORS = 56

class LinkCache(netlink.Cache):
    """Cache of network links"""

    def __init__(self, family=socket.AF_UNSPEC, cache=None):
        if not cache:
            cache = self._alloc_cache_name('route/link')

        self._info_module = None
        self._protocol = netlink.NETLINK_ROUTE
        self._nl_cache = cache
        self._set_arg1(family)

    def __getitem__(self, key):
        if type(key) is int:
            link = capi.rtnl_link_get(self._nl_cache, key)
        else:
            link = capi.rtnl_link_get_by_name(self._nl_cache, key)

        if link is None:
            raise KeyError()
        else:
            return Link.from_capi(link)

    @staticmethod
    def _new_object(obj):
        return Link(obj)

    def _new_cache(self, cache):
        return LinkCache(family=self.arg1, cache=cache)

class Link(netlink.Object):
    """Network link"""

    def __init__(self, obj=None):
        netlink.Object.__init__(self, 'route/link', 'link', obj)
        self._rtnl_link = self._obj2type(self._nl_object)

        if self.type:
            self._module_lookup('netlink.route.links.' + self.type)

        self.inet = inet.InetLink(self)
        self.af = {'inet' : self.inet }

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, tb):
        if exc_type is None:
            self.change()
        else:
            return False

    @classmethod
    def from_capi(cls, obj):
        return cls(capi.link2obj(obj))

    @staticmethod
    def _obj2type(obj):
        return capi.obj2link(obj)

    def __cmp__(self, other):
        return self.ifindex - other.ifindex

    @staticmethod
    def _new_instance(obj):
        if not obj:
            raise ValueError()

        return Link(obj)

    @property
    @netlink.nlattr(type=int, immutable=True, fmt=util.num)
    def ifindex(self):
        """interface index"""
        return capi.rtnl_link_get_ifindex(self._rtnl_link)

    @ifindex.setter
    def ifindex(self, value):
        capi.rtnl_link_set_ifindex(self._rtnl_link, int(value))

        # ifindex is immutable but we assume that if _orig does not
        # have an ifindex specified, it was meant to be given here
        if capi.rtnl_link_get_ifindex(self._orig) == 0:
            capi.rtnl_link_set_ifindex(self._orig, int(value))

    @property
    @netlink.nlattr(type=str, fmt=util.bold)
    def name(self):
        """Name of link"""
        return capi.rtnl_link_get_name(self._rtnl_link)

    @name.setter
    def name(self, value):
        capi.rtnl_link_set_name(self._rtnl_link, value)

        # name is the secondary identifier, if _orig does not have
        # the name specified yet, assume it was meant to be specified
        # here. ifindex will always take priority, therefore if ifindex
        # is specified as well, this will be ignored automatically.
        if capi.rtnl_link_get_name(self._orig) is None:
            capi.rtnl_link_set_name(self._orig, value)

    @property
    @netlink.nlattr(type=str, fmt=util.string)
    def flags(self):
        """Flags
        Setting this property will *Not* reset flags to value you supply in
        Examples:
        link.flags = '+xxx' # add xxx flag
        link.flags = 'xxx'  # exactly the same
        link.flags = '-xxx' # remove xxx flag
        link.flags = [ '+xxx', '-yyy' ] # list operation
        """
        flags = capi.rtnl_link_get_flags(self._rtnl_link)
        return capi.rtnl_link_flags2str(flags, 256)[0].split(',')

    def _set_flag(self, flag):
        if flag.startswith('-'):
            i = capi.rtnl_link_str2flags(flag[1:])
            capi.rtnl_link_unset_flags(self._rtnl_link, i)
        elif flag.startswith('+'):
            i = capi.rtnl_link_str2flags(flag[1:])
            capi.rtnl_link_set_flags(self._rtnl_link, i)
        else:
            i = capi.rtnl_link_str2flags(flag)
            capi.rtnl_link_set_flags(self._rtnl_link, i)

    @flags.setter
    def flags(self, value):
        if not (type(value) is str):
            for flag in value:
                self._set_flag(flag)
        else:
            self._set_flag(value)

    @property
    @netlink.nlattr(type=int, fmt=util.num)
    def mtu(self):
        """Maximum Transmission Unit"""
        return capi.rtnl_link_get_mtu(self._rtnl_link)

    @mtu.setter
    def mtu(self, value):
        capi.rtnl_link_set_mtu(self._rtnl_link, int(value))

    @property
    @netlink.nlattr(type=int, immutable=True, fmt=util.num)
    def family(self):
        """Address family"""
        return capi.rtnl_link_get_family(self._rtnl_link)

    @family.setter
    def family(self, value):
        capi.rtnl_link_set_family(self._rtnl_link, value)

    @property
    @netlink.nlattr(type=str, fmt=util.addr)
    def address(self):
        """Hardware address (MAC address)"""
        a = capi.rtnl_link_get_addr(self._rtnl_link)
        return netlink.AbstractAddress(a)

    @address.setter
    def address(self, value):
        capi.rtnl_link_set_addr(self._rtnl_link, value._addr)

    @property
    @netlink.nlattr(type=str, fmt=util.addr)
    def broadcast(self):
        """Hardware broadcast address"""
        a = capi.rtnl_link_get_broadcast(self._rtnl_link)
        return netlink.AbstractAddress(a)

    @broadcast.setter
    def broadcast(self, value):
        capi.rtnl_link_set_broadcast(self._rtnl_link, value._addr)

    @property
    @netlink.nlattr(type=str, immutable=True, fmt=util.string)
    def qdisc(self):
        """Name of qdisc (cannot be changed)"""
        return capi.rtnl_link_get_qdisc(self._rtnl_link)

    @qdisc.setter
    def qdisc(self, value):
        capi.rtnl_link_set_qdisc(self._rtnl_link, value)

    @property
    @netlink.nlattr(type=int, fmt=util.num)
    def txqlen(self):
        """Length of transmit queue"""
        return capi.rtnl_link_get_txqlen(self._rtnl_link)

    @txqlen.setter
    def txqlen(self, value):
        capi.rtnl_link_set_txqlen(self._rtnl_link, int(value))

    @property
    @netlink.nlattr(type=str, immutable=True, fmt=util.string)
    def arptype(self):
        """Type of link (cannot be changed)"""
        type_ = capi.rtnl_link_get_arptype(self._rtnl_link)
        return core_capi.nl_llproto2str(type_, 64)[0]

    @arptype.setter
    def arptype(self, value):
        i = core_capi.nl_str2llproto(value)
        capi.rtnl_link_set_arptype(self._rtnl_link, i)

    @property
    @netlink.nlattr(type=str, immutable=True, fmt=util.string, title='state')
    def operstate(self):
        """Operational status"""
        operstate = capi.rtnl_link_get_operstate(self._rtnl_link)
        return capi.rtnl_link_operstate2str(operstate, 32)[0]

    @operstate.setter
    def operstate(self, value):
        i = capi.rtnl_link_str2operstate(value)
        capi.rtnl_link_set_operstate(self._rtnl_link, i)

    @property
    @netlink.nlattr(type=str, immutable=True, fmt=util.string)
    def mode(self):
        """Link mode"""
        mode = capi.rtnl_link_get_linkmode(self._rtnl_link)
        return capi.rtnl_link_mode2str(mode, 32)[0]

    @mode.setter
    def mode(self, value):
        i = capi.rtnl_link_str2mode(value)
        capi.rtnl_link_set_linkmode(self._rtnl_link, i)

    @property
    @netlink.nlattr(type=str, fmt=util.string)
    def alias(self):
        """Interface alias (SNMP)"""
        return capi.rtnl_link_get_ifalias(self._rtnl_link)

    @alias.setter
    def alias(self, value):
        capi.rtnl_link_set_ifalias(self._rtnl_link, value)

    @property
    @netlink.nlattr(type=str, fmt=util.string)
    def type(self):
        """Link type"""
        return capi.rtnl_link_get_type(self._rtnl_link)

    @type.setter
    def type(self, value):
        if capi.rtnl_link_set_type(self._rtnl_link, value) < 0:
            raise NameError('unknown info type')

        self._module_lookup('netlink.route.links.' + value)

    def get_stat(self, stat):
        """Retrieve statistical information"""
        if type(stat) is str:
            stat = capi.rtnl_link_str2stat(stat)
            if stat < 0:
                raise NameError('unknown name of statistic')

        return capi.rtnl_link_get_stat(self._rtnl_link, stat)

    def enslave(self, slave, sock=None):
        if not sock:
            sock = netlink.lookup_socket(netlink.NETLINK_ROUTE)

        return capi.rtnl_link_enslave(sock._sock, self._rtnl_link, slave._rtnl_link)

    def release(self, slave, sock=None):
        if not sock:
            sock = netlink.lookup_socket(netlink.NETLINK_ROUTE)

        return capi.rtnl_link_release(sock._sock, self._rtnl_link, slave._rtnl_link)

    def add(self, sock=None, flags=None):
        if not sock:
            sock = netlink.lookup_socket(netlink.NETLINK_ROUTE)

        if not flags:
            flags = netlink.NLM_F_CREATE

        ret = capi.rtnl_link_add(sock._sock, self._rtnl_link, flags)
        if ret < 0:
            raise netlink.KernelError(ret)

    def change(self, sock=None, flags=0):
        """Commit changes made to the link object"""
        if sock is None:
            sock = netlink.lookup_socket(netlink.NETLINK_ROUTE)

        if not self._orig:
            raise netlink.NetlinkError('Original link not available')
        ret = capi.rtnl_link_change(sock._sock, self._orig, self._rtnl_link, flags)
        if ret < 0:
            raise netlink.KernelError(ret)

    def delete(self, sock=None):
        """Attempt to delete this link in the kernel"""
        if sock is None:
            sock = netlink.lookup_socket(netlink.NETLINK_ROUTE)

        ret = capi.rtnl_link_delete(sock._sock, self._rtnl_link)
        if ret < 0:
            raise netlink.KernelError(ret)

    ###################################################################
    # private properties
    #
    # Used for formatting output. USE AT OWN RISK
    @property
    def _state(self):
        if 'up' in self.flags:
            buf = util.good('up')
            if 'lowerup' not in self.flags:
                buf += ' ' + util.bad('no-carrier')
        else:
            buf = util.bad('down')
        return buf

    @property
    def _brief(self):
        return self._module_brief() + self._foreach_af('brief')

    @property
    def _flags(self):
        ignore = [
            'up',
            'running',
            'lowerup',
        ]
        return ','.join([flag for flag in self.flags if flag not in ignore])

    def _foreach_af(self, name, args=None):
        buf = ''
        for af in self.af:
            try:
                func = getattr(self.af[af], name)
                s = str(func(args))
                if len(s) > 0:
                    buf += ' ' + s
            except AttributeError:
                pass
        return buf

    def format(self, details=False, stats=False, indent=''):
        """Return link as formatted text"""
        fmt = util.MyFormatter(self, indent)

        buf = fmt.format('{a|ifindex} {a|name} {a|arptype} {a|address} '\
                 '{a|_state} <{a|_flags}> {a|_brief}')

        if details:
            buf += fmt.nl('\t{t|mtu} {t|txqlen} {t|weight} '\
                      '{t|qdisc} {t|operstate}')
            buf += fmt.nl('\t{t|broadcast} {t|alias}')

            buf += self._foreach_af('details', fmt)

        if stats:
            l = [['Packets', RX_PACKETS, TX_PACKETS],
                 ['Bytes', RX_BYTES, TX_BYTES],
                 ['Errors', RX_ERRORS, TX_ERRORS],
                 ['Dropped', RX_DROPPED, TX_DROPPED],
                 ['Compressed', RX_COMPRESSED, TX_COMPRESSED],
                 ['FIFO Errors', RX_FIFO_ERR, TX_FIFO_ERR],
                 ['Length Errors', RX_LEN_ERR, None],
                 ['Over Errors', RX_OVER_ERR, None],
                 ['CRC Errors', RX_CRC_ERR, None],
                 ['Frame Errors', RX_FRAME_ERR, None],
                 ['Missed Errors', RX_MISSED_ERR, None],
                 ['Abort Errors', None, TX_ABORT_ERR],
                 ['Carrier Errors', None, TX_CARRIER_ERR],
                 ['Heartbeat Errors', None, TX_HBEAT_ERR],
                 ['Window Errors', None, TX_WIN_ERR],
                 ['Collisions', None, COLLISIONS],
                 ['Multicast', None, MULTICAST],
                 ['', None, None],
                 ['Ipv6:', None, None],
                 ['Packets', IP6_INPKTS, IP6_OUTPKTS],
                 ['Bytes', IP6_INOCTETS, IP6_OUTOCTETS],
                 ['Discards', IP6_INDISCARDS, IP6_OUTDISCARDS],
                 ['Multicast Packets', IP6_INMCASTPKTS, IP6_OUTMCASTPKTS],
                 ['Multicast Bytes', IP6_INMCASTOCTETS, IP6_OUTMCASTOCTETS],
                 ['Broadcast Packets', IP6_INBCASTPKTS, IP6_OUTBCASTPKTS],
                 ['Broadcast Bytes', IP6_INBCASTOCTETS, IP6_OUTBCASTOCTETS],
                 ['Delivers', IP6_INDELIVERS, None],
                 ['Forwarded', None, IP6_OUTFORWDATAGRAMS],
                 ['No Routes', IP6_INNOROUTES, IP6_OUTNOROUTES],
                 ['Header Errors', IP6_INHDRERRORS, None],
                 ['Too Big Errors', IP6_INTOOBIGERRORS, None],
                 ['Address Errors', IP6_INADDRERRORS, None],
                 ['Unknown Protocol', IP6_INUNKNOWNPROTOS, None],
                 ['Truncated Packets', IP6_INTRUNCATEDPKTS, None],
                 ['Reasm Timeouts', IP6_REASMTIMEOUT, None],
                 ['Reasm Requests', IP6_REASMREQDS, None],
                 ['Reasm Failures', IP6_REASMFAILS, None],
                 ['Reasm OK', IP6_REASMOKS, None],
                 ['Frag Created', None, IP6_FRAGCREATES],
                 ['Frag Failures', None, IP6_FRAGFAILS],
                 ['Frag OK', None, IP6_FRAGOKS],
                 ['', None, None],
                 ['ICMPv6:', None, None],
                 ['Messages', ICMP6_INMSGS, ICMP6_OUTMSGS],
                 ['Errors', ICMP6_INERRORS, ICMP6_OUTERRORS]]

            buf += '\n\t%s%s%s%s\n' % (33 * ' ', util.title('RX'),
                           15 * ' ', util.title('TX'))

            for row in l:
                row[0] = util.kw(row[0])
                row[1] = self.get_stat(row[1]) if row[1] else ''
                row[2] = self.get_stat(row[2]) if row[2] else ''
                buf += '\t{0[0]:27} {0[1]:>16} {0[2]:>16}\n'.format(row)

            buf += self._foreach_af('stats')

        return buf

def get(name, sock=None):
    """Lookup Link object directly from kernel"""
    if not name:
        raise ValueError()

    if not sock:
        sock = netlink.lookup_socket(netlink.NETLINK_ROUTE)

    link = capi.get_from_kernel(sock._sock, 0, name)
    if not link:
        return None

    return Link.from_capi(link)

_link_cache = LinkCache()

def resolve(name):
    _link_cache.refill()
    return _link_cache[name]