Changeset - 9c7849a1fdfd
[Not reviewed]
default
0 1 0
Mads Kiilerich - 7 years ago 2019-01-05 16:47:08
mads@kiilerich.com
Grafted from: 06cad93246fd
tests: fix doctest for summarize_address_range - it is sensitive to linebreaks
1 file changed with 2 insertions and 4 deletions:
0 comments (0 inline, 0 general)
kallithea/lib/ipaddr.py
Show inline comments
 
# Copyright 2007 Google Inc.
 
#  Licensed to PSF under a Contributor Agreement.
 
#
 
# Licensed under the Apache License, Version 2.0 (the "License");
 
# you may not use this file except in compliance with the License.
 
# You may obtain a copy of the License at
 
#
 
#      http://www.apache.org/licenses/LICENSE-2.0
 
#
 
# Unless required by applicable law or agreed to in writing, software
 
# distributed under the License is distributed on an "AS IS" BASIS,
 
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 
# implied. See the License for the specific language governing
 
# permissions and limitations under the License.
 

	
 
"""A fast, lightweight IPv4/IPv6 manipulation library in Python.
 

	
 
This library is used to create/poke/manipulate IPv4 and IPv6 addresses
 
and networks.
 

	
 
"""
 

	
 
__version__ = 'trunk'
 

	
 
import struct
 

	
 
IPV4LENGTH = 32
 
IPV6LENGTH = 128
 

	
 

	
 
class AddressValueError(ValueError):
 
    """A Value Error related to the address."""
 

	
 

	
 
class NetmaskValueError(ValueError):
 
    """A Value Error related to the netmask."""
 

	
 

	
 
def IPAddress(address, version=None):
 
    """Take an IP string/int and return an object of the correct type.
 

	
 
    Args:
 
        address: A string or integer, the IP address.  Either IPv4 or
 
          IPv6 addresses may be supplied; integers less than 2**32 will
 
          be considered to be IPv4 by default.
 
        version: An Integer, 4 or 6. If set, don't try to automatically
 
          determine what the IP address type is. important for things
 
          like IPAddress(1), which could be IPv4, '0.0.0.1',  or IPv6,
 
          '::1'.
 

	
 
    Returns:
 
        An IPv4Address or IPv6Address object.
 

	
 
    Raises:
 
        ValueError: if the string passed isn't either a v4 or a v6
 
          address.
 

	
 
    """
 
    if version:
 
        if version == 4:
 
            return IPv4Address(address)
 
        elif version == 6:
 
            return IPv6Address(address)
 

	
 
    try:
 
        return IPv4Address(address)
 
    except (AddressValueError, NetmaskValueError):
 
        pass
 

	
 
    try:
 
        return IPv6Address(address)
 
    except (AddressValueError, NetmaskValueError):
 
        pass
 

	
 
    raise ValueError('%r does not appear to be an IPv4 or IPv6 address' %
 
                     address)
 

	
 

	
 
def IPNetwork(address, version=None, strict=False):
 
    """Take an IP string/int and return an object of the correct type.
 

	
 
    Args:
 
        address: A string or integer, the IP address.  Either IPv4 or
 
          IPv6 addresses may be supplied; integers less than 2**32 will
 
          be considered to be IPv4 by default.
 
        version: An Integer, if set, don't try to automatically
 
          determine what the IP address type is. important for things
 
          like IPNetwork(1), which could be IPv4, '0.0.0.1/32', or IPv6,
 
          '::1/128'.
 

	
 
    Returns:
 
        An IPv4Network or IPv6Network object.
 

	
 
    Raises:
 
        ValueError: if the string passed isn't either a v4 or a v6
 
          address. Or if a strict network was requested and a strict
 
          network wasn't given.
 

	
 
    """
 
    if version:
 
        if version == 4:
 
            return IPv4Network(address, strict)
 
        elif version == 6:
 
            return IPv6Network(address, strict)
 

	
 
    try:
 
        return IPv4Network(address, strict)
 
    except (AddressValueError, NetmaskValueError):
 
        pass
 

	
 
    try:
 
        return IPv6Network(address, strict)
 
    except (AddressValueError, NetmaskValueError):
 
        pass
 

	
 
    raise ValueError('%r does not appear to be an IPv4 or IPv6 network' %
 
                     address)
 

	
 

	
 
def v4_int_to_packed(address):
 
    """The binary representation of this address.
 

	
 
    Args:
 
        address: An integer representation of an IPv4 IP address.
 

	
 
    Returns:
 
        The binary representation of this address.
 

	
 
    Raises:
 
        ValueError: If the integer is too large to be an IPv4 IP
 
          address.
 
    """
 
    if address > _BaseV4._ALL_ONES:
 
        raise ValueError('Address too large for IPv4')
 
    return Bytes(struct.pack('!I', address))
 

	
 

	
 
def v6_int_to_packed(address):
 
    """The binary representation of this address.
 

	
 
    Args:
 
        address: An integer representation of an IPv6 IP address.
 

	
 
    Returns:
 
        The binary representation of this address.
 
    """
 
    return Bytes(struct.pack('!QQ', address >> 64, address & (2 ** 64 - 1)))
 

	
 

	
 
def _find_address_range(addresses):
 
    """Find a sequence of addresses.
 

	
 
    Args:
 
        addresses: a list of IPv4 or IPv6 addresses.
 

	
 
    Returns:
 
        A tuple containing the first and last IP addresses in the sequence.
 

	
 
    """
 
    first = last = addresses[0]
 
    for ip in addresses[1:]:
 
        if ip._ip == last._ip + 1:
 
            last = ip
 
        else:
 
            break
 
    return (first, last)
 

	
 

	
 
def _get_prefix_length(number1, number2, bits):
 
    """Get the number of leading bits that are same for two numbers.
 

	
 
    Args:
 
        number1: an integer.
 
        number2: another integer.
 
        bits: the maximum number of bits to compare.
 

	
 
    Returns:
 
        The number of leading bits that are the same for two numbers.
 

	
 
    """
 
    for i in range(bits):
 
        if number1 >> i == number2 >> i:
 
            return bits - i
 
    return 0
 

	
 

	
 
def _count_righthand_zero_bits(number, bits):
 
    """Count the number of zero bits on the right hand side.
 

	
 
    Args:
 
        number: an integer.
 
        bits: maximum number of bits to count.
 

	
 
    Returns:
 
        The number of zero bits on the right hand side of the number.
 

	
 
    """
 
    if number == 0:
 
        return bits
 
    for i in range(bits):
 
        if (number >> i) % 2:
 
            return i
 

	
 

	
 
def summarize_address_range(first, last):
 
    """Summarize a network range given the first and last IP addresses.
 

	
 
    Example:
 
        >>> summarize_address_range(IPv4Address('1.1.1.0'),
 
            IPv4Address('1.1.1.130'))
 
        [IPv4Network('1.1.1.0/25'), IPv4Network('1.1.1.128/31'),
 
        IPv4Network('1.1.1.130/32')]
 
        >>> summarize_address_range(IPv4Address('1.1.1.0'), IPv4Address('1.1.1.130'))
 
        [IPv4Network('1.1.1.0/25'), IPv4Network('1.1.1.128/31'), IPv4Network('1.1.1.130/32')]
 

	
 
    Args:
 
        first: the first IPv4Address or IPv6Address in the range.
 
        last: the last IPv4Address or IPv6Address in the range.
 

	
 
    Returns:
 
        The address range collapsed to a list of IPv4Network's or
 
        IPv6Network's.
 

	
 
    Raise:
 
        TypeError:
 
            If the first and last objects are not IP addresses.
 
            If the first and last objects are not the same version.
 
        ValueError:
 
            If the last object is not greater than the first.
 
            If the version is not 4 or 6.
 

	
 
    """
 
    if not (isinstance(first, _BaseIP) and isinstance(last, _BaseIP)):
 
        raise TypeError('first and last must be IP addresses, not networks')
 
    if first.version != last.version:
 
        raise TypeError("%s and %s are not of the same version" % (
 
                str(first), str(last)))
 
    if first > last:
 
        raise ValueError('last IP address must be greater than first')
 

	
 
    networks = []
 

	
 
    if first.version == 4:
 
        ip = IPv4Network
 
    elif first.version == 6:
 
        ip = IPv6Network
 
    else:
 
        raise ValueError('unknown IP version')
 

	
 
    ip_bits = first._max_prefixlen
 
    first_int = first._ip
 
    last_int = last._ip
 
    while first_int <= last_int:
 
        nbits = _count_righthand_zero_bits(first_int, ip_bits)
 
        current = None
 
        while nbits >= 0:
 
            addend = 2 ** nbits - 1
 
            current = first_int + addend
 
            nbits -= 1
 
            if current <= last_int:
 
                break
 
        prefix = _get_prefix_length(first_int, current, ip_bits)
 
        net = ip('%s/%d' % (str(first), prefix))
 
        networks.append(net)
 
        if current == ip._ALL_ONES:
 
            break
 
        first_int = current + 1
 
        first = IPAddress(first_int, version=first._version)
 
    return networks
 

	
 

	
 
def _collapse_address_list_recursive(addresses):
 
    """Loops through the addresses, collapsing concurrent netblocks.
 

	
 
    Example:
 

	
 
        ip1 = IPv4Network('1.1.0.0/24')
 
        ip2 = IPv4Network('1.1.1.0/24')
 
        ip3 = IPv4Network('1.1.2.0/24')
 
        ip4 = IPv4Network('1.1.3.0/24')
 
        ip5 = IPv4Network('1.1.4.0/24')
 
        ip6 = IPv4Network('1.1.0.1/22')
 

	
 
        _collapse_address_list_recursive([ip1, ip2, ip3, ip4, ip5, ip6]) ->
 
          [IPv4Network('1.1.0.0/22'), IPv4Network('1.1.4.0/24')]
 

	
 
        This shouldn't be called directly; it is called via
 
          collapse_address_list([]).
 

	
 
    Args:
 
        addresses: A list of IPv4Network's or IPv6Network's
 

	
 
    Returns:
 
        A list of IPv4Network's or IPv6Network's depending on what we were
 
        passed.
 

	
 
    """
 
    ret_array = []
 
    optimized = False
 

	
 
    for cur_addr in addresses:
 
        if not ret_array:
 
            ret_array.append(cur_addr)
 
            continue
 
        if cur_addr in ret_array[-1]:
 
            optimized = True
 
        elif cur_addr == ret_array[-1].supernet().subnet()[1]:
 
            ret_array.append(ret_array.pop().supernet())
 
            optimized = True
 
        else:
 
            ret_array.append(cur_addr)
 

	
 
    if optimized:
 
        return _collapse_address_list_recursive(ret_array)
 

	
 
    return ret_array
 

	
 

	
 
def collapse_address_list(addresses):
 
    """Collapse a list of IP objects.
 

	
 
    Example:
 
        collapse_address_list([IPv4('1.1.0.0/24'), IPv4('1.1.1.0/24')]) ->
 
          [IPv4('1.1.0.0/23')]
 

	
 
    Args:
 
        addresses: A list of IPv4Network or IPv6Network objects.
 

	
 
    Returns:
 
        A list of IPv4Network or IPv6Network objects depending on what we
 
        were passed.
 

	
 
    Raises:
 
        TypeError: If passed a list of mixed version objects.
 

	
 
    """
 
    i = 0
 
    addrs = []
 
    ips = []
 
    nets = []
 

	
 
    # split IP addresses and networks
 
    for ip in addresses:
 
        if isinstance(ip, _BaseIP):
 
            if ips and ips[-1]._version != ip._version:
 
                raise TypeError("%s and %s are not of the same version" % (
 
                        str(ip), str(ips[-1])))
 
            ips.append(ip)
 
        elif ip._prefixlen == ip._max_prefixlen:
 
            if ips and ips[-1]._version != ip._version:
 
                raise TypeError("%s and %s are not of the same version" % (
 
                        str(ip), str(ips[-1])))
 
            ips.append(ip.ip)
 
        else:
 
            if nets and nets[-1]._version != ip._version:
 
                raise TypeError("%s and %s are not of the same version" % (
 
                        str(ip), str(nets[-1])))
 
            nets.append(ip)
 

	
 
    # sort and dedup
 
    ips = sorted(set(ips))
 
    nets = sorted(set(nets))
 

	
 
    while i < len(ips):
 
        (first, last) = _find_address_range(ips[i:])
 
        i = ips.index(last) + 1
 
        addrs.extend(summarize_address_range(first, last))
 

	
 
    return _collapse_address_list_recursive(sorted(
 
        addrs + nets, key=_BaseNet._get_networks_key))
 

	
 

	
 
# backwards compatibility
 
CollapseAddrList = collapse_address_list
 

	
 

	
 
# We need to distinguish between the string and packed-bytes representations
 
# of an IP address.  For example, b'0::1' is the IPv4 address 48.58.58.49,
 
# while '0::1' is an IPv6 address.
 
#
 
# In Python 3, the native 'bytes' type already provides this functionality,
 
# so we use it directly.  For earlier implementations where bytes is not a
 
# distinct type, we create a subclass of str to serve as a tag.
 
#
 
# Usage example (Python 2):
 
#   ip = ipaddr.IPAddress(ipaddr.Bytes('xxxx'))
 
#
 
# Usage example (Python 3):
 
#   ip = ipaddr.IPAddress(b'xxxx')
 
try:
 
    if bytes is str:
 
        raise TypeError("bytes is not a distinct type")
 
    Bytes = bytes
 
except (NameError, TypeError):
 
    class Bytes(str):
 
        def __repr__(self):
 
            return 'Bytes(%s)' % str.__repr__(self)
 

	
 

	
 
def get_mixed_type_key(obj):
 
    """Return a key suitable for sorting between networks and addresses.
 

	
 
    Address and Network objects are not sortable by default; they're
 
    fundamentally different so the expression
 

	
 
        IPv4Address('1.1.1.1') <= IPv4Network('1.1.1.1/24')
 

	
 
    doesn't make any sense.  There are some times however, where you may wish
 
    to have ipaddr sort these for you anyway. If you need to do this, you
 
    can use this function as the key= argument to sorted().
 

	
 
    Args:
 
      obj: either a Network or Address object.
 
    Returns:
 
      appropriate key.
 

	
 
    """
 
    if isinstance(obj, _BaseNet):
 
        return obj._get_networks_key()
 
    elif isinstance(obj, _BaseIP):
 
        return obj._get_address_key()
 
    return NotImplemented
 

	
 

	
 
class _IPAddrBase(object):
 

	
 
    """The mother class."""
 

	
 
    def __index__(self):
 
        return self._ip
 

	
 
    def __int__(self):
 
        return self._ip
 

	
 
    def __hex__(self):
 
        return hex(self._ip)
 

	
 
    @property
 
    def exploded(self):
 
        """Return the longhand version of the IP address as a string."""
 
        return self._explode_shorthand_ip_string()
 

	
 
    @property
 
    def compressed(self):
 
        """Return the shorthand version of the IP address as a string."""
 
        return str(self)
 

	
 

	
 
class _BaseIP(_IPAddrBase):
 

	
 
    """A generic IP object.
 

	
 
    This IP class contains the version independent methods which are
 
    used by single IP addresses.
 

	
 
    """
 

	
 
    def __eq__(self, other):
 
        try:
 
            return (self._ip == other._ip
 
                    and self._version == other._version)
 
        except AttributeError:
 
            return NotImplemented
 

	
 
    def __ne__(self, other):
 
        eq = self.__eq__(other)
 
        if eq is NotImplemented:
 
            return NotImplemented
 
        return not eq
 

	
 
    def __le__(self, other):
 
        gt = self.__gt__(other)
 
        if gt is NotImplemented:
 
            return NotImplemented
 
        return not gt
 

	
 
    def __ge__(self, other):
 
        lt = self.__lt__(other)
 
        if lt is NotImplemented:
 
            return NotImplemented
 
        return not lt
 

	
 
    def __lt__(self, other):
 
        if self._version != other._version:
 
            raise TypeError('%s and %s are not of the same version' % (
 
                    str(self), str(other)))
 
        if not isinstance(other, _BaseIP):
 
            raise TypeError('%s and %s are not of the same type' % (
 
                    str(self), str(other)))
 
        if self._ip != other._ip:
 
            return self._ip < other._ip
 
        return False
 

	
 
    def __gt__(self, other):
 
        if self._version != other._version:
 
            raise TypeError('%s and %s are not of the same version' % (
 
                    str(self), str(other)))
 
        if not isinstance(other, _BaseIP):
 
            raise TypeError('%s and %s are not of the same type' % (
 
                    str(self), str(other)))
 
        if self._ip != other._ip:
 
            return self._ip > other._ip
 
        return False
 

	
 
    # Shorthand for Integer addition and subtraction. This is not
 
    # meant to ever support addition/subtraction of addresses.
 
    def __add__(self, other):
 
        if not isinstance(other, int):
 
            return NotImplemented
 
        return IPAddress(int(self) + other, version=self._version)
 

	
 
    def __sub__(self, other):
 
        if not isinstance(other, int):
 
            return NotImplemented
 
        return IPAddress(int(self) - other, version=self._version)
 

	
 
    def __repr__(self):
 
        return '%s(%r)' % (self.__class__.__name__, str(self))
 

	
 
    def __str__(self):
 
        return self._string_from_ip_int(self._ip)
 

	
 
    def __hash__(self):
 
        return hash(hex(long(self._ip)))
 

	
 
    def _get_address_key(self):
 
        return (self._version, self)
 

	
 
    @property
 
    def version(self):
 
        raise NotImplementedError('BaseIP has no version')
 

	
 

	
 
class _BaseNet(_IPAddrBase):
 

	
 
    """A generic IP object.
 

	
 
    This IP class contains the version independent methods which are
 
    used by networks.
 

	
 
    """
 

	
 
    def __init__(self, address):
 
        self._cache = {}
 

	
 
    def __repr__(self):
 
        return '%s(%r)' % (self.__class__.__name__, str(self))
 

	
 
    def iterhosts(self):
 
        """Generate Iterator over usable hosts in a network.
 

	
 
           This is like __iter__ except it doesn't return the network
 
           or broadcast addresses.
 

	
 
        """
 
        cur = int(self.network) + 1
 
        bcast = int(self.broadcast) - 1
 
        while cur <= bcast:
 
            cur += 1
 
            yield IPAddress(cur - 1, version=self._version)
 

	
 
    def __iter__(self):
 
        cur = int(self.network)
 
        bcast = int(self.broadcast)
 
        while cur <= bcast:
 
            cur += 1
 
            yield IPAddress(cur - 1, version=self._version)
 

	
 
    def __getitem__(self, n):
 
        network = int(self.network)
 
        broadcast = int(self.broadcast)
 
        if n >= 0:
 
            if network + n > broadcast:
 
                raise IndexError
 
            return IPAddress(network + n, version=self._version)
 
        else:
 
            n += 1
 
            if broadcast + n < network:
 
                raise IndexError
 
            return IPAddress(broadcast + n, version=self._version)
 

	
 
    def __lt__(self, other):
 
        if self._version != other._version:
 
            raise TypeError('%s and %s are not of the same version' % (
 
                    str(self), str(other)))
 
        if not isinstance(other, _BaseNet):
 
            raise TypeError('%s and %s are not of the same type' % (
 
                    str(self), str(other)))
 
        if self.network != other.network:
 
            return self.network < other.network
 
        if self.netmask != other.netmask:
 
            return self.netmask < other.netmask
 
        return False
 

	
 
    def __gt__(self, other):
 
        if self._version != other._version:
 
            raise TypeError('%s and %s are not of the same version' % (
 
                    str(self), str(other)))
0 comments (0 inline, 0 general)