<s> import re <EOL> import os <EOL> import sys <EOL> import time <EOL> import hmac <EOL> import base64 <EOL> import hashlib <EOL> import threading <EOL> import logging <EOL> import requests <EOL> from yubico_client . otp import OTP <EOL> from yubico_client . yubico_exceptions import ( StatusCodeError , <EOL> InvalidClientIdError , <EOL> InvalidValidationResponse , <EOL> SignatureVerificationError ) <EOL> from yubico_client . py3 import b <EOL> from yubico_client . py3 import urlencode <EOL> from yubico_client . py3 import unquote <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> COMMON_CA_LOCATIONS = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> DEFAULT_API_URLS = ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> DEFAULT_TIMEOUT = <NUM_LIT:10> <EOL> DEFAULT_MAX_TIME_WINDOW = <NUM_LIT:5> <EOL> BAD_STATUS_CODES = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> class Yubico ( object ) : <EOL> def __init__ ( self , client_id , key = None , verify_cert = True , <EOL> translate_otp = True , api_urls = DEFAULT_API_URLS , <EOL> ca_certs_bundle_path = None ) : <EOL> if ca_certs_bundle_path and not self . _is_valid_ca_bundle_file ( ca_certs_bundle_path ) : <EOL> raise ValueError ( ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> self . client_id = client_id <EOL> if key is not None : <EOL> key = base64 . b64decode ( key . encode ( '<STR_LIT:ascii>' ) ) <EOL> self . key = key <EOL> self . verify_cert = verify_cert <EOL> self . translate_otp = translate_otp <EOL> self . api_urls = self . _init_request_urls ( api_urls = api_urls ) <EOL> self . ca_certs_bundle_path = ca_certs_bundle_path <EOL> def verify ( self , otp , timestamp = False , sl = None , timeout = None , <EOL> return_response = False ) : <EOL> """<STR_LIT>""" <EOL> ca_bundle_path = self . _get_ca_bundle_path ( ) <EOL> otp = OTP ( otp , self . translate_otp ) <EOL> rand_str = b ( os . urandom ( <NUM_LIT:30> ) ) <EOL> nonce = base64 . b64encode ( rand_str , b ( '<STR_LIT>' ) ) [ : <NUM_LIT> ] . decode ( '<STR_LIT:utf-8>' ) <EOL> query_string = self . generate_query_string ( otp . otp , nonce , timestamp , <EOL> sl , timeout ) <EOL> threads = [ ] <EOL> timeout = timeout or DEFAULT_TIMEOUT <EOL> for url in self . api_urls : <EOL> thread = URLThread ( '<STR_LIT>' % ( url , query_string ) , timeout , <EOL> self . verify_cert , ca_bundle_path ) <EOL> thread . start ( ) <EOL> threads . append ( thread ) <EOL> start_time = time . time ( ) <EOL> while threads and ( start_time + timeout ) > time . time ( ) : <EOL> for thread in threads : <EOL> if not thread . is_alive ( ) : <EOL> if thread . exception : <EOL> raise thread . exception <EOL> elif thread . response : <EOL> status = self . verify_response ( thread . response , <EOL> otp . otp , nonce , <EOL> return_response ) <EOL> if status : <EOL> if return_response : <EOL> return status <EOL> else : <EOL> return True <EOL> threads . remove ( thread ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> raise Exception ( '<STR_LIT>' ) <EOL> def verify_multi ( self , otp_list , max_time_window = DEFAULT_MAX_TIME_WINDOW , <EOL> sl = None , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> otps = [ ] <EOL> for otp in otp_list : <EOL> otps . append ( OTP ( otp , self . translate_otp ) ) <EOL> if len ( otp_list ) < <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> device_ids = set ( ) <EOL> for otp in otps : <EOL> device_ids . add ( otp . device_id ) <EOL> if len ( device_ids ) != <NUM_LIT:1> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> for otp in otps : <EOL> response = self . verify ( otp . otp , True , sl , timeout , <EOL> return_response = True ) <EOL> if not response : <EOL> return False <EOL> otp . timestamp = int ( response [ '<STR_LIT>' ] ) <EOL> count = len ( otps ) <EOL> delta = otps [ count - <NUM_LIT:1> ] . timestamp - otps [ <NUM_LIT:0> ] . timestamp <EOL> delta = delta / <NUM_LIT:8> <EOL> if delta < <NUM_LIT:0> : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if delta > max_time_window : <EOL> raise Exception ( ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % <EOL> ( max_time_window ) ) <EOL> return True <EOL> def verify_response ( self , response , otp , nonce , return_response = False ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> status = re . search ( r'<STR_LIT>' , response ) . groups ( ) <EOL> if len ( status ) > <NUM_LIT:1> : <EOL> message = '<STR_LIT>' <EOL> raise InvalidValidationResponse ( message , response ) <EOL> status = status [ <NUM_LIT:0> ] <EOL> except ( AttributeError , IndexError ) : <EOL> return False <EOL> signature , parameters = self . parse_parameters_from_response ( response ) <EOL> if self . key : <EOL> generated_signature = self . generate_message_signature ( parameters ) <EOL> if signature != generated_signature : <EOL> logger . warn ( "<STR_LIT>" , parameters ) <EOL> raise SignatureVerificationError ( generated_signature , <EOL> signature ) <EOL> param_dict = self . get_parameters_as_dictionary ( parameters ) <EOL> if '<STR_LIT>' in param_dict and param_dict [ '<STR_LIT>' ] != otp : <EOL> message = '<STR_LIT>' <EOL> raise InvalidValidationResponse ( message , response , param_dict ) <EOL> if '<STR_LIT>' in param_dict and param_dict [ '<STR_LIT>' ] != nonce : <EOL> message = '<STR_LIT>' <EOL> raise InvalidValidationResponse ( message , response , param_dict ) <EOL> if status == '<STR_LIT:OK>' : <EOL> if return_response : <EOL> return param_dict <EOL> else : <EOL> return True <EOL> elif status == '<STR_LIT>' : <EOL> raise InvalidClientIdError ( self . client_id ) <EOL> elif status == '<STR_LIT>' : <EOL> raise StatusCodeError ( status ) <EOL> return False <EOL> def generate_query_string ( self , otp , nonce , timestamp = False , sl = None , <EOL> timeout = None ) : <EOL> """<STR_LIT>""" <EOL> data = [ ( '<STR_LIT:id>' , self . client_id ) , <EOL> ( '<STR_LIT>' , otp ) , <EOL> ( '<STR_LIT>' , nonce ) ] <EOL> if timestamp : <EOL> data . append ( ( '<STR_LIT>' , '<STR_LIT:1>' ) ) <EOL> if sl is not None : <EOL> if sl not in range ( <NUM_LIT:0> , <NUM_LIT> ) and sl not in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> data . append ( ( '<STR_LIT>' , sl ) ) <EOL> if timeout : <EOL> data . append ( ( '<STR_LIT>' , timeout ) ) <EOL> query_string = urlencode ( data ) <EOL> if self . key : <EOL> hmac_signature = self . generate_message_signature ( query_string ) <EOL> hmac_signature = hmac_signature <EOL> query_string += '<STR_LIT>' % ( hmac_signature . replace ( '<STR_LIT:+>' , '<STR_LIT>' ) ) <EOL> return query_string <EOL> def generate_message_signature ( self , query_string ) : <EOL> """<STR_LIT>""" <EOL> pairs = query_string . split ( '<STR_LIT:&>' ) <EOL> pairs = [ pair . split ( '<STR_LIT:=>' , <NUM_LIT:1> ) for pair in pairs ] <EOL> pairs_sorted = sorted ( pairs ) <EOL> pairs_string = '<STR_LIT:&>' . join ( [ '<STR_LIT:=>' . join ( pair ) for pair in pairs_sorted ] ) <EOL> digest = hmac . new ( self . key , b ( pairs_string ) , hashlib . sha1 ) . digest ( ) <EOL> signature = base64 . b64encode ( digest ) . decode ( '<STR_LIT:utf-8>' ) <EOL> return signature <EOL> def parse_parameters_from_response ( self , response ) : <EOL> """<STR_LIT>""" <EOL> lines = response . splitlines ( ) <EOL> pairs = [ line . strip ( ) . split ( '<STR_LIT:=>' , <NUM_LIT:1> ) for line in lines if '<STR_LIT:=>' in line ] <EOL> pairs = sorted ( pairs ) <EOL> signature = ( [ unquote ( v ) for k , v in pairs if k == '<STR_LIT:h>' ] or [ None ] ) [ <NUM_LIT:0> ] <EOL> query_string = '<STR_LIT:&>' . join ( [ k + '<STR_LIT:=>' + v for k , v in pairs if k != '<STR_LIT:h>' ] ) <EOL> return ( signature , query_string ) <EOL> def get_parameters_as_dictionary ( self , query_string ) : <EOL> """<STR_LIT>""" <EOL> pairs = ( x . split ( '<STR_LIT:=>' , <NUM_LIT:1> ) for x in query_string . split ( '<STR_LIT:&>' ) ) <EOL> return dict ( ( k , unquote ( v ) ) for k , v in pairs ) <EOL> def _init_request_urls ( self , api_urls ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( api_urls , ( str , list , tuple ) ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> if isinstance ( api_urls , str ) : <EOL> api_urls = ( api_urls , ) <EOL> api_urls = list ( api_urls ) <EOL> for url in api_urls : <EOL> if not url . startswith ( '<STR_LIT>' ) and not url . startswith ( '<STR_LIT>' ) : <EOL> raise ValueError ( ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( url ) ) ) <EOL> return list ( api_urls ) <EOL> def _get_ca_bundle_path ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . ca_certs_bundle_path : <EOL> return self . ca_certs_bundle_path <EOL> for file_path in COMMON_CA_LOCATIONS : <EOL> if self . _is_valid_ca_bundle_file ( file_path = file_path ) : <EOL> return file_path <EOL> return None <EOL> def _is_valid_ca_bundle_file ( self , file_path ) : <EOL> return os . path . exists ( file_path ) and os . path . isfile ( file_path ) <EOL> class URLThread ( threading . Thread ) : <EOL> def __init__ ( self , url , timeout , verify_cert , ca_bundle_path = None ) : <EOL> super ( URLThread , self ) . __init__ ( ) <EOL> self . url = url <EOL> self . timeout = timeout <EOL> self . verify_cert = verify_cert <EOL> self . ca_bundle_path = ca_bundle_path <EOL> self . exception = None <EOL> self . request = None <EOL> self . response = None <EOL> def run ( self ) : <EOL> logger . debug ( '<STR_LIT>' % ( self . url , <EOL> self . name ) ) <EOL> verify = self . verify_cert <EOL> if self . ca_bundle_path is not None : <EOL> verify = self . ca_bundle_path <EOL> logger . debug ( '<STR_LIT>' % ( self . ca_bundle_path ) ) <EOL> try : <EOL> self . request = requests . get ( url = self . url , timeout = self . timeout , <EOL> verify = verify ) <EOL> self . response = self . request . content . decode ( '<STR_LIT:utf-8>' ) <EOL> except requests . exceptions . SSLError : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> self . exception = e <EOL> self . response = None <EOL> except Exception : <EOL> e = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> logger . error ( '<STR_LIT>' + str ( e ) ) <EOL> self . response = None <EOL> args = ( self . url , self . name , self . response ) <EOL> logger . debug ( '<STR_LIT>' % args ) </s>
