1#@!#!123s

: / opt / imunify360 / venv / lib64 / python3.11 / site-packages / im360 / model /
Filename : global_whitelist.py
back
"""
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 defence360agent.files import IntegrityError
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 = []  # type: List[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.extend(
                                cls._load_file(
                                    local_index.localfilepath(item["url"])
                                )
                            )
        except (FileNotFoundError, IntegrityError) as e:
            logger.error("%s", e)
            return []

        return 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:
                    ipaddress.ip_network(item)
                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