File Manager

X7ROOT File Manager

Current Path: /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/utils

opt / imunify360 / venv / lib / python3.11 / site-packages / defence360agent / utils /

📁 ..
📄 __init__.py(52.55 KB)
📁 __pycache__
📄 _shutil.py(795 B)
📄 antivirus_mode.py(497 B)
📄 async_utils.py(718 B)
📄 benchmark.py(538 B)
📄 buffer.py(1.24 KB)
📄 check_db.py(7.35 KB)
📄 check_lock.py(636 B)
📄 cli.py(7.08 KB)
📄 common.py(14.41 KB)
📄 config.py(999 B)
📄 cronjob.py(902 B)
📄 doctor.py(1 KB)
📄 hyperscan.py(149 B)
📄 importer.py(2.67 KB)
📄 ipecho.py(1.9 KB)
📄 json.py(953 B)
📄 kwconfig.py(1.56 KB)
📄 parsers.py(11.12 KB)
📄 resource_limits.py(2.29 KB)
📄 safe_fileops.py(7.99 KB)
📄 safe_sequence.py(363 B)
📄 serialization.py(1.72 KB)
📄 sshutil.py(7.94 KB)
📄 subprocess.py(1.53 KB)
📄 support.py(5.2 KB)
📄 threads.py(1005 B)
📄 validate.py(4.27 KB)
📄 whmcs.py(7.6 KB)
📄 wordpress_mu_plugin.py(1.41 KB)

Editing: resource_limits.py

import asyncio
import logging

from enum import Enum
from os import fsdecode
from pathlib import Path
from typing import List

from defence360agent.utils import OsReleaseInfo

logger = logging.getLogger(__name__)

RUN_WITH_INTENSITY = "/usr/libexec/run-with-intensity"

LVECTL_BIN_PATH = Path("/usr/sbin/lvectl")
PROC_LVE_LIST_PATH = Path("/proc/lve/list")

class LimitsMethod(Enum):
    NICE = "nice"
    LVE = "lve"
    CGROUPS = "cgroups"

async def get_current_method() -> LimitsMethod:
    """Returns limit method, used in run-with-intensity tool."""
    proc = await asyncio.create_subprocess_exec(
        RUN_WITH_INTENSITY,
        "show",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

stdout, stderr = await proc.communicate()
    stdout = fsdecode(stdout).strip()

if stdout == "nice":
        return LimitsMethod.NICE
    if stdout == "lve":
        return LimitsMethod.LVE
    if stdout == "cgroups":
        return LimitsMethod.CGROUPS
    raise LookupError(
        "Parsing of used limitation method failed\nstdout: {}\nstderr: {}"
        .format(stdout, fsdecode(stderr).strip())
    )

async def create_subprocess(
    cmd: List[str],
    key: str,
    intensity_cpu: int,
    intensity_io: int,
    **subprocess_kwargs
) -> asyncio.subprocess.Process:
    """
    Creates asyncio.Process with limited resources (cpu & io),
    using run-with-intensity tool.

:param cmd: command to execute
    :param intensity_cpu: cpu intensity limit
    :param intensity_io: io intensity limit
    :param subprocess_kwargs: keyword arguments for create_subprocess_exec func
    :return: executed Process
    """
    limits_cmd = [
        RUN_WITH_INTENSITY,
        "run",
        "--intensity-cpu",
        str(intensity_cpu),
        "--intensity-io",
        str(intensity_io),
    ]
    limits_cmd.extend(["--key", key])

return await asyncio.create_subprocess_exec(
        *(limits_cmd + cmd), **subprocess_kwargs
    )

def is_lve_active() -> bool:
    """Checks that LVE-utils is active resource limiter."""
    # to avoid possible errors such as DEF-11941
    # make sure that OS is CL
    return PROC_LVE_LIST_PATH.exists() and OsReleaseInfo.is_cloudlinux()

def has_lvectl() -> bool:
    """Checks that LVE-utils is installed."""
    return LVECTL_BIN_PATH.exists()

X7ROOT File Manager

Editing: resource_limits.py

Upload File

Create Folder