C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// import asyncio import logging import time from pathlib import Path from defence360agent.internals.iaid import IAIDTokenError from defence360agent.api.server import APIError from defence360agent.api.server.events import EventsAPI from defence360agent.contracts.config import ( Core, IContactMessageType, ) from defence360agent.contracts.messages import MessageType from defence360agent.contracts.plugins import ( MessageSink, MessageSource, ) from defence360agent.internals.the_sink import TheSink from defence360agent.model.icontact import IContactThrottle from defence360agent.subsys.panels.cpanel import cPanel from defence360agent.subsys.panels.plesk import Plesk from defence360agent.subsys.panels.hosting_panel import HostingPanel from defence360agent.utils import ( await_for, create_task_and_log_exceptions, recurring_check, retry_on, Scope, ) from defence360agent.utils.common import DAY logger = logging.getLogger(__name__) async def async_log_on_error(e, i): logger.warning( "Can't get recommendations for the dashboard due to " "iaid token error, reason: %s. Attempt %s", e, i, ) await_for(seconds=100) class IContactSender(MessageSink, MessageSource): PROCESSING_ORDER = MessageSink.ProcessingOrder.ICONTACT_SENT SCOPE = Scope.AV_IM360 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._tasks = [] self._notification_flag_path = ( Path(Core.TMPDIR) / "icontact_generic_notifications" ) async def create_sink(self, loop): pass async def _send_icontact_message( self, *, message_type, params, period_limit, user=None, ): if message_type is None: return if not IContactThrottle.may_be_notified( message_type, period_limit, user=user, ): return template_args = await self._panel.notify( message_type=IContactMessageType.GENERIC, params=params, user=user, ) if template_args: IContactThrottle.refresh(message_type, user=user) sent_message = MessageType.IContactSent( message_type=message_type, timestamp=int(time.time()), template_args=template_args, ) await self._sink.process_message(sent_message) async def create_source(self, loop, sink: TheSink): self._sink = sink self._panel = HostingPanel() if self._panel.NAME in [cPanel.NAME, Plesk.NAME]: self._tasks = [ create_task_and_log_exceptions( loop, self