MAINHACK

Viewing / Editing File: config_provider.py

File is not writable. Editing disabled.
import json
import logging
import os
import pwd
from abc import abstractmethod
from contextlib import suppress
from textwrap import dedent
from typing import Mapping, Optional, Protocol

import sentry_sdk
import yaml

from defence360agent.utils import atomic_rewrite
from defence360agent.utils.fd_ops import open_dir_no_symlinks

logger = logging.getLogger(__name__)

# Don't read config if its file is larger than this.
_MAX_CONFIG_SIZE = 1 << 20  # 1MiB


class IConfigProvider(Protocol):
    @abstractmethod
    def read_config_file(
        self, force_read: bool = False, ignore_errors: bool = True
    ):
        raise NotImplementedError

    @abstractmethod
    def write_config_file(self, config: Mapping) -> None:
        raise NotImplementedError

    @abstractmethod
    def modified_since(self, timestamp: Optional[float]) -> bool:
        raise NotImplementedError


class ConfigError(Exception):
    pass


class JsonMessage:
    """Pretty-print given *obj* as JSON.

    To be used for logging. Example:

      logging.info("object: %s", JsonMessage(obj))

    """

    def __init__(self, obj):
        self._obj = obj

    def __str__(self):
        return json.dumps(self._obj, sort_keys=True)


def diff_section(prev_section: Optional[dict], section: Optional[dict]):
    """Return difference between config sections."""
    prev_section = prev_section or {}
    section = section or {}
    removed_settings = prev_section.keys() - section.keys()
    added_settings = section.keys() - prev_section.keys()
    return {
        "-": {v: prev_section[v] for v in removed_settings},
        "+": {v: section[v] for v in added_settings},
        # modified settings
        "?": {
            v: (prev_section[v], section[v])
            for v in (prev_section.keys() & section.keys())
            if prev_section[v] != section[v]
        },
    }


def diff_config(prev_conf: dict, conf: dict):
    """Compare *prev_conf* with the current *conf*."""
    removed_sections = prev_conf.keys() - conf.keys()
    yield {section: prev_conf[section] for section in removed_sections}
    added_sections = conf.keys() - prev_conf.keys()
    yield {section: conf[section] for section in added_sections}
    # changed sections
    yield {
        section: diff_section(prev_conf[section], conf[section])
        for section in (prev_conf.keys() & conf.keys())
        if prev_conf[section] != conf[section]
    }


def exclude_equals(*, main_conf: dict, base_conf: dict) -> dict:
    """
    Return dict derived from *main_conf* excluding parts
    that are equal in *base_conf*.
    For example,
    >>> base_conf = {
        "SECTION1": {"OPTION1": "default", "OPTION2": "default"},
        "SECTION2": {"OPTION1": "default"}
    }
    >>> main_conf = {
        "SECTION1": {"OPTION1": "value", "OPTION2": "default"},
        "SECTION2": {"OPTION1": "default"}
    }
    >>>
    >>> exclude_equals(main_conf=main_conf, base_conf=base_conf)
    {'SECTION1': {'OPTION1': 'value'}}
    >>>
    """
    _, added, changed = diff_config(base_conf, main_conf)
    result = {}
    for section, value in main_conf.items():
        if section in added.keys():
            result[section] = value
        if section in changed.keys():
            result.setdefault(section, {}).update(changed[section]["+"])
            result.setdefault(section, {}).update(
                {k: v[1] for k, v in changed[section]["?"].items()}
            )
    return result


class ConfigReader:
    """
    ConfigFile file for settings page.
    Location config file is PATH
    """

    def __init__(self, path, disclaimer="", permissions=None):
        self.path = path
        self.disclaimer = disclaimer
        self.permissions = permissions

    def __repr__(self):
        return "<{classname}({path})>".format(
            classname=self.__class__.__qualname__, path=self.path
        )

    def __str__(self):
        return f"ConfigReader at {self.path}"

    def read_config_file(
        self, force_read: bool = False, ignore_errors: bool = True
    ) -> dict:
        """Read config file into memory.

        Raises ConfigError.
        """
        try:
            if os.path.getsize(self.path) > _MAX_CONFIG_SIZE:
                raise ConfigError("Config file is too large")
            filename = self.path
            with open(filename, "r") as config_file:
                logger.info("Reading config file %s", filename)
                text = config_file.read()
        except UnicodeDecodeError as e:
            raise ConfigError("Unable to decode config file") from e
        except FileNotFoundError:
            return {}
        try:
            return self.load_config_body(text)
        except ConfigError as e:
            logger.error(e)
            if ignore_errors:
                return {}
            raise e

    def load_config_body(self, text: str) -> dict:
        try:
            config = yaml.safe_load(text)
        except yaml.YAMLError as e:
            raise ConfigError(
                f"Imunify360 config is not valid YAML document ({e})"
            ) from e

        if config is None:
            return {}

        if not isinstance(config, dict):
            raise ConfigError(
                "Imunify360 config is invalid or empty"
                ": path={!r}, text={!r}".format(self.path, text)
            )

        return config

    def _pre_write(self):
        pass

    def _post_write(self):
        pass

    def _serialize_config(self, config) -> str:
        config_text = ""
        if self.disclaimer:
            config_text += dedent(self.disclaimer)
            config_text += "\n"
        config_text += yaml.dump(config, default_flow_style=False)
        return config_text

    def write_config_file(self, config) -> str:
        self._pre_write()
        config_text = self._serialize_config(config)
        atomic_rewrite(
            self.path, config_text, backup=False, permissions=self.permissions
        )
        self._post_write()
        return config_text

    def modified_since(self, timestamp: Optional[float]) -> bool:
        return True


class CachedConfigReader(ConfigReader):
    def __init__(self, path, disclaimer="", permissions=None):
        super().__init__(path, disclaimer)
        self.mtime: Optional[float] = None
        self.size: Optional[float] = None
        self._config = {}
        self.permissions = permissions

    def __str__(self):
        return (
            "{classname} <'{path}', modified at {mtime}, {size} bytes>".format(
                classname=self.__class__.__qualname__,
                path=self.path,
                mtime=self.mtime,
                size=self.size,
            )
        )

    def read_config_file(
        self, force_read: bool = False, ignore_errors: bool = True
    ):
        """Update config if config file is modified"""
        if self.modified_since(self.mtime) or force_read:
            prev_config = self._config
            try:
                self._config = super().read_config_file(
                    ignore_errors=ignore_errors
                )
            except ConfigError as error:
                sentry_sdk.capture_exception(error)

                logger.warning(
                    "%s is invalid, using previous settings: %s",
                    self,
                    JsonMessage(self._config),
                )
                if not ignore_errors:
                    raise error
            else:
                if self.mtime is not None:  # don't log on startup
                    diffs = list(diff_config(prev_config, self._config))
                    if any(diffs):
                        # content has changed, log it
                        logger.info(
                            "%s modified: removed=%s, added=%s, changed=%s",
                            self,
                            *map(JsonMessage, diffs),
                        )

            self._refresh_stat_cache()

        return self._config

    def _refresh_stat_cache(self) -> None:
        """Sync cached mtime/size with the file on disk."""
        try:
            stat = os.stat(self.path)
            self.mtime = stat.st_mtime
            self.size = stat.st_size
        except FileNotFoundError:
            self.mtime = 0.0
            self.size = 0.0

    def modified_since(self, timestamp: Optional[float]) -> bool:
        """Whether the config has updated since *timestamp*.

        (as defined by its last modification time and size)
        :param timestamp: None means that the file has never been read before
        """
        # On startup consider timestamp to be None
        if timestamp is None:
            timestamp = 0.0
        try:
            stat = os.stat(self.path)
        except FileNotFoundError:
            st_mtime, st_size = 0.0, 0.0
        else:
            st_mtime, st_size = stat.st_mtime, stat.st_size
        return st_mtime > timestamp or st_size != self.size


class WriteOnlyConfigReader(CachedConfigReader):
    def __init__(self, path, disclaimer="", permissions=None):
        super().__init__(path, disclaimer, permissions)
        # write-only readers never hit the parent read path that populates
        # mtime/size, so seed them from disk now — otherwise the size
        # fallback in modified_since() (st_size != self.size) compares
        # against None forever and the check is stuck at True.
        self._refresh_stat_cache()

    def read_config_file(self, *_, **__):
        return self._config

    def write_config_file(self, config):
        config_text = super().write_config_file(config)
        self._config = self.load_config_body(config_text)
        self._refresh_stat_cache()
        return config_text


class UserConfigReader(CachedConfigReader):
    """Per-user config reader that resists TOCTOU symlink attacks.

    The user-specific subdirectory ``<USER_CONFDIR>/<username>/`` and the
    config file inside it must end up owned by ``root:<user-gid>`` with
    modes ``0750`` / ``0640``.  Earlier revisions performed the
    ``mkdir`` -> ``chown`` -> ``chmod`` sequence on path strings, which
    left a TOCTOU window: between the directory existing and the
    metadata syscalls, a swap to a symlink could redirect the chown to
    an arbitrary inode.  See DEF-41586 / CLOS-3965 for context.

    The hardened path opens the parent ``USER_CONFDIR`` once with
    ``O_NOFOLLOW`` at every component, then performs every subsequent
    operation (``mkdir``/``chown``/``chmod``/atomic write) relative to
    that fd or to a fresh ``O_NOFOLLOW`` fd of the user subdir.  No
    user-controlled path string is dereferenced more than once.
    """

    DIR_PERMISSIONS = 0o750
    FILE_PERMISSIONS = 0o640

    def __init__(self, path, username):
        super().__init__(path)
        self.username = username

    def __str__(self):
        return f"Config of user {self.username}"

    def _open_user_subdir(self, parent_fd: int, name: str) -> int:
        """Return an O_NOFOLLOW fd for ``name`` inside *parent_fd*.

        Creates the directory first if it does not already exist.  The
        ``O_NOFOLLOW`` flag guarantees that, if a symlink appears in the
        slot at any time after this call returns, every subsequent
        ``fchown``/``fchmod``/atomic-rewrite bound to the returned fd
        operates on the originally opened inode.
        """
        with suppress(FileExistsError):
            os.mkdir(name, mode=self.DIR_PERMISSIONS, dir_fd=parent_fd)
        return os.open(
            name,
            os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
            dir_fd=parent_fd,
        )

    def write_config_file(self, config) -> str:
        gid = pwd.getpwnam(self.username).pw_gid
        confdir, basename = os.path.split(self.path)
        userconfdir, username = os.path.split(confdir)

        # Open USER_CONFDIR (root-owned, package-controlled) with full
        # symlink protection at every path component.  Then descend to
        # the per-user subdir using a dir_fd-relative open with
        # O_NOFOLLOW so a symlink swap cannot redirect us.
        parent_fd = open_dir_no_symlinks(userconfdir)
        try:
            user_fd = self._open_user_subdir(parent_fd, username)
            try:
                # Apply directory ownership/permissions on the fd we
                # just opened — bound to the inode, not to the path.
                os.chown(user_fd, 0, gid)
                os.fchmod(user_fd, self.DIR_PERMISSIONS)

                config_text = self._serialize_config(config)

                # atomic_rewrite_fd creates a temp file via
                # O_CREAT|O_EXCL|O_NOFOLLOW relative to user_fd, chowns
                # and chmods the temp inode (not a path), then renames
                # it into place — all without leaving a TOCTOU window.
                atomic_rewrite(
                    basename,
                    config_text,
                    backup=False,
                    uid=0,
                    gid=gid,
                    permissions=self.FILE_PERMISSIONS,
                    dir_fd=user_fd,
                )
                # Re-normalize ownership/permissions on every call so
                # that an out-of-band ``chmod``/``chown`` between writes
                # cannot leave the file with weaker permissions.  When
                # ``atomic_rewrite_fd`` short-circuits on identical
                # content, no chown/chmod runs there, so we apply them
                # here.  ``O_NOFOLLOW`` keeps the fix TOCTOU-safe.
                file_fd = os.open(
                    basename,
                    os.O_RDONLY | os.O_NOFOLLOW,
                    dir_fd=user_fd,
                )
                try:
                    os.chown(file_fd, 0, gid)
                    os.fchmod(file_fd, self.FILE_PERMISSIONS)
                finally:
                    os.close(file_fd)
            finally:
                os.close(user_fd)
        finally:
            os.close(parent_fd)

        return config_text
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: config_provider.py

Server Info

System Features

User Info
