<s> '''<STR_LIT>''' <EOL> from __future__ import absolute_import <EOL> import logging <EOL> import msgpack <EOL> import socket <EOL> import os <EOL> import weakref <EOL> import time <EOL> import traceback <EOL> import salt . crypt <EOL> import salt . utils <EOL> import salt . utils . verify <EOL> import salt . utils . event <EOL> import salt . utils . async <EOL> import salt . payload <EOL> import salt . exceptions <EOL> import salt . transport . frame <EOL> import salt . transport . ipc <EOL> import salt . transport . client <EOL> import salt . transport . server <EOL> import salt . transport . mixins . auth <EOL> import salt . ext . six as six <EOL> from salt . exceptions import SaltReqTimeoutError , SaltClientError <EOL> from salt . transport import iter_transport_opts <EOL> import tornado <EOL> import tornado . tcpserver <EOL> import tornado . gen <EOL> import tornado . concurrent <EOL> import tornado . tcpclient <EOL> import tornado . netutil <EOL> if six . PY2 : <EOL> import urlparse <EOL> else : <EOL> import urllib . parse as urlparse <EOL> from Crypto . Cipher import PKCS1_OAEP <EOL> if six . PY3 and salt . utils . is_windows ( ) : <EOL> USE_LOAD_BALANCER = True <EOL> else : <EOL> USE_LOAD_BALANCER = False <EOL> if USE_LOAD_BALANCER : <EOL> import threading <EOL> import multiprocessing <EOL> import errno <EOL> import tornado . util <EOL> from salt . utils . process import SignalHandlingMultiprocessingProcess <EOL> log = logging . getLogger ( __name__ ) <EOL> def _set_tcp_keepalive ( sock , opts ) : <EOL> '''<STR_LIT>''' <EOL> if hasattr ( socket , '<STR_LIT>' ) : <EOL> if opts . get ( '<STR_LIT>' , False ) : <EOL> sock . setsockopt ( socket . SOL_SOCKET , socket . SO_KEEPALIVE , <NUM_LIT:1> ) <EOL> if hasattr ( socket , '<STR_LIT>' ) : <EOL> if hasattr ( socket , '<STR_LIT>' ) : <EOL> tcp_keepalive_idle = opts . get ( '<STR_LIT>' , - <NUM_LIT:1> ) <EOL> if tcp_keepalive_idle > <NUM_LIT:0> : <EOL> sock . setsockopt ( <EOL> socket . SOL_TCP , socket . TCP_KEEPIDLE , <EOL> int ( tcp_keepalive_idle ) ) <EOL> if hasattr ( socket , '<STR_LIT>' ) : <EOL> tcp_keepalive_cnt = opts . get ( '<STR_LIT>' , - <NUM_LIT:1> ) <EOL> if tcp_keepalive_cnt > <NUM_LIT:0> : <EOL> sock . setsockopt ( <EOL> socket . SOL_TCP , socket . TCP_KEEPCNT , <EOL> int ( tcp_keepalive_cnt ) ) <EOL> if hasattr ( socket , '<STR_LIT>' ) : <EOL> tcp_keepalive_intvl = opts . get ( '<STR_LIT>' , - <NUM_LIT:1> ) <EOL> if tcp_keepalive_intvl > <NUM_LIT:0> : <EOL> sock . setsockopt ( <EOL> socket . SOL_TCP , socket . TCP_KEEPINTVL , <EOL> int ( tcp_keepalive_intvl ) ) <EOL> if hasattr ( socket , '<STR_LIT>' ) : <EOL> tcp_keepalive_idle = opts . get ( '<STR_LIT>' , - <NUM_LIT:1> ) <EOL> tcp_keepalive_intvl = opts . get ( '<STR_LIT>' , - <NUM_LIT:1> ) <EOL> if tcp_keepalive_idle > <NUM_LIT:0> or tcp_keepalive_intvl > <NUM_LIT:0> : <EOL> if tcp_keepalive_idle <= <NUM_LIT:0> : <EOL> tcp_keepalive_idle = <NUM_LIT> <EOL> if tcp_keepalive_intvl <= <NUM_LIT:0> : <EOL> tcp_keepalive_intvl = <NUM_LIT:1> <EOL> sock . ioctl ( socket . SIO_KEEPALIVE_VALS , ( <EOL> <NUM_LIT:1> , int ( tcp_keepalive_idle * <NUM_LIT:1000> ) , <EOL> int ( tcp_keepalive_intvl * <NUM_LIT:1000> ) ) ) <EOL> else : <EOL> sock . setsockopt ( socket . SOL_SOCKET , socket . SO_KEEPALIVE , <NUM_LIT:0> ) <EOL> if USE_LOAD_BALANCER : <EOL> class LoadBalancerServer ( SignalHandlingMultiprocessingProcess ) : <EOL> '''<STR_LIT>''' <EOL> backlog = <NUM_LIT> <EOL> def __init__ ( self , opts , socket_queue , log_queue = None ) : <EOL> super ( LoadBalancerServer , self ) . __init__ ( log_queue = log_queue ) <EOL> self . opts = opts <EOL> self . socket_queue = socket_queue <EOL> self . _socket = None <EOL> def __setstate__ ( self , state ) : <EOL> self . _is_child = True <EOL> self . __init__ ( <EOL> state [ '<STR_LIT>' ] , <EOL> state [ '<STR_LIT>' ] , <EOL> log_queue = state [ '<STR_LIT>' ] <EOL> ) <EOL> def __getstate__ ( self ) : <EOL> return { '<STR_LIT>' : self . opts , <EOL> '<STR_LIT>' : self . socket_queue , <EOL> '<STR_LIT>' : self . log_queue } <EOL> def close ( self ) : <EOL> if self . _socket is not None : <EOL> self . _socket . shutdown ( socket . SHUT_RDWR ) <EOL> self . _socket . close ( ) <EOL> self . _socket = None <EOL> def __del__ ( self ) : <EOL> self . close ( ) <EOL> def run ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . _socket = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> self . _socket . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> _set_tcp_keepalive ( self . _socket , self . opts ) <EOL> self . _socket . setblocking ( <NUM_LIT:1> ) <EOL> self . _socket . bind ( ( self . opts [ '<STR_LIT>' ] , int ( self . opts [ '<STR_LIT>' ] ) ) ) <EOL> self . _socket . listen ( self . backlog ) <EOL> while True : <EOL> try : <EOL> connection , address = self . _socket . accept ( ) <EOL> self . socket_queue . put ( ( connection , address ) , True , None ) <EOL> except socket . error as e : <EOL> if tornado . util . errno_from_exception ( e ) == errno . ECONNABORTED : <EOL> continue <EOL> raise <EOL> class AsyncTCPReqChannel ( salt . transport . client . ReqChannel ) : <EOL> '''<STR_LIT>''' <EOL> instance_map = weakref . WeakKeyDictionary ( ) <EOL> def __new__ ( cls , opts , ** kwargs ) : <EOL> '''<STR_LIT>''' <EOL> io_loop = kwargs . get ( '<STR_LIT>' ) or tornado . ioloop . IOLoop . current ( ) <EOL> if io_loop not in cls . instance_map : <EOL> cls . instance_map [ io_loop ] = weakref . WeakValueDictionary ( ) <EOL> loop_instance_map = cls . instance_map [ io_loop ] <EOL> key = cls . __key ( opts , ** kwargs ) <EOL> if key not in loop_instance_map : <EOL> log . debug ( '<STR_LIT>' . format ( key ) ) <EOL> new_obj = object . __new__ ( cls ) <EOL> new_obj . __singleton_init__ ( opts , ** kwargs ) <EOL> loop_instance_map [ key ] = new_obj <EOL> else : <EOL> log . debug ( '<STR_LIT>' . format ( key ) ) <EOL> return loop_instance_map [ key ] <EOL> @ classmethod <EOL> def __key ( cls , opts , ** kwargs ) : <EOL> if '<STR_LIT>' in kwargs : <EOL> opts [ '<STR_LIT>' ] = kwargs [ '<STR_LIT>' ] <EOL> return ( opts [ '<STR_LIT>' ] , <EOL> opts [ '<STR_LIT:id>' ] , <EOL> opts [ '<STR_LIT>' ] , <EOL> kwargs . get ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ) <EOL> def __init__ ( self , opts , ** kwargs ) : <EOL> pass <EOL> def __singleton_init__ ( self , opts , ** kwargs ) : <EOL> self . opts = dict ( opts ) <EOL> self . serial = salt . payload . Serial ( self . opts ) <EOL> self . crypt = kwargs . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . io_loop = kwargs . get ( '<STR_LIT>' ) or tornado . ioloop . IOLoop . current ( ) <EOL> if self . crypt != '<STR_LIT>' : <EOL> self . auth = salt . crypt . AsyncAuth ( self . opts , io_loop = self . io_loop ) <EOL> resolver = kwargs . get ( '<STR_LIT>' ) <EOL> parse = urlparse . urlparse ( self . opts [ '<STR_LIT>' ] ) <EOL> host , port = parse . netloc . rsplit ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> self . master_addr = ( host , int ( port ) ) <EOL> self . _closing = False <EOL> self . message_client = SaltMessageClient ( <EOL> self . opts , host , int ( port ) , io_loop = self . io_loop , <EOL> resolver = resolver ) <EOL> def close ( self ) : <EOL> if self . _closing : <EOL> return <EOL> self . _closing = True <EOL> self . message_client . close ( ) <EOL> def __del__ ( self ) : <EOL> self . close ( ) <EOL> def _package_load ( self , load ) : <EOL> return { <EOL> '<STR_LIT>' : self . crypt , <EOL> '<STR_LIT>' : load , <EOL> } <EOL> @ tornado . gen . coroutine <EOL> def crypted_transfer_decode_dictentry ( self , load , dictkey = None , tries = <NUM_LIT:3> , timeout = <NUM_LIT> ) : <EOL> if not self . auth . authenticated : <EOL> yield self . auth . authenticate ( ) <EOL> ret = yield self . message_client . send ( self . _package_load ( self . auth . crypticle . dumps ( load ) ) , timeout = timeout ) <EOL> key = self . auth . get_keys ( ) <EOL> cipher = PKCS1_OAEP . new ( key ) <EOL> aes = cipher . decrypt ( ret [ '<STR_LIT:key>' ] ) <EOL> pcrypt = salt . crypt . Crypticle ( self . opts , aes ) <EOL> data = pcrypt . loads ( ret [ dictkey ] ) <EOL> if six . PY3 : <EOL> data = salt . transport . frame . decode_embedded_strs ( data ) <EOL> raise tornado . gen . Return ( data ) <EOL> @ tornado . gen . coroutine <EOL> def _crypted_transfer ( self , load , tries = <NUM_LIT:3> , timeout = <NUM_LIT> ) : <EOL> '''<STR_LIT>''' <EOL> @ tornado . gen . coroutine <EOL> def _do_transfer ( ) : <EOL> data = yield self . message_client . send ( self . _package_load ( self . auth . crypticle . dumps ( load ) ) , <EOL> timeout = timeout , <EOL> ) <EOL> if data : <EOL> data = self . auth . crypticle . loads ( data ) <EOL> if six . PY3 : <EOL> data = salt . transport . frame . decode_embedded_strs ( data ) <EOL> raise tornado . gen . Return ( data ) <EOL> if not self . auth . authenticated : <EOL> yield self . auth . authenticate ( ) <EOL> try : <EOL> ret = yield _do_transfer ( ) <EOL> raise tornado . gen . Return ( ret ) <EOL> except salt . crypt . AuthenticationError : <EOL> yield self . auth . authenticate ( ) <EOL> ret = yield _do_transfer ( ) <EOL> raise tornado . gen . Return ( ret ) <EOL> @ tornado . gen . coroutine <EOL> def _uncrypted_transfer ( self , load , tries = <NUM_LIT:3> , timeout = <NUM_LIT> ) : <EOL> ret = yield self . message_client . send ( self . _package_load ( load ) , timeout = timeout ) <EOL> raise tornado . gen . Return ( ret ) <EOL> @ tornado . gen . coroutine <EOL> def send ( self , load , tries = <NUM_LIT:3> , timeout = <NUM_LIT> ) : <EOL> '''<STR_LIT>''' <EOL> try : <EOL> if self . crypt == '<STR_LIT>' : <EOL> ret = yield self . _uncrypted_transfer ( load , tries = tries , timeout = timeout ) <EOL> else : <EOL> ret = yield self . _crypted_transfer ( load , tries = tries , timeout = timeout ) <EOL> except tornado . iostream . StreamClosedError : <EOL> raise SaltClientError ( '<STR_LIT>' ) <EOL> raise tornado . gen . Return ( ret ) <EOL> class AsyncTCPPubChannel ( salt . transport . mixins . auth . AESPubClientMixin , salt . transport . client . AsyncPubChannel ) : <EOL> def __init__ ( self , <EOL> opts , <EOL> ** kwargs ) : <EOL> self . opts = opts <EOL> self . serial = salt . payload . Serial ( self . opts ) <EOL> self . crypt = kwargs . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . io_loop = kwargs . get ( '<STR_LIT>' ) or tornado . ioloop . IOLoop . current ( ) <EOL> self . connected = False <EOL> self . _closing = False <EOL> self . _reconnected = False <EOL> self . event = salt . utils . event . get_event ( <EOL> '<STR_LIT>' , <EOL> opts = self . opts , <EOL> listen = False <EOL> ) <EOL> def close ( self ) : <EOL> if self . _closing : <EOL> return <EOL> self . _closing = True <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> self . message_client . close ( ) <EOL> def __del__ ( self ) : <EOL> self . close ( ) <EOL> def _package_load ( self , load ) : <EOL> return { <EOL> '<STR_LIT>' : self . crypt , <EOL> '<STR_LIT>' : load , <EOL> } <EOL> @ tornado . gen . coroutine <EOL> def send_id ( self , tok , force_auth ) : <EOL> '''<STR_LIT>''' <EOL> load = { '<STR_LIT:id>' : self . opts [ '<STR_LIT:id>' ] , '<STR_LIT>' : tok } <EOL> @ tornado . gen . coroutine <EOL> def _do_transfer ( ) : <EOL> msg = self . _package_load ( self . auth . crypticle . dumps ( load ) ) <EOL> package = salt . transport . frame . frame_msg ( msg , header = None ) <EOL> yield self . message_client . _stream . write ( package ) <EOL> raise tornado . gen . Return ( True ) <EOL> if force_auth or not self . auth . authenticated : <EOL> yield self . auth . authenticate ( ) <EOL> try : <EOL> ret = yield _do_transfer ( ) <EOL> raise tornado . gen . Return ( ret ) <EOL> except salt . crypt . AuthenticationError : <EOL> yield self . auth . authenticate ( ) <EOL> ret = yield _do_transfer ( ) <EOL> raise tornado . gen . Return ( ret ) <EOL> @ tornado . gen . coroutine <EOL> def connect_callback ( self , result ) : <EOL> if self . _closing : <EOL> return <EOL> yield self . send_id ( self . tok , self . _reconnected ) <EOL> self . connected = True <EOL> self . event . fire_event ( <EOL> { '<STR_LIT>' : self . opts [ '<STR_LIT>' ] } , <EOL> '<STR_LIT>' <EOL> ) <EOL> if self . _reconnected : <EOL> if self . opts . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> data = '<STR_LIT>' . format ( <EOL> self . opts [ '<STR_LIT:id>' ] , <EOL> time . asctime ( ) <EOL> ) <EOL> tag = salt . utils . event . tagify ( <EOL> [ self . opts [ '<STR_LIT:id>' ] , '<STR_LIT:start>' ] , <EOL> '<STR_LIT>' <EOL> ) <EOL> else : <EOL> data = '<STR_LIT>' . format ( <EOL> self . opts [ '<STR_LIT:id>' ] , <EOL> time . asctime ( ) <EOL> ) <EOL> tag = salt . utils . event . tagify ( <EOL> [ self . opts [ '<STR_LIT:id>' ] , '<STR_LIT:start>' ] , <EOL> '<STR_LIT>' <EOL> ) <EOL> load = { '<STR_LIT:id>' : self . opts [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : self . tok , <EOL> '<STR_LIT:data>' : data , <EOL> '<STR_LIT>' : tag } <EOL> req_channel = salt . utils . async . SyncWrapper ( <EOL> AsyncTCPReqChannel , ( self . opts , ) <EOL> ) <EOL> try : <EOL> req_channel . send ( load , timeout = <NUM_LIT> ) <EOL> except salt . exceptions . SaltReqTimeoutError : <EOL> log . info ( '<STR_LIT>' ) <EOL> except Exception : <EOL> log . info ( '<STR_LIT>' . format ( <EOL> traceback . format_exc ( ) ) <EOL> ) <EOL> else : <EOL> self . _reconnected = True <EOL> def disconnect_callback ( self ) : <EOL> if self . _closing : <EOL> return <EOL> self . connected = False <EOL> self . event . fire_event ( <EOL> { '<STR_LIT>' : self . opts [ '<STR_LIT>' ] } , <EOL> '<STR_LIT>' <EOL> ) <EOL> @ tornado . gen . coroutine <EOL> def connect ( self ) : <EOL> try : <EOL> self . auth = salt . crypt . AsyncAuth ( self . opts , io_loop = self . io_loop ) <EOL> self . tok = self . auth . gen_token ( '<STR_LIT>' ) <EOL> if not self . auth . authenticated : <EOL> yield self . auth . authenticate ( ) <EOL> self . message_client = SaltMessageClient ( <EOL> self . opts , <EOL> self . opts [ '<STR_LIT>' ] , <EOL> int ( self . auth . creds [ '<STR_LIT>' ] ) , <EOL> io_loop = self . io_loop , <EOL> connect_callback = self . connect_callback , <EOL> disconnect_callback = self . disconnect_callback ) <EOL> yield self . message_client . connect ( ) <EOL> self . connected = True <EOL> except KeyboardInterrupt : <EOL> raise <EOL> except : <EOL> raise SaltClientError ( '<STR_LIT>' ) <EOL> def on_recv ( self , callback ) : <EOL> '''<STR_LIT>''' <EOL> if callback is None : <EOL> return self . message_client . on_recv ( callback ) <EOL> @ tornado . gen . coroutine <EOL> def wrap_callback ( body ) : <EOL> if not isinstance ( body , dict ) : <EOL> body = msgpack . loads ( body ) <EOL> if six . PY3 : <EOL> body = salt . transport . frame . decode_embedded_strs ( body ) <EOL> ret = yield self . _decode_payload ( body ) <EOL> callback ( ret ) <EOL> return self . message_client . on_recv ( wrap_callback ) <EOL> class TCPReqServerChannel ( salt . transport . mixins . auth . AESReqServerMixin , salt . transport . server . ReqServerChannel ) : <EOL> backlog = <NUM_LIT:5> <EOL> def __init__ ( self , opts ) : <EOL> salt . transport . server . ReqServerChannel . __init__ ( self , opts ) <EOL> self . _socket = None <EOL> @ property <EOL> def socket ( self ) : <EOL> return self . _socket <EOL> def close ( self ) : <EOL> if self . _socket is not None : <EOL> self . _socket . shutdown ( socket . SHUT_RDWR ) <EOL> self . _socket . close ( ) <EOL> self . _socket = None <EOL> def __del__ ( self ) : <EOL> self . close ( ) <EOL> def pre_fork ( self , process_manager ) : <EOL> '''<STR_LIT>''' <EOL> salt . transport . mixins . auth . AESReqServerMixin . pre_fork ( self , process_manager ) <EOL> if USE_LOAD_BALANCER : <EOL> self . socket_queue = multiprocessing . Queue ( ) <EOL> process_manager . add_process ( <EOL> LoadBalancerServer , args = ( self . opts , self . socket_queue ) <EOL> ) <EOL> elif not salt . utils . is_windows ( ) : <EOL> self . _socket = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> self . _socket . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> _set_tcp_keepalive ( self . _socket , self . opts ) <EOL> self . _socket . setblocking ( <NUM_LIT:0> ) <EOL> self . _socket . bind ( ( self . opts [ '<STR_LIT>' ] , int ( self . opts [ '<STR_LIT>' ] ) ) ) <EOL> def post_fork ( self , payload_handler , io_loop ) : <EOL> '''<STR_LIT>''' <EOL> self . payload_handler = payload_handler <EOL> self . io_loop = io_loop <EOL> self . serial = salt . payload . Serial ( self . opts ) <EOL> if USE_LOAD_BALANCER : <EOL> self . req_server = LoadBalancerWorker ( <EOL> self . socket_queue , self . handle_message , io_loop = self . io_loop <EOL> ) <EOL> else : <EOL> if salt . utils . is_windows ( ) : <EOL> self . _socket = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> self . _socket . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> _set_tcp_keepalive ( self . _socket , self . opts ) <EOL> self . _socket . setblocking ( <NUM_LIT:0> ) <EOL> self . _socket . bind ( ( self . opts [ '<STR_LIT>' ] , int ( self . opts [ '<STR_LIT>' ] ) ) ) <EOL> self . req_server = SaltMessageServer ( self . handle_message , io_loop = self . io_loop ) <EOL> self . req_server . add_socket ( self . _socket ) <EOL> self . _socket . listen ( self . backlog ) <EOL> salt . transport . mixins . auth . AESReqServerMixin . post_fork ( self , payload_handler , io_loop ) <EOL> @ tornado . gen . coroutine <EOL> def handle_message ( self , stream , header , payload ) : <EOL> '''<STR_LIT>''' <EOL> try : <EOL> payload = self . _decode_payload ( payload ) <EOL> except Exception : <EOL> stream . write ( salt . transport . frame . frame_msg ( '<STR_LIT>' , header = header ) ) <EOL> raise tornado . gen . Return ( ) <EOL> if not isinstance ( payload , dict ) or not isinstance ( payload . get ( '<STR_LIT>' ) , dict ) : <EOL> yield stream . write ( salt . transport . frame . frame_msg ( <EOL> '<STR_LIT>' , header = header ) ) <EOL> raise tornado . gen . Return ( ) <EOL> if payload [ '<STR_LIT>' ] == '<STR_LIT>' and payload . get ( '<STR_LIT>' , { } ) . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> yield stream . write ( salt . transport . frame . frame_msg ( <EOL> self . _auth ( payload [ '<STR_LIT>' ] ) , header = header ) ) <EOL> raise tornado . gen . Return ( ) <EOL> try : <EOL> ret , req_opts = yield self . payload_handler ( payload ) <EOL> except Exception as e : <EOL> stream . write ( '<STR_LIT>' ) <EOL> log . error ( '<STR_LIT>' , exc_info = True ) <EOL> stream . close ( ) <EOL> raise tornado . gen . Return ( ) <EOL> req_fun = req_opts . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if req_fun == '<STR_LIT>' : <EOL> stream . write ( salt . transport . frame . frame_msg ( ret , header = header ) ) <EOL> elif req_fun == '<STR_LIT>' : <EOL> stream . write ( salt . transport . frame . frame_msg ( self . crypticle . dumps ( ret ) , header = header ) ) <EOL> elif req_fun == '<STR_LIT>' : <EOL> stream . write ( salt . transport . frame . frame_msg ( self . _encrypt_private ( ret , <EOL> req_opts [ '<STR_LIT:key>' ] , <EOL> req_opts [ '<STR_LIT>' ] , <EOL> ) , header = header ) ) <EOL> else : <EOL> log . error ( '<STR_LIT>' . format ( req_fun ) ) <EOL> stream . write ( '<STR_LIT>' ) <EOL> stream . close ( ) <EOL> raise tornado . gen . Return ( ) <EOL> class SaltMessageServer ( tornado . tcpserver . TCPServer , object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , message_handler , * args , ** kwargs ) : <EOL> super ( SaltMessageServer , self ) . __init__ ( * args , ** kwargs ) <EOL> self . clients = [ ] <EOL> self . message_handler = message_handler <EOL> @ tornado . gen . coroutine <EOL> def handle_stream ( self , stream , address ) : <EOL> '''<STR_LIT>''' <EOL> log . trace ( '<STR_LIT>' . format ( address ) ) <EOL> self . clients . append ( ( stream , address ) ) <EOL> unpacker = msgpack . Unpacker ( ) <EOL> try : <EOL> while True : <EOL> wire_bytes = yield stream . read_bytes ( <NUM_LIT> , partial = True ) <EOL> unpacker . feed ( wire_bytes ) <EOL> for framed_msg in unpacker : <EOL> if six . PY3 : <EOL> framed_msg = salt . transport . frame . decode_embedded_strs ( <EOL> framed_msg <EOL> ) <EOL> header = framed_msg [ '<STR_LIT>' ] <EOL> self . io_loop . spawn_callback ( self . message_handler , stream , header , framed_msg [ '<STR_LIT:body>' ] ) <EOL> except tornado . iostream . StreamClosedError : <EOL> log . trace ( '<STR_LIT>' . format ( address ) ) <EOL> self . clients . remove ( ( stream , address ) ) <EOL> except Exception as e : <EOL> log . trace ( '<STR_LIT>' . format ( e ) ) <EOL> self . clients . remove ( ( stream , address ) ) <EOL> stream . close ( ) <EOL> def shutdown ( self ) : <EOL> '''<STR_LIT>''' <EOL> for item in self . clients : <EOL> client , address = item <EOL> client . close ( ) <EOL> self . clients . remove ( item ) <EOL> if USE_LOAD_BALANCER : <EOL> class LoadBalancerWorker ( SaltMessageServer ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , socket_queue , message_handler , * args , ** kwargs ) : <EOL> super ( LoadBalancerWorker , self ) . __init__ ( <EOL> message_handler , * args , ** kwargs ) <EOL> self . socket_queue = socket_queue <EOL> t = threading . Thread ( target = self . socket_queue_thread ) <EOL> t . start ( ) <EOL> def socket_queue_thread ( self ) : <EOL> try : <EOL> while True : <EOL> client_socket , address = self . socket_queue . get ( True , None ) <EOL> self . io_loop . spawn_callback ( <EOL> self . _handle_connection , client_socket , address ) <EOL> except ( KeyboardInterrupt , SystemExit ) : <EOL> pass <EOL> class TCPClientKeepAlive ( tornado . tcpclient . TCPClient ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , opts , resolver = None , io_loop = None ) : <EOL> self . opts = opts <EOL> super ( TCPClientKeepAlive , self ) . __init__ ( <EOL> resolver = resolver , io_loop = io_loop ) <EOL> def _create_stream ( self , max_buffer_size , af , addr ) : <EOL> sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> _set_tcp_keepalive ( sock , self . opts ) <EOL> stream = tornado . iostream . IOStream ( <EOL> sock , <EOL> io_loop = self . io_loop , <EOL> max_buffer_size = max_buffer_size ) <EOL> return stream . connect ( addr ) <EOL> class SaltMessageClient ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , opts , host , port , io_loop = None , resolver = None , <EOL> connect_callback = None , disconnect_callback = None ) : <EOL> self . host = host <EOL> self . port = port <EOL> self . connect_callback = connect_callback <EOL> self . disconnect_callback = disconnect_callback <EOL> self . io_loop = io_loop or tornado . ioloop . IOLoop . current ( ) <EOL> self . _tcp_client = TCPClientKeepAlive ( <EOL> opts , io_loop = self . io_loop , resolver = resolver ) <EOL> self . _mid = <NUM_LIT:1> <EOL> self . _max_messages = int ( ( <NUM_LIT:1> << <NUM_LIT> ) - <NUM_LIT:2> ) <EOL> self . send_queue = [ ] <EOL> self . send_future_map = { } <EOL> self . send_timeout_map = { } <EOL> self . _read_until_future = None <EOL> self . _on_recv = None <EOL> self . _closing = False <EOL> self . _connecting_future = self . connect ( ) <EOL> self . io_loop . spawn_callback ( self . _stream_return ) <EOL> def close ( self ) : <EOL> if self . _closing : <EOL> return <EOL> self . _closing = True <EOL> if hasattr ( self , '<STR_LIT>' ) and not self . _stream . closed ( ) : <EOL> self . _stream . close ( ) <EOL> if self . _read_until_future is not None : <EOL> self . _read_until_future . exc_info ( ) <EOL> self . _tcp_client . close ( ) <EOL> self . connect_callback = None <EOL> self . disconnect_callback = None <EOL> def __del__ ( self ) : <EOL> self . close ( ) <EOL> def connect ( self ) : <EOL> '''<STR_LIT>''' <EOL> if hasattr ( self , '<STR_LIT>' ) and not self . _connecting_future . done ( ) : <EOL> future = self . _connecting_future <EOL> else : <EOL> future = tornado . concurrent . Future ( ) <EOL> self . _connecting_future = future <EOL> self . io_loop . add_callback ( self . _connect ) <EOL> if self . connect_callback is not None : <EOL> def handle_future ( future ) : <EOL> response = future . result ( ) <EOL> self . io_loop . add_callback ( self . connect_callback , response ) <EOL> future . add_done_callback ( handle_future ) <EOL> return future <EOL> @ tornado . gen . coroutine <EOL> def _connect ( self ) : <EOL> '''<STR_LIT>''' <EOL> while True : <EOL> if self . _closing : <EOL> break <EOL> try : <EOL> self . _stream = yield self . _tcp_client . connect ( self . host , self . port ) <EOL> self . _connecting_future . set_result ( True ) <EOL> break <EOL> except Exception as e : <EOL> yield tornado . gen . sleep ( <NUM_LIT:1> ) <EOL> @ tornado . gen . coroutine <EOL> def _stream_return ( self ) : <EOL> while not self . _closing and ( <EOL> not self . _connecting_future . done ( ) or <EOL> self . _connecting_future . result ( ) is not True ) : <EOL> yield self . _connecting_future <EOL> unpacker = msgpack . Unpacker ( ) <EOL> while not self . _closing : <EOL> try : <EOL> self . _read_until_future = self . _stream . read_bytes ( <NUM_LIT> , partial = True ) <EOL> wire_bytes = yield self . _read_until_future <EOL> unpacker . feed ( wire_bytes ) <EOL> for framed_msg in unpacker : <EOL> if six . PY3 : <EOL> framed_msg = salt . transport . frame . decode_embedded_strs ( <EOL> framed_msg <EOL> ) <EOL> header = framed_msg [ '<STR_LIT>' ] <EOL> body = framed_msg [ '<STR_LIT:body>' ] <EOL> message_id = header . get ( '<STR_LIT>' ) <EOL> if message_id in self . send_future_map : <EOL> self . send_future_map . pop ( message_id ) . set_result ( body ) <EOL> self . remove_message_timeout ( message_id ) <EOL> else : <EOL> if self . _on_recv is not None : <EOL> self . io_loop . spawn_callback ( self . _on_recv , header , body ) <EOL> else : <EOL> log . error ( '<STR_LIT>' . format ( message_id ) ) <EOL> except tornado . iostream . StreamClosedError as e : <EOL> log . debug ( '<STR_LIT>' . format ( self . host , self . port ) ) <EOL> for future in six . itervalues ( self . send_future_map ) : <EOL> future . set_exception ( e ) <EOL> self . send_future_map = { } <EOL> if self . _closing : <EOL> return <EOL> if self . disconnect_callback : <EOL> self . disconnect_callback ( ) <EOL> if self . _connecting_future . done ( ) : <EOL> self . _connecting_future = self . connect ( ) <EOL> yield self . _connecting_future <EOL> except Exception as e : <EOL> log . error ( '<STR_LIT>' , exc_info = True ) <EOL> for future in six . itervalues ( self . send_future_map ) : <EOL> future . set_exception ( e ) <EOL> self . send_future_map = { } <EOL> if self . _closing : <EOL> return <EOL> if self . disconnect_callback : <EOL> self . disconnect_callback ( ) <EOL> if self . _connecting_future . done ( ) : <EOL> self . _connecting_future = self . connect ( ) <EOL> yield self . _connecting_future <EOL> @ tornado . gen . coroutine <EOL> def _stream_send ( self ) : <EOL> while not self . _connecting_future . done ( ) or self . _connecting_future . result ( ) is not True : <EOL> yield self . _connecting_future <EOL> while len ( self . send_queue ) > <NUM_LIT:0> : <EOL> message_id , item = self . send_queue [ <NUM_LIT:0> ] <EOL> try : <EOL> yield self . _stream . write ( item ) <EOL> del self . send_queue [ <NUM_LIT:0> ] <EOL> except tornado . iostream . StreamClosedError as e : <EOL> if message_id in self . send_future_map : <EOL> self . send_future_map . pop ( message_id ) . set_exception ( e ) <EOL> self . remove_message_timeout ( message_id ) <EOL> del self . send_queue [ <NUM_LIT:0> ] <EOL> if self . _closing : <EOL> return <EOL> if self . disconnect_callback : <EOL> self . disconnect_callback ( ) <EOL> if self . _connecting_future . done ( ) : <EOL> self . _connecting_future = self . connect ( ) <EOL> yield self . _connecting_future <EOL> def _message_id ( self ) : <EOL> wrap = False <EOL> while self . _mid in self . send_future_map : <EOL> if self . _mid >= self . _max_messages : <EOL> if wrap : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> self . _mid = <NUM_LIT:1> <EOL> wrap = True <EOL> else : <EOL> self . _mid += <NUM_LIT:1> <EOL> return self . _mid <EOL> def on_recv ( self , callback ) : <EOL> '''<STR_LIT>''' <EOL> if callback is None : <EOL> self . _on_recv = callback <EOL> else : <EOL> def wrap_recv ( header , body ) : <EOL> callback ( body ) <EOL> self . _on_recv = wrap_recv <EOL> def remove_message_timeout ( self , message_id ) : <EOL> if message_id not in self . send_timeout_map : <EOL> return <EOL> timeout = self . send_timeout_map . pop ( message_id ) <EOL> self . io_loop . remove_timeout ( timeout ) <EOL> def timeout_message ( self , message_id ) : <EOL> if message_id in self . send_timeout_map : <EOL> del self . send_timeout_map [ message_id ] <EOL> if message_id in self . send_future_map : <EOL> self . send_future_map . pop ( message_id ) . set_exception ( <EOL> SaltReqTimeoutError ( '<STR_LIT>' ) <EOL> ) <EOL> def send ( self , msg , timeout = None , callback = None ) : <EOL> '''<STR_LIT>''' <EOL> message_id = self . _message_id ( ) <EOL> header = { '<STR_LIT>' : message_id } <EOL> future = tornado . concurrent . Future ( ) <EOL> if callback is not None : <EOL> def handle_future ( future ) : <EOL> response = future . result ( ) <EOL> self . io_loop . add_callback ( callback , response ) <EOL> future . add_done_callback ( handle_future ) <EOL> self . send_future_map [ message_id ] = future <EOL> if timeout is not None : <EOL> send_timeout = self . io_loop . call_later ( timeout , self . timeout_message , message_id ) <EOL> self . send_timeout_map [ message_id ] = send_timeout <EOL> if len ( self . send_queue ) == <NUM_LIT:0> : <EOL> self . io_loop . spawn_callback ( self . _stream_send ) <EOL> self . send_queue . append ( ( message_id , salt . transport . frame . frame_msg ( msg , header = header ) ) ) <EOL> return future <EOL> class Subscriber ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , stream , address ) : <EOL> self . stream = stream <EOL> self . address = address <EOL> self . _closing = False <EOL> self . _read_until_future = None <EOL> self . id_ = None <EOL> def close ( self ) : <EOL> if self . _closing : <EOL> return <EOL> self . _closing = True <EOL> if not self . stream . closed ( ) : <EOL> self . stream . close ( ) <EOL> if self . _read_until_future is not None : <EOL> self . _read_until_future . exc_info ( ) <EOL> def __del__ ( self ) : <EOL> self . close ( ) <EOL> class PubServer ( tornado . tcpserver . TCPServer , object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , opts , io_loop = None ) : <EOL> super ( PubServer , self ) . __init__ ( io_loop = io_loop ) <EOL> self . opts = opts <EOL> self . _closing = False <EOL> self . clients = set ( ) <EOL> self . aes_funcs = salt . master . AESFuncs ( self . opts ) <EOL> self . present = { } <EOL> self . presence_events = False <EOL> if self . opts . get ( '<STR_LIT>' , False ) : <EOL> tcp_only = True <EOL> for transport , _ in iter_transport_opts ( self . opts ) : <EOL> if transport != '<STR_LIT>' : <EOL> tcp_only = False <EOL> if tcp_only : <EOL> self . presence_events = True <EOL> if self . presence_events : <EOL> self . event = salt . utils . event . get_event ( <EOL> '<STR_LIT>' , <EOL> opts = self . opts , <EOL> listen = False <EOL> ) <EOL> def close ( self ) : <EOL> if self . _closing : <EOL> return <EOL> self . _closing = True <EOL> def __del__ ( self ) : <EOL> self . close ( ) <EOL> def _add_client_present ( self , client ) : <EOL> id_ = client . id_ <EOL> if id_ in self . present : <EOL> clients = self . present [ id_ ] <EOL> clients . add ( client ) <EOL> else : <EOL> self . present [ id_ ] = set ( [ client ] ) <EOL> if self . presence_events : <EOL> data = { '<STR_LIT>' : [ id_ ] , <EOL> '<STR_LIT>' : [ ] } <EOL> self . event . fire_event ( <EOL> data , <EOL> salt . utils . event . tagify ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ) <EOL> data = { '<STR_LIT>' : list ( self . present . keys ( ) ) } <EOL> self . event . fire_event ( <EOL> data , <EOL> salt . utils . event . tagify ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ) <EOL> def _remove_client_present ( self , client ) : <EOL> id_ = client . id_ <EOL> if id_ is None or id_ not in self . present : <EOL> return <EOL> clients = self . present [ id_ ] <EOL> if client not in clients : <EOL> return <EOL> clients . remove ( client ) <EOL> if len ( clients ) == <NUM_LIT:0> : <EOL> del self . present [ id_ ] <EOL> if self . presence_events : <EOL> data = { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : [ id_ ] } <EOL> self . event . fire_event ( <EOL> data , <EOL> salt . utils . event . tagify ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ) <EOL> data = { '<STR_LIT>' : list ( self . present . keys ( ) ) } <EOL> self . event . fire_event ( <EOL> data , <EOL> salt . utils . event . tagify ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ) <EOL> @ tornado . gen . coroutine <EOL> def _stream_read ( self , client ) : <EOL> unpacker = msgpack . Unpacker ( ) <EOL> while not self . _closing : <EOL> try : <EOL> client . _read_until_future = client . stream . read_bytes ( <NUM_LIT> , partial = True ) <EOL> wire_bytes = yield client . _read_until_future <EOL> unpacker . feed ( wire_bytes ) <EOL> for framed_msg in unpacker : <EOL> if six . PY3 : <EOL> framed_msg = salt . transport . frame . decode_embedded_strs ( <EOL> framed_msg <EOL> ) <EOL> body = framed_msg [ '<STR_LIT:body>' ] <EOL> if body [ '<STR_LIT>' ] != '<STR_LIT>' : <EOL> continue <EOL> crypticle = salt . crypt . Crypticle ( self . opts , salt . master . SMaster . secrets [ '<STR_LIT>' ] [ '<STR_LIT>' ] . value ) <EOL> load = crypticle . loads ( body [ '<STR_LIT>' ] ) <EOL> if six . PY3 : <EOL> load = salt . transport . frame . decode_embedded_strs ( load ) <EOL> if not self . aes_funcs . verify_minion ( load [ '<STR_LIT:id>' ] , load [ '<STR_LIT>' ] ) : <EOL> continue <EOL> client . id_ = load [ '<STR_LIT:id>' ] <EOL> self . _add_client_present ( client ) <EOL> except tornado . iostream . StreamClosedError as e : <EOL> log . debug ( '<STR_LIT>' . format ( client . address ) ) <EOL> client . close ( ) <EOL> self . _remove_client_present ( client ) <EOL> self . clients . discard ( client ) <EOL> break <EOL> except Exception as e : <EOL> log . error ( '<STR_LIT>' , exc_info = True ) <EOL> continue <EOL> def handle_stream ( self , stream , address ) : <EOL> log . trace ( '<STR_LIT>' . format ( address ) ) <EOL> client = Subscriber ( stream , address ) <EOL> self . clients . add ( client ) <EOL> self . io_loop . spawn_callback ( self . _stream_read , client ) <EOL> @ tornado . gen . coroutine <EOL> def publish_payload ( self , package , _ ) : <EOL> log . debug ( '<STR_LIT>' . format ( package ) ) <EOL> payload = salt . transport . frame . frame_msg ( package [ '<STR_LIT>' ] ) <EOL> to_remove = [ ] <EOL> if '<STR_LIT>' in package : <EOL> topic_lst = package [ '<STR_LIT>' ] <EOL> for topic in topic_lst : <EOL> if topic in self . present : <EOL> for client in self . present [ topic ] : <EOL> try : <EOL> f = client . stream . write ( payload ) <EOL> self . io_loop . add_future ( f , lambda f : True ) <EOL> except tornado . iostream . StreamClosedError : <EOL> to_remove . append ( client ) <EOL> else : <EOL> log . debug ( '<STR_LIT>' . format ( topic ) ) <EOL> else : <EOL> for client in self . clients : <EOL> try : <EOL> f = client . stream . write ( payload ) <EOL> self . io_loop . add_future ( f , lambda f : True ) <EOL> except tornado . iostream . StreamClosedError : <EOL> to_remove . append ( client ) <EOL> for client in to_remove : <EOL> log . debug ( '<STR_LIT>' . format ( client . address ) ) <EOL> client . close ( ) <EOL> self . _remove_client_present ( client ) <EOL> self . clients . discard ( client ) <EOL> log . trace ( '<STR_LIT>' ) <EOL> class TCPPubServerChannel ( salt . transport . server . PubServerChannel ) : <EOL> backlog = <NUM_LIT> <EOL> def __init__ ( self , opts ) : <EOL> self . opts = opts <EOL> self . serial = salt . payload . Serial ( self . opts ) <EOL> self . io_loop = None <EOL> def __setstate__ ( self , state ) : <EOL> salt . master . SMaster . secrets = state [ '<STR_LIT>' ] <EOL> self . __init__ ( state [ '<STR_LIT>' ] ) <EOL> def __getstate__ ( self ) : <EOL> return { '<STR_LIT>' : self . opts , <EOL> '<STR_LIT>' : salt . master . SMaster . secrets } <EOL> def _publish_daemon ( self , log_queue = None ) : <EOL> '''<STR_LIT>''' <EOL> salt . utils . appendproctitle ( self . __class__ . __name__ ) <EOL> if log_queue is not None : <EOL> salt . log . setup . set_multiprocessing_logging_queue ( log_queue ) <EOL> salt . log . setup . setup_multiprocessing_logging ( log_queue ) <EOL> if self . io_loop is None : <EOL> self . io_loop = tornado . ioloop . IOLoop . current ( ) <EOL> pub_server = PubServer ( self . opts , io_loop = self . io_loop ) <EOL> sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> sock . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> _set_tcp_keepalive ( sock , self . opts ) <EOL> sock . setblocking ( <NUM_LIT:0> ) <EOL> sock . bind ( ( self . opts [ '<STR_LIT>' ] , int ( self . opts [ '<STR_LIT>' ] ) ) ) <EOL> sock . listen ( self . backlog ) <EOL> pub_server . add_socket ( sock ) <EOL> if self . opts . get ( '<STR_LIT>' , '<STR_LIT>' ) == '<STR_LIT>' : <EOL> pull_uri = int ( self . opts . get ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> else : <EOL> pull_uri = os . path . join ( self . opts [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> pull_sock = salt . transport . ipc . IPCMessageServer ( <EOL> pull_uri , <EOL> io_loop = self . io_loop , <EOL> payload_handler = pub_server . publish_payload , <EOL> ) <EOL> log . info ( '<STR_LIT>' . format ( pull_uri ) ) <EOL> old_umask = os . umask ( <NUM_LIT> ) <EOL> try : <EOL> pull_sock . start ( ) <EOL> finally : <EOL> os . umask ( old_umask ) <EOL> try : <EOL> self . io_loop . start ( ) <EOL> except ( KeyboardInterrupt , SystemExit ) : <EOL> salt . log . setup . shutdown_multiprocessing_logging ( ) <EOL> def pre_fork ( self , process_manager ) : <EOL> '''<STR_LIT>''' <EOL> kwargs = { } <EOL> if salt . utils . is_windows ( ) : <EOL> kwargs [ '<STR_LIT>' ] = ( <EOL> salt . log . setup . get_multiprocessing_logging_queue ( ) <EOL> ) <EOL> process_manager . add_process ( self . _publish_daemon , kwargs = kwargs ) <EOL> def publish ( self , load ) : <EOL> '''<STR_LIT>''' <EOL> payload = { '<STR_LIT>' : '<STR_LIT>' } <EOL> crypticle = salt . crypt . Crypticle ( self . opts , salt . master . SMaster . secrets [ '<STR_LIT>' ] [ '<STR_LIT>' ] . value ) <EOL> payload [ '<STR_LIT>' ] = crypticle . dumps ( load ) <EOL> if self . opts [ '<STR_LIT>' ] : <EOL> master_pem_path = os . path . join ( self . opts [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> log . debug ( "<STR_LIT>" ) <EOL> payload [ '<STR_LIT>' ] = salt . crypt . sign_message ( master_pem_path , payload [ '<STR_LIT>' ] ) <EOL> if self . opts . get ( '<STR_LIT>' , '<STR_LIT>' ) == '<STR_LIT>' : <EOL> pull_uri = int ( self . opts . get ( '<STR_LIT>' , <NUM_LIT> ) ) <EOL> else : <EOL> pull_uri = os . path . join ( self . opts [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> pub_sock = salt . utils . async . SyncWrapper ( <EOL> salt . transport . ipc . IPCMessageClient , <EOL> ( pull_uri , ) <EOL> ) <EOL> pub_sock . connect ( ) <EOL> int_payload = { '<STR_LIT>' : self . serial . dumps ( payload ) } <EOL> if load [ '<STR_LIT>' ] == '<STR_LIT:list>' : <EOL> int_payload [ '<STR_LIT>' ] = load [ '<STR_LIT>' ] <EOL> pub_sock . send ( int_payload ) </s>
