Edit File - File Manager

Edit File: redirect.py

import logging
from typing import Iterable, List

from defence360agent.contracts.config import (
    PORT_BLOCKING_MODE_ALLOW,
    PORT_BLOCKING_MODE_DENY,
)
from im360.contracts.config import Firewall
from im360.internals.core import (
    FirewallRules,
    IPSetCmdBuilder,
    is_nat_available,
)
from im360.internals.core.ipset import (
    IP_SET_PREFIX,
    AbstractIPSet,
    IPSetAtomicRestoreBase,
    IPSetCount,
    libipset,
)
from im360.internals.core.ipset.port_deny import InputPortBlockingDenyModeIPSet
from im360.subsys import webshield
from im360.subsys.webshield_mode import Mode as WebshieldMode
from im360.utils.net import ALL, TCP
from defence360agent.utils.validate import IPVersion
from im360.model.firewall import BlockedPort

logger = logging.getLogger(__name__)

class SingleIPSetNoRedirectPort(IPSetAtomicRestoreBase):
    _NAME = "{prefix}.{ip_version}.no-redirect-port"

def gen_ipset_create_ops(self, ip_version: IPVersion) -> List[str]:
        return [
            "create {name} bitmap:port range 0-65535 timeout 0 -exist".format(
                name=self.gen_ipset_name_for_ip_version(ip_version)
            )
        ]

def gen_ipset_destroy_ops(self, ip_version: IPVersion) -> List[str]:
        ipset_name = self.gen_ipset_name_for_ip_version(ip_version)
        return [IPSetCmdBuilder.get_destroy_cmd(ipset_name)]

def gen_ipset_flush_ops(self, ip_version: IPVersion) -> List[str]:
        return [
            IPSetCmdBuilder.get_flush_cmd(
                self.gen_ipset_name_for_ip_version(ip_version)
            )
        ]

async def gen_ipset_restore_ops(self, ip_version: IPVersion) -> List[str]:
        lines = []
        name = self.gen_ipset_name_for_ip_version(ip_version)
        current_mode = WebshieldMode.get()
        if Firewall.port_blocking_mode == PORT_BLOCKING_MODE_DENY:
            white_listed_ports = []
            for (
                port_or_port_range
            ) in InputPortBlockingDenyModeIPSet().get_config_option(
                # redirection to webshield rules work only for TCP,
                # so get conf value for TCP
                ip_version,
                TCP,
            ):
                if port_or_port_range.isdigit():
                    white_listed_ports.append(int(port_or_port_range))
                else:
                    white_listed_ports += [
                        p
                        for p in range(
                            *list(map(int, port_or_port_range.split("-")))
                        )
                    ]
            for port in sorted(
                webshield.redirected_to_webshield_ports(current_mode)
            ):
                if port not in white_listed_ports:
                    lines.append(
                        " ".join(
                            libipset.prepare_ipset_command("add", name, port)
                        )
                    )
        elif Firewall.port_blocking_mode == PORT_BLOCKING_MODE_ALLOW:
            black_listed_ports = []
            from im360.internals.core import IPSetPort

for item in IPSetPort()._fetch():  # noqa
                port, generic_proto = item
                if generic_proto in (TCP, ALL):
                    # webshield work by tcp proto
                    black_listed_ports.append(port)
            for port in sorted(
                webshield.redirected_to_webshield_ports(current_mode)
            ):
                if port in black_listed_ports:
                    lines.append(
                        " ".join(
                            libipset.prepare_ipset_command("add", name, port)
                        )
                    )
        else:
            raise NotImplementedError()
        return lines

def gen_ipset_name_for_ip_version(self, ip_version: IPVersion) -> str:
        return self.custom_ipset_name or self._NAME.format(
            prefix=IP_SET_PREFIX, ip_version=ip_version
        )

async def get_db_count(self, ip_version: IPVersion):
        return BlockedPort.fetch_count()

class IPSetNoRedirectPort(AbstractIPSet):
    def get_all_ipsets(self, ip_version: IPVersion):
        return frozenset(
            [
                ipset.gen_ipset_name_for_ip_version(ip_version)
                for ipset in self.get_all_ipset_instances(ip_version)
            ]
        )

def get_all_ipset_instances(
        self, ip_version: IPVersion
    ) -> List[IPSetAtomicRestoreBase]:
        return [SingleIPSetNoRedirectPort()]

def get_rules(self, ip_version: IPVersion, **kwargs) -> Iterable[dict]:
        # there is no any reason why we should do some redirection
        # in case port is blocked. if port is block it means
        # it's blocked without any exception, which can happend,
        # for example, because of some redirection on webshield
        # or anywhere

# see IPSetPort.block also
        return [
            dict(
                rule=FirewallRules.stop_redirection(
                    SingleIPSetNoRedirectPort().gen_ipset_name_for_ip_version(
                        ip_version
                    )
                ),
                table=FirewallRules.NAT
                if is_nat_available(ip_version)
                else FirewallRules.MANGLE,
                chain=FirewallRules.IMUNIFY_INPUT_CHAIN,
                position=1,
                priority=FirewallRules.HIGHEST_PRIORITY,
            )
        ]

async def restore(self, ip_version: IPVersion) -> None:
        ipset = SingleIPSetNoRedirectPort()
        name = ipset.gen_ipset_name_for_ip_version(ip_version)
        await libipset.flush_set(name)
        lines = await ipset.gen_ipset_restore_ops(ip_version)
        await libipset.restore(lines, name=name)

@classmethod
    def gen_ipset_create_ops(cls, ip_version: IPVersion) -> List[str]:
        return SingleIPSetNoRedirectPort().gen_ipset_create_ops(ip_version)

async def add_item(self, port, ip_version: IPVersion):
        ipset = SingleIPSetNoRedirectPort()
        await libipset.add_item(
            ipset.gen_ipset_name_for_ip_version(ip_version), port, timeout=0
        )

async def delete_item(self, port, ip_version: IPVersion):
        ipset = SingleIPSetNoRedirectPort()
        await libipset.delete_item(
            ipset.gen_ipset_name_for_ip_version(ip_version), port
        )

async def get_ipsets_count(self, ip_version: IPVersion) -> list:
        ipsets = []
        for ip_set in self.get_all_ipset_instances(ip_version):
            if ip_set.is_enabled():
                set_name = ip_set.gen_ipset_name_for_ip_version(ip_version)
                expected_count = await ip_set.get_db_count(ip_version)
                ipset_count = await libipset.get_ipset_count(set_name)
                ipsets.append(
                    IPSetCount(
                        name=set_name,
                        db_count=expected_count,
                        ipset_count=ipset_count,
                    )
                )
        return ipsets

class IPSetWebshieldPort(AbstractIPSet):
    """
    Used to insert chain to check access to the webshield ports.
    Only redirected and local connections are allowed.
    Rules to the chain should be added by the corresponding
    WebshieldEnabledIPSet.
    """

def get_all_ipsets(self, ip_version: IPVersion):
        return frozenset()

def get_all_ipset_instances(
        self, ip_version: IPVersion
    ) -> List[IPSetAtomicRestoreBase]:
        return []

def get_rules(self, ip_version: IPVersion, **kwargs):
        # insert rule at the top of IMUNIFY_INPUT_CHAIN
        current_mode = WebshieldMode.get()
        if not WebshieldMode.wants_redirect(current_mode):
            return []
        return [
            dict(
                rule=FirewallRules.block_dst_port_list(
                    webshield.destination_webshield_ports(),
                    policy=FirewallRules.WEBSHIELD_PORTS_INPUT_CHAIN,
                ),
                chain=FirewallRules.IMUNIFY_INPUT_CHAIN,
                table=FirewallRules.FILTER,
                priority=FirewallRules.HIGHEST_PRIORITY,
            )
        ]

async def restore(self, ip_version: IPVersion) -> None:
        pass

def gen_ipset_create_ops(self, ip_version: IPVersion):
        return []

async def get_ipsets_count(self, ip_version: IPVersion):
        return []

Back to File Manager