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/cli.py (9273B)

from collections import defaultdict
import json
import os
import subprocess
import sys
import time
import yaml

PRETTY_JSON_ARGS = {"sort_keys": True, "indent": 2, "separators": (",", ": ")}

EXITCODE_NOT_FOUND = 2
EXITCODE_WARNING = 3
EXITCODE_GENERAL_ERROR = 11

PAGERS = ["/bin/less", "/bin/more"]

SUCCESS, WARNING, ERROR = "success", "warnings", "error"  # see simple_rpc

_CLI_MSG_PREFIX = {WARNING: "WARNING", ERROR: "ERROR"}

EXIT_CODES = {
    SUCCESS: 0,
    WARNING: EXITCODE_WARNING,
    ERROR: EXITCODE_GENERAL_ERROR,
}

def pager(data):
    pager = os.environ.get(
        "PAGER", next((p for p in PAGERS if os.path.isfile(p)), None)
    )
    if pager is None:
        print(data)
    else:
        subprocess.run([pager], input=data.encode(), stdout=sys.stdout)

class TablePrinter:
    def __init__(self):
        self._headers = {}
        self._mappers = defaultdict(list)
        self._right_aligned = {}
        self._widths = {}

def set_field_properties(
        self,
        field,
        mappers=None,
        max_width=None,
        right_align=False,
        header=None,
    ):
        if mappers:
            self._mappers[field] = mappers
        if max_width:
            self._widths[field] = max_width
        self._right_aligned[field] = right_align
        self._headers[field] = header if header else field.upper()

def print(self, fields, items, file=sys.stdout):
        headers = [self._headers.get(field, field.upper()) for field in fields]
        widths = [len(field) for field in headers]
        rows = []
        for item in items:
            row = []
            for i, field in enumerate(fields):
                v = item.get(field)
                for mapper in self._mappers[field]:
                    v = mapper(v)
                v = str(v)
                if len(v) > widths[i]:
                    max_width = self._widths.get(field)
                    if max_width and len(v) > max_width:
                        v = v[: max_width - 3] + "..."
                    widths[i] = len(v)
                row.append(v)
            rows.append(row)
        print(self._format_row(headers, widths, False))
        for row in rows:
            print(
                self._format_row(
                    row, widths, self._right_aligned.get(field, False)
                )
            )

@staticmethod
    def _add_padding(value, width, right_align):
        if right_align:
            return value.rjust(width)
        return value.ljust(width)

@staticmethod
    def _format_row(columns, widths, right_aligned):
        cols = [
            TablePrinter._add_padding(value, widths[i], right_aligned)
            for i, value in enumerate(columns)
        ]
        return "  ".join(cols)

def n_a(value):
    return value if value is not None else "n/a"

def to_int(value):
    return int(value) if value is not None else value

def extract_field(field):
    def extractor(value):
        if isinstance(value, dict):
            return value.get(field)
        return value

return extractor

def print_table(data, field_props):
    table = TablePrinter()
    for props in field_props:
        table.set_field_properties(*props)
    table.print([item[0] for item in field_props], data)

def print_incidents(data):
    field_props = (
        ("timestamp", [to_int]),
        ("abuser", [n_a]),
        ("country", [extract_field("code")]),
        ("times", [n_a]),
        ("name", [n_a]),
        ("severity", [n_a]),
    )
    print_table(data, field_props)

def add_ttl(data):
    now = int(time.time())
    for item in data:
        expiration = item.get("expiration", 0)
        if expiration > 0:
            item["ttl"] = expiration - now
        else:
            item["ttl"] = 0

def print_graylist(data):
    add_ttl(data)
    field_props = (
        ("ip",),
        ("ttl",),
        ("country", [extract_field("code")]),
    )
    print_table(data, field_props)

def print_bwlist(data):
    add_ttl(data)
    field_props = (
        ("ip",),
        ("ttl",),
        ("country", [extract_field("code")]),
        ("imported_from",),
        ("comment",),
    )
    print_table(data, field_props)

def guess_printer(data):
    if isinstance(data, (list, tuple)):
        if len(data):
            printer = TablePrinter()
            if isinstance(data[0], dict):
                keys = sorted(data[0].keys())
                printer.set_field_properties(
                    "country", mappers=[extract_field("code")]
                )
                printer.print(keys, data)
            else:
                for item in data:
                    print(item)
    else:
        print(data)

def yaml_printer(data):
    if isinstance(data, str):
        print(data)
    else:
        print(yaml.dump(data, default_flow_style=False))

def json_printer(data):
    if isinstance(data, str):
        print(data)
    else:
        print(json.dumps(data))

def hook_printer(data):
    if isinstance(data, dict):
        print("Status: {}".format(data["status"]))
    else:
        result = []
        for hook in data:
            result.append(
                "Event: {}, Path: {}{}".format(
                    hook["event"],
                    hook["path"],
                    "  native" if hook["native"] else "",
                )
            )
        print("\n".join(result))

def waf_set_printer(items):
    if not items:
        print("No users targeted.")
        return
    counts = {"succeeded": 0, "skipped": 0, "failed": 0}
    for item in items:
        counts[item["status"]] += 1
    print(
        "{succeeded} succeeded, {skipped} skipped, {failed} failed.".format(
            **counts
        )
    )
    print()
    print_table(items, (("user",), ("status",), ("reason",)))

def waf_status_printer(result):
    header = "Global WAF: " + result.get("global_waf", "unknown")
    if not result.get("security_plugin_enabled", True):
        header += " (plugin off)"
    print(header)
    print(
        "Default (no override): " + result.get("global_waf_default", "unknown")
    )
    items = result.get("items") or []
    total = result.get("total_count", len(items))
    if len(items) < total:
        # Page or 500-cap truncated the list — make the gap explicit so a
        # human doesn't read the table as the complete set.
        print("Total accounts: {} (showing {})".format(total, len(items)))
    else:
        print("Total accounts: {}".format(total))
    print()
    if not items:
        print("No accounts.")
        return
    print_table(
        items,
        (("name",), ("waf_status",), ("source",), ("wp_sites",)),
    )

PRINTERS = {
    ("config", "show"): json_printer,
    ("eula", "show"): pager,
    ("get",): print_incidents,
    ("whitelist",): print_bwlist,
    ("whitelist", "ip", "list"): print_bwlist,
    ("blacklist",): print_bwlist,
    ("blacklist", "ip", "list"): print_bwlist,
    ("graylist",): print_graylist,
    ("graylist", "ip", "list"): print_graylist,
    ("malware", "on-demand", "status"): yaml_printer,
    ("feature-management", "defaults"): yaml_printer,
    ("feature-management", "show"): yaml_printer,
    ("feature-management", "enable"): yaml_printer,
    ("feature-management", "disable"): yaml_printer,
    ("feature-management", "get"): yaml_printer,
    ("hook", "add"): hook_printer,
    ("hook", "delete"): hook_printer,
    ("hook", "list"): hook_printer,
    ("hook", "add-native"): hook_printer,
    ("wordpress-plugin", "waf", "set"): waf_set_printer,
    ("wordpress-plugin", "waf", "status"): waf_status_printer,
}

# Printers that consume the full response dict (globals + items), not just
# result["items"] — needed for the header line the waf status table carries.
_FULL_RESULT_PRINTERS = {("wordpress-plugin", "waf", "status")}

def _get_default_output(result):
    return result["items"] if result.get("items") is not None else "OK"

def _print_json_response(result, is_verbose=False):
    pretty_args = PRETTY_JSON_ARGS if is_verbose else {}
    print(json.dumps(result, **pretty_args))

def _print_plain_response(method, result):
    """Print result in plain text format using appropriate printer."""
    print_fun = PRINTERS.get(method, guess_printer)
    if method in _FULL_RESULT_PRINTERS:
        print_fun(result)
    else:
        print_fun(_get_default_output(result))

def print_response(method, result, is_json=False, is_verbose=False):
    if is_json:
        _print_json_response(result, is_verbose)
    else:
        _print_plain_response(method, result)

def print_warnings(data: dict):
    if not isinstance(data, dict):
        # This can happen, for example, if validation of cli args fails
        return

for warning in data.get("warnings", []):
        print(warning, file=sys.stderr)

def print_error(
    result, messages, is_json=False, is_verbose=False, *, file=sys.stderr
):
    if is_json:
        pretty_args = PRETTY_JSON_ARGS if is_verbose else {}
        print(json.dumps({result: messages}, **pretty_args))
    else:
        if isinstance(messages, (list, tuple)):
            for msg in messages:
                print("%s: %s" % (_CLI_MSG_PREFIX[result], msg), file=file)
        else:
            print(messages, file=file)

cmd: