1#@!#!123s

: / proc / 3 / root / opt / support / lib /
Filename : firewall_tools.py
back
from typing import Literal
from pathlib import Path
import re
import subprocess
import netaddr
from output import err_exit, print_listed, print_fail2ban_listed, warn
from run_cmd import is_exe, which
from rads.color import yellow


def fw_info() -> tuple[
    Literal['APF', 'CSF', 'ipset+fail2ban'],
    Literal['/usr/local/sbin/apf', '/usr/sbin/csf', None],
    list[dict[str, str]] | str,
]:
    """Yields a tuple of fw_name, fw_command, fw_data.
    fw_name will be "APF", "CSF", or "ipset+fail2ban".

    If fw_name was "APF" or "CSF", fw_command will be the path to its exe.

    If fw_name was "APF" or "CSF", fw_data will be the contents of its deny
    file. Otherwise, fw_data will be a list of dicts containing "listname"
    and "ip".


    Returns:
        tuple[str, str | None, list[dict[str, str] | None]]: see above
    """
    deny_file = None
    fw_cmd = None
    name = ''
    if is_exe('/usr/local/sbin/apf'):
        fw_cmd = '/usr/local/sbin/apf'
        deny_file = Path('/etc/apf/deny_hosts.rules')
        name = 'APF'
    elif is_exe('/usr/sbin/csf'):
        fw_cmd = '/usr/sbin/csf'
        deny_file = Path('/etc/csf/csf.deny')
        name = 'CSF'
    elif is_exe('/usr/bin/fail2ban-client') and which('ipset'):
        name = 'ipset+fail2ban'
    else:
        err_exit('Cannot identify firewall')
    if deny_file is None:
        deny_data = list(read_ipset_save())
    else:
        try:
            deny_data = deny_file.read_text(encoding='utf-8')
        except FileNotFoundError:
            err_exit(f'Cannot read {deny_file}. Firewall is misconfigured.')
    return name, fw_cmd, deny_data


def read_ipset_save():
    irgx = re.compile(r'add (?P<listname>[a-zA-Z0-9\-_]+) (?P<ip>[0-9\./]+)$')
    with subprocess.Popen(
        ['ipset', 'save'],
        encoding='utf-8',
        stdout=subprocess.PIPE,
        universal_newlines=True,
    ) as proc:
        for line in proc.stdout:
            if match := irgx.match(line.rstrip()):
                yield match.groupdict()


def ipset_list_action(
    listname: str,
) -> Literal['ACCEPT', 'DROP', 'DENY', 'UNKNOWN']:
    """Check whether an ipset list is set to ACCEPT, DROP, or DENY"""
    set_re = re.compile(r'--match-set ([a-z0-9\-\_]+)')
    action_re = re.compile(r'-j ([A-Z]+)')
    iptables_file = Path('/etc/sysconfig/iptables')
    iptables = iptables_file.read_text(encoding='utf-8').splitlines()
    for line in iptables:
        if set_match := set_re.findall(line):
            if set_match[0] == listname:
                if action_match := action_re.findall(line):
                    action = action_match[0]
                    if action != 'DOCKER':
                        return action
    return 'UNKNOWN'


def get_jail_ports():
    jail_re = re.compile(r'--match-set ([a-z0-9\-]+)')
    ports_re = re.compile(r'--dports ([0-9\,]+)')
    iptables_file = Path('/etc/sysconfig/iptables')
    iptables = iptables_file.read_text(encoding='utf-8').splitlines()
    for line in iptables:
        if 'INPUT' in line and 'f2b' in line:
            jail_match = jail_re.findall(line)
            if not jail_match:
                continue

            jail_name = jail_match[0]
            if 'multiport' in line:
                port_match = ports_re.findall(line)
                if port_match:
                    ports = port_match[0]
                    yield jail_name, ports
            else:
                yield jail_name, 'ALL'


def ipset_fail2ban_check(
    fw_data: list[dict[str, str]], ipaddr: netaddr.IPAddress
) -> tuple[bool, str | None]:
    """Check deny_data ``fw_info()`` for an IP address. If found, return whether
    it's blocked and in what fail2ban list if it was automatically blocked

    Args:
        fw_data (list[dict[str, str]]): third arg returned by ``fw_info()``
        ipaddr (netaddr.IPAddress): IP address to check

    Returns:
        tuple[bool, str | None]]: if blocked and in what fail2ban list if any
    """
    jail_port_map = dict(get_jail_ports())
    within_lists = []
    within_jails = []
    list_name = None
    for tnet in fw_data:
        try:
            listed = ipaddr in netaddr.IPNetwork(tnet['ip'])
            if listed:
                list_name = tnet['listname']
                within_lists.append(list_name)
        except netaddr.AddrFormatError:
            continue

    if not within_lists:
        print_listed(ipaddr, False, 'any ipset or fail2ban list')
        return False, None
    listed = True

    for list_name in within_lists:
        list_action = ipset_list_action(list_name)
        print_listed(ipaddr, True, f'the {list_name} {list_action} list')
        if list_name.startswith('f2b-'):
            jail_name = list_name.replace('f2b-', '')
            jail_ports = jail_port_map.get(list_name, 'ALL')
            within_jails.append(jail_name)
            print_fail2ban_listed(jail_name, jail_ports)
        else:
            if list_action == 'ACCEPT':
                warn(
                    f'{ipaddr} is NOT BLOCKED. It is whitelisted.', color=yellow
                )
                return False, None

    return listed, within_jails


def check_iptables(ipaddr: netaddr.IPAddress) -> bool:
    """Search iptables -nL for a line containing an IP which does not start with
    ACCEPT"""
    try:
        fw_data = subprocess.check_output(['iptables', '-nL'], encoding='utf-8')
    except (OSError, subprocess.CalledProcessError):
        # stderr will print to tty
        err_exit('could not run iptables -nL')
    for line in fw_data.splitlines():
        if not line.startswith('ACCEPT') and str(ipaddr) in line:
            return True
    return False