22

✏️ Editing: cphulk.py

import json
import logging
import subprocess
from ipaddress import summarize_address_range, IPv4Address

from defence360agent.subsys.panels.cpanel import cPanel
from defence360agent.utils import run

WHMAPI = "/usr/sbin/whmapi1"

logger = logging.getLogger(__name__)

async def is_running():
    try:
        rc, *_ = await run(
            ["/usr/local/cpanel/scripts/restartsrv_cphulkd", "--status"]
        )
        return rc == 0
    except FileNotFoundError:
        return False

def cphulk_parse_ip(range_str, comment):
    parsed_ips = []
    ips = range_str.split("-")
    if len(ips) == 1:
        return [(ips[0], comment)]
    try:
        parsed_ips = [
            (str(cidr), comment)
            for cidr in summarize_address_range(
                IPv4Address(ips[0]), IPv4Address(ips[1])
            )
        ]
    except ValueError as e:
        logger.warning("Got invalid IP error: %s", str(e))
    finally:
        return parsed_ips

def ips_from_list(list_name):
    """
    Obtain white/black lists from cphulk using cpanel api
        # whmapi1 read_cphulk_records list_name=black --output=json
    :param list_name: name of the list to obtain from
    :return: list of ips and networks
    """
    list_name = list_name.lower()
    ips = []

if not cPanel.is_installed():
        return ips

logger.info("Loading from chpulk %s list", list_name)
    try:
        try:
            cp = subprocess.run(
                [
                    WHMAPI,
                    "read_cphulk_records",
                    "--output=json",
                    "list_name=" + list_name,
                ],
                stdout=subprocess.PIPE,
                check=True,
                timeout=20,
            )
        except OSError:
            logger.info("whmapi not found, skipping import from cphulk")
            return []

try:
            output = json.loads(cp.stdout.decode())
            if not output["metadata"]["result"]:
                logger.warning(
                    "Got error from whmapi1: %s", output["metadata"]["reason"]
                )
            else:
                for rng, comment in output["data"]["ips_in_list"].items():
                    ips.extend(cphulk_parse_ip(rng, comment))
        except (json.JSONDecodeError, KeyError):
            logger.error("Broken output from whmapi1: %s", cp.stdout)

except Exception:
        logger.exception("Failed to obtain ip list from cphulk")

return ips

MR SKB