__init__.py (10916B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\eve\common\lib\jwt\__init__.py 2 from __future__ import unicode_literals 3 import base64 4 import binascii 5 import hashlib 6 import hmac 7 import six 8 import struct 9 import sys 10 from datetime import datetime 11 from calendar import timegm 12 from collections import Mapping 13 try: 14 import json 15 except ImportError: 16 import simplejson as json 17 18 if sys.version_info >= (3, 0, 0): 19 unicode = str 20 basestring = str 21 __version__ = u'0.3.0' 22 __all__ = [u'encode', u'decode', u'DecodeError'] 23 24 class DecodeError(Exception): 25 pass 26 27 28 class ExpiredSignature(Exception): 29 pass 30 31 32 class InvalidAudience(Exception): 33 pass 34 35 36 class InvalidIssuer(Exception): 37 pass 38 39 40 signing_methods = {u'none': lambda msg, key: '', 41 u'HS256': lambda msg, key: hmac.new(key, msg, hashlib.sha256).digest(), 42 u'HS384': lambda msg, key: hmac.new(key, msg, hashlib.sha384).digest(), 43 u'HS512': lambda msg, key: hmac.new(key, msg, hashlib.sha512).digest()} 44 verify_methods = {u'HS256': lambda msg, key: hmac.new(key, msg, hashlib.sha256).digest(), 45 u'HS384': lambda msg, key: hmac.new(key, msg, hashlib.sha384).digest(), 46 u'HS512': lambda msg, key: hmac.new(key, msg, hashlib.sha512).digest()} 47 48 def prepare_HS_key(key): 49 if not isinstance(key, basestring) and not isinstance(key, bytes): 50 raise TypeError(u'Expecting a string- or bytes-formatted key.') 51 if isinstance(key, unicode): 52 key = key.encode(u'utf-8') 53 return key 54 55 56 prepare_key_methods = {u'none': lambda key: None, 57 u'HS256': prepare_HS_key, 58 u'HS384': prepare_HS_key, 59 u'HS512': prepare_HS_key} 60 try: 61 from Crypto.Signature import PKCS1_v1_5 62 from Crypto.Hash import SHA256 63 from Crypto.Hash import SHA384 64 from Crypto.Hash import SHA512 65 from Crypto.PublicKey import RSA 66 signing_methods.update({u'RS256': lambda msg, key: PKCS1_v1_5.new(key).sign(SHA256.new(msg)), 67 u'RS384': lambda msg, key: PKCS1_v1_5.new(key).sign(SHA384.new(msg)), 68 u'RS512': lambda msg, key: PKCS1_v1_5.new(key).sign(SHA512.new(msg))}) 69 verify_methods.update({u'RS256': lambda msg, key, sig: PKCS1_v1_5.new(key).verify(SHA256.new(msg), sig), 70 u'RS384': lambda msg, key, sig: PKCS1_v1_5.new(key).verify(SHA384.new(msg), sig), 71 u'RS512': lambda msg, key, sig: PKCS1_v1_5.new(key).verify(SHA512.new(msg), sig)}) 72 73 def prepare_RS_key(key): 74 if isinstance(key, RSA._RSAobj): 75 return key 76 if isinstance(key, basestring): 77 if isinstance(key, unicode): 78 key = key.encode(u'utf-8') 79 key = RSA.importKey(key) 80 else: 81 raise TypeError(u'Expecting a PEM- or RSA-formatted key.') 82 return key 83 84 85 prepare_key_methods.update({u'RS256': prepare_RS_key, 86 u'RS384': prepare_RS_key, 87 u'RS512': prepare_RS_key}) 88 except ImportError: 89 pass 90 91 def base64_urlsafe_decode_int(data): 92 if isinstance(data, six.text_type): 93 data = data.encode(u'ascii') 94 _d = base64.urlsafe_b64decode(bytes(data) + '==') 95 arr = struct.unpack(str(u'%sB' % len(_d)), _d) 96 return int(u''.join([ u'%02x' % byte for byte in arr ]), 16) 97 98 99 try: 100 import rsa 101 from rsa.pkcs1 import VerificationError 102 103 def prepare_key_rs256(key): 104 return key 105 106 107 def verify_rs256(signing_input, key, signature): 108 key = rsa.PublicKey.load_pkcs1(key, format=u'PEM') 109 try: 110 rsa.verify(signing_input, signature, key) 111 except VerificationError: 112 return False 113 114 return True 115 116 117 def sign_rs256(msg, key): 118 return rsa.sign(msg, key, u'SHA-256') 119 120 121 if u'RS256' not in verify_methods: 122 prepare_key_methods.update({u'RS256': prepare_key_rs256}) 123 verify_methods.update({u'RS256': verify_rs256}) 124 if u'RS256' not in signing_methods: 125 signing_methods.update({u'RS256': sign_rs256}) 126 except ImportError: 127 pass 128 129 try: 130 import ecdsa 131 from Crypto.Hash import SHA256 132 from Crypto.Hash import SHA384 133 from Crypto.Hash import SHA512 134 signing_methods.update({u'ES256': lambda msg, key: key.sign(msg, hashfunc=hashlib.sha256, sigencode=ecdsa.util.sigencode_der), 135 u'ES384': lambda msg, key: key.sign(msg, hashfunc=hashlib.sha384, sigencode=ecdsa.util.sigencode_der), 136 u'ES512': lambda msg, key: key.sign(msg, hashfunc=hashlib.sha512, sigencode=ecdsa.util.sigencode_der)}) 137 verify_methods.update({u'ES256': lambda msg, key, sig: key.verify(sig, msg, hashfunc=hashlib.sha256, sigdecode=ecdsa.util.sigdecode_der), 138 u'ES384': lambda msg, key, sig: key.verify(sig, msg, hashfunc=hashlib.sha384, sigdecode=ecdsa.util.sigdecode_der), 139 u'ES512': lambda msg, key, sig: key.verify(sig, msg, hashfunc=hashlib.sha512, sigdecode=ecdsa.util.sigdecode_der)}) 140 141 def prepare_ES_key(key): 142 if isinstance(key, ecdsa.SigningKey) or isinstance(key, ecdsa.VerifyingKey): 143 return key 144 if isinstance(key, basestring): 145 if isinstance(key, unicode): 146 key = key.encode(u'utf-8') 147 try: 148 key = ecdsa.VerifyingKey.from_pem(key) 149 except ecdsa.der.UnexpectedDER: 150 try: 151 key = ecdsa.SigningKey.from_pem(key) 152 except: 153 raise 154 155 else: 156 raise TypeError(u'Expecting a PEM-formatted key.') 157 return key 158 159 160 prepare_key_methods.update({u'ES256': prepare_ES_key, 161 u'ES384': prepare_ES_key, 162 u'ES512': prepare_ES_key}) 163 except ImportError: 164 pass 165 166 def constant_time_compare(val1, val2): 167 if len(val1) != len(val2): 168 return False 169 result = 0 170 if sys.version_info >= (3, 0, 0): 171 for x, y in zip(val1, val2): 172 result |= x ^ y 173 174 else: 175 for x, y in zip(val1, val2): 176 result |= ord(x) ^ ord(y) 177 178 return result == 0 179 180 181 def base64url_decode(input): 182 rem = len(input) % 4 183 if rem > 0: 184 input += '=' * (4 - rem) 185 return base64.urlsafe_b64decode(input) 186 187 188 def base64url_encode(input): 189 return base64.urlsafe_b64encode(input).replace('=', '') 190 191 192 def header(jwt): 193 header_segment = jwt.split('.', 1)[0] 194 try: 195 header_data = base64url_decode(header_segment).decode(u'utf-8') 196 return json.loads(header_data) 197 except (ValueError, TypeError): 198 raise DecodeError(u'Invalid header encoding') 199 200 201 def encode(payload, key, algorithm = u'HS256', headers = None): 202 segments = [] 203 if algorithm is None: 204 algorithm = u'none' 205 if not isinstance(payload, Mapping): 206 raise TypeError(u'Expecting a mapping object, as json web token onlysupport json objects.') 207 header = {u'typ': u'JWT', 208 u'alg': algorithm} 209 if headers: 210 header.update(headers) 211 json_header = json.dumps(header, separators=(u',', u':')).encode(u'utf-8') 212 segments.append(base64url_encode(json_header)) 213 for time_claim in [u'exp', u'iat', u'nbf']: 214 if isinstance(payload.get(time_claim), datetime): 215 payload[time_claim] = timegm(payload[time_claim].utctimetuple()) 216 217 json_payload = json.dumps(payload, separators=(u',', u':')).encode(u'utf-8') 218 segments.append(base64url_encode(json_payload)) 219 signing_input = '.'.join(segments) 220 try: 221 key = prepare_key_methods[algorithm](key) 222 signature = signing_methods[algorithm](signing_input, key) 223 except KeyError: 224 raise NotImplementedError(u'Algorithm not supported') 225 226 segments.append(base64url_encode(signature)) 227 return '.'.join(segments) 228 229 230 def decode(jwt, key = u'', verify = True, **kwargs): 231 payload, signing_input, header, signature = load(jwt) 232 if verify: 233 verify_expiration = kwargs.pop(u'verify_expiration', True) 234 leeway = kwargs.pop(u'leeway', 0) 235 verify_signature(payload, signing_input, header, signature, key, verify_expiration, leeway, **kwargs) 236 return payload 237 238 239 def load(jwt): 240 if isinstance(jwt, unicode): 241 jwt = jwt.encode(u'utf-8') 242 try: 243 signing_input, crypto_segment = jwt.rsplit('.', 1) 244 header_segment, payload_segment = signing_input.split('.', 1) 245 except ValueError: 246 raise DecodeError(u'Not enough segments') 247 248 try: 249 header_data = base64url_decode(header_segment) 250 except (TypeError, binascii.Error): 251 raise DecodeError(u'Invalid header padding') 252 253 try: 254 header = json.loads(header_data.decode(u'utf-8')) 255 except ValueError as e: 256 raise DecodeError(u'Invalid header string: %s' % e) 257 258 try: 259 payload_data = base64url_decode(payload_segment) 260 except (TypeError, binascii.Error): 261 raise DecodeError(u'Invalid payload padding') 262 263 try: 264 payload = json.loads(payload_data.decode(u'utf-8')) 265 except ValueError as e: 266 raise DecodeError(u'Invalid payload string: %s' % e) 267 268 try: 269 signature = base64url_decode(crypto_segment) 270 except (TypeError, binascii.Error): 271 raise DecodeError(u'Invalid crypto padding') 272 273 return (payload, 274 signing_input, 275 header, 276 signature) 277 278 279 def verify_signature(payload, signing_input, header, signature, key = u'', verify_expiration = True, leeway = 0, **kwargs): 280 try: 281 algorithm = header[u'alg'].upper() 282 key = prepare_key_methods[algorithm](key) 283 if algorithm.startswith(u'HS'): 284 expected = verify_methods[algorithm](signing_input, key) 285 if not constant_time_compare(signature, expected): 286 raise DecodeError(u'Signature verification failed') 287 elif not verify_methods[algorithm](signing_input, key, signature): 288 raise DecodeError(u'Signature verification failed') 289 except KeyError: 290 raise DecodeError(u'Algorithm not supported') 291 292 if u'nbf' in payload and verify_expiration: 293 utc_timestamp = timegm(datetime.utcnow().utctimetuple()) 294 if payload[u'nbf'] > utc_timestamp + leeway: 295 raise ExpiredSignature(u'Signature not yet valid') 296 if u'exp' in payload and verify_expiration: 297 utc_timestamp = timegm(datetime.utcnow().utctimetuple()) 298 if payload[u'exp'] < utc_timestamp - leeway: 299 raise ExpiredSignature(u'Signature has expired') 300 audience = kwargs.get(u'audience') 301 if audience: 302 if isinstance(audience, list): 303 audiences = audience 304 else: 305 audiences = [audience] 306 if payload.get(u'aud') not in audiences: 307 raise InvalidAudience(u'Invalid audience') 308 issuer = kwargs.get(u'issuer') 309 if issuer: 310 if payload.get(u'iss') != issuer: 311 raise InvalidIssuer(u'Invalid issuer') 312 313 314 def is_jwt(token): 315 try: 316 return token.count(u'.') == 2 317 except AttributeError: 318 return False 319 320 321 def fabricate_local_dev_user_token(tenant, user_id): 322 subject = u'USER:EVE:{}'.format(user_id) 323 payload = {u'sub': subject, 324 u'tenant': tenant, 325 u'tier': u'dev', 326 u'region': u'local'} 327 return encode(payload, u'dev')