✏️ Editing: csf_imports.py Read Only

import logging
import pprint

from im360.internals import geo
from defence360agent.rpc_tools import lookup
from defence360agent.rpc_tools.utils import run_in_executor_decorator
from defence360agent.utils import Scope
from im360.api.ips import IgnoredByPortAPI, PortAPI
from im360.model.firewall import IPList
from im360.subsys import csf
from im360.subsys.panels.cpanel import cphulk
from defence360agent.utils.validate import IP
from im360.subsys import webshield

logger = logging.getLogger(__name__)

class CSFImportsEndpoints(lookup.RootEndpoints):
    SCOPE = Scope.IM360

@lookup.bind("import", "wblist")
    @run_in_executor_decorator
    def import_wblist(self):
        logger.info("Loading w/b lists from cPHulk")
        counter = 0
        with geo.reader() as geo_reader:
            for list_ in (IPList.WHITE, IPList.BLACK):
                for ip, comment in cphulk.ips_from_list(list_):
                    if not IP.is_valid_ip_network(ip):
                        logger.warning("Invalid IPv4 %s, skipping", ip)
                        continue
                    _, created = IPList.create_or_get(
                        ip=ip,
                        listname=list_,
                        imported_from="cPHulk",
                        comment=comment,
                        country=geo_reader.get_id(ip),
                    )
                    counter += int(created)
                    logger.info("Added {}".format(ip))
        return "Loaded {} ip addresses from cPHulk".format(counter)

@lookup.bind("import", "blocked-ports")
    async def import_blocked_ports(self, dry_run=True):
        port_protos = {}
        allowed_ip = csf.ignore_ports_from_file(csf.CSF_ALLOW_FILE)

captcha_ports = set(range(*webshield.port_range()))
        for proto in (csf.TCP, csf.UDP):
            # Skipping captcha ports
            for port in csf.closed_ports(proto) - captcha_ports:
                port_proto = (port, proto)
                port_protos[port_proto] = []
                for _port, _proto, ip, comment in allowed_ip:
                    if port == _port and proto == _proto:
                        port_protos[port_proto].append((ip, comment))

if dry_run:
            return pprint.pformat(port_protos)

affected, _ = await PortAPI.block(
            list(port_protos.keys()), comment="Imported from CSF"
        )

for port, proto in affected:
            for ip, comment in port_protos[(port, proto)]:
                await IgnoredByPortAPI.block(
                    [ip], port=port, proto=proto, comment=comment
                )
        return "Added {} blocked ports from CSF".format(len(affected))

🔐 Sid Gifari File Manager Pro