✏️ Editing: server.py Read Only

"""
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 argparse
import asyncio
import gc
import logging
import os
import signal
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager, suppress
from functools import partial
from pathlib import Path
from subprocess import CalledProcessError, check_output
from typing import Tuple

import daemon
from lockfile import AlreadyLocked
import daemon.pidfile
import psutil

import defence360agent.internals.logger
from defence360agent import files
from defence360agent.api import health, inactivity
from defence360agent.contracts.config import (
    ConfigsValidator,
    Core,
    Merger,
    Model,
    SimpleRpc,
)
from defence360agent.contracts.hook_events import HookEvent
from defence360agent.contracts.license import LicenseCLN
from defence360agent.contracts.plugins import MessageSink, MessageSource
from defence360agent.internals.global_scope import g
from defence360agent.internals.iaid import IndependentAgentIDAPI
from defence360agent.internals.the_sink import TheSink
from defence360agent.model import instance, simplification, tls_check
from defence360agent.simple_rpc import (
    NonRootRpcServer,
    NonRootRpcServerAV,
    RpcServer,
    RpcServerAV,
    is_running,
)
from defence360agent.subsys import persistent_state, systemd_notifier
from defence360agent.utils import (
    Task,
    create_task_and_log_exceptions,
    is_root_user,
    is_systemd_boot,
)
from defence360agent.utils.safe_fileops import shutdown_process_pools
from defence360agent.utils.check_db import is_db_corrupted
from defence360agent.utils.cli import EXITCODE_GENERAL_ERROR
from defence360agent.utils.common import DAY, rate_limit
from defence360agent.sentry import flush_sentry
from imav.malwarelib.config import (
    MalwareHitStatus,
    MalwareScanResourceType,
    VulnerabilityHitStatus,
)
from imav.malwarelib.model import MalwareHit, VulnerabilityHit
import sentry_sdk

# Increase recursion depth to allow malware scanner into deeply nested
# directories with absolute path length up to 4096 symbols
_MAX_RECURSION_DEPTH = 2100
_DB_IS_CORRUPTED_FLAG = Path("%s.is_corrupted" % Model.PATH)
_DB_IS_CORRUPTED_MSG = (
    "Imunify360 database is corrupt. "
    "Application cannot run with corrupt database. "
    "Please, contact Imunify360 support team at "
    "https://cloudlinux.zendesk.com"
)

logger = logging.getLogger(__name__)
throttled_log_error = rate_limit(period=DAY)(logger.error)

class TaskFactory:
    def __init__(self):
        self.pool = set()

def __call__(self, loop, coro):
        task = Task(coro, loop=loop)
        self.pool.add(task)
        task.add_done_callback(self.pool.discard)
        return task

@contextmanager
def log_and_suppress_error(message):
    """Log *message* on any error & suppress it."""
    try:
        yield
    except Exception as e:
        logger.error("caught error %r on %s", e, message)
        sentry_sdk.capture_exception(e)

_PACKAGE_MANAGER_PROCESS_NAMES = frozenset({"rpm", "yum", "dnf", "rpm2cpio"})
_PACKAGE_MANAGER_WAIT_TIMEOUT = 35

def _wait_for_package_manager_children() -> None:
    """Wait for child rpm/yum/dnf processes to finish before exiting.

Exiting while a child package-manager process holds the Berkeley DB
    lock corrupts rpmdb (BDB1507). See DEF-39911.
    """
    try:
        current = psutil.Process()
        for child in current.children(recursive=True):
            try:
                name = child.name()
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
            if name in _PACKAGE_MANAGER_PROCESS_NAMES:
                logger.info(
                    "Waiting for child %s (pid=%d) to finish "
                    "to avoid rpmdb corruption",
                    name,
                    child.pid,
                )
                try:
                    child.wait(timeout=_PACKAGE_MANAGER_WAIT_TIMEOUT)
                except psutil.TimeoutExpired:
                    logger.warning(
                        "Child %s (pid=%d) did not finish in %ds",
                        name,
                        child.pid,
                        _PACKAGE_MANAGER_WAIT_TIMEOUT,
                    )
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        pass
    except Exception as e:
        logger.warning("Error waiting for package manager children: %s", e)

def _log_shutdown_debug_info() -> None:
    """Log debug information about threads and child processes at shutdown.

This helps diagnose why the process might hang during shutdown.
    """
    if not g.get("DEBUG"):
        return

# Log all alive threads
    current_thread = threading.current_thread()
    alive_threads = [t for t in threading.enumerate() if t.is_alive()]
    non_daemon_threads = [
        t for t in alive_threads if not t.daemon and t != current_thread
    ]

logger.info(
        "Shutdown debug: %d threads alive (%d non-daemon, excluding main)",
        len(alive_threads),
        len(non_daemon_threads),
    )

for t in alive_threads:
        logger.info(
            "  Thread: name=%r, daemon=%s, ident=%s",
            t.name,
            t.daemon,
            t.ident,
        )

# Log child processes
    try:
        current_process = psutil.Process()
        children = current_process.children(recursive=True)
        if children:
            logger.info("Shutdown debug: %d child processes", len(children))
            for child in children:
                try:
                    logger.info(
                        "  Child process: pid=%d, name=%r, status=%s",
                        child.pid,
                        child.name(),
                        child.status(),
                    )
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    pass
        else:
            logger.info("Shutdown debug: no child processes")
    except Exception as e:
        logger.warning("Could not enumerate child processes: %s", e)

_SHUTDOWN_DEADLINE = 50  # seconds — must be < TimeoutStopSec (60s)

def _shutdown_watchdog(deadline, stop_event):
    """Daemon thread that force-exits if shutdown exceeds *deadline* seconds.

This is defense-in-depth: if the asyncio event loop or any shutdown
    phase hangs (e.g. a task survives CancelledError — see DEF-40570),
    the process is killed regardless.
    """
    if stop_event.wait(deadline):
        return  # Normal shutdown completed — no need to force-exit.
    logger.error(
        "Shutdown watchdog: deadline of %ds exceeded, forcing exit",
        deadline,
    )
    try:
        flush_sentry()
    except Exception:
        pass
    os._exit(1)

async def _shutdown_task(loop, the_sink, plugin_list):
    # Start a watchdog thread that force-exits if shutdown hangs.
    # The thread is daemonic so it won't prevent normal exit.
    stop_watchdog = threading.Event()
    watchdog = threading.Thread(
        target=_shutdown_watchdog,
        args=(_SHUTDOWN_DEADLINE, stop_watchdog),
        daemon=True,
    )
    watchdog.start()

with log_and_suppress_error("marking the start of the shutdown process"):
        health.sensor.shutting_down(time.time())

logger.info("shutdown task starting, pid=%s", os.getpid())

# Phase 1: Stop external sources from sending new messages
    with log_and_suppress_error("closing external sources"):
        async with asyncio.timeout(10):
            if "sensor_server" in g:
                g.sensor_server.close()
                await g.sensor_server.wait_closed()

# Phase 2: Shutdown plugins by priority (they can still send to TheSink)
    for plugin in sorted(plugin_list, key=lambda p: p.SHUTDOWN_PRIORITY):
        with log_and_suppress_error("shutting down plugin"):
            async with asyncio.timeout(10):
                logger.info(
                    "Shutting down %s.%s (priority %d)...",
                    plugin.__class__.__module__,
                    plugin.__class__.__name__,
                    plugin.SHUTDOWN_PRIORITY,
                )
                await plugin.shutdown()

# Phase 3: Shutdown TheSink after all plugins have flushed
    with log_and_suppress_error("draining message queue"):
        logger.info("Shutting down TheSink...")
        async with asyncio.timeout(30):
            await the_sink.shutdown()

with log_and_suppress_error("shutting down IAID API"):
        async with asyncio.timeout(10):
            await IndependentAgentIDAPI.shutdown()

if (restart_task := g.get("web_server_restart_task")) is not None:
        with log_and_suppress_error("waiting for web server restart"):
            await asyncio.wait_for(restart_task, timeout=60)

with log_and_suppress_error("stopping loop"):
        loop.stop()

flush_sentry()

stop_watchdog.set()
    logger.info("shutdown task finished, pid=%s", os.getpid())

def _daemonize(pidfilepath):
    logger.info("Run as daemon [pidfile = %s]", pidfilepath)

dc = daemon.DaemonContext()
    dc.pidfile = daemon.pidfile.PIDLockFile(pidfilepath)

dc.prevent_core = False
    dc.umask = Core.FILE_UMASK
    if is_systemd_boot():
        dc.detach_process = False
    else:
        dc.detach_process = True
    dc.files_preserve = defence360agent.internals.logger.get_fds()
    try:
        dc.open()
    except AlreadyLocked:
        logger.error("PID file already locked by another process")
        sys.exit(EXITCODE_GENERAL_ERROR)
    gc.collect()

# quirk: somehow this is needed for root logger messages to do not
    #        propagate to specialized loggers, e.g. 'perf', 'nework'
    defence360agent.internals.logger.reconfigure()

async def _initial_files_update():
    """Perform update files on start."""
    await files.update_all_no_fail_if_files_exist()

def _tls_check_reset(loop):
    # init thread id for simplification.run_in_executor() worker thread
    loop.run_until_complete(
        simplification.run_in_executor(loop, tls_check.reset)
    )

# mark current thread as "main_thread" for more informative error messages
    # PSSST! simplification.run_in_executor() is main thread now! :-X
    # tls_check.reset("main_thread")

def plugin_instances(objs, pclass):
    return [p for p in objs if isinstance(p, pclass)]

def _start_plugins(loop, plugin_classes) -> Tuple[TheSink, list, list]:
    plugins = [plugin_class() for plugin_class in plugin_classes]

# instantiate sinks
    sinks = plugin_instances(plugins, MessageSink)
    for s in sinks:
        logger.info("Creating sink %r", s)
        loop.run_until_complete(s.create_sink(loop))

# instantiate sources
    the_sink = TheSink(sinks, loop)
    sources = plugin_instances(plugins, MessageSource)
    for s in sources:
        logger.info("Creating source %r", s)
        loop.run_until_complete(s.create_source(loop, the_sink))

the_sink.start()

return the_sink, sinks, sources

def _start_rpc(loop, the_sink: TheSink):
    logger.info("Starting RpcServers...")
    if SimpleRpc.SOCKET_ACTIVATION:
        rpc_servers = (RpcServerAV, NonRootRpcServerAV)
    else:
        rpc_servers = (RpcServer, NonRootRpcServer)
    for rpc in rpc_servers:
        loop.run_until_complete(rpc.create(loop, the_sink))

def _get_pids_open(*files):
    try:
        out = check_output(
            ["lsof", "+wt"] + list(files),
            env={"PATH": "/usr/sbin:/usr/bin", **os.environ},
        )
    except CalledProcessError as e:
        out = bytes(e.output)
    except FileNotFoundError:
        logger.warning("There is no lsof in /usr/sbin:/usr/bin")
        return []
    except IOError:
        return []
    lines = out.strip().split(b"\n")
    pids = [int(line) for line in lines if line]
    return list(set(pids))

def _check_able_to_start(pidfile):
    if is_running():
        # get parent process info
        ppid = os.getppid()
        if ppid != 0:
            parent = psutil.Process(ppid).name()
            pids_used_socket = _get_pids_open(
                SimpleRpc.SOCKET_PATH, SimpleRpc.NON_ROOT_SOCKET_PATH
            )
            process_used_socket = []
            for pid in pids_used_socket:
                try:
                    _pr = psutil.Process(pid)
                except psutil.NoSuchProcess:
                    continue
                _local_parent = _pr.parent()
                if _local_parent:
                    _parent_name = _local_parent.name()
                else:
                    _parent_name = "None"
                process_used_socket.append(
                    (
                        pid,
                        _pr.name(),
                        "parent process = %s" % str(_parent_name),
                    )
                )
            try:
                with open(pidfile) as file:
                    written_pid = file.read()
            except (OSError, IOError):
                written_pid = None
            throttled_log_error(
                "Instance of %s is already running. "
                'Parent process "%s" with pid "%s". '
                "Sockets are in use by %s. "
                "%s file contents %s pid"
                % (
                    Core.SVC_NAME,
                    parent,
                    ppid,
                    str(process_used_socket),
                    pidfile,
                    written_pid,
                )
            )
            sys.exit(EXITCODE_GENERAL_ERROR)

if is_db_corrupted(db_path=Model.PATH):
        if not _DB_IS_CORRUPTED_FLAG.exists():
            logger.error(_DB_IS_CORRUPTED_MSG)
            _DB_IS_CORRUPTED_FLAG.touch()
        else:
            logger.warning(_DB_IS_CORRUPTED_MSG)
        sys.exit(EXITCODE_GENERAL_ERROR)
    else:
        with suppress(FileNotFoundError):
            _DB_IS_CORRUPTED_FLAG.unlink()

def start(plugin_classes: list, init_actions) -> None:
    """Common function for agent service startup.

plugin_classes is a list of classes implementing message processing
    plugins. init_actions is a coroutine that will be called prior to starting
    RPC and message processing."""
    if not is_root_user():
        logger.info("Imunify agent could be started by the root user only!")
        sys.exit(EXITCODE_GENERAL_ERROR)

args = parse_cli()

defence360agent.internals.logger.setLogLevel(args.verbose)
    if args.log_config or os.environ.get("IMUNIFY360_LOGGING_CONFIG_FILE"):
        defence360agent.internals.logger.update_logging_config_from_file(
            args.log_config or os.environ.get("IMUNIFY360_LOGGING_CONFIG_FILE")
        )

sys.setrecursionlimit(_MAX_RECURSION_DEPTH)

_check_able_to_start(args.pidfile)

if args.daemon:
        _daemonize(args.pidfile)
        systemd_notifier.notify(systemd_notifier.AgentState.DAEMONIZED)

health.sensor.starting(time.time())
    if not LicenseCLN.is_registered():
        health.sensor.unregistered()

loop = asyncio.get_event_loop()
    _cpu = os.cpu_count()
    # https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor
    # default's in Python 3.8
    loop.set_default_executor(
        ThreadPoolExecutor(max_workers=min(32, _cpu + 4 if _cpu else 5))
    )
    loop.set_task_factory(TaskFactory())
    try:
        _tls_check_reset(loop)
        instance.db.init(Model.PATH)

validate_configs_on_start(loop)
        Merger.update_merged_config()

loop.run_until_complete(init_actions())
        try:
            for _stop_outdated in [_stop_pending_cleanup, _stop_pending_patch]:
                _stop_outdated()
        except simplification.PeeweeException as e:
            # we intentionally capture all exceptions here and log them
            # it may happened on package update or other reasons, we don't
            # want to start agent in such case
            logger.error(
                "Failed to stop pending cleanup/patch. Reason: %s", repr(e)
            )
            sys.exit(EXITCODE_GENERAL_ERROR)

# If this is first agent run - we SHOULD download
        # all of the static files
        # If it isn't first agent run - essential files already downloaded
        # and will be updated asynchronously
        if not loop.run_until_complete(files.essential_files_exist()):
            logger.info(
                "Essential files are missing. Performing initial files update."
            )
            try:
                loop.run_until_complete(_initial_files_update())
            except files.UpdateError as e:
                logger.error("Failed to perform initial files update: %s", e)
                sys.exit(EXITCODE_GENERAL_ERROR)
        inactivity.track.set_timeout(SimpleRpc.INACTIVITY_TIMEOUT)

the_sink, sinks, sources = _start_plugins(loop, plugin_classes)
        _start_rpc(loop, the_sink)
        logger.info("Message Bus started")
        agent_started = HookEvent.AgentStarted(
            version=Core.VERSION, resident=False
        )
        create_task_and_log_exceptions(
            loop, the_sink.process_message, agent_started
        )

try:
            persistent_state.remove_unused_locks()
        except Exception as e:
            logger.error("Failed to remove unused locks: %s", e)
        # note: plugins are started before the shutdown task has been setup
        #  therefore plugin.shutdown() won't be called before create_source()
        _setup_signal_handlers(
            loop, partial(_shutdown_task, loop, the_sink, sinks + sources)
        )
        loop.run_forever()
        logger.info("loop stopped")
    finally:
        # Log debug info about threads/processes before cleanup
        logger.info("Starting executor cleanup")
        _log_shutdown_debug_info()

# Cancel any remaining pending tasks to prevent blocking during cleanup
        pending = [t for t in asyncio.all_tasks(loop) if not t.done()]
        if pending:
            logger.info("Cancelling %d pending tasks", len(pending))
            for task in pending:
                task.cancel()
            # Give tasks a chance to handle cancellation
            with suppress(Exception):
                loop.run_until_complete(asyncio.wait(pending, timeout=2.0))

# Fix: Explicitly shutdown the executor without waiting for threads.
        # Python 3.9+ loop.close() calls executor.shutdown(wait=True) which
        # blocks if any thread is still running (e.g., blocked on network I/O).
        # asyncio.wait_for() timeouts only cancel the coroutine, not the
        # underlying thread, so threads with long socket timeouts (45s) can
        # block shutdown for up to 60s causing systemd to SIGKILL the process.
        executor = getattr(loop, "_default_executor", None)
        if executor:
            logger.info("Shutting down ThreadPoolExecutor")
            executor.shutdown(wait=False, cancel_futures=True)

# Shutdown any active ProcessPoolExecutors used for privilege-dropping
        logger.info("Shutting down ProcessPoolExecutors")
        shutdown_process_pools()

# Log debug info after cleanup to see what's still alive
        logger.info("Executor cleanup complete")
        _log_shutdown_debug_info()

# DEF-39911: wait for any child rpm/yum/dnf processes to finish
        # before exiting, otherwise rpmdb gets corrupted
        _wait_for_package_manager_children()

# closing the loop after loop.stop() cuts off pending tasks stacktraces
        loop.close()
        logger.info("Event loop closed")

def validate_configs_on_start(loop):
    try:
        ConfigsValidator.validate_config_layers()
    except Exception as e:
        from defence360agent.hooks.execute import execute_hooks

agent_misconfig = HookEvent.AgentMisconfig(error=repr(e))
        loop.run_until_complete(execute_hooks(agent_misconfig))
        logger.warning(str(e))
        sys.exit(EXITCODE_GENERAL_ERROR)

def _setup_signal_handlers(loop, shutdowntask):
    called = False  # whether the signal handler was called already

def _sighandler(loop, sig):
        nonlocal called
        if not called:
            called = True
            logger.info("Caught %s", sig)
            # note: store ref, to keep the task alive, just in case
            called = create_task_and_log_exceptions(loop, shutdowntask)
        else:
            logger.info(
                "Caught %s. Shutdown task is already running, please wait.",
                sig,
            )

for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGUSR1, signal.SIGUSR2):
        loop.add_signal_handler(sig, _sighandler, loop, sig)

def parse_cli():
    parser = argparse.ArgumentParser(description="Run imunify agent")
    parser.add_argument(
        "-v",
        dest="verbose",
        action="count",
        default=0,
        help=(
            "Level of logging. Each value corresponds to:"
            "1 - console only log level,"
            "2 - previous plus add network log,"
            "3 - all previous plus add process message log,"
            "4 - all previous plus add debug log"
        ),
    )
    parser.add_argument("--daemon", action="store_true", help="run as daemon")
    parser.add_argument(
        "--pidfile",
        default="/var/run/imunify360.pid",
        help="use with --daemon",
    )
    parser.add_argument("--log-config", help="logging config filename")
    return parser.parse_args(sys.argv[1:])

def _stop_pending_cleanup():
    """
    Get back to FOUND all malware hits which have stuck in CLEANUP_STARTED
    """
    hits = MalwareHit.select().where(
        MalwareHit.status == MalwareHitStatus.CLEANUP_STARTED,
        MalwareHit.resource_type == MalwareScanResourceType.FILE.value,
    )
    MalwareHit.set_status(hits, MalwareHitStatus.FOUND)

def _stop_pending_patch():
    """
    Get back to VULNERABLE all vulnerabilities which have stuck in PATCH_IN_PROGRESS
    """
    hits = VulnerabilityHit.select().where(
        VulnerabilityHit.status == VulnerabilityHitStatus.PATCH_IN_PROGRESS,
    )
    VulnerabilityHit.set_status(hits, VulnerabilityHitStatus.VULNERABLE)

🔐 Sid Gifari File Manager Pro