📁
SKYSHELL MANAGER
PHP v8.1.34
Create
Create
Path:
root
/
home
/
terracebizon
/
public_html
/
wp-includes
/
js
/
tinymce
/
themes
/
Name
Size
Perm
Actions
📁
inlite
-
0755
🗑️
🏷️
🔒
📁
modern
-
0755
🗑️
🏷️
🔒
📄
config.php
6.83 KB
0444
🗑️
🏷️
⬇️
✏️
🔒
📄
error_log
12934.91 KB
0644
🗑️
🏷️
⬇️
✏️
🔒
Edit: service_manager.py
"""IM360 services manager plugin. Manages firewall-related services (DoS protector, UAL, auditd, scanlogd) based on imunify360 config changes. """ import logging import shutil from pathlib import Path from random import randint from tempfile import NamedTemporaryFile from defence360agent import utils from defence360agent.contracts import plugins from defence360agent.plugins.service_manager import BaseServiceManager from defence360agent.subsys import svcctl from defence360agent.subsys.persistent_state import load_state, save_state from im360.contracts import config, messages as im360_messages from im360.subsys import csf __all__ = ["ServiceManager"] logger = logging.getLogger(__name__) UAL_CRON_TEMPLATE_PATH = Path( "/opt/imunify360/venv/share/imunify360/imunify360-ual.cron.template" ) UAL_OLD_CRON_PATH = Path("/etc/cron.d/imunify360-ual.cron") UAL_CRON_PATH = Path("/etc/cron.d/imunify360-ual") class ServiceManager(BaseServiceManager): """IM360 service manager: manages firewall services.""" SCOPE = utils.Scope.IM360 AUDITD_SHOULD_BE_RUNNING = config.FromConfig("LOGGER", "syscall_monitor") def __init__(self, *, unitctl=None): super().__init__() self._services = [ self._ensure_consistent_dos_protector_state, self._ensure_consistent_ual_state, self._ensure_consistend_auditd_state, self._ensure_consistent_scanlogd_state, ] self._units = { "dos_protector": unitctl or svcctl.imunify360_dos_protector_service(), "ual": unitctl or svcctl.imunify360_ual_service(), "auditd": unitctl or svcctl.imunify360_auditd_service(), "scanlogd": unitctl or svcctl.imunify360_scanlogd_service(), } self._configs = load_state("service_manager") if not self._configs: self._configs = {"dos_protector": {}} async def create_sink(self, loop): pass async def _ensure_consistent_dos_protector_state(self): unitctl = self._units["dos_protector"] if not unitctl: # unsupported platform return old_config = self._configs["dos_protector"] new_config = config.EnhancedDOS.as_dict() should_be_running = ( config.EnhancedDOS.ENABLED and not await csf.is_running() ) await self._ensure_service_status( unitctl, "DosProtector", should_be_running, reload=(old_config != new_config), ) self._configs["dos_protector"] = new_config save_state("service_manager", self._configs) @plugins.expect(im360_messages.StrategyChange) async def on_strategy_change( self, _message: im360_messages.StrategyChange ): async with self._lock: await self._ensure_consistent_dos_protector_state() async def _ensure_consistent_ual_state(self): should_be_running = config.UnifiedAccessLogger.ENABLED unitctl = self._units["ual"] if should_be_running: UAL_CRON_PATH.unlink(missing_ok=True) await self._ensure_service_status( unitctl, "UnifiedAccessLogger", should_be_running, reload=False ) if not should_be_running: self._create_ual_cronjob() def _create_ual_cronjob(self): if UAL_OLD_CRON_PATH.exists(): UAL_OLD_CRON_PATH.unlink() try: cronjob_content = UAL_CRON_TEMPLATE_PATH.read_text().format( random_minute=randint(0, 59), report_interval="5m" ) except FileNotFoundError: logger.warning( "UAL cron template not found: %s", UAL_CRON_TEMPLATE_PATH ) return if UAL_CRON_PATH.exists(): old_cronjob_content = UAL_CRON_PATH.read_text() identical = True for old_line, new_line in zip( old_cronjob_content.splitlines(), cronjob_content.splitlines() ): # {random_minute} * * * * root /usr/sbin/imunify36... if "* * *" in old_line: # is cronjob line # ignore minute part old_line = old_line[old_line.find(" ") + 1 :] new_line = new_line[new_line.find(" ") + 1 :] if old_line != new_line: identical = False break if identical: return with NamedTemporaryFile("w", delete=False) as f: temp_cronjob_path = Path(f.name) f.write(cronjob_content) temp_cronjob_path.chmod(0o644) shutil.move(temp_cronjob_path, UAL_CRON_PATH) async def _ensure_consistent_scanlogd_state(self): should_be_running = config.Scanlogd.ENABLE unitctl = self._units["scanlogd"] await self._ensure_service_status( unitctl, "Scanlogd", should_be_running, reload=False ) async def _ensure_consistend_auditd_state(self): unitctl = self._units["auditd"] if not unitctl: # unsupported platform return await self._ensure_service_status( unitctl, "AuditD", self.AUDITD_SHOULD_BE_RUNNING, reload=False )
Save