Da

✏️ Editing: imunify360-webshield-compose-lists

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

import glob
import ipaddress
import json
import os
import random
import shutil
import string
import sys

DESCRIPTION_PATH = '/var/imunify360/files/whitelist/v2/description.json'
COMMON_PROXY_PATH = '/etc/imunify360-webshield/common-proxies.conf'
WHITELISTED_PATH = ('/etc/imunify360-webshield/webshield-http.conf.d'
                    '/static-whitelist.conf')
GEO_SRC_DIR = '/var/imunify360/files/geo/v1'
GEO_DST_CFG = '/etc/imunify360-webshield/country_ips.conf'
CUSTOM_WHITELIST_GLOB = '/etc/imunify360/whitelist/*.txt'
CUSTOM_WHITELIST_CONF = '/etc/imunify360-webshield/custom-whitelisted.conf'
CUSTOM_BLACKLIST_GLOB = '/etc/imunify360/blacklist/*.txt'
CUSTOM_BLACKLIST_CONF = '/etc/imunify360-webshield/custom-blacklisted.conf'
GEO_DST_CFG_HDR = """# THIS FILE IS GENERATED AUTOMATICALLY
# BY IMUNIFY360-WEBSHIELD. DO NOT MODIFY IT
"""

def subnet_valid(subnet):
    if not subnet:
        return False
    try:
        if '.' in subnet:
            ipaddress.IPv4Network(subnet, strict=False)
        else:
            ipaddress.IPv6Network(subnet, strict=False)
    except ipaddress.AddressValueError:
        return False
    return True

def get_config():
    """
    Reads JSON data from description.json and returns parsed dict
    :return: config -> dict
    """
    try:
        with open(DESCRIPTION_PATH) as i:
            return json.load(i)
    except Exception as e:
        return

def get_files(config):
    """
    Gets paths of IP list files. Depending on is the list is a proxy list
    or not.
    :param config: dict -> descriptions
    :return: tuple -> tuple of lists (proxies and whitelisted)
    """
    proxies = []
    whitelisted = []
    items = config.get('items')
    if not items:
        return
    for item in items:
        url = item.get('url')
        if not url:
            continue
        _dir = os.path.dirname(DESCRIPTION_PATH)
        _base = os.path.basename(url)
        path = os.path.join(_dir, _base)
        name = item.get('name', _base)
        if os.path.extsep in name:
            name, _ = os.path.splitext(name)
        if 'proxy' in item.get('groups', []):
            proxies.append((path, name))
    # whitelists are no more used so empty list returned
    return proxies, whitelisted

def get_ips_from_files(pairs):
    """
    Reads all filepaths and places its addresses into common list
    :param paths: list -> filepaths to be read
    :return: list -> list of IP addresses
    """
    ips = []
    for path, name in pairs:
        try:
            with open(path) as i:
                for line in i:
                    if line.startswith('#'):
                        continue
                    ip = line.strip()
                    if not ip:
                        continue
                    ips.append((ip, name))
        except Exception:
            continue
    return ips

def generate(length=8):
    """
    Generates random string or specified length
    :param length: int -> random string length
    :return: str -> generated random string
    """
    sample = string.digits + string.ascii_letters
    return ''.join(random.sample(sample, length))

def save(ips, path, wrap=False):
    """
    Saves IP lists as nginx configs
    :param ips: list -> list of IP addresses to be saved
    :param path: str -> path to config to generate
    :param wrap: boolean -> if the IPs in config are to be wrapped in nginx
                'geo' structure
    """
    temp = path + '.' + generate()
    tpl = '{} {};\n'
    try:
        with open(temp, 'w') as o:
            if wrap:
                o.write('geo $static_whitelisted {\n')
            if ips:
                for ip, token in dict(ips).items():
                    o.write(tpl.format(ip, token))
            if wrap:
                o.write('}\n')
    except Exception as e:
        return

shutil.move(temp, path)

def make_geo():
    """
    Generates mapping subnet -> country for webshield
    Update: as we moved blacklisted countries processing into WAFD,
    there's no more need to keep all countries IPs in the webshield.
    However we still need to keep China IP addresses because of
    splash_as_captcha functionality
    """
    if not os.path.isdir(GEO_SRC_DIR):
        return

temp = GEO_DST_CFG + '.' + generate()
    line_fmt = '{} {};\n'
    with open(temp, 'w') as w:
        w.write(GEO_DST_CFG_HDR)
        for name in os.listdir(GEO_SRC_DIR):
            if name.startswith('CountrySubnets-') and name.endswith('.txt'):
                code = name[15:17]
                if code != "CN":    # Keep China addresses only
                    continue
                full_path = os.path.join(GEO_SRC_DIR, name)
                try:
                    with open(full_path) as r:
                        for line in r:
                            stripped = line.strip()
                            if not stripped:
                                continue
                            w.write(line_fmt.format(stripped, code))
                except Exception:
                    continue

os.rename(temp, GEO_DST_CFG)

def make_custom():
    """
    Reads all IP addresses from custom directories and forms webshield configs
    from them
    """
    pairs = (
        (CUSTOM_WHITELIST_GLOB, CUSTOM_WHITELIST_CONF),
        (CUSTOM_BLACKLIST_GLOB, CUSTOM_BLACKLIST_CONF))

fmt = '{} 1;\n'

for read_glob, conf_path in pairs:
        read_paths = glob.glob(read_glob)
        if not read_paths:
            continue
        temp_path = conf_path + '.' + generate()
        try:
            with open(temp_path, 'w') as w:
                for read_path in read_paths:
                    with open(read_path) as r:
                        for line in r:
                            subnet = line.partition('#')[0].strip()
                            if not subnet:
                                # blank line / [full line/end-of-line] comment
                                continue
                            if not subnet_valid(subnet):
                                continue
                            w.write(fmt.format(subnet))
            os.rename(temp_path, conf_path)
        except Exception:
            continue

def main():
    """
    The main workflow routine. Read config, parse files, make lists, save them
    """
    make_geo()
    config = get_config()
    if config is None:
        save(None, WHITELISTED_PATH, True)
        return
    paths = get_files(config)
    if not paths or len(paths) != 2:
        print("Incomplete data. Return", file=sys.stderr)
        save(None, WHITELISTED_PATH, True)
        return
    proxies, whitelisted = paths
    proxies_ips = get_ips_from_files(proxies)
    whitelisted_ips = get_ips_from_files(whitelisted)
    destinations = ((proxies_ips, COMMON_PROXY_PATH, False),
                    (whitelisted_ips, WHITELISTED_PATH, True))
    for ips, path, wrap in destinations:
        save(ips, path, wrap)

def dispatcher():
    if len(sys.argv) > 1 and sys.argv[1] == '--custom-lists-only':
        make_custom()
    else:
        main()

if __name__ == '__main__':
    dispatcher()

📁 Sid Gifari File Manager