fix validators, updated gluon/contrib/ipaddr

This commit is contained in:
ilvalle
2016-06-07 19:50:49 +02:00
parent 92374741ff
commit fd850ab46f
9 changed files with 221 additions and 148 deletions
+1 -1
View File
@@ -40,7 +40,7 @@ before_script:
script: export COVERAGE_PROCESS_START=gluon/tests/coverage.ini; ./web2py.py --run_system_tests --with_coverage script: export COVERAGE_PROCESS_START=gluon/tests/coverage.ini; ./web2py.py --run_system_tests --with_coverage
after_success: after_success:
- if [[ $TRAVIS_PYTHON_VERSION == '2.7' ]]; then coverage combine; fi - coverage combine;
- if [[ $TRAVIS_PYTHON_VERSION == '2.7' ]]; then codecov; fi - if [[ $TRAVIS_PYTHON_VERSION == '2.7' ]]; then codecov; fi
notifications: notifications:
+145 -87
View File
@@ -22,7 +22,7 @@ and networks.
""" """
__version__ = '2.1.11' __version__ = 'trunk'
import struct import struct
@@ -156,16 +156,19 @@ def _find_address_range(addresses):
addresses: a list of IPv4 or IPv6 addresses. addresses: a list of IPv4 or IPv6 addresses.
Returns: Returns:
A tuple containing the first and last IP addresses in the sequence. A tuple containing the first and last IP addresses in the sequence,
and the index of the last IP address in the sequence.
""" """
first = last = addresses[0] first = last = addresses[0]
last_index = 0
for ip in addresses[1:]: for ip in addresses[1:]:
if ip._ip == last._ip + 1: if ip._ip == last._ip + 1:
last = ip last = ip
last_index += 1
else: else:
break break
return (first, last) return (first, last, last_index)
def _get_prefix_length(number1, number2, bits): def _get_prefix_length(number1, number2, bits):
"""Get the number of leading bits that are same for two numbers. """Get the number of leading bits that are same for two numbers.
@@ -358,8 +361,8 @@ def collapse_address_list(addresses):
nets = sorted(set(nets)) nets = sorted(set(nets))
while i < len(ips): while i < len(ips):
(first, last) = _find_address_range(ips[i:]) (first, last, last_index) = _find_address_range(ips[i:])
i = ips.index(last) + 1 i += last_index + 1
addrs.extend(summarize_address_range(first, last)) addrs.extend(summarize_address_range(first, last))
return _collapse_address_list_recursive(sorted( return _collapse_address_list_recursive(sorted(
@@ -876,6 +879,26 @@ class _BaseNet(_IPAddrBase):
else: else:
raise NetmaskValueError('Bit pattern does not match /1*0*/') raise NetmaskValueError('Bit pattern does not match /1*0*/')
def _prefix_from_prefix_int(self, prefixlen):
"""Validate and return a prefix length integer.
Args:
prefixlen: An integer containing the prefix length.
Returns:
The input, possibly converted from long to int.
Raises:
NetmaskValueError: If the input is not an integer, or out of range.
"""
if not isinstance(prefixlen, (int, long)):
raise NetmaskValueError('%r is not an integer' % prefixlen)
prefixlen = int(prefixlen)
if not (0 <= prefixlen <= self._max_prefixlen):
raise NetmaskValueError('%d is not a valid prefix length' %
prefixlen)
return prefixlen
def _prefix_from_prefix_string(self, prefixlen_str): def _prefix_from_prefix_string(self, prefixlen_str):
"""Turn a prefix length string into an integer. """Turn a prefix length string into an integer.
@@ -893,12 +916,10 @@ class _BaseNet(_IPAddrBase):
if not _BaseV4._DECIMAL_DIGITS.issuperset(prefixlen_str): if not _BaseV4._DECIMAL_DIGITS.issuperset(prefixlen_str):
raise ValueError raise ValueError
prefixlen = int(prefixlen_str) prefixlen = int(prefixlen_str)
if not (0 <= prefixlen <= self._max_prefixlen):
raise ValueError
except ValueError: except ValueError:
raise NetmaskValueError('%s is not a valid prefix length' % raise NetmaskValueError('%s is not a valid prefix length' %
prefixlen_str) prefixlen_str)
return prefixlen return self._prefix_from_prefix_int(prefixlen)
def _prefix_from_ip_string(self, ip_str): def _prefix_from_ip_string(self, ip_str):
"""Turn a netmask/hostmask string into a prefix length. """Turn a netmask/hostmask string into a prefix length.
@@ -1239,6 +1260,11 @@ class IPv4Address(_BaseV4, _BaseIP):
""" """
_BaseV4.__init__(self, address) _BaseV4.__init__(self, address)
# Efficient copy constructor.
if isinstance(address, IPv4Address):
self._ip = address._ip
return
# Efficient constructor from integer. # Efficient constructor from integer.
if isinstance(address, (int, long)): if isinstance(address, (int, long)):
self._ip = address self._ip = address
@@ -1279,29 +1305,32 @@ class IPv4Network(_BaseV4, _BaseNet):
"""Instantiate a new IPv4 network object. """Instantiate a new IPv4 network object.
Args: Args:
address: A string or integer representing the IP [& network]. address: The IPv4 network as a string, 2-tuple, or any format
'192.168.1.1/24' supported by the IPv4Address constructor.
'192.168.1.1/255.255.255.0'
'192.168.1.1/0.0.0.255'
are all functionally the same in IPv4. Similarly,
'192.168.1.1'
'192.168.1.1/255.255.255.255'
'192.168.1.1/32'
are also functionaly 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 Strings typically use CIDR format, such as '192.0.2.0/24'.
dotted quad form, it is treated as a netmask if it starts with a If a dotted-quad is provided after the '/', it is treated as
non-zero field (e.g. /255.0.0.0 == /8) and as a hostmask if it a netmask if it starts with a nonzero bit (e.g. 255.0.0.0 == /8)
starts with a zero field (e.g. 0.255.255.255 == /8), with the or a hostmask if it starts with a zero bit
single exception of an all-zero mask which is treated as a (e.g. /0.0.0.255 == /8), with the single exception of an all-zero
netmask == /0. If no mask is given, a default of /32 is used. mask which is treated as /0.
Additionally, an integer can be passed, so The 2-tuple format consists of an (ip, prefixlen), where ip is any
IPv4Network('192.168.1.1') == IPv4Network(3232235777). format recognized by the IPv4Address constructor, and prefixlen is
or, more generally an integer from 0 through 32.
IPv4Network(int(IPv4Network('192.168.1.1'))) ==
IPv4Network('192.168.1.1') A plain IPv4 address (in any format) will be forwarded to the
IPv4Address constructor, with an implied prefixlen of 32.
For example, the following inputs are equivalent:
IPv4Network('192.0.2.1/32')
IPv4Network('192.0.2.1/255.255.255.255')
IPv4Network('192.0.2.1')
IPv4Network(0xc0000201)
IPv4Network(IPv4Address('192.0.2.1'))
IPv4Network(('192.0.2.1', 32))
IPv4Network((0xc0000201, 32))
IPv4Network((IPv4Address('192.0.2.1'), 32))
strict: A boolean. If true, ensure that we have been passed strict: A boolean. If true, ensure that we have been passed
A true network address, eg, 192.168.1.0/24 and not an A true network address, eg, 192.168.1.0/24 and not an
@@ -1318,41 +1347,51 @@ class IPv4Network(_BaseV4, _BaseNet):
_BaseNet.__init__(self, address) _BaseNet.__init__(self, address)
_BaseV4.__init__(self, address) _BaseV4.__init__(self, address)
# Constructing from an integer or packed bytes. # Constructing from a single IP address.
if isinstance(address, (int, long, Bytes)): if isinstance(address, (int, long, Bytes, IPv4Address)):
self.ip = IPv4Address(address) self.ip = IPv4Address(address)
self._ip = self.ip._ip self._ip = self.ip._ip
self._prefixlen = self._max_prefixlen self._prefixlen = self._max_prefixlen
self.netmask = IPv4Address(self._ALL_ONES) self.netmask = IPv4Address(self._ALL_ONES)
return return
# Assume input argument to be string or any object representation # Constructing from an (ip, prefixlen) tuple.
# which converts into a formatted IP prefix string. if isinstance(address, tuple):
addr = str(address).split('/')
if len(addr) > 2:
raise AddressValueError(address)
self._ip = self._ip_int_from_string(addr[0])
self.ip = IPv4Address(self._ip)
if len(addr) == 2:
try: try:
# Check for a netmask in prefix length form. ip, prefixlen = address
self._prefixlen = self._prefix_from_prefix_string(addr[1]) except ValueError:
except NetmaskValueError: raise AddressValueError(address)
# Check for a netmask or hostmask in dotted-quad form. self.ip = IPv4Address(ip)
# This may raise NetmaskValueError. self._ip = self.ip._ip
self._prefixlen = self._prefix_from_ip_string(addr[1]) self._prefixlen = self._prefix_from_prefix_int(prefixlen)
else: else:
self._prefixlen = self._max_prefixlen # Assume input argument to be string or any object representation
# which converts into a formatted IP prefix string.
addr = str(address).split('/')
if len(addr) > 2:
raise AddressValueError(address)
self._ip = self._ip_int_from_string(addr[0])
self.ip = IPv4Address(self._ip)
if len(addr) == 2:
try:
# Check for a netmask in prefix length form.
self._prefixlen = self._prefix_from_prefix_string(addr[1])
except NetmaskValueError:
# Check for a netmask or hostmask in dotted-quad form.
# This may raise NetmaskValueError.
self._prefixlen = self._prefix_from_ip_string(addr[1])
else:
self._prefixlen = self._max_prefixlen
self.netmask = IPv4Address(self._ip_int_from_prefix(self._prefixlen)) self.netmask = IPv4Address(self._ip_int_from_prefix(self._prefixlen))
if strict: if strict:
if self.ip != self.network: if self.ip != self.network:
raise ValueError('%s has host bits set' % raise ValueError('%s has host bits set' % self.ip)
self.ip)
if self._prefixlen == (self._max_prefixlen - 1): if self._prefixlen == (self._max_prefixlen - 1):
self.iterhosts = self.__iter__ self.iterhosts = self.__iter__
@@ -1447,7 +1486,7 @@ class _BaseV6(object):
try: try:
# Now, parse the hextets into a 128-bit integer. # Now, parse the hextets into a 128-bit integer.
ip_int = 0 ip_int = 0L
for i in xrange(parts_hi): for i in xrange(parts_hi):
ip_int <<= 16 ip_int <<= 16
ip_int |= self._parse_hextet(parts[i]) ip_int |= self._parse_hextet(parts[i])
@@ -1752,6 +1791,11 @@ class IPv6Address(_BaseV6, _BaseIP):
""" """
_BaseV6.__init__(self, address) _BaseV6.__init__(self, address)
# Efficient copy constructor.
if isinstance(address, IPv6Address):
self._ip = address._ip
return
# Efficient constructor from integer. # Efficient constructor from integer.
if isinstance(address, (int, long)): if isinstance(address, (int, long)):
self._ip = address self._ip = address
@@ -1771,9 +1815,6 @@ class IPv6Address(_BaseV6, _BaseIP):
# Assume input argument to be string or any object representation # Assume input argument to be string or any object representation
# which converts into a formatted IP string. # which converts into a formatted IP string.
addr_str = str(address) addr_str = str(address)
if not addr_str:
raise AddressValueError('')
self._ip = self._ip_int_from_string(addr_str) self._ip = self._ip_int_from_string(addr_str)
@@ -1793,28 +1834,34 @@ class IPv6Network(_BaseV6, _BaseNet):
def __init__(self, address, strict=False): def __init__(self, address, strict=False):
"""Instantiate a new IPv6 Network object. """Instantiate a new IPv6 network object.
Args: Args:
address: A string or integer representing the IPv6 network or the IP address: The IPv6 network as a string, 2-tuple, or any format
and prefix/netmask. supported by the IPv6Address constructor.
'2001:4860::/128'
'2001:4860:0000:0000:0000:0000:0000:0000/128'
'2001:4860::'
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 Strings should be in CIDR format, such as '2001:db8::/32'.
IPv6Network('2001:4860::') ==
IPv6Network(42541956101370907050197289607612071936L). The 2-tuple format consists of an (ip, prefixlen), where ip is any
or, more generally format recognized by the IPv6Address constructor, and prefixlen is
IPv6Network(IPv6Network('2001:4860::')._ip) == an integer from 0 through 128.
IPv6Network('2001:4860::')
A plain IPv6 address (in any format) will be forwarded to the
IPv6Address constructor, with an implied prefixlen of 128.
For example, the following inputs are equivalent:
IPv6Network('2001:db8::/128')
IPv6Network('2001:db8:0:0:0:0:0:0/128')
IPv6Network('2001:db8::')
IPv6Network(0x20010db8 << 96)
IPv6Network(IPv6Address('2001:db8::'))
IPv6Network(('2001:db8::', 128))
IPv6Network((0x20010db8 << 96, 128))
IPv6Network((IPv6Address('2001:db8::'), 128))
strict: A boolean. If true, ensure that we have been passed strict: A boolean. If true, ensure that we have been passed
A true network address, eg, 192.168.1.0/24 and not an A true network address, eg, 2001:db8::/32 and not an
IP address on a network, eg, 192.168.1.1/24. IP address on a network, eg, 2001:db8::1/32.
Raises: Raises:
AddressValueError: If address isn't a valid IPv6 address. AddressValueError: If address isn't a valid IPv6 address.
@@ -1827,29 +1874,40 @@ class IPv6Network(_BaseV6, _BaseNet):
_BaseNet.__init__(self, address) _BaseNet.__init__(self, address)
_BaseV6.__init__(self, address) _BaseV6.__init__(self, address)
# Constructing from an integer or packed bytes. # Constructing from a single IP address.
if isinstance(address, (int, long, Bytes)): if isinstance(address, (int, long, Bytes, IPv6Address)):
self.ip = IPv6Address(address) self.ip = IPv6Address(address)
self._ip = self.ip._ip self._ip = self.ip._ip
self._prefixlen = self._max_prefixlen self._prefixlen = self._max_prefixlen
self.netmask = IPv6Address(self._ALL_ONES) self.netmask = IPv6Address(self._ALL_ONES)
return return
# Assume input argument to be string or any object representation # Constructing from an (ip, prefixlen) tuple.
# which converts into a formatted IP prefix string. if isinstance(address, tuple):
addr = str(address).split('/') try:
ip, prefixlen = address
except ValueError:
raise AddressValueError(address)
self.ip = IPv6Address(ip)
self._ip = self.ip._ip
self._prefixlen = self._prefix_from_prefix_int(prefixlen)
if len(addr) > 2:
raise AddressValueError(address)
self._ip = self._ip_int_from_string(addr[0])
self.ip = IPv6Address(self._ip)
if len(addr) == 2:
# This may raise NetmaskValueError
self._prefixlen = self._prefix_from_prefix_string(addr[1])
else: else:
self._prefixlen = self._max_prefixlen # Assume input argument to be string or any object representation
# which converts into a formatted IP prefix string.
addr = str(address).split('/')
if len(addr) > 2:
raise AddressValueError(address)
self._ip = self._ip_int_from_string(addr[0])
self.ip = IPv6Address(self._ip)
if len(addr) == 2:
# This may raise NetmaskValueError
self._prefixlen = self._prefix_from_prefix_string(addr[1])
else:
self._prefixlen = self._max_prefixlen
self.netmask = IPv6Address(self._ip_int_from_prefix(self._prefixlen)) self.netmask = IPv6Address(self._ip_int_from_prefix(self._prefixlen))
+1 -1
View File
@@ -31,7 +31,7 @@ from gluon.contrib.redis_utils import RConn
from gluon.contrib.redis_scheduler import RScheduler from gluon.contrib.redis_scheduler import RScheduler
def demo1(*args,**vars): def demo1(*args,**vars):
print('you passed args=%s and vars=%s') % (args, vars) print('you passed args=%s and vars=%s' % (args, vars))
return 'done!' return 'done!'
def demo2(): def demo2():
+1 -1
View File
@@ -41,7 +41,7 @@ Create File: app/models/scheduler.py ======
from gluon.scheduler import Scheduler from gluon.scheduler import Scheduler
def demo1(*args,**vars): def demo1(*args,**vars):
print('you passed args=%s and vars=%s') % (args, vars) print('you passed args=%s and vars=%s' % (args, vars))
return 'done!' return 'done!'
def demo2(): def demo2():
+1 -1
View File
@@ -13,13 +13,13 @@ from .test_html import *
from .test_contribs import * from .test_contribs import *
from .test_routes import * from .test_routes import *
from .test_router import * from .test_router import *
from .test_validators import *
if sys.version[:3] == '2.7': if sys.version[:3] == '2.7':
from .test_compileapp import * from .test_compileapp import *
from .test_is_url import * from .test_is_url import *
from .test_languages import * from .test_languages import *
from .test_serializers import * from .test_serializers import *
from .test_validators import *
from .test_utils import * from .test_utils import *
from .test_tools import * from .test_tools import *
from .test_appadmin import * from .test_appadmin import *
+1 -1
View File
@@ -307,7 +307,7 @@ class TestsForSchedulerRunner(testForSchedulerRunnerBase):
self.db.commit() self.db.commit()
self.writefunction(r""" self.writefunction(r"""
def demo1(*args,**vars): def demo1(*args,**vars):
print('you passed args=%s and vars=%s') % (args, vars) print('you passed args=%s and vars=%s' % (args, vars))
return args[0] return args[0]
def demo4(): def demo4():
+35 -21
View File
@@ -13,7 +13,7 @@ fix_sys_path(__file__)
from gluon.validators import * from gluon.validators import *
from gluon._compat import PY2, to_bytes
class TestValidators(unittest.TestCase): class TestValidators(unittest.TestCase):
@@ -55,7 +55,10 @@ class TestValidators(unittest.TestCase):
rtn = IS_MATCH('^.hell$', strict=True)('shell') rtn = IS_MATCH('^.hell$', strict=True)('shell')
self.assertEqual(rtn, ('shell', None)) self.assertEqual(rtn, ('shell', None))
rtn = IS_MATCH(u'hell', is_unicode=True)('àòè') rtn = IS_MATCH(u'hell', is_unicode=True)('àòè')
self.assertEqual(rtn, ('\xc3\xa0\xc3\xb2\xc3\xa8', 'Invalid expression')) if PY2:
self.assertEqual(rtn, ('\xc3\xa0\xc3\xb2\xc3\xa8', 'Invalid expression'))
else:
self.assertEqual(rtn, ('àòè', 'Invalid expression'))
rtn = IS_MATCH(u'hell', is_unicode=True)(u'hell') rtn = IS_MATCH(u'hell', is_unicode=True)(u'hell')
self.assertEqual(rtn, (u'hell', None)) self.assertEqual(rtn, (u'hell', None))
rtn = IS_MATCH('hell', is_unicode=True)(u'hell') rtn = IS_MATCH('hell', is_unicode=True)(u'hell')
@@ -111,9 +114,15 @@ class TestValidators(unittest.TestCase):
self.assertEqual(rtn, (cpstr, 'Enter from 0 to 3 characters')) self.assertEqual(rtn, (cpstr, 'Enter from 0 to 3 characters'))
# test unicode # test unicode
rtn = IS_LENGTH(2)(u'°2') rtn = IS_LENGTH(2)(u'°2')
self.assertEqual(rtn, ('\xc2\xb02', None)) if PY2:
self.assertEqual(rtn, ('\xc2\xb02', None))
else:
self.assertEqual(rtn, (u'°2', None))
rtn = IS_LENGTH(2)(u'°12') rtn = IS_LENGTH(2)(u'°12')
self.assertEqual(rtn, (u'\xb012', 'Enter from 0 to 2 characters')) if PY2:
self.assertEqual(rtn, (u'\xb012', 'Enter from 0 to 2 characters'))
else:
self.assertEqual(rtn, (u'°12', 'Enter from 0 to 2 characters'))
# test automatic str() # test automatic str()
rtn = IS_LENGTH(minsize=1)(1) rtn = IS_LENGTH(minsize=1)(1)
self.assertEqual(rtn, ('1', None)) self.assertEqual(rtn, ('1', None))
@@ -121,19 +130,19 @@ class TestValidators(unittest.TestCase):
self.assertEqual(rtn, (1, 'Enter from 2 to 255 characters')) self.assertEqual(rtn, (1, 'Enter from 2 to 255 characters'))
# test FieldStorage # test FieldStorage
import cgi import cgi
from StringIO import StringIO from io import BytesIO
a = cgi.FieldStorage() a = cgi.FieldStorage()
a.file = StringIO('abc') a.file = BytesIO(b'abc')
rtn = IS_LENGTH(minsize=4)(a) rtn = IS_LENGTH(minsize=4)(a)
self.assertEqual(rtn, (a, 'Enter from 4 to 255 characters')) self.assertEqual(rtn, (a, 'Enter from 4 to 255 characters'))
urlencode_data = "key2=value2x&key3=value3&key4=value4" urlencode_data = b"key2=value2x&key3=value3&key4=value4"
urlencode_environ = { urlencode_environ = {
'CONTENT_LENGTH': str(len(urlencode_data)), 'CONTENT_LENGTH': str(len(urlencode_data)),
'CONTENT_TYPE': 'application/x-www-form-urlencoded', 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
'QUERY_STRING': 'key1=value1&key2=value2y', 'QUERY_STRING': 'key1=value1&key2=value2y',
'REQUEST_METHOD': 'POST', 'REQUEST_METHOD': 'POST',
} }
fake_stdin = StringIO(urlencode_data) fake_stdin = BytesIO(urlencode_data)
fake_stdin.seek(0) fake_stdin.seek(0)
a = cgi.FieldStorage(fp=fake_stdin, environ=urlencode_environ) a = cgi.FieldStorage(fp=fake_stdin, environ=urlencode_environ)
rtn = IS_LENGTH(minsize=6)(a) rtn = IS_LENGTH(minsize=6)(a)
@@ -692,15 +701,15 @@ class TestValidators(unittest.TestCase):
def test_IS_LOWER(self): def test_IS_LOWER(self):
rtn = IS_LOWER()('ABC') rtn = IS_LOWER()('ABC')
self.assertEqual(rtn, ('abc', None)) self.assertEqual(rtn, (b'abc', None))
rtn = IS_LOWER()('Ñ') rtn = IS_LOWER()('Ñ')
self.assertEqual(rtn, ('\xc3\xb1', None)) self.assertEqual(rtn, (b'\xc3\xb1', None))
def test_IS_UPPER(self): def test_IS_UPPER(self):
rtn = IS_UPPER()('abc') rtn = IS_UPPER()('abc')
self.assertEqual(rtn, ('ABC', None)) self.assertEqual(rtn, (b'ABC', None))
rtn = IS_UPPER()('ñ') rtn = IS_UPPER()('ñ')
self.assertEqual(rtn, ('\xc3\x91', None)) self.assertEqual(rtn, (b'\xc3\x91', None))
def test_IS_SLUG(self): def test_IS_SLUG(self):
rtn = IS_SLUG()('abc123') rtn = IS_SLUG()('abc123')
@@ -821,7 +830,10 @@ class TestValidators(unittest.TestCase):
rtn = IS_STRONG(es=True, entropy=100)('a1d') rtn = IS_STRONG(es=True, entropy=100)('a1d')
self.assertEqual(rtn, ('a1d', 'Entropy (15.97) less than required (100)')) self.assertEqual(rtn, ('a1d', 'Entropy (15.97) less than required (100)'))
rtn = IS_STRONG(es=True, entropy=100)('añd') rtn = IS_STRONG(es=True, entropy=100)('añd')
self.assertEqual(rtn, ('a\xc3\xb1d', 'Entropy (18.13) less than required (100)')) if PY2:
self.assertEqual(rtn, ('a\xc3\xb1d', 'Entropy (18.13) less than required (100)'))
else:
self.assertEqual(rtn, ('añd', 'Entropy (18.13) less than required (100)'))
rtn = IS_STRONG()('********') rtn = IS_STRONG()('********')
self.assertEqual(rtn, ('********', None)) self.assertEqual(rtn, ('********', None))
rtn = IS_STRONG(es=True, max=4)('abcde') rtn = IS_STRONG(es=True, max=4)('abcde')
@@ -855,10 +867,10 @@ class TestValidators(unittest.TestCase):
class DummyImageFile(object): class DummyImageFile(object):
def __init__(self, filename, ext, width, height): def __init__(self, filename, ext, width, height):
from StringIO import StringIO from io import BytesIO
import struct import struct
self.filename = filename + '.' + ext self.filename = filename + '.' + ext
self.file = StringIO() self.file = BytesIO()
if ext == 'bmp': if ext == 'bmp':
self.file.write(b'BM') self.file.write(b'BM')
self.file.write(b' ' * 16) self.file.write(b' ' * 16)
@@ -915,7 +927,7 @@ class TestValidators(unittest.TestCase):
def test_IS_UPLOAD_FILENAME(self): def test_IS_UPLOAD_FILENAME(self):
import cgi import cgi
from StringIO import StringIO from io import BytesIO
def gen_fake(filename): def gen_fake(filename):
formdata_file_data = """ formdata_file_data = """
@@ -937,7 +949,7 @@ this is the content of the fake file
'QUERY_STRING': 'key1=value1&key2=value2x', 'QUERY_STRING': 'key1=value1&key2=value2x',
'REQUEST_METHOD': 'POST', 'REQUEST_METHOD': 'POST',
} }
return cgi.FieldStorage(fp=StringIO(formdata_file_data), environ=formdata_file_environ)['file_attach'] return cgi.FieldStorage(fp=BytesIO(to_bytes(formdata_file_data)), environ=formdata_file_environ)['file_attach']
fake = gen_fake('example.pdf') fake = gen_fake('example.pdf')
rtn = IS_UPLOAD_FILENAME(extension='pdf')(fake) rtn = IS_UPLOAD_FILENAME(extension='pdf')(fake)
@@ -1016,8 +1028,12 @@ this is the content of the fake file
self.assertEqual(rtn, ('2001::126c:8ffa:fe22:b3af', 'Enter valid IPv6 address')) self.assertEqual(rtn, ('2001::126c:8ffa:fe22:b3af', 'Enter valid IPv6 address'))
rtn = IS_IPV6(is_multicast=True)('ff00::126c:8ffa:fe22:b3af') rtn = IS_IPV6(is_multicast=True)('ff00::126c:8ffa:fe22:b3af')
self.assertEqual(rtn, ('ff00::126c:8ffa:fe22:b3af', None)) self.assertEqual(rtn, ('ff00::126c:8ffa:fe22:b3af', None))
rtn = IS_IPV6(is_routeable=True)('2001::126c:8ffa:fe22:b3af') # TODO:
self.assertEqual(rtn, ('2001::126c:8ffa:fe22:b3af', None)) # with py3.ipaddress '2001::126c:8ffa:fe22:b3af' is considered private
# with py2.ipaddress '2001::126c:8ffa:fe22:b3af' is considered private
# with gluon.contrib.ipaddr(both current and trunk) is not considered private
# rtn = IS_IPV6(is_routeable=True)('2001::126c:8ffa:fe22:b3af')
# self.assertEqual(rtn, ('2001::126c:8ffa:fe22:b3af', None))
rtn = IS_IPV6(is_routeable=True)('ff00::126c:8ffa:fe22:b3af') rtn = IS_IPV6(is_routeable=True)('ff00::126c:8ffa:fe22:b3af')
self.assertEqual(rtn, ('ff00::126c:8ffa:fe22:b3af', 'Enter valid IPv6 address')) self.assertEqual(rtn, ('ff00::126c:8ffa:fe22:b3af', 'Enter valid IPv6 address'))
rtn = IS_IPV6(subnets='2001::/32')('2001::8ffa:fe22:b3af') rtn = IS_IPV6(subnets='2001::/32')('2001::8ffa:fe22:b3af')
@@ -1091,8 +1107,6 @@ this is the content of the fake file
self.assertEqual(rtn, ('2001::126c:8ffa:fe22:b3af', 'Enter valid IP address')) self.assertEqual(rtn, ('2001::126c:8ffa:fe22:b3af', 'Enter valid IP address'))
rtn = IS_IPADDRESS(is_multicast=True)('ff00::126c:8ffa:fe22:b3af') rtn = IS_IPADDRESS(is_multicast=True)('ff00::126c:8ffa:fe22:b3af')
self.assertEqual(rtn, ('ff00::126c:8ffa:fe22:b3af', None)) self.assertEqual(rtn, ('ff00::126c:8ffa:fe22:b3af', None))
rtn = IS_IPADDRESS(is_routeable=True)('2001::126c:8ffa:fe22:b3af')
self.assertEqual(rtn, ('2001::126c:8ffa:fe22:b3af', None))
rtn = IS_IPADDRESS(is_routeable=True)('ff00::126c:8ffa:fe22:b3af') rtn = IS_IPADDRESS(is_routeable=True)('ff00::126c:8ffa:fe22:b3af')
self.assertEqual(rtn, ('ff00::126c:8ffa:fe22:b3af', 'Enter valid IP address')) self.assertEqual(rtn, ('ff00::126c:8ffa:fe22:b3af', 'Enter valid IP address'))
rtn = IS_IPADDRESS(subnets='2001::/32')('2001::8ffa:fe22:b3af') rtn = IS_IPADDRESS(subnets='2001::/32')('2001::8ffa:fe22:b3af')
+3 -3
View File
@@ -23,7 +23,7 @@ import logging
import socket import socket
import base64 import base64
import zlib import zlib
from gluon._compat import basestring, pickle, PY2, xrange, to_bytes from gluon._compat import basestring, pickle, PY2, xrange, to_bytes, to_native
_struct_2_long_long = struct.Struct('=QQ') _struct_2_long_long = struct.Struct('=QQ')
@@ -102,8 +102,8 @@ def simple_hash(text, key='', salt='', digest_alg='md5'):
h = digest_alg(text + key + salt) h = digest_alg(text + key + salt)
elif digest_alg.startswith('pbkdf2'): # latest and coolest! elif digest_alg.startswith('pbkdf2'): # latest and coolest!
iterations, keylen, alg = digest_alg[7:-1].split(',') iterations, keylen, alg = digest_alg[7:-1].split(',')
return pbkdf2_hex(text, salt, int(iterations), return to_native(pbkdf2_hex(text, salt, int(iterations),
int(keylen), get_digest(alg)) int(keylen), get_digest(alg)))
elif key: # use hmac elif key: # use hmac
digest_alg = get_digest(digest_alg) digest_alg = get_digest(digest_alg)
h = hmac.new(key + salt, text, digest_alg) h = hmac.new(key + salt, text, digest_alg)
+33 -32
View File
@@ -21,7 +21,7 @@ import urllib
import struct import struct
import decimal import decimal
import unicodedata import unicodedata
from gluon._compat import StringIO, long, unicodeT, to_unicode, urllib_unquote, unichr, to_bytes from gluon._compat import StringIO, long, unicodeT, to_unicode, urllib_unquote, unichr, to_bytes, PY2, to_unicode, to_native
from gluon.utils import simple_hash, web2py_uuid, DIGEST_ALG_BY_SIZE from gluon.utils import simple_hash, web2py_uuid, DIGEST_ALG_BY_SIZE
from pydal.objects import Field, FieldVirtual, FieldMethod from pydal.objects import Field, FieldVirtual, FieldMethod
from functools import reduce from functools import reduce
@@ -192,10 +192,13 @@ class IS_MATCH(Validator):
self.regex = re.compile(expression) self.regex = re.compile(expression)
self.error_message = error_message self.error_message = error_message
self.extract = extract self.extract = extract
self.is_unicode = is_unicode self.is_unicode = is_unicode or (not(PY2))
def __call__(self, value): def __call__(self, value):
if self.is_unicode: if not(PY2): # PY3 convert bytes to unicode
value = to_unicode(value)
if self.is_unicode or not(PY2):
if not isinstance(value, unicodeT): if not isinstance(value, unicodeT):
match = self.regex.search(str(value).decode('utf8')) match = self.regex.search(str(value).decode('utf8'))
else: else:
@@ -267,7 +270,7 @@ class IS_EXPR(Validator):
return (value, self.expression(value)) return (value, self.expression(value))
# for backward compatibility # for backward compatibility
self.environment.update(value=value) self.environment.update(value=value)
exec ('__ret__=' + self.expression) in self.environment exec ('__ret__=' + self.expression, self.environment)
if self.environment['__ret__']: if self.environment['__ret__']:
return (value, None) return (value, None)
return (value, translate(self.error_message)) return (value, translate(self.error_message))
@@ -333,7 +336,7 @@ class IS_LENGTH(Validator):
return (value, None) return (value, None)
elif isinstance(value, str): elif isinstance(value, str):
try: try:
lvalue = len(value.decode('utf8')) lvalue = len(to_unicode(value))
except: except:
lvalue = len(value) lvalue = len(value)
if self.minsize <= lvalue <= self.maxsize: if self.minsize <= lvalue <= self.maxsize:
@@ -341,6 +344,9 @@ class IS_LENGTH(Validator):
elif isinstance(value, unicodeT): elif isinstance(value, unicodeT):
if self.minsize <= len(value) <= self.maxsize: if self.minsize <= len(value) <= self.maxsize:
return (value.encode('utf8'), None) return (value.encode('utf8'), None)
elif isinstance(value, (bytes, bytearray)):
if self.minsize <= len(value) <= self.maxsize:
return (value, None)
elif isinstance(value, (tuple, list)): elif isinstance(value, (tuple, list)):
if self.minsize <= len(value) <= self.maxsize: if self.minsize <= len(value) <= self.maxsize:
return (value, None) return (value, None)
@@ -448,7 +454,7 @@ class IS_IN_SET(Validator):
else: else:
items = [(k, self.labels[i]) for (i, k) in enumerate(self.theset)] items = [(k, self.labels[i]) for (i, k) in enumerate(self.theset)]
if self.sort: if self.sort:
items.sort(options_sorter) items.sort(key=lambda o: str(o[1]).upper())
if zero and not self.zero is None and not self.multiple: if zero and not self.zero is None and not self.multiple:
items.insert(0, ('', self.zero)) items.insert(0, ('', self.zero))
return items return items
@@ -594,7 +600,7 @@ class IS_IN_DB(Validator):
self.build_set() self.build_set()
items = [(k, self.labels[i]) for (i, k) in enumerate(self.theset)] items = [(k, self.labels[i]) for (i, k) in enumerate(self.theset)]
if self.sort: if self.sort:
items.sort(options_sorter) items.sort(key=lambda o: str(o[1]).upper())
if zero and self.zero is not None and not self.multiple: if zero and self.zero is not None and not self.multiple:
items.insert(0, ('', self.zero)) items.insert(0, ('', self.zero))
return items return items
@@ -717,10 +723,7 @@ class IS_NOT_IN_DB(Validator):
self.record_id = id self.record_id = id
def __call__(self, value): def __call__(self, value):
if isinstance(value, unicodeT): value = to_native(str(value))
value = value.encode('utf8')
else:
value = str(value)
if not value.strip(): if not value.strip():
return (value, translate(self.error_message)) return (value, translate(self.error_message))
if value in self.allowed_override: if value in self.allowed_override:
@@ -1455,7 +1458,7 @@ def unicode_to_ascii_authority(authority):
import encodings.idna import encodings.idna
for label in labels: for label in labels:
if label: if label:
asciiLabels.append(encodings.idna.ToASCII(label)) asciiLabels.append(to_native(encodings.idna.ToASCII(label)))
else: else:
# encodings.idna.ToASCII does not accept an empty string, but # encodings.idna.ToASCII does not accept an empty string, but
# it is necessary for us to allow for empty labels so that we # it is necessary for us to allow for empty labels so that we
@@ -1525,6 +1528,7 @@ def unicode_to_ascii_url(url, prepend_scheme):
scheme = str(scheme) + '://' scheme = str(scheme) + '://'
else: else:
scheme = '' scheme = ''
return scheme + unicode_to_ascii_authority(authority) +\ return scheme + unicode_to_ascii_authority(authority) +\
escape_unicode(path) + escape_unicode(query) + str(fragment) escape_unicode(path) + escape_unicode(query) + str(fragment)
@@ -2083,7 +2087,6 @@ class IS_URL(Validator):
may be modified to (1) prepend a scheme, and/or (2) convert a may be modified to (1) prepend a scheme, and/or (2) convert a
non-compliant unicode URL into a compliant US-ASCII version. non-compliant unicode URL into a compliant US-ASCII version.
""" """
if self.mode == 'generic': if self.mode == 'generic':
subMethod = IS_GENERIC_URL(error_message=self.error_message, subMethod = IS_GENERIC_URL(error_message=self.error_message,
allowed_schemes=self.allowed_schemes, allowed_schemes=self.allowed_schemes,
@@ -2101,7 +2104,7 @@ class IS_URL(Validator):
else: else:
try: try:
asciiValue = unicode_to_ascii_url(value, self.prepend_scheme) asciiValue = unicode_to_ascii_url(value, self.prepend_scheme)
except Exception: except Exception as e:
# If we are not able to convert the unicode url into a # If we are not able to convert the unicode url into a
# US-ASCII URL, then the URL is not valid # US-ASCII URL, then the URL is not valid
return (value, translate(self.error_message)) return (value, translate(self.error_message))
@@ -2477,7 +2480,7 @@ class IS_LOWER(Validator):
""" """
def __call__(self, value): def __call__(self, value):
return (value.decode('utf8').lower().encode('utf8'), None) return (to_bytes(to_unicode(value).lower()), None)
class IS_UPPER(Validator): class IS_UPPER(Validator):
@@ -2492,7 +2495,7 @@ class IS_UPPER(Validator):
""" """
def __call__(self, value): def __call__(self, value):
return (value.decode('utf8').upper().encode('utf8'), None) return (to_bytes(to_unicode(value).upper()), None)
def urlify(s, maxlen=80, keep_underscores=False): def urlify(s, maxlen=80, keep_underscores=False):
@@ -2501,11 +2504,10 @@ def urlify(s, maxlen=80, keep_underscores=False):
if (keep_underscores): underscores are retained in the string if (keep_underscores): underscores are retained in the string
else: underscores are translated to hyphens (default) else: underscores are translated to hyphens (default)
""" """
if isinstance(s, str): s = to_unicode(s) # to unicode
s = s.decode('utf-8') # to unicode
s = s.lower() # to lowercase s = s.lower() # to lowercase
s = unicodedata.normalize('NFKD', s) # replace special characters s = unicodedata.normalize('NFKD', s) # replace special characters
s = s.encode('ascii', 'ignore') # encode as ASCII s = to_native(s, charset='ascii', errors='ignore') # encode as ASCII
s = re.sub('&\w+?;', '', s) # strip html entities s = re.sub('&\w+?;', '', s) # strip html entities
if keep_underscores: if keep_underscores:
s = re.sub('\s+', '-', s) # whitespace to hyphens s = re.sub('\s+', '-', s) # whitespace to hyphens
@@ -2912,8 +2914,7 @@ def calc_entropy(string):
other = set() other = set()
seen = set() seen = set()
lastset = None lastset = None
if isinstance(string, str): string = to_unicode(string)
string = unicode(string, encoding='utf8')
for c in string: for c in string:
# classify this character # classify this character
inset = otherset inset = otherset
@@ -3057,7 +3058,7 @@ class IS_STRONG(object):
if not self.error_message: if not self.error_message:
if self.estring: if self.estring:
return (value, '|'.join(failures)) return (value, '|'.join(failures))
from html import XML from gluon.html import XML
return (value, XML('<br />'.join(failures))) return (value, XML('<br />'.join(failures)))
else: else:
return (value, translate(self.error_message)) return (value, translate(self.error_message))
@@ -3134,24 +3135,24 @@ class IS_IMAGE(Validator):
and self.minsize[1] <= height <= self.maxsize[1] and self.minsize[1] <= height <= self.maxsize[1]
value.file.seek(0) value.file.seek(0)
return (value, None) return (value, None)
except: except Exception as e:
return (value, translate(self.error_message)) return (value, translate(self.error_message))
def __bmp(self, stream): def __bmp(self, stream):
if stream.read(2) == 'BM': if stream.read(2) == b'BM':
stream.read(16) stream.read(16)
return struct.unpack("<LL", stream.read(8)) return struct.unpack("<LL", stream.read(8))
return (-1, -1) return (-1, -1)
def __gif(self, stream): def __gif(self, stream):
if stream.read(6) in ('GIF87a', 'GIF89a'): if stream.read(6) in (b'GIF87a', b'GIF89a'):
stream = stream.read(5) stream = stream.read(5)
if len(stream) == 5: if len(stream) == 5:
return tuple(struct.unpack("<HHB", stream)[:-1]) return tuple(struct.unpack("<HHB", stream)[:-1])
return (-1, -1) return (-1, -1)
def __jpeg(self, stream): def __jpeg(self, stream):
if stream.read(2) == '\xFF\xD8': if stream.read(2) == b'\xFF\xD8':
while True: while True:
(marker, code, length) = struct.unpack("!BBH", stream.read(4)) (marker, code, length) = struct.unpack("!BBH", stream.read(4))
if marker != 0xFF: if marker != 0xFF:
@@ -3164,9 +3165,9 @@ class IS_IMAGE(Validator):
return (-1, -1) return (-1, -1)
def __png(self, stream): def __png(self, stream):
if stream.read(8) == '\211PNG\r\n\032\n': if stream.read(8) == b'\211PNG\r\n\032\n':
stream.read(4) stream.read(4)
if stream.read(4) == "IHDR": if stream.read(4) == b"IHDR":
return struct.unpack("!LL", stream.read(8)) return struct.unpack("!LL", stream.read(8))
return (-1, -1) return (-1, -1)
@@ -3367,7 +3368,7 @@ class IS_IPV4(Validator):
if isinstance(value, str): if isinstance(value, str):
temp.append(value.split('.')) temp.append(value.split('.'))
elif isinstance(value, (list, tuple)): elif isinstance(value, (list, tuple)):
if len(value) == len(filter(lambda item: isinstance(item, int), value)) == 4: if len(value) == len(list(filter(lambda item: isinstance(item, int), value))) == 4:
temp.append(value) temp.append(value)
else: else:
for item in value: for item in value:
@@ -3510,7 +3511,7 @@ class IS_IPV6(Validator):
from gluon.contrib import ipaddr as ipaddress from gluon.contrib import ipaddr as ipaddress
try: try:
ip = ipaddress.IPv6Address(value.decode('utf-8')) ip = ipaddress.IPv6Address(to_unicode(value))
ok = True ok = True
except ipaddress.AddressValueError: except ipaddress.AddressValueError:
return (value, translate(self.error_message)) return (value, translate(self.error_message))
@@ -3522,7 +3523,7 @@ class IS_IPV6(Validator):
self.subnets = [self.subnets] self.subnets = [self.subnets]
for network in self.subnets: for network in self.subnets:
try: try:
ipnet = ipaddress.IPv6Network(network.decode('utf-8')) ipnet = ipaddress.IPv6Network(to_unicode(network))
except (ipaddress.NetmaskValueError, ipaddress.AddressValueError): except (ipaddress.NetmaskValueError, ipaddress.AddressValueError):
return (value, translate('invalid subnet provided')) return (value, translate('invalid subnet provided'))
if ip in ipnet: if ip in ipnet:
@@ -3739,7 +3740,7 @@ class IS_IPADDRESS(Validator):
IPv6Address) IPv6Address)
try: try:
ip = IPAddress(value.decode('utf-8')) ip = IPAddress(to_unicode(value))
except ValueError: except ValueError:
return (value, translate(self.error_message)) return (value, translate(self.error_message))