dinislage/web2py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

many pep8 improvements

Browse Source
This commit is contained in:
mdipierro
2012-10-19 12:33:53 -05:00
parent b716df1a05
commit 4b2ba185ae
117 changed files with 7160 additions and 5670 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gluon/utils.py
+66 -53
View File
@@ -38,7 +38,8 @@ except ImportError:
logger = logging.getLogger("web2py")
def compare(a,b):
def compare(a, b):
""" compares two strings and not vulnerable to timing attacks """
if len(a) != len(b):
return False
@@ -47,36 +48,39 @@ def compare(a,b):
result |= ord(x) ^ ord(y)
return result == 0
def md5_hash(text):
""" Generate a md5 hash with the given text """
return hashlib.md5(text).hexdigest()
def simple_hash(text, key='', salt = '', digest_alg = 'md5'):
def simple_hash(text, key='', salt='', digest_alg='md5'):
"""
Generates hash with the given text using the specified
digest hashing algorithm
"""
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!
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 pbkdf2_hex(text, salt, int(iterations),
int(keylen),get_digest(alg))
elif key: # use hmac
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 = hmac.new(key + salt, text, digest_alg)
else: # compatible with third party systems
h = hashlib.new(digest_alg)
h.update(text+salt)
h.update(text + salt)
return h.hexdigest()
def get_digest(value):
"""
Returns a hashlib digest algorithm from a string
"""
if not isinstance(value,str):
if not isinstance(value, str):
return value
value = value.lower()
if value == "md5":
@@ -95,50 +99,55 @@ def get_digest(value):
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',
}
128 / 4: 'md5',
160 / 4: 'sha1',
224 / 4: 'sha224',
256 / 4: 'sha256',
384 / 4: 'sha384',
512 / 4: 'sha512',
}
def pad(s,n=32,padchar='.'):
def pad(s, n=32, padchar='.'):
return s + (32 - len(s) % 32) * padchar
def secure_dumps(data,encryption_key,hash_key=None,compression_level=None):
def secure_dumps(data, encryption_key, hash_key=None, compression_level=None):
if not hash_key:
hash_key = hashlib.sha1(encryption_key).hexdigest()
dump = cPickle.dumps(data)
if compression_level:
dump = zlib.compress(dump, compression_level)
key = pad(encryption_key[:32])
cipher = AES.new(key,IV=key[:16])
cipher = AES.new(key, IV=key[:16])
encrypted_data = base64.urlsafe_b64encode(cipher.encrypt(pad(dump)))
signature = hmac.new(hash_key,encrypted_data).hexdigest()
return signature+':'+encrypted_data
signature = hmac.new(hash_key, encrypted_data).hexdigest()
return signature + ':' + encrypted_data
def secure_loads(data,encryption_key,hash_key=None, compression_level=None):
def secure_loads(data, encryption_key, hash_key=None, compression_level=None):
if not ':' in data:
return None
if not hash_key:
hash_key = hashlib.sha1(encryption_key).hexdigest()
signature, encrypted_data = data.split(':',1)
actual_signature = hmac.new(hash_key,encrypted_data).hexdigest()
if signature!=actual_signature:
signature, encrypted_data = data.split(':', 1)
actual_signature = hmac.new(hash_key, encrypted_data).hexdigest()
if signature != actual_signature:
return None
key = pad(encryption_key[:32])
cipher = AES.new(key,IV=key[:16])
cipher = AES.new(key, IV=key[:16])
try:
data = cipher.decrypt(base64.urlsafe_b64decode(encrypted_data))
data = data.rstrip(' ')
if compression_level:
data = zlib.decompress(data)
return cPickle.loads(data)
except (TypeError,cPickle.UnpicklingError):
except (TypeError, cPickle.UnpicklingError):
return None
### compute constant CTOKENS
def initialize_urandom():
"""
This function and the web2py_uuid follow from the following discussion:
@@ -154,14 +163,15 @@ def initialize_urandom():
"""
node_id = uuid.getnode()
microseconds = int(time.time() * 1e6)
ctokens = [((node_id + microseconds) >> ((i%6)*8)) % 256 for i in range(16)]
ctokens = [((node_id + microseconds) >> ((i % 6) * 8)) %
256 for i in range(16)]
random.seed(node_id + microseconds)
try:
os.urandom(1)
have_urandom = True
try:
# try to add process-specific entropy
frandom = open('/dev/urandom','wb')
frandom = open('/dev/urandom', 'wb')
try:
frandom.write(''.join(chr(t) for t in ctokens))
finally:
@@ -172,15 +182,16 @@ def initialize_urandom():
except NotImplementedError:
have_urandom = False
logger.warning(
"""Cryptographically secure session management is not possible on your system because
"""Cryptographically secure session management is not possible on your system because
your system does not provide a cryptographically secure entropy source.
This is not specific to web2py; consider deploying on a different operating system.""")
unpacked_ctokens = struct.unpack('=QQ',string.join(
(chr(x) for x in ctokens),''))
unpacked_ctokens = struct.unpack('=QQ', string.join(
(chr(x) for x in ctokens), ''))
return unpacked_ctokens, have_urandom
UNPACKED_CTOKENS, HAVE_URANDOM = initialize_urandom()
def fast_urandom16(urandom=[], locker = threading.RLock()):
def fast_urandom16(urandom=[], locker=threading.RLock()):
"""
this is 4x faster than calling os.urandom(16) and prevents
the "too many files open" issue with concurrent access to os.urandom()
@@ -190,12 +201,13 @@ def fast_urandom16(urandom=[], locker = threading.RLock()):
except IndexError:
try:
locker.acquire()
ur = os.urandom(16*1024)
urandom += [ur[i:i+16] for i in xrange(16,1024*16,16)]
ur = os.urandom(16 * 1024)
urandom += [ur[i:i + 16] for i in xrange(16, 1024 * 16, 16)]
return ur[0:16]
finally:
locker.release()
def web2py_uuid(ctokens=UNPACKED_CTOKENS):
"""
This function follows from the following discussion:
@@ -204,20 +216,21 @@ def web2py_uuid(ctokens=UNPACKED_CTOKENS):
It works like uuid.uuid4 except that tries to use os.urandom() if possible
and it XORs the output with the tokens uniquely associated with this machine.
"""
rand_longs = (random.getrandbits(64),random.getrandbits(64))
rand_longs = (random.getrandbits(64), random.getrandbits(64))
if HAVE_URANDOM:
urand_longs = struct.unpack('=QQ', fast_urandom16())
byte_s = struct.pack('=QQ',
rand_longs[0]^urand_longs[0]^ctokens[0],
rand_longs[1]^urand_longs[1]^ctokens[1])
rand_longs[0] ^ urand_longs[0] ^ ctokens[0],
rand_longs[1] ^ urand_longs[1] ^ ctokens[1])
else:
byte_s = struct.pack('=QQ',
rand_longs[0]^ctokens[0],
rand_longs[1]^ctokens[1])
rand_longs[0] ^ ctokens[0],
rand_longs[1] ^ ctokens[1])
return str(uuid.UUID(bytes=byte_s, version=4))
REGEX_IPv4 = re.compile('(\d+)\.(\d+)\.(\d+)\.(\d+)')
def is_valid_ip_address(address):
"""
>>> is_valid_ip_address('127.0')
@@ -228,29 +241,29 @@ def is_valid_ip_address(address):
True
"""
# deal with special cases
if address.lower() in ('127.0.0.1','localhost','::1','::ffff:127.0.0.1'):
if address.lower() in ('127.0.0.1', 'localhost', '::1', '::ffff:127.0.0.1'):
return True
elif address.lower() in ('unkown',''):
elif address.lower() in ('unkown', ''):
return False
elif address.count('.')==3: # assume IPv4
elif address.count('.') == 3: # assume IPv4
if address.startswith('::ffff:'):
address = address[7:]
if hasattr(socket,'inet_aton'): # try validate using the OS
if hasattr(socket, 'inet_aton'): # try validate using the OS
try:
addr = socket.inet_aton(address)
return True
except socket.error: # invalid address
except socket.error: # invalid address
return False
else: # try validate using Regex
else: # try validate using Regex
match = REGEX_IPv4.match(address)
if match and all(0<=int(match.group(i))<256 for i in (1,2,3,4)):
if match and all(0 <= int(match.group(i)) < 256 for i in (1, 2, 3, 4)):
return True
return False
elif hasattr(socket,'inet_pton'): # assume IPv6, try using the OS
elif hasattr(socket, 'inet_pton'): # assume IPv6, try using the OS
try:
addr = socket.inet_pton(socket.AF_INET6, address)
return True
except socket.error: # invalid address
except socket.error: # invalid address
return False
else: # do not know what to do? assume it is a valid address
else: # do not know what to do? assume it is a valid address
return True