tԑu v d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl m
Z
d dl m Z m
Z
mZmZ d dlmZ d dlmZ d dlmZmZmZ d dlmZmZ d dlm Z m Z d d l m Z d d
l m!Z!m"Z"m#Z# d d l$m%Z% ej& e' Z( e Z) e Z* e Z+ G d d
Z, G d de,e Z-dS ) N) Generator)APIErrorAPIErrorTooManyRequests
APITokenError send_message)license)Core)Message MessageList MessageType) MessageSinkexpect)Gen publisher)PersistentMessagesQueue)recurring_checksafe_cancel_taskScope)ServerJSONEncoderc e Zd ZdZ eej dd ZdZ dZ
de j fdZ
dd
Zd Zed Zej deej d d f fd
Z eej dedd fd Z e e d Z e d d Z dede fdZ!de defdZ"d Z#d Z$d Z%ddZ&d S )SendToServerClienta Send messages to server.
* process Reportable messages;
* add them to a pending messages list;
* send all pending messages to server when list is full (contains
_PENDING_MESSAGES_LIMIT items or more);
* send all pending messages on plugin shutdown. IMUNIFYAV_MESSAGES_COUNT_TO_SEND i, 2 loopc ~ K || _ t | _ t j | _ t j | _ t j | _ | |
| _ | | | _
d S N)_loopr _pendingasyncioEvent _try_sendLock_lock_shutting_down create_task_send _sender_task_invoke_send_message_invoke_send_message_task)selfr s S/opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins/client.py create_sinkz SendToServerClient.create_sink7 s
/11
\^^
%moo , ,TZZ\\ : :)-)9)9 % % ' '*
*
&&& returnNc x K | j t j | | j d{V nh# t j $ rV t d| j | j
st | j d{V Y nw xY w| j j
dk rrt d| j j
| j t d| j dS dS )a~
When shutdown begins it signals any in-flight HTTP sends to
abort immediately (via _shutting_down event), then gives 50
seconds to finish the stop() sequence. If stop() isn't done
in 50 seconds it force-cancels the sender task.
Finally, any messages still in the buffer are flushed to
persistent storage so nothing is lost.
Nz5Timeout (%ds) sending messages to server on shutdown.r z&Save %s messages to persistent storagezStored queue %r)r% setr wait_forstop_SHUTDOWN_SEND_TIMEOUT TimeoutErrorloggererrorr( cancelledr r buffer_sizewarningpush_buffer_to_storageqsizer+ s r, shutdownzSendToServerClient.shutdownB sJ
!!! :"499;;0KLL L L L L L L L L# : : : LLG+
$..00
:&t'8999999999 : = $q ( ( NN8
)
M 0 0 2 2 2 NN,dm.A.A.C.C D D D D D
) (s 2A A"B54B5c K t d t | j d{V t d | j 4 d{V t d t | j d{V | j | d{V ddd d{V dS # 1 d{V swxY w Y dS )aq
Stop sending.
1. wait for the lock being available
i.e., while _sender_task finishes the current round
of sending message (if it takes too long, then
the timeout in shutdown() is triggered
2. once the sending round complete (we got the lock),
cancel the next iteration of the _sender_task (it exits)
3. send _pending messages (again, if it takes too long,
the timeout in shutdown() is triggered
and the coroutine is cancelled
That method makes sure that the coroutine
that was started in it has ended.
It excludes a situation when:
-> The result of a coroutine that started
BEFORE shutdown() is started.
-> And the process of sending messages
from _pending is interrupted because of it
z2SendToServer.stop cancel _invoke_send_message_taskNzSendToServer.stop wait lockz4SendToServer.stop lock acquired, cancel _sender_task) r6 infor r* r$ r( r% clear_send_pending_messagesr= s r, r3 zSendToServerClient.stope s 0 HIII t =>>>>>>>>> 1222: 0 0 0 0 0 0 0 0
KKN O O O"4#455 5 5 5 5 5 5 5
% % ' ' '--// / / / / / / / 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0s A(C
C#&C#c , | t j | t j | t j | S r )set_product_namer
LicenseCLNget_product_name
set_server_id
get_server_id set_license get_token)apis r, _set_api_attrsz!SendToServerClient._set_api_attrs sn W / @ @ B BCCC ',::<<=== *4466777
r. c # * K t j d }t j d 5 }t
j t j || }|
| V d d d d S # 1 swxY w Y d S )NIMUNIFYAV_API_BASE ) max_workers)executor) osenvironget
concurrentfuturesThreadPoolExecutorr SendMessageAPIr VERSIONrL )r+ base_urlrQ rK s r, _get_apizSendToServerClient._get_api s :>>"677
2
2q
2
A
A +X - h C %%c** * * * + + + + + + + + + + + + + + + + + +s 9BB B messagec @ K d|vrt j |d< d|vrt j j |d< | j | | | j t j
|t d d S )N timestamp
message_idz agent-queuedstage) timeuuiduuid4hexr put _encode_data_to_put_in_queuer" r1 r report_reporter_gen_queued)r+ r\ s r, send_to_serverz!SendToServerClient.send_to_server s g % %#'9;;GK w & &$(JLL$4GL !
$;;GDDEEE "6nMMMMMMr. c < K | j d S r )r" r1 r= s r, r) z'SendToServerClient._invoke_send_message s r. r c t K | j d {V | j | j | j k rt d d }| j 4 d {V t d | d {V n8# t j $ r&}t d |}Y d }~nd }~ww xY wd d d d {V n# 1 d {V swxY w Y t d |r|d S d S )Nz SendToServer._send wait lockz SendToServer._send lock acquiredz&SendToServer._send cancelled unlockingz SendToServer._send lock released) r" waitrA r r< _PENDING_MESSAGES_LIMITr6 r@ r$ rB r CancelledError)r+ need_to_canceles r, r' zSendToServerClient._send s n!!######### = D$@ @ @ KK6 7 7 7 !Nz
'
'
'
'
'
'
'
' >???' 557777777777 - ' ' 'KK HIII%&NNNNNN'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
'
KK: ; ; ;
%$$ A @
%
%s< D"B=<D=C2 C-(D-C22D
DDdatac f t j |t dz }| S )N)cls
)jsondumpsr encode)r+ rr msgs r, rg z/SendToServerClient._encode_data_to_put_in_queue s* j#4555<zz||r. c t j | }| d r"t | d S t | S )Nlist)rv loadsrT r r
)r+ r\ rr s r, _decode_messagez"SendToServerClient._decode_message sK z'"" 88F 1 txx / /00 0t}} r. c | dd dz |d< | j | | | d S )Napi_retries_countr r^ )rT r rf rg )r+ r\ r^ s r, _requeue_messagez#SendToServerClient._requeue_message s] '.{{3F'J'JQ'N#$
- -g 6 6)
r. c l K t j | | }t j | j }t j ||ht j d{V \ }}|D ]}t
| d{V ||v r| dS dS )zRace the HTTP send against the shutdown signal.
Returns True on success, raises on API error,
or returns False if shutdown interrupted the send.
) return_whenNTF)r
ensure_futurer r% rm FIRST_COMPLETEDr result)r+ rK r\ send_task
shutdown_taskdone
pending_taskstasks r, _send_one_messagez$SendToServerClient._send_one_message s )#*:*:7*C*CDD -d.A.F.F.H.HII
$+L
& /%
%
%
m " ) )D"4(( ( ( ( ( ( ( ( (
4 ur. c K | j r8t d | j || dS | | }| d | d d} t j |t d | || d {V }|s8t d
| j || dS t j |t d t
d | d
S # t t f$ r=}t d|| | || Y d }~dS d }~wt" $ r=}t d|| | || Y d }~dS d }~ww xY w)zTry sending a single message.
Returns (stop, failed) where *stop* is True if further sends
should stop (shutdown or server-level error) and *failed* is
True when the message could not be delivered.
z3Shutdown signal received, saving remaining messagesr )TFmethodr_ )r r_ z
agent-sendingr` Nz?Shutdown signal received during send, saving remaining messagesz
agent-sentzmessage sent %s)FFz'Failed to send message %s to server: %s)TT)FT)r% is_setr6 r: r rf r} rT r rh _reporter_gen_sendingr _reporter_gen_sentr@ r r r r )r+ rK r^
message_bytesr\ msg_infosentexcs r,
_try_send_onez SendToServerClient._try_send_one s % % ' ' NNE
M my A A A ;&&}55 kk(++!++l33
.o
//W========D
#1
!!-9!EEE"{ W&8 M M M M KK)8 4 4 4 <' 7 NN98S
! !'9 5 5 5 ::::: NN98S
! !'9 5 5 5 ;;;;; s+ A0E 7E G2F
G2GGc f K | j }t dt | d}d}| 5 }|j 2|D ]!\ }}| j || " d d d d S |D ]\ }}| ||| d {V \ }}|dz
}|r|dz
}|rS||d } | D ]!\ }
} | j | |
"|r|t | z
}t | } n||d D ]!\ }}| j || "n1# ||d D ]!\ }}| j || "w xY w d d d n # 1 swxY w Y t d| d S )NzSending %s messagesr r r z Unsuccessful to send %s messages) r pop_allr6 r@ lenr[ server_idrf r ) r+ messages
failure_count processedrK r^ r r3 failed remainingtsmbs r, rB z)SendToServerClient._send_pending_messages s =((** )3x==999
]]__ J}$08 J J,I}M%%my%IIII J J J J J J J J
J08 ,I})-););Y
* * $ $ $ $ $ $LD& NI +%*
$,YZZ$8 &/ @ @FB M - -bB - ? ? ? ?! < *S^^ ;M$'MM
190D J J,I}M%%my%IIIIJ0D J J,I}M%%my%IIIIJ J7 J J J J J J J J J J J J J J J: 6
FFFFFs+ -F B E
-F
.E;;F FF)r/ N)'__name__
__module__ __qualname____doc__intrR rS rT rn _SEND_MESSAGE_RECURRING_TIMEr4 r AbstractEventLoopr- r> r3 staticmethodrL
contextlibcontextmanagerr r rX r[ r r
Reportabler
rj r r) r' bytesrg r} r r r rB r. r, r r ' s 7 7 "c
92>> $*
g&?
!E !E !E !EF$0 $0 $0L \ +)L$?t$KL + + + + VK "## NG N N N N $# N _122 32 _Q% % %$ U u
4- - - ^"G "G "G "G "G "Gr. r c e Zd Zej ZdZdS ) SendToServeri N)r r r r AVSCOPESHUTDOWN_PRIORITYr r. r, r r > s HEr. r ).r concurrent.futuresrU r rv loggingrR rb rc typingr defence360agent.api.serverr r r r defence360agent.contractsr defence360agent.contracts.configr "defence360agent.contracts.messagesr
r r !defence360agent.contracts.pluginsr
r 2defence360agent.internals.message_status_publisherr r ,defence360agent.internals.persistent_messager defence360agent.utilsr r r defence360agent.utils.jsonr getLoggerr r6 ri r r r r r r. r, r s# . - - - - - 1 1 1 1 1 1
B A A A A A A A M M M M M M M M K J J J J J J J J J 8 8 8 8 8 8 8 $ $suu SUU TG TG TG TG TG TG TG TGn %{ r.