:: **Zploit** v1.0 | Current Path: **/usr/libexec/imunify360/**
:: Editing File: 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 signal import socket import sys from tempfile import NamedTemporaryFile from typing import Callable, List, TextIO from defence360agent import sentry 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[[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 check_upload(ctx: Context) -> int: logger.info("upload action") suffix = "-key" path = "" filename = "" 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"] found_key = False for k, v in args.items(): if k.endswith(suffix): found_key = True path = args[k[: -len(suffix)]] # "file-eicar.com-key":"file-0" # [len("file-"):-len(suffix)] -> eicar.com filename = k[len("file-"):-len(suffix)] break if not found_key: logger.warning( "upload action: no '-key' suffix found in cPanel hook args; " "scan skipped (user '%s', dir '%s', available arg keys: %s)", _log_safe(data.get("user")), _log_safe(args.get("dir")), [_log_safe(k) for k in sorted(args.keys())], ) if path != "": allowed = ctx.checker(path) ctx.stdout.write( status_text( allowed, "upload", filename, args.get("dir"), data["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: with NamedTemporaryFile(mode="w") as ntf: ntf.write(content) ntf.flush() allowed = ctx.checker(ntf.name) 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(file_to_scan: 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 the uploaded file for scanning # create PID.upload_job in the resident dir aibolit_job.create_upload_job( files=[file_to_scan], resident_dir_path=resident_dir_path, timeout=remaining_time(), ) logger.info("file %s is sent for scanning", file_to_scan) # 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", file_to_scan) 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)