__ __ __ __ _____ _ _ _____ _ _ _
| \/ | \ \ / / | __ \ (_) | | / ____| | | | |
| \ / |_ __\ V / | |__) | __ ___ ____ _| |_ ___ | (___ | |__ ___| | |
| |\/| | '__|> < | ___/ '__| \ \ / / _` | __/ _ \ \___ \| '_ \ / _ \ | |
| | | | |_ / . \ | | | | | |\ V / (_| | || __/ ____) | | | | __/ | |
|_| |_|_(_)_/ \_\ |_| |_| |_| \_/ \__,_|\__\___| |_____/|_| |_|\___V 2.1
if you need WebShell for Seo everyday contact me on Telegram
Telegram Address : @jackleet
For_More_Tools:
"""Restart the agent when a Coraza panel feature flag changes.
``coraza_for_cpanel_apache`` and ``coraza_for_cloudways_hybrid`` select which
panel ``HostingPanel()`` builds (plain vs Coraza-driven), and that choice is
made once, at startup. Syncing a new flag value into the feature-flags file
does not rebuild the panel on its own, so it would otherwise take effect only
on the next agent restart.
This watcher watches the feature-flags directory via inotify and restarts the
agent whenever either flag's value changes, applying the new selection without
waiting for an unrelated restart.
IM360-only: the flags affect panels/wafd, which do not exist in AV mode.
"""
from logging import getLogger
from pathlib import Path
from defence360agent.contracts.plugins import MessageSource
from defence360agent.internals.feature_flags import FLAGS_PATH, is_enabled
from defence360agent.subsys.svcctl import imunify360_agent_service
from defence360agent.utils import Scope, recurring_check, safe_cancel_task
from im360.contracts.config import Webshield
from imav.malwarelib.subsys.ainotify import Inotify, Watcher
log = getLogger(__name__)
class CorazaFFWatcher(MessageSource):
SCOPE = Scope.IM360
WATCH_DIR = str(Path(FLAGS_PATH).parent).encode("ascii")
WATCH_FILE = Path(FLAGS_PATH).name.encode("ascii")
def __init__(self) -> None:
self.__last = None
self.__task = None
self.__watcher = None
@staticmethod
def __get_flags() -> dict[str, bool]:
# fmt: off
return {
key: is_enabled(key)
for key in (
"coraza_for_cpanel_apache",
"coraza_for_cloudways_hybrid",
)
}
# fmt: on
async def create_source(self, loop, sink, watcher=None):
self.__last = self.__get_flags()
self.__watcher = Watcher(loop) if watcher is None else watcher
self.__watcher.watch(self.WATCH_DIR, Inotify.MOVED_TO)
self.__task = loop.create_task(self.__process_events())
log.info("Init Coraza feature flags: %s.", self.__last)
async def shutdown(self) -> None:
if self.__task is not None:
await safe_cancel_task(self.__task)
self.__task = None
if self.__watcher is not None:
self.__watcher.close()
self.__watcher = None
@recurring_check(0)
async def __process_events(self):
event = await self.__watcher.get_event()
if event.name != self.WATCH_FILE:
return
cur = self.__get_flags()
prev = self.__last
self.__last = cur
if prev == cur:
return
if Webshield.USE_FEATURE_FLAGS:
log.info("Coraza feature flags changed: %s -> %s.", prev, cur)
await imunify360_agent_service().restart()
else:
log.info("Coraza feature flags disabled.")