|e>kQ7 d Z ddlZddlZddlZddlZddlZddlZddlZddl Z
ddl Z
ddl Z ddl
mZ ej e ZdZej dd Ze d dz Zej d d
Zd Zd Zd
Z G d d Z de fdZ G d d Z e Z ej! e j" e Z#dS )u
Lightweight message status publisher for asyncclient.
Publishes MESSAGE_STATUS events to the local proxy which relays them
to the EMQX broker via MQTT.
Each call to report() submits an HTTP POST to a thread pool — no
batching or internal queue.
Usage::
publisher = MessageStatusPublisher()
message_id_gen = Gen()
msg = {...}
message_id_gen.enrich(msg) # adds message_reporter_id / message_reporter_increment
publisher.report(msg, reporter_id_gen)
N)
is_enabledz/var/imunify360/iaidIMUNIFY_PROXY_URLzhttp://127.0.0.1:11234/z/api/v1/mqtt-publishIMUNIFY_PROXY_API_KEY c 6 e Zd ZdZddZdefdZdeddfdZdS ) Genz_ID + monotonic counter generator.
Each instance has its own UUID and its own counter.
returnNc ~ t j j | _ d| _ t j | _ d S )Nr )uuiduuid4hexid_counter threadingLock_lockselfs g/opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/internals/message_status_publisher.py__init__z Gen.__init__: s, *,,"
^%%
c v | j 5 | j }| xj dz
c_ |cd d d S # 1 swxY w Y d S )N )r r )r values r _nextz Gen._next? s
Z ME MMQ MM s .22msgc H | j |d< | |d< dS )z>Add message_reporter_id and message_reporter_increment to msg.message_reporter_idmessage_reporter_incrementN)r r )r r s r enrichz
Gen.enrichE s( %)W !",0JJLL ()))r r
N) __name__
__module__ __qualname____doc__r intr dictr$ r r r r 4 so
& & & &
s 9$ 94 9 9 9 9 9 9r r r
c t t 5 } | cd d d S # 1 swxY w Y d S # t $ r Y dS w xY w)Nr )open
_IAID_PATHreadstripOSError)fs r
_read_iaidr4 K s
*
$6688>>## $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ $ rrs3 A &A A A
A A
A
A$#A$c V e Zd Zd
dZd
dZdedededdfdZd eded
eddfd Z d
d Z
dS )MessageStatusPublisherr
Nc d| _ t j | _ d| _ t
j t d | _ t j
t | _ d S )Nr Fz
msg-status) max_workersthread_name_prefix)
_iaidr r
_init_lock _initialized
concurrentfuturesThreadPoolExecutor _MAX_WORKERS_poolBoundedSemaphore
_MAX_INFLIGHT _inflightr s r r z MessageStatusPublisher.__init__T s]
#.** ! '::$ + ;
#3MBBr c | j rd S | j 5 | j r d d d d S t | _ | j s.t dt d d d d S d| _ d d d d S # 1 swxY w Y d S )NzImsg-status: iaid not available yet (file %s missing or empty), will retryT)r< r; r4 r: loggerinfor/ r s r _ensure_initializedz*MessageStatusPublisher._ensure_initialized^ s F
_ % %
% % % % % % % % $ DJ:
*
% % % % % % % % !%D % % % % % % % % % % % % % % % % % %s B;B/BB
Br reporter_genstagec v t d sdS | dd }| d sdS j d s t d|| dS t
j |j | | dd | d d
||d } j
j ||| }n*# t $ r j
Y dS w xY w| fd dS )
z3Publish a status record via HTTP POST to the proxy.
mqtt_trackingNmethodr r" F)blockingz3msg-status: queue full, dropping stage=%s method=%sr# r ) timestamp reporter_idreporter_incrementr" r# message_typerJ c 6 j S )N)rD release)_r s r z/MessageStatusPublisher.report.. s 4>+A+A+C+C r )r getrD acquirerF warningtimer r rA submit_do_post RuntimeErrorrT add_done_callback)r r rI rJ rM recordfutures ` r reportz MessageStatusPublisher.reportn se
/** F2&&ww,-- F
~%%u%55 NNE
F '?"."4"4"6"6#&77+@"#E#E*-'',a+ + #
Z&&t}feVLLFF N " " $ $ $ FF !C!C!C!CDDDDDs "C7 7#D D r_ rM c , | | j sd S | j |d< t j | }ddi}t
r
t
|d< t j t ||d }t j
|t 5 }| d d d d S # 1 swxY w Y d S # t $ r(}t d||| Y d }~d S d }~ww xY w) Niaidz Content-Typezapplication/jsonz X-API-KeyPOST)dataheadersrM )timeoutz.msg-status: POST failed stage=%s method=%s: %r)rH r: jsondumpsencode_PROXY_API_KEYurllibrequestRequest_PUBLISH_ENDPOINTurlopen
_POST_TIMEOUTr0 ExceptionrF rY ) r r_ rJ rM payloadrf reqrespes r r\ z MessageStatusPublisher._do_post s} """z F v j((//11G%'9:G
6'5 $.((! ) C '']'CC
t
NN@
s< B C! 2C C! CC! C C! !
D+ DDc < | j d d S )NF)wait)rA shutdownr s r ry z MessageStatusPublisher.shutdown s!
'''''r r% ) r& r' r( r rH r+ r strra r\ ry r, r r r6 r6 S s C C C C% % % % 'E$ 'Ec 'E# 'E$ 'E 'E 'E 'ERt C 4( ( ( ( ( (r r6 )$r) atexitconcurrent.futuresr= rh loggingosr rZ urllib.errorrl urllib.requestr 'defence360agent.internals.feature_flagsr getLoggerr& rF r/ environrW
_PROXY_URLrstripro rk rq r@ rC r rz r4 r6 publisherregisterry message_id_genr, r r r s (
> > > > > > 8 $ $
#
Z^^ /1I
J
J
%%c**-CC
7<<
9 9 9 9 9 9 9 9.C _( _( _( _( _( _( _( _(D
# " $ $ " # # #r