<s> from deployer import std <EOL> from deployer . cli import HandlerType <EOL> from deployer . daemonize import daemonize <EOL> from deployer . exceptions import ActionException <EOL> from deployer . loggers import LoggerInterface <EOL> from deployer . loggers . default import DefaultLogger , IndentedDefaultLogger <EOL> from deployer . options import Options <EOL> from deployer . pseudo_terminal import Pty <EOL> from deployer . shell import Shell , ShellHandler , BuiltinType <EOL> from twisted . internet import reactor , defer , abstract , fdesc <EOL> from twisted . internet import threads <EOL> from twisted . internet . protocol import Protocol , Factory <EOL> from twisted . internet . error import CannotListenError <EOL> from contextlib import nested <EOL> from setproctitle import setproctitle <EOL> from tempfile import NamedTemporaryFile <EOL> import StringIO <EOL> import datetime <EOL> import getpass <EOL> import logging <EOL> import os <EOL> import pickle <EOL> import sys <EOL> __all__ = ( '<STR_LIT:start>' , ) <EOL> """<STR_LIT>""" <EOL> OPEN_TERMINAL_TIMEOUT = <NUM_LIT:2> <EOL> class SocketHandlerType ( HandlerType ) : <EOL> color = '<STR_LIT>' <EOL> postfix = '<STR_LIT>' <EOL> class NewShell ( ShellHandler ) : <EOL> """<STR_LIT>""" <EOL> is_leaf = True <EOL> handler_type = SocketHandlerType ( ) <EOL> def __call__ ( self ) : <EOL> print '<STR_LIT>' <EOL> self . shell . session . openNewShellFromThread ( ) <EOL> class Jobs ( ShellHandler ) : <EOL> is_leaf = True <EOL> handler_type = SocketHandlerType ( ) <EOL> def __call__ ( self ) : <EOL> print '<STR_LIT>' <EOL> class Monitor ( ShellHandler ) : <EOL> is_leaf = True <EOL> handler_type = SocketHandlerType ( ) <EOL> def __call__ ( self ) : <EOL> def monitor ( pty ) : <EOL> logger = IndentedDefaultLogger ( pty . stdout ) <EOL> self . shell . logger_interface . attach ( logger ) <EOL> pty . stdout . write ( '<STR_LIT>' ) <EOL> pty . stdin . read ( <NUM_LIT:1> ) <EOL> self . shell . logger_interface . detach ( logger ) <EOL> self . shell . session . connection . runInNewPtys ( monitor , focus = False ) <EOL> class CloseDeadPanes ( ShellHandler ) : <EOL> is_leaf = True <EOL> handler_type = BuiltinType ( ) <EOL> def __call__ ( self ) : <EOL> for c in set ( self . shell . connection . connectionPool ) : <EOL> if c . process_done : <EOL> c . send_close ( ) <EOL> class SocketShell ( Shell ) : <EOL> """<STR_LIT:U+0020>""" <EOL> def __init__ ( self , * a , ** kw ) : <EOL> self . connection = kw . pop ( '<STR_LIT>' ) <EOL> Shell . __init__ ( self , * a , ** kw ) <EOL> @ property <EOL> def extensions ( self ) : <EOL> return { <EOL> '<STR_LIT>' : NewShell , <EOL> '<STR_LIT>' : Monitor , <EOL> '<STR_LIT>' : CloseDeadPanes , <EOL> } <EOL> class SocketPty ( Pty ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , stdin , stdout , run_in_new_pty , interactive ) : <EOL> Pty . __init__ ( self , stdin , stdout , interactive ) <EOL> self . _run_in_new_ptys = run_in_new_pty <EOL> def run_in_auxiliary_ptys ( self , callbacks ) : <EOL> return self . _run_in_new_ptys ( callbacks ) <EOL> @ property <EOL> def auxiliary_ptys_are_available ( self ) : <EOL> return True <EOL> class SelectableFile ( abstract . FileDescriptor ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fp , callback ) : <EOL> self . fp = fp <EOL> fdesc . setNonBlocking ( fp ) <EOL> self . callback = callback <EOL> self . fileno = self . fp . fileno <EOL> abstract . FileDescriptor . __init__ ( self , reactor ) <EOL> def doRead ( self ) : <EOL> buf = self . fp . read ( <NUM_LIT> ) <EOL> if buf : <EOL> self . callback ( buf ) <EOL> class Connection ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , protocol ) : <EOL> self . root_node = protocol . factory . root_node <EOL> self . extra_loggers = protocol . factory . extra_loggers <EOL> self . runtime_options = protocol . factory . runtime_options <EOL> self . transportHandle = protocol . _handle <EOL> self . doneCallback = protocol . transport . loseConnection <EOL> self . connectionPool = protocol . factory . connectionPool <EOL> self . connection_shell = None <EOL> self . exit_status = <NUM_LIT:0> <EOL> self . process_done = False <EOL> master , self . slave = os . openpty ( ) <EOL> self . shell_in = os . fdopen ( master , '<STR_LIT:w>' , <NUM_LIT:0> ) <EOL> self . shell_out = os . fdopen ( master , '<STR_LIT:r>' , <NUM_LIT:0> ) <EOL> stdin = os . fdopen ( self . slave , '<STR_LIT:r>' , <NUM_LIT:0> ) <EOL> stdout = os . fdopen ( self . slave , '<STR_LIT:w>' , <NUM_LIT:0> ) <EOL> self . pty = SocketPty ( stdin , stdout , self . runInNewPtys , <EOL> interactive = protocol . factory . interactive ) <EOL> self . _startReading ( ) <EOL> def _startReading ( self ) : <EOL> """<STR_LIT>""" <EOL> def writeCallback ( data ) : <EOL> self . transportHandle ( '<STR_LIT>' , data ) <EOL> self . reader = SelectableFile ( self . shell_out , writeCallback ) <EOL> self . reader . startReading ( ) <EOL> def set_term_var ( self , value ) : <EOL> self . pty . set_term_var ( value ) <EOL> def finish ( self , exit_status = <NUM_LIT:0> , always_close = False ) : <EOL> """<STR_LIT>""" <EOL> self . exit_status = exit_status <EOL> self . process_done = True <EOL> close = ( always_close or not self . runtime_options [ '<STR_LIT>' ] . get_value_as_bool ( ) ) <EOL> self . transportHandle ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : self . exit_status , <EOL> '<STR_LIT>' : not close <EOL> } ) <EOL> if close : <EOL> self . send_close ( ) <EOL> def send_close ( self ) : <EOL> self . reader . stopReading ( ) <EOL> self . doneCallback ( ) <EOL> def __del__ ( self ) : <EOL> try : <EOL> self . shell_in . close ( ) <EOL> self . shell_out . close ( ) <EOL> except : <EOL> pass <EOL> def startShell ( self , clone_shell = None , cd_path = None , action_name = None , <EOL> parameters = None , open_scp_shell = False ) : <EOL> """<STR_LIT>""" <EOL> self . connection_shell = ConnectionShell ( self , clone_shell = clone_shell , cd_path = cd_path ) <EOL> self . connection_shell . startThread ( action_name = action_name , parameters = parameters , <EOL> open_scp_shell = open_scp_shell ) <EOL> def openNewConnection ( self , focus = False ) : <EOL> """<STR_LIT>""" <EOL> def _open ( ) : <EOL> self . transportHandle ( '<STR_LIT>' , { '<STR_LIT>' : focus } ) <EOL> reactor . callFromThread ( _open ) <EOL> def runInNewPtys ( self , funcs , focus = False ) : <EOL> """<STR_LIT>""" <EOL> if callable ( funcs ) : <EOL> funcs = [ funcs ] <EOL> else : <EOL> funcs = funcs [ : ] <EOL> def getConnection ( ) : <EOL> self . openNewConnection ( focus = focus ) <EOL> return PtyManager . getNewConnectionFromThread ( ) <EOL> connections = [ ] <EOL> for f in funcs : <EOL> try : <EOL> connections . append ( getConnection ( ) ) <EOL> except PtyManager . NoPtyConnection as e : <EOL> print '<STR_LIT>' <EOL> break <EOL> results = [ ] <EOL> done_d = defer . Deferred ( ) <EOL> todo = [ len ( funcs ) ] <EOL> def countDown ( ) : <EOL> todo [ <NUM_LIT:0> ] -= <NUM_LIT:1> <EOL> if todo [ <NUM_LIT:0> ] == <NUM_LIT:0> : <EOL> done_d . callback ( results ) <EOL> def thread ( f , index , connection ) : <EOL> """<STR_LIT>""" <EOL> sys . stdout . set_handler ( connection . pty . stdout ) <EOL> sys . stdin . set_handler ( connection . pty . stdin ) <EOL> try : <EOL> result = f ( connection . pty ) <EOL> except Exception as e : <EOL> print str ( e ) <EOL> result = '<STR_LIT>' <EOL> sys . stdout . del_handler ( ) <EOL> sys . stdin . del_handler ( ) <EOL> reactor . callFromThread ( connection . finish ) <EOL> return result <EOL> def startAllThreads ( ) : <EOL> """<STR_LIT>""" <EOL> def startThread ( f , conn ) : <EOL> index = len ( results ) <EOL> results . append ( None ) <EOL> d = threads . deferToThread ( thread , f , index , conn ) <EOL> @ d . addCallback <EOL> def done ( result ) : <EOL> results [ index ] = result <EOL> countDown ( ) <EOL> @ d . addErrback <EOL> def err ( failure ) : <EOL> results [ index ] = str ( failure ) <EOL> countDown ( ) <EOL> while connections : <EOL> conn = connections . pop ( ) <EOL> f = funcs . pop ( ) <EOL> startThread ( f , conn ) <EOL> threads . blockingCallFromThread ( reactor , startAllThreads ) <EOL> def handleRemainingInCurrentPty ( ) : <EOL> while funcs : <EOL> f = funcs . pop ( ) <EOL> try : <EOL> result = f ( self . pty ) <EOL> results . append ( result ) <EOL> except Exception as e : <EOL> results . append ( str ( e ) ) <EOL> countDown ( ) <EOL> handleRemainingInCurrentPty ( ) <EOL> class ForkResult ( object ) : <EOL> """<STR_LIT>""" <EOL> def join ( self ) : <EOL> """<STR_LIT>""" <EOL> if todo [ <NUM_LIT:0> ] == <NUM_LIT:0> : <EOL> return results <EOL> else : <EOL> return threads . blockingCallFromThread ( reactor , lambda : done_d ) <EOL> @ property <EOL> def result ( self ) : <EOL> if todo [ <NUM_LIT:0> ] == <NUM_LIT:0> : <EOL> return results <EOL> else : <EOL> raise AttributeError ( '<STR_LIT>' ) <EOL> return ForkResult ( ) <EOL> class ConnectionShell ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , connection , clone_shell = None , cd_path = None ) : <EOL> self . connection = connection <EOL> self . logger_interface = LoggerInterface ( ) <EOL> self . shell = SocketShell ( connection . root_node , connection . pty , connection . runtime_options , <EOL> self . logger_interface , clone_shell = clone_shell , connection = connection ) <EOL> self . cd_path = cd_path <EOL> def openNewShellFromThread ( self ) : <EOL> """<STR_LIT>""" <EOL> self . connection . openNewConnection ( focus = True ) <EOL> try : <EOL> new_connection = PtyManager . getNewConnectionFromThread ( ) <EOL> ConnectionShell ( new_connection , clone_shell = self . shell ) . startThread ( ) <EOL> except PtyManager . NoPtyConnection as e : <EOL> print '<STR_LIT>' <EOL> def openNewShellFromReactor ( self ) : <EOL> self . connection . openNewConnection ( focus = True ) <EOL> d = PtyManager . getNewConnection ( ) <EOL> @ d . addCallback <EOL> def openShell ( new_connection ) : <EOL> new_connection . startShell ( clone_shell = self . shell ) <EOL> @ d . addErrback <EOL> def failed ( failure ) : <EOL> pass <EOL> def startThread ( self , * a , ** kw ) : <EOL> threads . deferToThread ( lambda : self . thread ( * a , ** kw ) ) <EOL> def thread ( self , action_name = None , parameters = None , open_scp_shell = False ) : <EOL> parameters = parameters or [ ] <EOL> sys . stdout . set_handler ( self . connection . pty . stdout ) <EOL> sys . stdin . set_handler ( self . connection . pty . stdin ) <EOL> self . shell . session = self <EOL> with self . logger_interface . attach_in_block ( DefaultLogger ( print_group = False ) ) : <EOL> with nested ( * [ self . logger_interface . attach_in_block ( l ) for l in self . connection . extra_loggers ] ) : <EOL> if self . cd_path : <EOL> self . shell . cd ( self . cd_path ) <EOL> if action_name and open_scp_shell : <EOL> print "<STR_LIT>" <EOL> exit_status = <NUM_LIT:1> <EOL> elif open_scp_shell : <EOL> self . shell . open_scp_shell ( ) <EOL> exit_status = <NUM_LIT:0> <EOL> elif action_name : <EOL> try : <EOL> self . shell . run_action ( action_name , * parameters ) <EOL> exit_status = <NUM_LIT:0> <EOL> except ActionException : <EOL> exit_status = <NUM_LIT:1> <EOL> except Exception : <EOL> import traceback <EOL> traceback . print_exc ( ) <EOL> exit_status = <NUM_LIT:1> <EOL> else : <EOL> self . shell . cmdloop ( ) <EOL> exit_status = <NUM_LIT:0> <EOL> self . shell . session = None <EOL> self . shell = None <EOL> sys . stdout . del_handler ( ) <EOL> sys . stdin . del_handler ( ) <EOL> reactor . callFromThread ( lambda : self . connection . finish ( <EOL> exit_status = exit_status , always_close = True ) ) <EOL> class PtyManager ( object ) : <EOL> need_pty_callback = None <EOL> class NoPtyConnection ( Exception ) : <EOL> pass <EOL> @ classmethod <EOL> def getNewConnectionFromThread ( cls ) : <EOL> """<STR_LIT>""" <EOL> return threads . blockingCallFromThread ( reactor , cls . getNewConnection ) <EOL> @ classmethod <EOL> def getNewConnection ( cls ) : <EOL> d = defer . Deferred ( ) <EOL> def callback ( connection ) : <EOL> cls . need_pty_callback = None <EOL> timeout . cancel ( ) <EOL> d . callback ( connection ) <EOL> def timeout ( ) : <EOL> cls . need_pty_callback = None <EOL> d . errback ( cls . NoPtyConnection ( ) ) <EOL> timeout = reactor . callLater ( OPEN_TERMINAL_TIMEOUT , timeout ) <EOL> cls . need_pty_callback = staticmethod ( callback ) <EOL> return d <EOL> class CliClientProtocol ( Protocol ) : <EOL> def __init__ ( self ) : <EOL> self . _buffer = [ ] <EOL> self . connection = None <EOL> self . created = datetime . datetime . now ( ) <EOL> def dataReceived ( self , data ) : <EOL> try : <EOL> io = StringIO . StringIO ( '<STR_LIT>' . join ( self . _buffer + [ data ] ) ) <EOL> action , data = pickle . load ( io ) <EOL> if action == '<STR_LIT>' : <EOL> self . connection . shell_in . write ( data ) <EOL> elif action == '<STR_LIT>' : <EOL> self . connection . pty . set_size ( * data ) <EOL> elif action == '<STR_LIT>' : <EOL> processes = [ { <EOL> '<STR_LIT>' : c . connection_shell . shell . state . _node . __class__ . __name__ , <EOL> '<STR_LIT>' : c . connection_shell . shell . state . _node . __module__ , <EOL> '<STR_LIT>' : c . connection_shell . shell . currently_running or '<STR_LIT>' <EOL> } for c in self . factory . connectionPool if c . connection_shell and c . connection_shell . shell ] <EOL> self . _handle ( '<STR_LIT>' , { <EOL> '<STR_LIT>' : self . created . isoformat ( ) , <EOL> '<STR_LIT>' : self . connection . root_node . __class__ . __name__ , <EOL> '<STR_LIT>' : self . connection . root_node . __module__ , <EOL> '<STR_LIT>' : processes , <EOL> } ) <EOL> self . transport . loseConnection ( ) <EOL> elif action == '<STR_LIT>' : <EOL> logging . info ( '<STR_LIT>' ) <EOL> if self . connection . connection_shell : <EOL> self . connection . connection_shell . openNewShellFromReactor ( ) <EOL> else : <EOL> self . _handle ( '<STR_LIT>' , { '<STR_LIT>' : True } ) <EOL> elif action == '<STR_LIT>' : <EOL> logging . info ( '<STR_LIT>' ) <EOL> cd_path = data . get ( '<STR_LIT>' , None ) <EOL> action_name = data . get ( '<STR_LIT>' , None ) <EOL> parameters = data . get ( '<STR_LIT>' , None ) <EOL> open_scp_shell = data . get ( '<STR_LIT>' , False ) <EOL> term_var = data . get ( '<STR_LIT>' , False ) <EOL> self . connection . set_term_var ( term_var ) <EOL> if PtyManager . need_pty_callback and not action_name : <EOL> PtyManager . need_pty_callback ( self . connection ) <EOL> else : <EOL> self . connection . startShell ( cd_path = cd_path , <EOL> action_name = action_name , parameters = parameters , <EOL> open_scp_shell = open_scp_shell ) <EOL> remainder = io . read ( ) <EOL> self . _buffer = [ remainder ] <EOL> if len ( remainder ) : <EOL> self . dataReceived ( '<STR_LIT>' ) <EOL> except ( EOFError , ValueError ) as e : <EOL> if data : <EOL> self . _buffer . append ( data ) <EOL> def connectionLost ( self , reason ) : <EOL> """<STR_LIT>""" <EOL> logging . info ( '<STR_LIT>' ) <EOL> self . factory . connectionPool . remove ( self . connection ) <EOL> self . connection = None <EOL> if len ( self . factory . connectionPool ) == <NUM_LIT:0> and self . factory . shutdownOnLastDisconnect : <EOL> logging . info ( '<STR_LIT>' ) <EOL> reactor . stop ( ) <EOL> def _handle ( self , action , data ) : <EOL> self . transport . write ( pickle . dumps ( ( action , data ) ) ) <EOL> def connectionMade ( self ) : <EOL> self . connection = Connection ( self ) <EOL> self . factory . connectionPool . add ( self . connection ) <EOL> def startSocketServer ( root_node , shutdownOnLastDisconnect , interactive , socket = None , extra_loggers = None ) : <EOL> """<STR_LIT>""" <EOL> factory = Factory ( ) <EOL> factory . connectionPool = set ( ) <EOL> factory . protocol = CliClientProtocol <EOL> factory . shutdownOnLastDisconnect = shutdownOnLastDisconnect <EOL> factory . root_node = root_node <EOL> factory . interactive = interactive <EOL> factory . extra_loggers = extra_loggers or [ ] <EOL> factory . runtime_options = Options ( ) <EOL> if socket : <EOL> reactor . listenUNIX ( socket , factory ) <EOL> else : <EOL> i = <NUM_LIT:0> <EOL> while True : <EOL> try : <EOL> socket = '<STR_LIT>' % ( getpass . getuser ( ) , i ) <EOL> reactor . listenUNIX ( socket , factory ) <EOL> break <EOL> except CannotListenError : <EOL> i += <NUM_LIT:1> <EOL> if i == <NUM_LIT:100> : <EOL> logging . warning ( '<STR_LIT>' ) <EOL> raise <EOL> return socket <EOL> def start ( root_node , daemonized = False , shutdown_on_last_disconnect = False , thread_pool_size = <NUM_LIT:50> , <EOL> interactive = True , logfile = None , socket = None , extra_loggers = None ) : <EOL> """<STR_LIT>""" <EOL> root_node = root_node ( ) <EOL> socket2 = startSocketServer ( root_node , shutdownOnLastDisconnect = shutdown_on_last_disconnect , <EOL> interactive = interactive , socket = socket , extra_loggers = extra_loggers ) <EOL> def set_title ( stream_name ) : <EOL> suffix = ( '<STR_LIT>' % stream_name if stream_name else '<STR_LIT>' ) <EOL> setproctitle ( '<STR_LIT>' % <EOL> ( root_node . __class__ . __name__ , socket2 , suffix ) ) <EOL> def run_server ( ) : <EOL> stream = None <EOL> if logfile : <EOL> logging . basicConfig ( filename = logfile , level = logging . DEBUG ) <EOL> elif not daemonized : <EOL> logging . basicConfig ( filename = '<STR_LIT>' , level = logging . DEBUG ) <EOL> else : <EOL> stream = NamedTemporaryFile ( delete = True , suffix = socket2 . replace ( '<STR_LIT:/>' , '<STR_LIT:->' ) ) <EOL> logging . basicConfig ( stream = stream , level = logging . DEBUG ) <EOL> logging . info ( '<STR_LIT>' % socket2 ) <EOL> std . setup ( ) <EOL> if thread_pool_size : <EOL> reactor . suggestThreadPoolSize ( thread_pool_size ) <EOL> set_title ( stream . name if stream else logfile ) <EOL> reactor . run ( ) <EOL> if stream : <EOL> stream . close ( ) <EOL> if daemonized : <EOL> if daemonize ( ) : <EOL> run_server ( ) <EOL> sys . exit ( ) <EOL> else : <EOL> return socket2 <EOL> else : <EOL> run_server ( ) </s>
