Mercurial > kallithea
comparison rhodecode/lib/ipaddr.py @ 3125:9b92cf5a0cca beta
Added UserIpMap interface for allowed IP addresses and IP restriction access
ref #264 IP restriction for users and user groups
author | Marcin Kuzminski <marcin@python-works.com> |
---|---|
date | Sun, 30 Dec 2012 23:06:03 +0100 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
3124:6659c5af04e7 | 3125:9b92cf5a0cca |
---|---|
1 # Copyright 2007 Google Inc. | |
2 # Licensed to PSF under a Contributor Agreement. | |
3 # | |
4 # Licensed under the Apache License, Version 2.0 (the "License"); | |
5 # you may not use this file except in compliance with the License. | |
6 # You may obtain a copy of the License at | |
7 # | |
8 # http://www.apache.org/licenses/LICENSE-2.0 | |
9 # | |
10 # Unless required by applicable law or agreed to in writing, software | |
11 # distributed under the License is distributed on an "AS IS" BASIS, | |
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or | |
13 # implied. See the License for the specific language governing | |
14 # permissions and limitations under the License. | |
15 | |
16 """A fast, lightweight IPv4/IPv6 manipulation library in Python. | |
17 | |
18 This library is used to create/poke/manipulate IPv4 and IPv6 addresses | |
19 and networks. | |
20 | |
21 """ | |
22 | |
23 __version__ = 'trunk' | |
24 | |
25 import struct | |
26 | |
27 IPV4LENGTH = 32 | |
28 IPV6LENGTH = 128 | |
29 | |
30 | |
31 class AddressValueError(ValueError): | |
32 """A Value Error related to the address.""" | |
33 | |
34 | |
35 class NetmaskValueError(ValueError): | |
36 """A Value Error related to the netmask.""" | |
37 | |
38 | |
39 def IPAddress(address, version=None): | |
40 """Take an IP string/int and return an object of the correct type. | |
41 | |
42 Args: | |
43 address: A string or integer, the IP address. Either IPv4 or | |
44 IPv6 addresses may be supplied; integers less than 2**32 will | |
45 be considered to be IPv4 by default. | |
46 version: An Integer, 4 or 6. If set, don't try to automatically | |
47 determine what the IP address type is. important for things | |
48 like IPAddress(1), which could be IPv4, '0.0.0.1', or IPv6, | |
49 '::1'. | |
50 | |
51 Returns: | |
52 An IPv4Address or IPv6Address object. | |
53 | |
54 Raises: | |
55 ValueError: if the string passed isn't either a v4 or a v6 | |
56 address. | |
57 | |
58 """ | |
59 if version: | |
60 if version == 4: | |
61 return IPv4Address(address) | |
62 elif version == 6: | |
63 return IPv6Address(address) | |
64 | |
65 try: | |
66 return IPv4Address(address) | |
67 except (AddressValueError, NetmaskValueError): | |
68 pass | |
69 | |
70 try: | |
71 return IPv6Address(address) | |
72 except (AddressValueError, NetmaskValueError): | |
73 pass | |
74 | |
75 raise ValueError('%r does not appear to be an IPv4 or IPv6 address' % | |
76 address) | |
77 | |
78 | |
79 def IPNetwork(address, version=None, strict=False): | |
80 """Take an IP string/int and return an object of the correct type. | |
81 | |
82 Args: | |
83 address: A string or integer, the IP address. Either IPv4 or | |
84 IPv6 addresses may be supplied; integers less than 2**32 will | |
85 be considered to be IPv4 by default. | |
86 version: An Integer, if set, don't try to automatically | |
87 determine what the IP address type is. important for things | |
88 like IPNetwork(1), which could be IPv4, '0.0.0.1/32', or IPv6, | |
89 '::1/128'. | |
90 | |
91 Returns: | |
92 An IPv4Network or IPv6Network object. | |
93 | |
94 Raises: | |
95 ValueError: if the string passed isn't either a v4 or a v6 | |
96 address. Or if a strict network was requested and a strict | |
97 network wasn't given. | |
98 | |
99 """ | |
100 if version: | |
101 if version == 4: | |
102 return IPv4Network(address, strict) | |
103 elif version == 6: | |
104 return IPv6Network(address, strict) | |
105 | |
106 try: | |
107 return IPv4Network(address, strict) | |
108 except (AddressValueError, NetmaskValueError): | |
109 pass | |
110 | |
111 try: | |
112 return IPv6Network(address, strict) | |
113 except (AddressValueError, NetmaskValueError): | |
114 pass | |
115 | |
116 raise ValueError('%r does not appear to be an IPv4 or IPv6 network' % | |
117 address) | |
118 | |
119 | |
120 def v4_int_to_packed(address): | |
121 """The binary representation of this address. | |
122 | |
123 Args: | |
124 address: An integer representation of an IPv4 IP address. | |
125 | |
126 Returns: | |
127 The binary representation of this address. | |
128 | |
129 Raises: | |
130 ValueError: If the integer is too large to be an IPv4 IP | |
131 address. | |
132 """ | |
133 if address > _BaseV4._ALL_ONES: | |
134 raise ValueError('Address too large for IPv4') | |
135 return Bytes(struct.pack('!I', address)) | |
136 | |
137 | |
138 def v6_int_to_packed(address): | |
139 """The binary representation of this address. | |
140 | |
141 Args: | |
142 address: An integer representation of an IPv6 IP address. | |
143 | |
144 Returns: | |
145 The binary representation of this address. | |
146 """ | |
147 return Bytes(struct.pack('!QQ', address >> 64, address & (2 ** 64 - 1))) | |
148 | |
149 | |
150 def _find_address_range(addresses): | |
151 """Find a sequence of addresses. | |
152 | |
153 Args: | |
154 addresses: a list of IPv4 or IPv6 addresses. | |
155 | |
156 Returns: | |
157 A tuple containing the first and last IP addresses in the sequence. | |
158 | |
159 """ | |
160 first = last = addresses[0] | |
161 for ip in addresses[1:]: | |
162 if ip._ip == last._ip + 1: | |
163 last = ip | |
164 else: | |
165 break | |
166 return (first, last) | |
167 | |
168 | |
169 def _get_prefix_length(number1, number2, bits): | |
170 """Get the number of leading bits that are same for two numbers. | |
171 | |
172 Args: | |
173 number1: an integer. | |
174 number2: another integer. | |
175 bits: the maximum number of bits to compare. | |
176 | |
177 Returns: | |
178 The number of leading bits that are the same for two numbers. | |
179 | |
180 """ | |
181 for i in range(bits): | |
182 if number1 >> i == number2 >> i: | |
183 return bits - i | |
184 return 0 | |
185 | |
186 | |
187 def _count_righthand_zero_bits(number, bits): | |
188 """Count the number of zero bits on the right hand side. | |
189 | |
190 Args: | |
191 number: an integer. | |
192 bits: maximum number of bits to count. | |
193 | |
194 Returns: | |
195 The number of zero bits on the right hand side of the number. | |
196 | |
197 """ | |
198 if number == 0: | |
199 return bits | |
200 for i in range(bits): | |
201 if (number >> i) % 2: | |
202 return i | |
203 | |
204 | |
205 def summarize_address_range(first, last): | |
206 """Summarize a network range given the first and last IP addresses. | |
207 | |
208 Example: | |
209 >>> summarize_address_range(IPv4Address('1.1.1.0'), | |
210 IPv4Address('1.1.1.130')) | |
211 [IPv4Network('1.1.1.0/25'), IPv4Network('1.1.1.128/31'), | |
212 IPv4Network('1.1.1.130/32')] | |
213 | |
214 Args: | |
215 first: the first IPv4Address or IPv6Address in the range. | |
216 last: the last IPv4Address or IPv6Address in the range. | |
217 | |
218 Returns: | |
219 The address range collapsed to a list of IPv4Network's or | |
220 IPv6Network's. | |
221 | |
222 Raise: | |
223 TypeError: | |
224 If the first and last objects are not IP addresses. | |
225 If the first and last objects are not the same version. | |
226 ValueError: | |
227 If the last object is not greater than the first. | |
228 If the version is not 4 or 6. | |
229 | |
230 """ | |
231 if not (isinstance(first, _BaseIP) and isinstance(last, _BaseIP)): | |
232 raise TypeError('first and last must be IP addresses, not networks') | |
233 if first.version != last.version: | |
234 raise TypeError("%s and %s are not of the same version" % ( | |
235 str(first), str(last))) | |
236 if first > last: | |
237 raise ValueError('last IP address must be greater than first') | |
238 | |
239 networks = [] | |
240 | |
241 if first.version == 4: | |
242 ip = IPv4Network | |
243 elif first.version == 6: | |
244 ip = IPv6Network | |
245 else: | |
246 raise ValueError('unknown IP version') | |
247 | |
248 ip_bits = first._max_prefixlen | |
249 first_int = first._ip | |
250 last_int = last._ip | |
251 while first_int <= last_int: | |
252 nbits = _count_righthand_zero_bits(first_int, ip_bits) | |
253 current = None | |
254 while nbits >= 0: | |
255 addend = 2 ** nbits - 1 | |
256 current = first_int + addend | |
257 nbits -= 1 | |
258 if current <= last_int: | |
259 break | |
260 prefix = _get_prefix_length(first_int, current, ip_bits) | |
261 net = ip('%s/%d' % (str(first), prefix)) | |
262 networks.append(net) | |
263 if current == ip._ALL_ONES: | |
264 break | |
265 first_int = current + 1 | |
266 first = IPAddress(first_int, version=first._version) | |
267 return networks | |
268 | |
269 | |
270 def _collapse_address_list_recursive(addresses): | |
271 """Loops through the addresses, collapsing concurrent netblocks. | |
272 | |
273 Example: | |
274 | |
275 ip1 = IPv4Network('1.1.0.0/24') | |
276 ip2 = IPv4Network('1.1.1.0/24') | |
277 ip3 = IPv4Network('1.1.2.0/24') | |
278 ip4 = IPv4Network('1.1.3.0/24') | |
279 ip5 = IPv4Network('1.1.4.0/24') | |
280 ip6 = IPv4Network('1.1.0.1/22') | |
281 | |
282 _collapse_address_list_recursive([ip1, ip2, ip3, ip4, ip5, ip6]) -> | |
283 [IPv4Network('1.1.0.0/22'), IPv4Network('1.1.4.0/24')] | |
284 | |
285 This shouldn't be called directly; it is called via | |
286 collapse_address_list([]). | |
287 | |
288 Args: | |
289 addresses: A list of IPv4Network's or IPv6Network's | |
290 | |
291 Returns: | |
292 A list of IPv4Network's or IPv6Network's depending on what we were | |
293 passed. | |
294 | |
295 """ | |
296 ret_array = [] | |
297 optimized = False | |
298 | |
299 for cur_addr in addresses: | |
300 if not ret_array: | |
301 ret_array.append(cur_addr) | |
302 continue | |
303 if cur_addr in ret_array[-1]: | |
304 optimized = True | |
305 elif cur_addr == ret_array[-1].supernet().subnet()[1]: | |
306 ret_array.append(ret_array.pop().supernet()) | |
307 optimized = True | |
308 else: | |
309 ret_array.append(cur_addr) | |
310 | |
311 if optimized: | |
312 return _collapse_address_list_recursive(ret_array) | |
313 | |
314 return ret_array | |
315 | |
316 | |
317 def collapse_address_list(addresses): | |
318 """Collapse a list of IP objects. | |
319 | |
320 Example: | |
321 collapse_address_list([IPv4('1.1.0.0/24'), IPv4('1.1.1.0/24')]) -> | |
322 [IPv4('1.1.0.0/23')] | |
323 | |
324 Args: | |
325 addresses: A list of IPv4Network or IPv6Network objects. | |
326 | |
327 Returns: | |
328 A list of IPv4Network or IPv6Network objects depending on what we | |
329 were passed. | |
330 | |
331 Raises: | |
332 TypeError: If passed a list of mixed version objects. | |
333 | |
334 """ | |
335 i = 0 | |
336 addrs = [] | |
337 ips = [] | |
338 nets = [] | |
339 | |
340 # split IP addresses and networks | |
341 for ip in addresses: | |
342 if isinstance(ip, _BaseIP): | |
343 if ips and ips[-1]._version != ip._version: | |
344 raise TypeError("%s and %s are not of the same version" % ( | |
345 str(ip), str(ips[-1]))) | |
346 ips.append(ip) | |
347 elif ip._prefixlen == ip._max_prefixlen: | |
348 if ips and ips[-1]._version != ip._version: | |
349 raise TypeError("%s and %s are not of the same version" % ( | |
350 str(ip), str(ips[-1]))) | |
351 ips.append(ip.ip) | |
352 else: | |
353 if nets and nets[-1]._version != ip._version: | |
354 raise TypeError("%s and %s are not of the same version" % ( | |
355 str(ip), str(nets[-1]))) | |
356 nets.append(ip) | |
357 | |
358 # sort and dedup | |
359 ips = sorted(set(ips)) | |
360 nets = sorted(set(nets)) | |
361 | |
362 while i < len(ips): | |
363 (first, last) = _find_address_range(ips[i:]) | |
364 i = ips.index(last) + 1 | |
365 addrs.extend(summarize_address_range(first, last)) | |
366 | |
367 return _collapse_address_list_recursive(sorted( | |
368 addrs + nets, key=_BaseNet._get_networks_key)) | |
369 | |
370 # backwards compatibility | |
371 CollapseAddrList = collapse_address_list | |
372 | |
373 # We need to distinguish between the string and packed-bytes representations | |
374 # of an IP address. For example, b'0::1' is the IPv4 address 48.58.58.49, | |
375 # while '0::1' is an IPv6 address. | |
376 # | |
377 # In Python 3, the native 'bytes' type already provides this functionality, | |
378 # so we use it directly. For earlier implementations where bytes is not a | |
379 # distinct type, we create a subclass of str to serve as a tag. | |
380 # | |
381 # Usage example (Python 2): | |
382 # ip = ipaddr.IPAddress(ipaddr.Bytes('xxxx')) | |
383 # | |
384 # Usage example (Python 3): | |
385 # ip = ipaddr.IPAddress(b'xxxx') | |
386 try: | |
387 if bytes is str: | |
388 raise TypeError("bytes is not a distinct type") | |
389 Bytes = bytes | |
390 except (NameError, TypeError): | |
391 class Bytes(str): | |
392 def __repr__(self): | |
393 return 'Bytes(%s)' % str.__repr__(self) | |
394 | |
395 | |
396 def get_mixed_type_key(obj): | |
397 """Return a key suitable for sorting between networks and addresses. | |
398 | |
399 Address and Network objects are not sortable by default; they're | |
400 fundamentally different so the expression | |
401 | |
402 IPv4Address('1.1.1.1') <= IPv4Network('1.1.1.1/24') | |
403 | |
404 doesn't make any sense. There are some times however, where you may wish | |
405 to have ipaddr sort these for you anyway. If you need to do this, you | |
406 can use this function as the key= argument to sorted(). | |
407 | |
408 Args: | |
409 obj: either a Network or Address object. | |
410 Returns: | |
411 appropriate key. | |
412 | |
413 """ | |
414 if isinstance(obj, _BaseNet): | |
415 return obj._get_networks_key() | |
416 elif isinstance(obj, _BaseIP): | |
417 return obj._get_address_key() | |
418 return NotImplemented | |
419 | |
420 | |
421 class _IPAddrBase(object): | |
422 | |
423 """The mother class.""" | |
424 | |
425 def __index__(self): | |
426 return self._ip | |
427 | |
428 def __int__(self): | |
429 return self._ip | |
430 | |
431 def __hex__(self): | |
432 return hex(self._ip) | |
433 | |
434 @property | |
435 def exploded(self): | |
436 """Return the longhand version of the IP address as a string.""" | |
437 return self._explode_shorthand_ip_string() | |
438 | |
439 @property | |
440 def compressed(self): | |
441 """Return the shorthand version of the IP address as a string.""" | |
442 return str(self) | |
443 | |
444 | |
445 class _BaseIP(_IPAddrBase): | |
446 | |
447 """A generic IP object. | |
448 | |
449 This IP class contains the version independent methods which are | |
450 used by single IP addresses. | |
451 | |
452 """ | |
453 | |
454 def __eq__(self, other): | |
455 try: | |
456 return (self._ip == other._ip | |
457 and self._version == other._version) | |
458 except AttributeError: | |
459 return NotImplemented | |
460 | |
461 def __ne__(self, other): | |
462 eq = self.__eq__(other) | |
463 if eq is NotImplemented: | |
464 return NotImplemented | |
465 return not eq | |
466 | |
467 def __le__(self, other): | |
468 gt = self.__gt__(other) | |
469 if gt is NotImplemented: | |
470 return NotImplemented | |
471 return not gt | |
472 | |
473 def __ge__(self, other): | |
474 lt = self.__lt__(other) | |
475 if lt is NotImplemented: | |
476 return NotImplemented | |
477 return not lt | |
478 | |
479 def __lt__(self, other): | |
480 if self._version != other._version: | |
481 raise TypeError('%s and %s are not of the same version' % ( | |
482 str(self), str(other))) | |
483 if not isinstance(other, _BaseIP): | |
484 raise TypeError('%s and %s are not of the same type' % ( | |
485 str(self), str(other))) | |
486 if self._ip != other._ip: | |
487 return self._ip < other._ip | |
488 return False | |
489 | |
490 def __gt__(self, other): | |
491 if self._version != other._version: | |
492 raise TypeError('%s and %s are not of the same version' % ( | |
493 str(self), str(other))) | |
494 if not isinstance(other, _BaseIP): | |
495 raise TypeError('%s and %s are not of the same type' % ( | |
496 str(self), str(other))) | |
497 if self._ip != other._ip: | |
498 return self._ip > other._ip | |
499 return False | |
500 | |
501 # Shorthand for Integer addition and subtraction. This is not | |
502 # meant to ever support addition/subtraction of addresses. | |
503 def __add__(self, other): | |
504 if not isinstance(other, int): | |
505 return NotImplemented | |
506 return IPAddress(int(self) + other, version=self._version) | |
507 | |
508 def __sub__(self, other): | |
509 if not isinstance(other, int): | |
510 return NotImplemented | |
511 return IPAddress(int(self) - other, version=self._version) | |
512 | |
513 def __repr__(self): | |
514 return '%s(%r)' % (self.__class__.__name__, str(self)) | |
515 | |
516 def __str__(self): | |
517 return '%s' % self._string_from_ip_int(self._ip) | |
518 | |
519 def __hash__(self): | |
520 return hash(hex(long(self._ip))) | |
521 | |
522 def _get_address_key(self): | |
523 return (self._version, self) | |
524 | |
525 @property | |
526 def version(self): | |
527 raise NotImplementedError('BaseIP has no version') | |
528 | |
529 | |
530 class _BaseNet(_IPAddrBase): | |
531 | |
532 """A generic IP object. | |
533 | |
534 This IP class contains the version independent methods which are | |
535 used by networks. | |
536 | |
537 """ | |
538 | |
539 def __init__(self, address): | |
540 self._cache = {} | |
541 | |
542 def __repr__(self): | |
543 return '%s(%r)' % (self.__class__.__name__, str(self)) | |
544 | |
545 def iterhosts(self): | |
546 """Generate Iterator over usable hosts in a network. | |
547 | |
548 This is like __iter__ except it doesn't return the network | |
549 or broadcast addresses. | |
550 | |
551 """ | |
552 cur = int(self.network) + 1 | |
553 bcast = int(self.broadcast) - 1 | |
554 while cur <= bcast: | |
555 cur += 1 | |
556 yield IPAddress(cur - 1, version=self._version) | |
557 | |
558 def __iter__(self): | |
559 cur = int(self.network) | |
560 bcast = int(self.broadcast) | |
561 while cur <= bcast: | |
562 cur += 1 | |
563 yield IPAddress(cur - 1, version=self._version) | |
564 | |
565 def __getitem__(self, n): | |
566 network = int(self.network) | |
567 broadcast = int(self.broadcast) | |
568 if n >= 0: | |
569 if network + n > broadcast: | |
570 raise IndexError | |
571 return IPAddress(network + n, version=self._version) | |
572 else: | |
573 n += 1 | |
574 if broadcast + n < network: | |
575 raise IndexError | |
576 return IPAddress(broadcast + n, version=self._version) | |
577 | |
578 def __lt__(self, other): | |
579 if self._version != other._version: | |
580 raise TypeError('%s and %s are not of the same version' % ( | |
581 str(self), str(other))) | |
582 if not isinstance(other, _BaseNet): | |
583 raise TypeError('%s and %s are not of the same type' % ( | |
584 str(self), str(other))) | |
585 if self.network != other.network: | |
586 return self.network < other.network | |
587 if self.netmask != other.netmask: | |
588 return self.netmask < other.netmask | |
589 return False | |
590 | |
591 def __gt__(self, other): | |
592 if self._version != other._version: | |
593 raise TypeError('%s and %s are not of the same version' % ( | |
594 str(self), str(other))) | |
595 if not isinstance(other, _BaseNet): | |
596 raise TypeError('%s and %s are not of the same type' % ( | |
597 str(self), str(other))) | |
598 if self.network != other.network: | |
599 return self.network > other.network | |
600 if self.netmask != other.netmask: | |
601 return self.netmask > other.netmask | |
602 return False | |
603 | |
604 def __le__(self, other): | |
605 gt = self.__gt__(other) | |
606 if gt is NotImplemented: | |
607 return NotImplemented | |
608 return not gt | |
609 | |
610 def __ge__(self, other): | |
611 lt = self.__lt__(other) | |
612 if lt is NotImplemented: | |
613 return NotImplemented | |
614 return not lt | |
615 | |
616 def __eq__(self, other): | |
617 try: | |
618 return (self._version == other._version | |
619 and self.network == other.network | |
620 and int(self.netmask) == int(other.netmask)) | |
621 except AttributeError: | |
622 if isinstance(other, _BaseIP): | |
623 return (self._version == other._version | |
624 and self._ip == other._ip) | |
625 | |
626 def __ne__(self, other): | |
627 eq = self.__eq__(other) | |
628 if eq is NotImplemented: | |
629 return NotImplemented | |
630 return not eq | |
631 | |
632 def __str__(self): | |
633 return '%s/%s' % (str(self.ip), | |
634 str(self._prefixlen)) | |
635 | |
636 def __hash__(self): | |
637 return hash(int(self.network) ^ int(self.netmask)) | |
638 | |
639 def __contains__(self, other): | |
640 # always false if one is v4 and the other is v6. | |
641 if self._version != other._version: | |
642 return False | |
643 # dealing with another network. | |
644 if isinstance(other, _BaseNet): | |
645 return (self.network <= other.network and | |
646 self.broadcast >= other.broadcast) | |
647 # dealing with another address | |
648 else: | |
649 return (int(self.network) <= int(other._ip) <= | |
650 int(self.broadcast)) | |
651 | |
652 def overlaps(self, other): | |
653 """Tell if self is partly contained in other.""" | |
654 return self.network in other or self.broadcast in other or ( | |
655 other.network in self or other.broadcast in self) | |
656 | |
657 @property | |
658 def network(self): | |
659 x = self._cache.get('network') | |
660 if x is None: | |
661 x = IPAddress(self._ip & int(self.netmask), version=self._version) | |
662 self._cache['network'] = x | |
663 return x | |
664 | |
665 @property | |
666 def broadcast(self): | |
667 x = self._cache.get('broadcast') | |
668 if x is None: | |
669 x = IPAddress(self._ip | int(self.hostmask), version=self._version) | |
670 self._cache['broadcast'] = x | |
671 return x | |
672 | |
673 @property | |
674 def hostmask(self): | |
675 x = self._cache.get('hostmask') | |
676 if x is None: | |
677 x = IPAddress(int(self.netmask) ^ self._ALL_ONES, | |
678 version=self._version) | |
679 self._cache['hostmask'] = x | |
680 return x | |
681 | |
682 @property | |
683 def with_prefixlen(self): | |
684 return '%s/%d' % (str(self.ip), self._prefixlen) | |
685 | |
686 @property | |
687 def with_netmask(self): | |
688 return '%s/%s' % (str(self.ip), str(self.netmask)) | |
689 | |
690 @property | |
691 def with_hostmask(self): | |
692 return '%s/%s' % (str(self.ip), str(self.hostmask)) | |
693 | |
694 @property | |
695 def numhosts(self): | |
696 """Number of hosts in the current subnet.""" | |
697 return int(self.broadcast) - int(self.network) + 1 | |
698 | |
699 @property | |
700 def version(self): | |
701 raise NotImplementedError('BaseNet has no version') | |
702 | |
703 @property | |
704 def prefixlen(self): | |
705 return self._prefixlen | |
706 | |
707 def address_exclude(self, other): | |
708 """Remove an address from a larger block. | |
709 | |
710 For example: | |
711 | |
712 addr1 = IPNetwork('10.1.1.0/24') | |
713 addr2 = IPNetwork('10.1.1.0/26') | |
714 addr1.address_exclude(addr2) = | |
715 [IPNetwork('10.1.1.64/26'), IPNetwork('10.1.1.128/25')] | |
716 | |
717 or IPv6: | |
718 | |
719 addr1 = IPNetwork('::1/32') | |
720 addr2 = IPNetwork('::1/128') | |
721 addr1.address_exclude(addr2) = [IPNetwork('::0/128'), | |
722 IPNetwork('::2/127'), | |
723 IPNetwork('::4/126'), | |
724 IPNetwork('::8/125'), | |
725 ... | |
726 IPNetwork('0:0:8000::/33')] | |
727 | |
728 Args: | |
729 other: An IPvXNetwork object of the same type. | |
730 | |
731 Returns: | |
732 A sorted list of IPvXNetwork objects addresses which is self | |
733 minus other. | |
734 | |
735 Raises: | |
736 TypeError: If self and other are of difffering address | |
737 versions, or if other is not a network object. | |
738 ValueError: If other is not completely contained by self. | |
739 | |
740 """ | |
741 if not self._version == other._version: | |
742 raise TypeError("%s and %s are not of the same version" % ( | |
743 str(self), str(other))) | |
744 | |
745 if not isinstance(other, _BaseNet): | |
746 raise TypeError("%s is not a network object" % str(other)) | |
747 | |
748 if other not in self: | |
749 raise ValueError('%s not contained in %s' % (str(other), | |
750 str(self))) | |
751 if other == self: | |
752 return [] | |
753 | |
754 ret_addrs = [] | |
755 | |
756 # Make sure we're comparing the network of other. | |
757 other = IPNetwork('%s/%s' % (str(other.network), str(other.prefixlen)), | |
758 version=other._version) | |
759 | |
760 s1, s2 = self.subnet() | |
761 while s1 != other and s2 != other: | |
762 if other in s1: | |
763 ret_addrs.append(s2) | |
764 s1, s2 = s1.subnet() | |
765 elif other in s2: | |
766 ret_addrs.append(s1) | |
767 s1, s2 = s2.subnet() | |
768 else: | |
769 # If we got here, there's a bug somewhere. | |
770 assert True == False, ('Error performing exclusion: ' | |
771 's1: %s s2: %s other: %s' % | |
772 (str(s1), str(s2), str(other))) | |
773 if s1 == other: | |
774 ret_addrs.append(s2) | |
775 elif s2 == other: | |
776 ret_addrs.append(s1) | |
777 else: | |
778 # If we got here, there's a bug somewhere. | |
779 assert True == False, ('Error performing exclusion: ' | |
780 's1: %s s2: %s other: %s' % | |
781 (str(s1), str(s2), str(other))) | |
782 | |
783 return sorted(ret_addrs, key=_BaseNet._get_networks_key) | |
784 | |
785 def compare_networks(self, other): | |
786 """Compare two IP objects. | |
787 | |
788 This is only concerned about the comparison of the integer | |
789 representation of the network addresses. This means that the | |
790 host bits aren't considered at all in this method. If you want | |
791 to compare host bits, you can easily enough do a | |
792 'HostA._ip < HostB._ip' | |
793 | |
794 Args: | |
795 other: An IP object. | |
796 | |
797 Returns: | |
798 If the IP versions of self and other are the same, returns: | |
799 | |
800 -1 if self < other: | |
801 eg: IPv4('1.1.1.0/24') < IPv4('1.1.2.0/24') | |
802 IPv6('1080::200C:417A') < IPv6('1080::200B:417B') | |
803 0 if self == other | |
804 eg: IPv4('1.1.1.1/24') == IPv4('1.1.1.2/24') | |
805 IPv6('1080::200C:417A/96') == IPv6('1080::200C:417B/96') | |
806 1 if self > other | |
807 eg: IPv4('1.1.1.0/24') > IPv4('1.1.0.0/24') | |
808 IPv6('1080::1:200C:417A/112') > | |
809 IPv6('1080::0:200C:417A/112') | |
810 | |
811 If the IP versions of self and other are different, returns: | |
812 | |
813 -1 if self._version < other._version | |
814 eg: IPv4('10.0.0.1/24') < IPv6('::1/128') | |
815 1 if self._version > other._version | |
816 eg: IPv6('::1/128') > IPv4('255.255.255.0/24') | |
817 | |
818 """ | |
819 if self._version < other._version: | |
820 return -1 | |
821 if self._version > other._version: | |
822 return 1 | |
823 # self._version == other._version below here: | |
824 if self.network < other.network: | |
825 return -1 | |
826 if self.network > other.network: | |
827 return 1 | |
828 # self.network == other.network below here: | |
829 if self.netmask < other.netmask: | |
830 return -1 | |
831 if self.netmask > other.netmask: | |
832 return 1 | |
833 # self.network == other.network and self.netmask == other.netmask | |
834 return 0 | |
835 | |
836 def _get_networks_key(self): | |
837 """Network-only key function. | |
838 | |
839 Returns an object that identifies this address' network and | |
840 netmask. This function is a suitable "key" argument for sorted() | |
841 and list.sort(). | |
842 | |
843 """ | |
844 return (self._version, self.network, self.netmask) | |
845 | |
846 def _ip_int_from_prefix(self, prefixlen=None): | |
847 """Turn the prefix length netmask into a int for comparison. | |
848 | |
849 Args: | |
850 prefixlen: An integer, the prefix length. | |
851 | |
852 Returns: | |
853 An integer. | |
854 | |
855 """ | |
856 if not prefixlen and prefixlen != 0: | |
857 prefixlen = self._prefixlen | |
858 return self._ALL_ONES ^ (self._ALL_ONES >> prefixlen) | |
859 | |
860 def _prefix_from_ip_int(self, ip_int, mask=32): | |
861 """Return prefix length from the decimal netmask. | |
862 | |
863 Args: | |
864 ip_int: An integer, the IP address. | |
865 mask: The netmask. Defaults to 32. | |
866 | |
867 Returns: | |
868 An integer, the prefix length. | |
869 | |
870 """ | |
871 while mask: | |
872 if ip_int & 1 == 1: | |
873 break | |
874 ip_int >>= 1 | |
875 mask -= 1 | |
876 | |
877 return mask | |
878 | |
879 def _ip_string_from_prefix(self, prefixlen=None): | |
880 """Turn a prefix length into a dotted decimal string. | |
881 | |
882 Args: | |
883 prefixlen: An integer, the netmask prefix length. | |
884 | |
885 Returns: | |
886 A string, the dotted decimal netmask string. | |
887 | |
888 """ | |
889 if not prefixlen: | |
890 prefixlen = self._prefixlen | |
891 return self._string_from_ip_int(self._ip_int_from_prefix(prefixlen)) | |
892 | |
893 def iter_subnets(self, prefixlen_diff=1, new_prefix=None): | |
894 """The subnets which join to make the current subnet. | |
895 | |
896 In the case that self contains only one IP | |
897 (self._prefixlen == 32 for IPv4 or self._prefixlen == 128 | |
898 for IPv6), return a list with just ourself. | |
899 | |
900 Args: | |
901 prefixlen_diff: An integer, the amount the prefix length | |
902 should be increased by. This should not be set if | |
903 new_prefix is also set. | |
904 new_prefix: The desired new prefix length. This must be a | |
905 larger number (smaller prefix) than the existing prefix. | |
906 This should not be set if prefixlen_diff is also set. | |
907 | |
908 Returns: | |
909 An iterator of IPv(4|6) objects. | |
910 | |
911 Raises: | |
912 ValueError: The prefixlen_diff is too small or too large. | |
913 OR | |
914 prefixlen_diff and new_prefix are both set or new_prefix | |
915 is a smaller number than the current prefix (smaller | |
916 number means a larger network) | |
917 | |
918 """ | |
919 if self._prefixlen == self._max_prefixlen: | |
920 yield self | |
921 return | |
922 | |
923 if new_prefix is not None: | |
924 if new_prefix < self._prefixlen: | |
925 raise ValueError('new prefix must be longer') | |
926 if prefixlen_diff != 1: | |
927 raise ValueError('cannot set prefixlen_diff and new_prefix') | |
928 prefixlen_diff = new_prefix - self._prefixlen | |
929 | |
930 if prefixlen_diff < 0: | |
931 raise ValueError('prefix length diff must be > 0') | |
932 new_prefixlen = self._prefixlen + prefixlen_diff | |
933 | |
934 if not self._is_valid_netmask(str(new_prefixlen)): | |
935 raise ValueError( | |
936 'prefix length diff %d is invalid for netblock %s' % ( | |
937 new_prefixlen, str(self))) | |
938 | |
939 first = IPNetwork('%s/%s' % (str(self.network), | |
940 str(self._prefixlen + prefixlen_diff)), | |
941 version=self._version) | |
942 | |
943 yield first | |
944 current = first | |
945 while True: | |
946 broadcast = current.broadcast | |
947 if broadcast == self.broadcast: | |
948 return | |
949 new_addr = IPAddress(int(broadcast) + 1, version=self._version) | |
950 current = IPNetwork('%s/%s' % (str(new_addr), str(new_prefixlen)), | |
951 version=self._version) | |
952 | |
953 yield current | |
954 | |
955 def masked(self): | |
956 """Return the network object with the host bits masked out.""" | |
957 return IPNetwork('%s/%d' % (self.network, self._prefixlen), | |
958 version=self._version) | |
959 | |
960 def subnet(self, prefixlen_diff=1, new_prefix=None): | |
961 """Return a list of subnets, rather than an iterator.""" | |
962 return list(self.iter_subnets(prefixlen_diff, new_prefix)) | |
963 | |
964 def supernet(self, prefixlen_diff=1, new_prefix=None): | |
965 """The supernet containing the current network. | |
966 | |
967 Args: | |
968 prefixlen_diff: An integer, the amount the prefix length of | |
969 the network should be decreased by. For example, given a | |
970 /24 network and a prefixlen_diff of 3, a supernet with a | |
971 /21 netmask is returned. | |
972 | |
973 Returns: | |
974 An IPv4 network object. | |
975 | |
976 Raises: | |
977 ValueError: If self.prefixlen - prefixlen_diff < 0. I.e., you have a | |
978 negative prefix length. | |
979 OR | |
980 If prefixlen_diff and new_prefix are both set or new_prefix is a | |
981 larger number than the current prefix (larger number means a | |
982 smaller network) | |
983 | |
984 """ | |
985 if self._prefixlen == 0: | |
986 return self | |
987 | |
988 if new_prefix is not None: | |
989 if new_prefix > self._prefixlen: | |
990 raise ValueError('new prefix must be shorter') | |
991 if prefixlen_diff != 1: | |
992 raise ValueError('cannot set prefixlen_diff and new_prefix') | |
993 prefixlen_diff = self._prefixlen - new_prefix | |
994 | |
995 if self.prefixlen - prefixlen_diff < 0: | |
996 raise ValueError( | |
997 'current prefixlen is %d, cannot have a prefixlen_diff of %d' % | |
998 (self.prefixlen, prefixlen_diff)) | |
999 return IPNetwork('%s/%s' % (str(self.network), | |
1000 str(self.prefixlen - prefixlen_diff)), | |
1001 version=self._version) | |
1002 | |
1003 # backwards compatibility | |
1004 Subnet = subnet | |
1005 Supernet = supernet | |
1006 AddressExclude = address_exclude | |
1007 CompareNetworks = compare_networks | |
1008 Contains = __contains__ | |
1009 | |
1010 | |
1011 class _BaseV4(object): | |
1012 | |
1013 """Base IPv4 object. | |
1014 | |
1015 The following methods are used by IPv4 objects in both single IP | |
1016 addresses and networks. | |
1017 | |
1018 """ | |
1019 | |
1020 # Equivalent to 255.255.255.255 or 32 bits of 1's. | |
1021 _ALL_ONES = (2 ** IPV4LENGTH) - 1 | |
1022 _DECIMAL_DIGITS = frozenset('0123456789') | |
1023 | |
1024 def __init__(self, address): | |
1025 self._version = 4 | |
1026 self._max_prefixlen = IPV4LENGTH | |
1027 | |
1028 def _explode_shorthand_ip_string(self): | |
1029 return str(self) | |
1030 | |
1031 def _ip_int_from_string(self, ip_str): | |
1032 """Turn the given IP string into an integer for comparison. | |
1033 | |
1034 Args: | |
1035 ip_str: A string, the IP ip_str. | |
1036 | |
1037 Returns: | |
1038 The IP ip_str as an integer. | |
1039 | |
1040 Raises: | |
1041 AddressValueError: if ip_str isn't a valid IPv4 Address. | |
1042 | |
1043 """ | |
1044 octets = ip_str.split('.') | |
1045 if len(octets) != 4: | |
1046 raise AddressValueError(ip_str) | |
1047 | |
1048 packed_ip = 0 | |
1049 for oc in octets: | |
1050 try: | |
1051 packed_ip = (packed_ip << 8) | self._parse_octet(oc) | |
1052 except ValueError: | |
1053 raise AddressValueError(ip_str) | |
1054 return packed_ip | |
1055 | |
1056 def _parse_octet(self, octet_str): | |
1057 """Convert a decimal octet into an integer. | |
1058 | |
1059 Args: | |
1060 octet_str: A string, the number to parse. | |
1061 | |
1062 Returns: | |
1063 The octet as an integer. | |
1064 | |
1065 Raises: | |
1066 ValueError: if the octet isn't strictly a decimal from [0..255]. | |
1067 | |
1068 """ | |
1069 # Whitelist the characters, since int() allows a lot of bizarre stuff. | |
1070 if not self._DECIMAL_DIGITS.issuperset(octet_str): | |
1071 raise ValueError | |
1072 octet_int = int(octet_str, 10) | |
1073 # Disallow leading zeroes, because no clear standard exists on | |
1074 # whether these should be interpreted as decimal or octal. | |
1075 if octet_int > 255 or (octet_str[0] == '0' and len(octet_str) > 1): | |
1076 raise ValueError | |
1077 return octet_int | |
1078 | |
1079 def _string_from_ip_int(self, ip_int): | |
1080 """Turns a 32-bit integer into dotted decimal notation. | |
1081 | |
1082 Args: | |
1083 ip_int: An integer, the IP address. | |
1084 | |
1085 Returns: | |
1086 The IP address as a string in dotted decimal notation. | |
1087 | |
1088 """ | |
1089 octets = [] | |
1090 for _ in xrange(4): | |
1091 octets.insert(0, str(ip_int & 0xFF)) | |
1092 ip_int >>= 8 | |
1093 return '.'.join(octets) | |
1094 | |
1095 @property | |
1096 def max_prefixlen(self): | |
1097 return self._max_prefixlen | |
1098 | |
1099 @property | |
1100 def packed(self): | |
1101 """The binary representation of this address.""" | |
1102 return v4_int_to_packed(self._ip) | |
1103 | |
1104 @property | |
1105 def version(self): | |
1106 return self._version | |
1107 | |
1108 @property | |
1109 def is_reserved(self): | |
1110 """Test if the address is otherwise IETF reserved. | |
1111 | |
1112 Returns: | |
1113 A boolean, True if the address is within the | |
1114 reserved IPv4 Network range. | |
1115 | |
1116 """ | |
1117 return self in IPv4Network('240.0.0.0/4') | |
1118 | |
1119 @property | |
1120 def is_private(self): | |
1121 """Test if this address is allocated for private networks. | |
1122 | |
1123 Returns: | |
1124 A boolean, True if the address is reserved per RFC 1918. | |
1125 | |
1126 """ | |
1127 return (self in IPv4Network('10.0.0.0/8') or | |
1128 self in IPv4Network('172.16.0.0/12') or | |
1129 self in IPv4Network('192.168.0.0/16')) | |
1130 | |
1131 @property | |
1132 def is_multicast(self): | |
1133 """Test if the address is reserved for multicast use. | |
1134 | |
1135 Returns: | |
1136 A boolean, True if the address is multicast. | |
1137 See RFC 3171 for details. | |
1138 | |
1139 """ | |
1140 return self in IPv4Network('224.0.0.0/4') | |
1141 | |
1142 @property | |
1143 def is_unspecified(self): | |
1144 """Test if the address is unspecified. | |
1145 | |
1146 Returns: | |
1147 A boolean, True if this is the unspecified address as defined in | |
1148 RFC 5735 3. | |
1149 | |
1150 """ | |
1151 return self in IPv4Network('0.0.0.0') | |
1152 | |
1153 @property | |
1154 def is_loopback(self): | |
1155 """Test if the address is a loopback address. | |
1156 | |
1157 Returns: | |
1158 A boolean, True if the address is a loopback per RFC 3330. | |
1159 | |
1160 """ | |
1161 return self in IPv4Network('127.0.0.0/8') | |
1162 | |
1163 @property | |
1164 def is_link_local(self): | |
1165 """Test if the address is reserved for link-local. | |
1166 | |
1167 Returns: | |
1168 A boolean, True if the address is link-local per RFC 3927. | |
1169 | |
1170 """ | |
1171 return self in IPv4Network('169.254.0.0/16') | |
1172 | |
1173 | |
1174 class IPv4Address(_BaseV4, _BaseIP): | |
1175 | |
1176 """Represent and manipulate single IPv4 Addresses.""" | |
1177 | |
1178 def __init__(self, address): | |
1179 | |
1180 """ | |
1181 Args: | |
1182 address: A string or integer representing the IP | |
1183 '192.168.1.1' | |
1184 | |
1185 Additionally, an integer can be passed, so | |
1186 IPv4Address('192.168.1.1') == IPv4Address(3232235777). | |
1187 or, more generally | |
1188 IPv4Address(int(IPv4Address('192.168.1.1'))) == | |
1189 IPv4Address('192.168.1.1') | |
1190 | |
1191 Raises: | |
1192 AddressValueError: If ipaddr isn't a valid IPv4 address. | |
1193 | |
1194 """ | |
1195 _BaseV4.__init__(self, address) | |
1196 | |
1197 # Efficient constructor from integer. | |
1198 if isinstance(address, (int, long)): | |
1199 self._ip = address | |
1200 if address < 0 or address > self._ALL_ONES: | |
1201 raise AddressValueError(address) | |
1202 return | |
1203 | |
1204 # Constructing from a packed address | |
1205 if isinstance(address, Bytes): | |
1206 try: | |
1207 self._ip, = struct.unpack('!I', address) | |
1208 except struct.error: | |
1209 raise AddressValueError(address) # Wrong length. | |
1210 return | |
1211 | |
1212 # Assume input argument to be string or any object representation | |
1213 # which converts into a formatted IP string. | |
1214 addr_str = str(address) | |
1215 self._ip = self._ip_int_from_string(addr_str) | |
1216 | |
1217 | |
1218 class IPv4Network(_BaseV4, _BaseNet): | |
1219 | |
1220 """This class represents and manipulates 32-bit IPv4 networks. | |
1221 | |
1222 Attributes: [examples for IPv4Network('1.2.3.4/27')] | |
1223 ._ip: 16909060 | |
1224 .ip: IPv4Address('1.2.3.4') | |
1225 .network: IPv4Address('1.2.3.0') | |
1226 .hostmask: IPv4Address('0.0.0.31') | |
1227 .broadcast: IPv4Address('1.2.3.31') | |
1228 .netmask: IPv4Address('255.255.255.224') | |
1229 .prefixlen: 27 | |
1230 | |
1231 """ | |
1232 | |
1233 # the valid octets for host and netmasks. only useful for IPv4. | |
1234 _valid_mask_octets = set((255, 254, 252, 248, 240, 224, 192, 128, 0)) | |
1235 | |
1236 def __init__(self, address, strict=False): | |
1237 """Instantiate a new IPv4 network object. | |
1238 | |
1239 Args: | |
1240 address: A string or integer representing the IP [& network]. | |
1241 '192.168.1.1/24' | |
1242 '192.168.1.1/255.255.255.0' | |
1243 '192.168.1.1/0.0.0.255' | |
1244 are all functionally the same in IPv4. Similarly, | |
1245 '192.168.1.1' | |
1246 '192.168.1.1/255.255.255.255' | |
1247 '192.168.1.1/32' | |
1248 are also functionaly equivalent. That is to say, failing to | |
1249 provide a subnetmask will create an object with a mask of /32. | |
1250 | |
1251 If the mask (portion after the / in the argument) is given in | |
1252 dotted quad form, it is treated as a netmask if it starts with a | |
1253 non-zero field (e.g. /255.0.0.0 == /8) and as a hostmask if it | |
1254 starts with a zero field (e.g. 0.255.255.255 == /8), with the | |
1255 single exception of an all-zero mask which is treated as a | |
1256 netmask == /0. If no mask is given, a default of /32 is used. | |
1257 | |
1258 Additionally, an integer can be passed, so | |
1259 IPv4Network('192.168.1.1') == IPv4Network(3232235777). | |
1260 or, more generally | |
1261 IPv4Network(int(IPv4Network('192.168.1.1'))) == | |
1262 IPv4Network('192.168.1.1') | |
1263 | |
1264 strict: A boolean. If true, ensure that we have been passed | |
1265 A true network address, eg, 192.168.1.0/24 and not an | |
1266 IP address on a network, eg, 192.168.1.1/24. | |
1267 | |
1268 Raises: | |
1269 AddressValueError: If ipaddr isn't a valid IPv4 address. | |
1270 NetmaskValueError: If the netmask isn't valid for | |
1271 an IPv4 address. | |
1272 ValueError: If strict was True and a network address was not | |
1273 supplied. | |
1274 | |
1275 """ | |
1276 _BaseNet.__init__(self, address) | |
1277 _BaseV4.__init__(self, address) | |
1278 | |
1279 # Constructing from an integer or packed bytes. | |
1280 if isinstance(address, (int, long, Bytes)): | |
1281 self.ip = IPv4Address(address) | |
1282 self._ip = self.ip._ip | |
1283 self._prefixlen = self._max_prefixlen | |
1284 self.netmask = IPv4Address(self._ALL_ONES) | |
1285 return | |
1286 | |
1287 # Assume input argument to be string or any object representation | |
1288 # which converts into a formatted IP prefix string. | |
1289 addr = str(address).split('/') | |
1290 | |
1291 if len(addr) > 2: | |
1292 raise AddressValueError(address) | |
1293 | |
1294 self._ip = self._ip_int_from_string(addr[0]) | |
1295 self.ip = IPv4Address(self._ip) | |
1296 | |
1297 if len(addr) == 2: | |
1298 mask = addr[1].split('.') | |
1299 if len(mask) == 4: | |
1300 # We have dotted decimal netmask. | |
1301 if self._is_valid_netmask(addr[1]): | |
1302 self.netmask = IPv4Address(self._ip_int_from_string( | |
1303 addr[1])) | |
1304 elif self._is_hostmask(addr[1]): | |
1305 self.netmask = IPv4Address( | |
1306 self._ip_int_from_string(addr[1]) ^ self._ALL_ONES) | |
1307 else: | |
1308 raise NetmaskValueError('%s is not a valid netmask' | |
1309 % addr[1]) | |
1310 | |
1311 self._prefixlen = self._prefix_from_ip_int(int(self.netmask)) | |
1312 else: | |
1313 # We have a netmask in prefix length form. | |
1314 if not self._is_valid_netmask(addr[1]): | |
1315 raise NetmaskValueError(addr[1]) | |
1316 self._prefixlen = int(addr[1]) | |
1317 self.netmask = IPv4Address(self._ip_int_from_prefix( | |
1318 self._prefixlen)) | |
1319 else: | |
1320 self._prefixlen = self._max_prefixlen | |
1321 self.netmask = IPv4Address(self._ip_int_from_prefix( | |
1322 self._prefixlen)) | |
1323 if strict: | |
1324 if self.ip != self.network: | |
1325 raise ValueError('%s has host bits set' % | |
1326 self.ip) | |
1327 if self._prefixlen == (self._max_prefixlen - 1): | |
1328 self.iterhosts = self.__iter__ | |
1329 | |
1330 def _is_hostmask(self, ip_str): | |
1331 """Test if the IP string is a hostmask (rather than a netmask). | |
1332 | |
1333 Args: | |
1334 ip_str: A string, the potential hostmask. | |
1335 | |
1336 Returns: | |
1337 A boolean, True if the IP string is a hostmask. | |
1338 | |
1339 """ | |
1340 bits = ip_str.split('.') | |
1341 try: | |
1342 parts = [int(x) for x in bits if int(x) in self._valid_mask_octets] | |
1343 except ValueError: | |
1344 return False | |
1345 if len(parts) != len(bits): | |
1346 return False | |
1347 if parts[0] < parts[-1]: | |
1348 return True | |
1349 return False | |
1350 | |
1351 def _is_valid_netmask(self, netmask): | |
1352 """Verify that the netmask is valid. | |
1353 | |
1354 Args: | |
1355 netmask: A string, either a prefix or dotted decimal | |
1356 netmask. | |
1357 | |
1358 Returns: | |
1359 A boolean, True if the prefix represents a valid IPv4 | |
1360 netmask. | |
1361 | |
1362 """ | |
1363 mask = netmask.split('.') | |
1364 if len(mask) == 4: | |
1365 if [x for x in mask if int(x) not in self._valid_mask_octets]: | |
1366 return False | |
1367 if [y for idx, y in enumerate(mask) if idx > 0 and | |
1368 y > mask[idx - 1]]: | |
1369 return False | |
1370 return True | |
1371 try: | |
1372 netmask = int(netmask) | |
1373 except ValueError: | |
1374 return False | |
1375 return 0 <= netmask <= self._max_prefixlen | |
1376 | |
1377 # backwards compatibility | |
1378 IsRFC1918 = lambda self: self.is_private | |
1379 IsMulticast = lambda self: self.is_multicast | |
1380 IsLoopback = lambda self: self.is_loopback | |
1381 IsLinkLocal = lambda self: self.is_link_local | |
1382 | |
1383 | |
1384 class _BaseV6(object): | |
1385 | |
1386 """Base IPv6 object. | |
1387 | |
1388 The following methods are used by IPv6 objects in both single IP | |
1389 addresses and networks. | |
1390 | |
1391 """ | |
1392 | |
1393 _ALL_ONES = (2 ** IPV6LENGTH) - 1 | |
1394 _HEXTET_COUNT = 8 | |
1395 _HEX_DIGITS = frozenset('0123456789ABCDEFabcdef') | |
1396 | |
1397 def __init__(self, address): | |
1398 self._version = 6 | |
1399 self._max_prefixlen = IPV6LENGTH | |
1400 | |
1401 def _ip_int_from_string(self, ip_str): | |
1402 """Turn an IPv6 ip_str into an integer. | |
1403 | |
1404 Args: | |
1405 ip_str: A string, the IPv6 ip_str. | |
1406 | |
1407 Returns: | |
1408 A long, the IPv6 ip_str. | |
1409 | |
1410 Raises: | |
1411 AddressValueError: if ip_str isn't a valid IPv6 Address. | |
1412 | |
1413 """ | |
1414 parts = ip_str.split(':') | |
1415 | |
1416 # An IPv6 address needs at least 2 colons (3 parts). | |
1417 if len(parts) < 3: | |
1418 raise AddressValueError(ip_str) | |
1419 | |
1420 # If the address has an IPv4-style suffix, convert it to hexadecimal. | |
1421 if '.' in parts[-1]: | |
1422 ipv4_int = IPv4Address(parts.pop())._ip | |
1423 parts.append('%x' % ((ipv4_int >> 16) & 0xFFFF)) | |
1424 parts.append('%x' % (ipv4_int & 0xFFFF)) | |
1425 | |
1426 # An IPv6 address can't have more than 8 colons (9 parts). | |
1427 if len(parts) > self._HEXTET_COUNT + 1: | |
1428 raise AddressValueError(ip_str) | |
1429 | |
1430 # Disregarding the endpoints, find '::' with nothing in between. | |
1431 # This indicates that a run of zeroes has been skipped. | |
1432 try: | |
1433 skip_index, = ( | |
1434 [i for i in xrange(1, len(parts) - 1) if not parts[i]] or | |
1435 [None]) | |
1436 except ValueError: | |
1437 # Can't have more than one '::' | |
1438 raise AddressValueError(ip_str) | |
1439 | |
1440 # parts_hi is the number of parts to copy from above/before the '::' | |
1441 # parts_lo is the number of parts to copy from below/after the '::' | |
1442 if skip_index is not None: | |
1443 # If we found a '::', then check if it also covers the endpoints. | |
1444 parts_hi = skip_index | |
1445 parts_lo = len(parts) - skip_index - 1 | |
1446 if not parts[0]: | |
1447 parts_hi -= 1 | |
1448 if parts_hi: | |
1449 raise AddressValueError(ip_str) # ^: requires ^:: | |
1450 if not parts[-1]: | |
1451 parts_lo -= 1 | |
1452 if parts_lo: | |
1453 raise AddressValueError(ip_str) # :$ requires ::$ | |
1454 parts_skipped = self._HEXTET_COUNT - (parts_hi + parts_lo) | |
1455 if parts_skipped < 1: | |
1456 raise AddressValueError(ip_str) | |
1457 else: | |
1458 # Otherwise, allocate the entire address to parts_hi. The endpoints | |
1459 # could still be empty, but _parse_hextet() will check for that. | |
1460 if len(parts) != self._HEXTET_COUNT: | |
1461 raise AddressValueError(ip_str) | |
1462 parts_hi = len(parts) | |
1463 parts_lo = 0 | |
1464 parts_skipped = 0 | |
1465 | |
1466 try: | |
1467 # Now, parse the hextets into a 128-bit integer. | |
1468 ip_int = 0L | |
1469 for i in xrange(parts_hi): | |
1470 ip_int <<= 16 | |
1471 ip_int |= self._parse_hextet(parts[i]) | |
1472 ip_int <<= 16 * parts_skipped | |
1473 for i in xrange(-parts_lo, 0): | |
1474 ip_int <<= 16 | |
1475 ip_int |= self._parse_hextet(parts[i]) | |
1476 return ip_int | |
1477 except ValueError: | |
1478 raise AddressValueError(ip_str) | |
1479 | |
1480 def _parse_hextet(self, hextet_str): | |
1481 """Convert an IPv6 hextet string into an integer. | |
1482 | |
1483 Args: | |
1484 hextet_str: A string, the number to parse. | |
1485 | |
1486 Returns: | |
1487 The hextet as an integer. | |
1488 | |
1489 Raises: | |
1490 ValueError: if the input isn't strictly a hex number from [0..FFFF]. | |
1491 | |
1492 """ | |
1493 # Whitelist the characters, since int() allows a lot of bizarre stuff. | |
1494 if not self._HEX_DIGITS.issuperset(hextet_str): | |
1495 raise ValueError | |
1496 if len(hextet_str) > 4: | |
1497 raise ValueError | |
1498 hextet_int = int(hextet_str, 16) | |
1499 if hextet_int > 0xFFFF: | |
1500 raise ValueError | |
1501 return hextet_int | |
1502 | |
1503 def _compress_hextets(self, hextets): | |
1504 """Compresses a list of hextets. | |
1505 | |
1506 Compresses a list of strings, replacing the longest continuous | |
1507 sequence of "0" in the list with "" and adding empty strings at | |
1508 the beginning or at the end of the string such that subsequently | |
1509 calling ":".join(hextets) will produce the compressed version of | |
1510 the IPv6 address. | |
1511 | |
1512 Args: | |
1513 hextets: A list of strings, the hextets to compress. | |
1514 | |
1515 Returns: | |
1516 A list of strings. | |
1517 | |
1518 """ | |
1519 best_doublecolon_start = -1 | |
1520 best_doublecolon_len = 0 | |
1521 doublecolon_start = -1 | |
1522 doublecolon_len = 0 | |
1523 for index in range(len(hextets)): | |
1524 if hextets[index] == '0': | |
1525 doublecolon_len += 1 | |
1526 if doublecolon_start == -1: | |
1527 # Start of a sequence of zeros. | |
1528 doublecolon_start = index | |
1529 if doublecolon_len > best_doublecolon_len: | |
1530 # This is the longest sequence of zeros so far. | |
1531 best_doublecolon_len = doublecolon_len | |
1532 best_doublecolon_start = doublecolon_start | |
1533 else: | |
1534 doublecolon_len = 0 | |
1535 doublecolon_start = -1 | |
1536 | |
1537 if best_doublecolon_len > 1: | |
1538 best_doublecolon_end = (best_doublecolon_start + | |
1539 best_doublecolon_len) | |
1540 # For zeros at the end of the address. | |
1541 if best_doublecolon_end == len(hextets): | |
1542 hextets += [''] | |
1543 hextets[best_doublecolon_start:best_doublecolon_end] = [''] | |
1544 # For zeros at the beginning of the address. | |
1545 if best_doublecolon_start == 0: | |
1546 hextets = [''] + hextets | |
1547 | |
1548 return hextets | |
1549 | |
1550 def _string_from_ip_int(self, ip_int=None): | |
1551 """Turns a 128-bit integer into hexadecimal notation. | |
1552 | |
1553 Args: | |
1554 ip_int: An integer, the IP address. | |
1555 | |
1556 Returns: | |
1557 A string, the hexadecimal representation of the address. | |
1558 | |
1559 Raises: | |
1560 ValueError: The address is bigger than 128 bits of all ones. | |
1561 | |
1562 """ | |
1563 if not ip_int and ip_int != 0: | |
1564 ip_int = int(self._ip) | |
1565 | |
1566 if ip_int > self._ALL_ONES: | |
1567 raise ValueError('IPv6 address is too large') | |
1568 | |
1569 hex_str = '%032x' % ip_int | |
1570 hextets = [] | |
1571 for x in range(0, 32, 4): | |
1572 hextets.append('%x' % int(hex_str[x:x + 4], 16)) | |
1573 | |
1574 hextets = self._compress_hextets(hextets) | |
1575 return ':'.join(hextets) | |
1576 | |
1577 def _explode_shorthand_ip_string(self): | |
1578 """Expand a shortened IPv6 address. | |
1579 | |
1580 Args: | |
1581 ip_str: A string, the IPv6 address. | |
1582 | |
1583 Returns: | |
1584 A string, the expanded IPv6 address. | |
1585 | |
1586 """ | |
1587 if isinstance(self, _BaseNet): | |
1588 ip_str = str(self.ip) | |
1589 else: | |
1590 ip_str = str(self) | |
1591 | |
1592 ip_int = self._ip_int_from_string(ip_str) | |
1593 parts = [] | |
1594 for i in xrange(self._HEXTET_COUNT): | |
1595 parts.append('%04x' % (ip_int & 0xFFFF)) | |
1596 ip_int >>= 16 | |
1597 parts.reverse() | |
1598 if isinstance(self, _BaseNet): | |
1599 return '%s/%d' % (':'.join(parts), self.prefixlen) | |
1600 return ':'.join(parts) | |
1601 | |
1602 @property | |
1603 def max_prefixlen(self): | |
1604 return self._max_prefixlen | |
1605 | |
1606 @property | |
1607 def packed(self): | |
1608 """The binary representation of this address.""" | |
1609 return v6_int_to_packed(self._ip) | |
1610 | |
1611 @property | |
1612 def version(self): | |
1613 return self._version | |
1614 | |
1615 @property | |
1616 def is_multicast(self): | |
1617 """Test if the address is reserved for multicast use. | |
1618 | |
1619 Returns: | |
1620 A boolean, True if the address is a multicast address. | |
1621 See RFC 2373 2.7 for details. | |
1622 | |
1623 """ | |
1624 return self in IPv6Network('ff00::/8') | |
1625 | |
1626 @property | |
1627 def is_reserved(self): | |
1628 """Test if the address is otherwise IETF reserved. | |
1629 | |
1630 Returns: | |
1631 A boolean, True if the address is within one of the | |
1632 reserved IPv6 Network ranges. | |
1633 | |
1634 """ | |
1635 return (self in IPv6Network('::/8') or | |
1636 self in IPv6Network('100::/8') or | |
1637 self in IPv6Network('200::/7') or | |
1638 self in IPv6Network('400::/6') or | |
1639 self in IPv6Network('800::/5') or | |
1640 self in IPv6Network('1000::/4') or | |
1641 self in IPv6Network('4000::/3') or | |
1642 self in IPv6Network('6000::/3') or | |
1643 self in IPv6Network('8000::/3') or | |
1644 self in IPv6Network('A000::/3') or | |
1645 self in IPv6Network('C000::/3') or | |
1646 self in IPv6Network('E000::/4') or | |
1647 self in IPv6Network('F000::/5') or | |
1648 self in IPv6Network('F800::/6') or | |
1649 self in IPv6Network('FE00::/9')) | |
1650 | |
1651 @property | |
1652 def is_unspecified(self): | |
1653 """Test if the address is unspecified. | |
1654 | |
1655 Returns: | |
1656 A boolean, True if this is the unspecified address as defined in | |
1657 RFC 2373 2.5.2. | |
1658 | |
1659 """ | |
1660 return self._ip == 0 and getattr(self, '_prefixlen', 128) == 128 | |
1661 | |
1662 @property | |
1663 def is_loopback(self): | |
1664 """Test if the address is a loopback address. | |
1665 | |
1666 Returns: | |
1667 A boolean, True if the address is a loopback address as defined in | |
1668 RFC 2373 2.5.3. | |
1669 | |
1670 """ | |
1671 return self._ip == 1 and getattr(self, '_prefixlen', 128) == 128 | |
1672 | |
1673 @property | |
1674 def is_link_local(self): | |
1675 """Test if the address is reserved for link-local. | |
1676 | |
1677 Returns: | |
1678 A boolean, True if the address is reserved per RFC 4291. | |
1679 | |
1680 """ | |
1681 return self in IPv6Network('fe80::/10') | |
1682 | |
1683 @property | |
1684 def is_site_local(self): | |
1685 """Test if the address is reserved for site-local. | |
1686 | |
1687 Note that the site-local address space has been deprecated by RFC 3879. | |
1688 Use is_private to test if this address is in the space of unique local | |
1689 addresses as defined by RFC 4193. | |
1690 | |
1691 Returns: | |
1692 A boolean, True if the address is reserved per RFC 3513 2.5.6. | |
1693 | |
1694 """ | |
1695 return self in IPv6Network('fec0::/10') | |
1696 | |
1697 @property | |
1698 def is_private(self): | |
1699 """Test if this address is allocated for private networks. | |
1700 | |
1701 Returns: | |
1702 A boolean, True if the address is reserved per RFC 4193. | |
1703 | |
1704 """ | |
1705 return self in IPv6Network('fc00::/7') | |
1706 | |
1707 @property | |
1708 def ipv4_mapped(self): | |
1709 """Return the IPv4 mapped address. | |
1710 | |
1711 Returns: | |
1712 If the IPv6 address is a v4 mapped address, return the | |
1713 IPv4 mapped address. Return None otherwise. | |
1714 | |
1715 """ | |
1716 if (self._ip >> 32) != 0xFFFF: | |
1717 return None | |
1718 return IPv4Address(self._ip & 0xFFFFFFFF) | |
1719 | |
1720 @property | |
1721 def teredo(self): | |
1722 """Tuple of embedded teredo IPs. | |
1723 | |
1724 Returns: | |
1725 Tuple of the (server, client) IPs or None if the address | |
1726 doesn't appear to be a teredo address (doesn't start with | |
1727 2001::/32) | |
1728 | |
1729 """ | |
1730 if (self._ip >> 96) != 0x20010000: | |
1731 return None | |
1732 return (IPv4Address((self._ip >> 64) & 0xFFFFFFFF), | |
1733 IPv4Address(~self._ip & 0xFFFFFFFF)) | |
1734 | |
1735 @property | |
1736 def sixtofour(self): | |
1737 """Return the IPv4 6to4 embedded address. | |
1738 | |
1739 Returns: | |
1740 The IPv4 6to4-embedded address if present or None if the | |
1741 address doesn't appear to contain a 6to4 embedded address. | |
1742 | |
1743 """ | |
1744 if (self._ip >> 112) != 0x2002: | |
1745 return None | |
1746 return IPv4Address((self._ip >> 80) & 0xFFFFFFFF) | |
1747 | |
1748 | |
1749 class IPv6Address(_BaseV6, _BaseIP): | |
1750 | |
1751 """Represent and manipulate single IPv6 Addresses. | |
1752 """ | |
1753 | |
1754 def __init__(self, address): | |
1755 """Instantiate a new IPv6 address object. | |
1756 | |
1757 Args: | |
1758 address: A string or integer representing the IP | |
1759 | |
1760 Additionally, an integer can be passed, so | |
1761 IPv6Address('2001:4860::') == | |
1762 IPv6Address(42541956101370907050197289607612071936L). | |
1763 or, more generally | |
1764 IPv6Address(IPv6Address('2001:4860::')._ip) == | |
1765 IPv6Address('2001:4860::') | |
1766 | |
1767 Raises: | |
1768 AddressValueError: If address isn't a valid IPv6 address. | |
1769 | |
1770 """ | |
1771 _BaseV6.__init__(self, address) | |
1772 | |
1773 # Efficient constructor from integer. | |
1774 if isinstance(address, (int, long)): | |
1775 self._ip = address | |
1776 if address < 0 or address > self._ALL_ONES: | |
1777 raise AddressValueError(address) | |
1778 return | |
1779 | |
1780 # Constructing from a packed address | |
1781 if isinstance(address, Bytes): | |
1782 try: | |
1783 hi, lo = struct.unpack('!QQ', address) | |
1784 except struct.error: | |
1785 raise AddressValueError(address) # Wrong length. | |
1786 self._ip = (hi << 64) | lo | |
1787 return | |
1788 | |
1789 # Assume input argument to be string or any object representation | |
1790 # which converts into a formatted IP string. | |
1791 addr_str = str(address) | |
1792 if not addr_str: | |
1793 raise AddressValueError('') | |
1794 | |
1795 self._ip = self._ip_int_from_string(addr_str) | |
1796 | |
1797 | |
1798 class IPv6Network(_BaseV6, _BaseNet): | |
1799 | |
1800 """This class represents and manipulates 128-bit IPv6 networks. | |
1801 | |
1802 Attributes: [examples for IPv6('2001:658:22A:CAFE:200::1/64')] | |
1803 .ip: IPv6Address('2001:658:22a:cafe:200::1') | |
1804 .network: IPv6Address('2001:658:22a:cafe::') | |
1805 .hostmask: IPv6Address('::ffff:ffff:ffff:ffff') | |
1806 .broadcast: IPv6Address('2001:658:22a:cafe:ffff:ffff:ffff:ffff') | |
1807 .netmask: IPv6Address('ffff:ffff:ffff:ffff::') | |
1808 .prefixlen: 64 | |
1809 | |
1810 """ | |
1811 | |
1812 def __init__(self, address, strict=False): | |
1813 """Instantiate a new IPv6 Network object. | |
1814 | |
1815 Args: | |
1816 address: A string or integer representing the IPv6 network or the IP | |
1817 and prefix/netmask. | |
1818 '2001:4860::/128' | |
1819 '2001:4860:0000:0000:0000:0000:0000:0000/128' | |
1820 '2001:4860::' | |
1821 are all functionally the same in IPv6. That is to say, | |
1822 failing to provide a subnetmask will create an object with | |
1823 a mask of /128. | |
1824 | |
1825 Additionally, an integer can be passed, so | |
1826 IPv6Network('2001:4860::') == | |
1827 IPv6Network(42541956101370907050197289607612071936L). | |
1828 or, more generally | |
1829 IPv6Network(IPv6Network('2001:4860::')._ip) == | |
1830 IPv6Network('2001:4860::') | |
1831 | |
1832 strict: A boolean. If true, ensure that we have been passed | |
1833 A true network address, eg, 192.168.1.0/24 and not an | |
1834 IP address on a network, eg, 192.168.1.1/24. | |
1835 | |
1836 Raises: | |
1837 AddressValueError: If address isn't a valid IPv6 address. | |
1838 NetmaskValueError: If the netmask isn't valid for | |
1839 an IPv6 address. | |
1840 ValueError: If strict was True and a network address was not | |
1841 supplied. | |
1842 | |
1843 """ | |
1844 _BaseNet.__init__(self, address) | |
1845 _BaseV6.__init__(self, address) | |
1846 | |
1847 # Constructing from an integer or packed bytes. | |
1848 if isinstance(address, (int, long, Bytes)): | |
1849 self.ip = IPv6Address(address) | |
1850 self._ip = self.ip._ip | |
1851 self._prefixlen = self._max_prefixlen | |
1852 self.netmask = IPv6Address(self._ALL_ONES) | |
1853 return | |
1854 | |
1855 # Assume input argument to be string or any object representation | |
1856 # which converts into a formatted IP prefix string. | |
1857 addr = str(address).split('/') | |
1858 | |
1859 if len(addr) > 2: | |
1860 raise AddressValueError(address) | |
1861 | |
1862 self._ip = self._ip_int_from_string(addr[0]) | |
1863 self.ip = IPv6Address(self._ip) | |
1864 | |
1865 if len(addr) == 2: | |
1866 if self._is_valid_netmask(addr[1]): | |
1867 self._prefixlen = int(addr[1]) | |
1868 else: | |
1869 raise NetmaskValueError(addr[1]) | |
1870 else: | |
1871 self._prefixlen = self._max_prefixlen | |
1872 | |
1873 self.netmask = IPv6Address(self._ip_int_from_prefix(self._prefixlen)) | |
1874 | |
1875 if strict: | |
1876 if self.ip != self.network: | |
1877 raise ValueError('%s has host bits set' % | |
1878 self.ip) | |
1879 if self._prefixlen == (self._max_prefixlen - 1): | |
1880 self.iterhosts = self.__iter__ | |
1881 | |
1882 def _is_valid_netmask(self, prefixlen): | |
1883 """Verify that the netmask/prefixlen is valid. | |
1884 | |
1885 Args: | |
1886 prefixlen: A string, the netmask in prefix length format. | |
1887 | |
1888 Returns: | |
1889 A boolean, True if the prefix represents a valid IPv6 | |
1890 netmask. | |
1891 | |
1892 """ | |
1893 try: | |
1894 prefixlen = int(prefixlen) | |
1895 except ValueError: | |
1896 return False | |
1897 return 0 <= prefixlen <= self._max_prefixlen | |
1898 | |
1899 @property | |
1900 def with_netmask(self): | |
1901 return self.with_prefixlen |