<s> """<STR_LIT>""" <EOL> import base64 <EOL> try : <EOL> import json <EOL> except ImportError : <EOL> import simplejson as json <EOL> import logging <EOL> import os <EOL> import time <EOL> import urllib <EOL> import google <EOL> from google . appengine . api import memcache <EOL> from google . appengine . api import oauth <EOL> from google . appengine . api import urlfetch <EOL> from google . appengine . api import users <EOL> try : <EOL> from Crypto . Hash import SHA256 <EOL> from Crypto . PublicKey import RSA <EOL> _CRYPTO_LOADED = True <EOL> except ImportError : <EOL> _CRYPTO_LOADED = False <EOL> __all__ = [ '<STR_LIT>' ] <EOL> _CLOCK_SKEW_SECS = <NUM_LIT> <EOL> _MAX_TOKEN_LIFETIME_SECS = <NUM_LIT> <EOL> _DEFAULT_CERT_URI = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> _ENV_USE_OAUTH_SCOPE = '<STR_LIT>' <EOL> _ENV_AUTH_EMAIL = '<STR_LIT>' <EOL> _ENV_AUTH_DOMAIN = '<STR_LIT>' <EOL> _EMAIL_SCOPE = '<STR_LIT>' <EOL> _TOKENINFO_URL = '<STR_LIT>' <EOL> class _AppIdentityError ( Exception ) : <EOL> pass <EOL> def get_current_user ( ) : <EOL> """<STR_LIT>""" <EOL> if not _is_auth_info_available ( ) : <EOL> logging . error ( '<STR_LIT>' ) <EOL> return None <EOL> if _ENV_USE_OAUTH_SCOPE in os . environ : <EOL> return oauth . get_current_user ( os . environ [ _ENV_USE_OAUTH_SCOPE ] ) <EOL> if ( _ENV_AUTH_EMAIL in os . environ and <EOL> _ENV_AUTH_DOMAIN in os . environ ) : <EOL> if not os . environ [ _ENV_AUTH_EMAIL ] : <EOL> return None <EOL> return users . User ( os . environ [ _ENV_AUTH_EMAIL ] , <EOL> os . environ [ _ENV_AUTH_DOMAIN ] or None ) <EOL> return None <EOL> def _is_auth_info_available ( ) : <EOL> """<STR_LIT>""" <EOL> return ( ( _ENV_AUTH_EMAIL in os . environ and <EOL> _ENV_AUTH_DOMAIN in os . environ ) or <EOL> _ENV_USE_OAUTH_SCOPE in os . environ ) <EOL> def _maybe_set_current_user_vars ( method , api_info = None , request = None ) : <EOL> """<STR_LIT>""" <EOL> if _is_auth_info_available ( ) : <EOL> return <EOL> os . environ [ _ENV_AUTH_EMAIL ] = '<STR_LIT>' <EOL> os . environ [ _ENV_AUTH_DOMAIN ] = '<STR_LIT>' <EOL> try : <EOL> api_info = api_info or method . im_self . api_info <EOL> except AttributeError : <EOL> logging . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> method . __name__ ) <EOL> scopes = method . method_info . scopes <EOL> audiences = method . method_info . audiences <EOL> allowed_client_ids = method . method_info . allowed_client_ids <EOL> else : <EOL> scopes = ( method . method_info . scopes <EOL> if method . method_info . scopes is not None <EOL> else api_info . scopes ) <EOL> audiences = ( method . method_info . audiences <EOL> if method . method_info . audiences is not None <EOL> else api_info . audiences ) <EOL> allowed_client_ids = ( method . method_info . allowed_client_ids <EOL> if method . method_info . allowed_client_ids is not None <EOL> else api_info . allowed_client_ids ) <EOL> if not scopes and not audiences and not allowed_client_ids : <EOL> return <EOL> token = _get_token ( request ) <EOL> if not token : <EOL> return None <EOL> if ( ( scopes == [ _EMAIL_SCOPE ] or scopes == ( _EMAIL_SCOPE , ) ) and <EOL> allowed_client_ids ) : <EOL> logging . info ( '<STR_LIT>' ) <EOL> time_now = long ( time . time ( ) ) <EOL> user = _get_id_token_user ( token , audiences , allowed_client_ids , time_now , <EOL> memcache ) <EOL> if user : <EOL> os . environ [ _ENV_AUTH_EMAIL ] = user . email ( ) <EOL> os . environ [ _ENV_AUTH_DOMAIN ] = user . auth_domain ( ) <EOL> return <EOL> if scopes : <EOL> logging . info ( '<STR_LIT>' ) <EOL> result = urlfetch . fetch ( <EOL> '<STR_LIT>' % ( _TOKENINFO_URL , urllib . urlencode ( { '<STR_LIT>' : token } ) ) ) <EOL> if result . status_code == <NUM_LIT:200> : <EOL> token_info = json . loads ( result . content ) <EOL> _set_oauth_user_vars ( token_info , audiences , allowed_client_ids , <EOL> scopes , _is_local_dev ( ) ) <EOL> def _get_token ( request ) : <EOL> """<STR_LIT>""" <EOL> auth_header = os . environ . get ( '<STR_LIT>' ) <EOL> if auth_header : <EOL> allowed_auth_schemes = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> for auth_scheme in allowed_auth_schemes : <EOL> if auth_header . startswith ( auth_scheme ) : <EOL> return auth_header [ len ( auth_scheme ) + <NUM_LIT:1> : ] <EOL> return None <EOL> if request : <EOL> for key in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> token , _ = request . get_unrecognized_field_info ( key ) <EOL> if token : <EOL> return token <EOL> def _get_id_token_user ( token , audiences , allowed_client_ids , time_now , cache ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> parsed_token = _verify_signed_jwt_with_certs ( token , time_now , cache ) <EOL> except _AppIdentityError , e : <EOL> logging . warning ( '<STR_LIT>' , e ) <EOL> return None <EOL> except : <EOL> logging . warning ( '<STR_LIT>' ) <EOL> return None <EOL> if _verify_parsed_token ( parsed_token , audiences , allowed_client_ids ) : <EOL> email = parsed_token [ '<STR_LIT:email>' ] <EOL> return users . User ( email ) <EOL> def _set_oauth_user_vars ( token_info , audiences , allowed_client_ids , scopes , <EOL> local_dev ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT:email>' not in token_info : <EOL> logging . warning ( '<STR_LIT>' ) <EOL> return <EOL> if not token_info . get ( '<STR_LIT>' ) : <EOL> logging . warning ( '<STR_LIT>' ) <EOL> return <EOL> if audiences or allowed_client_ids : <EOL> if '<STR_LIT>' not in token_info : <EOL> logging . warning ( '<STR_LIT>' ) <EOL> return <EOL> if token_info [ '<STR_LIT>' ] in audiences : <EOL> pass <EOL> elif ( token_info [ '<STR_LIT>' ] == token_info . get ( '<STR_LIT>' ) and <EOL> allowed_client_ids is not None and <EOL> token_info [ '<STR_LIT>' ] in allowed_client_ids ) : <EOL> pass <EOL> else : <EOL> logging . warning ( '<STR_LIT>' ) <EOL> return <EOL> token_scopes = token_info . get ( '<STR_LIT>' , '<STR_LIT>' ) . split ( '<STR_LIT:U+0020>' ) <EOL> if not any ( scope in scopes for scope in token_scopes ) : <EOL> logging . warning ( '<STR_LIT>' ) <EOL> return <EOL> if local_dev : <EOL> os . environ [ _ENV_AUTH_EMAIL ] = token_info [ '<STR_LIT:email>' ] <EOL> os . environ [ _ENV_AUTH_DOMAIN ] = '<STR_LIT>' <EOL> return <EOL> for scope in scopes : <EOL> try : <EOL> oauth_user = oauth . get_current_user ( scope ) <EOL> oauth_scope = scope <EOL> break <EOL> except oauth . Error : <EOL> pass <EOL> else : <EOL> logging . warning ( '<STR_LIT>' ) <EOL> return None <EOL> if oauth_user . email ( ) == token_info [ '<STR_LIT:email>' ] : <EOL> os . environ [ _ENV_USE_OAUTH_SCOPE ] = oauth_scope <EOL> return <EOL> logging . warning ( '<STR_LIT>' ) <EOL> return None <EOL> def _is_local_dev ( ) : <EOL> return os . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) . startswith ( '<STR_LIT>' ) <EOL> def _verify_parsed_token ( parsed_token , audiences , allowed_client_ids ) : <EOL> if parsed_token . get ( '<STR_LIT>' ) != '<STR_LIT>' : <EOL> logging . warning ( '<STR_LIT>' , parsed_token . get ( '<STR_LIT>' ) ) <EOL> return False <EOL> aud = parsed_token . get ( '<STR_LIT>' ) <EOL> if not aud : <EOL> logging . warning ( '<STR_LIT>' ) <EOL> return False <EOL> cid = parsed_token . get ( '<STR_LIT>' ) <EOL> if aud != cid and aud not in audiences : <EOL> logging . warning ( '<STR_LIT>' , aud ) <EOL> return False <EOL> if not allowed_client_ids : <EOL> logging . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return False <EOL> elif not cid or cid not in allowed_client_ids : <EOL> logging . warning ( '<STR_LIT>' , cid ) <EOL> return False <EOL> if '<STR_LIT:email>' not in parsed_token : <EOL> return False <EOL> return True <EOL> def _urlsafe_b64decode ( b64string ) : <EOL> b64string = b64string . encode ( '<STR_LIT:ascii>' ) <EOL> padded = b64string + '<STR_LIT:=>' * ( ( <NUM_LIT:4> - len ( b64string ) ) % <NUM_LIT:4> ) <EOL> return base64 . urlsafe_b64decode ( padded ) <EOL> def _get_cached_certs ( cert_uri , cache ) : <EOL> certs = cache . get ( cert_uri , namespace = '<STR_LIT>' ) <EOL> if certs is None : <EOL> logging . info ( '<STR_LIT>' ) <EOL> try : <EOL> result = urlfetch . fetch ( cert_uri ) <EOL> except AssertionError : <EOL> return None <EOL> if result . status_code == <NUM_LIT:200> : <EOL> certs = json . loads ( result . content ) <EOL> cache . set ( cert_uri , certs , time = <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , namespace = '<STR_LIT>' ) <EOL> else : <EOL> logging . error ( <EOL> '<STR_LIT>' , result . status_code ) <EOL> return certs <EOL> def _b64_to_long ( b ) : <EOL> b = b . encode ( '<STR_LIT:ascii>' ) <EOL> b += '<STR_LIT:=>' * ( ( <NUM_LIT:4> - len ( b ) ) % <NUM_LIT:4> ) <EOL> b = base64 . b64decode ( b ) <EOL> return long ( b . encode ( '<STR_LIT>' ) , <NUM_LIT:16> ) <EOL> def _verify_signed_jwt_with_certs ( <EOL> jwt , time_now , cache , <EOL> cert_uri = _DEFAULT_CERT_URI ) : <EOL> """<STR_LIT>""" <EOL> segments = jwt . split ( '<STR_LIT:.>' ) <EOL> if len ( segments ) != <NUM_LIT:3> : <EOL> raise _AppIdentityError ( '<STR_LIT>' % jwt ) <EOL> signed = '<STR_LIT>' % ( segments [ <NUM_LIT:0> ] , segments [ <NUM_LIT:1> ] ) <EOL> signature = _urlsafe_b64decode ( segments [ <NUM_LIT:2> ] ) <EOL> lsignature = long ( signature . encode ( '<STR_LIT>' ) , <NUM_LIT:16> ) <EOL> header_body = _urlsafe_b64decode ( segments [ <NUM_LIT:0> ] ) <EOL> try : <EOL> header = json . loads ( header_body ) <EOL> except : <EOL> raise _AppIdentityError ( '<STR_LIT>' % header_body ) <EOL> if header . get ( '<STR_LIT>' ) != '<STR_LIT>' : <EOL> raise _AppIdentityError ( '<STR_LIT>' % <EOL> header . get ( '<STR_LIT>' ) ) <EOL> json_body = _urlsafe_b64decode ( segments [ <NUM_LIT:1> ] ) <EOL> try : <EOL> parsed = json . loads ( json_body ) <EOL> except : <EOL> raise _AppIdentityError ( '<STR_LIT>' % json_body ) <EOL> certs = _get_cached_certs ( cert_uri , cache ) <EOL> if certs is None : <EOL> raise _AppIdentityError ( <EOL> '<STR_LIT>' % jwt ) <EOL> if not _CRYPTO_LOADED : <EOL> raise _AppIdentityError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> verified = False <EOL> for keyvalue in certs [ '<STR_LIT>' ] : <EOL> modulus = _b64_to_long ( keyvalue [ '<STR_LIT>' ] ) <EOL> exponent = _b64_to_long ( keyvalue [ '<STR_LIT>' ] ) <EOL> key = RSA . construct ( ( modulus , exponent ) ) <EOL> local_hash = SHA256 . new ( signed ) . hexdigest ( ) <EOL> local_hash = local_hash . zfill ( <NUM_LIT:64> ) <EOL> hexsig = '<STR_LIT>' % key . encrypt ( lsignature , '<STR_LIT>' ) [ <NUM_LIT:0> ] <EOL> verified = ( hexsig [ - <NUM_LIT:64> : ] == local_hash ) <EOL> if verified : <EOL> break <EOL> if not verified : <EOL> raise _AppIdentityError ( '<STR_LIT>' % jwt ) <EOL> iat = parsed . get ( '<STR_LIT>' ) <EOL> if iat is None : <EOL> raise _AppIdentityError ( '<STR_LIT>' % json_body ) <EOL> earliest = iat - _CLOCK_SKEW_SECS <EOL> exp = parsed . get ( '<STR_LIT>' ) <EOL> if exp is None : <EOL> raise _AppIdentityError ( '<STR_LIT>' % json_body ) <EOL> if exp >= time_now + _MAX_TOKEN_LIFETIME_SECS : <EOL> raise _AppIdentityError ( '<STR_LIT>' % json_body ) <EOL> latest = exp + _CLOCK_SKEW_SECS <EOL> if time_now < earliest : <EOL> raise _AppIdentityError ( '<STR_LIT>' % <EOL> ( time_now , earliest , json_body ) ) <EOL> if time_now > latest : <EOL> raise _AppIdentityError ( '<STR_LIT>' % <EOL> ( time_now , latest , json_body ) ) <EOL> return parsed </s>
