xref: /aosp_15_r20/external/libnl/python/netlink/route/tc.py (revision 4dc78e53d49367fa8e61b07018507c90983a077d)

#
# Copyright (c) 2011 Thomas Graf <[email protected]>
#
from __future__ import absolute_import

__all__ = [
    "TcCache",
    "Tc",
    "QdiscCache",
    "Qdisc",
    "TcClassCache",
    "TcClass",
]

from .. import core as netlink
from . import capi as capi
from .. import util as util
from . import link as Link

TC_PACKETS = 0
TC_BYTES = 1
TC_RATE_BPS = 2
TC_RATE_PPS = 3
TC_QLEN = 4
TC_BACKLOG = 5
TC_DROPS = 6
TC_REQUEUES = 7
TC_OVERLIMITS = 9

TC_H_ROOT = 0xFFFFFFFF
TC_H_INGRESS = 0xFFFFFFF1

STAT_PACKETS = 0
STAT_BYTES = 1
STAT_RATE_BPS = 2
STAT_RATE_PPS = 3
STAT_QLEN = 4
STAT_BACKLOG = 5
STAT_DROPS = 6
STAT_REQUEUES = 7
STAT_OVERLIMITS = 8
STAT_MAX = STAT_OVERLIMITS


class Handle(object):
    """Traffic control handle

    Representation of a traffic control handle which uniquely identifies
    each traffic control object in its link namespace.

    handle = tc.Handle('10:20')
    handle = tc.handle('root')
    print int(handle)
    print str(handle)
    """

    def __init__(self, val=None):
        if type(val) is str:
            val = capi.tc_str2handle(val)
        elif not val:
            val = 0

        self._val = int(val)

    def __cmp__(self, other):
        if other is None:
            other = 0

        if isinstance(other, Handle):
            return int(self) - int(other)
        elif isinstance(other, int):
            return int(self) - other
        else:
            raise TypeError()

    def __int__(self):
        return self._val

    def __str__(self):
        return capi.rtnl_tc_handle2str(self._val, 64)[0]

    def isroot(self):
        return self._val == TC_H_ROOT or self._val == TC_H_INGRESS


class TcCache(netlink.Cache):
    """Cache of traffic control object"""

    def __getitem__(self, key):
        raise NotImplementedError()


class Tc(netlink.Object):
    def __cmp__(self, other):
        diff = self.ifindex - other.ifindex
        if diff == 0:
            diff = int(self.handle) - int(other.handle)
        return diff

    def _tc_module_lookup(self):
        self._module_lookup(self._module_path + self.kind, "init_" + self._name)

    @property
    def root(self):
        """True if tc object is a root object"""
        return self.parent.isroot()

    @property
    def ifindex(self):
        """interface index"""
        return capi.rtnl_tc_get_ifindex(self._rtnl_tc)

    @ifindex.setter
    def ifindex(self, value):
        capi.rtnl_tc_set_ifindex(self._rtnl_tc, int(value))

    @property
    def link(self):
        link = capi.rtnl_tc_get_link(self._rtnl_tc)
        if not link:
            return None

        return Link.Link.from_capi(link)

    @link.setter
    def link(self, value):
        capi.rtnl_tc_set_link(self._rtnl_tc, value._link)

    @property
    def mtu(self):
        return capi.rtnl_tc_get_mtu(self._rtnl_tc)

    @mtu.setter
    def mtu(self, value):
        capi.rtnl_tc_set_mtu(self._rtnl_tc, int(value))

    @property
    def mpu(self):
        return capi.rtnl_tc_get_mpu(self._rtnl_tc)

    @mpu.setter
    def mpu(self, value):
        capi.rtnl_tc_set_mpu(self._rtnl_tc, int(value))

    @property
    def overhead(self):
        return capi.rtnl_tc_get_overhead(self._rtnl_tc)

    @overhead.setter
    def overhead(self, value):
        capi.rtnl_tc_set_overhead(self._rtnl_tc, int(value))

    @property
    def linktype(self):
        return capi.rtnl_tc_get_linktype(self._rtnl_tc)

    @linktype.setter
    def linktype(self, value):
        capi.rtnl_tc_set_linktype(self._rtnl_tc, int(value))

    @property
    @netlink.nlattr(fmt=util.handle)
    def handle(self):
        return Handle(capi.rtnl_tc_get_handle(self._rtnl_tc))

    @handle.setter
    def handle(self, value):
        capi.rtnl_tc_set_handle(self._rtnl_tc, int(value))

    @property
    @netlink.nlattr(fmt=util.handle)
    def parent(self):
        return Handle(capi.rtnl_tc_get_parent(self._rtnl_tc))

    @parent.setter
    def parent(self, value):
        capi.rtnl_tc_set_parent(self._rtnl_tc, int(value))

    @property
    @netlink.nlattr(fmt=util.bold)
    def kind(self):
        return capi.rtnl_tc_get_kind(self._rtnl_tc)

    @kind.setter
    def kind(self, value):
        capi.rtnl_tc_set_kind(self._rtnl_tc, value)
        self._tc_module_lookup()

    def get_stat(self, id):
        return capi.rtnl_tc_get_stat(self._rtnl_tc, id)

    @property
    def _dev(self):
        buf = util.kw("dev") + " "

        if self.link:
            return buf + util.string(self.link.name)
        else:
            return buf + util.num(self.ifindex)

    def brief(self, title, nodev=False, noparent=False):
        ret = title + " {a|kind} {a|handle}"

        if not nodev:
            ret += " {a|_dev}"

        if not noparent:
            ret += " {t|parent}"

        return ret + self._module_brief()

    @staticmethod
    def details():
        return "{t|mtu} {t|mpu} {t|overhead} {t|linktype}"

    @property
    def packets(self):
        return self.get_stat(STAT_PACKETS)

    @property
    def bytes(self):
        return self.get_stat(STAT_BYTES)

    @property
    def qlen(self):
        return self.get_stat(STAT_QLEN)

    @staticmethod
    def stats(fmt):
        return fmt.nl("{t|packets} {t|bytes} {t|qlen}")


class QdiscCache(netlink.Cache):
    """Cache of qdiscs"""

    def __init__(self, cache=None):
        if not cache:
            cache = self._alloc_cache_name("route/qdisc")

        self._protocol = netlink.NETLINK_ROUTE
        self._nl_cache = cache

    # 	def __getitem__(self, key):
    #        	if type(key) is int:
    #                        link = capi.rtnl_link_get(self._this, key)
    #                elif type(key) is str:
    #                        link = capi.rtnl_link_get_by_name(self._this, key)
    #
    # 		if qdisc is None:
    #                        raise KeyError()
    # 		else:
    #                        return Qdisc._from_capi(capi.qdisc2obj(qdisc))

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

    @staticmethod
    def _new_cache(cache):
        return QdiscCache(cache=cache)


class Qdisc(Tc):
    """Queueing discipline"""

    def __init__(self, obj=None):
        netlink.Object.__init__(self, "route/qdisc", "qdisc", obj)
        self._module_path = "netlink.route.qdisc."
        self._rtnl_qdisc = self._obj2type(self._nl_object)
        self._rtnl_tc = capi.obj2tc(self._nl_object)

        if self.kind:
            self._tc_module_lookup()

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

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

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

        return Qdisc(obj)

    @property
    def childs(self):
        ret = []

        if int(self.handle):
            ret += get_cls(self.ifindex, parent=self.handle)

            if self.root:
                ret += get_class(self.ifindex, parent=TC_H_ROOT)

            ret += get_class(self.ifindex, parent=self.handle)

        return ret

    # 	def add(self, socket, flags=None):
    #        	if not flags:
    #                        flags = netlink.NLM_F_CREATE
    #
    # 		ret = capi.rtnl_link_add(socket._sock, self._link, flags)
    # 		if ret < 0:
    # 			raise netlink.KernelError(ret)
    #
    # 	def change(self, socket, flags=0):
    # 		"""Commit changes made to the link object"""
    # 		if not self._orig:
    # 			raise NetlinkError('Original link not available')
    #        	ret = capi.rtnl_link_change(socket._sock, self._orig, self._link, flags)
    #                if ret < 0:
    #                        raise netlink.KernelError(ret)
    #
    # 	def delete(self, socket):
    # 		"""Attempt to delete this link in the kernel"""
    #        	ret = capi.rtnl_link_delete(socket._sock, self._link)
    #                if ret < 0:
    #                        raise netlink.KernelError(ret)

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

        buf = fmt.format(self.brief("qdisc", nodev, noparent))

        if details:
            buf += fmt.nl("\t" + self.details())

        if stats:
            buf += self.stats(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:27} {1:>16} {2:>16}\n'.format(*row)

        return buf


class TcClassCache(netlink.Cache):
    """Cache of traffic classes"""

    def __init__(self, ifindex, cache=None):
        if not cache:
            cache = self._alloc_cache_name("route/class")

        self._protocol = netlink.NETLINK_ROUTE
        self._nl_cache = cache
        self._set_arg1(ifindex)

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

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


class TcClass(Tc):
    """Traffic Class"""

    def __init__(self, obj=None):
        netlink.Object.__init__(self, "route/class", "class", obj)
        self._module_path = "netlink.route.qdisc."
        self._rtnl_class = self._obj2type(self._nl_object)
        self._rtnl_tc = capi.obj2tc(self._nl_object)

        if self.kind:
            self._tc_module_lookup()

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

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

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

        return TcClass(obj)

    @property
    def childs(self):
        ret = []

        # classes can have classifiers, child classes and leaf
        # qdiscs
        ret += get_cls(self.ifindex, parent=self.handle)
        ret += get_class(self.ifindex, parent=self.handle)
        ret += get_qdisc(self.ifindex, parent=self.handle)

        return ret

    def format(
        self, details=False, _stats=False, nodev=False, noparent=False, indent=""
    ):
        """Return class as formatted text"""
        fmt = util.MyFormatter(self, indent)

        buf = fmt.format(self.brief("class", nodev, noparent))

        if details:
            buf += fmt.nl("\t" + self.details())

        return buf


class ClassifierCache(netlink.Cache):
    """Cache of traffic classifiers objects"""

    def __init__(self, ifindex, parent, cache=None):
        if not cache:
            cache = self._alloc_cache_name("route/cls")

        self._protocol = netlink.NETLINK_ROUTE
        self._nl_cache = cache
        self._set_arg1(ifindex)
        self._set_arg2(int(parent))

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

    def _new_cache(self, cache):
        return ClassifierCache(self.arg1, self.arg2, cache=cache)


class Classifier(Tc):
    """Classifier"""

    def __init__(self, obj=None):
        netlink.Object.__init__(self, "route/cls", "cls", obj)
        self._module_path = "netlink.route.cls."
        self._rtnl_cls = self._obj2type(self._nl_object)
        self._rtnl_tc = capi.obj2tc(self._nl_object)

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

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

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

        return Classifier(obj)

    @property
    def priority(self):
        return capi.rtnl_cls_get_prio(self._rtnl_cls)

    @priority.setter
    def priority(self, value):
        capi.rtnl_cls_set_prio(self._rtnl_cls, int(value))

    @property
    def protocol(self):
        return capi.rtnl_cls_get_protocol(self._rtnl_cls)

    @protocol.setter
    def protocol(self, value):
        capi.rtnl_cls_set_protocol(self._rtnl_cls, int(value))

    @property
    def childs(self):
        return []

    def format(
        self, details=False, _stats=False, nodev=False, noparent=False, indent=""
    ):
        """Return class as formatted text"""
        fmt = util.MyFormatter(self, indent)

        buf = fmt.format(self.brief("classifier", nodev, noparent))
        buf += fmt.format(" {t|priority} {t|protocol}")

        if details:
            buf += fmt.nl("\t" + self.details())

        return buf


_qdisc_cache = QdiscCache()


def get_qdisc(ifindex, handle=None, parent=None):
    lst = []

    _qdisc_cache.refill()

    for qdisc in _qdisc_cache:
        if qdisc.ifindex != ifindex:
            continue
        if (handle is not None) and (qdisc.handle != handle):
            continue
        if (parent is not None) and (qdisc.parent != parent):
            continue
        lst.append(qdisc)

    return lst


_class_cache = {}


def get_class(ifindex, parent, handle=None):
    lst = []

    try:
        cache = _class_cache[ifindex]
    except KeyError:
        cache = TcClassCache(ifindex)
        _class_cache[ifindex] = cache

    cache.refill()

    for cl in cache:
        if (parent is not None) and (cl.parent != parent):
            continue
        if (handle is not None) and (cl.handle != handle):
            continue
        lst.append(cl)

    return lst


_cls_cache = {}


def get_cls(ifindex, parent, handle=None):

    chain = _cls_cache.get(ifindex, {})

    try:
        cache = chain[parent]
    except KeyError:
        cache = ClassifierCache(ifindex, parent)
        chain[parent] = cache

    cache.refill()

    if handle is None:
        return list(cache)

    return [cls for cls in cache if cls.handle == handle]