22

✏️ Editing: global_whitelist.py

"""
Global whitelist as it was initially defined by
https://cloudlinux.atlassian.net/browse/DEF-2307
"""
import ipaddress
import logging
from typing import Iterable, List

from im360 import files

logger = logging.getLogger(__name__)

IM360_FILE_NAME = "imunify360.txt"

class GlobalWhitelist:
    @staticmethod
    def _allow_load_from_file(file_name):
        """Check if file file_name should be loaded.
        Do not load imunify360.txt as it will be loaded separately in
        GlobalImunifyWhitelist.
        """
        return file_name != IM360_FILE_NAME

@classmethod
    async def load(cls, group=None) -> List[str]:
        result = set[str]()
        try:
            async with files.Index.locked(files.WHITELISTS):
                local_index = files.Index(files.WHITELISTS)
                for item in local_index.items():
                    if group is None or (group in item.get("groups", [])):
                        if cls._allow_load_from_file(item["name"]):
                            result.update(
                                cls._load_file(
                                    local_index.localfilepath(item["url"])
                                )
                            )
        except Exception:
            logger.exception("Failed to load global whitelist")
            return []

return list(result)

@classmethod
    def _load_file(cls, file, *, log_error=None) -> Iterable[str]:
        """
        :param file: file path
        :param log_error: what logging method to use [default: logger.error]
        """
        if log_error is None:
            log_error = logger.error
        with open(file) as f:
            for ln, item in enumerate(f):
                item = item.partition("#")[0].strip()
                if not item:
                    # allow blank lines and comments (including end-of-line)
                    continue
                try:
                    ip = ipaddress.ip_network(item)
                    if ip.version == 4:
                        item = item.removesuffix("/32")
                except ValueError as e:
                    log_error(
                        "%s in %s line %d",
                        e,
                        file,
                        # base 1 for files line numbering
                        ln + 1,
                    )
                else:
                    yield item

class GlobalImunifyWhitelist(GlobalWhitelist):
    @staticmethod
    def _allow_load_from_file(name):
        return name == IM360_FILE_NAME

MR SKB