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