golden hour

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

File/Folder	Size	Actions
__init__.py	68.07 KB
__pycache__	-
_shutil.py	3.92 KB
antivirus_mode.py	497 B
async_utils.py	718 B
benchmark.py	538 B
buffer.py	1.9 KB
check_db.py	7.8 KB
check_lock.py	856 B
cli.py	9.06 KB
common.py	14.41 KB
completions.py	9.75 KB
config.py	1.66 KB
cronjob.py	902 B
doctor.py	5.37 KB
fd_ops.py	7.4 KB
hyperscan.py	149 B
importer.py	2.66 KB
ipecho.py	3.17 KB
json.py	953 B
kwconfig.py	1.71 KB
net.py	595 B
net_transport.py	15.91 KB
parsers.py	11.6 KB
resource_limits.py	5.04 KB
safe_fileops.py	13.35 KB
safe_sequence.py	363 B
serialization.py	2.35 KB
sshutil.py	14.82 KB
subprocess.py	1.53 KB
support.py	5.2 KB
tenant_path.py	4.35 KB
threads.py	1005 B
validate.py	4.27 KB
whmcs.py	7.6 KB
wordpress_mu_plugin.py	1.41 KB
zipsafe.py	720 B

Edit: cli.py

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)