REDROOM

Edit File Size: 2.65 KB

/proc/thread-self/root/opt/imunify360/venv/lib64/python3.11/site-packages/imav/malwarelib/subsys/aibolit.py

Text Base64

"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License,
or (at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. 
See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License
 along with this program.  If not, see <https://www.gnu.org/licenses/>.

This software is also available under ImunifyAV commercial license,
see <https://www.imunify360.com/legal/eula>
"""
import logging

from defence360agent.subsys import svcctl
from defence360agent.utils import CheckRunError  # NOSONAR

logger = logging.getLogger(__name__)

AIBOLIT_SOCKET_NAME = "aibolit-resident.socket"
AIBOLIT_SVCNAME = "aibolit-resident"

async def restart_on_sigs_or_config_update(_, is_updated):
    if is_updated:
        logger.info("ai-bolit service will be restarted")
        aibolit_socket = svcctl.adaptor(AIBOLIT_SOCKET_NAME)
        aibolit_service = svcctl.adaptor(AIBOLIT_SVCNAME)
        try:
            await _ensure_socket_active(aibolit_service, aibolit_socket)
            await aibolit_service.restart()
        except CheckRunError as e:
            if "is masked" in str(e):
                logger.warning(
                    "Aibolit service is masked, skipping restart during"
                    " installation"
                )
                return
            raise
        logger.debug("ai-bolit service restarted")

async def _ensure_socket_active(aibolit_service, aibolit_socket):
    """Restart aibolit-resident.socket if it was killed by systemd rate
    limiting.  When the service hits StartLimitBurst, systemd stops
    both the service *and* the socket (via PartOf) and marks them
    failed.  A plain ``restart`` of the service won't help because
    the socket stays dead.  We reset-failed + restart the socket so
    that the next scan can trigger socket activation again."""
    try:
        if not await aibolit_socket.is_active():
            logger.warning("aibolit-resident.socket is not active, recovering")
            await aibolit_service.reset_failed()
            await aibolit_socket.reset_failed()
            await aibolit_socket.restart()
    except Exception:
        logger.exception("failed to recover aibolit-resident.socket")

async def restart_on_detect_admin_tools_update():
    await restart_on_sigs_or_config_update(
        None,
        True,
    )

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Name	Size	Perms	Modified	Actions
__pycache__ DIR	-	drwxr-xr-x	2026-06-08 20:24:16	Edit Download recursive recursive
aibolit.py	2.65 KB	lrw-r--r--	2026-05-26 21:31:52	Edit Download recursive recursive
ainotify.py	8.59 KB	lrw-r--r--	2026-05-26 21:31:52	Edit Download recursive recursive
cxs.py	1.37 KB	lrw-r--r--	2026-05-26 21:31:52	Edit Download recursive recursive
malware.py	41.54 KB	lrw-r--r--	2026-05-26 21:31:52	Edit Download recursive recursive
pure_ftpd.py	5.50 KB	lrw-r--r--	2026-05-26 21:31:52	Edit Download recursive recursive
restore_from_backup.py	2.42 KB	lrw-r--r--	2026-05-26 21:31:52	Edit Download recursive recursive
__init__.py	0 B	lrw-r--r--	2026-05-26 21:31:52	Edit Download recursive recursive

Directory Contents