File: //opt/saltstack/salt/lib/python3.10/site-packages/salt/channel/__pycache__/client.cpython-310.pyc
o
�N�gGT � @ s d Z ddlZddlZddlZddlZddlZddlZddlZddl Zddl
Zddl Zddl Zddl
ZddlZddlZddlZddlmZ e�e�ZdZdZG dd� d�ZG dd � d �ZG d
d � d �ZG d d
� d
�ZG dd� d�Z G dd� d�Z G dd� d�Z dS )z�
Encapsulate the different transports available to Salt.
This includes client side transport, for the ReqServer and the Publisher
� N)� SyncWrapper�< � c @ � e Zd ZdZedd� �ZdS )�
ReqChannelz�
Factory class to create a sychronous communication channels to the master's
ReqServer. ReqChannels use transports to connect to the ReqServer.
c K � t tj| f|dd�S �N�io_loop)Z
loop_kwarg)r �AsyncReqChannel�factory��opts�kwargs� r �G/opt/saltstack/salt/lib/python3.10/site-packages/salt/channel/client.pyr % � �zReqChannel.factoryN��__name__�
__module__� __qualname__�__doc__� staticmethodr r r r r r s r c @ r )� PushChannelzM
Factory class to create Sync channel for push side of push/pull IPC
c K r r )r �AsyncPushChannelr r r r r r 4 r zPushChannel.factoryNr r r r r r / � r c @ r )� PullChannelzM
Factory class to create Sync channel for pull side of push/pull IPC
c K r r )r �AsyncPullChannelr r r r r r C r zPullChannel.factoryNr r r r r r > r r c @ s� e Zd ZdZg d�ZdgZedd� �Zee fdd�Z
e dd � �Z e d
d � �Z
d d
� Zejjjjdd� �Zejjjj d$dd��Zdd� Zejjjjd%dd��Zejjjjdd� �Zejjjjdd� �Zejjjjd&d d ��Zd d � Zd d!� Z d"d#� Z dS )'r
z�
Factory class to create a asynchronous communication channels to the
master's ReqServer. ReqChannels connect to the master's ReqServerChannel.
)�!crypted_transfer_decode_dictentry�_crypted_transfer�_uncrypted_transfer�send�connect�closec
K s� d}d|v r |d }nd|� di �� di �v r |d d d }d|vr-d|v r-|d |d< |� d�}|d u r>tjjjj�� }|� dt�}|� dt�}|� d d
�}|d kr]tj j
||d �}nd }tj j ||d �} | || |||d
�S )N�zeromq� transport�pillar�master�
master_urir Zrequest_channel_timeoutZrequest_channel_tries�crypt�aes�clear�r )�tries�timeout)
�get�salt�ext�tornado�ioloop�IOLoop�current�REQUEST_CHANNEL_TIMEOUT�REQUEST_CHANNEL_TRIESr( � AsyncAuthr$ Zrequest_client)
�clsr
r �ttyper r- r, r( �authr$ r r r r ^ s$
zAsyncReqChannel.factoryc K sR t |�| _|| _|| _d | _| jr tj�| jd | jj�| _d| _ || _
|| _ d S )N�pki_dirF) �dictr
r$ r: �master_pubkey_path�os�path�join�mpub�_closingr- r, )�selfr
r$ r: r- r, r r r r �__init__| s
zAsyncReqChannel.__init__c C � | j rdS dS �Nr) r* �r: �rC r r r r( � s zAsyncReqChannel.cryptc C s | j jS �N)r$ r9 rH r r r r9 � s zAsyncReqChannel.ttypec C s8 | j |dd�}| j dkr| jd |d< | jd |d< |S )N� ��enc�load�versionr) �encryption_algorithmZenc_algo�signing_algorithm�sig_algo)r( r
)rC rM �retr r r �
_package_load� s �
z AsyncReqChannel._package_loadc
c sr � d} z | j j||d�V }W n! ty0 } zt�d|� ||kr"� |d7 }W Y d }~qd }~ww tjjj� |��)N� T�r- �Failed to send msg %r)
r$ r � Exception�log�tracer/ r0 r1 �gen�Return)rC rM r, r- �_tryrR �excr r r �_send_with_retry� s$ �� ��z AsyncReqChannel._send_with_retryNc c sN � |d u r| j }|d u r| j}t�� j}||d< | jjs"| j�� V | �| � | jj
� |��||�V }| j� � }d|vrQ| j�� V | �| � | jj
� |��||�V }|�
|d | jd �}tj�| j|�} | �|| �}
| �|
d |
d �s{tj�d��tj�|
d �} | d |d kr�tj�d��| d |kr�tj�d��tjjj�| d ��)
N�nonce�keyrO �data�sigz,Pillar payload signature failed to validate.zKey verification failed.z!Pillar nonce verification failed.r% )r- r, �uuid�uuid4�hexr: �
authenticated� authenticater^ rS � crypticle�dumpsZget_keysZdecryptr
r/ r( Z Crypticle�loads�verify_signature�AuthenticationError�payloadr0 r1 rZ r[ ) rC rM Zdictkeyr- r, r_ rR r` r) ZpcryptZ
signed_msgra r r r r � sF �
�
�� z1AsyncReqChannel.crypted_transfer_decode_dictentryc C s t j�| j��||| jd �S )NrP )r/ r( Z PublicKeyr= Zverifyr
)rC ra rb r r r rk � s �z AsyncReqChannel.verify_signatureFc # s� � t �� j�� rt� t�r�� d< tjjjj � ����fdd��}�j
j s+�j
� � V z|� V }W n tj
jyF �j
� � V |� V }Y nw tjjj�|��)a�
Send a load across the wire, with encryption
In case of authentication errors, try to renegotiate authentication
and retry the method.
Indeed, we can fail too early in case of a master restart during a
minion state execution call
:param dict load: A load to send across the wire
:param int timeout: The number of seconds on a response before failing
r_ c 3 sh � �j j���jj�� ���d�V } | r �jjj| ��d�} �r%�jdkr,tj j �
| �} tj j j
�| ��)NrU )r_ Ztcp)r$ r rS r: rh ri rj r9 r/ �frameZdecode_embedded_strsr0 r1 rZ r[ )ra �rM r_ �rawrC r- r r � _do_transfer s ��z7AsyncReqChannel._crypted_transfer.<locals>._do_transfer)rc rd re �
isinstancer< r/ r0 r1 rZ � coroutiner: rf rg r( rl r[ )rC rM r- rp rq rR r ro r r � s �
�z!AsyncReqChannel._crypted_transferc c s* � | j j| �|�|d�V }tjjj�|��)z�
Send a load across the wire in cleartext
:param dict load: A load to send across the wire
:param int timeout: The number of seconds on a response before failing
rU )r$ r rS r/ r0 r1 rZ r[ )rC rM r- rR r r r r s ��z#AsyncReqChannel._uncrypted_transferc c s � | j �� V d S rI )r$ r! rH r r r r! . s �zAsyncReqChannel.connectc
c s� � |du r| j }|du r| j}d} z%| jdkr't�d|� | j||d�V }nt�d|� | j|||d�V }W n! tyX } zt�d |� ||krJ� |d7 }W Y d}~qd}~ww tj j
j � |��)
a,
Send a request, return a future which will complete when we send the message
:param dict load: A load to send across the wire
:param int tries: The number of times to make before failure
:param int timeout: The number of seconds on a response before failing
NrT Tr* z ReqChannel send clear load=%rrU z ReqChannel send crypt load=%r)r- rp rV )
r- r, r( rX rY r r rW r/ r0 r1 rZ r[ )rC rM r, r- rp r\ rR r] r r r r 2 s. �
��zAsyncReqChannel.sendc C s. | j rdS t�d| jj� d| _ | j�� dS )z�
Since the message_client creates sockets and assigns them to the IOLoop we have to
specifically destroy them, since we aren't the only ones with references to the FDs
NzClosing %s instanceT)rB rX �debug� __class__r r$ r" rH r r r r" R s
zAsyncReqChannel.closec C � | S rI r rH r r r � __enter__] � zAsyncReqChannel.__enter__c G � | � � d S rI �r" �rC �argsr r r �__exit__` � zAsyncReqChannel.__exit__)NNN)F)NNF) r r r r �
async_methods�
close_methods� classmethodr r5 r6 rD �propertyr( r9 rS r/ r0 r1 rZ rs r^ r rk r r r! r r" rw r} r r r r r
M sD �
"
�
�3
.
r
c @ s� e Zd ZdZddgZdgZedd� �Zd"dd �Ze d
d � �Z
e j j
jjd d
� �Zdd� Zd"dd�Zdd� Ze j j
jjdd� �Ze j j
jjdd� �Zdd� Zdd� Ze j j
jjd d � �Zd d � Zd d!� ZdS )#�AsyncPubChannelzQ
Factory class to create subscription channels to the master's Publisher
r! �_decode_payloadr" c K s� d}d|v r |d }nd|� di �� di �v r |d d d }d|vr-d|v r-|d |d< |dkr;d|d< t�d |� |� d
�}|d u rLtjjjj�� }tj j
||d �}tj � ||�}| ||||�S ) Nr# r$ r% r& r' ZdetectTZ detect_modez$Transport is set to detect; using %sr r+ )
r. rX �infor/ r0 r1 r2 r3 r4 r( r7 r$ Zpublish_client)r8 r
r r9 r r: r$ r r r r q s
zAsyncPubChannel.factoryNc C sh || _ || _|| _| j�d�| _|| _d| _d| _tj j
j d| j dd�| _
t j
�| j d | jj�| _d S )Ns saltF�minion)r
�listenr; )r
r r: Z gen_token�tokenr$ rB � _reconnectedr/ �utils�eventZ get_eventr> r? r@ rA r= )rC r
r$ r: r r r r rD � s zAsyncPubChannel.__init__c C rE rF rG rH r r r r( � s zAsyncPubChannel.cryptc
c s� � z0| j js | j �� V t| j�dd��dkr | j�d�}n| j jd }| j�|| j | j
�V W dS t y9 � t yZ } zdt
|�vrOtj�d|� ���W Y d}~dS d}~ww )zX
Return a future which completes when connected to the remote publisher
� publish_porti� z -|RETRY|-z Unable to sign_in to master: N)r: rf rg �intr
r. Zcredsr$ r! �connect_callback�disconnect_callback�KeyboardInterruptrW �strr/ �
exceptions�SaltClientError)rC r� r] r r r r! � s( �
� ����zAsyncPubChannel.connectc C s, | j �� | jdur| j�� d| _dS dS )z#
Close the channel
N)r$ r" r� ZdestroyrH r r r r" � s
�zAsyncPubChannel.closec s: � du r
�j �d�S tjjjj� �fdd��}�j �|�S )zH
When jobs are received pass them (decoded) to callback
Nc 3 s@ � �j �| �V }��|�V }t�d|� |d ur � |� d S d S )NzPubChannel received: %r)r$ Z_decode_messagesr� rX rt )�messagesrm �decoded��callbackrC r r �
wrap_callback� s � �z.AsyncPubChannel.on_recv.<locals>.wrap_callback)r$ �on_recvr/ r0 r1 rZ rs )rC r� r� r r� r r� � s
zAsyncPubChannel.on_recvc C s | j |dd�S )NrJ rK )r( )rC rM r r r rS � s �z AsyncPubChannel._package_loadc
# s � �j d |d�� tjjjj� �fdd��}|s �jjs`d}|�j d ks,�j d dk r`z�j�� V W n+ tj j
yQ } zt � |� |d7 }W Y d}~nd}~ww |�j d ks,�j d dk s,z |� V }tjjj�
|�� tjjy� �j�� V |� V }tjjj�
|��w ) z�
Send the minion id to the master so that the master may better
track the connection state of the minion.
In case of authentication errors, try to renegotiate authentication
and retry the method.
�id)r� �tokc 3 sF � �� �jj�� ��} tjjj| d d�}�j�|�V tj j
j � d��)N)�headerT)
rS r: rh ri r/ r$ rn Z frame_msgr r0 r1 rZ r[ )�msg�package�rM rC r r rq � s
�z-AsyncPubChannel.send_id.<locals>._do_transferr Ztcp_authentication_retriesrT N)r
r/ r0 r1 rZ rs r: rf rg r� r� rX rt r[ r( rl )rC r� Z
force_authrq �countr] rR r r� r �send_id� s4 �
��� �zAsyncPubChannel.send_idc
c s� � | j rd S z�| �| j| j�V d| _| j�d| jd id� | jr�| j�d�dkrCd� | jd t
� � �}t j
j�| jd dgd�}nd � | jd t
� � �}t j
j�| jd dgd
�}| jd d d | j||d �}t�| j��2}z
|j|d
d�V W n t jjy� t�d� Y n ty� tjddd� Y nw W d � n 1 s�w Y W d S W d S d| _W d S ty� } ztjd|dd� W Y d }~d S d }~ww )NTr& Z__master_connectedZ__roleZsyndiczSyndic {} started at {}r� �startzMinion {} started at {}r� Z
_minion_event)r� �cmdZpretagr� ra �tagr rU zEfire_master failed: master could not be contacted. Request timed out.zfire_master failed)�exc_infoz2Caught exception in PubChannel connect callback %r)rB r� r� r� � connectedr� �
fire_eventr
r. �format�time�asctimer/ r� Ztagifyr
r r r� ZSaltReqTimeoutErrorrX r� rW �error)rC �resultra r� rM Zchannelr] r r r r� � sV �� ��� ��(� ���z AsyncPubChannel.connect_callbackc C s, | j rd S d| _| j�d| jd id� d S )NFr& Z__master_disconnected)rB r� r� r� r
rH r r r r� . s z#AsyncPubChannel.disconnect_callbackc C s\ | j �d�r*|�dd�stj�d��tjj| j|d |�d�|d d�s,tj�d��d S d S ) NZsign_pub_messagesrb Fz<Message signing is enabled but the payload has no signature.rM rQ )� algorithmz%Message signature failed to validate.)r
r. r/ r( rl rk r= )rC rm r r r �_verify_master_signature4 s ����z(AsyncPubChannel._verify_master_signaturec c s� � t �d|� d}|d dkrY| �|� z
| jj�|d �|d< W n
tjjy. d}Y nw |rYz| j� � V | jj�|d �|d< W n tjjyX t �
d| j d � Y nw tj j
j�|��) NzDecoding payload: %sFrL r) rM TzEPayload decryption failed even after re-authenticating with master %sZ master_ip)rX rY r� r: rh rj r/ r( rl rg r� r
r0 r1 rZ r[ )rC rm Zreauthr r r r� F s* �
� ��z AsyncPubChannel._decode_payloadc C rv rI r rH r r r rw \ rx zAsyncPubChannel.__enter__c G ry rI rz r{ r r r r} _ r~ zAsyncPubChannel.__exit__rI )r r r r r r� r� r rD r� r( r/ r0 r1 rZ rs r! r" r� rS r� r� r� r� r� rw r} r r r r r� d s6 ��
%
/
r� c @ r )r z3
Factory class to create IPC Push channels
c K � ddl }|jjj| fi |��S )z\
If we have additional IPC transports other than UxD and TCP, add them here
r N)�salt.transport.ipcr$ �ipcZIPCMessageClient�r
r r/ r r r r h s zAsyncPushChannel.factoryNr r r r r r c r r c @ r )r z3
Factory class to create IPC pull channels
c K r� )z\
If we have additional IPC transports other than UXD and TCP, add them here
r N)r� r$ r� ZIPCMessageServerr� r r r r y s zAsyncPullChannel.factoryNr r r r r r t r r ) r �loggingr> r� rc Z
salt.cryptr/ Zsalt.exceptionsZsalt.ext.tornado.genZsalt.ext.tornado.ioloopZ salt.payloadZsalt.transport.frameZsalt.utils.eventZsalt.utils.filesZsalt.utils.minionsZsalt.utils.stringutilsZsalt.utils.verifyZsalt.utils.asynchronousr � getLoggerr rX r5 r6 r r r r
r� r r r r r r �<module> s>