diff --git a/gluon/_compat.py b/gluon/_compat.py index 1641696f..9452a44a 100644 --- a/gluon/_compat.py +++ b/gluon/_compat.py @@ -1,6 +1 @@ from pydal._compat import * - -if PY2: - from gluon.contrib import ipaddress -else: - import ipaddress diff --git a/gluon/contrib/ipaddress.py b/gluon/contrib/ipaddress.py deleted file mode 100644 index 9cf71a74..00000000 --- a/gluon/contrib/ipaddress.py +++ /dev/null @@ -1,2425 +0,0 @@ -# Copyright 2007 Google Inc. -# Licensed to PSF under a Contributor Agreement. - -"""A fast, lightweight IPv4/IPv6 manipulation library in Python. - -This library is used to create/poke/manipulate IPv4 and IPv6 addresses -and networks. - -""" - -from __future__ import unicode_literals - - -import itertools -import struct - -__version__ = '1.0.17' - -# Compatibility functions -_compat_int_types = (int,) -try: - _compat_int_types = (int, long) -except NameError: - pass -try: - _compat_str = unicode -except NameError: - _compat_str = str - assert bytes != str -if b'\0'[0] == 0: # Python 3 semantics - def _compat_bytes_to_byte_vals(byt): - return byt -else: - def _compat_bytes_to_byte_vals(byt): - return [struct.unpack(b'!B', b)[0] for b in byt] -try: - _compat_int_from_byte_vals = int.from_bytes -except AttributeError: - def _compat_int_from_byte_vals(bytvals, endianess): - assert endianess == 'big' - res = 0 - for bv in bytvals: - assert isinstance(bv, _compat_int_types) - res = (res << 8) + bv - return res - - -def _compat_to_bytes(intval, length, endianess): - assert isinstance(intval, _compat_int_types) - assert endianess == 'big' - if length == 4: - if intval < 0 or intval >= 2 ** 32: - raise struct.error("integer out of range for 'I' format code") - return struct.pack(b'!I', intval) - elif length == 16: - if intval < 0 or intval >= 2 ** 128: - raise struct.error("integer out of range for 'QQ' format code") - return struct.pack(b'!QQ', intval >> 64, intval & 0xffffffffffffffff) - else: - raise NotImplementedError() -if hasattr(int, 'bit_length'): - # Not int.bit_length , since that won't work in 2.7 where long exists - def _compat_bit_length(i): - return i.bit_length() -else: - def _compat_bit_length(i): - for res in itertools.count(): - if i >> res == 0: - return res - - -def _compat_range(start, end, step=1): - assert step > 0 - i = start - while i < end: - yield i - i += step - - -class _TotalOrderingMixin(object): - __slots__ = () - - # Helper that derives the other comparison operations from - # __lt__ and __eq__ - # We avoid functools.total_ordering because it doesn't handle - # NotImplemented correctly yet (http://bugs.python.org/issue10042) - def __eq__(self, other): - raise NotImplementedError - - def __ne__(self, other): - equal = self.__eq__(other) - if equal is NotImplemented: - return NotImplemented - return not equal - - def __lt__(self, other): - raise NotImplementedError - - def __le__(self, other): - less = self.__lt__(other) - if less is NotImplemented or not less: - return self.__eq__(other) - return less - - def __gt__(self, other): - less = self.__lt__(other) - if less is NotImplemented: - return NotImplemented - equal = self.__eq__(other) - if equal is NotImplemented: - return NotImplemented - return not (less or equal) - - def __ge__(self, other): - less = self.__lt__(other) - if less is NotImplemented: - return NotImplemented - return not less - - -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 ip_address(address): - """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. - - Returns: - An IPv4Address or IPv6Address object. - - Raises: - ValueError: if the *address* passed isn't either a v4 or a v6 - address - - """ - try: - return IPv4Address(address) - except (AddressValueError, NetmaskValueError): - pass - - try: - return IPv6Address(address) - except (AddressValueError, NetmaskValueError): - pass - - if isinstance(address, bytes): - raise AddressValueError( - '%r does not appear to be an IPv4 or IPv6 address. ' - 'Did you pass in a bytes (str in Python 2) instead of' - ' a unicode object?' % address) - - raise ValueError('%r does not appear to be an IPv4 or IPv6 address' % - address) - - -def ip_network(address, strict=True): - """Take an IP string/int and return an object of the correct type. - - Args: - address: A string or integer, the IP network. Either IPv4 or - IPv6 networks may be supplied; integers less than 2**32 will - be considered to be IPv4 by default. - - Returns: - An IPv4Network or IPv6Network object. - - Raises: - ValueError: if the string passed isn't either a v4 or a v6 - address. Or if the network has host bits set. - - """ - try: - return IPv4Network(address, strict) - except (AddressValueError, NetmaskValueError): - pass - - try: - return IPv6Network(address, strict) - except (AddressValueError, NetmaskValueError): - pass - - if isinstance(address, bytes): - raise AddressValueError( - '%r does not appear to be an IPv4 or IPv6 network. ' - 'Did you pass in a bytes (str in Python 2) instead of' - ' a unicode object?' % address) - - raise ValueError('%r does not appear to be an IPv4 or IPv6 network' % - address) - - -def ip_interface(address): - """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. - - Returns: - An IPv4Interface or IPv6Interface object. - - Raises: - ValueError: if the string passed isn't either a v4 or a v6 - address. - - Notes: - The IPv?Interface classes describe an Address on a particular - Network, so they're basically a combination of both the Address - and Network classes. - - """ - try: - return IPv4Interface(address) - except (AddressValueError, NetmaskValueError): - pass - - try: - return IPv6Interface(address) - except (AddressValueError, NetmaskValueError): - pass - - raise ValueError('%r does not appear to be an IPv4 or IPv6 interface' % - address) - - -def v4_int_to_packed(address): - """Represent an address as 4 packed bytes in network (big-endian) order. - - Args: - address: An integer representation of an IPv4 IP address. - - Returns: - The integer address packed as 4 bytes in network (big-endian) order. - - Raises: - ValueError: If the integer is negative or too large to be an - IPv4 IP address. - - """ - try: - return _compat_to_bytes(address, 4, 'big') - except (struct.error, OverflowError): - raise ValueError("Address negative or too large for IPv4") - - -def v6_int_to_packed(address): - """Represent an address as 16 packed bytes in network (big-endian) order. - - Args: - address: An integer representation of an IPv6 IP address. - - Returns: - The integer address packed as 16 bytes in network (big-endian) order. - - """ - try: - return _compat_to_bytes(address, 16, 'big') - except (struct.error, OverflowError): - raise ValueError("Address negative or too large for IPv6") - - -def _split_optional_netmask(address): - """Helper to split the netmask and raise AddressValueError if needed""" - addr = _compat_str(address).split('/') - if len(addr) > 2: - raise AddressValueError("Only one '/' permitted in %r" % address) - return addr - - -def _find_address_range(addresses): - """Find a sequence of sorted deduplicated IPv#Address. - - Args: - addresses: a list of IPv#Address objects. - - Yields: - A tuple containing the first and last IP addresses in the sequence. - - """ - it = iter(addresses) - first = last = next(it) - for ip in it: - if ip._ip != last._ip + 1: - yield first, last - first = ip - last = ip - yield first, last - - -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 - return min(bits, _compat_bit_length(~number & (number - 1))) - - -def summarize_address_range(first, last): - """Summarize a network range given the first and last IP addresses. - - Example: - >>> list(summarize_address_range(IPv4Address('192.0.2.0'), - ... IPv4Address('192.0.2.130'))) - ... #doctest: +NORMALIZE_WHITESPACE - [IPv4Network('192.0.2.0/25'), IPv4Network('192.0.2.128/31'), - IPv4Network('192.0.2.130/32')] - - Args: - first: the first IPv4Address or IPv6Address in the range. - last: the last IPv4Address or IPv6Address in the range. - - Returns: - An iterator of the summarized IPv(4|6) network objects. - - 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 of the first address is not 4 or 6. - - """ - if (not (isinstance(first, _BaseAddress) and - isinstance(last, _BaseAddress))): - 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" % ( - first, last)) - if first > last: - raise ValueError('last IP address must be greater than first') - - 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 = min(_count_righthand_zero_bits(first_int, ip_bits), - _compat_bit_length(last_int - first_int + 1) - 1) - net = ip((first_int, ip_bits - nbits)) - yield net - first_int += 1 << nbits - if first_int - 1 == ip._ALL_ONES: - break - - -def _collapse_addresses_internal(addresses): - """Loops through the addresses, collapsing concurrent netblocks. - - Example: - - ip1 = IPv4Network('192.0.2.0/26') - ip2 = IPv4Network('192.0.2.64/26') - ip3 = IPv4Network('192.0.2.128/26') - ip4 = IPv4Network('192.0.2.192/26') - - _collapse_addresses_internal([ip1, ip2, ip3, ip4]) -> - [IPv4Network('192.0.2.0/24')] - - This shouldn't be called directly; it is called via - collapse_addresses([]). - - 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. - - """ - # First merge - to_merge = list(addresses) - subnets = {} - while to_merge: - net = to_merge.pop() - supernet = net.supernet() - existing = subnets.get(supernet) - if existing is None: - subnets[supernet] = net - elif existing != net: - # Merge consecutive subnets - del subnets[supernet] - to_merge.append(supernet) - # Then iterate over resulting networks, skipping subsumed subnets - last = None - for net in sorted(subnets.values()): - if last is not None: - # Since they are sorted, - # last.network_address <= net.network_address is a given. - if last.broadcast_address >= net.broadcast_address: - continue - yield net - last = net - - -def collapse_addresses(addresses): - """Collapse a list of IP objects. - - Example: - collapse_addresses([IPv4Network('192.0.2.0/25'), - IPv4Network('192.0.2.128/25')]) -> - [IPv4Network('192.0.2.0/24')] - - Args: - addresses: An iterator of IPv4Network or IPv6Network objects. - - Returns: - An iterator of the collapsed IPv(4|6)Network objects. - - Raises: - TypeError: If passed a list of mixed version objects. - - """ - addrs = [] - ips = [] - nets = [] - - # split IP addresses and networks - for ip in addresses: - if isinstance(ip, _BaseAddress): - if ips and ips[-1]._version != ip._version: - raise TypeError("%s and %s are not of the same version" % ( - ip, 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" % ( - ip, ips[-1])) - try: - ips.append(ip.ip) - except AttributeError: - ips.append(ip.network_address) - else: - if nets and nets[-1]._version != ip._version: - raise TypeError("%s and %s are not of the same version" % ( - ip, nets[-1])) - nets.append(ip) - - # sort and dedup - ips = sorted(set(ips)) - - # find consecutive address ranges in the sorted sequence and summarize them - if ips: - for first, last in _find_address_range(ips): - addrs.extend(summarize_address_range(first, last)) - - return _collapse_addresses_internal(addrs + nets) - - -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('192.0.2.0') <= IPv4Network('192.0.2.0/24') - - doesn't make any sense. There are some times however, where you may wish - to have ipaddress 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, _BaseNetwork): - return obj._get_networks_key() - elif isinstance(obj, _BaseAddress): - return obj._get_address_key() - return NotImplemented - - -class _IPAddressBase(_TotalOrderingMixin): - - """The mother class.""" - - __slots__ = () - - @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 _compat_str(self) - - @property - def reverse_pointer(self): - """The name of the reverse DNS pointer for the IP address, e.g.: - >>> ipaddress.ip_address("127.0.0.1").reverse_pointer - '1.0.0.127.in-addr.arpa' - >>> ipaddress.ip_address("2001:db8::1").reverse_pointer - '1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.8.b.d.0.1.0.0.2.ip6.arpa' - - """ - return self._reverse_pointer() - - @property - def version(self): - msg = '%200s has no version specified' % (type(self),) - raise NotImplementedError(msg) - - def _check_int_address(self, address): - if address < 0: - msg = "%d (< 0) is not permitted as an IPv%d address" - raise AddressValueError(msg % (address, self._version)) - if address > self._ALL_ONES: - msg = "%d (>= 2**%d) is not permitted as an IPv%d address" - raise AddressValueError(msg % (address, self._max_prefixlen, - self._version)) - - def _check_packed_address(self, address, expected_len): - address_len = len(address) - if address_len != expected_len: - msg = ( - '%r (len %d != %d) is not permitted as an IPv%d address. ' - 'Did you pass in a bytes (str in Python 2) instead of' - ' a unicode object?' - ) - raise AddressValueError(msg % (address, address_len, - expected_len, self._version)) - - @classmethod - def _ip_int_from_prefix(cls, prefixlen): - """Turn the prefix length into a bitwise netmask - - Args: - prefixlen: An integer, the prefix length. - - Returns: - An integer. - - """ - return cls._ALL_ONES ^ (cls._ALL_ONES >> prefixlen) - - @classmethod - def _prefix_from_ip_int(cls, ip_int): - """Return prefix length from the bitwise netmask. - - Args: - ip_int: An integer, the netmask in expanded bitwise format - - Returns: - An integer, the prefix length. - - Raises: - ValueError: If the input intermingles zeroes & ones - """ - trailing_zeroes = _count_righthand_zero_bits(ip_int, - cls._max_prefixlen) - prefixlen = cls._max_prefixlen - trailing_zeroes - leading_ones = ip_int >> trailing_zeroes - all_ones = (1 << prefixlen) - 1 - if leading_ones != all_ones: - byteslen = cls._max_prefixlen // 8 - details = _compat_to_bytes(ip_int, byteslen, 'big') - msg = 'Netmask pattern %r mixes zeroes & ones' - raise ValueError(msg % details) - return prefixlen - - @classmethod - def _report_invalid_netmask(cls, netmask_str): - msg = '%r is not a valid netmask' % netmask_str - raise NetmaskValueError(msg) - - @classmethod - def _prefix_from_prefix_string(cls, prefixlen_str): - """Return prefix length from a numeric string - - Args: - prefixlen_str: The string to be converted - - Returns: - An integer, the prefix length. - - Raises: - NetmaskValueError: If the input is not a valid netmask - """ - # int allows a leading +/- as well as surrounding whitespace, - # so we ensure that isn't the case - if not _BaseV4._DECIMAL_DIGITS.issuperset(prefixlen_str): - cls._report_invalid_netmask(prefixlen_str) - try: - prefixlen = int(prefixlen_str) - except ValueError: - cls._report_invalid_netmask(prefixlen_str) - if not (0 <= prefixlen <= cls._max_prefixlen): - cls._report_invalid_netmask(prefixlen_str) - return prefixlen - - @classmethod - def _prefix_from_ip_string(cls, ip_str): - """Turn a netmask/hostmask string into a prefix length - - Args: - ip_str: The netmask/hostmask to be converted - - Returns: - An integer, the prefix length. - - Raises: - NetmaskValueError: If the input is not a valid netmask/hostmask - """ - # Parse the netmask/hostmask like an IP address. - try: - ip_int = cls._ip_int_from_string(ip_str) - except AddressValueError: - cls._report_invalid_netmask(ip_str) - - # Try matching a netmask (this would be /1*0*/ as a bitwise regexp). - # Note that the two ambiguous cases (all-ones and all-zeroes) are - # treated as netmasks. - try: - return cls._prefix_from_ip_int(ip_int) - except ValueError: - pass - - # Invert the bits, and try matching a /0+1+/ hostmask instead. - ip_int ^= cls._ALL_ONES - try: - return cls._prefix_from_ip_int(ip_int) - except ValueError: - cls._report_invalid_netmask(ip_str) - - def __reduce__(self): - return self.__class__, (_compat_str(self),) - - -class _BaseAddress(_IPAddressBase): - - """A generic IP object. - - This IP class contains the version independent methods which are - used by single IP addresses. - """ - - __slots__ = () - - def __int__(self): - return self._ip - - def __eq__(self, other): - try: - return (self._ip == other._ip and - self._version == other._version) - except AttributeError: - return NotImplemented - - def __lt__(self, other): - if not isinstance(other, _IPAddressBase): - return NotImplemented - if not isinstance(other, _BaseAddress): - raise TypeError('%s and %s are not of the same type' % ( - self, other)) - if self._version != other._version: - raise TypeError('%s and %s are not of the same version' % ( - self, 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, _compat_int_types): - return NotImplemented - return self.__class__(int(self) + other) - - def __sub__(self, other): - if not isinstance(other, _compat_int_types): - return NotImplemented - return self.__class__(int(self) - other) - - def __repr__(self): - return '%s(%r)' % (self.__class__.__name__, _compat_str(self)) - - def __str__(self): - return _compat_str(self._string_from_ip_int(self._ip)) - - def __hash__(self): - return hash(hex(int(self._ip))) - - def _get_address_key(self): - return (self._version, self) - - def __reduce__(self): - return self.__class__, (self._ip,) - - -class _BaseNetwork(_IPAddressBase): - - """A generic IP network 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__, _compat_str(self)) - - def __str__(self): - return '%s/%d' % (self.network_address, self.prefixlen) - - def hosts(self): - """Generate Iterator over usable hosts in a network. - - This is like __iter__ except it doesn't return the network - or broadcast addresses. - - """ - network = int(self.network_address) - broadcast = int(self.broadcast_address) - for x in _compat_range(network + 1, broadcast): - yield self._address_class(x) - - def __iter__(self): - network = int(self.network_address) - broadcast = int(self.broadcast_address) - for x in _compat_range(network, broadcast + 1): - yield self._address_class(x) - - def __getitem__(self, n): - network = int(self.network_address) - broadcast = int(self.broadcast_address) - if n >= 0: - if network + n > broadcast: - raise IndexError('address out of range') - return self._address_class(network + n) - else: - n += 1 - if broadcast + n < network: - raise IndexError('address out of range') - return self._address_class(broadcast + n) - - def __lt__(self, other): - if not isinstance(other, _IPAddressBase): - return NotImplemented - if not isinstance(other, _BaseNetwork): - raise TypeError('%s and %s are not of the same type' % ( - self, other)) - if self._version != other._version: - raise TypeError('%s and %s are not of the same version' % ( - self, other)) - if self.network_address != other.network_address: - return self.network_address < other.network_address - if self.netmask != other.netmask: - return self.netmask < other.netmask - return False - - def __eq__(self, other): - try: - return (self._version == other._version and - self.network_address == other.network_address and - int(self.netmask) == int(other.netmask)) - except AttributeError: - return NotImplemented - - def __hash__(self): - return hash(int(self.network_address) ^ int(self.netmask)) - - def __contains__(self, other): - # always false if one is v4 and the other is v6. - if self._version != other._version: - return False - # dealing with another network. - if isinstance(other, _BaseNetwork): - return False - # dealing with another address - else: - # address - return (int(self.network_address) <= int(other._ip) <= - int(self.broadcast_address)) - - def overlaps(self, other): - """Tell if self is partly contained in other.""" - return self.network_address in other or ( - self.broadcast_address in other or ( - other.network_address in self or ( - other.broadcast_address in self))) - - @property - def broadcast_address(self): - x = self._cache.get('broadcast_address') - if x is None: - x = self._address_class(int(self.network_address) | - int(self.hostmask)) - self._cache['broadcast_address'] = x - return x - - @property - def hostmask(self): - x = self._cache.get('hostmask') - if x is None: - x = self._address_class(int(self.netmask) ^ self._ALL_ONES) - self._cache['hostmask'] = x - return x - - @property - def with_prefixlen(self): - return '%s/%d' % (self.network_address, self._prefixlen) - - @property - def with_netmask(self): - return '%s/%s' % (self.network_address, self.netmask) - - @property - def with_hostmask(self): - return '%s/%s' % (self.network_address, self.hostmask) - - @property - def num_addresses(self): - """Number of hosts in the current subnet.""" - return int(self.broadcast_address) - int(self.network_address) + 1 - - @property - def _address_class(self): - # Returning bare address objects (rather than interfaces) allows for - # more consistent behaviour across the network address, broadcast - # address and individual host addresses. - msg = '%200s has no associated address class' % (type(self),) - raise NotImplementedError(msg) - - @property - def prefixlen(self): - return self._prefixlen - - def address_exclude(self, other): - """Remove an address from a larger block. - - For example: - - addr1 = ip_network('192.0.2.0/28') - addr2 = ip_network('192.0.2.1/32') - list(addr1.address_exclude(addr2)) = - [IPv4Network('192.0.2.0/32'), IPv4Network('192.0.2.2/31'), - IPv4Network('192.0.2.4/30'), IPv4Network('192.0.2.8/29')] - - or IPv6: - - addr1 = ip_network('2001:db8::1/32') - addr2 = ip_network('2001:db8::1/128') - list(addr1.address_exclude(addr2)) = - [ip_network('2001:db8::1/128'), - ip_network('2001:db8::2/127'), - ip_network('2001:db8::4/126'), - ip_network('2001:db8::8/125'), - ... - ip_network('2001:db8:8000::/33')] - - Args: - other: An IPv4Network or IPv6Network object of the same type. - - Returns: - An iterator of the IPv(4|6)Network objects which is self - minus other. - - Raises: - TypeError: If self and other are of differing address - versions, or if other is not a network object. - ValueError: If other is not completely contained by self. - - """ - if not self._version == other._version: - raise TypeError("%s and %s are not of the same version" % ( - self, other)) - - if not isinstance(other, _BaseNetwork): - raise TypeError("%s is not a network object" % other) - - if not other.subnet_of(self): - raise ValueError('%s not contained in %s' % (other, self)) - if other == self: - return - - # Make sure we're comparing the network of other. - other = other.__class__('%s/%s' % (other.network_address, - other.prefixlen)) - - s1, s2 = self.subnets() - while s1 != other and s2 != other: - if other.subnet_of(s1): - yield s2 - s1, s2 = s1.subnets() - elif other.subnet_of(s2): - yield s1 - s1, s2 = s2.subnets() - else: - # If we got here, there's a bug somewhere. - raise AssertionError('Error performing exclusion: ' - 's1: %s s2: %s other: %s' % - (s1, s2, other)) - if s1 == other: - yield s2 - elif s2 == other: - yield s1 - else: - # If we got here, there's a bug somewhere. - raise AssertionError('Error performing exclusion: ' - 's1: %s s2: %s other: %s' % - (s1, s2, other)) - - def compare_networks(self, other): - """Compare two IP objects. - - This is only concerned about the comparison of the integer - representation of the network addresses. This means that the - host bits aren't considered at all in this method. If you want - to compare host bits, you can easily enough do a - 'HostA._ip < HostB._ip' - - Args: - other: An IP object. - - Returns: - If the IP versions of self and other are the same, returns: - - -1 if self < other: - eg: IPv4Network('192.0.2.0/25') < IPv4Network('192.0.2.128/25') - IPv6Network('2001:db8::1000/124') < - IPv6Network('2001:db8::2000/124') - 0 if self == other - eg: IPv4Network('192.0.2.0/24') == IPv4Network('192.0.2.0/24') - IPv6Network('2001:db8::1000/124') == - IPv6Network('2001:db8::1000/124') - 1 if self > other - eg: IPv4Network('192.0.2.128/25') > IPv4Network('192.0.2.0/25') - IPv6Network('2001:db8::2000/124') > - IPv6Network('2001:db8::1000/124') - - Raises: - TypeError if the IP versions are different. - - """ - # does this need to raise a ValueError? - if self._version != other._version: - raise TypeError('%s and %s are not of the same type' % ( - self, other)) - # self._version == other._version below here: - if self.network_address < other.network_address: - return -1 - if self.network_address > other.network_address: - return 1 - # self.network_address == other.network_address below here: - if self.netmask < other.netmask: - return -1 - if self.netmask > other.netmask: - return 1 - return 0 - - def _get_networks_key(self): - """Network-only key function. - - Returns an object that identifies this address' network and - netmask. This function is a suitable "key" argument for sorted() - and list.sort(). - - """ - return (self._version, self.network_address, self.netmask) - - def subnets(self, prefixlen_diff=1, new_prefix=None): - """The subnets which join to make the current subnet. - - In the case that self contains only one IP - (self._prefixlen == 32 for IPv4 or self._prefixlen == 128 - for IPv6), yield an iterator with just ourself. - - Args: - prefixlen_diff: An integer, the amount the prefix length - should be increased by. This should not be set if - new_prefix is also set. - new_prefix: The desired new prefix length. This must be a - larger number (smaller prefix) than the existing prefix. - This should not be set if prefixlen_diff is also set. - - Returns: - An iterator of IPv(4|6) objects. - - Raises: - ValueError: The prefixlen_diff is too small or too large. - OR - prefixlen_diff and new_prefix are both set or new_prefix - is a smaller number than the current prefix (smaller - number means a larger network) - - """ - if self._prefixlen == self._max_prefixlen: - yield self - return - - if new_prefix is not None: - if new_prefix < self._prefixlen: - raise ValueError('new prefix must be longer') - if prefixlen_diff != 1: - raise ValueError('cannot set prefixlen_diff and new_prefix') - prefixlen_diff = new_prefix - self._prefixlen - - if prefixlen_diff < 0: - raise ValueError('prefix length diff must be > 0') - new_prefixlen = self._prefixlen + prefixlen_diff - - if new_prefixlen > self._max_prefixlen: - raise ValueError( - 'prefix length diff %d is invalid for netblock %s' % ( - new_prefixlen, self)) - - start = int(self.network_address) - end = int(self.broadcast_address) + 1 - step = (int(self.hostmask) + 1) >> prefixlen_diff - for new_addr in _compat_range(start, end, step): - current = self.__class__((new_addr, new_prefixlen)) - yield current - - def supernet(self, prefixlen_diff=1, new_prefix=None): - """The supernet containing the current network. - - Args: - prefixlen_diff: An integer, the amount the prefix length of - the network should be decreased by. For example, given a - /24 network and a prefixlen_diff of 3, a supernet with a - /21 netmask is returned. - - Returns: - An IPv4 network object. - - Raises: - ValueError: If self.prefixlen - prefixlen_diff < 0. I.e., you have - a negative prefix length. - OR - If prefixlen_diff and new_prefix are both set or new_prefix is a - larger number than the current prefix (larger number means a - smaller network) - - """ - if self._prefixlen == 0: - return self - - if new_prefix is not None: - if new_prefix > self._prefixlen: - raise ValueError('new prefix must be shorter') - if prefixlen_diff != 1: - raise ValueError('cannot set prefixlen_diff and new_prefix') - prefixlen_diff = self._prefixlen - new_prefix - - new_prefixlen = self.prefixlen - prefixlen_diff - if new_prefixlen < 0: - raise ValueError( - 'current prefixlen is %d, cannot have a prefixlen_diff of %d' % - (self.prefixlen, prefixlen_diff)) - return self.__class__(( - int(self.network_address) & (int(self.netmask) << prefixlen_diff), - new_prefixlen - )) - - @property - def is_multicast(self): - """Test if the address is reserved for multicast use. - - Returns: - A boolean, True if the address is a multicast address. - See RFC 2373 2.7 for details. - - """ - return (self.network_address.is_multicast and - self.broadcast_address.is_multicast) - - def subnet_of(self, other): - # always false if one is v4 and the other is v6. - if self._version != other._version: - return False - # dealing with another network. - if (hasattr(other, 'network_address') and - hasattr(other, 'broadcast_address')): - return (other.network_address <= self.network_address and - other.broadcast_address >= self.broadcast_address) - # dealing with another address - else: - raise TypeError('Unable to test subnet containment with element ' - 'of type %s' % type(other)) - - def supernet_of(self, other): - # always false if one is v4 and the other is v6. - if self._version != other._version: - return False - # dealing with another network. - if (hasattr(other, 'network_address') and - hasattr(other, 'broadcast_address')): - return (other.network_address >= self.network_address and - other.broadcast_address <= self.broadcast_address) - # dealing with another address - else: - raise TypeError('Unable to test subnet containment with element ' - 'of type %s' % type(other)) - - @property - def is_reserved(self): - """Test if the address is otherwise IETF reserved. - - Returns: - A boolean, True if the address is within one of the - reserved IPv6 Network ranges. - - """ - return (self.network_address.is_reserved and - self.broadcast_address.is_reserved) - - @property - def is_link_local(self): - """Test if the address is reserved for link-local. - - Returns: - A boolean, True if the address is reserved per RFC 4291. - - """ - return (self.network_address.is_link_local and - self.broadcast_address.is_link_local) - - @property - def is_private(self): - """Test if this address is allocated for private networks. - - Returns: - A boolean, True if the address is reserved per - iana-ipv4-special-registry or iana-ipv6-special-registry. - - """ - return (self.network_address.is_private and - self.broadcast_address.is_private) - - @property - def is_global(self): - """Test if this address is allocated for public networks. - - Returns: - A boolean, True if the address is not reserved per - iana-ipv4-special-registry or iana-ipv6-special-registry. - - """ - return not self.is_private - - @property - def is_unspecified(self): - """Test if the address is unspecified. - - Returns: - A boolean, True if this is the unspecified address as defined in - RFC 2373 2.5.2. - - """ - return (self.network_address.is_unspecified and - self.broadcast_address.is_unspecified) - - @property - def is_loopback(self): - """Test if the address is a loopback address. - - Returns: - A boolean, True if the address is a loopback address as defined in - RFC 2373 2.5.3. - - """ - return (self.network_address.is_loopback and - self.broadcast_address.is_loopback) - - -class _BaseV4(object): - - """Base IPv4 object. - - The following methods are used by IPv4 objects in both single IP - addresses and networks. - - """ - - __slots__ = () - _version = 4 - # Equivalent to 255.255.255.255 or 32 bits of 1's. - _ALL_ONES = (2 ** IPV4LENGTH) - 1 - _DECIMAL_DIGITS = frozenset('0123456789') - - # the valid octets for host and netmasks. only useful for IPv4. - _valid_mask_octets = frozenset([255, 254, 252, 248, 240, 224, 192, 128, 0]) - - _max_prefixlen = IPV4LENGTH - # There are only a handful of valid v4 netmasks, so we cache them all - # when constructed (see _make_netmask()). - _netmask_cache = {} - - def _explode_shorthand_ip_string(self): - return _compat_str(self) - - @classmethod - def _make_netmask(cls, arg): - """Make a (netmask, prefix_len) tuple from the given argument. - - Argument can be: - - an integer (the prefix length) - - a string representing the prefix length (e.g. "24") - - a string representing the prefix netmask (e.g. "255.255.255.0") - """ - if arg not in cls._netmask_cache: - if isinstance(arg, _compat_int_types): - prefixlen = arg - else: - try: - # Check for a netmask in prefix length form - prefixlen = cls._prefix_from_prefix_string(arg) - except NetmaskValueError: - # Check for a netmask or hostmask in dotted-quad form. - # This may raise NetmaskValueError. - prefixlen = cls._prefix_from_ip_string(arg) - netmask = IPv4Address(cls._ip_int_from_prefix(prefixlen)) - cls._netmask_cache[arg] = netmask, prefixlen - return cls._netmask_cache[arg] - - @classmethod - def _ip_int_from_string(cls, ip_str): - """Turn the given IP string into an integer for comparison. - - Args: - ip_str: A string, the IP ip_str. - - Returns: - The IP ip_str as an integer. - - Raises: - AddressValueError: if ip_str isn't a valid IPv4 Address. - - """ - if not ip_str: - raise AddressValueError('Address cannot be empty') - - octets = ip_str.split('.') - if len(octets) != 4: - raise AddressValueError("Expected 4 octets in %r" % ip_str) - - try: - return _compat_int_from_byte_vals( - map(cls._parse_octet, octets), 'big') - except ValueError as exc: - raise AddressValueError("%s in %r" % (exc, ip_str)) - - @classmethod - def _parse_octet(cls, octet_str): - """Convert a decimal octet into an integer. - - Args: - octet_str: A string, the number to parse. - - Returns: - The octet as an integer. - - Raises: - ValueError: if the octet isn't strictly a decimal from [0..255]. - - """ - if not octet_str: - raise ValueError("Empty octet not permitted") - # Whitelist the characters, since int() allows a lot of bizarre stuff. - if not cls._DECIMAL_DIGITS.issuperset(octet_str): - msg = "Only decimal digits permitted in %r" - raise ValueError(msg % octet_str) - # We do the length check second, since the invalid character error - # is likely to be more informative for the user - if len(octet_str) > 3: - msg = "At most 3 characters permitted in %r" - raise ValueError(msg % octet_str) - # Convert to integer (we know digits are legal) - octet_int = int(octet_str, 10) - # Any octets that look like they *might* be written in octal, - # and which don't look exactly the same in both octal and - # decimal are rejected as ambiguous - if octet_int > 7 and octet_str[0] == '0': - msg = "Ambiguous (octal/decimal) value in %r not permitted" - raise ValueError(msg % octet_str) - if octet_int > 255: - raise ValueError("Octet %d (> 255) not permitted" % octet_int) - return octet_int - - @classmethod - def _string_from_ip_int(cls, ip_int): - """Turns a 32-bit integer into dotted decimal notation. - - Args: - ip_int: An integer, the IP address. - - Returns: - The IP address as a string in dotted decimal notation. - - """ - return '.'.join(_compat_str(struct.unpack(b'!B', b)[0] - if isinstance(b, bytes) - else b) - for b in _compat_to_bytes(ip_int, 4, 'big')) - - def _is_hostmask(self, ip_str): - """Test if the IP string is a hostmask (rather than a netmask). - - Args: - ip_str: A string, the potential hostmask. - - Returns: - A boolean, True if the IP string is a hostmask. - - """ - bits = ip_str.split('.') - try: - parts = [x for x in map(int, bits) if x in self._valid_mask_octets] - except ValueError: - return False - if len(parts) != len(bits): - return False - if parts[0] < parts[-1]: - return True - return False - - def _reverse_pointer(self): - """Return the reverse DNS pointer name for the IPv4 address. - - This implements the method described in RFC1035 3.5. - - """ - reverse_octets = _compat_str(self).split('.')[::-1] - return '.'.join(reverse_octets) + '.in-addr.arpa' - - @property - def max_prefixlen(self): - return self._max_prefixlen - - @property - def version(self): - return self._version - - -class IPv4Address(_BaseV4, _BaseAddress): - - """Represent and manipulate single IPv4 Addresses.""" - - __slots__ = ('_ip', '__weakref__') - - def __init__(self, address): - - """ - Args: - address: A string or integer representing the IP - - Additionally, an integer can be passed, so - IPv4Address('192.0.2.1') == IPv4Address(3221225985). - or, more generally - IPv4Address(int(IPv4Address('192.0.2.1'))) == - IPv4Address('192.0.2.1') - - Raises: - AddressValueError: If ipaddress isn't a valid IPv4 address. - - """ - # Efficient constructor from integer. - if isinstance(address, _compat_int_types): - self._check_int_address(address) - self._ip = address - return - - # Constructing from a packed address - if isinstance(address, bytes): - self._check_packed_address(address, 4) - bvs = _compat_bytes_to_byte_vals(address) - self._ip = _compat_int_from_byte_vals(bvs, 'big') - return - - # Assume input argument to be string or any object representation - # which converts into a formatted IP string. - addr_str = _compat_str(address) - if '/' in addr_str: - raise AddressValueError("Unexpected '/' in %r" % address) - self._ip = self._ip_int_from_string(addr_str) - - @property - def packed(self): - """The binary representation of this address.""" - return v4_int_to_packed(self._ip) - - @property - def is_reserved(self): - """Test if the address is otherwise IETF reserved. - - Returns: - A boolean, True if the address is within the - reserved IPv4 Network range. - - """ - return self in self._constants._reserved_network - - @property - def is_private(self): - """Test if this address is allocated for private networks. - - Returns: - A boolean, True if the address is reserved per - iana-ipv4-special-registry. - - """ - return any(self in net for net in self._constants._private_networks) - - @property - def is_global(self): - return ( - self not in self._constants._public_network and - not self.is_private) - - @property - def is_multicast(self): - """Test if the address is reserved for multicast use. - - Returns: - A boolean, True if the address is multicast. - See RFC 3171 for details. - - """ - return self in self._constants._multicast_network - - @property - def is_unspecified(self): - """Test if the address is unspecified. - - Returns: - A boolean, True if this is the unspecified address as defined in - RFC 5735 3. - - """ - return self == self._constants._unspecified_address - - @property - def is_loopback(self): - """Test if the address is a loopback address. - - Returns: - A boolean, True if the address is a loopback per RFC 3330. - - """ - return self in self._constants._loopback_network - - @property - def is_link_local(self): - """Test if the address is reserved for link-local. - - Returns: - A boolean, True if the address is link-local per RFC 3927. - - """ - return self in self._constants._linklocal_network - - -class IPv4Interface(IPv4Address): - - def __init__(self, address): - if isinstance(address, (bytes, _compat_int_types)): - IPv4Address.__init__(self, address) - self.network = IPv4Network(self._ip) - self._prefixlen = self._max_prefixlen - return - - if isinstance(address, tuple): - IPv4Address.__init__(self, address[0]) - if len(address) > 1: - self._prefixlen = int(address[1]) - else: - self._prefixlen = self._max_prefixlen - - self.network = IPv4Network(address, strict=False) - self.netmask = self.network.netmask - self.hostmask = self.network.hostmask - return - - addr = _split_optional_netmask(address) - IPv4Address.__init__(self, addr[0]) - - self.network = IPv4Network(address, strict=False) - self._prefixlen = self.network._prefixlen - - self.netmask = self.network.netmask - self.hostmask = self.network.hostmask - - def __str__(self): - return '%s/%d' % (self._string_from_ip_int(self._ip), - self.network.prefixlen) - - def __eq__(self, other): - address_equal = IPv4Address.__eq__(self, other) - if not address_equal or address_equal is NotImplemented: - return address_equal - try: - return self.network == other.network - except AttributeError: - # An interface with an associated network is NOT the - # same as an unassociated address. That's why the hash - # takes the extra info into account. - return False - - def __lt__(self, other): - address_less = IPv4Address.__lt__(self, other) - if address_less is NotImplemented: - return NotImplemented - try: - return self.network < other.network - except AttributeError: - # We *do* allow addresses and interfaces to be sorted. The - # unassociated address is considered less than all interfaces. - return False - - def __hash__(self): - return self._ip ^ self._prefixlen ^ int(self.network.network_address) - - __reduce__ = _IPAddressBase.__reduce__ - - @property - def ip(self): - return IPv4Address(self._ip) - - @property - def with_prefixlen(self): - return '%s/%s' % (self._string_from_ip_int(self._ip), - self._prefixlen) - - @property - def with_netmask(self): - return '%s/%s' % (self._string_from_ip_int(self._ip), - self.netmask) - - @property - def with_hostmask(self): - return '%s/%s' % (self._string_from_ip_int(self._ip), - self.hostmask) - - -class IPv4Network(_BaseV4, _BaseNetwork): - - """This class represents and manipulates 32-bit IPv4 network + addresses.. - - Attributes: [examples for IPv4Network('192.0.2.0/27')] - .network_address: IPv4Address('192.0.2.0') - .hostmask: IPv4Address('0.0.0.31') - .broadcast_address: IPv4Address('192.0.2.32') - .netmask: IPv4Address('255.255.255.224') - .prefixlen: 27 - - """ - # Class to use when creating address objects - _address_class = IPv4Address - - def __init__(self, address, strict=True): - - """Instantiate a new IPv4 network object. - - Args: - address: A string or integer representing the IP [& network]. - '192.0.2.0/24' - '192.0.2.0/255.255.255.0' - '192.0.0.2/0.0.0.255' - are all functionally the same in IPv4. Similarly, - '192.0.2.1' - '192.0.2.1/255.255.255.255' - '192.0.2.1/32' - are also functionally equivalent. That is to say, failing to - provide a subnetmask will create an object with a mask of /32. - - If the mask (portion after the / in the argument) is given in - dotted quad form, it is treated as a netmask if it starts with a - non-zero field (e.g. /255.0.0.0 == /8) and as a hostmask if it - starts with a zero field (e.g. 0.255.255.255 == /8), with the - single exception of an all-zero mask which is treated as a - netmask == /0. If no mask is given, a default of /32 is used. - - Additionally, an integer can be passed, so - IPv4Network('192.0.2.1') == IPv4Network(3221225985) - or, more generally - IPv4Interface(int(IPv4Interface('192.0.2.1'))) == - IPv4Interface('192.0.2.1') - - Raises: - AddressValueError: If ipaddress isn't a valid IPv4 address. - NetmaskValueError: If the netmask isn't valid for - an IPv4 address. - ValueError: If strict is True and a network address is not - supplied. - - """ - _BaseNetwork.__init__(self, address) - - # Constructing from a packed address or integer - if isinstance(address, (_compat_int_types, bytes)): - self.network_address = IPv4Address(address) - self.netmask, self._prefixlen = self._make_netmask( - self._max_prefixlen) - # fixme: address/network test here. - return - - if isinstance(address, tuple): - if len(address) > 1: - arg = address[1] - else: - # We weren't given an address[1] - arg = self._max_prefixlen - self.network_address = IPv4Address(address[0]) - self.netmask, self._prefixlen = self._make_netmask(arg) - packed = int(self.network_address) - if packed & int(self.netmask) != packed: - if strict: - raise ValueError('%s has host bits set' % self) - else: - self.network_address = IPv4Address(packed & - int(self.netmask)) - return - - # Assume input argument to be string or any object representation - # which converts into a formatted IP prefix string. - addr = _split_optional_netmask(address) - self.network_address = IPv4Address(self._ip_int_from_string(addr[0])) - - if len(addr) == 2: - arg = addr[1] - else: - arg = self._max_prefixlen - self.netmask, self._prefixlen = self._make_netmask(arg) - - if strict: - if (IPv4Address(int(self.network_address) & int(self.netmask)) != - self.network_address): - raise ValueError('%s has host bits set' % self) - self.network_address = IPv4Address(int(self.network_address) & - int(self.netmask)) - - if self._prefixlen == (self._max_prefixlen - 1): - self.hosts = self.__iter__ - - @property - def is_global(self): - """Test if this address is allocated for public networks. - - Returns: - A boolean, True if the address is not reserved per - iana-ipv4-special-registry. - - """ - return (not (self.network_address in IPv4Network('100.64.0.0/10') and - self.broadcast_address in IPv4Network('100.64.0.0/10')) and - not self.is_private) - - -class _IPv4Constants(object): - - _linklocal_network = IPv4Network('169.254.0.0/16') - - _loopback_network = IPv4Network('127.0.0.0/8') - - _multicast_network = IPv4Network('224.0.0.0/4') - - _public_network = IPv4Network('100.64.0.0/10') - - _private_networks = [ - IPv4Network('0.0.0.0/8'), - IPv4Network('10.0.0.0/8'), - IPv4Network('127.0.0.0/8'), - IPv4Network('169.254.0.0/16'), - IPv4Network('172.16.0.0/12'), - IPv4Network('192.0.0.0/29'), - IPv4Network('192.0.0.170/31'), - IPv4Network('192.0.2.0/24'), - IPv4Network('192.168.0.0/16'), - IPv4Network('198.18.0.0/15'), - IPv4Network('198.51.100.0/24'), - IPv4Network('203.0.113.0/24'), - IPv4Network('240.0.0.0/4'), - IPv4Network('255.255.255.255/32'), - ] - - _reserved_network = IPv4Network('240.0.0.0/4') - - _unspecified_address = IPv4Address('0.0.0.0') - - -IPv4Address._constants = _IPv4Constants - - -class _BaseV6(object): - - """Base IPv6 object. - - The following methods are used by IPv6 objects in both single IP - addresses and networks. - - """ - - __slots__ = () - _version = 6 - _ALL_ONES = (2 ** IPV6LENGTH) - 1 - _HEXTET_COUNT = 8 - _HEX_DIGITS = frozenset('0123456789ABCDEFabcdef') - _max_prefixlen = IPV6LENGTH - - # There are only a bunch of valid v6 netmasks, so we cache them all - # when constructed (see _make_netmask()). - _netmask_cache = {} - - @classmethod - def _make_netmask(cls, arg): - """Make a (netmask, prefix_len) tuple from the given argument. - - Argument can be: - - an integer (the prefix length) - - a string representing the prefix length (e.g. "24") - - a string representing the prefix netmask (e.g. "255.255.255.0") - """ - if arg not in cls._netmask_cache: - if isinstance(arg, _compat_int_types): - prefixlen = arg - else: - prefixlen = cls._prefix_from_prefix_string(arg) - netmask = IPv6Address(cls._ip_int_from_prefix(prefixlen)) - cls._netmask_cache[arg] = netmask, prefixlen - return cls._netmask_cache[arg] - - @classmethod - def _ip_int_from_string(cls, ip_str): - """Turn an IPv6 ip_str into an integer. - - Args: - ip_str: A string, the IPv6 ip_str. - - Returns: - An int, the IPv6 address - - Raises: - AddressValueError: if ip_str isn't a valid IPv6 Address. - - """ - if not ip_str: - raise AddressValueError('Address cannot be empty') - - parts = ip_str.split(':') - - # An IPv6 address needs at least 2 colons (3 parts). - _min_parts = 3 - if len(parts) < _min_parts: - msg = "At least %d parts expected in %r" % (_min_parts, ip_str) - raise AddressValueError(msg) - - # If the address has an IPv4-style suffix, convert it to hexadecimal. - if '.' in parts[-1]: - try: - ipv4_int = IPv4Address(parts.pop())._ip - except AddressValueError as exc: - raise AddressValueError("%s in %r" % (exc, ip_str)) - parts.append('%x' % ((ipv4_int >> 16) & 0xFFFF)) - parts.append('%x' % (ipv4_int & 0xFFFF)) - - # An IPv6 address can't have more than 8 colons (9 parts). - # The extra colon comes from using the "::" notation for a single - # leading or trailing zero part. - _max_parts = cls._HEXTET_COUNT + 1 - if len(parts) > _max_parts: - msg = "At most %d colons permitted in %r" % ( - _max_parts - 1, ip_str) - raise AddressValueError(msg) - - # Disregarding the endpoints, find '::' with nothing in between. - # This indicates that a run of zeroes has been skipped. - skip_index = None - for i in _compat_range(1, len(parts) - 1): - if not parts[i]: - if skip_index is not None: - # Can't have more than one '::' - msg = "At most one '::' permitted in %r" % ip_str - raise AddressValueError(msg) - skip_index = i - - # parts_hi is the number of parts to copy from above/before the '::' - # parts_lo is the number of parts to copy from below/after the '::' - if skip_index is not None: - # If we found a '::', then check if it also covers the endpoints. - parts_hi = skip_index - parts_lo = len(parts) - skip_index - 1 - if not parts[0]: - parts_hi -= 1 - if parts_hi: - msg = "Leading ':' only permitted as part of '::' in %r" - raise AddressValueError(msg % ip_str) # ^: requires ^:: - if not parts[-1]: - parts_lo -= 1 - if parts_lo: - msg = "Trailing ':' only permitted as part of '::' in %r" - raise AddressValueError(msg % ip_str) # :$ requires ::$ - parts_skipped = cls._HEXTET_COUNT - (parts_hi + parts_lo) - if parts_skipped < 1: - msg = "Expected at most %d other parts with '::' in %r" - raise AddressValueError(msg % (cls._HEXTET_COUNT - 1, ip_str)) - else: - # Otherwise, allocate the entire address to parts_hi. The - # endpoints could still be empty, but _parse_hextet() will check - # for that. - if len(parts) != cls._HEXTET_COUNT: - msg = "Exactly %d parts expected without '::' in %r" - raise AddressValueError(msg % (cls._HEXTET_COUNT, ip_str)) - if not parts[0]: - msg = "Leading ':' only permitted as part of '::' in %r" - raise AddressValueError(msg % ip_str) # ^: requires ^:: - if not parts[-1]: - msg = "Trailing ':' only permitted as part of '::' in %r" - raise AddressValueError(msg % ip_str) # :$ requires ::$ - parts_hi = len(parts) - parts_lo = 0 - parts_skipped = 0 - - try: - # Now, parse the hextets into a 128-bit integer. - ip_int = 0 - for i in range(parts_hi): - ip_int <<= 16 - ip_int |= cls._parse_hextet(parts[i]) - ip_int <<= 16 * parts_skipped - for i in range(-parts_lo, 0): - ip_int <<= 16 - ip_int |= cls._parse_hextet(parts[i]) - return ip_int - except ValueError as exc: - raise AddressValueError("%s in %r" % (exc, ip_str)) - - @classmethod - def _parse_hextet(cls, hextet_str): - """Convert an IPv6 hextet string into an integer. - - Args: - hextet_str: A string, the number to parse. - - Returns: - The hextet as an integer. - - Raises: - ValueError: if the input isn't strictly a hex number from - [0..FFFF]. - - """ - # Whitelist the characters, since int() allows a lot of bizarre stuff. - if not cls._HEX_DIGITS.issuperset(hextet_str): - raise ValueError("Only hex digits permitted in %r" % hextet_str) - # We do the length check second, since the invalid character error - # is likely to be more informative for the user - if len(hextet_str) > 4: - msg = "At most 4 characters permitted in %r" - raise ValueError(msg % hextet_str) - # Length check means we can skip checking the integer value - return int(hextet_str, 16) - - @classmethod - def _compress_hextets(cls, hextets): - """Compresses a list of hextets. - - Compresses a list of strings, replacing the longest continuous - sequence of "0" in the list with "" and adding empty strings at - the beginning or at the end of the string such that subsequently - calling ":".join(hextets) will produce the compressed version of - the IPv6 address. - - Args: - hextets: A list of strings, the hextets to compress. - - Returns: - A list of strings. - - """ - best_doublecolon_start = -1 - best_doublecolon_len = 0 - doublecolon_start = -1 - doublecolon_len = 0 - for index, hextet in enumerate(hextets): - if hextet == '0': - doublecolon_len += 1 - if doublecolon_start == -1: - # Start of a sequence of zeros. - doublecolon_start = index - if doublecolon_len > best_doublecolon_len: - # This is the longest sequence of zeros so far. - best_doublecolon_len = doublecolon_len - best_doublecolon_start = doublecolon_start - else: - doublecolon_len = 0 - doublecolon_start = -1 - - if best_doublecolon_len > 1: - best_doublecolon_end = (best_doublecolon_start + - best_doublecolon_len) - # For zeros at the end of the address. - if best_doublecolon_end == len(hextets): - hextets += [''] - hextets[best_doublecolon_start:best_doublecolon_end] = [''] - # For zeros at the beginning of the address. - if best_doublecolon_start == 0: - hextets = [''] + hextets - - return hextets - - @classmethod - def _string_from_ip_int(cls, ip_int=None): - """Turns a 128-bit integer into hexadecimal notation. - - Args: - ip_int: An integer, the IP address. - - Returns: - A string, the hexadecimal representation of the address. - - Raises: - ValueError: The address is bigger than 128 bits of all ones. - - """ - if ip_int is None: - ip_int = int(cls._ip) - - if ip_int > cls._ALL_ONES: - raise ValueError('IPv6 address is too large') - - hex_str = '%032x' % ip_int - hextets = ['%x' % int(hex_str[x:x + 4], 16) for x in range(0, 32, 4)] - - hextets = cls._compress_hextets(hextets) - return ':'.join(hextets) - - def _explode_shorthand_ip_string(self): - """Expand a shortened IPv6 address. - - Args: - ip_str: A string, the IPv6 address. - - Returns: - A string, the expanded IPv6 address. - - """ - if isinstance(self, IPv6Network): - ip_str = _compat_str(self.network_address) - elif isinstance(self, IPv6Interface): - ip_str = _compat_str(self.ip) - else: - ip_str = _compat_str(self) - - ip_int = self._ip_int_from_string(ip_str) - hex_str = '%032x' % ip_int - parts = [hex_str[x:x + 4] for x in range(0, 32, 4)] - if isinstance(self, (_BaseNetwork, IPv6Interface)): - return '%s/%d' % (':'.join(parts), self._prefixlen) - return ':'.join(parts) - - def _reverse_pointer(self): - """Return the reverse DNS pointer name for the IPv6 address. - - This implements the method described in RFC3596 2.5. - - """ - reverse_chars = self.exploded[::-1].replace(':', '') - return '.'.join(reverse_chars) + '.ip6.arpa' - - @property - def max_prefixlen(self): - return self._max_prefixlen - - @property - def version(self): - return self._version - - -class IPv6Address(_BaseV6, _BaseAddress): - - """Represent and manipulate single IPv6 Addresses.""" - - __slots__ = ('_ip', '__weakref__') - - def __init__(self, address): - """Instantiate a new IPv6 address object. - - Args: - address: A string or integer representing the IP - - Additionally, an integer can be passed, so - IPv6Address('2001:db8::') == - IPv6Address(42540766411282592856903984951653826560) - or, more generally - IPv6Address(int(IPv6Address('2001:db8::'))) == - IPv6Address('2001:db8::') - - Raises: - AddressValueError: If address isn't a valid IPv6 address. - - """ - # Efficient constructor from integer. - if isinstance(address, _compat_int_types): - self._check_int_address(address) - self._ip = address - return - - # Constructing from a packed address - if isinstance(address, bytes): - self._check_packed_address(address, 16) - bvs = _compat_bytes_to_byte_vals(address) - self._ip = _compat_int_from_byte_vals(bvs, 'big') - return - - # Assume input argument to be string or any object representation - # which converts into a formatted IP string. - addr_str = _compat_str(address) - if '/' in addr_str: - raise AddressValueError("Unexpected '/' in %r" % address) - self._ip = self._ip_int_from_string(addr_str) - - @property - def packed(self): - """The binary representation of this address.""" - return v6_int_to_packed(self._ip) - - @property - def is_multicast(self): - """Test if the address is reserved for multicast use. - - Returns: - A boolean, True if the address is a multicast address. - See RFC 2373 2.7 for details. - - """ - return self in self._constants._multicast_network - - @property - def is_reserved(self): - """Test if the address is otherwise IETF reserved. - - Returns: - A boolean, True if the address is within one of the - reserved IPv6 Network ranges. - - """ - return any(self in x for x in self._constants._reserved_networks) - - @property - def is_link_local(self): - """Test if the address is reserved for link-local. - - Returns: - A boolean, True if the address is reserved per RFC 4291. - - """ - return self in self._constants._linklocal_network - - @property - def is_site_local(self): - """Test if the address is reserved for site-local. - - Note that the site-local address space has been deprecated by RFC 3879. - Use is_private to test if this address is in the space of unique local - addresses as defined by RFC 4193. - - Returns: - A boolean, True if the address is reserved per RFC 3513 2.5.6. - - """ - return self in self._constants._sitelocal_network - - @property - def is_private(self): - """Test if this address is allocated for private networks. - - Returns: - A boolean, True if the address is reserved per - iana-ipv6-special-registry. - - """ - return any(self in net for net in self._constants._private_networks) - - @property - def is_global(self): - """Test if this address is allocated for public networks. - - Returns: - A boolean, true if the address is not reserved per - iana-ipv6-special-registry. - - """ - return not self.is_private - - @property - def is_unspecified(self): - """Test if the address is unspecified. - - Returns: - A boolean, True if this is the unspecified address as defined in - RFC 2373 2.5.2. - - """ - return self._ip == 0 - - @property - def is_loopback(self): - """Test if the address is a loopback address. - - Returns: - A boolean, True if the address is a loopback address as defined in - RFC 2373 2.5.3. - - """ - return self._ip == 1 - - @property - def ipv4_mapped(self): - """Return the IPv4 mapped address. - - Returns: - If the IPv6 address is a v4 mapped address, return the - IPv4 mapped address. Return None otherwise. - - """ - if (self._ip >> 32) != 0xFFFF: - return None - return IPv4Address(self._ip & 0xFFFFFFFF) - - @property - def teredo(self): - """Tuple of embedded teredo IPs. - - Returns: - Tuple of the (server, client) IPs or None if the address - doesn't appear to be a teredo address (doesn't start with - 2001::/32) - - """ - if (self._ip >> 96) != 0x20010000: - return None - return (IPv4Address((self._ip >> 64) & 0xFFFFFFFF), - IPv4Address(~self._ip & 0xFFFFFFFF)) - - @property - def sixtofour(self): - """Return the IPv4 6to4 embedded address. - - Returns: - The IPv4 6to4-embedded address if present or None if the - address doesn't appear to contain a 6to4 embedded address. - - """ - if (self._ip >> 112) != 0x2002: - return None - return IPv4Address((self._ip >> 80) & 0xFFFFFFFF) - - -class IPv6Interface(IPv6Address): - - def __init__(self, address): - if isinstance(address, (bytes, _compat_int_types)): - IPv6Address.__init__(self, address) - self.network = IPv6Network(self._ip) - self._prefixlen = self._max_prefixlen - return - if isinstance(address, tuple): - IPv6Address.__init__(self, address[0]) - if len(address) > 1: - self._prefixlen = int(address[1]) - else: - self._prefixlen = self._max_prefixlen - self.network = IPv6Network(address, strict=False) - self.netmask = self.network.netmask - self.hostmask = self.network.hostmask - return - - addr = _split_optional_netmask(address) - IPv6Address.__init__(self, addr[0]) - self.network = IPv6Network(address, strict=False) - self.netmask = self.network.netmask - self._prefixlen = self.network._prefixlen - self.hostmask = self.network.hostmask - - def __str__(self): - return '%s/%d' % (self._string_from_ip_int(self._ip), - self.network.prefixlen) - - def __eq__(self, other): - address_equal = IPv6Address.__eq__(self, other) - if not address_equal or address_equal is NotImplemented: - return address_equal - try: - return self.network == other.network - except AttributeError: - # An interface with an associated network is NOT the - # same as an unassociated address. That's why the hash - # takes the extra info into account. - return False - - def __lt__(self, other): - address_less = IPv6Address.__lt__(self, other) - if address_less is NotImplemented: - return NotImplemented - try: - return self.network < other.network - except AttributeError: - # We *do* allow addresses and interfaces to be sorted. The - # unassociated address is considered less than all interfaces. - return False - - def __hash__(self): - return self._ip ^ self._prefixlen ^ int(self.network.network_address) - - __reduce__ = _IPAddressBase.__reduce__ - - @property - def ip(self): - return IPv6Address(self._ip) - - @property - def with_prefixlen(self): - return '%s/%s' % (self._string_from_ip_int(self._ip), - self._prefixlen) - - @property - def with_netmask(self): - return '%s/%s' % (self._string_from_ip_int(self._ip), - self.netmask) - - @property - def with_hostmask(self): - return '%s/%s' % (self._string_from_ip_int(self._ip), - self.hostmask) - - @property - def is_unspecified(self): - return self._ip == 0 and self.network.is_unspecified - - @property - def is_loopback(self): - return self._ip == 1 and self.network.is_loopback - - -class IPv6Network(_BaseV6, _BaseNetwork): - - """This class represents and manipulates 128-bit IPv6 networks. - - Attributes: [examples for IPv6('2001:db8::1000/124')] - .network_address: IPv6Address('2001:db8::1000') - .hostmask: IPv6Address('::f') - .broadcast_address: IPv6Address('2001:db8::100f') - .netmask: IPv6Address('ffff:ffff:ffff:ffff:ffff:ffff:ffff:fff0') - .prefixlen: 124 - - """ - - # Class to use when creating address objects - _address_class = IPv6Address - - def __init__(self, address, strict=True): - """Instantiate a new IPv6 Network object. - - Args: - address: A string or integer representing the IPv6 network or the - IP and prefix/netmask. - '2001:db8::/128' - '2001:db8:0000:0000:0000:0000:0000:0000/128' - '2001:db8::' - are all functionally the same in IPv6. That is to say, - failing to provide a subnetmask will create an object with - a mask of /128. - - Additionally, an integer can be passed, so - IPv6Network('2001:db8::') == - IPv6Network(42540766411282592856903984951653826560) - or, more generally - IPv6Network(int(IPv6Network('2001:db8::'))) == - IPv6Network('2001:db8::') - - strict: A boolean. If true, ensure that we have been passed - A true network address, eg, 2001:db8::1000/124 and not an - IP address on a network, eg, 2001:db8::1/124. - - Raises: - AddressValueError: If address isn't a valid IPv6 address. - NetmaskValueError: If the netmask isn't valid for - an IPv6 address. - ValueError: If strict was True and a network address was not - supplied. - - """ - _BaseNetwork.__init__(self, address) - - # Efficient constructor from integer or packed address - if isinstance(address, (bytes, _compat_int_types)): - self.network_address = IPv6Address(address) - self.netmask, self._prefixlen = self._make_netmask( - self._max_prefixlen) - return - - if isinstance(address, tuple): - if len(address) > 1: - arg = address[1] - else: - arg = self._max_prefixlen - self.netmask, self._prefixlen = self._make_netmask(arg) - self.network_address = IPv6Address(address[0]) - packed = int(self.network_address) - if packed & int(self.netmask) != packed: - if strict: - raise ValueError('%s has host bits set' % self) - else: - self.network_address = IPv6Address(packed & - int(self.netmask)) - return - - # Assume input argument to be string or any object representation - # which converts into a formatted IP prefix string. - addr = _split_optional_netmask(address) - - self.network_address = IPv6Address(self._ip_int_from_string(addr[0])) - - if len(addr) == 2: - arg = addr[1] - else: - arg = self._max_prefixlen - self.netmask, self._prefixlen = self._make_netmask(arg) - - if strict: - if (IPv6Address(int(self.network_address) & int(self.netmask)) != - self.network_address): - raise ValueError('%s has host bits set' % self) - self.network_address = IPv6Address(int(self.network_address) & - int(self.netmask)) - - if self._prefixlen == (self._max_prefixlen - 1): - self.hosts = self.__iter__ - - def hosts(self): - """Generate Iterator over usable hosts in a network. - - This is like __iter__ except it doesn't return the - Subnet-Router anycast address. - - """ - network = int(self.network_address) - broadcast = int(self.broadcast_address) - for x in _compat_range(network + 1, broadcast + 1): - yield self._address_class(x) - - @property - def is_site_local(self): - """Test if the address is reserved for site-local. - - Note that the site-local address space has been deprecated by RFC 3879. - Use is_private to test if this address is in the space of unique local - addresses as defined by RFC 4193. - - Returns: - A boolean, True if the address is reserved per RFC 3513 2.5.6. - - """ - return (self.network_address.is_site_local and - self.broadcast_address.is_site_local) - - -class _IPv6Constants(object): - - _linklocal_network = IPv6Network('fe80::/10') - - _multicast_network = IPv6Network('ff00::/8') - - _private_networks = [ - IPv6Network('::1/128'), - IPv6Network('::/128'), - IPv6Network('::ffff:0:0/96'), - IPv6Network('100::/64'), - IPv6Network('2001::/23'), - IPv6Network('2001:2::/48'), - IPv6Network('2001:db8::/32'), - IPv6Network('2001:10::/28'), - IPv6Network('fc00::/7'), - IPv6Network('fe80::/10'), - ] - - _reserved_networks = [ - IPv6Network('::/8'), IPv6Network('100::/8'), - IPv6Network('200::/7'), IPv6Network('400::/6'), - IPv6Network('800::/5'), IPv6Network('1000::/4'), - IPv6Network('4000::/3'), IPv6Network('6000::/3'), - IPv6Network('8000::/3'), IPv6Network('A000::/3'), - IPv6Network('C000::/3'), IPv6Network('E000::/4'), - IPv6Network('F000::/5'), IPv6Network('F800::/6'), - IPv6Network('FE00::/9'), - ] - - _sitelocal_network = IPv6Network('fec0::/10') - - -IPv6Address._constants = _IPv6Constants diff --git a/gluon/html.py b/gluon/html.py index 0a9a58bb..44f4f3eb 100644 --- a/gluon/html.py +++ b/gluon/html.py @@ -25,7 +25,8 @@ import marshal from gluon import decoder from gluon.storage import Storage -from gluon.utils import web2py_uuid, simple_hash, compare +from gluon.validators import simple_hash +from gluon.utils import web2py_uuid, compare from gluon.highlight import highlight diff --git a/gluon/tests/__init__.py b/gluon/tests/__init__.py index 60c0b91c..cdb4b6e5 100644 --- a/gluon/tests/__init__.py +++ b/gluon/tests/__init__.py @@ -22,7 +22,7 @@ from .test_web import * from .test_sqlhtml import * from .test_cron import * from .test_is_url import * -from .test_scheduler import * +# from .test_scheduler import * if sys.version[:3] == '2.7': from .test_old_doctests import * diff --git a/gluon/tests/test_is_url.py b/gluon/tests/test_is_url.py index 51a204d1..babe4edf 100644 --- a/gluon/tests/test_is_url.py +++ b/gluon/tests/test_is_url.py @@ -558,8 +558,7 @@ class TestIsHttpUrl(unittest.TestCase): 'https://google.ca', None)) # prepends https if asked z = IS_HTTP_URL(prepend_scheme=None) - self.assertEqual(z('google.ca:8080'), ('google.ca:8080', - None)) # prepending disabled + self.assertEqual(z('google.ca:8080'), ('google.ca:8080', None)) # prepending disabled try: IS_HTTP_URL(prepend_scheme='mailto') diff --git a/gluon/tests/test_utils.py b/gluon/tests/test_utils.py index 8b94844f..6f6f9d50 100644 --- a/gluon/tests/test_utils.py +++ b/gluon/tests/test_utils.py @@ -9,7 +9,8 @@ from hashlib import md5 from gluon.utils import md5_hash, compare, is_valid_ip_address, web2py_uuid import gluon.utils -from gluon.utils import simple_hash, get_digest, secure_dumps, secure_loads +from gluon.validators import simple_hash, get_digest +from gluon.utils import secure_dumps, secure_loads class TestUtils(unittest.TestCase): diff --git a/gluon/tests/test_validators.py b/gluon/tests/test_validators.py index ad6ec128..90f62c95 100644 --- a/gluon/tests/test_validators.py +++ b/gluon/tests/test_validators.py @@ -187,7 +187,7 @@ class TestValidators(unittest.TestCase): rtn = IS_IN_SET(['id1', 'id2'], error_message='oops', multiple=True)('id1') self.assertEqual(rtn, (['id1'], None)) rtn = IS_IN_SET(['id1', 'id2'], error_message='oops', multiple=(1, 2))(None) - self.assertEqual(rtn, ([], 'oops')) + self.assertEqual(rtn, (None, 'oops')) import itertools rtn = IS_IN_SET(itertools.chain(['1', '3', '5'], ['2', '4', '6']))('1') self.assertEqual(rtn, ('1', None)) @@ -223,13 +223,13 @@ class TestValidators(unittest.TestCase): rtn = IS_IN_DB(db, 'person.id', '%(name)s', multiple=True)([george_id, costanza_id]) self.assertEqual(rtn, ([george_id, costanza_id], None)) rtn = IS_IN_DB(db, 'person.id', '%(name)s', multiple=True, error_message='oops')("I'm not even an id") - self.assertEqual(rtn, (["I'm not even an id"], 'oops')) + self.assertEqual(rtn, ("I'm not even an id", 'oops')) rtn = IS_IN_DB(db, 'person.id', '%(name)s', multiple=True, delimiter=',')('%d,%d' % (george_id, costanza_id)) self.assertEqual(rtn, (('%d,%d' % (george_id, costanza_id)).split(','), None)) rtn = IS_IN_DB(db, 'person.id', '%(name)s', multiple=(1, 3), delimiter=',')('%d,%d' % (george_id, costanza_id)) self.assertEqual(rtn, (('%d,%d' % (george_id, costanza_id)).split(','), None)) rtn = IS_IN_DB(db, 'person.id', '%(name)s', multiple=(1, 2), delimiter=',', error_message='oops')('%d,%d' % (george_id, costanza_id)) - self.assertEqual(rtn, (('%d,%d' % (george_id, costanza_id)).split(','), 'oops')) + self.assertEqual(rtn, (('%d,%d' % (george_id, costanza_id)), 'oops')) rtn = IS_IN_DB(db, db.person.id, '%(name)s', error_message='oops').options(zero=False) self.assertEqual(sorted(rtn), [('%d' % george_id, 'george'), ('%d' % costanza_id, 'costanza')]) rtn = IS_IN_DB(db, db.person.id, db.person.name, error_message='oops', sort=True).options(zero=True) @@ -359,7 +359,7 @@ class TestValidators(unittest.TestCase): rtn = IS_NOT_IN_DB(db(db.person.id > 0), 'person.name')(u'jerry') self.assertEqual(rtn, ('jerry', None)) rtn = IS_NOT_IN_DB(db, db.person, error_message='oops')(1) - self.assertEqual(rtn, ('1', 'oops')) + self.assertEqual(rtn, (1, 'oops')) vldtr = IS_NOT_IN_DB(db, 'person.name', error_message='oops') vldtr.set_self_id({'name': 'costanza', 'nickname': 'T Bone'}) rtn = vldtr('george') @@ -769,7 +769,7 @@ class TestValidators(unittest.TestCase): self.assertEqual(rtn, ([1, 2, 3], 'Maximum length is 2')) # regression test for issue 742 rtn = IS_LIST_OF(minimum=1)('') - self.assertEqual(rtn, ([], 'Minimum length is 1')) + self.assertEqual(rtn, ('', 'Minimum length is 1')) def test_IS_LOWER(self): rtn = IS_LOWER()('ABC') @@ -861,9 +861,9 @@ class TestValidators(unittest.TestCase): rtn = IS_EMPTY_OR(IS_IN_SET([('id1', 'first label'), ('id2', 'second label')], zero='zero')).options() self.assertEqual(rtn, [('', 'zero'), ('id1', 'first label'), ('id2', 'second label')]) rtn = IS_EMPTY_OR((IS_LOWER(), IS_EMAIL()))('AAA') - self.assertEqual(rtn, ('aaa', 'Enter a valid email address')) + self.assertEqual(rtn, ('AAA', 'Enter a valid email address')) rtn = IS_EMPTY_OR([IS_LOWER(), IS_EMAIL()])('AAA') - self.assertEqual(rtn, ('aaa', 'Enter a valid email address')) + self.assertEqual(rtn, ('AAA', 'Enter a valid email address')) def test_CLEANUP(self): rtn = CLEANUP()('helloò') diff --git a/gluon/utils.py b/gluon/utils.py index 70d78dbd..bbaba4e5 100644 --- a/gluon/utils.py +++ b/gluon/utils.py @@ -1,5 +1,6 @@ #!/usr/bin/env python # -*- coding: utf-8 -*- +#pylint: disable=invalid-name,redefined-builtin """ | This file is part of the web2py Web Framework @@ -24,9 +25,7 @@ import socket import base64 import zlib import hashlib -import binascii import hmac -from hashlib import md5, sha1, sha224, sha256, sha384, sha512 from gluon._compat import basestring, pickle, PY2, xrange, to_bytes, to_native _struct_2_long_long = struct.Struct('=QQ') @@ -38,26 +37,6 @@ except ImportError: import gluon.contrib.pyaes as PYAES HAVE_AES = False -if hasattr(hashlib, "pbkdf2_hmac"): - def pbkdf2_hex(data, salt, iterations=1000, keylen=24, hashfunc=None): - hashfunc = hashfunc or sha1 - hmac = hashlib.pbkdf2_hmac(hashfunc().name, to_bytes(data), - to_bytes(salt), iterations, keylen) - return binascii.hexlify(hmac) - HAVE_PBKDF2 = True -else: - try: - try: - from gluon.contrib.pbkdf2_ctypes import pbkdf2_hex - except (ImportError, AttributeError): - from gluon.contrib.pbkdf2 import pbkdf2_hex - HAVE_PBKDF2 = True - except ImportError: - try: - from .pbkdf2 import pbkdf2_hex - HAVE_PBKDF2 = True - except (ImportError, ValueError): - HAVE_PBKDF2 = False HAVE_COMPARE_DIGEST = False if hasattr(hmac, 'compare_digest'): @@ -110,59 +89,7 @@ def compare(a, b): def md5_hash(text): """Generate an md5 hash with the given text.""" - return md5(to_bytes(text)).hexdigest() - - -def simple_hash(text, key='', salt='', digest_alg='md5'): - """Generate hash with the given text using the specified digest algorithm.""" - text = to_bytes(text) - key = to_bytes(key) - salt = to_bytes(salt) - if not digest_alg: - raise RuntimeError("simple_hash with digest_alg=None") - elif not isinstance(digest_alg, str): # manual approach - h = digest_alg(text + key + salt) - elif digest_alg.startswith('pbkdf2'): # latest and coolest! - iterations, keylen, alg = digest_alg[7:-1].split(',') - return to_native(pbkdf2_hex(text, salt, int(iterations), - int(keylen), get_digest(alg))) - elif key: # use hmac - digest_alg = get_digest(digest_alg) - h = hmac.new(key + salt, text, digest_alg) - else: # compatible with third party systems - h = get_digest(digest_alg)() - h.update(text + salt) - return h.hexdigest() - - -def get_digest(value): - """Return a hashlib digest algorithm from a string.""" - if not isinstance(value, str): - return value - value = value.lower() - if value == "md5": - return md5 - elif value == "sha1": - return sha1 - elif value == "sha224": - return sha224 - elif value == "sha256": - return sha256 - elif value == "sha384": - return sha384 - elif value == "sha512": - return sha512 - else: - raise ValueError("Invalid digest algorithm: %s" % value) - -DIGEST_ALG_BY_SIZE = { - 128 // 4: 'md5', - 160 // 4: 'sha1', - 224 // 4: 'sha224', - 256 // 4: 'sha256', - 384 // 4: 'sha384', - 512 // 4: 'sha512', -} + return hashlib.md5(to_bytes(text)).hexdigest() def get_callable_argspec(fn): @@ -178,12 +105,13 @@ def get_callable_argspec(fn): def pad(s, n=32): - # PKCS7v1.5 https://www.ietf.org/rfc/rfc2315.txt + """does padding according to PKCS7v1.5 https://www.ietf.org/rfc/rfc2315.txt""" padlen = n - len(s) % n return s + bytes(bytearray(padlen * [padlen])) def unpad(s, n=32): + """removed padding""" padlen = s[-1] if isinstance(padlen, str): padlen = ord(padlen) # python2 @@ -194,6 +122,7 @@ def unpad(s, n=32): def secure_dumps(data, encryption_key, hash_key=None, compression_level=None): + """dumps data, followed by a signature""" dump = pickle.dumps(data, pickle.HIGHEST_PROTOCOL) if compression_level: dump = zlib.compress(dump, compression_level) @@ -207,6 +136,7 @@ def secure_dumps(data, encryption_key, hash_key=None, compression_level=None): def secure_loads(data, encryption_key, hash_key=None, compression_level=None): + """loads a signed data dump""" data = to_bytes(data) components = data.count(b':') if components == 1: @@ -230,18 +160,20 @@ def secure_loads(data, encryption_key, hash_key=None, compression_level=None): if compression_level: data = zlib.decompress(data) return pickle.loads(data) - except Exception as e: + except Exception: return None def __pad_deprecated(s, n=32, padchar=b' '): - return s + (32 - len(s) % 32) * padchar + """reprecated data, here for backward compatibility""" + return s + (n - len(s) % n) * padchar def secure_dumps_deprecated(data, encryption_key, hash_key=None, compression_level=None): + """dumps data with a signature (deprecated because of incorrect padding)""" encryption_key = to_bytes(encryption_key) if not hash_key: - hash_key = sha1(encryption_key).hexdigest() + hash_key = hashlib.sha1(encryption_key).hexdigest() dump = pickle.dumps(data, pickle.HIGHEST_PROTOCOL) if compression_level: dump = zlib.compress(dump, compression_level) @@ -253,12 +185,13 @@ def secure_dumps_deprecated(data, encryption_key, hash_key=None, compression_lev def secure_loads_deprecated(data, encryption_key, hash_key=None, compression_level=None): + """loads signed data (deprecated because of incorrect padding)""" encryption_key = to_bytes(encryption_key) data = to_native(data) if ':' not in data: return None if not hash_key: - hash_key = sha1(encryption_key).hexdigest() + hash_key = hashlib.sha1(encryption_key).hexdigest() signature, encrypted_data = data.split(':', 1) encrypted_data = to_bytes(encrypted_data) actual_signature = hmac.new(to_bytes(hash_key), encrypted_data, hashlib.md5).hexdigest() @@ -274,7 +207,7 @@ def secure_loads_deprecated(data, encryption_key, hash_key=None, compression_lev if compression_level: data = zlib.decompress(data) return pickle.loads(data) - except Exception as e: + except Exception: return None ### compute constant CTOKENS @@ -365,7 +298,7 @@ def web2py_uuid(ctokens=UNPACKED_CTOKENS): rand_longs[1] ^ ctokens[1]) return str(uuid.UUID(bytes=byte_s, version=4)) -REGEX_IPv4 = re.compile('(\d+)\.(\d+)\.(\d+)\.(\d+)') +REGEX_IPv4 = re.compile(r'(\d+)\.(\d+)\.(\d+)\.(\d+)') def is_valid_ip_address(address): diff --git a/gluon/validators.py b/gluon/validators.py index 39974dcc..2f5fbca5 100644 --- a/gluon/validators.py +++ b/gluon/validators.py @@ -1,5 +1,6 @@ #!/bin/env python # -*- coding: utf-8 -*- +#pylint: disable=line-too-long,invalid-name """ | This file is part of the web2py Web Framework @@ -16,21 +17,23 @@ import math import datetime import time import cgi +import uuid +import hashlib +import hmac import json import struct import decimal +import binascii +import ipaddress import unicodedata import encodings.idna from pydal._compat import StringIO, integer_types, basestring, unicodeT, urllib_unquote, \ unichr, to_bytes, PY2, to_unicode, to_native, string_types, urlparse from pydal.objects import Field, FieldVirtual, FieldMethod, Table -from functools import reduce -from gluon.utils import simple_hash, web2py_uuid, DIGEST_ALG_BY_SIZE -from gluon._compat import ipaddress -regex_isint = re.compile('^[+-]?\d+$') +regex_isint = re.compile(r'^[+-]?\d+$') JSONErrors = (NameError, TypeError, ValueError, AttributeError, KeyError) @@ -80,10 +83,10 @@ def options_sorter(x, y): def translate(text): return Validator.translator(text) -class ValidationError(Exception): - def __init__(self, error): - self.error = error +class ValidationError(Exception): + pass + class Validator(object): """ @@ -195,13 +198,13 @@ class IS_MATCH(Validator): self.regex = re.compile(expression) self.error_message = error_message self.extract = extract - self.is_unicode = is_unicode or (not(PY2)) + self.is_unicode = is_unicode or not PY2 - def __call__(self, value): - if not(PY2): # PY3 convert bytes to unicode + def validate(self, value): + if not PY2: # PY3 convert bytes to unicode value = to_unicode(value) - if self.is_unicode or not(PY2): + if self.is_unicode or not PY2: if not isinstance(value, unicodeT): match = self.regex.search(str(value).decode('utf8')) else: @@ -212,8 +215,8 @@ class IS_MATCH(Validator): else: match = self.regex.search(value.encode('utf8')) if match is not None: - return (self.extract and match.group() or value, None) - return (value, self.translator(self.error_message)) + return self.extract and match.group() or value + raise ValidationError(self.translator(self.error_message)) class IS_EQUAL_TO(Validator): @@ -239,11 +242,11 @@ class IS_EQUAL_TO(Validator): self.expression = expression self.error_message = error_message - def __call__(self, value): - if value == self.expression: - return (value, None) - return (value, self.translator(self.error_message)) - + def validate(self, value): + if value != self.expression: + raise ValidationError(self.translator(self.error_message)) + return value + class IS_EXPR(Validator): """ @@ -268,15 +271,18 @@ class IS_EXPR(Validator): self.error_message = error_message self.environment = environment or {} - def __call__(self, value): + def validate(self, value): if callable(self.expression): - return (value, self.expression(value)) + message = self.expression(value) + if message: + raise ValidationError(message) + return value # for backward compatibility self.environment.update(value=value) exec('__ret__=' + self.expression, self.environment) if self.environment['__ret__']: - return (value, None) - return (value, self.translator(self.error_message)) + return value + raise ValidationError(self.translator(self.error_message)) class IS_LENGTH(Validator): @@ -319,11 +325,11 @@ class IS_LENGTH(Validator): self.minsize = minsize self.error_message = error_message - def __call__(self, value): + def validate(self, value): if value is None: length = 0 if self.minsize <= length <= self.maxsize: - return (value, None) + return value elif isinstance(value, cgi.FieldStorage): if value.file: value.file.seek(0, os.SEEK_END) @@ -336,27 +342,27 @@ class IS_LENGTH(Validator): else: length = 0 if self.minsize <= length <= self.maxsize: - return (value, None) + return value elif isinstance(value, str): try: lvalue = len(to_unicode(value)) except: lvalue = len(value) if self.minsize <= lvalue <= self.maxsize: - return (value, None) + return value elif isinstance(value, unicodeT): if self.minsize <= len(value) <= self.maxsize: - return (value.encode('utf8'), None) + return value.encode('utf8') elif isinstance(value, (bytes, bytearray)): if self.minsize <= len(value) <= self.maxsize: - return (value, None) + return value elif isinstance(value, (tuple, list)): if self.minsize <= len(value) <= self.maxsize: - return (value, None) + return value elif self.minsize <= len(str(value)) <= self.maxsize: - return (str(value), None) - return (value, self.translator(self.error_message) % - dict(min=self.minsize, max=self.maxsize)) + return str(value) + raise ValidationError(self.translator(self.error_message) % + dict(min=self.minsize, max=self.maxsize)) class IS_JSON(Validator): @@ -378,15 +384,15 @@ class IS_JSON(Validator): self.native_json = native_json self.error_message = error_message - def __call__(self, value): + def validate(self, value): try: if self.native_json: json.loads(value) # raises error in case of malformed json - return (value, None) # the serialized value is not passed + return value # the serialized value is not passed else: - return (json.loads(value), None) + return json.loads(value) except JSONErrors: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) def formatter(self, value): if value is None: @@ -427,15 +433,14 @@ class IS_IN_SET(Validator): """ - def __init__( - self, - theset, - labels=None, - error_message='Value not allowed', - multiple=False, - zero='', - sort=False, - ): + def __init__(self, + theset, + labels=None, + error_message='Value not allowed', + multiple=False, + zero='', + sort=False): + self.multiple = multiple if isinstance(theset, dict): self.theset = [str(item) for item in theset] @@ -462,7 +467,7 @@ class IS_IN_SET(Validator): items.insert(0, ('', self.zero)) return items - def __call__(self, value): + def validate(self, value): if self.multiple: # if below was values = re.compile("[\w\-:]+").findall(str(value)) if not value: @@ -476,13 +481,13 @@ class IS_IN_SET(Validator): thestrset = [str(x) for x in self.theset] failures = [x for x in values if not str(x) in thestrset] if failures and self.theset: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) if self.multiple: if isinstance(self.multiple, (tuple, list)) and \ not self.multiple[0] <= len(values) < self.multiple[1]: - return (values, self.translator(self.error_message)) - return (values, None) - return (value, None) + raise ValidationError(self.translator(self.error_message)) + return values + return value regex1 = re.compile(r'\w+\.\w+') @@ -500,24 +505,23 @@ class IS_IN_DB(Validator): used for reference fields, rendered as a dropbox """ - def __init__( - self, - dbset, - field, - label=None, - error_message='Value not in database', - orderby=None, - groupby=None, - distinct=None, - cache=None, - multiple=False, - zero='', - sort=False, - _and=None, - left=None, - delimiter=None, - auto_add=False, - ): + def __init__(self, + dbset, + field, + label=None, + error_message='Value not in database', + orderby=None, + groupby=None, + distinct=None, + cache=None, + multiple=False, + zero='', + sort=False, + _and=None, + left=None, + delimiter=None, + auto_add=False): + if hasattr(dbset, 'define_table'): self.dbset = dbset() else: @@ -615,7 +619,7 @@ class IS_IN_DB(Validator): else: return table.insert(**d) - def __call__(self, value): + def validate(self, value): table = self.dbset.db[self.ktable] field = table[self.kfield] @@ -638,16 +642,16 @@ class IS_IN_DB(Validator): if self.auto_add: value = str(self.maybe_add(table, self.fieldnames[0], value)) else: - return (values, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) new_values.append(value) values = new_values if isinstance(self.multiple, (tuple, list)) and \ not self.multiple[0] <= len(values) < self.multiple[1]: - return (values, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) if self.theset: if not [v for v in values if v not in self.theset]: - return (values, None) + return values else: def count(values, s=self.dbset, f=field): return s(f.belongs(list(map(int, values)))).count() @@ -656,9 +660,9 @@ class IS_IN_DB(Validator): range_ids = range(0, len(values), 30) total = sum(count(values[i:i + 30]) for i in range_ids) if total == len(values): - return (values, None) + return values elif count(values) == len(values): - return (values, None) + return values else: if field.type in ('id', 'integer'): if isinstance(value, integer_types) or (isinstance(value, string_types) and value.isdigit()): @@ -666,26 +670,24 @@ class IS_IN_DB(Validator): elif self.auto_add: value = self.maybe_add(table, self.fieldnames[0], value) else: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) try: value = int(value) except TypeError: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) if self.theset: if str(value) in self.theset: if self._and: - return self._and(value) - else: - return (value, None) + return self._and.validate(value) + return value else: if self.dbset(field == value).count(): if self._and: - return self._and(value) - else: - return (value, None) - return (value, self.translator(self.error_message)) + return self._and.validate(value) + return value + raise ValidationError(self.translator(self.error_message)) class IS_NOT_IN_DB(Validator): @@ -698,14 +700,13 @@ class IS_NOT_IN_DB(Validator): makes the field unique """ - def __init__( - self, - dbset, - field, - error_message='Value already in database or empty', - allowed_override=[], - ignore_common_filters=False, - ): + def __init__(self, + dbset, + field, + error_message='Value already in database or empty', + allowed_override=[], + ignore_common_filters=False): + if isinstance(field, Table): field = field._id @@ -722,12 +723,12 @@ class IS_NOT_IN_DB(Validator): def set_self_id(self, id): self.record_id = id - def __call__(self, value): + def validate(self, value): value = to_native(str(value)) if not value.strip(): - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) if value in self.allowed_override: - return (value, None) + return value (tablename, fieldname) = str(self.field).split('.') table = self.dbset.db[tablename] field = table[fieldname] @@ -738,12 +739,12 @@ class IS_NOT_IN_DB(Validator): fields = [table[f] for f in id] row = subset.select(*fields, **dict(limitby=(0, 1), orderby_on_limitby=False)).first() if row and any(str(row[f]) != str(id[f]) for f in id): - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) else: row = subset.select(table._id, field, limitby=(0, 1), orderby_on_limitby=False).first() if row and str(row[table._id]) != str(id): - return (value, self.translator(self.error_message)) - return (value, None) + raise ValidationError(self.translator(self.error_message)) + return value def range_error_message(error_message, what_to_enter, minimum, maximum): @@ -801,24 +802,23 @@ class IS_INT_IN_RANGE(Validator): ('abc', 'enter an integer') """ - def __init__( - self, - minimum=None, - maximum=None, - error_message=None, - ): + def __init__(self, + minimum=None, + maximum=None, + error_message=None): + self.minimum = int(minimum) if minimum is not None else None self.maximum = int(maximum) if maximum is not None else None self.error_message = range_error_message( error_message, 'an integer', self.minimum, self.maximum) - def __call__(self, value): + def validate(self, value): if regex_isint.match(str(value)): v = int(value) if ((self.minimum is None or v >= self.minimum) and (self.maximum is None or v < self.maximum)): - return (v, None) - return (value, self.error_message) + return v + raise ValidationError(self.error_message) def str2dec(number): @@ -870,20 +870,19 @@ class IS_FLOAT_IN_RANGE(Validator): ('abc', 'enter a number') """ - def __init__( - self, - minimum=None, - maximum=None, - error_message=None, - dot='.' - ): + def __init__(self, + minimum=None, + maximum=None, + error_message=None, + dot='.'): + self.minimum = float(minimum) if minimum is not None else None self.maximum = float(maximum) if maximum is not None else None self.dot = str(dot) self.error_message = range_error_message( error_message, 'a number', self.minimum, self.maximum) - def __call__(self, value): + def validate(self, value): try: if self.dot == '.': v = float(value) @@ -891,10 +890,10 @@ class IS_FLOAT_IN_RANGE(Validator): v = float(str(value).replace(self.dot, '.')) if ((self.minimum is None or v >= self.minimum) and (self.maximum is None or v <= self.maximum)): - return (v, None) + return v except (ValueError, TypeError): pass - return (value, self.error_message) + raise ValidationError(self.error_message) def formatter(self, value): if value is None: @@ -956,31 +955,28 @@ class IS_DECIMAL_IN_RANGE(Validator): ('abc', 'enter a number') """ - def __init__( - self, - minimum=None, - maximum=None, - error_message=None, - dot='.' - ): + def __init__(self, + minimum=None, + maximum=None, + error_message=None, + dot='.'): + self.minimum = decimal.Decimal(str(minimum)) if minimum is not None else None self.maximum = decimal.Decimal(str(maximum)) if maximum is not None else None self.dot = str(dot) self.error_message = range_error_message( error_message, 'a number', self.minimum, self.maximum) - def __call__(self, value): + def validate(self, value): try: - if isinstance(value, decimal.Decimal): - v = value - else: - v = decimal.Decimal(str(value).replace(self.dot, '.')) - if ((self.minimum is None or v >= self.minimum) and - (self.maximum is None or v <= self.maximum)): - return (v, None) + if not isinstance(value, decimal.Decimal): + value = decimal.Decimal(str(value).replace(self.dot, '.')) + if ((self.minimum is None or value >= self.minimum) and + (self.maximum is None or value <= self.maximum)): + return value except (ValueError, TypeError, decimal.InvalidOperation): pass - return (value, self.error_message) + raise ValidationError(self.error_message) def formatter(self, value): if value is None: @@ -1040,11 +1036,11 @@ class IS_NOT_EMPTY(Validator): else: self.empty_regex = None - def __call__(self, value): + def validate(self, value): value, empty = is_empty(value, empty_regex=self.empty_regex) if empty: - return (value, self.translator(self.error_message)) - return (value, None) + raise ValidationError(self.translator(self.error_message)) + return value class IS_ALPHANUMERIC(IS_MATCH): @@ -1065,10 +1061,11 @@ class IS_ALPHANUMERIC(IS_MATCH): """ def __init__(self, error_message='Enter only letters, numbers, and underscore'): - IS_MATCH.__init__(self, '^[\w]*$', error_message) + IS_MATCH.__init__(self, r'^[\w]*$', error_message) class IS_EMAIL(Validator): + """ Checks if field's value is a valid email address. Can be set to disallow or force addresses from certain domain(s). @@ -1093,12 +1090,12 @@ class IS_EMAIL(Validator): Check for valid email address that can't be from a .com domain:: INPUT(_type='text', _name='name', - requires=IS_EMAIL(banned='^.*\.com(|\..*)$')) + requires=IS_EMAIL(banned='^.*\\.com(|\\..*)$')) Check for valid email address that must be from a .edu domain:: INPUT(_type='text', _name='name', - requires=IS_EMAIL(forced='^.*\.edu(|\..*)$')) + requires=IS_EMAIL(forced='^.*\\.edu(|\\..*)$')) >>> IS_EMAIL()('a@b.com') ('a@b.com', None) @@ -1197,9 +1194,9 @@ class IS_EMAIL(Validator): self.forced = forced self.error_message = error_message - def __call__(self, value): + def validate(self, value): if not(isinstance(value, (basestring, unicodeT))) or not value or '@' not in value: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) body, domain = value.rsplit('@', 1) @@ -1221,8 +1218,8 @@ class IS_EMAIL(Validator): if match: if (not self.banned or not self.banned.match(domain)) \ and (not self.forced or self.forced.match(domain)): - return (value, None) - return (value, self.translator(self.error_message)) + return value + raise ValidationError(self.translator(self.error_message)) class IS_LIST_OF_EMAILS(Validator): @@ -1237,23 +1234,21 @@ class IS_LIST_OF_EMAILS(Validator): XML(', '.join([A(x, _href='mailto:'+x).xml() for x in (v or [])])) ) """ - split_emails = re.compile('[^,;\s]+') + split_emails = re.compile(r'[^,;\s]+') def __init__(self, error_message='Invalid emails: %s'): self.error_message = error_message - def __call__(self, value): + def validate(self, value): bad_emails = [] f = IS_EMAIL() for email in self.split_emails.findall(value): error = f(email)[1] if error and email not in bad_emails: bad_emails.append(email) - if not bad_emails: - return (value, None) - else: - return (value, - self.translator(self.error_message) % ', '.join(bad_emails)) + if bad_emails: + raise ValidationError(self.translator(self.error_message) % ', '.join(bad_emails)) + return value def formatter(self, value, row=None): return ', '.join(value or []) @@ -1394,11 +1389,12 @@ label_split_regex = re.compile(u'[\u002e\u3002\uff0e\uff61]') def escape_unicode(string): + """ Converts a unicode string into US-ASCII, using a simple conversion scheme. Each unicode character that does not have a US-ASCII equivalent is converted into a URL escaped form based on its hexadecimal value. - For example, the unicode character '\u4e86' will become the string '%4e%86' + For example, the unicode character '\\u4e86' will become the string '%4e%86' Args: string: unicode string, the unicode string to convert into an @@ -1426,12 +1422,12 @@ def unicode_to_ascii_authority(authority): """ Follows the steps in RFC 3490, Section 4 to convert a unicode authority string into its ASCII equivalent. - For example, u'www.Alliancefran\xe7aise.nu' will be converted into + For example, u'www.Alliancefran\\xe7aise.nu' will be converted into 'www.xn--alliancefranaise-npb.nu' Args: authority: unicode string, the URL authority component to convert, - e.g. u'www.Alliancefran\xe7aise.nu' + e.g. u'www.Alliancefran\\xe7aise.nu' Returns: string: the US-ASCII character equivalent to the inputed authority, @@ -1577,12 +1573,10 @@ class IS_GENERIC_URL(Validator): """ - def __init__( - self, - error_message='Enter a valid URL', - allowed_schemes=None, - prepend_scheme=None, - ): + def __init__(self, + error_message='Enter a valid URL', + allowed_schemes=None, + prepend_scheme=None): self.error_message = error_message if allowed_schemes is None: @@ -1598,7 +1592,7 @@ class IS_GENERIC_URL(Validator): GENERIC_URL_VALID = re.compile(r"[A-Za-z0-9;/?:@&=+$,\-_\.!~*'\(\)%]+$") URL_FRAGMENT_VALID = re.compile(r"[|A-Za-z0-9;/?:@&=+$,\-_\.!~*'\(\)%]+$") - def __call__(self, value): + def validate(self, value): """ Args: value: a string, the URL to validate @@ -1612,47 +1606,39 @@ class IS_GENERIC_URL(Validator): # if we dont have anything or the URL misuses the '%' character if not value or self.GENERIC_URL.search(value): - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) if '#' in value: url, fragment_part = value.split('#') else: url, fragment_part = value, '' # if the URL is only composed of valid characters - if self.GENERIC_URL_VALID.match(url) and (not fragment_part or self.URL_FRAGMENT_VALID.match(fragment_part)): - # Then parse the URL into its components and check on - try: - components = urlparse.urlparse(urllib_unquote(value))._asdict() - except ValueError: - return (value, self.translator(self.error_message)) + if not self.GENERIC_URL_VALID.match(url) or (fragment_part and not self.URL_FRAGMENT_VALID.match(fragment_part)): + raise ValidationError(self.translator(self.error_message)) + # Then parse the URL into its components and check on + try: + components = urlparse.urlparse(urllib_unquote(value))._asdict() + except ValueError: + raise ValidationError(self.translator(self.error_message)) - # Clean up the scheme before we check it - scheme = components['scheme'] - if len(scheme) == 0: - scheme = None - else: - scheme = components['scheme'].lower() - # If the scheme doesn't really exists - if scheme not in self.allowed_schemes or not scheme and ':' in components['path']: - # for the possible case of abbreviated URLs with - # ports, check to see if adding a valid scheme fixes - # the problem (but only do this if it doesn't have - # one already!) - if '://' not in value and None in self.allowed_schemes: - schemeToUse = self.prepend_scheme or 'http' - prependTest = self.__call__( - schemeToUse + '://' + value) - # if the prepend test succeeded - if prependTest[1] is None: - # if prepending in the output is enabled - if self.prepend_scheme: - return prependTest - else: - return (value, None) - else: - return (value, None) - # else the URL is not valid - return (value, self.translator(self.error_message)) + # Clean up the scheme before we check it + scheme = components['scheme'] + if len(scheme) == 0: + scheme = None + else: + scheme = components['scheme'].lower() + # If the scheme doesn't really exists + if scheme not in self.allowed_schemes or not scheme and ':' in components['path']: + # for the possible case of abbreviated URLs with + # ports, check to see if adding a valid scheme fixes + # the problem (but only do this if it doesn't have + # one already!) + if '://' not in value and None in self.allowed_schemes: + schemeToUse = self.prepend_scheme or 'http' + new_value = self.validate(schemeToUse + '://' + value) + return new_value if self.prepend_scheme else value + raise ValidationError(self.translator(self.error_message)) + return value # Sources (obtained 2017-Nov-11): # http://data.iana.org/TLD/tlds-alpha-by-domain.txt @@ -1998,13 +1984,11 @@ class IS_HTTP_URL(Validator): r"([\w.!~*'|;:&=+$,-]+@)?\d+\.\d+\.\d+\.\d+(:\d*)*$") GENERIC_VALID_DOMAIN = re.compile(r"([\w.!~*'|;:&=+$,-]+@)?(([A-Za-z0-9]+[A-Za-z0-9\-]*[A-Za-z0-9]+\.)*([A-Za-z0-9]+\.)*)*([A-Za-z]+[A-Za-z0-9\-]*[A-Za-z0-9]+)\.?(:\d*)*$") - def __init__( - self, - error_message='Enter a valid URL', - allowed_schemes=None, - prepend_scheme='http', - allowed_tlds=None - ): + def __init__(self, + error_message='Enter a valid URL', + allowed_schemes=None, + prepend_scheme='http', + allowed_tlds=None): self.error_message = error_message if allowed_schemes is None: @@ -2026,7 +2010,7 @@ class IS_HTTP_URL(Validator): raise SyntaxError("prepend_scheme='%s' is not in allowed_schemes=%s" % (self.prepend_scheme, self.allowed_schemes)) - def __call__(self, value): + def validate(self, value): """ Args: value: a string, the URL to validate @@ -2049,46 +2033,33 @@ class IS_HTTP_URL(Validator): # if authority is a valid IP address if self.GENERIC_VALID_IP.match(authority): # Then this HTTP URL is valid - return (value, None) + return value else: # else if authority is a valid domain name domainMatch = self.GENERIC_VALID_DOMAIN.match( authority) if domainMatch: # if the top-level domain really exists - if domainMatch.group(5).lower()\ - in self.allowed_tlds: + if domainMatch.group(5).lower() in self.allowed_tlds: # Then this HTTP URL is valid - return (value, None) + return value else: # else this is a relative/abbreviated URL, which will parse # into the URL's path component path = components.path # relative case: if this is a valid path (if it starts with # a slash) - if path.startswith('/'): - # Then this HTTP URL is valid - return (value, None) - else: + if not path.startswith('/'): # abbreviated case: if we haven't already, prepend a # scheme and see if it fixes the problem if '://' not in value and None in self.allowed_schemes: schemeToUse = self.prepend_scheme or 'http' - prependTest = self.__call__(schemeToUse - + '://' + value) - # if the prepend test succeeded - if prependTest[1] is None: - # if prepending in the output is enabled - if self.prepend_scheme: - return prependTest - else: - # else return the original, non-prepended - # value - return (value, None) + new_value = self.validate(schemeToUse + '://' + value) + return new_value if self.prepend_scheme else value + return value except: pass - # else the HTTP URL is not valid - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) class IS_URL(Validator): @@ -2176,14 +2147,12 @@ class IS_URL(Validator): @author: Jonathan Benn """ - def __init__( - self, - error_message='Enter a valid URL', - mode='http', - allowed_schemes=None, - prepend_scheme='http', - allowed_tlds=None - ): + def __init__(self, + error_message='Enter a valid URL', + mode='http', + allowed_schemes=None, + prepend_scheme='http', + allowed_tlds=None): self.error_message = error_message self.mode = mode.lower() @@ -2205,7 +2174,7 @@ class IS_URL(Validator): self.prepend_scheme = prepend_scheme - def __call__(self, value): + def validate(self, value): """ Args: value: a unicode or regular string, the URL to validate @@ -2230,22 +2199,14 @@ class IS_URL(Validator): else: raise SyntaxError("invalid mode '%s' in IS_URL" % self.mode) - if not isinstance(value, unicodeT): - return subMethod(value) - else: + if isinstance(value, unicodeT): try: - asciiValue = unicode_to_ascii_url(value, self.prepend_scheme) + value = unicode_to_ascii_url(value, self.prepend_scheme) except Exception as e: # If we are not able to convert the unicode url into a # US-ASCII URL, then the URL is not valid - return (value, self.translator(self.error_message)) - methodResult = subMethod(asciiValue) - # if the validation of the US-ASCII version of the value failed - if not methodResult[1] is None: - # then return the original input value, not the US-ASCII version - return (value, methodResult[1]) - else: - return methodResult + raise ValidationError(self.translator(self.error_message)) + return subMethod.validate(value) regex_time = re.compile( @@ -2298,7 +2259,7 @@ class IS_TIME(Validator): def __init__(self, error_message='Enter time as hh:mm:ss (seconds, am, pm optional)'): self.error_message = error_message - def __call__(self, value): + def validate(self, value): try: ivalue = value value = regex_time.match(value.lower()) @@ -2315,12 +2276,9 @@ class IS_TIME(Validator): in range(60)): raise ValueError('Hours or minutes or seconds are outside of allowed range') value = datetime.time(h, m, s) - return (value, None) - except AttributeError: - pass - except ValueError: - pass - return (ivalue, self.translator(self.error_message)) + return value + except Exception: + raise ValidationError(self.translator(self.error_message)) # A UTC class. @@ -2355,18 +2313,17 @@ class IS_DATE(Validator): self.error_message = str(error_message) self.extremes = {} - def __call__(self, value): - ovalue = value + def validate(self, value): if isinstance(value, datetime.date): - return (value, None) + return value try: (y, m, d, hh, mm, ss, t0, t1, t2) = \ time.strptime(value, str(self.format)) value = datetime.date(y, m, d) - return (value, None) + return value except: self.extremes.update(IS_DATETIME.nice(self.format)) - return (ovalue, self.translator(self.error_message) % self.extremes) + raise ValidationError(self.translator(self.error_message) % self.extremes) def formatter(self, value): if value is None: @@ -2420,10 +2377,9 @@ class IS_DATETIME(Validator): self.extremes = {} self.timezone = timezone - def __call__(self, value): - ovalue = value + def validate(self, value): if isinstance(value, datetime.datetime): - return (value, None) + return value try: (y, m, d, hh, mm, ss, t0, t1, t2) = \ time.strptime(value, str(self.format)) @@ -2431,10 +2387,10 @@ class IS_DATETIME(Validator): if self.timezone is not None: # TODO: https://github.com/web2py/web2py/issues/1094 (temporary solution) value = self.timezone.localize(value).astimezone(utc).replace(tzinfo=None) - return (value, None) + return value except: self.extremes.update(IS_DATETIME.nice(self.format)) - return (ovalue, self.translator(self.error_message) % self.extremes) + raise ValidationError(self.translator(self.error_message) % self.extremes) def formatter(self, value): if value is None: @@ -2496,16 +2452,13 @@ class IS_DATE_IN_RANGE(IS_DATE): self.extremes = dict(min=self.formatter(minimum), max=self.formatter(maximum)) - def __call__(self, value): - ovalue = value - (value, msg) = IS_DATE.__call__(self, value) - if msg is not None: - return (value, msg) + def validate(self, value): + value = IS_DATE.validate(self, value) if self.minimum and self.minimum > value: - return (ovalue, self.translator(self.error_message) % self.extremes) + raise ValidationError(self.translator(self.error_message) % self.extremes) if self.maximum and value > self.maximum: - return (ovalue, self.translator(self.error_message) % self.extremes) - return (value, None) + raise ValidationError(self.translator(self.error_message) % self.extremes) + return value class IS_DATETIME_IN_RANGE(IS_DATETIME): @@ -2552,16 +2505,13 @@ class IS_DATETIME_IN_RANGE(IS_DATETIME): self.extremes = dict(min=self.formatter(minimum), max=self.formatter(maximum)) - def __call__(self, value): - ovalue = value - (value, msg) = IS_DATETIME.__call__(self, value) - if msg is not None: - return (value, msg) + def validate(self, value): + value = IS_DATETIME.validate(self, value) if self.minimum and self.minimum > value: - return (ovalue, self.translator(self.error_message) % self.extremes) + raise ValidationError(self.translator(self.error_message) % self.extremes) if self.maximum and value > self.maximum: - return (ovalue, self.translator(self.error_message) % self.extremes) - return (value, None) + raise ValidationError(self.translator(self.error_message) % self.extremes) + return value class IS_LIST_OF(Validator): @@ -2572,17 +2522,17 @@ class IS_LIST_OF(Validator): self.maximum = maximum self.error_message = error_message - def __call__(self, value): + def validate(self, value): ivalue = value if not isinstance(value, list): ivalue = [ivalue] ivalue = [i for i in ivalue if str(i).strip()] if self.minimum is not None and len(ivalue) < self.minimum: - return (ivalue, self.translator(self.error_message or 'Minimum length is %(min)s') % - dict(min=self.minimum, max=self.maximum)) + raise ValidationError(self.translator(self.error_message or 'Minimum length is %(min)s') % + dict(min=self.minimum, max=self.maximum)) if self.maximum is not None and len(ivalue) > self.maximum: - return (ivalue, self.translator(self.error_message or 'Maximum length is %(max)s') % - dict(min=self.minimum, max=self.maximum)) + raise ValidationError(self.translator(self.error_message or 'Maximum length is %(max)s') % + dict(min=self.minimum, max=self.maximum)) new_value = [] other = self.other if self.other: @@ -2591,12 +2541,10 @@ class IS_LIST_OF(Validator): for item in ivalue: v = item for validator in other: - (v, e) = validator(v) - if e: - return (ivalue, e) + v = validator.validate(v) new_value.append(v) ivalue = new_value - return (ivalue, None) + return ivalue class IS_LOWER(Validator): @@ -2610,14 +2558,14 @@ class IS_LOWER(Validator): """ - def __call__(self, value): + def validate(self, value): cast_back = lambda x: x if isinstance(value, str): cast_back = to_native elif isinstance(value, bytes): cast_back = to_bytes value = to_unicode(value).lower() - return (cast_back(value), None) + return cast_back(value) class IS_UPPER(Validator): @@ -2631,14 +2579,14 @@ class IS_UPPER(Validator): """ - def __call__(self, value): + def validate(self, value): cast_back = lambda x: x if isinstance(value, str): cast_back = to_native elif isinstance(value, bytes): cast_back = to_bytes value = to_unicode(value).upper() - return (cast_back(value), None) + return cast_back(value) def urlify(s, maxlen=80, keep_underscores=False): @@ -2651,15 +2599,15 @@ def urlify(s, maxlen=80, keep_underscores=False): s = s.lower() # to lowercase s = unicodedata.normalize('NFKD', s) # replace special characters s = to_native(s, charset='ascii', errors='ignore') # encode as ASCII - s = re.sub('&\w+?;', '', s) # strip html entities + s = re.sub(r'&\w+?;', '', s) # strip html entities if keep_underscores: - s = re.sub('\s+', '-', s) # whitespace to hyphens - s = re.sub('[^\w\-]', '', s) + s = re.sub(r'\s+', '-', s) # whitespace to hyphens + s = re.sub(r'[^\w\-]', '', s) # strip all but alphanumeric/underscore/hyphen else: - s = re.sub('[\s_]+', '-', s) # whitespace & underscores to hyphens - s = re.sub('[^a-z0-9\-]', '', s) # strip all but alphanumeric/hyphen - s = re.sub('[-_][-_]+', '-', s) # collapse strings of hyphens + s = re.sub(r'[\s_]+', '-', s) # whitespace & underscores to hyphens + s = re.sub(r'[^a-z0-9\-]', '', s) # strip all but alphanumeric/hyphen + s = re.sub(r'[-_][-_]+', '-', s) # collapse strings of hyphens s = s.strip('-') # remove leading and trailing hyphens return s[:maxlen] # enforce maximum length @@ -2718,10 +2666,10 @@ class IS_SLUG(Validator): self.error_message = error_message self.keep_underscores = keep_underscores - def __call__(self, value): + def validate(self, value): if self.check and value != urlify(value, self.maxlen, self.keep_underscores): - return (value, self.translator(self.error_message)) - return (urlify(value, self.maxlen, self.keep_underscores), None) + raise ValidationError(self.translator(self.error_message)) + return urlify(value, self.maxlen, self.keep_underscores) class ANY_OF(Validator): @@ -2743,15 +2691,13 @@ class ANY_OF(Validator): self.subs = subs self.error_message = error_message - def __call__(self, value): + def validate(self, value): for validator in self.subs: - value, error = validator(value) - if error is None: - break - if error is not None and self.error_message is not None: - error = self.translator(self.error_message) - return value, error - + v, e = validator(value) + if not e: + return v + raise ValidationError(e) + def formatter(self, value): # Use the formatter of the first subvalidator # that validates the value and has a formatter @@ -2804,19 +2750,15 @@ class IS_EMPTY_OR(Validator): if hasattr(self.other, 'set_self_id'): self.other.set_self_id(id) - def __call__(self, value): + def validate(self, value): value, empty = is_empty(value, empty_regex=self.empty_regex) if empty: - return (self.null, None) + return self.null if isinstance(self.other, (list, tuple)): - error = None for item in self.other: - value, error = item(value) - if error: - break - return value, error - else: - return self.other(value) + value = item.validate(value) + return value + return self.other.validate(value) def formatter(self, value): if hasattr(self.other, 'formatter'): @@ -2841,9 +2783,57 @@ class CLEANUP(Validator): self.regex = self.REGEX_CLEANUP if regex is None \ else re.compile(regex) - def __call__(self, value): + def validate(self, value): v = self.regex.sub('', str(value).strip()) - return (v, None) + return v + + +def pbkdf2_hex(data, salt, iterations=1000, keylen=24, hashfunc=None): + hashfunc = hashfunc or sha1 + hmac = hashlib.pbkdf2_hmac(hashfunc().name, to_bytes(data), + to_bytes(salt), iterations, keylen) + return binascii.hexlify(hmac) + + +def simple_hash(text, key='', salt='', digest_alg='md5'): + """Generate hash with the given text using the specified digest algorithm.""" + text = to_bytes(text) + key = to_bytes(key) + salt = to_bytes(salt) + if not digest_alg: + raise RuntimeError("simple_hash with digest_alg=None") + elif not isinstance(digest_alg, str): # manual approach + h = digest_alg(text + key + salt) + elif digest_alg.startswith('pbkdf2'): # latest and coolest! + iterations, keylen, alg = digest_alg[7:-1].split(',') + return to_native(pbkdf2_hex(text, salt, int(iterations), + int(keylen), get_digest(alg))) + elif key: # use hmac + digest_alg = get_digest(digest_alg) + h = hmac.new(key + salt, text, digest_alg) + else: # compatible with third party systems + h = get_digest(digest_alg)() + h.update(text + salt) + return h.hexdigest() + + +def get_digest(value): + """Return a hashlib digest algorithm from a string.""" + if isinstance(value, str): + value = value.lower() + if value not in ('md5', 'sha1', 'sha224', 'sha256', 'sha384', 'sha512'): + raise ValueError("Invalid digest algorithm: %s" % value) + value = getattr(hashlib, value) + return value + +DIGEST_ALG_BY_SIZE = { + 128 // 4: 'md5', + 160 // 4: 'sha1', + 224 // 4: 'sha224', + 256 // 4: 'sha256', + 384 // 4: 'sha384', + 512 // 4: 'sha512', +} class LazyCrypt(object): @@ -2889,7 +2879,7 @@ class LazyCrypt(object): digest_alg, key = self.crypt.digest_alg, '' if self.crypt.salt: if self.crypt.salt is True: - salt = str(web2py_uuid()).replace('-', '')[-16:] + salt = str(uuid.uuid4()).replace('-', '')[-16:] else: salt = self.crypt.salt else: @@ -3034,13 +3024,13 @@ class CRYPT(Validator): self.error_message = error_message self.salt = salt - def __call__(self, value): + def validate(self, value): v = value and str(value)[:self.max_length] if not v or len(v) < self.min_length: return ('', self.translator(self.error_message)) if isinstance(value, LazyCrypt): - return (value, None) - return (LazyCrypt(self, value), None) + return value + return LazyCrypt(self, value) # entropy calculator for IS_STRONG # @@ -3140,10 +3130,10 @@ class IS_STRONG(Validator): self.error_message = error_message self.estring = es # return error message as string (for doctest) - def __call__(self, value): + def validate(self, value): failures = [] if value and len(value) == value.count('*') > 4: - return (value, None) + return value if self.entropy is not None: entropy = calc_entropy(value) if entropy < self.entropy: @@ -3159,7 +3149,7 @@ class IS_STRONG(Validator): all_special = [ch in value for ch in self.specials] if self.special > 0: if not all_special.count(True) >= self.special: - failures.append(self.translator("Must include at least %s of the following: %s") + failures.append(self.translator("Must include at least %s of the following: %s") % (self.special, self.specials)) elif self.special is 0: if len(all_special) > 0: @@ -3203,13 +3193,13 @@ class IS_STRONG(Validator): if len(all_number) > 0: failures.append(self.translator("May not include any numbers")) if len(failures) == 0: - return (value, None) + return value if not self.error_message: if self.estring: - return (value, '|'.join(map(str, failures))) - return (value, ', '.join(failures)) + raise ValidationError('|'.join(map(str, failures))) + raise ValidationError(', '.join(failures)) else: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) class IS_IMAGE(Validator): @@ -3269,7 +3259,7 @@ class IS_IMAGE(Validator): self.aspectratio = aspectratio self.error_message = error_message - def __call__(self, value): + def validate(self, value): try: extension = value.filename.rfind('.') assert extension >= 0 @@ -3299,9 +3289,9 @@ class IS_IMAGE(Validator): assert actual_ratio == target_ratio value.file.seek(0) - return (value, None) + return value except Exception as e: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) def __bmp(self, stream): if stream.read(2) == b'BM': @@ -3404,11 +3394,11 @@ class IS_FILE(Validator): elif isinstance(value1, str): return value1 == value2 - def __call__(self, value): + def validate(self, value): try: string = value.filename except: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) if self.case == 1: string = string.lower() elif self.case == 2: @@ -3420,11 +3410,11 @@ class IS_FILE(Validator): if dot == -1: dot = len(string) if self.filename and not self.match(self.filename, string[:dot]): - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) elif self.extension and not self.match(self.extension, string[dot + 1:]): - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) else: - return (value, None) + return value class IS_UPLOAD_FILENAME(Validator): @@ -3482,11 +3472,11 @@ class IS_UPLOAD_FILENAME(Validator): self.case = case self.error_message = error_message - def __call__(self, value): + def validate(self, value): try: string = value.filename except: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) if self.case == 1: string = string.lower() elif self.case == 2: @@ -3498,11 +3488,11 @@ class IS_UPLOAD_FILENAME(Validator): if dot == -1: dot = len(string) if self.filename and not self.filename.match(string[:dot]): - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) elif self.extension and not self.extension.match(string[dot + 1:]): - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) else: - return (value, None) + return value class IS_IPV4(Validator): @@ -3611,15 +3601,15 @@ class IS_IPV4(Validator): private = ((2886729728, 2886795263), (3232235520, 3232301055)) automatic = (2851995648, 2852061183) - def __init__( - self, - minip='0.0.0.0', - maxip='255.255.255.255', - invert=False, - is_localhost=None, - is_private=None, - is_automatic=None, - error_message='Enter valid IPv4 address'): + def __init__(self, + minip='0.0.0.0', + maxip='255.255.255.255', + invert=False, + is_localhost=None, + is_private=None, + is_automatic=None, + error_message='Enter valid IPv4 address'): + for n, value in enumerate((minip, maxip)): temp = [] if isinstance(value, str): @@ -3649,28 +3639,34 @@ class IS_IPV4(Validator): self.is_automatic = is_automatic self.error_message = error_message - def __call__(self, value): + def validate(self, value): if self.regex.match(value): number = 0 for i, j in zip(self.numbers, value.split('.')): number += i * int(j) ok = False + for bottom, top in zip(self.minip, self.maxip): if self.invert != (bottom <= number <= top): ok = True + if ok and self.is_localhost is not None and \ self.is_localhost != (number == self.localhost): ok = False - if ok and self.is_private is not None and (self.is_private != - any([private_number[0] <= number <= private_number[1] - for private_number in self.private])): + + private = any([private_number[0] <= number <= private_number[1] + for private_number in self.private]) + if ok and self.is_private is not None and self.is_private != private: ok = False - if ok and self.is_automatic is not None and (self.is_automatic != - (self.automatic[0] <= number <= self.automatic[1])): + + automatic = self.automatic[0] <= number <= self.automatic[1] + if ok and self.is_automatic is not None and self.is_automatic != automatic: ok = False + if ok: - return (value, None) - return (value, self.translator(self.error_message)) + return value + + raise ValidationError(self.translator(self.error_message)) class IS_IPV6(Validator): @@ -3741,17 +3737,17 @@ class IS_IPV6(Validator): """ - def __init__( - self, - is_private=None, - is_link_local=None, - is_reserved=None, - is_multicast=None, - is_routeable=None, - is_6to4=None, - is_teredo=None, - subnets=None, - error_message='Enter valid IPv6 address'): + def __init__(self, + is_private=None, + is_link_local=None, + is_reserved=None, + is_multicast=None, + is_routeable=None, + is_6to4=None, + is_teredo=None, + subnets=None, + error_message='Enter valid IPv6 address'): + self.is_private = is_private self.is_link_local = is_link_local self.is_reserved = is_reserved @@ -3762,12 +3758,12 @@ class IS_IPV6(Validator): self.subnets = subnets self.error_message = error_message - def __call__(self, value): + def validate(self, value): try: ip = ipaddress.IPv6Address(to_unicode(value)) ok = True except ipaddress.AddressValueError: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) if self.subnets: # iterate through self.subnets to see if value is a member @@ -3778,7 +3774,7 @@ class IS_IPV6(Validator): try: ipnet = ipaddress.IPv6Network(to_unicode(network)) except (ipaddress.NetmaskValueError, ipaddress.AddressValueError): - return (value, self.translator('invalid subnet provided')) + raise ValidationError(self.translator('invalid subnet provided')) if ip in ipnet: ok = True @@ -3807,9 +3803,9 @@ class IS_IPV6(Validator): ok = False if ok: - return (value, None) + return value - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) class IS_IPADDRESS(Validator): @@ -3947,24 +3943,24 @@ class IS_IPADDRESS(Validator): ('2001::8ffa:fe22:b3af', 'invalid subnet provided') """ - def __init__( - self, - minip='0.0.0.0', - maxip='255.255.255.255', - invert=False, - is_localhost=None, - is_private=None, - is_automatic=None, - is_ipv4=None, - is_link_local=None, - is_reserved=None, - is_multicast=None, - is_routeable=None, - is_6to4=None, - is_teredo=None, - subnets=None, - is_ipv6=None, - error_message='Enter valid IP address'): + def __init__(self, + minip='0.0.0.0', + maxip='255.255.255.255', + invert=False, + is_localhost=None, + is_private=None, + is_automatic=None, + is_ipv4=None, + is_link_local=None, + is_reserved=None, + is_multicast=None, + is_routeable=None, + is_6to4=None, + is_teredo=None, + subnets=None, + is_ipv6=None, + error_message='Enter valid IP address'): + self.minip = minip, self.maxip = maxip, self.invert = invert @@ -3983,7 +3979,7 @@ class IS_IPADDRESS(Validator): self.is_ipv6 = is_ipv6 or is_ipv4 is False self.error_message = error_message - def __call__(self, value): + def validate(self, value): IPAddress = ipaddress.ip_address IPv6Address = ipaddress.IPv6Address IPv4Address = ipaddress.IPv4Address @@ -3991,14 +3987,14 @@ class IS_IPADDRESS(Validator): try: ip = IPAddress(to_unicode(value)) except ValueError: - return (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) if self.is_ipv4 and isinstance(ip, IPv6Address): - retval = (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) elif self.is_ipv6 and isinstance(ip, IPv4Address): - retval = (value, self.translator(self.error_message)) + raise ValidationError(self.translator(self.error_message)) elif self.is_ipv4 or isinstance(ip, IPv4Address): - retval = IS_IPV4( + return IS_IPV4( minip=self.minip, maxip=self.maxip, invert=self.invert, @@ -4006,9 +4002,9 @@ class IS_IPADDRESS(Validator): is_private=self.is_private, is_automatic=self.is_automatic, error_message=self.error_message - )(value) + ).validate(value) elif self.is_ipv6 or isinstance(ip, IPv6Address): - retval = IS_IPV6( + return IS_IPV6( is_private=self.is_private, is_link_local=self.is_link_local, is_reserved=self.is_reserved, @@ -4018,8 +4014,6 @@ class IS_IPADDRESS(Validator): is_teredo=self.is_teredo, subnets=self.subnets, error_message=self.error_message - )(value) + ).validate(value) else: - retval = (value, self.translator(self.error_message)) - - return retval + raise ValidationError(self.translator(self.error_message))