ACIL FM

Nama	Ukuran	Aksi
__pycache__	-	Open
accumulate.py	3.63 MB	View DL Edit
analyst_cleanup_update.py	5.53 MB	View DL Edit
backup_info_sender.py	3.06 MB	View DL Edit
cagefs.py	5.17 MB	View DL Edit
checkpoint.py	1.23 MB	View DL Edit
client.py	10.6 MB	View DL Edit
config_merger.py	828 B	View DL Edit
config_watcher.py	1.89 MB	View DL Edit
event_hook_executor.py	777 B	View DL Edit
event_monitor.py	3.32 MB	View DL Edit
event_monitor_message_processor.py	6.33 MB	View DL Edit
files_recurring_update.py	1.09 MB	View DL Edit
icontact_sender.py	4.42 MB	View DL Edit
idle_time_out.py	1.21 MB	View DL Edit
lve_utils_install.py	1.58 MB	View DL Edit
myimunify.py	1.98 MB	View DL Edit
ping.py	536 B	View DL Edit
send_domain_list.py	2.78 MB	View DL Edit
send_server_config.py	10.96 MB	View DL Edit
__init__.py	0 B	View DL Edit

Edit file: /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins/send_domain_list.py

import datetime
import logging
import pwd
import time

from defence360agent.contracts.config import (
    int_from_envvar,
)
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.myimunify_id import get_myimunify_users
from defence360agent.contracts.plugins import (
    MessageSink,
    MessageSource,
)
from defence360agent.subsys.panels.hosting_panel import HostingPanel
from defence360agent.utils import (
    Scope,
    recurring_check,
    split_for_chunk,
)

logger = logging.getLogger(__name__)

class SendDomainList(MessageSink, MessageSource):
    SCOPE = Scope.AV_IM360

def __init__(self, period=None):
        self._task = None
        if period:
            self._period = period
        else:
            self._period = int_from_envvar(
                "IMUNIFY360_SEND_DOMAIN_PERIOD",
                int(datetime.timedelta(days=1).total_seconds()),
            )

async def create_sink(self, loop):
        """MessageSink method"""

async def create_source(self, loop, sink):
        self._loop = loop
        self._sink = sink

self._task = self._loop.create_task(
            recurring_check(self._period)(self._send_domain_list)()
        )

async def shutdown(self):
        if self._task is not None:
            self._task, t = None, self._task
            t.cancel()
            await t

def _panel_domain_type_to_imunify(self, panel_type: str) -> str:
        return {
            "main": "primary",
            "parked": "alias",
        }.get(panel_type, panel_type)

async def _create_domain_list_msg(self):
        hp = HostingPanel()
        logger.info("HostingsPanel: %s", hp.NAME)
        domains = []
        myimunify_users = await get_myimunify_users()
        for user in myimunify_users:
            username = user["username"]
            user_pwd = pwd.getpwnam(username)
            for domain_data in await hp.get_user_domains_details(username):
                domains.append(
                    {
                        "username": username,
                        "docroot": domain_data.docroot,
                        "name": domain_data.domain,
                        "securesite_user_id": user["myimunify_id"],
                        "uid": user_pwd.pw_uid,
                        "type": self._panel_domain_type_to_imunify(
                            domain_data.type
                        ),
                    }
                )
        timestamp = time.time()
        for chunk in split_for_chunk(domains, chunk_size=3000):
            msg = MessageType.DomainList()
            msg["timestamp"] = timestamp
            msg["domains"] = chunk
            yield msg

async def _send_domain_list(self):
        async for msg in self._create_domain_list_msg():
            await self._sink.process_message(msg)

Batal