[ Avaa Bypassed ]




Upload:

Command:

hmhc3928@3.17.187.254: ~ $
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()

Filemanager

Name Type Size Permission Actions
__pycache__ Folder 0755
__init__.py File 51.54 KB 0644
_shutil.py File 795 B 0644
antivirus_mode.py File 497 B 0644
benchmark.py File 576 B 0644
buffer.py File 1.24 KB 0644
check_db.py File 7.35 KB 0644
cli.py File 7.08 KB 0644
common.py File 14.41 KB 0644
config.py File 999 B 0644
cronjob.py File 902 B 0644
hyperscan.py File 149 B 0644
importer.py File 2.29 KB 0644
json.py File 953 B 0644
kwconfig.py File 1.56 KB 0644
parsers.py File 11.1 KB 0644
resource_limits.py File 2.29 KB 0644
safe_fileops.py File 7.96 KB 0644
safe_sequence.py File 363 B 0644
serialization.py File 1.72 KB 0644
subprocess.py File 1.53 KB 0644
support.py File 5.19 KB 0644
threads.py File 1005 B 0644
whmcs.py File 7.6 KB 0644
wordpress_mu_plugin.py File 2.47 KB 0644