<s> from __future__ import unicode_literals <EOL> import base64 <EOL> import logging <EOL> import mimetypes <EOL> import os <EOL> import random <EOL> import shutil <EOL> import sys <EOL> from io import BytesIO <EOL> from json import loads as json_loads <EOL> import six <EOL> from six . moves . http_client import UNAUTHORIZED , NOT_MODIFIED <EOL> from six . moves . http_cookiejar import Cookie , CookieJar , MozillaCookieJar <EOL> from six . moves . urllib . error import HTTPError , URLError <EOL> from six . moves . urllib . parse import parse_qsl , urlencode , urlparse , urlunparse <EOL> from six . moves . urllib . request import ( <EOL> BaseHandler , <EOL> HTTPBasicAuthHandler , <EOL> HTTPCookieProcessor , <EOL> HTTPDigestAuthHandler , <EOL> HTTPErrorProcessor , <EOL> HTTPPasswordMgr , <EOL> ProxyHandler , <EOL> Request as URLRequest , <EOL> build_opener , <EOL> install_opener , <EOL> urlopen ) <EOL> from rbtools import get_package_version <EOL> from rbtools . api . cache import APICache <EOL> from rbtools . api . errors import APIError , create_api_error , ServerInterfaceError <EOL> from rbtools . utils . filesystem import get_home_path <EOL> try : <EOL> import ssl <EOL> from six . moves . urllib . request import HTTPSHandler <EOL> except ImportError : <EOL> ssl = None <EOL> HTTPSHandler = None <EOL> RBTOOLS_COOKIE_FILE = '<STR_LIT>' <EOL> RB_COOKIE_NAME = '<STR_LIT>' <EOL> class HttpRequest ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url , method = '<STR_LIT:GET>' , query_args = { } ) : <EOL> self . method = method <EOL> self . headers = { } <EOL> self . _fields = { } <EOL> self . _files = { } <EOL> query_args = dict ( [ <EOL> ( key . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) , value ) <EOL> for key , value in six . iteritems ( query_args ) <EOL> ] ) <EOL> url_parts = list ( urlparse ( url ) ) <EOL> query = dict ( parse_qsl ( url_parts [ <NUM_LIT:4> ] ) ) <EOL> query . update ( query_args ) <EOL> url_parts [ <NUM_LIT:4> ] = urlencode ( query ) <EOL> self . url = urlunparse ( url_parts ) <EOL> def add_field ( self , name , value ) : <EOL> self . _fields [ name ] = value <EOL> def add_file ( self , name , filename , content ) : <EOL> self . _files [ name ] = { <EOL> '<STR_LIT:filename>' : filename , <EOL> '<STR_LIT:content>' : content , <EOL> } <EOL> def del_field ( self , name ) : <EOL> del self . _fields [ name ] <EOL> def del_file ( self , filename ) : <EOL> del self . _files [ filename ] <EOL> def encode_multipart_formdata ( self ) : <EOL> """<STR_LIT>""" <EOL> if not ( self . _fields or self . _files ) : <EOL> return None , None <EOL> NEWLINE = b'<STR_LIT:\r\n>' <EOL> BOUNDARY = self . _make_mime_boundary ( ) <EOL> content = BytesIO ( ) <EOL> for key in self . _fields : <EOL> content . write ( b'<STR_LIT>' + BOUNDARY + NEWLINE ) <EOL> content . write ( b'<STR_LIT>' <EOL> b'<STR_LIT>' % key . encode ( '<STR_LIT:utf-8>' ) ) <EOL> content . write ( NEWLINE + NEWLINE ) <EOL> if isinstance ( self . _fields [ key ] , six . binary_type ) : <EOL> content . write ( self . _fields [ key ] + NEWLINE ) <EOL> else : <EOL> content . write ( six . text_type ( self . _fields [ key ] ) . encode ( '<STR_LIT:utf-8>' ) <EOL> + NEWLINE ) <EOL> for key in self . _files : <EOL> filename = self . _files [ key ] [ '<STR_LIT:filename>' ] <EOL> value = self . _files [ key ] [ '<STR_LIT:content>' ] <EOL> mime_type = mimetypes . guess_type ( filename ) [ <NUM_LIT:0> ] <EOL> if mime_type : <EOL> mime_type = mime_type . encode ( '<STR_LIT:utf-8>' ) <EOL> else : <EOL> mime_type = b'<STR_LIT>' <EOL> content . write ( b'<STR_LIT>' + BOUNDARY + NEWLINE ) <EOL> content . write ( b'<STR_LIT>' <EOL> % key . encode ( '<STR_LIT:utf-8>' ) ) <EOL> content . write ( b'<STR_LIT>' % filename . encode ( '<STR_LIT:utf-8>' ) <EOL> + NEWLINE ) <EOL> content . write ( b'<STR_LIT>' % mime_type + NEWLINE ) <EOL> content . write ( NEWLINE ) <EOL> if isinstance ( value , six . text_type ) : <EOL> content . write ( value . encode ( '<STR_LIT:utf-8>' ) ) <EOL> else : <EOL> content . write ( value ) <EOL> content . write ( NEWLINE ) <EOL> content . write ( b'<STR_LIT>' + BOUNDARY + b'<STR_LIT>' + NEWLINE + NEWLINE ) <EOL> content_type = b'<STR_LIT>' % BOUNDARY <EOL> return content_type , content . getvalue ( ) <EOL> def _make_mime_boundary ( self ) : <EOL> """<STR_LIT>""" <EOL> fmt = '<STR_LIT>' % len ( repr ( sys . maxsize - <NUM_LIT:1> ) ) <EOL> token = random . randrange ( sys . maxsize ) <EOL> return ( b'<STR_LIT:=>' * <NUM_LIT:15> ) + ( fmt % token ) . encode ( '<STR_LIT:utf-8>' ) + b'<STR_LIT>' <EOL> class Request ( URLRequest ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url , body = '<STR_LIT>' , headers = { } , method = '<STR_LIT>' ) : <EOL> URLRequest . __init__ ( self , url , body , headers ) <EOL> self . method = method <EOL> def get_method ( self ) : <EOL> return self . method <EOL> class PresetHTTPAuthHandler ( BaseHandler ) : <EOL> """<STR_LIT>""" <EOL> handler_order = <NUM_LIT> <EOL> AUTH_HEADER = '<STR_LIT>' <EOL> def __init__ ( self , url , password_mgr ) : <EOL> self . url = url <EOL> self . password_mgr = password_mgr <EOL> self . used = False <EOL> def reset ( self , username , password ) : <EOL> self . password_mgr . rb_user = username <EOL> self . password_mgr . rb_pass = password <EOL> self . used = False <EOL> def http_request ( self , request ) : <EOL> if not self . used : <EOL> if self . password_mgr . api_token : <EOL> request . add_header ( self . AUTH_HEADER , <EOL> '<STR_LIT>' % self . password_mgr . api_token ) <EOL> self . used = True <EOL> elif self . password_mgr . rb_user : <EOL> username , password = self . password_mgr . find_user_password ( '<STR_LIT>' , self . url ) <EOL> raw = '<STR_LIT>' % ( username , password ) <EOL> request . add_header ( self . AUTH_HEADER , <EOL> '<STR_LIT>' % base64 . b64encode ( raw ) . strip ( ) ) <EOL> self . used = True <EOL> return request <EOL> https_request = http_request <EOL> class ReviewBoardHTTPErrorProcessor ( HTTPErrorProcessor ) : <EOL> """<STR_LIT>""" <EOL> def http_response ( self , request , response ) : <EOL> if not ( <NUM_LIT:200> <= response . code < <NUM_LIT> or <EOL> response . code == NOT_MODIFIED ) : <EOL> response = self . parent . error ( '<STR_LIT:http>' , request , response , <EOL> response . code , response . msg , <EOL> response . info ( ) ) <EOL> return response <EOL> https_response = http_response <EOL> class ReviewBoardHTTPBasicAuthHandler ( HTTPBasicAuthHandler ) : <EOL> """<STR_LIT>""" <EOL> OTP_TOKEN_HEADER = '<STR_LIT>' <EOL> MAX_OTP_TOKEN_ATTEMPTS = <NUM_LIT:5> <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> HTTPBasicAuthHandler . __init__ ( self , * args , ** kwargs ) <EOL> self . _retried = False <EOL> self . _lasturl = "<STR_LIT>" <EOL> self . _needs_otp_token = False <EOL> self . _otp_token_attempts = <NUM_LIT:0> <EOL> def retry_http_basic_auth ( self , host , request , realm , * args , ** kwargs ) : <EOL> if self . _lasturl != host : <EOL> self . _retried = False <EOL> self . _lasturl = host <EOL> if self . _retried : <EOL> return None <EOL> self . _retried = True <EOL> response = self . _do_http_basic_auth ( host , request , realm ) <EOL> if response and response . code != UNAUTHORIZED : <EOL> self . _retried = False <EOL> return response <EOL> def _do_http_basic_auth ( self , host , request , realm ) : <EOL> user , password = self . passwd . find_user_password ( realm , host ) <EOL> if password is None : <EOL> return None <EOL> raw = '<STR_LIT>' % ( user , password ) <EOL> auth = '<STR_LIT>' % base64 . b64encode ( raw ) . strip ( ) <EOL> if ( request . headers . get ( self . auth_header , None ) == auth and <EOL> ( not self . _needs_otp_token or <EOL> self . _otp_token_attempts > self . MAX_OTP_TOKEN_ATTEMPTS ) ) : <EOL> return None <EOL> request . add_unredirected_header ( self . auth_header , auth ) <EOL> try : <EOL> response = self . parent . open ( request , timeout = request . timeout ) <EOL> return response <EOL> except HTTPError as e : <EOL> if e . code == UNAUTHORIZED : <EOL> headers = e . info ( ) <EOL> otp_header = headers . get ( self . OTP_TOKEN_HEADER , '<STR_LIT>' ) <EOL> if otp_header . startswith ( '<STR_LIT>' ) : <EOL> self . _needs_otp_token = True <EOL> required , token_method = otp_header . split ( '<STR_LIT:;>' ) <EOL> token = self . passwd . get_otp_token ( request . get_full_url ( ) , <EOL> token_method . strip ( ) ) <EOL> if not token : <EOL> return None <EOL> request . add_unredirected_header ( self . OTP_TOKEN_HEADER , <EOL> token ) <EOL> self . _otp_token_attempts += <NUM_LIT:1> <EOL> return self . _do_http_basic_auth ( host , request , realm ) <EOL> raise <EOL> return None <EOL> class ReviewBoardHTTPPasswordMgr ( HTTPPasswordMgr ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , reviewboard_url , rb_user = None , rb_pass = None , <EOL> api_token = None , auth_callback = None , otp_token_callback = None ) : <EOL> HTTPPasswordMgr . __init__ ( self ) <EOL> self . passwd = { } <EOL> self . rb_url = reviewboard_url <EOL> self . rb_user = rb_user <EOL> self . rb_pass = rb_pass <EOL> self . api_token = api_token <EOL> self . auth_callback = auth_callback <EOL> self . otp_token_callback = otp_token_callback <EOL> def find_user_password ( self , realm , uri ) : <EOL> if realm == '<STR_LIT>' : <EOL> if self . auth_callback : <EOL> username , password = self . auth_callback ( realm , uri , <EOL> username = self . rb_user , <EOL> password = self . rb_pass ) <EOL> self . rb_user = username <EOL> self . rb_pass = password <EOL> return self . rb_user , self . rb_pass <EOL> else : <EOL> return HTTPPasswordMgr . find_user_password ( self , realm , uri ) <EOL> def get_otp_token ( self , uri , method ) : <EOL> if self . otp_token_callback : <EOL> return self . otp_token_callback ( uri , method ) <EOL> def create_cookie_jar ( cookie_file = None ) : <EOL> """<STR_LIT>""" <EOL> home_path = get_home_path ( ) <EOL> if not cookie_file : <EOL> cookie_file = os . path . join ( home_path , RBTOOLS_COOKIE_FILE ) <EOL> post_review_cookies = os . path . join ( home_path , <EOL> '<STR_LIT>' ) <EOL> if ( not os . path . isfile ( cookie_file ) and <EOL> os . path . isfile ( post_review_cookies ) ) : <EOL> try : <EOL> shutil . copyfile ( post_review_cookies , cookie_file ) <EOL> os . chmod ( cookie_file , <NUM_LIT> ) <EOL> except IOError as e : <EOL> logging . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" , e ) <EOL> if not os . path . isfile ( cookie_file ) : <EOL> try : <EOL> open ( cookie_file , '<STR_LIT:w>' ) . close ( ) <EOL> os . chmod ( cookie_file , <NUM_LIT> ) <EOL> except IOError as e : <EOL> logging . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' , e ) <EOL> return MozillaCookieJar ( cookie_file ) , cookie_file <EOL> class ReviewBoardServer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url , cookie_file = None , username = None , password = None , <EOL> api_token = None , agent = None , session = None , disable_proxy = False , <EOL> auth_callback = None , otp_token_callback = None , <EOL> verify_ssl = True , save_cookies = True , ext_auth_cookies = None ) : <EOL> if not url . endswith ( '<STR_LIT:/>' ) : <EOL> url += '<STR_LIT:/>' <EOL> self . url = url + '<STR_LIT>' <EOL> self . save_cookies = save_cookies <EOL> self . ext_auth_cookies = ext_auth_cookies <EOL> if self . save_cookies : <EOL> self . cookie_jar , self . cookie_file = create_cookie_jar ( <EOL> cookie_file = cookie_file ) <EOL> try : <EOL> self . cookie_jar . load ( ignore_expires = True ) <EOL> except IOError : <EOL> pass <EOL> else : <EOL> self . cookie_jar = CookieJar ( ) <EOL> self . cookie_file = None <EOL> if self . ext_auth_cookies : <EOL> try : <EOL> self . cookie_jar . load ( ext_auth_cookies , ignore_expires = True ) <EOL> except IOError as e : <EOL> logging . critical ( '<STR_LIT>' <EOL> '<STR_LIT>' , e ) <EOL> pass <EOL> parsed_url = urlparse ( url ) <EOL> self . domain = parsed_url [ <NUM_LIT:1> ] . partition ( '<STR_LIT::>' ) [ <NUM_LIT:0> ] <EOL> if self . domain . count ( '<STR_LIT:.>' ) < <NUM_LIT:1> : <EOL> self . domain = '<STR_LIT>' % self . domain <EOL> if session : <EOL> cookie = Cookie ( <EOL> version = <NUM_LIT:0> , <EOL> name = RB_COOKIE_NAME , <EOL> value = session , <EOL> port = None , <EOL> port_specified = False , <EOL> domain = self . domain , <EOL> domain_specified = True , <EOL> domain_initial_dot = True , <EOL> path = parsed_url [ <NUM_LIT:2> ] , <EOL> path_specified = True , <EOL> secure = False , <EOL> expires = None , <EOL> discard = False , <EOL> comment = None , <EOL> comment_url = None , <EOL> rest = { '<STR_LIT>' : None } ) <EOL> self . cookie_jar . set_cookie ( cookie ) <EOL> if self . save_cookies : <EOL> self . cookie_jar . save ( ) <EOL> if username : <EOL> try : <EOL> self . cookie_jar . clear ( self . domain , parsed_url [ <NUM_LIT:2> ] , <EOL> RB_COOKIE_NAME ) <EOL> except KeyError : <EOL> pass <EOL> password_mgr = ReviewBoardHTTPPasswordMgr ( self . url , <EOL> username , <EOL> password , <EOL> api_token , <EOL> auth_callback , <EOL> otp_token_callback ) <EOL> self . preset_auth_handler = PresetHTTPAuthHandler ( self . url , <EOL> password_mgr ) <EOL> handlers = [ ] <EOL> if not verify_ssl : <EOL> context = ssl . _create_unverified_context ( ) <EOL> handlers . append ( HTTPSHandler ( context = context ) ) <EOL> if disable_proxy : <EOL> handlers . append ( ProxyHandler ( { } ) ) <EOL> handlers += [ <EOL> HTTPCookieProcessor ( self . cookie_jar ) , <EOL> ReviewBoardHTTPBasicAuthHandler ( password_mgr ) , <EOL> HTTPDigestAuthHandler ( password_mgr ) , <EOL> self . preset_auth_handler , <EOL> ReviewBoardHTTPErrorProcessor ( ) , <EOL> ] <EOL> if agent : <EOL> self . agent = agent <EOL> else : <EOL> self . agent = ( '<STR_LIT>' + get_package_version ( ) ) . encode ( '<STR_LIT:utf-8>' ) <EOL> opener = build_opener ( * handlers ) <EOL> opener . addheaders = [ <EOL> ( b'<STR_LIT>' , self . agent ) , <EOL> ] <EOL> install_opener ( opener ) <EOL> self . _cache = None <EOL> self . _urlopen = urlopen <EOL> def enable_cache ( self , cache_location = None , in_memory = False ) : <EOL> """<STR_LIT>""" <EOL> if not self . _cache : <EOL> self . _cache = APICache ( create_db_in_memory = in_memory , <EOL> db_location = cache_location ) <EOL> self . _urlopen = self . _cache . make_request <EOL> def login ( self , username , password ) : <EOL> """<STR_LIT>""" <EOL> self . preset_auth_handler . reset ( username , password ) <EOL> def logout ( self ) : <EOL> """<STR_LIT>""" <EOL> self . preset_auth_handler . reset ( None , None ) <EOL> self . make_request ( HttpRequest ( '<STR_LIT>' % self . url , <EOL> method = '<STR_LIT>' ) ) <EOL> self . cookie_jar . clear ( self . domain ) <EOL> if self . save_cookies : <EOL> self . cookie_jar . save ( ) <EOL> def process_error ( self , http_status , data ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> rsp = json_loads ( data ) <EOL> assert rsp [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> logging . debug ( '<STR_LIT>' % <EOL> ( rsp [ '<STR_LIT>' ] [ '<STR_LIT:code>' ] , http_status , rsp [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) ) <EOL> logging . debug ( '<STR_LIT>' % rsp ) <EOL> raise create_api_error ( http_status , rsp [ '<STR_LIT>' ] [ '<STR_LIT:code>' ] , rsp , <EOL> rsp [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> except ValueError : <EOL> logging . debug ( '<STR_LIT>' % ( http_status , data ) ) <EOL> raise APIError ( http_status , None , None , data ) <EOL> def make_request ( self , request ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> content_type , body = request . encode_multipart_formdata ( ) <EOL> headers = request . headers <EOL> if body : <EOL> headers . update ( { <EOL> b'<STR_LIT:Content-Type>' : content_type , <EOL> b'<STR_LIT>' : str ( len ( body ) ) , <EOL> } ) <EOL> else : <EOL> headers [ b'<STR_LIT>' ] = '<STR_LIT:0>' <EOL> r = Request ( request . url . encode ( '<STR_LIT:utf-8>' ) , body , headers , <EOL> request . method . encode ( '<STR_LIT:utf-8>' ) ) <EOL> rsp = self . _urlopen ( r ) <EOL> except HTTPError as e : <EOL> self . process_error ( e . code , e . read ( ) ) <EOL> except URLError as e : <EOL> raise ServerInterfaceError ( '<STR_LIT:%s>' % e . reason ) <EOL> if self . save_cookies : <EOL> try : <EOL> self . cookie_jar . save ( ) <EOL> except IOError : <EOL> pass <EOL> return rsp </s>
