U d*Y@sdZddlZddlZddlZddlZddlZddlmZddlmZddl m Z m Z ddl m Z ddlmZmZdd lmZejejejejhZd ZeZd Zd Zed ZddddddZefddZ GdddZ!Gddde!Z"Gddde!Z#dddZ$dS)zTransport implementation.N)contextmanager)SSLError)packunpack)UnexpectedFrame)KNOWN_TCP_OPTSSOL_TCP) set_cloexeci(isAMQP z\[([\.0-9a-f:]+)\](?::(\d+))?i< ) TCP_NODELAYTCP_USER_TIMEOUT TCP_KEEPIDLE TCP_KEEPINTVL TCP_KEEPCNTcCs^|}t|}|r6|d}|drVt|d}n d|krV|dd\}}t|}||fS)z1Convert hostname:port string to host, port tuple.r:) IPV6_LITERALmatchgroupintrsplit)hostdefaultportmr2/tmp/pip-unpacked-wheel-sfh45kzb/amqp/transport.py to_host_port(s   r c@seZdZdZd$ddZdZddZd d Zed d Z d dZ ddZ ddZ ddZ d%ddZddZddZddZddZefd d!Zd"d#ZdS)&_AbstractTransportaCommon superclass for TCP and SSL transports. PARAMETERS: host: str Broker address in format ``HOSTNAME:PORT``. connect_timeout: int Timeout of creating new connection. read_timeout: int sets ``SO_RCVTIMEO`` parameter of socket. write_timeout: int sets ``SO_SNDTIMEO`` parameter of socket. socket_settings: dict dictionary containing `optname` and ``optval`` passed to ``setsockopt(2)``. raise_on_initial_eintr: bool when True, ``socket.timeout`` is raised when exception is received during first read. See ``_read()`` for details. NTcKsDd|_d|_||_t|_t|\|_|_||_||_ ||_ ||_ dSNF) connectedsockraise_on_initial_eintr EMPTY_BUFFER _read_bufferr rrconnect_timeout read_timeout write_timeoutsocket_settings)selfrr(r)r*r+r%kwargsrrr__init__Wsz_AbstractTransport.__init__) connectionr$r%r'rrr(r)r*r+__dict__ __weakref__c Cs|jrt|jdd|jd}|jdd|jd}dt|jd|d|dt|dd Sdt|jd t|dd SdS) Nrrr z at z#x>z: (disconnected) at )r$ getsockname getpeernametype__name__id)r,srcdstrrr__repr__ts ""*z_AbstractTransport.__repr__c Cszz>|jrWdS||j|j|j||j|j|jd|_Wn6t t fk rt|j rn|jsn|j d|_ YnXdS)NT) r#_connectrrr( _init_socketr+r)r*OSErrorrr$closer,rrrconnect|s   z_AbstractTransport.connectc cs|dkr|jVn|j}|}||kr2||zz |jVWntk r}z0dt|krhtndt|kr|tW5d}~XYn8tk r}z|jtj krtW5d}~XYnXW5||kr||XdS)N timed outzThe operation did not complete) r$ gettimeout settimeoutrstrsockettimeoutr>errno EWOULDBLOCK)r,rGr$prevexcrrrhaving_timeouts(       z!_AbstractTransport.having_timeoutc Cst||tjtjt}t|D]\}}|\}}} } } zRt||| |_zt|jdWntk rjYnX|j ||j | Wn>tj k r|jr|j d|_|dt |krYqXqqdS)NTr)rF getaddrinfo AF_UNSPEC SOCK_STREAMr enumerater$r NotImplementedErrorrDrAerrorr?len) r,rrrGentriesiresafsocktypeproto canonnamesarrrr<s0  z_AbstractTransport._connectc Cs|jd|jtjtjd||tj|ftj|ffD]B\}}|dk r>t |}t ||d}|jtj|t d||q>| | t dS)Nri@BZll)r$rD setsockoptrF SOL_SOCKET SO_KEEPALIVE_set_socket_options SO_SNDTIMEO SO_RCVTIMEOrr_setup_transport_writeAMQP_PROTOCOL_HEADER)r,r+r)r*rGintervalsecZusecrrrr=s     z_AbstractTransport._init_socketc Csi}tD]}d}|dkrDzddlm}WqXtk r@d}YqXXntt|rXtt|}|r|tkrrt|||<qtt|r|ttt|||<q|S)Nrr)r) rrFr ImportErrorhasattrgetattrDEFAULT_SOCKET_SETTINGS getsockoptr )r,r$tcp_optsoptenumrrr_get_tcp_socket_defaultss&     z+_AbstractTransport._get_tcp_socket_defaultscCs@||j}|r|||D]\}}|jt||q"dSN)rpr$updateitemsr\r )r,r+rmrnvalrrrr_s   z&_AbstractTransport._set_socket_optionsFcCs tddS)z#Read exactly n bytes from the peer.Must be overridden in subclassNrQ)r,ninitialrrr_readsz_AbstractTransport._readcCsdS)z.Do any additional initialization of the class.Nrr@rrrrbsz#_AbstractTransport._setup_transportcCsdS)z8Do any preliminary work in shutting down the connection.Nrr@rrr_shutdown_transportsz&_AbstractTransport._shutdown_transportcCs tddS)z&Completely write a string to the peer.ruNrv)r,srrrrcsz_AbstractTransport._writecCs|jdk r~z |Wntk r*YnXz|jtjWntk rRYnXz|jWntk rvYnXd|_d|_dSr")r$rzr>shutdownrF SHUT_RDWRr?r#r@rrrr?s  z_AbstractTransport.closec Cs|j}t}z|dd}||7}|d|\}}}|tkr|t}z||t} Wn&tjttfk rt||7}YnXd|| g} n||} || 7}t|d} Wntjk r||j |_ Ynttfk rl} zt | tj rt j dkr| jtjkr||j |_ tt | trHdt| krH||j |_ t| jtkrZd|_W5d } ~ XYnX| d kr||| fStd | d d d S)aParse AMQP frame. Frame has following format:: 0 1 3 7 size+7 size+8 +------+---------+---------+ +-------------+ +-----------+ | type | channel | size | | payload | | frame-end | +------+---------+---------+ +-------------+ +-----------+ octet short long 'size' octets octet Tz>BHIrntrBFNzReceived frame_end z#04xz while expecting 0xce)ryr&SIGNED_INT_MAXrFrGr>rjoinordr' isinstancerRosnamerHrIrE_UNAVAILr#r) r,rreadZread_frame_bufferZ frame_headerZ frame_typeZchannelsizeZpart1Zpart2payloadZ frame_endrKrrr read_framesP           z_AbstractTransport.read_framec Cs^z||WnJtjk r&Yn4tk rX}z|jtkrFd|_W5d}~XYnXdSr")rcrFrGr>rHrr#)r,r{rKrrrwriteYs z_AbstractTransport.write)NNNNT)F)r7 __module__ __qualname____doc__r. __slots__r;rArrLr<r=rpr_ryrbrzrcr?rrrrrrrr!7s.    Br!c sveZdZdZdfdd ZdZddZddd Zdd d ZdddZ ddZ d e j e j e jffddZddZZS) SSLTransportaTransport that works over SSL. PARAMETERS: host: str Broker address in format ``HOSTNAME:PORT``. connect_timeout: int Timeout of creating new connection. ssl: bool|dict parameters of TLS subsystem. - when ``ssl`` is not dictionary, defaults of TLS are used - otherwise: - if ``ssl`` dictionary contains ``context`` key, :attr:`~SSLTransport._wrap_context` is used for wrapping socket. ``context`` is a dictionary passed to :attr:`~SSLTransport._wrap_context` as context parameter. All others items from ``ssl`` argument are passed as ``sslopts``. - if ``ssl`` dictionary does not contain ``context`` key, :attr:`~SSLTransport._wrap_socket_sni` is used for wrapping socket. All items in ``ssl`` argument are passed to :attr:`~SSLTransport._wrap_socket_sni` as parameters. kwargs: additional arguments of :class:`~amqp.transport._AbstractTransport` class Nc s6t|tr|ni|_t|_tj|fd|i|dS)Nr()rdictssloptsr&r'superr.)r,rr(sslr- __class__rrr.szSSLTransport.__init__)rcCs,|j|jf|j|_|j|jj|_dS)z!Wrap the socket in an SSL object.N) _wrap_socketr$r do_handshaker _quick_recvr@rrrrbs zSSLTransport._setup_transportcKs"|r|j||f|S|j|f|Srq) _wrap_context_wrap_socket_sni)r,r$contextrrrrrszSSLTransport._wrap_socketcKs tjf|}||_|j|f|S)uWrap socket without SNI headers. PARAMETERS: sock: socket.socket Socket to be wrapped. sslopts: dict Parameters of :attr:`ssl.SSLContext.wrap_socket`. check_hostname Whether to match the peer cert’s hostname. See :attr:`ssl.SSLContext.check_hostname` for details. ctx_options Parameters of :attr:`ssl.create_default_context`. )rcreate_default_contextcheck_hostname wrap_socket)r,r$rrZ ctx_optionsctxrrrrs zSSLTransport._wrap_contextFTc Cs||||| d} | dkr(|r"tjntj} t| } |dk rF| |||dk rX| || dk rj| | ztjox| dk | _Wnt k rYnX|dk r|| _ |dkr| j tj kr|rtj j ntj j}| || jf| }|S)uSocket wrap with SNI headers. stdlib :attr:`ssl.SSLContext.wrap_socket` method augmented with support for setting the server_hostname field required for SNI hostname header. PARAMETERS: sock: socket.socket Socket to be wrapped. keyfile: str Path to the private key certfile: str Path to the certificate server_side: bool Identifies whether server-side or client-side behavior is desired from this socket. See :attr:`~ssl.SSLContext.wrap_socket` for details. cert_reqs: ssl.VerifyMode When set to other than :attr:`ssl.CERT_NONE`, peers certificate is checked. Possible values are :attr:`ssl.CERT_NONE`, :attr:`ssl.CERT_OPTIONAL` and :attr:`ssl.CERT_REQUIRED`. ca_certs: str Path to “certification authority” (CA) certificates used to validate other peers’ certificates when ``cert_reqs`` is other than :attr:`ssl.CERT_NONE`. do_handshake_on_connect: bool Specifies whether to do the SSL handshake automatically. See :attr:`~ssl.SSLContext.wrap_socket` for details. suppress_ragged_eofs (bool): See :attr:`~ssl.SSLContext.wrap_socket` for details. server_hostname: str Specifies the hostname of the service which we are connecting to. See :attr:`~ssl.SSLContext.wrap_socket` for details. ciphers: str Available ciphers for sockets created with this context. See :attr:`ssl.SSLContext.set_ciphers` ssl_version: Protocol of the SSL Context. The value is one of ``ssl.PROTOCOL_*`` constants. )r$ server_sidedo_handshake_on_connectsuppress_ragged_eofsserver_hostnameN)rPROTOCOL_TLS_SERVERPROTOCOL_TLS_CLIENT SSLContextload_cert_chainload_verify_locations set_ciphersHAS_SNIrAttributeError verify_mode CERT_NONEPurpose CLIENT_AUTH SERVER_AUTHload_default_certsr)r,r$keyfilecertfiler cert_reqsca_certsrrrciphers ssl_versionoptsrpurposerrrrsBD        zSSLTransport._wrap_socket_snicCs|jdk r|j|_dS)z/Unwrap a SSL socket, so we can call shutdown().N)r$unwrapr@rrrrz-s z SSLTransport._shutdown_transportc Cs|j}|j}zt||krz||t|}WnHtk rv}z*|j|krd|r\|jr\tWYqW5d}~XYnX|std||7}qWn||_YnX|d|||d}|_|S)N%Server unexpectedly closed connectionrr'rSr>rHr%rFrG r,rwrxZ_errnosrecvZrbufr{rKresultrrrry2s(   zSSLTransport._readcCsP|jj}|rLz ||}Wntk r0d}YnX|s>td||d}qdS)z+Write a string out to the SSL socket fully.rz Socket closedN)r$r ValueErrorr>)r,r{rrwrrrrcNs  zSSLTransport._write)NN)N)N) NNFNNFTNNN)r7rrrr.rrbrrrrzrHENOENTEAGAINEINTRryrc __classcell__rrrrrds*"   x rc@s.eZdZdZddZdejejffddZdS) TCPTransportz~Transport that deals directly with TCP socket. All parameters are :class:`~amqp.transport._AbstractTransport` class. cCs|jj|_t|_|jj|_dSrq)r$sendallrcr&r'rrr@rrrrbes zTCPTransport._setup_transportFc Cs|j}|j}zt||krz||t|}WnHtk rv}z*|j|krd|r\|jr\tWYqW5d}~XYnX|std||7}qWn||_YnX|d|||d}|_|S)z%Read exactly n bytes from the socket.Nrrrrrrryls(   zTCPTransport._readN) r7rrrrbrHrrryrrrrr_srFcKs"|rtnt}||f||d|S)aCreate transport. Given a few parameters from the Connection constructor, select and create a subclass of :class:`~amqp.transport._AbstractTransport`. PARAMETERS: host: str Broker address in format ``HOSTNAME:PORT``. connect_timeout: int Timeout of creating new connection. ssl: bool|dict If set, :class:`~amqp.transport.SSLTransport` is used and ``ssl`` parameter is passed to it. Otherwise :class:`~amqp.transport.TCPTransport` is used. kwargs: additional arguments of :class:`~amqp.transport._AbstractTransport` class )r(r)rr)rr(rr- transportrrr Transports r)NF)%rrHrrerFr contextlibrrstructrr exceptionsrplatformrr utilsr rrrrIrZ AMQP_PORTbytesr&rrdcompilerrkr r!rrrrrrrs>      /|&