<s> from twisted . internet import defer , reactor , protocol <EOL> from twisted . internet . error import DNSLookupError <EOL> from twisted . web . client import readBody , HTTPConnectionPool , Agent <EOL> from twisted . web . http_headers import Headers <EOL> from twisted . web . _newclient import ResponseDone <EOL> from synapse . http . endpoint import matrix_federation_endpoint <EOL> from synapse . util . async import sleep <EOL> from synapse . util . logcontext import preserve_context_over_fn <EOL> import synapse . metrics <EOL> from canonicaljson import encode_canonical_json <EOL> from synapse . api . errors import ( <EOL> SynapseError , Codes , HttpResponseException , <EOL> ) <EOL> from signedjson . sign import sign_json <EOL> import simplejson as json <EOL> import logging <EOL> import random <EOL> import sys <EOL> import urllib <EOL> import urlparse <EOL> logger = logging . getLogger ( __name__ ) <EOL> outbound_logger = logging . getLogger ( "<STR_LIT>" ) <EOL> metrics = synapse . metrics . get_metrics_for ( __name__ ) <EOL> outgoing_requests_counter = metrics . register_counter ( <EOL> "<STR_LIT>" , <EOL> labels = [ "<STR_LIT>" ] , <EOL> ) <EOL> incoming_responses_counter = metrics . register_counter ( <EOL> "<STR_LIT>" , <EOL> labels = [ "<STR_LIT>" , "<STR_LIT:code>" ] , <EOL> ) <EOL> MAX_LONG_RETRIES = <NUM_LIT:10> <EOL> MAX_SHORT_RETRIES = <NUM_LIT:3> <EOL> class MatrixFederationEndpointFactory ( object ) : <EOL> def __init__ ( self , hs ) : <EOL> self . tls_server_context_factory = hs . tls_server_context_factory <EOL> def endpointForURI ( self , uri ) : <EOL> destination = uri . netloc <EOL> return matrix_federation_endpoint ( <EOL> reactor , destination , timeout = <NUM_LIT:10> , <EOL> ssl_context_factory = self . tls_server_context_factory <EOL> ) <EOL> class MatrixFederationHttpClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , hs ) : <EOL> self . hs = hs <EOL> self . signing_key = hs . config . signing_key [ <NUM_LIT:0> ] <EOL> self . server_name = hs . hostname <EOL> pool = HTTPConnectionPool ( reactor ) <EOL> pool . maxPersistentPerHost = <NUM_LIT:10> <EOL> self . agent = Agent . usingEndpointFactory ( <EOL> reactor , MatrixFederationEndpointFactory ( hs ) , pool = pool <EOL> ) <EOL> self . clock = hs . get_clock ( ) <EOL> self . version_string = hs . version_string <EOL> self . _next_id = <NUM_LIT:1> <EOL> def _create_url ( self , destination , path_bytes , param_bytes , query_bytes ) : <EOL> return urlparse . urlunparse ( <EOL> ( "<STR_LIT>" , destination , path_bytes , param_bytes , query_bytes , "<STR_LIT>" ) <EOL> ) <EOL> @ defer . inlineCallbacks <EOL> def _create_request ( self , destination , method , path_bytes , <EOL> body_callback , headers_dict = { } , param_bytes = b"<STR_LIT>" , <EOL> query_bytes = b"<STR_LIT>" , retry_on_dns_fail = True , <EOL> timeout = None , long_retries = False ) : <EOL> """<STR_LIT>""" <EOL> headers_dict [ b"<STR_LIT>" ] = [ self . version_string ] <EOL> headers_dict [ b"<STR_LIT>" ] = [ destination ] <EOL> url_bytes = self . _create_url ( <EOL> destination , path_bytes , param_bytes , query_bytes <EOL> ) <EOL> txn_id = "<STR_LIT>" % ( method , self . _next_id ) <EOL> self . _next_id = ( self . _next_id + <NUM_LIT:1> ) % ( sys . maxint - <NUM_LIT:1> ) <EOL> outbound_logger . info ( <EOL> "<STR_LIT>" , <EOL> txn_id , destination , method , url_bytes <EOL> ) <EOL> if long_retries : <EOL> retries_left = MAX_LONG_RETRIES <EOL> else : <EOL> retries_left = MAX_SHORT_RETRIES <EOL> http_url_bytes = urlparse . urlunparse ( <EOL> ( "<STR_LIT>" , "<STR_LIT>" , path_bytes , param_bytes , query_bytes , "<STR_LIT>" ) <EOL> ) <EOL> log_result = None <EOL> try : <EOL> while True : <EOL> producer = None <EOL> if body_callback : <EOL> producer = body_callback ( method , http_url_bytes , headers_dict ) <EOL> try : <EOL> def send_request ( ) : <EOL> request_deferred = preserve_context_over_fn ( <EOL> self . agent . request , <EOL> method , <EOL> url_bytes , <EOL> Headers ( headers_dict ) , <EOL> producer <EOL> ) <EOL> return self . clock . time_bound_deferred ( <EOL> request_deferred , <EOL> time_out = timeout / <NUM_LIT> if timeout else <NUM_LIT> , <EOL> ) <EOL> response = yield preserve_context_over_fn ( <EOL> send_request , <EOL> ) <EOL> log_result = "<STR_LIT>" % ( response . code , response . phrase , ) <EOL> break <EOL> except Exception as e : <EOL> if not retry_on_dns_fail and isinstance ( e , DNSLookupError ) : <EOL> logger . warn ( <EOL> "<STR_LIT>" , <EOL> destination , <EOL> e <EOL> ) <EOL> log_result = "<STR_LIT>" % ( <EOL> destination , e <EOL> ) <EOL> raise <EOL> logger . warn ( <EOL> "<STR_LIT>" , <EOL> txn_id , <EOL> destination , <EOL> method , <EOL> url_bytes , <EOL> type ( e ) . __name__ , <EOL> _flatten_response_never_received ( e ) , <EOL> ) <EOL> log_result = "<STR_LIT>" % ( <EOL> type ( e ) . __name__ , _flatten_response_never_received ( e ) , <EOL> ) <EOL> if retries_left and not timeout : <EOL> if long_retries : <EOL> delay = <NUM_LIT:4> ** ( MAX_LONG_RETRIES + <NUM_LIT:1> - retries_left ) <EOL> delay = min ( delay , <NUM_LIT> ) <EOL> delay *= random . uniform ( <NUM_LIT> , <NUM_LIT> ) <EOL> else : <EOL> delay = <NUM_LIT:0.5> * <NUM_LIT:2> ** ( MAX_SHORT_RETRIES - retries_left ) <EOL> delay = min ( delay , <NUM_LIT:2> ) <EOL> delay *= random . uniform ( <NUM_LIT> , <NUM_LIT> ) <EOL> yield sleep ( delay ) <EOL> retries_left -= <NUM_LIT:1> <EOL> else : <EOL> raise <EOL> finally : <EOL> outbound_logger . info ( <EOL> "<STR_LIT>" , <EOL> txn_id , <EOL> destination , <EOL> log_result , <EOL> ) <EOL> if <NUM_LIT:200> <= response . code < <NUM_LIT> : <EOL> pass <EOL> else : <EOL> body = yield preserve_context_over_fn ( readBody , response ) <EOL> raise HttpResponseException ( <EOL> response . code , response . phrase , body <EOL> ) <EOL> defer . returnValue ( response ) <EOL> def sign_request ( self , destination , method , url_bytes , headers_dict , <EOL> content = None ) : <EOL> request = { <EOL> "<STR_LIT>" : method , <EOL> "<STR_LIT>" : url_bytes , <EOL> "<STR_LIT>" : self . server_name , <EOL> "<STR_LIT>" : destination , <EOL> } <EOL> if content is not None : <EOL> request [ "<STR_LIT:content>" ] = content <EOL> request = sign_json ( request , self . server_name , self . signing_key ) <EOL> auth_headers = [ ] <EOL> for key , sig in request [ "<STR_LIT>" ] [ self . server_name ] . items ( ) : <EOL> auth_headers . append ( bytes ( <EOL> "<STR_LIT>" % ( <EOL> self . server_name , key , sig , <EOL> ) <EOL> ) ) <EOL> headers_dict [ b"<STR_LIT>" ] = auth_headers <EOL> @ defer . inlineCallbacks <EOL> def put_json ( self , destination , path , data = { } , json_data_callback = None , <EOL> long_retries = False ) : <EOL> """<STR_LIT>""" <EOL> if not json_data_callback : <EOL> def json_data_callback ( ) : <EOL> return data <EOL> def body_callback ( method , url_bytes , headers_dict ) : <EOL> json_data = json_data_callback ( ) <EOL> self . sign_request ( <EOL> destination , method , url_bytes , headers_dict , json_data <EOL> ) <EOL> producer = _JsonProducer ( json_data ) <EOL> return producer <EOL> response = yield self . _create_request ( <EOL> destination . encode ( "<STR_LIT:ascii>" ) , <EOL> "<STR_LIT>" , <EOL> path . encode ( "<STR_LIT:ascii>" ) , <EOL> body_callback = body_callback , <EOL> headers_dict = { "<STR_LIT:Content-Type>" : [ "<STR_LIT:application/json>" ] } , <EOL> long_retries = long_retries , <EOL> ) <EOL> if <NUM_LIT:200> <= response . code < <NUM_LIT> : <EOL> c_type = response . headers . getRawHeaders ( "<STR_LIT:Content-Type>" ) <EOL> if "<STR_LIT:application/json>" not in c_type : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" <EOL> ) <EOL> body = yield preserve_context_over_fn ( readBody , response ) <EOL> defer . returnValue ( json . loads ( body ) ) <EOL> @ defer . inlineCallbacks <EOL> def post_json ( self , destination , path , data = { } , long_retries = True ) : <EOL> """<STR_LIT>""" <EOL> def body_callback ( method , url_bytes , headers_dict ) : <EOL> self . sign_request ( <EOL> destination , method , url_bytes , headers_dict , data <EOL> ) <EOL> return _JsonProducer ( data ) <EOL> response = yield self . _create_request ( <EOL> destination . encode ( "<STR_LIT:ascii>" ) , <EOL> "<STR_LIT:POST>" , <EOL> path . encode ( "<STR_LIT:ascii>" ) , <EOL> body_callback = body_callback , <EOL> headers_dict = { "<STR_LIT:Content-Type>" : [ "<STR_LIT:application/json>" ] } , <EOL> long_retries = True , <EOL> ) <EOL> if <NUM_LIT:200> <= response . code < <NUM_LIT> : <EOL> c_type = response . headers . getRawHeaders ( "<STR_LIT:Content-Type>" ) <EOL> if "<STR_LIT:application/json>" not in c_type : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" <EOL> ) <EOL> body = yield preserve_context_over_fn ( readBody , response ) <EOL> defer . returnValue ( json . loads ( body ) ) <EOL> @ defer . inlineCallbacks <EOL> def get_json ( self , destination , path , args = { } , retry_on_dns_fail = True , <EOL> timeout = None ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" , args ) <EOL> encoded_args = { } <EOL> for k , vs in args . items ( ) : <EOL> if isinstance ( vs , basestring ) : <EOL> vs = [ vs ] <EOL> encoded_args [ k ] = [ v . encode ( "<STR_LIT>" ) for v in vs ] <EOL> query_bytes = urllib . urlencode ( encoded_args , True ) <EOL> logger . debug ( "<STR_LIT>" , args , retry_on_dns_fail ) <EOL> def body_callback ( method , url_bytes , headers_dict ) : <EOL> self . sign_request ( destination , method , url_bytes , headers_dict ) <EOL> return None <EOL> response = yield self . _create_request ( <EOL> destination . encode ( "<STR_LIT:ascii>" ) , <EOL> "<STR_LIT:GET>" , <EOL> path . encode ( "<STR_LIT:ascii>" ) , <EOL> query_bytes = query_bytes , <EOL> body_callback = body_callback , <EOL> retry_on_dns_fail = retry_on_dns_fail , <EOL> timeout = timeout , <EOL> ) <EOL> if <NUM_LIT:200> <= response . code < <NUM_LIT> : <EOL> c_type = response . headers . getRawHeaders ( "<STR_LIT:Content-Type>" ) <EOL> if "<STR_LIT:application/json>" not in c_type : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" <EOL> ) <EOL> body = yield preserve_context_over_fn ( readBody , response ) <EOL> defer . returnValue ( json . loads ( body ) ) <EOL> @ defer . inlineCallbacks <EOL> def get_file ( self , destination , path , output_stream , args = { } , <EOL> retry_on_dns_fail = True , max_size = None ) : <EOL> """<STR_LIT>""" <EOL> encoded_args = { } <EOL> for k , vs in args . items ( ) : <EOL> if isinstance ( vs , basestring ) : <EOL> vs = [ vs ] <EOL> encoded_args [ k ] = [ v . encode ( "<STR_LIT>" ) for v in vs ] <EOL> query_bytes = urllib . urlencode ( encoded_args , True ) <EOL> logger . debug ( "<STR_LIT>" , args , retry_on_dns_fail ) <EOL> def body_callback ( method , url_bytes , headers_dict ) : <EOL> self . sign_request ( destination , method , url_bytes , headers_dict ) <EOL> return None <EOL> response = yield self . _create_request ( <EOL> destination . encode ( "<STR_LIT:ascii>" ) , <EOL> "<STR_LIT:GET>" , <EOL> path . encode ( "<STR_LIT:ascii>" ) , <EOL> query_bytes = query_bytes , <EOL> body_callback = body_callback , <EOL> retry_on_dns_fail = retry_on_dns_fail <EOL> ) <EOL> headers = dict ( response . headers . getAllRawHeaders ( ) ) <EOL> try : <EOL> length = yield preserve_context_over_fn ( <EOL> _readBodyToFile , <EOL> response , output_stream , max_size <EOL> ) <EOL> except : <EOL> logger . exception ( "<STR_LIT>" ) <EOL> raise <EOL> defer . returnValue ( ( length , headers ) ) <EOL> class _ReadBodyToFileProtocol ( protocol . Protocol ) : <EOL> def __init__ ( self , stream , deferred , max_size ) : <EOL> self . stream = stream <EOL> self . deferred = deferred <EOL> self . length = <NUM_LIT:0> <EOL> self . max_size = max_size <EOL> def dataReceived ( self , data ) : <EOL> self . stream . write ( data ) <EOL> self . length += len ( data ) <EOL> if self . max_size is not None and self . length >= self . max_size : <EOL> self . deferred . errback ( SynapseError ( <EOL> <NUM_LIT> , <EOL> "<STR_LIT>" % ( self . max_size , ) , <EOL> Codes . TOO_LARGE , <EOL> ) ) <EOL> self . deferred = defer . Deferred ( ) <EOL> self . transport . loseConnection ( ) <EOL> def connectionLost ( self , reason ) : <EOL> if reason . check ( ResponseDone ) : <EOL> self . deferred . callback ( self . length ) <EOL> else : <EOL> self . deferred . errback ( reason ) <EOL> def _readBodyToFile ( response , stream , max_size ) : <EOL> d = defer . Deferred ( ) <EOL> response . deliverBody ( _ReadBodyToFileProtocol ( stream , d , max_size ) ) <EOL> return d <EOL> class _JsonProducer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , jsn ) : <EOL> self . reset ( jsn ) <EOL> def reset ( self , jsn ) : <EOL> self . body = encode_canonical_json ( jsn ) <EOL> self . length = len ( self . body ) <EOL> def startProducing ( self , consumer ) : <EOL> consumer . write ( self . body ) <EOL> return defer . succeed ( None ) <EOL> def pauseProducing ( self ) : <EOL> pass <EOL> def stopProducing ( self ) : <EOL> pass <EOL> def resumeProducing ( self ) : <EOL> pass <EOL> def _flatten_response_never_received ( e ) : <EOL> if hasattr ( e , "<STR_LIT>" ) : <EOL> return "<STR_LIT:U+002CU+0020>" . join ( <EOL> _flatten_response_never_received ( f . value ) <EOL> for f in e . reasons <EOL> ) <EOL> else : <EOL> return "<STR_LIT>" % ( type ( e ) . __name__ , e . message , ) </s>
