import sys
import socket
-__all__ = ['Message', 'Socket', 'DumpParams', 'Object', 'Cache', 'KernelError',
- 'NetlinkError']
-__version__ = "0.1"
+__all__ = [
+ 'Message',
+ 'Socket',
+ 'DumpParams',
+ 'Object',
+ 'Cache',
+ 'KernelError',
+ 'NetlinkError',
+]
+
+__version__ = '0.1'
# netlink protocols
NETLINK_ROUTE = 0
class KernelError(NetlinkError):
def __str__(self):
- return "Kernel returned: {0}".format(self._msg)
+ return 'Kernel returned: {0}'.format(self._msg)
class ImmutableError(NetlinkError):
def __init__(self, msg):
self._msg = msg
def __str__(self):
- return "Immutable attribute: {0}".format(self._msg)
+ return 'Immutable attribute: {0}'.format(self._msg)
class Message(object):
"""Netlink message"""
self._msg = capi.nlmsg_alloc_size(size)
if self._msg is None:
- raise Exception("Message allocation returned NULL")
+ raise Exception('Message allocation returned NULL')
def __del__(self):
capi.nlmsg_free(self._msg)
self._sock = capi.nl_socket_alloc_cb(cb)
if self._sock is None:
- raise Exception("NULL pointer returned while allocating socket")
+ raise Exception('NULL pointer returned while allocating socket')
def __del__(self):
capi.nl_socket_free(self._sock)
def __str__(self):
- return "nlsock<{0}>".format(self.localPort)
+ return 'nlsock<{0}>'.format(self.localPort)
@property
def local_port(self):
def sendto(self, buf):
ret = capi.nl_sendto(self._sock, buf, len(buf))
if ret < 0:
- raise Exception("Failed to send")
+ raise Exception('Failed to send')
else:
return ret
def __init__(self, type=NL_DUMP_LINE):
self._dp = capi.alloc_dump_params()
if not self._dp:
- raise Exception("Unable to allocate struct nl_dump_params")
+ raise Exception('Unable to allocate struct nl_dump_params')
self._dp.dp_type = type
def apply(self, attr, val):
try:
- d = attrs[self._name + "." + attr]
+ d = attrs[self._name + '.' + attr]
except KeyError:
- raise KeyError("Unknown " + self._name +
- " attribute: " + attr)
+ raise KeyError('Unknown ' + self._name +
+ ' attribute: ' + attr)
if 'immutable' in d:
raise ImmutableError(attr)
if not self._hasattr(attr):
- raise KeyError("Invalid " + self._name +
- " attribute: " + attr)
+ raise KeyError('Invalid ' + self._name +
+ ' attribute: ' + attr)
self._setattr(attr, val)
class ObjIterator(object):
if self._nl_addr:
return capi.nl_addr2str(self._nl_addr, 64)[0]
else:
- return "none"
+ return 'none'
@property
def shared(self):
from __future__ import absolute_import
-__version__ = "1.0"
+__version__ = '1.0'
__all__ = [
'AddressCache',
'Address']
def __init__(self, cache=None):
if not cache:
- cache = self._alloc_cache_name("route/addr")
+ cache = self._alloc_cache_name('route/addr')
self._protocol = netlink.NETLINK_ROUTE
self._nl_cache = cache
"""Network address"""
def __init__(self, obj=None):
- netlink.Object.__init__(self, "route/addr", "address", obj)
+ netlink.Object.__init__(self, 'route/addr', 'address', obj)
self._rtnl_addr = self._obj2type(self._nl_object)
@classmethod
from __future__ import absolute_import
-__version__ = "0.1"
+__version__ = '0.1'
__all__ = [
'LinkCache',
'Link',
- 'get_from_kernel']
+ 'get_from_kernel',
+]
import socket
import sys
def __init__(self, family=socket.AF_UNSPEC, cache=None):
if not cache:
- cache = self._alloc_cache_name("route/link")
+ cache = self._alloc_cache_name('route/link')
self._info_module = None
self._protocol = netlink.NETLINK_ROUTE
"""Network link"""
def __init__(self, obj=None):
- netlink.Object.__init__(self, "route/link", "link", obj)
+ netlink.Object.__init__(self, 'route/link', 'link', obj)
self._rtnl_link = self._obj2type(self._nl_object)
if self.type:
@type.setter
def type(self, value):
if capi.rtnl_link_set_type(self._rtnl_link, value) < 0:
- raise NameError("unknown info type")
+ raise NameError('unknown info type')
self._module_lookup('netlink.route.links.' + value)
if type(stat) is str:
stat = capi.rtnl_link_str2stat(stat)
if stat < 0:
- raise NameError("unknown name of statistic")
+ raise NameError('unknown name of statistic')
return capi.rtnl_link_get_stat(self._rtnl_link, stat)
socket = netlink.lookup_socket(netlink.NETLINK_ROUTE)
if not self._orig:
- raise NetlinkError("Original link not available")
+ raise NetlinkError('Original link not available')
ret = capi.rtnl_link_change(socket._sock, self._orig, self._rtnl_link, flags)
if ret < 0:
raise netlink.KernelError(ret)
@property
def _flags(self):
- ignore = ['up', 'running', 'lowerup']
+ ignore = [
+ 'up',
+ 'running',
+ 'lowerup',
+ ]
return ','.join([flag for flag in self.flags if flag not in ignore])
def _foreach_af(self, name, args=None):
"""
from __future__ import absolute_import
-__version__ = "1.0"
-__all__ = ['init']
+__version__ = '1.0'
+__all__ = [
+ 'init',
+]
from ... import core as netlink
from __future__ import absolute_import
-__all__ = ['']
+__all__ = [
+ '',
+]
from ... import core as netlink
from .. import capi as capi
if type(id) is str:
id = capi.rtnl_link_inet_str2devconf(id)[0]
if id < 0:
- raise NameError("unknown configuration id")
+ raise NameError('unknown configuration id')
return id
class InetLink(object):
'QdiscCache',
'Qdisc',
'TcClassCache',
- 'TcClass']
+ 'TcClass',
+]
import socket
import sys
def __init__(self, cache=None):
if not cache:
- cache = self._alloc_cache_name("route/qdisc")
+ cache = self._alloc_cache_name('route/qdisc')
self._protocol = netlink.NETLINK_ROUTE
self._nl_cache = cache
"""Queueing discipline"""
def __init__(self, obj=None):
- netlink.Object.__init__(self, "route/qdisc", "qdisc", obj)
+ 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)
# def change(self, socket, flags=0):
# """Commit changes made to the link object"""
# if not self._orig:
-# raise NetlinkError("Original link not available")
+# 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 __init__(self, ifindex, cache=None):
if not cache:
- cache = self._alloc_cache_name("route/class")
+ cache = self._alloc_cache_name('route/class')
self._protocol = netlink.NETLINK_ROUTE
self._nl_cache = cache
"""Traffic Class"""
def __init__(self, obj=None):
- netlink.Object.__init__(self, "route/class", "class", obj)
+ 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)
def __init__(self, ifindex, parent, cache=None):
if not cache:
- cache = self._alloc_cache_name("route/cls")
+ cache = self._alloc_cache_name('route/cls')
self._protocol = netlink.NETLINK_ROUTE
self._nl_cache = cache
"""Classifier"""
def __init__(self, obj=None):
- netlink.Object.__init__(self, "route/cls", "cls", obj)
+ 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)
from string import Formatter
import types
-__version__ = "1.0"
+__version__ = '1.0'
def _color(t, c):
return b'{esc}[{color}m{text}{esc}[0m'.format(esc=b'\x1b', color=c, text=t)
elif conversion is None:
return value
- raise ValueError("Unknown converion specifier {0!s}".format(conversion))
+ raise ValueError('Unknown converion specifier {0!s}'.format(conversion))
def nl(self, format_string=''):
return '\n' + self._indent + self.format(format_string)