summaryrefslogtreecommitdiffstats
path: root/python/netlink/route/address.py
blob: 29b80728838b074ad7db7059abbf7167c670d431 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
#
# Copyright (c) 2011 Thomas Graf <tgraf@suug.ch>
#

"""Module providing access to network addresses
"""

from __future__ import absolute_import


__version__ = "1.0"
__all__ = [
	'AddressCache',
	'Address']

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

###########################################################################
# Address Cache
class AddressCache(netlink.Cache):
	"""Cache containing network addresses"""

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

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

	def __getitem__(self, key):
		# Using ifindex=0 here implies that the local address itself
		# is unique, otherwise the first occurence is returned.
                return self.lookup(0, key)

	def lookup(self, ifindex, local):
        	if type(local) is str:
                        local = netlink.AbstractAddress(local)

		addr = capi.rtnl_addr_get(self._nl_cache, ifindex,
					  local._nl_addr)
		if addr is None:
			raise KeyError()

                return Address._from_capi(addr)

	def _new_object(self, obj):
        	return Address(obj)

	def _new_cache(self, cache):
        	return AddressCache(cache=cache)

###########################################################################
# Address Object
class Address(netlink.Object):
	"""Network address"""

	def __init__(self, obj=None):
		netlink.Object.__init__(self, "route/addr", "address", obj)
		self._rtnl_addr = self._obj2type(self._nl_object)

	@classmethod
        def _from_capi(cls, obj):
        	return cls(capi.addr2obj(obj))

	def _obj2type(self, obj):
		return capi.obj2addr(obj)

	def __cmp__(self, other):
		# sort by:
		#    1. network link
		#    2. address family
		#    3. local address (including prefixlen)
		diff = self.ifindex - other.ifindex

		if diff == 0:
                        diff = self.family - other.family
                        if diff == 0:
                                diff = capi.nl_addr_cmp(self.local, other.local)

		return diff

	def _new_instance(self, obj):
		return Address(obj)

	#####################################################################
	# ifindex
        @netlink.nlattr('address.ifindex', type=int, immutable=True,
        		fmt=util.num)
        @property
        def ifindex(self):
                """interface index"""
                return capi.rtnl_addr_get_ifindex(self._rtnl_addr)

	@ifindex.setter
        def ifindex(self, value):
        	link = Link.resolve(value)
                if not link:
                        raise ValueError()

                self.link = link

	#####################################################################
	# link
        @netlink.nlattr('address.link', type=str, fmt=util.string)
        @property
        def link(self):
		link = capi.rtnl_addr_get_link(self._rtnl_addr)
                if not link:
                        return None

                return Link.Link.from_capi(link)

        @link.setter
        def link(self, value):
                if type(value) is str:
                        try:
                                value = Link.resolve(value)
                        except KeyError:
                                raise ValueError()

        	capi.rtnl_addr_set_link(self._rtnl_addr, value._rtnl_link)

                # 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_addr_get_ifindex(self._orig) == 0:
                        capi.rtnl_addr_set_ifindex(self._orig, value.ifindex)

	#####################################################################
	# label
        @netlink.nlattr('address.label', type=str, fmt=util.string)
	@property
        def label(self):
        	"""address label"""
        	return capi.rtnl_addr_get_label(self._rtnl_addr)

	@label.setter
        def label(self, value):
        	capi.rtnl_addr_set_label(self._rtnl_addr, value)

	#####################################################################
	# flags
        @netlink.nlattr('address.flags', type=str, fmt=util.string)
	@property
        def flags(self):
        	"""Flags"""
        	flags = capi.rtnl_addr_get_flags(self._rtnl_addr)
                return capi.rtnl_addr_flags2str(flags, 256)[0].split(',')

	def _set_flag(self, flag):
                if flag[0] == '-':
                        i = capi.rtnl_addr_str2flags(flag[1:])
                        capi.rtnl_addr_unset_flags(self._rtnl_addr, i)
                else:
                        i = capi.rtnl_addr_str2flags(flag[1:])
                        capi.rtnl_addr_set_flags(self._rtnl_addr, i)

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

	#####################################################################
	# family
        @netlink.nlattr('address.family', type=int, immutable=True,
        		fmt=util.num)
	@property
        def family(self):
        	"""Address family"""
                fam = capi.rtnl_addr_get_family(self._rtnl_addr)
                return netlink.AddressFamily(fam)

	@family.setter
        def family(self, value):
        	if not isinstance(value, AddressFamily):
                        value = AddressFamily(value)

        	capi.rtnl_addr_set_family(self._rtnl_addr, int(value))

	#####################################################################
	# scope
        @netlink.nlattr('address.scope', type=int, fmt=util.num)
	@property
        def scope(self):
        	"""Address scope"""
        	scope = capi.rtnl_addr_get_scope(self._rtnl_addr)
        	return capi.rtnl_scope2str(scope, 32)[0]

	@scope.setter
        def scope(self, value):
        	if type(value) is str:
                        value = capi.rtnl_str2scope(value)
        	capi.rtnl_addr_set_scope(self._rtnl_addr, value)

	#####################################################################
	# local address
        @netlink.nlattr('address.local', type=str, immutable=True,
        		fmt=util.addr)
	@property
        def local(self):
        	"""Local address"""
        	a = capi.rtnl_addr_get_local(self._rtnl_addr)
        	return netlink.AbstractAddress(a)

	@local.setter
        def local(self, value):
		a = netlink.AbstractAddress(value)
        	capi.rtnl_addr_set_local(self._rtnl_addr, a._nl_addr)

                # local is immutable but we assume that if _orig does not
                # have a local address specified, it was meant to be given here
                if capi.rtnl_addr_get_local(self._orig) is None:
                        capi.rtnl_addr_set_local(self._orig, a._nl_addr)

	#####################################################################
	# Peer address
        @netlink.nlattr('address.peer', type=str, fmt=util.addr)
	@property
        def peer(self):
        	"""Peer address"""
        	a = capi.rtnl_addr_get_peer(self._rtnl_addr)
        	return netlink.AbstractAddress(a)

	@peer.setter
        def peer(self, value):
		a = netlink.AbstractAddress(value)
        	capi.rtnl_addr_set_peer(self._rtnl_addr, a._nl_addr)

	#####################################################################
	# Broadcast address
        @netlink.nlattr('address.broadcast', type=str, fmt=util.addr)
	@property
        def broadcast(self):
        	"""Broadcast address"""
        	a = capi.rtnl_addr_get_broadcast(self._rtnl_addr)
        	return netlink.AbstractAddress(a)

	@broadcast.setter
        def broadcast(self, value):
		a = netlink.AbstractAddress(value)
        	capi.rtnl_addr_set_broadcast(self._rtnl_addr, a._nl_addr)

	#####################################################################
	# Multicast address
        @netlink.nlattr('address.multicast', type=str, fmt=util.addr)
	@property
        def multicast(self):
        	"""multicast address"""
        	a = capi.rtnl_addr_get_multicast(self._rtnl_addr)
        	return netlink.AbstractAddress(a)

	@multicast.setter
        def multicast(self, value):
        	try:
                        a = netlink.AbstractAddress(value)
                except ValueError as err:
                	raise AttributeError('multicast', err)

        	capi.rtnl_addr_set_multicast(self._rtnl_addr, a._nl_addr)

	#####################################################################
	# Anycast address
        @netlink.nlattr('address.anycast', type=str, fmt=util.addr)
	@property
        def anycast(self):
        	"""anycast address"""
        	a = capi.rtnl_addr_get_anycast(self._rtnl_addr)
        	return netlink.AbstractAddress(a)

	@anycast.setter
        def anycast(self, value):
		a = netlink.AbstractAddress(value)
        	capi.rtnl_addr_set_anycast(self._rtnl_addr, a._nl_addr)

	#####################################################################
	# Valid lifetime
        @netlink.nlattr('address.valid_lifetime', type=int, immutable=True,
        		fmt=util.num)
	@property
        def valid_lifetime(self):
        	"""Valid lifetime"""
        	msecs = capi.rtnl_addr_get_valid_lifetime(self._rtnl_addr)
		if msecs == 0xFFFFFFFF:
			return None
		else:
			return datetime.timedelta(seconds=msecs)

	@valid_lifetime.setter
        def valid_lifetime(self, value):
        	capi.rtnl_addr_set_valid_lifetime(self._rtnl_addr, int(value))

	#####################################################################
	# Preferred lifetime
        @netlink.nlattr('address.preferred_lifetime', type=int,
        		immutable=True, fmt=util.num)
	@property
        def preferred_lifetime(self):
        	"""Preferred lifetime"""
        	msecs = capi.rtnl_addr_get_preferred_lifetime(self._rtnl_addr)
		if msecs == 0xFFFFFFFF:
			return None
		else:
			return datetime.timedelta(seconds=msecs)

	@preferred_lifetime.setter
        def preferred_lifetime(self, value):
        	capi.rtnl_addr_set_preferred_lifetime(self._rtnl_addr, int(value))

	#####################################################################
	# Creation Time
        @netlink.nlattr('address.create_time', type=int, immutable=True,
        		fmt=util.num)
	@property
        def create_time(self):
        	"""Creation time"""
        	hsec = capi.rtnl_addr_get_create_time(self._rtnl_addr)
		return datetime.timedelta(milliseconds=10*hsec)

	#####################################################################
	# Last Update
        @netlink.nlattr('address.last_update', type=int, immutable=True,
        		fmt=util.num)
	@property
        def last_update(self):
        	"""Last update"""
        	hsec = capi.rtnl_addr_get_last_update_time(self._rtnl_addr)
		return datetime.timedelta(milliseconds=10*hsec)

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

        	if not flags:
			flags = netlink.NLM_F_CREATE

		ret = capi.rtnl_addr_add(socket._sock, self._rtnl_addr, flags)
		if ret < 0:
			raise netlink.KernelError(ret)

	#####################################################################
	# delete()
	def delete(self, socket, flags=0):
		"""Attempt to delete this address in the kernel"""
		ret = capi.rtnl_addr_delete(socket._sock, self._rtnl_addr, flags)
                if ret < 0:
                        raise netlink.KernelError(ret)

	###################################################################
	# private properties
	#
	# Used for formatting output. USE AT OWN RISK
	@property
	def _flags(self):
		return ','.join(self.flags)

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

		buf = fmt.format('{a|local!b}')

		if not nodev:
			buf += fmt.format(' {a|ifindex}')
		
		buf += fmt.format(' {a|scope}')

		if self.label:
			buf += fmt.format(' "{a|label}"')
		
		buf += fmt.format(' <{a|_flags}>')

		if details:
			buf += fmt.nl('\t{t|broadcast} {t|multicast}') \
			     + fmt.nl('\t{t|peer} {t|anycast}')

			if self.valid_lifetime:
				buf += fmt.nl('\t{s|valid-lifetime!k} '\
				       '{a|valid_lifetime}')

			if self.preferred_lifetime:
				buf += fmt.nl('\t{s|preferred-lifetime!k} '\
				       '{a|preferred_lifetime}')

		if stats and (self.create_time or self.last_update):
			buf += self.nl('\t{s|created!k} {a|create_time}'\
			       ' {s|last-updated!k} {a|last_update}')

		return buf