FM / utils

/opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/utils/

/opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/utils

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
antivirus_mode.py	497	0644	edit dl rm
async_utils.py	718	0644	edit dl rm
benchmark.py	538	0644	edit dl rm
buffer.py	1945	0644	edit dl rm
check_db.py	7988	0644	edit dl rm
check_lock.py	856	0644	edit dl rm
cli.py	9273	0644	edit dl rm
common.py	14757	0644	edit dl rm
completions.py	9988	0644	edit dl rm
config.py	1695	0644	edit dl rm
cronjob.py	902	0644	edit dl rm
doctor.py	5503	0644	edit dl rm
fd_ops.py	7573	0644	edit dl rm
hyperscan.py	149	0644	edit dl rm
importer.py	2722	0644	edit dl rm
ipecho.py	3247	0644	edit dl rm
json.py	953	0644	edit dl rm
kwconfig.py	1756	0644	edit dl rm
net.py	595	0644	edit dl rm
net_transport.py	16289	0644	edit dl rm
parsers.py	11879	0644	edit dl rm
resource_limits.py	5158	0644	edit dl rm
safe_fileops.py	13672	0644	edit dl rm
safe_sequence.py	363	0644	edit dl rm
serialization.py	2410	0644	edit dl rm
sshutil.py	15174	0644	edit dl rm
subprocess.py	1570	0644	edit dl rm
support.py	5329	0644	edit dl rm
tenant_path.py	4456	0644	edit dl rm
threads.py	1005	0644	edit dl rm
validate.py	4375	0644	edit dl rm
whmcs.py	7784	0644	edit dl rm
wordpress_mu_plugin.py	1440	0644	edit dl rm
zipsafe.py	720	0644	edit dl rm
_shutil.py	4019	0644	edit dl rm
__init__.py	69706	0644	edit dl rm

Edit: /opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/utils/safe_fileops.py (13672B)

import asyncio
import atexit
import errno
import functools
import logging
import os
import pathlib
import pwd
import shutil
import stat
from concurrent.futures import ProcessPoolExecutor
from contextlib import contextmanager, suppress
from itertools import chain
from typing import Set, Tuple, Union

from defence360agent import utils

R_FLAGS = os.O_RDONLY
W_FLAGS = os.O_TRUNC | os.O_CREAT | os.O_WRONLY
IN_PLACE_W_FLAGS = os.O_CREAT | os.O_WRONLY

logger = logging.getLogger(__name__)

# Track active ProcessPoolExecutors so they can be cleaned up during shutdown.
# Each call to drop() permanently changes the worker process identity, so we
# must use a fresh executor per call. We track them to prevent orphaned worker
# processes from blocking agent shutdown (causing systemd SIGKILL).
_active_pools: Set[ProcessPoolExecutor] = set()

async def _run_in_fresh_executor(loop: asyncio.AbstractEventLoop, *args):
    pool = ProcessPoolExecutor(max_workers=1)
    _active_pools.add(pool)
    try:
        return await loop.run_in_executor(pool, *args)
    finally:
        try:
            pool.shutdown(wait=False)
        finally:
            _active_pools.discard(pool)

def shutdown_process_pools() -> None:
    """Shutdown all tracked ProcessPoolExecutors.

Should be called during agent shutdown to ensure clean process termination.
    """
    for pool in list(_active_pools):
        try:
            pool.shutdown(wait=False, cancel_futures=True)
        except Exception as e:
            logger.warning("Error shutting down ProcessPoolExecutor: %s", e)
    _active_pools.clear()

# Register cleanup at exit as a fallback
atexit.register(shutdown_process_pools)

def drop(fun, uid, gid, *args):
    os.setgroups([])
    os.setgid(gid)
    os.setuid(uid)
    return fun(*args)

class UnsafeFileOperation(Exception):
    pass

def ensure_regular_file(path: str) -> None:
    """Verify path is a regular file; remove and raise FileNotFoundError if not.

Uses os.lstat() to avoid following symlinks. If the file is a FIFO,
    symlink, socket, device, etc., it is deleted so the caller can
    recreate it as a regular file.
    """
    st = os.lstat(path)  # raises FileNotFoundError if missing
    if not stat.S_ISREG(st.st_mode):
        logger.warning(
            "Identity file %s is not a regular file (mode=%s), removing",
            path,
            stat.filemode(st.st_mode),
        )
        os.unlink(path)
        raise FileNotFoundError(f"Removed non-regular identity file: {path}")

def check_non_admin_file(file):
    st = os.stat(str(file))
    if st.st_uid < utils.get_min_uid():
        raise UnsafeFileOperation(
            "The file belongs to admin user: " + str(file)
        )
    return True

def safe(missing_ok=False):
    def _safe(fun):
        @functools.wraps(fun)
        async def wrapper(filename, *args, loop=None):
            if not os.path.exists(filename) and not missing_ok:
                raise FileNotFoundError(
                    "No such file or directory: " + filename
                )
            path = pathlib.Path(filename)
            paths = chain(reversed(path.parents), [path])
            if missing_ok:
                paths = reversed(path.parents)
            for p in paths:
                st = os.stat(str(p))
                if st.st_uid != 0 and st.st_gid != 0:
                    uid, gid = st.st_uid, st.st_gid
                    break
            else:
                raise UnsafeFileOperation(
                    "Unsafe file operation under root: " + str(path)
                )

loop = loop or asyncio.get_event_loop()

return await _run_in_fresh_executor(
                loop,
                drop,
                fun,
                uid,
                gid,
                filename,
                *args,
            )

return wrapper

return _safe

def _touch(filename: str):
    pathlib.Path(filename).touch()

def _write_text(filename: str, data: str):
    pathlib.Path(filename).write_text(data)

# This is the only way to make _write_text and _touch pickable.
# If we use decorator syntax instead - it's impossible
# to use them in multiprocessing
async def write_text(filename: str, data: str):
    return await safe(missing_ok=True)(_write_text)(filename, data)

async def touch(filename: str):
    return await safe(missing_ok=True)(_touch)(filename)

chmod = safe(os.chmod)
unlink = safe(os.unlink)

@contextmanager
def safe_open_file(filename, mode, user, respect_homedir=True):
    if "w" in mode:
        raise UnsafeFileOperation("'w' mode is not permitted")
    with open(filename, mode) as f:
        st = os.fstat(f.fileno())
        passwd = pwd.getpwnam(user)
        real_path = os.readlink(f"/proc/self/fd/{f.fileno()}")
        filename_str = str(filename)

# Checking if no symlinks along the pathway...
        # Unfortunately, that is going to fail for hosters that mapped
        # /home dir to be e.g.
        # /home -> /mnt/sdb1/home
        if (filename_str != real_path) or (st.st_uid != passwd.pw_uid):
            raise UnsafeFileOperation(f"Unable to safely read {filename_str}")

if (
            respect_homedir
            and pathlib.Path(passwd.pw_dir)
            not in pathlib.Path(filename_str).parents
        ):
            raise UnsafeFileOperation(
                f"Unable to safely read {filename_str}. "
                "File is not in user homedir"
            )
        yield f

@contextmanager
def open_fd(*args, **kwargs):
    """
    Context manager which wraps os.open and close file descriptor at the end

:param args: positional arguments for os.open
    :param kwargs: keyword arguments for os.open
    """
    fd = os.open(*args, **kwargs)
    try:
        yield fd
    finally:
        with suppress(OSError):  # fd is already closed
            os.close(fd)

@contextmanager
def opendir_fd(name: str, *args, **kwargs):
    """
    Context manager to get a directory file descriptor
    It also checks if a directory doesn't contain a symlink in the path

:param name: full directory name
    :param args: positional arguments for os.open
    :param kwargs: keyword arguments for os.open
    """
    with open_fd(name, *args, flags=os.O_DIRECTORY, **kwargs) as dir_fd:
        real = os.readlink("/proc/self/fd/{}".format(dir_fd))
        if name != real:
            raise UnsafeFileOperation("Operations on symlinks are prohibited")
        yield dir_fd

@contextmanager
def open_fobj(f: Union[str, int], dir_fd=None, flags=0, mode=None):
    """
    Context manager to open file object from file name or from file descriptor
    File object extended with 'st' attribute that contains os.stat_result of
    the opened file

:param f: file name or file descriptor to open
    :param dir_fd: directory descriptor, ignored if 'f' is a file descriptor
    :param flags: flags for os.open, ignored if 'f' is a file descriptor
    :param mode: mode for built-in open
    """

st = None
    if isinstance(f, str):
        # safe_* == False
        with suppress(OSError):
            # make a file readable/writable by an owner
            st = os.stat(f, dir_fd=dir_fd)
            os.chmod(
                f, mode=st.st_mode | stat.S_IRUSR | stat.S_IWUSR, dir_fd=dir_fd
            )

f = os.open(f, flags=flags, dir_fd=dir_fd)

with open(f, mode=mode) as fo:
        fo.st = st or os.stat(f)
        try:
            yield fo
        finally:
            if st:
                # revert file permissions
                with suppress(OSError):
                    os.chmod(f, mode=st.st_mode)

@contextmanager
def safe_tuple(name: str, dir_fd: int, flags: int, is_safe: bool):
    """
    If is_safe flag is True, open file descriptor using name and dir_fd
    If is_safe is False, return name and dir_fd as is
    """
    if is_safe:
        with open_fd(name, dir_fd=dir_fd, flags=flags) as fd:
            yield fd, None
    else:
        yield name, dir_fd

def _move(
    src: Union[Tuple[str, int], Tuple[int, None]],
    dst: Union[Tuple[str, int], Tuple[int, None]],
    src_unlink,
    dst_overwrite,
    racecall,
):
    src_f, src_dir_fd = src
    dst_f, dst_dir_fd = dst

w_flags = W_FLAGS | (0 if dst_overwrite else os.O_EXCL)

with open_fobj(
        src_f, dir_fd=src_dir_fd, flags=R_FLAGS, mode="rb"
    ) as src_fo:
        with open_fobj(
            dst_f, dir_fd=dst_dir_fd, flags=w_flags, mode="wb"
        ) as dst_fo:
            if racecall:
                racecall[0]()
            shutil.copyfileobj(src_fo, dst_fo)

if isinstance(dst_f, str):
                # safe_dst == False
                os.chmod(dst_fo.fileno(), mode=src_fo.st.st_mode)

if src_unlink and isinstance(src_f, str):
            # safe_src == False
            if racecall:
                racecall[1]()
            os.unlink(src_f, dir_fd=src_dir_fd)

async def safe_move(
    src: str,
    dst: str,
    safe_src=False,
    safe_dst=False,
    src_unlink=True,
    dst_overwrite=False,
    racecall=None,
):
    src_dir, src_name = os.path.split(src)
    dst_dir, dst_name = os.path.split(dst)

with opendir_fd(src_dir) as src_dir_fd, opendir_fd(
        dst_dir
    ) as dst_dir_fd, safe_tuple(
        src_name, src_dir_fd, R_FLAGS, safe_src
    ) as src_tuple, safe_tuple(
        dst_name, dst_dir_fd, W_FLAGS, safe_dst
    ) as dst_tuple:
        src_st = os.stat(src_name, dir_fd=src_dir_fd)

loop = asyncio.get_event_loop()
        await _run_in_fresh_executor(
            loop,
            drop,
            _move,
            src_st.st_uid,
            src_st.st_gid,
            src_tuple,
            dst_tuple,
            src_unlink,
            dst_overwrite,
            racecall,
        )

if src_unlink and safe_src:
            if racecall:
                racecall[1]()
            os.unlink(src_name, dir_fd=src_dir_fd)

if safe_dst:
            os.chown(dst_name, src_st.st_uid, src_st.st_gid, dir_fd=dst_dir_fd)
            os.chmod(dst_name, src_st.st_mode, dir_fd=dst_dir_fd)

def _check_room(src_size: int, dst_occupied: int, dst_dir_fd: int) -> None:
    vfs = os.statvfs(dst_dir_fd)
    # a sparse or compressed destination frees what it occupies, not its size
    frsize = vfs.f_frsize or vfs.f_bsize or 4096
    needed = max(0, (src_size + frsize - 1) // frsize * frsize - dst_occupied)
    available = vfs.f_bavail * frsize
    if available < needed:
        raise OSError(
            errno.ENOSPC,
            "{} bytes needed, {} available".format(needed, available),
        )

def _copy_in_place(
    src: Union[Tuple[str, int], Tuple[int, None]],
    dst: Union[Tuple[str, int], Tuple[int, None]],
):
    src_f, src_dir_fd = src
    dst_f, dst_dir_fd = dst

with open_fobj(
        src_f, dir_fd=src_dir_fd, flags=R_FLAGS, mode="rb"
    ) as src_fo:
        with open_fobj(
            dst_f, dir_fd=dst_dir_fd, flags=IN_PLACE_W_FLAGS, mode="wb"
        ) as dst_fo:
            shutil.copyfileobj(src_fo, dst_fo)
            dst_fo.flush()
            written = dst_fo.tell()
            if written != src_fo.st.st_size:
                raise OSError(
                    errno.EIO,
                    "copied {} of {} bytes".format(written, src_fo.st.st_size),
                )
            os.ftruncate(dst_fo.fileno(), written)

if isinstance(dst_f, str):
                # safe_dst == False; open_fobj puts back the mode of a
                # destination that was already there, so this dresses only
                # one we had to create
                os.chmod(dst_fo.fileno(), mode=src_fo.st.st_mode)

async def safe_copy_in_place(
    src: str,
    dst: str,
    safe_src=False,
    safe_dst=False,
):
    """Copy src over dst without emptying it first: the destination keeps its
    inode, and a copy that fails leaves the bytes written so far followed by
    the tail of the previous content rather than an empty file.
    """
    src_dir, src_name = os.path.split(src)
    dst_dir, dst_name = os.path.split(dst)

with opendir_fd(src_dir) as src_dir_fd, opendir_fd(dst_dir) as dst_dir_fd:
        src_st = os.stat(src_name, dir_fd=src_dir_fd)
        dst_occupied = 0
        dst_existed = True
        try:
            dst_st = os.stat(dst_name, dir_fd=dst_dir_fd)
        except FileNotFoundError:
            dst_existed = False
        else:
            dst_occupied = dst_st.st_blocks * 512
        # before opening the destination: opening it creates one we then keep
        _check_room(src_st.st_size, dst_occupied, dst_dir_fd)

with safe_tuple(
            src_name, src_dir_fd, R_FLAGS, safe_src
        ) as src_tuple, safe_tuple(
            dst_name, dst_dir_fd, IN_PLACE_W_FLAGS, safe_dst
        ) as dst_tuple:
            loop = asyncio.get_event_loop()
            try:
                await _run_in_fresh_executor(
                    loop,
                    drop,
                    _copy_in_place,
                    src_st.st_uid,
                    src_st.st_gid,
                    src_tuple,
                    dst_tuple,
                )
            except BaseException:
                if not dst_existed:
                    with suppress(OSError):
                        if not os.stat(dst_name, dir_fd=dst_dir_fd).st_size:
                            os.unlink(dst_name, dir_fd=dst_dir_fd)
                raise

if safe_dst:
                os.chown(
                    dst_name, src_st.st_uid, src_st.st_gid, dir_fd=dst_dir_fd
                )
                os.chmod(dst_name, src_st.st_mode, dir_fd=dst_dir_fd)

cmd: