SWAMPy

✎ CARVE FLESH

EDITING: cpanel_fileman_hook

#!/opt/imunify360/venv/bin/python3
"""This script is a cPanel hook script for several Filemanager related events.

Based on:

* https://documentation.cpanel.net/display/DD/Guide+to+Standardized+Hooks
* https://documentation.cpanel.net/display/DD/Guide+to+Standardized+Hooks+-+Hook+Action+Code
"""  # noqa: E501

import logging
import json
import os
import signal
import socket
import sys
from tempfile import NamedTemporaryFile
from typing import Callable, List, Optional, TextIO

from defence360agent import sentry
from defence360agent.utils import get_abspath_from_user_dir
from im360 import aibolit_job

logger = logging.getLogger("cpanel_fileman_hook")

# Control-char escapes for log lines parsed by ossec.
_CONTROL_CHAR_TRANSLATIONS = {
    ord("\n"): "\\n",
    ord("\r"): "\\r",
    **{c: f"\\x{c:02x}" for c in range(0x00, 0x09)},
    **{c: f"\\x{c:02x}" for c in (0x0B, 0x0C)},
    **{c: f"\\x{c:02x}" for c in range(0x0E, 0x20)},
    0x7F: "\\x7f",
}

def _log_safe(value):
    """Escape newlines and control chars; pass tab and None through."""
    if value is None:
        return None
    if not isinstance(value, str):
        value = str(value)
    return value.translate(_CONTROL_CHAR_TRANSLATIONS)

# _PATH_HOOK contains the location of this hook on the server
_PATH_HOOK = "/usr/libexec/imunify360/cpanel_fileman_hook"

# Defensive caps against unbounded reads. The stdin cap bounds the
# whole JSON envelope; the content cap is a tighter bound on what
# gets spooled to /tmp via NamedTemporaryFile (it must be strictly
# smaller than the stdin cap, otherwise the JSON-envelope overhead
# guarantees content < raw < MAX_CONTENT_SIZE and the check is dead).
MAX_INPUT_SIZE = 10 * 1024 * 1024
MAX_CONTENT_SIZE = 5 * 1024 * 1024

DESCRIBE_DATA = [
    {
        "blocking": 1,
        "escalateprivs": 0,
        "category": "Cpanel",
        "event": "UAPI::Fileman::upload_files",
        "stage": "pre",
        "hook": _PATH_HOOK + " --upload",
        "exectype": "script",
    },
    {
        "blocking": 1,
        "escalateprivs": 0,
        "category": "Cpanel",
        "event": "UAPI::Fileman::save_file_content",
        "stage": "pre",
        "hook": _PATH_HOOK + " --save",
        "exectype": "script",
    },
    {
        "blocking": 1,
        "escalateprivs": 0,
        "category": "Cpanel",
        "event": "Api2::Fileman::savefile",
        "stage": "pre",
        "hook": _PATH_HOOK + " --save",
        "exectype": "script",
    },
]

class Context:
    def __init__(
        self,
        stdin: TextIO,
        stdout: TextIO,
        stderr: TextIO,
        args: List[str],
        checker: Callable[[List[str], List[Optional[str]]], bool],
    ):
        self.stdin = stdin
        self.stdout = stdout
        self.stderr = stderr
        self.args = args
        self.checker = checker

def status_text(
    allowed: bool, method=None, filename=None, folder=None, user=None
) -> str:
    if not allowed:
        logger.info(
            "0 BAILOUT malware detected when %s '%s' in %s for user %s",
            _log_safe(method),
            _log_safe(filename),
            _log_safe(folder),
            _log_safe(user),
        )
    return "1" if allowed else "0 BAILOUT malware detected"

def status_code(allowed: bool) -> int:
    return 0

def describe_action(ctx: Context) -> int:
    ctx.stdout.write(json.dumps(DESCRIBE_DATA))
    return 0

def _resolve_dest_path(folder, filename, user):
    # cPanel sends an absolute dir for UAPI events, a home-relative path for
    # the legacy Api2 savefile event.
    if not folder or not filename:
        return None
    if os.path.isabs(folder):
        joined = os.path.join(folder, filename)
    else:
        try:
            joined = str(
                get_abspath_from_user_dir(user, os.path.join(folder, filename))
            )
        except (ValueError, TypeError):
            return None
    # Collapse '.'/'..' so a crafted path can't lexically match a whitelisted
    # ancestor while cPanel writes the file to a different real location.
    return os.path.normpath(joined)

def check_upload(ctx: Context) -> int:
    logger.info("upload action")
    suffix = "-key"
    allowed = True
    raw = ctx.stdin.read(MAX_INPUT_SIZE + 1)
    if len(raw) > MAX_INPUT_SIZE:
        logger.warning(
            "upload: input too large (>%d bytes); rejecting",
            MAX_INPUT_SIZE,
        )
        raise ValueError("input too large")
    data = json.loads(raw)["data"]
    args = data["args"]
    user = data.get("user")
    # A plural upload_files request carries one "file-<name>-key" entry per
    # uploaded file (value "file-N"), with the temp spool path under the
    # matching "file-<name>". Collect every file so the whole batch is scanned
    # in one job — the resident blocks the upload if any file is malware.
    files = []
    dest_paths = []
    filenames = []
    for k in args:
        if k.endswith(suffix):
            path = args.get(k[: -len(suffix)])
            filename = k[len("file-"):-len(suffix)]
            if path:
                files.append(path)
                dest_paths.append(
                    _resolve_dest_path(args.get("dir"), filename, user)
                )
                filenames.append(filename)
    if not files:
        logger.warning(
            "upload action: no '-key' suffix found in cPanel hook args; "
            "scan skipped (user '%s', dir '%s', available arg keys: %s)",
            _log_safe(user),
            _log_safe(args.get("dir")),
            [_log_safe(k) for k in sorted(args.keys())],
        )
    else:
        allowed = ctx.checker(files, dest_paths)
    ctx.stdout.write(
        status_text(
            allowed,
            "upload",
            ", ".join(filenames),
            args.get("dir"),
            user,
        )
    )
    return status_code(allowed)

def check_save(ctx: Context) -> int:
    logger.info("save action")
    allowed = True
    raw = ctx.stdin.read(MAX_INPUT_SIZE + 1)
    if len(raw) > MAX_INPUT_SIZE:
        logger.warning(
            "save: input too large (>%d bytes); rejecting",
            MAX_INPUT_SIZE,
        )
        raise ValueError("input too large")
    data = json.loads(raw)["data"]
    args = data["args"]
    filename = args.get("filename") or args.get("file")
    folder = args.get("dir") or args.get("path")
    user = data["user"]
    if "content" not in args:
        logger.warning(
            "save action: 'content' field missing from cPanel hook "
            "payload; scan skipped (file '%s' in '%s' for user '%s')",
            _log_safe(filename),
            _log_safe(folder),
            _log_safe(user),
        )
        content = None
    else:
        content = args["content"]
        if content != "" and not isinstance(content, str):
            logger.warning(
                "save action: 'content' is %s, expected str; "
                "scan skipped (file '%s' in '%s' for user '%s')",
                type(content).__name__,
                _log_safe(filename),
                _log_safe(folder),
                _log_safe(user),
            )
    if isinstance(content, str) and len(content) > MAX_CONTENT_SIZE:
        logger.warning(
            "save: content too large (%d bytes > %d); rejecting",
            len(content),
            MAX_CONTENT_SIZE,
        )
        raise ValueError("content too large")
    if isinstance(content, str) and content:
        dest = _resolve_dest_path(folder, filename, user)
        with NamedTemporaryFile(mode="w") as ntf:
            ntf.write(content)
            ntf.flush()
            allowed = ctx.checker([ntf.name], [dest])
    ctx.stdout.write(
        status_text(
            allowed,
            "save",
            filename,
            folder,
            user,
        )
    )
    return status_code(allowed)

KNOWN_ACTIONS = {
    "describe": describe_action,
    "upload": check_upload,
    "save": check_save,
}

def aibolit_checker(files: List[str], dest_paths: List[Optional[str]]) -> bool:
    # FOLLOWING IS MOSTLY COPIED FROM modsec_scan_real.py

resident_dir_path = aibolit_job.RESIDENT_DIR
    # to include the import time, we could read the start time of the
    # process https://gist.github.com/westhood/1073585
    remaining_time = aibolit_job.create_remaining_time_func(
        aibolit_job.UPLOAD_TIMEOUT
    )

# signals we'll be waiting for from aibolit
    sigset = {signal.SIGUSR1, signal.SIGUSR2}
    # block the signal in all threads
    signal.pthread_sigmask(signal.SIG_BLOCK, sigset)

# submit every uploaded file for scanning in one job; the resident returns
    # SIGUSR1 (block) if any of them is malware.
    aibolit_job.create_upload_job(
        files=files,
        dest_paths=dest_paths,
        resident_dir_path=resident_dir_path,
        timeout=remaining_time(),
    )
    logger.info("files sent for scanning: %s", files)

# notify aibolit about the new job
    aibolit_job.notify_aibolit_start_it_if_necessary()

# wait for response
    while True:
        # use sigtimedwait() instead of signal() to get the uid

# note: ignore a possible race on retry inside sigtimedwait() on
        # receiving a signal (see sigtimedwait()'s Python docs)
        si = signal.sigtimedwait(sigset, remaining_time())
        if si is None:
            # timed out
            logger.warning("timed out while scanning %s", files)
            return True
        if si.si_uid == 0:  # the signal is from root
            if si.si_signo == signal.SIGUSR1:
                return False
            elif si.si_signo == signal.SIGUSR2:
                return True
            else:
                assert 0, "shouldn't happen"  # pragma: no cover

def setup_logging() -> None:
    """Setup logging carefully for ossec to capture.

When the hook prints logs on stderr, cpanel captures logs and prints
    them to /usr/local/cpanel/logs/error_log file.

We need to make sure the logging format matches the syslog format
    for ossec to decode it perfectly.
    """
    global logger
    hostname = socket.getfqdn()
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler()
    formatter = logging.Formatter(
        f"%(asctime)s {hostname} %(name)s[%(process)d]: %(message)s",
        datefmt="%b %d %H:%M:%S",
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    sentry.configure_sentry()

def do_main(ctx: Context) -> int:
    if len(ctx.args) < 2:
        print("No command is given.", file=ctx.stderr)
        return 1
    if not ctx.args[1].startswith("--"):
        print("Wrong argument:", ctx.args[1], file=ctx.stderr)
        return 1
    action = ctx.args[1][2:]
    if action not in KNOWN_ACTIONS:
        print("Unknown action:", action, file=ctx.stderr)
        return 1
    return KNOWN_ACTIONS[action](ctx)

def main(ctx: Context) -> int:
    try:
        return do_main(ctx)
    except Exception as e:
        print("1 Exception:", e, file=ctx.stderr)
        logger.exception("internal error: %s", e)
        return 1

if __name__ == "__main__":
    setup_logging()
    # First line of log does not get its own line but prepended with
    # cpanel logging format, e.g.(/usr/local/cpanel/logs/error_log):
    #
    # [2024-07-16 07:42:40 +0000] info [uapi] STDERR output from hook: /usr/libexec/imunify360/cpanel_fileman_hook --upload
    # [2024-07-16 07:42:40 +0000] info [uapi] Jul 16 07:42:39 cl7x64.cltest.com cpanel_fileman_hook[101676]: Starting imunify fileman hook
    # Jul 16 07:42:39 cl7x64.cltest.com cpanel_fileman_hook[101676]: upload action
    # Jul 16 07:42:39 cl7x64.cltest.com cpanel_fileman_hook[101676]: file /home/user228/tmp/Cpanel_Form_file.upload.bb6355b0 is sent for scanning
    # Jul 16 07:42:40 cl7x64.cltest.com cpanel_fileman_hook[101676]: 0 BAILOUT malware detected when upload 'eicar.com' in public_html for user user228
    # Jul 16 07:42:40 cl7x64.cltest.com cpanel_fileman_hook[101676]: exiting with code 0
    #
    # [2024-07-16 07:42:40 +0000] info [uapi] End STDERR from hook
    #
    # This is the reason we issue a log during startup
    logger.info("Starting imunify fileman hook")
    ctx = Context(
        sys.stdin,
        sys.stdout,
        sys.stderr,
        sys.argv,
        aibolit_checker,
    )
    code = main(ctx)
    ctx.stdout.flush()
    ctx.stderr.flush()
    logger.info("exiting with code %s", code)
    sys.exit(code)

CANCEL

Name	Type	Size	Modified	Actions
↩ ..	DIR	—	—
📄 cpanel_fileman_hook	?	12.2 KB	2026-08-11 10:41	EDIT