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))