httpauth.py (7017B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\cherrypy\lib\httpauth.py 2 __version__ = (1, 0, 1) 3 __author__ = 'Tiago Cogumbreiro <cogumbreiro@users.sf.net>' 4 __credits__ = '\n Peter van Kampen for its recipe which implement most of Digest authentication:\n http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/302378\n' 5 __license__ = '\nCopyright (c) 2005, Tiago Cogumbreiro <cogumbreiro@users.sf.net>\nAll rights reserved.\n\nRedistribution and use in source and binary forms, with or without modification, \nare permitted provided that the following conditions are met:\n\n * Redistributions of source code must retain the above copyright notice, \n this list of conditions and the following disclaimer.\n * Redistributions in binary form must reproduce the above copyright notice, \n this list of conditions and the following disclaimer in the documentation \n and/or other materials provided with the distribution.\n * Neither the name of Sylvain Hellegouarch nor the names of his contributors \n may be used to endorse or promote products derived from this software \n without specific prior written permission.\n\nTHIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND \nANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED \nWARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE \nDISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE \nFOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL \nDAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR \nSERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER \nCAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, \nOR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE \nOF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n' 6 __all__ = ('digestAuth', 'basicAuth', 'doAuth', 'checkResponse', 'parseAuthorization', 'SUPPORTED_ALGORITHM', 'md5SessionKey', 'calculateNonce', 'SUPPORTED_QOP') 7 import time 8 from cherrypy._cpcompat import base64_decode, ntob, md5 9 from cherrypy._cpcompat import parse_http_list, parse_keqv_list 10 MD5 = 'MD5' 11 MD5_SESS = 'MD5-sess' 12 AUTH = 'auth' 13 AUTH_INT = 'auth-int' 14 SUPPORTED_ALGORITHM = (MD5, MD5_SESS) 15 SUPPORTED_QOP = (AUTH, AUTH_INT) 16 DIGEST_AUTH_ENCODERS = {MD5: lambda val: md5(ntob(val)).hexdigest(), 17 MD5_SESS: lambda val: md5(ntob(val)).hexdigest()} 18 19 def calculateNonce(realm, algorithm = MD5): 20 global DIGEST_AUTH_ENCODERS 21 try: 22 encoder = DIGEST_AUTH_ENCODERS[algorithm] 23 except KeyError: 24 raise NotImplementedError('The chosen algorithm (%s) does not have an implementation yet' % algorithm) 25 26 return encoder('%d:%s' % (time.time(), realm)) 27 28 29 def digestAuth(realm, algorithm = MD5, nonce = None, qop = AUTH): 30 if nonce is None: 31 nonce = calculateNonce(realm, algorithm) 32 return 'Digest realm="%s", nonce="%s", algorithm="%s", qop="%s"' % (realm, 33 nonce, 34 algorithm, 35 qop) 36 37 38 def basicAuth(realm): 39 return 'Basic realm="%s"' % realm 40 41 42 def doAuth(realm): 43 return digestAuth(realm) + ' ' + basicAuth(realm) 44 45 46 def _parseDigestAuthorization(auth_params): 47 items = parse_http_list(auth_params) 48 params = parse_keqv_list(items) 49 required = ['username', 50 'realm', 51 'nonce', 52 'uri', 53 'response'] 54 for k in required: 55 if k not in params: 56 return None 57 58 if 'qop' in params and not ('cnonce' in params and 'nc' in params): 59 return None 60 if ('cnonce' in params or 'nc' in params) and 'qop' not in params: 61 return None 62 return params 63 64 65 def _parseBasicAuthorization(auth_params): 66 username, password = base64_decode(auth_params).split(':', 1) 67 return {'username': username, 68 'password': password} 69 70 71 AUTH_SCHEMES = {'basic': _parseBasicAuthorization, 72 'digest': _parseDigestAuthorization} 73 74 def parseAuthorization(credentials): 75 global AUTH_SCHEMES 76 auth_scheme, auth_params = credentials.split(' ', 1) 77 auth_scheme = auth_scheme.lower() 78 parser = AUTH_SCHEMES[auth_scheme] 79 params = parser(auth_params) 80 if params is None: 81 return 82 params['auth_scheme'] = auth_scheme 83 return params 84 85 86 def md5SessionKey(params, password): 87 keys = ('username', 'realm', 'nonce', 'cnonce') 88 params_copy = {} 89 for key in keys: 90 params_copy[key] = params[key] 91 92 params_copy['algorithm'] = MD5_SESS 93 return _A1(params_copy, password) 94 95 96 def _A1(params, password): 97 algorithm = params.get('algorithm', MD5) 98 H = DIGEST_AUTH_ENCODERS[algorithm] 99 if algorithm == MD5: 100 return '%s:%s:%s' % (params['username'], params['realm'], password) 101 if algorithm == MD5_SESS: 102 h_a1 = H('%s:%s:%s' % (params['username'], params['realm'], password)) 103 return '%s:%s:%s' % (h_a1, params['nonce'], params['cnonce']) 104 105 106 def _A2(params, method, kwargs): 107 qop = params.get('qop', 'auth') 108 if qop == 'auth': 109 return method + ':' + params['uri'] 110 if qop == 'auth-int': 111 entity_body = kwargs.get('entity_body', '') 112 H = kwargs['H'] 113 return '%s:%s:%s' % (method, params['uri'], H(entity_body)) 114 raise NotImplementedError("The 'qop' method is unknown: %s" % qop) 115 116 117 def _computeDigestResponse(auth_map, password, method = 'GET', A1 = None, **kwargs): 118 params = auth_map 119 algorithm = params.get('algorithm', MD5) 120 H = DIGEST_AUTH_ENCODERS[algorithm] 121 KD = lambda secret, data: H(secret + ':' + data) 122 qop = params.get('qop', None) 123 H_A2 = H(_A2(params, method, kwargs)) 124 if algorithm == MD5_SESS and A1 is not None: 125 H_A1 = H(A1) 126 else: 127 H_A1 = H(_A1(params, password)) 128 if qop in ('auth', 'auth-int'): 129 request = '%s:%s:%s:%s:%s' % (params['nonce'], 130 params['nc'], 131 params['cnonce'], 132 params['qop'], 133 H_A2) 134 elif qop is None: 135 request = '%s:%s' % (params['nonce'], H_A2) 136 return KD(H_A1, request) 137 138 139 def _checkDigestResponse(auth_map, password, method = 'GET', A1 = None, **kwargs): 140 if auth_map['realm'] != kwargs.get('realm', None): 141 return False 142 response = _computeDigestResponse(auth_map, password, method, A1, **kwargs) 143 return response == auth_map['response'] 144 145 146 def _checkBasicResponse(auth_map, password, method = 'GET', encrypt = None, **kwargs): 147 try: 148 return encrypt(auth_map['password'], auth_map['username']) == password 149 except TypeError: 150 return encrypt(auth_map['password']) == password 151 152 153 AUTH_RESPONSES = {'basic': _checkBasicResponse, 154 'digest': _checkDigestResponse} 155 156 def checkResponse(auth_map, password, method = 'GET', encrypt = None, **kwargs): 157 checker = AUTH_RESPONSES[auth_map['auth_scheme']] 158 return checker(auth_map, password, method=method, encrypt=encrypt, **kwargs) 159 160 161 global SUPPORTED_ALGORITHM ## Warning: Unused global 162 global SUPPORTED_QOP ## Warning: Unused global