22

✏️ Editing: captcha.py

"""Captcha specific rules customization."""
from typing import AbstractSet, Iterator, Mapping

from im360.subsys.panels.hosting_panel import HostingPanel
from im360.contracts.config import Webshield
from im360.internals.core.firewall import FirewallRules, is_nat_available
from im360.subsys import webshield
from defence360agent.utils.validate import IPVersion, LocalhostIP

from .types_ import FirewallRule, WebshieldRuleBuilder

class CaptchaRuleBuilder(WebshieldRuleBuilder):
    """Implement steps specific to generating captcha related iptables
    rules.

"""

def __init__(self, include_webshield_ports_rules=False):  # type: ignore
        self.__include_webshield_ports_rules = include_webshield_ports_rules

def open_webshield_ports_for_localhost_rules(
        self,
        ip_version: IPVersion,
    ) -> Iterator[FirewallRule]:
        """Return FirewallRule to open webshields ports for localhost."""
        if not self.__include_webshield_ports_rules:
            return
        yield FirewallRule(
            chain=FirewallRules.WEBSHIELD_PORTS_INPUT_CHAIN,
            rule=FirewallRules.open_all_for_src_net(
                LocalhostIP[ip_version].value
            ),
        )

def block_webshield_ports_rules(
        self, redirect_map: Mapping[int, int], dest_ports: AbstractSet
    ) -> Iterator[FirewallRule]:
        """Return FirewallRules to log redirected traffic."""
        if not self.__include_webshield_ports_rules:
            return
        # insert rule to the bottom of WEBSHIELD_PORTS_INPUT_CHAIN
        yield FirewallRule(
            chain=FirewallRules.WEBSHIELD_PORTS_INPUT_CHAIN,
            rule=FirewallRules.compose_action(
                FirewallRules.LOG_BLOCK_PORT_CHAIN
            ),
            priority=FirewallRules.LOWEST_PRIORITY,
        )

def redirect_panel_ports(
        self, ip_version: IPVersion
    ) -> Iterator[FirewallRule]:
        """
        Used to create redirect to webshield for panel ports
        for example 2082, 2083 for cPanel.
        Turned on by config option panel_protection
        """
        if not self.__include_webshield_ports_rules:
            return
        ports = HostingPanel().get_webshield_protected_ports()
        if (
            ports
            and Webshield.PANEL_PROTECTION
            and Webshield.ENABLE
            and webshield.expects_traffic()
        ):
            for dst_port, target_port in ports.items():
                yield FirewallRule(
                    rule=FirewallRules.protected_by_webshield(
                        dst_port, target_port
                    ),
                    chain=FirewallRules.IMUNIFY_INPUT_CHAIN,
                    table=FirewallRules.NAT
                    if is_nat_available(ip_version)
                    else FirewallRules.MANGLE,
                    priority=FirewallRules.HIGHEST_PRIORITY,
                )
            yield FirewallRule(
                rule=FirewallRules.block_dst_port_list(
                    set(ports.values()),
                    policy=FirewallRules.ACCEPT,
                ),
                chain=FirewallRules.WEBSHIELD_PORTS_INPUT_CHAIN,
                table=FirewallRules.FILTER,
                priority=FirewallRules.HIGHEST_PRIORITY,
            )

def logdrop_chain_rules(self, ipset_name: str) -> Iterator[FirewallRule]:
        """Generator yield drop chain"""
        yield FirewallRule(
            rule=FirewallRules.ipset_rule(
                ipset_name, FirewallRules.LOG_GRAYLIST_CHAIN
            ),
        )

def drop_tproxy_rules(self, ipset_name: str) -> Iterator[FirewallRule]:
        """Generator yield firewall drop rule for tproxy"""
        yield FirewallRule(
            rule=FirewallRules.traffic_not_from_tproxy(ipset_name)
        )

MR SKB