|
Package Dropbox ::
Package web2py ::
Package gluon ::
Module utils
|
|
1
2
3
4 """
5 This file is part of the web2py Web Framework
6 Copyrighted by Massimo Di Pierro <mdipierro@cs.depaul.edu>
7 License: LGPLv3 (http://www.gnu.org/licenses/lgpl.html)
8
9 This file specifically includes utilities for security.
10 """
11
12 import threading
13 import struct
14 import hashlib
15 import hmac
16 import uuid
17 import random
18 import time
19 import os
20 import re
21 import sys
22 import logging
23 import socket
24 import base64
25 import zlib
26
27 python_version = sys.version_info[0]
28
29 if python_version == 2:
30 import cPickle as pickle
31 else:
32 import pickle
33
34
35 try:
36 from Crypto.Cipher import AES
37 except ImportError:
38 import contrib.aes as AES
39
40 try:
41 from contrib.pbkdf2 import pbkdf2_hex
42 HAVE_PBKDF2 = True
43 except ImportError:
44 try:
45 from .pbkdf2 import pbkdf2_hex
46 HAVE_PBKDF2 = True
47 except (ImportError, ValueError):
48 HAVE_PBKDF2 = False
49
50 logger = logging.getLogger("web2py")
51
52
54 """ compares two strings and not vulnerable to timing attacks """
55 if len(a) != len(b):
56 return False
57 result = 0
58 for x, y in zip(a, b):
59 result |= ord(x) ^ ord(y)
60 return result == 0
61
62
64 """ Generate a md5 hash with the given text """
65 return hashlib.md5(text).hexdigest()
66
67
68 -def simple_hash(text, key='', salt='', digest_alg='md5'):
69 """
70 Generates hash with the given text using the specified
71 digest hashing algorithm
72 """
73 if not digest_alg:
74 raise RuntimeError("simple_hash with digest_alg=None")
75 elif not isinstance(digest_alg, str):
76 h = digest_alg(text + key + salt)
77 elif digest_alg.startswith('pbkdf2'):
78 iterations, keylen, alg = digest_alg[7:-1].split(',')
79 return pbkdf2_hex(text, salt, int(iterations),
80 int(keylen), get_digest(alg))
81 elif key:
82 digest_alg = get_digest(digest_alg)
83 h = hmac.new(key + salt, text, digest_alg)
84 else:
85 h = hashlib.new(digest_alg)
86 h.update(text + salt)
87 return h.hexdigest()
88
89
91 """
92 Returns a hashlib digest algorithm from a string
93 """
94 if not isinstance(value, str):
95 return value
96 value = value.lower()
97 if value == "md5":
98 return hashlib.md5
99 elif value == "sha1":
100 return hashlib.sha1
101 elif value == "sha224":
102 return hashlib.sha224
103 elif value == "sha256":
104 return hashlib.sha256
105 elif value == "sha384":
106 return hashlib.sha384
107 elif value == "sha512":
108 return hashlib.sha512
109 else:
110 raise ValueError("Invalid digest algorithm: %s" % value)
111
112 DIGEST_ALG_BY_SIZE = {
113 128 / 4: 'md5',
114 160 / 4: 'sha1',
115 224 / 4: 'sha224',
116 256 / 4: 'sha256',
117 384 / 4: 'sha384',
118 512 / 4: 'sha512',
119 }
120
121
122 -def pad(s, n=32, padchar=' '):
123 return s + (32 - len(s) % 32) * padchar
124
125
126 -def secure_dumps(data, encryption_key, hash_key=None, compression_level=None):
127 if not hash_key:
128 hash_key = hashlib.sha1(encryption_key).hexdigest()
129 dump = pickle.dumps(data)
130 if compression_level:
131 dump = zlib.compress(dump, compression_level)
132 key = pad(encryption_key[:32])
133 cipher = AES.new(key, IV=key[:16])
134 encrypted_data = base64.urlsafe_b64encode(cipher.encrypt(pad(dump)))
135 signature = hmac.new(hash_key, encrypted_data).hexdigest()
136 return signature + ':' + encrypted_data
137
138
139 -def secure_loads(data, encryption_key, hash_key=None, compression_level=None):
140 if not ':' in data:
141 return None
142 if not hash_key:
143 hash_key = hashlib.sha1(encryption_key).hexdigest()
144 signature, encrypted_data = data.split(':', 1)
145 actual_signature = hmac.new(hash_key, encrypted_data).hexdigest()
146 if signature != actual_signature:
147 return None
148 key = pad(encryption_key[:32])
149 cipher = AES.new(key, IV=key[:16])
150 try:
151 data = cipher.decrypt(base64.urlsafe_b64decode(encrypted_data))
152 data = data.rstrip(' ')
153 if compression_level:
154 data = zlib.decompress(data)
155 return pickle.loads(data)
156 except (TypeError, pickle.UnpicklingError):
157 return None
158
159
160
161
163 """
164 This function and the web2py_uuid follow from the following discussion:
165 http://groups.google.com/group/web2py-developers/browse_thread/thread/7fd5789a7da3f09
166
167 At startup web2py compute a unique ID that identifies the machine by adding
168 uuid.getnode() + int(time.time() * 1e3)
169
170 This is a 48-bit number. It converts the number into 16 8-bit tokens.
171 It uses this value to initialize the entropy source ('/dev/urandom') and to seed random.
172
173 If os.random() is not supported, it falls back to using random and issues a warning.
174 """
175 node_id = uuid.getnode()
176 microseconds = int(time.time() * 1e6)
177 ctokens = [((node_id + microseconds) >> ((i % 6) * 8)) %
178 256 for i in range(16)]
179 random.seed(node_id + microseconds)
180 try:
181 os.urandom(1)
182 have_urandom = True
183 try:
184
185 frandom = open('/dev/urandom', 'wb')
186 try:
187 if python_version == 2:
188 frandom.write(''.join(chr(t) for t in ctokens))
189 else:
190 frandom.write(bytes([]).join(bytes([t]) for t in ctokens))
191 finally:
192 frandom.close()
193 except IOError:
194
195 pass
196 except NotImplementedError:
197 have_urandom = False
198 logger.warning(
199 """Cryptographically secure session management is not possible on your system because
200 your system does not provide a cryptographically secure entropy source.
201 This is not specific to web2py; consider deploying on a different operating system.""")
202 if python_version == 2:
203 packed = ''.join(chr(x) for x in ctokens)
204 else:
205 packed = bytes([]).join(bytes([x]) for x in ctokens)
206 unpacked_ctokens = struct.unpack('=QQ', packed)
207 return unpacked_ctokens, have_urandom
208 UNPACKED_CTOKENS, HAVE_URANDOM = initialize_urandom()
209
210
212 """
213 this is 4x faster than calling os.urandom(16) and prevents
214 the "too many files open" issue with concurrent access to os.urandom()
215 """
216 try:
217 return urandom.pop()
218 except IndexError:
219 try:
220 locker.acquire()
221 ur = os.urandom(16 * 1024)
222 urandom += [ur[i:i + 16] for i in xrange(16, 1024 * 16, 16)]
223 return ur[0:16]
224 finally:
225 locker.release()
226
227
229 """
230 This function follows from the following discussion:
231 http://groups.google.com/group/web2py-developers/browse_thread/thread/7fd5789a7da3f09
232
233 It works like uuid.uuid4 except that tries to use os.urandom() if possible
234 and it XORs the output with the tokens uniquely associated with this machine.
235 """
236 rand_longs = (random.getrandbits(64), random.getrandbits(64))
237 if HAVE_URANDOM:
238 urand_longs = struct.unpack('=QQ', fast_urandom16())
239 byte_s = struct.pack('=QQ',
240 rand_longs[0] ^ urand_longs[0] ^ ctokens[0],
241 rand_longs[1] ^ urand_longs[1] ^ ctokens[1])
242 else:
243 byte_s = struct.pack('=QQ',
244 rand_longs[0] ^ ctokens[0],
245 rand_longs[1] ^ ctokens[1])
246 return str(uuid.UUID(bytes=byte_s, version=4))
247
248 REGEX_IPv4 = re.compile('(\d+)\.(\d+)\.(\d+)\.(\d+)')
249
250
252 """
253 >>> is_valid_ip_address('127.0')
254 False
255 >>> is_valid_ip_address('127.0.0.1')
256 True
257 >>> is_valid_ip_address('2001:660::1')
258 True
259 """
260
261 if address.lower() in ('127.0.0.1', 'localhost', '::1', '::ffff:127.0.0.1'):
262 return True
263 elif address.lower() in ('unknown', ''):
264 return False
265 elif address.count('.') == 3:
266 if address.startswith('::ffff:'):
267 address = address[7:]
268 if hasattr(socket, 'inet_aton'):
269 try:
270 socket.inet_aton(address)
271 return True
272 except socket.error:
273 return False
274 else:
275 match = REGEX_IPv4.match(address)
276 if match and all(0 <= int(match.group(i)) < 256 for i in (1, 2, 3, 4)):
277 return True
278 return False
279 elif hasattr(socket, 'inet_pton'):
280 try:
281 socket.inet_pton(socket.AF_INET6, address)
282 return True
283 except socket.error:
284 return False
285 else:
286 return True
287
288
290 """Determines whether the IP address appears to be a loopback address.
291
292 This assumes that the IP is valid. The IPv6 check is limited to '::1'.
293
294 """
295 if not ip:
296 return False
297 if ip.count('.') == 3:
298 return ip.startswith('127') or ip.startswith('::ffff:127')
299 return ip == '::1'
300