22

✏️ Editing: gen_ports_conf.py

#!/opt/imunify360/venv/bin/python3

import os
import re
import subprocess

PREFIX = '/etc/imunify360-webshield'
PORTS = os.path.join(PREFIX, 'ports.conf')
SSL_PORTS = os.path.join(PREFIX, 'ssl_ports.conf')
PRESETS = os.path.join(PREFIX, 'presets.cfg')
DA_CONFIG = '/usr/local/directadmin/conf/directadmin.conf'
HTTP_INCLUDES = os.path.join(PREFIX, 'webshield-http.conf.d')
IPV6_CHECK_PATH = '/sys/module/ipv6/parameters/disable'
RESOLV = '/etc/resolv.conf'
RESOLVER = os.path.join(HTTP_INCLUDES, 'resolver.conf')
IPV4_TITLE = '# IPv4\n'
IPV6_TITLE = '# IPv6\n'
IPV4_ONLY_TITLE = '# IPv4 only (IPv6 is disabled)\n'
IPV4_FMT = 'listen      *:{}{};\n'
IPV6_FMT = 'listen      [::]:{}{};\n'
IPV4_SSL_FMT = 'listen      *:{} ssl{};\n'
IPV6_SSL_FMT = 'listen      [::]:{} ssl{};\n'
RESOLVER_FMT = "resolver {}{};\n"

class BasePanel:

cmd = None
    ports = [52224]
    ssl_ports = [52223]

@classmethod
    def check(cls):
        if not cls.cmd:
            return False
        try:
            subprocess.check_call(
                cls.cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        except (FileNotFoundError, subprocess.CalledProcessError):
            return False
        return True

@classmethod
    def define_ports(cls):
        return cls.ports, cls.ssl_ports

class Cpanel(BasePanel):
    cmd = ('/usr/local/cpanel/cpanel', '-V')
    modularity_flag = '/usr/share/imunify360-webshield/modularity_mode'
    ports = [52224, 52228, 52230, 52232]
    apache_mode_ports = [52228, 52230, 52232]
    ssl_ports = [52223, 52227, 52229, 52231]
    apache_mode_ssl_ports = [52227, 52229, 52231]

@classmethod
    def define_ports(cls):
        """
        Redefinition of parent class method. When in 'apache' mode, all the HTTP/HTTPS
        traffic is supposed to be handled by apache and non-standard ports like 2082/2083
        are expected to be handled by webshield. So as there's no point for the webshield
        to listen on standard HTTP/HTTPS ports, we don't include them in the webshield ports
        config
        """
        try:
            with open(cls.modularity_flag) as f:
                mode = f.read().strip()
            if mode == 'apache':
                return cls.apache_mode_ports, cls.apache_mode_ssl_ports
            return cls.ports, cls.ssl_ports
        except Exception:
            return cls.ports, cls.ssl_ports

class Plesk(BasePanel):
    cmd = ('/usr/sbin/plesk', 'version')
    ports = [52224, 52234]
    ssl_ports = [52223, 52233]

class DirectAdmin(BasePanel):
    cmd = ('/usr/local/directadmin/custombuild/build', 'version')
    config = '/usr/local/directadmin/conf/directadmin.conf'
    patt = re.compile(r'SSL\s*=\s*(?P<ssl>1|0)', re.IGNORECASE)
    ports = [52224]
    ssl_ports = [52223]
    panel_ports = [52235]

@classmethod
    def _check_ssl(cls):
        with open(cls.config) as f:
            for line in f:
                if line.startswith('#'):
                    continue
                m = cls.patt.match(line)
                if m:
                    ssl = True if m.group('ssl') == '1' else False
                    return ssl
            return False

@classmethod
    def define_ports(cls):
        if cls._check_ssl():
            return cls.ports, cls.ssl_ports + cls.panel_ports
        return cls.ports + cls.panel_ports, cls.ssl_ports

def get_ports():
    for panel_cls in Cpanel, Plesk, DirectAdmin:
        if panel_cls.check():
            return panel_cls.define_ports()
        else:
            continue
    return BasePanel.define_ports()

def is_ipv6_on():
    """
    Checks if IPv6 is enabled on the host
    """
    try:
        with open(IPV6_CHECK_PATH) as p:
            val = p.read().strip()
        if val == "0":
            return True
        return False
    except Exception:
        return True

def is_proxy_enabled():
    """
    Checks if 'proxy_protocol' is enabled
    """
    try:
        with open(PRESETS) as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                if line.startswith('#'):
                    continue
                if '=' not in line:
                    continue
                key, value = [i.strip() for i in line.split('=', 1)]
                if key != 'proxy_protocol':
                    continue
                if value.lower() in ('yes', 'on', 'true'):
                    return True
        return False
    except Exception:
        return False

def write_ports(ipv6=True):
    """
    Writes IPv4/IPv6 ports configs for webshield
    """
    ports_list, ssl_ports_list = get_ports()
    proxy_on = is_proxy_enabled()
    title = IPV4_TITLE if ipv6 else IPV4_ONLY_TITLE
    for is_ssl, path, ports in (
            (False, PORTS, ports_list), (True, SSL_PORTS, ssl_ports_list)):
        with open(path, 'w') as w:
            w.write(title)
            fmt = IPV4_SSL_FMT if is_ssl else IPV4_FMT
            proto = ''
            if proxy_on:
                proto = ' proxy_protocol'
            for port in ports:
                w.write(fmt.format(port, proto))
            if ipv6:
                w.write(IPV6_TITLE)
                fmt = IPV6_SSL_FMT if is_ssl else IPV6_FMT
                for port in ports:
                    w.write(fmt.format(port, proto))
            if is_ssl:
                opts = [
                    '# Enable HTTP/2.',
                    'http2 on;',
                    '# Set timeouts for HTTP/2.',
                    'keepalive_timeout 5;',
                    'client_header_timeout 5;',
                ]
                w.write('\n'.join(opts) + '\n')

def write_resolver():
    """
    Writes resolver for webshield (based on /etc/resolv.conf content)
    """
    has_ipv6 = False
    if not os.path.isdir(HTTP_INCLUDES):
        return
    patt_v4 = re.compile(
        r'nameserver\s+(?P<ip>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})')
    patt_v6 = re.compile(
        r'nameserver\s+(?P<ip>(?:(?:[a-fA-F0-9]{1,4})?\:){2,7}(?:[a-fA-F0-9]{1,4})?)')
    ips = []
    try:
        with open(RESOLV) as f:
            for line in f:
                m = patt_v4.match(line)
                if m:
                    ip = m.group('ip')
                else:
                    m = patt_v6.match(line)
                    if m:
                        has_ipv6 = True
                        ip = '[' + m.group('ip') + ']'
                    else:
                        continue
                if ip not in ips:
                    ips.append(ip)
    except Exception:
        pass

ips_string = ' '.join(ips) if ips else '127.0.0.1'
    v6_part = '' if has_ipv6 else ' ipv6=off'

with open(RESOLVER, 'w') as f:
        f.write(RESOLVER_FMT.format(ips_string, v6_part))

if __name__ == '__main__':
    ipv6 = is_ipv6_on()
    write_ports(ipv6=ipv6)
    write_resolver()

MR SKB