<s> __all__ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> import os <EOL> import sys <EOL> import socket <EOL> import errno <EOL> import time <EOL> import tempfile <EOL> import itertools <EOL> import _multiprocessing <EOL> from multiprocessing import current_process , AuthenticationError <EOL> from multiprocessing . util import get_temp_dir , Finalize , sub_debug , debug <EOL> from multiprocessing . forking import duplicate , close <EOL> BUFSIZE = <NUM_LIT> <EOL> _mmap_counter = itertools . count ( ) <EOL> default_family = '<STR_LIT>' <EOL> families = [ '<STR_LIT>' ] <EOL> if hasattr ( socket , '<STR_LIT>' ) : <EOL> default_family = '<STR_LIT>' <EOL> families += [ '<STR_LIT>' ] <EOL> if sys . platform == '<STR_LIT:win32>' : <EOL> default_family = '<STR_LIT>' <EOL> families += [ '<STR_LIT>' ] <EOL> def arbitrary_address ( family ) : <EOL> '''<STR_LIT>''' <EOL> if family == '<STR_LIT>' : <EOL> return ( '<STR_LIT:localhost>' , <NUM_LIT:0> ) <EOL> elif family == '<STR_LIT>' : <EOL> return tempfile . mktemp ( prefix = '<STR_LIT>' , dir = get_temp_dir ( ) ) <EOL> elif family == '<STR_LIT>' : <EOL> return tempfile . mktemp ( prefix = r'<STR_LIT>' % <EOL> ( os . getpid ( ) , _mmap_counter . next ( ) ) ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> def address_type ( address ) : <EOL> '''<STR_LIT>''' <EOL> if type ( address ) == tuple : <EOL> return '<STR_LIT>' <EOL> elif type ( address ) is str and address . startswith ( '<STR_LIT>' ) : <EOL> return '<STR_LIT>' <EOL> elif type ( address ) is str : <EOL> return '<STR_LIT>' <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % address ) <EOL> class Listener ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , address = None , family = None , backlog = <NUM_LIT:1> , authkey = None ) : <EOL> family = family or ( address and address_type ( address ) ) or default_family <EOL> address = address or arbitrary_address ( family ) <EOL> if family == '<STR_LIT>' : <EOL> self . _listener = PipeListener ( address , backlog ) <EOL> else : <EOL> self . _listener = SocketListener ( address , family , backlog ) <EOL> if authkey is not None and not isinstance ( authkey , bytes ) : <EOL> raise TypeError , '<STR_LIT>' <EOL> self . _authkey = authkey <EOL> def accept ( self ) : <EOL> '''<STR_LIT>''' <EOL> c = self . _listener . accept ( ) <EOL> if self . _authkey : <EOL> deliver_challenge ( c , self . _authkey ) <EOL> answer_challenge ( c , self . _authkey ) <EOL> return c <EOL> def close ( self ) : <EOL> '''<STR_LIT>''' <EOL> return self . _listener . close ( ) <EOL> address = property ( lambda self : self . _listener . _address ) <EOL> last_accepted = property ( lambda self : self . _listener . _last_accepted ) <EOL> def Client ( address , family = None , authkey = None ) : <EOL> '''<STR_LIT>''' <EOL> family = family or address_type ( address ) <EOL> if family == '<STR_LIT>' : <EOL> c = PipeClient ( address ) <EOL> else : <EOL> c = SocketClient ( address ) <EOL> if authkey is not None and not isinstance ( authkey , bytes ) : <EOL> raise TypeError , '<STR_LIT>' <EOL> if authkey is not None : <EOL> answer_challenge ( c , authkey ) <EOL> deliver_challenge ( c , authkey ) <EOL> return c <EOL> if sys . platform != '<STR_LIT:win32>' : <EOL> def Pipe ( duplex = True ) : <EOL> '''<STR_LIT>''' <EOL> if duplex : <EOL> s1 , s2 = socket . socketpair ( ) <EOL> c1 = _multiprocessing . Connection ( os . dup ( s1 . fileno ( ) ) ) <EOL> c2 = _multiprocessing . Connection ( os . dup ( s2 . fileno ( ) ) ) <EOL> s1 . close ( ) <EOL> s2 . close ( ) <EOL> else : <EOL> fd1 , fd2 = os . pipe ( ) <EOL> c1 = _multiprocessing . Connection ( fd1 , writable = False ) <EOL> c2 = _multiprocessing . Connection ( fd2 , readable = False ) <EOL> return c1 , c2 <EOL> else : <EOL> from . _multiprocessing import win32 <EOL> def Pipe ( duplex = True ) : <EOL> '''<STR_LIT>''' <EOL> address = arbitrary_address ( '<STR_LIT>' ) <EOL> if duplex : <EOL> openmode = win32 . PIPE_ACCESS_DUPLEX <EOL> access = win32 . GENERIC_READ | win32 . GENERIC_WRITE <EOL> obsize , ibsize = BUFSIZE , BUFSIZE <EOL> else : <EOL> openmode = win32 . PIPE_ACCESS_INBOUND <EOL> access = win32 . GENERIC_WRITE <EOL> obsize , ibsize = <NUM_LIT:0> , BUFSIZE <EOL> h1 = win32 . CreateNamedPipe ( <EOL> address , openmode , <EOL> win32 . PIPE_TYPE_MESSAGE | win32 . PIPE_READMODE_MESSAGE | <EOL> win32 . PIPE_WAIT , <EOL> <NUM_LIT:1> , obsize , ibsize , win32 . NMPWAIT_WAIT_FOREVER , win32 . NULL <EOL> ) <EOL> h2 = win32 . CreateFile ( <EOL> address , access , <NUM_LIT:0> , win32 . NULL , win32 . OPEN_EXISTING , <NUM_LIT:0> , win32 . NULL <EOL> ) <EOL> win32 . SetNamedPipeHandleState ( <EOL> h2 , win32 . PIPE_READMODE_MESSAGE , None , None <EOL> ) <EOL> try : <EOL> win32 . ConnectNamedPipe ( h1 , win32 . NULL ) <EOL> except WindowsError , e : <EOL> if e . args [ <NUM_LIT:0> ] != win32 . ERROR_PIPE_CONNECTED : <EOL> raise <EOL> c1 = _multiprocessing . PipeConnection ( h1 , writable = duplex ) <EOL> c2 = _multiprocessing . PipeConnection ( h2 , readable = duplex ) <EOL> return c1 , c2 <EOL> class SocketListener ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , address , family , backlog = <NUM_LIT:1> ) : <EOL> self . _socket = socket . socket ( getattr ( socket , family ) ) <EOL> self . _socket . bind ( address ) <EOL> self . _socket . listen ( backlog ) <EOL> self . _address = self . _socket . getsockname ( ) <EOL> self . _family = family <EOL> self . _last_accepted = None <EOL> if family == '<STR_LIT>' : <EOL> self . _unlink = Finalize ( <EOL> self , os . unlink , args = ( address , ) , exitpriority = <NUM_LIT:0> <EOL> ) <EOL> else : <EOL> self . _unlink = None <EOL> def accept ( self ) : <EOL> s , self . _last_accepted = self . _socket . accept ( ) <EOL> fd = duplicate ( s . fileno ( ) ) <EOL> conn = _multiprocessing . Connection ( fd ) <EOL> s . close ( ) <EOL> return conn <EOL> def close ( self ) : <EOL> self . _socket . close ( ) <EOL> if self . _unlink is not None : <EOL> self . _unlink ( ) <EOL> def SocketClient ( address ) : <EOL> '''<STR_LIT>''' <EOL> family = address_type ( address ) <EOL> s = socket . socket ( getattr ( socket , family ) ) <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> s . connect ( address ) <EOL> except socket . error , e : <EOL> if e . args [ <NUM_LIT:0> ] != errno . ECONNREFUSED : <EOL> debug ( '<STR_LIT>' , address ) <EOL> raise <EOL> time . sleep ( <NUM_LIT> ) <EOL> else : <EOL> break <EOL> else : <EOL> raise <EOL> fd = duplicate ( s . fileno ( ) ) <EOL> conn = _multiprocessing . Connection ( fd ) <EOL> s . close ( ) <EOL> return conn <EOL> if sys . platform == '<STR_LIT:win32>' : <EOL> class PipeListener ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , address , backlog = None ) : <EOL> self . _address = address <EOL> handle = win32 . CreateNamedPipe ( <EOL> address , win32 . PIPE_ACCESS_DUPLEX , <EOL> win32 . PIPE_TYPE_MESSAGE | win32 . PIPE_READMODE_MESSAGE | <EOL> win32 . PIPE_WAIT , <EOL> win32 . PIPE_UNLIMITED_INSTANCES , BUFSIZE , BUFSIZE , <EOL> win32 . NMPWAIT_WAIT_FOREVER , win32 . NULL <EOL> ) <EOL> self . _handle_queue = [ handle ] <EOL> self . _last_accepted = None <EOL> sub_debug ( '<STR_LIT>' , self . _address ) <EOL> self . close = Finalize ( <EOL> self , PipeListener . _finalize_pipe_listener , <EOL> args = ( self . _handle_queue , self . _address ) , exitpriority = <NUM_LIT:0> <EOL> ) <EOL> def accept ( self ) : <EOL> newhandle = win32 . CreateNamedPipe ( <EOL> self . _address , win32 . PIPE_ACCESS_DUPLEX , <EOL> win32 . PIPE_TYPE_MESSAGE | win32 . PIPE_READMODE_MESSAGE | <EOL> win32 . PIPE_WAIT , <EOL> win32 . PIPE_UNLIMITED_INSTANCES , BUFSIZE , BUFSIZE , <EOL> win32 . NMPWAIT_WAIT_FOREVER , win32 . NULL <EOL> ) <EOL> self . _handle_queue . append ( newhandle ) <EOL> handle = self . _handle_queue . pop ( <NUM_LIT:0> ) <EOL> try : <EOL> win32 . ConnectNamedPipe ( handle , win32 . NULL ) <EOL> except WindowsError , e : <EOL> if e . args [ <NUM_LIT:0> ] != win32 . ERROR_PIPE_CONNECTED : <EOL> raise <EOL> return _multiprocessing . PipeConnection ( handle ) <EOL> @ staticmethod <EOL> def _finalize_pipe_listener ( queue , address ) : <EOL> sub_debug ( '<STR_LIT>' , address ) <EOL> for handle in queue : <EOL> close ( handle ) <EOL> def PipeClient ( address ) : <EOL> '''<STR_LIT>''' <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> win32 . WaitNamedPipe ( address , <NUM_LIT:1000> ) <EOL> h = win32 . CreateFile ( <EOL> address , win32 . GENERIC_READ | win32 . GENERIC_WRITE , <EOL> <NUM_LIT:0> , win32 . NULL , win32 . OPEN_EXISTING , <NUM_LIT:0> , win32 . NULL <EOL> ) <EOL> except WindowsError , e : <EOL> if e . args [ <NUM_LIT:0> ] not in ( win32 . ERROR_SEM_TIMEOUT , <EOL> win32 . ERROR_PIPE_BUSY ) : <EOL> raise <EOL> else : <EOL> break <EOL> else : <EOL> raise <EOL> win32 . SetNamedPipeHandleState ( <EOL> h , win32 . PIPE_READMODE_MESSAGE , None , None <EOL> ) <EOL> return _multiprocessing . PipeConnection ( h ) <EOL> MESSAGE_LENGTH = <NUM_LIT:20> <EOL> CHALLENGE = b'<STR_LIT>' <EOL> WELCOME = b'<STR_LIT>' <EOL> FAILURE = b'<STR_LIT>' <EOL> def deliver_challenge ( connection , authkey ) : <EOL> import hmac <EOL> assert isinstance ( authkey , bytes ) <EOL> message = os . urandom ( MESSAGE_LENGTH ) <EOL> connection . send_bytes ( CHALLENGE + message ) <EOL> digest = hmac . new ( authkey , message ) . digest ( ) <EOL> response = connection . recv_bytes ( <NUM_LIT> ) <EOL> if response == digest : <EOL> connection . send_bytes ( WELCOME ) <EOL> else : <EOL> connection . send_bytes ( FAILURE ) <EOL> raise AuthenticationError ( '<STR_LIT>' ) <EOL> def answer_challenge ( connection , authkey ) : <EOL> import hmac <EOL> assert isinstance ( authkey , bytes ) <EOL> message = connection . recv_bytes ( <NUM_LIT> ) <EOL> assert message [ : len ( CHALLENGE ) ] == CHALLENGE , '<STR_LIT>' % message <EOL> message = message [ len ( CHALLENGE ) : ] <EOL> digest = hmac . new ( authkey , message ) . digest ( ) <EOL> connection . send_bytes ( digest ) <EOL> response = connection . recv_bytes ( <NUM_LIT> ) <EOL> if response != WELCOME : <EOL> raise AuthenticationError ( '<STR_LIT>' ) <EOL> class ConnectionWrapper ( object ) : <EOL> def __init__ ( self , conn , dumps , loads ) : <EOL> self . _conn = conn <EOL> self . _dumps = dumps <EOL> self . _loads = loads <EOL> for attr in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> obj = getattr ( conn , attr ) <EOL> setattr ( self , attr , obj ) <EOL> def send ( self , obj ) : <EOL> s = self . _dumps ( obj ) <EOL> self . _conn . send_bytes ( s ) <EOL> def recv ( self ) : <EOL> s = self . _conn . recv_bytes ( ) <EOL> return self . _loads ( s ) <EOL> def _xml_dumps ( obj ) : <EOL> return xmlrpclib . dumps ( ( obj , ) , None , None , None , <NUM_LIT:1> ) . encode ( '<STR_LIT:utf8>' ) <EOL> def _xml_loads ( s ) : <EOL> ( obj , ) , method = xmlrpclib . loads ( s . decode ( '<STR_LIT:utf8>' ) ) <EOL> return obj <EOL> class XmlListener ( Listener ) : <EOL> def accept ( self ) : <EOL> global xmlrpclib <EOL> import xmlrpclib <EOL> obj = Listener . accept ( self ) <EOL> return ConnectionWrapper ( obj , _xml_dumps , _xml_loads ) <EOL> def XmlClient ( * args , ** kwds ) : <EOL> global xmlrpclib <EOL> import xmlrpclib <EOL> return ConnectionWrapper ( Client ( * args , ** kwds ) , _xml_dumps , _xml_loads ) </s>
