<s> import requests <EOL> import logging <EOL> import uuid <EOL> import hashlib <EOL> import hmac <EOL> import base64 <EOL> import re <EOL> import sys <EOL> from requests . auth import AuthBase <EOL> from time import gmtime , strftime <EOL> if sys . version_info [ <NUM_LIT:0> ] >= <NUM_LIT:3> : <EOL> from urllib . parse import urlparse , parse_qsl , urlunparse <EOL> else : <EOL> from urlparse import urlparse , parse_qsl , urlunparse <EOL> import urllib3 . contrib . pyopenssl <EOL> urllib3 . contrib . pyopenssl . inject_into_urllib3 ( ) <EOL> logger = logging . getLogger ( __name__ ) <EOL> __all__ = [ '<STR_LIT>' ] <EOL> def eg_timestamp ( ) : <EOL> return strftime ( '<STR_LIT>' , gmtime ( ) ) <EOL> def new_nonce ( ) : <EOL> return uuid . uuid4 ( ) <EOL> def base64_hmac_sha256 ( data , key ) : <EOL> return base64 . b64encode ( <EOL> hmac . new ( key . encode ( '<STR_LIT:utf8>' ) , data . encode ( '<STR_LIT:utf8>' ) , hashlib . sha256 ) . digest ( ) <EOL> ) . decode ( '<STR_LIT:utf8>' ) <EOL> def base64_sha256 ( data ) : <EOL> return base64 . b64encode ( hashlib . sha256 ( data . encode ( '<STR_LIT:utf8>' ) ) . digest ( ) ) . decode ( '<STR_LIT:utf8>' ) <EOL> class EdgeGridAuth ( AuthBase ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client_token , client_secret , access_token , <EOL> headers_to_sign = None , max_body = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> self . client_token = client_token <EOL> self . client_secret = client_secret <EOL> self . access_token = access_token <EOL> if headers_to_sign : <EOL> self . headers_to_sign = [ h . lower ( ) for h in headers_to_sign ] <EOL> else : <EOL> self . headers_to_sign = [ ] <EOL> self . max_body = max_body <EOL> self . redirect_location = None <EOL> @ staticmethod <EOL> def from_edgerc ( rcinput , section = '<STR_LIT:default>' ) : <EOL> """<STR_LIT>""" <EOL> from . edgerc import EdgeRc <EOL> if isinstance ( rcinput , EdgeRc ) : <EOL> rc = rcinput <EOL> else : <EOL> rc = EdgeRc ( rcinput ) <EOL> return EdgeGridAuth ( <EOL> client_token = rc . get ( section , '<STR_LIT>' ) , <EOL> client_secret = rc . get ( section , '<STR_LIT>' ) , <EOL> access_token = rc . get ( section , '<STR_LIT>' ) , <EOL> headers_to_sign = rc . getlist ( section , '<STR_LIT>' ) , <EOL> max_body = rc . getint ( section , '<STR_LIT>' ) <EOL> ) <EOL> def make_signing_key ( self , timestamp ) : <EOL> signing_key = base64_hmac_sha256 ( timestamp , self . client_secret ) <EOL> logger . debug ( '<STR_LIT>' , signing_key ) <EOL> return signing_key <EOL> def canonicalize_headers ( self , r ) : <EOL> spaces_re = re . compile ( '<STR_LIT>' ) <EOL> return '<STR_LIT:\t>' . join ( [ <EOL> "<STR_LIT>" % ( h , spaces_re . sub ( '<STR_LIT:U+0020>' , r . headers [ h ] . strip ( ) ) ) <EOL> for h in self . headers_to_sign if h in r . headers <EOL> ] ) <EOL> def make_content_hash ( self , r ) : <EOL> content_hash = "<STR_LIT>" <EOL> prepared_body = ( r . body or '<STR_LIT>' ) <EOL> logger . debug ( "<STR_LIT>" , prepared_body ) <EOL> if r . method == '<STR_LIT:POST>' and len ( prepared_body ) > <NUM_LIT:0> : <EOL> logger . debug ( "<STR_LIT>" , prepared_body ) <EOL> if len ( prepared_body ) > self . max_body : <EOL> logger . debug ( <EOL> "<STR_LIT>" , <EOL> len ( prepared_body ) , self . max_body <EOL> ) <EOL> prepared_body = prepared_body [ <NUM_LIT:0> : self . max_body ] <EOL> logger . debug ( "<STR_LIT>" , len ( prepared_body ) ) <EOL> content_hash = base64_sha256 ( prepared_body ) <EOL> logger . debug ( "<STR_LIT>" , content_hash ) <EOL> return content_hash <EOL> def make_data_to_sign ( self , r , auth_header ) : <EOL> parsed_url = urlparse ( r . url ) <EOL> if ( r . headers . get ( '<STR_LIT>' , False ) ) : <EOL> netloc = r . headers [ '<STR_LIT>' ] <EOL> else : <EOL> netloc = parsed_url . netloc <EOL> data_to_sign = '<STR_LIT:\t>' . join ( [ <EOL> r . method , <EOL> parsed_url . scheme , <EOL> netloc , <EOL> parsed_url . path + ( '<STR_LIT:?>' + parsed_url . query if parsed_url . query else "<STR_LIT>" ) , <EOL> self . canonicalize_headers ( r ) , <EOL> self . make_content_hash ( r ) , <EOL> auth_header <EOL> ] ) <EOL> logger . debug ( '<STR_LIT>' , '<STR_LIT>' . join ( data_to_sign . split ( '<STR_LIT:\t>' ) ) ) <EOL> return data_to_sign <EOL> def sign_request ( self , r , timestamp , auth_header ) : <EOL> return base64_hmac_sha256 ( <EOL> self . make_data_to_sign ( r , auth_header ) , <EOL> self . make_signing_key ( timestamp ) <EOL> ) <EOL> def make_auth_header ( self , r , timestamp , nonce ) : <EOL> kvps = [ <EOL> ( '<STR_LIT>' , self . client_token ) , <EOL> ( '<STR_LIT>' , self . access_token ) , <EOL> ( '<STR_LIT>' , timestamp ) , <EOL> ( '<STR_LIT>' , nonce ) , <EOL> ] <EOL> auth_header = "<STR_LIT>" + '<STR_LIT:;>' . join ( [ "<STR_LIT>" % kvp for kvp in kvps ] ) + '<STR_LIT:;>' <EOL> logger . debug ( '<STR_LIT>' , auth_header ) <EOL> signed_auth_header = auth_header + '<STR_LIT>' + self . sign_request ( r , timestamp , auth_header ) <EOL> logger . debug ( '<STR_LIT>' , signed_auth_header ) <EOL> return signed_auth_header <EOL> def handle_redirect ( self , res , ** kwargs ) : <EOL> if res . is_redirect : <EOL> redirect_location = res . headers [ '<STR_LIT:location>' ] <EOL> logger . debug ( "<STR_LIT>" , redirect_location ) <EOL> request_to_sign = res . request . copy ( ) <EOL> request_to_sign . url = redirect_location <EOL> res . request . headers [ '<STR_LIT>' ] = self . make_auth_header ( <EOL> request_to_sign , eg_timestamp ( ) , new_nonce ( ) <EOL> ) <EOL> def __call__ ( self , r ) : <EOL> timestamp = eg_timestamp ( ) <EOL> nonce = new_nonce ( ) <EOL> r . headers [ '<STR_LIT>' ] = self . make_auth_header ( r , timestamp , nonce ) <EOL> r . register_hook ( '<STR_LIT>' , self . handle_redirect ) <EOL> return r </s>
