FM / autotracing

/opt/cloudlinux/venv/lib/python3.11/site-packages/ssa/autotracing/

/opt/cloudlinux/venv/lib/python3.11/site-packages/ssa/autotracing

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
config.py	18664	0644	edit dl rm
__init__.py	140	0644	edit dl rm

Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/ssa/autotracing/config.py (18664B)

# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

"""
This module contains a config parser for cloudlinux-xray-autotracing
"""

import logging
import os
import pwd
import stat
import configparser
from configparser import ConfigParser
from dataclasses import dataclass
from enum import Enum
from typing import Any, Union, Tuple, Iterator, List, Optional

from clcommon.clpwd import ClPwd
from clcommon.const import Feature
from clcommon.cpapi import cpusers, is_panel_feature_supported
from clcommon.cpapi.cpapiexceptions import CPAPIException

from ..internal.constants import flag_file
from ..internal.exceptions import SSAError
from ..internal.utils import (
    umask_0,
    set_privileges,
    is_xray_user_agent_active,
    xray_version,
    is_kernel_version_supported,
)

logger = logging.getLogger("autotracing.config")

def is_autotracing_supported() -> bool:
    """Currently Auto tracing feature is not supported on Shared edition"""
    return is_panel_feature_supported(Feature.AUTOTRACING)

class Status(Enum):
    """
    Autotracing statuses
    """

ENABLED = "enabled"
    DISABLED = "disabled"

@dataclass
class User:
    """
    User container
    """

uid: int
    name: str
    home: str

class AutotracingConfig(ConfigParser):
    """
    Autotracing basic config parser
    """

main_section = "conf"

# uid the config file must belong to, or None to skip the ownership check.
    # UserLevelConfig sets this to the owning tenant's uid so a tenant cannot
    # redirect the privileged read to a root- or other-user-owned file.
    expected_owner_uid = None

def _read_config_file(self) -> None:
        """
        Read self.config_file, hardened against symlink redirection without a
        process-wide privilege drop (safe inside the threaded ssa-agent daemon).

Uses O_NOFOLLOW so a symlink as the final path component is refused, and
        an fstat() ownership check (expected_owner_uid) so an intermediate
        directory symlink that redirects the read to a root/other-user file is
        refused too. A missing/unreadable file is not an error (matches
        ConfigParser.read, which silently skips it); a symlink/ownership/type
        violation IS refused (file is not parsed).

O_NONBLOCK + an fstat() S_ISREG regular-file check close the FIFO /
        special-file DoS: O_NOFOLLOW does NOT reject a FIFO, and a tenant who
        replaces the path with a writer-less FIFO would otherwise hang the
        privileged read forever (a blocking syscall raises nothing, so no
        per-user handler fires). O_NONBLOCK makes the open return immediately
        and the S_ISREG check refuses FIFOs, sockets, devices and directories
        before any read. A regular file ignores O_NONBLOCK on read, so the
        subsequent parse is unaffected.
        """
        try:
            fd = os.open(self.config_file, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC)
        except OSError:
            # ConfigParser.read silently ignores any file it cannot open
            # (missing, permission-denied, O_NOFOLLOW ELOOP on a symlink,
            # ENXIO on a writer-less FIFO, ...); preserve that and fall back
            # to defaults. A suspicious file that DOES open is still refused
            # by the S_ISREG / ownership checks below.
            return
        # errors='replace' so a non-UTF-8 (binary) config degrades to
        # replacement chars instead of raising UnicodeDecodeError; at worst it
        # yields a configparser.Error at parse time, which callers handle.
        with os.fdopen(fd, encoding="utf-8", errors="replace") as config_stream:
            st = os.fstat(fd)
            if not stat.S_ISREG(st.st_mode):
                raise SSAError(f"Refusing to read {self.config_file}: not a regular file (mode {st.st_mode:#o})")
            if self.expected_owner_uid is not None and st.st_uid != self.expected_owner_uid:
                raise SSAError(
                    f"Refusing to read {self.config_file}: owned by uid {st.st_uid}, expected {self.expected_owner_uid}"
                )
            self.read_file(config_stream)

def check_config_dir(self) -> None:
        """
        If subdirectory location for autotracing config file does not exist,
        create it
        """
        subdir_path = os.path.dirname(self.config_file)
        if not os.path.exists(subdir_path):
            os.mkdir(subdir_path)

def set_config_value(self, key: Any, value: Any) -> None:
        """
        Set given config item 'key' to given value 'value'
        """
        self[self.main_section][key] = value
        self.check_config_dir()
        # O_NOFOLLOW: refuse a tenant-planted symlink at the config path rather
        # than following it and truncating the link target under root context.
        try:
            fd = os.open(self.config_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o644)
        except OSError as e:
            raise SSAError(f"Refusing to write autotracing config {self.config_file}: {e}") from e
        with os.fdopen(fd, "w") as configfile:
            self.write(configfile)

def get_config_value(self, key: Any) -> Any:
        """
        Set given config item 'key' to given value 'value'
        """
        # config_file may be a user-writable path (~/.ssa/autotracing) read in
        # root context; a malformed/unreadable file must degrade to the built-in
        # defaults rather than aborting an all-users enumeration (disabled_users).
        try:
            self._read_config_file()
        except (configparser.Error, OSError, UnicodeDecodeError) as e:
            logger.warning("Ignoring invalid autotracing config %s: %s", self.config_file, e)
        try:
            return self[self.main_section][key]
        except (KeyError, configparser.Error) as e:
            # A missing key, or a syntactically-valid file whose value breaks
            # ConfigParser interpolation (e.g. a bare '%' in a user-writable
            # ~/.ssa/autotracing such as "status = 50%"), must degrade to the
            # built-in default rather than propagate and abort the all-users
            # disabled_users() enumeration run in root context.
            # NOTE: self.defaults() cannot be used as the fallback here because
            # main_section == default_section, so read() merges the (possibly
            # malformed) file values into the default section; we keep an
            # untouched copy of the built-in defaults instead.
            if not isinstance(e, KeyError):
                logger.warning(
                    "Ignoring invalid autotracing config value in %s: %s",
                    self.config_file,
                    e,
                )
            return self._builtin_defaults.get(key)

def set_status(self, value: Any) -> None:
        """
        Set given status
        """
        self.set_config_value("status", value)

def get_status(self) -> Any:
        """
        Set given status
        """
        return self.get_config_value("status")

class AdminLevelConfig(AutotracingConfig):
    """Admin level autotracing config"""

def __init__(self):
        defaults = {"status": "disabled"}
        self.config_file = "/usr/share/clos_ssa/autotracing"
        # Snapshot the built-in defaults before read() can merge file values
        # into the default section (main_section == default_section).
        self._builtin_defaults = dict(defaults)
        super().__init__(defaults, default_section=self.main_section, strict=False)

class UserLevelConfig(AutotracingConfig):
    """User level autotracing config"""

def __init__(self, configpath: str):
        defaults = {"status": AdminLevelConfig().get_status()}
        self.config_file = f"{configpath}/.ssa/autotracing"
        # The config lives in the tenant's own home dir, so the privileged
        # daemon read must belong to that tenant. Resolve the expected owner
        # from the home dir itself; if it cannot be resolved, leave the
        # ownership check disabled (O_NOFOLLOW still blocks a final-component
        # symlink) and let the read surface whatever the path actually is.
        try:
            self.expected_owner_uid = os.stat(configpath).st_uid
        except OSError:
            self.expected_owner_uid = None
        # Snapshot the built-in defaults before read() can merge file values
        # into the default section (main_section == default_section).
        self._builtin_defaults = dict(defaults)
        super().__init__(defaults, default_section=self.main_section, strict=False)

def who_am_i() -> User:
    """
    Get current user and his details
    """
    pw_entry = pwd.getpwuid(os.getuid())
    return User(pw_entry.pw_uid, pw_entry.pw_name, pw_entry.pw_dir)

def config_instance(user_home: str = None) -> Union[AdminLevelConfig, UserLevelConfig]:
    """
    Initialize correct config file instance depending on context
    """
    current_user = who_am_i()
    if current_user.uid == 0:
        # in Admin mode: globally or for particular user
        if user_home:
            # for a particular user
            conf_instance = UserLevelConfig(user_home)
        else:
            # globally
            conf_instance = AdminLevelConfig()
    else:
        # in User mode: user's config only
        if is_xray_user_agent_active():
            conf_instance = UserLevelConfig(current_user.home)
        else:
            # if no X-Ray App available, do not allow manipulations
            raise SSAError(
                "Auto tracing management is not available. Reason: X-Ray End-User plugin is not enabled, please contact your system administrator for help."
            )

return conf_instance

def set_config_value(value: str, user: str = None) -> None:
    """ """
    if user:
        # try to modify user's config with dropping privileges
        try:
            pw_data = pwd.getpwnam(user)
        except KeyError as e:
            raise SSAError(f"User '{user}' not found") from e
        try:
            with set_privileges(target_uid=pw_data.pw_uid, target_gid=pw_data.pw_gid):
                config_instance(pw_data.pw_dir).set_status(value)
        except PermissionError as e:
            raise SSAError(e.strerror) from e
    else:
        with umask_0(0o022):
            # remove write for group
            config_instance().set_status(value)

def _audit_autotracing(operation, target, status="success", error=None):
    try:
        caller = who_am_i()
        caller_uid, caller_name = caller.uid, caller.name
    except Exception:
        caller_uid, caller_name = os.getuid(), "unknown"
    try:
        if error is None:
            logger.info(
                "[audit] operation=%s status=%s uid=%d user=%s target=%s",
                operation,
                status,
                caller_uid,
                caller_name,
                target,
            )
        else:
            logger.info(
                "[audit] operation=%s status=%s uid=%d user=%s target=%s error=%r",
                operation,
                status,
                caller_uid,
                caller_name,
                target,
                str(error),
            )
    except Exception:
        pass

def enable(username: str = None, mode_all: bool = False) -> None:
    """
    Enable autotracing.
    If username is given, the user's config is changed in Admin's mode.
    Perform some misconfiguration checks before enabling and
    do not enable if some of them appear
    """
    target = username if username else ("all_users" if mode_all else "global")
    try:
        misconfiguration_checks()
    except SSAError as e:
        issue = e.reason
    else:
        issue = None

try:
        if mode_all and username is None:
            remove_custom_users_configs()
        set_config_value(Status.ENABLED.value, username)
    except Exception as e:
        _audit_autotracing("autotracing_enable", target, status="failure", error=e)
        raise
    _audit_autotracing("autotracing_enable", target)
    return issue

def disable(username: str = None, mode_all: bool = False) -> None:
    """
    Disable autotracing.
    If username is given, the user's config is changed in Admin's mode
    """
    target = username if username else ("all_users" if mode_all else "global")
    try:
        if mode_all and username is None:
            remove_custom_users_configs()
        set_config_value(Status.DISABLED.value, username)
    except Exception as e:
        _audit_autotracing("autotracing_disable", target, status="failure", error=e)
        raise
    _audit_autotracing("autotracing_disable", target)

def status(username: str = None) -> Optional[Tuple[str, Optional[str]]]:
    """
    Get status of autotracing.
    If username is given, the status for a particular user is returned
    """
    try:
        misconfiguration_checks()
    except SSAError as e:
        issue = e.reason
    else:
        issue = None

if username is not None:
        try:
            return UserLevelConfig(ClPwd().get_homedir(username)).get_status(), None
        except ClPwd.NoSuchUserException as e:
            raise SSAError(str(e)) from e
    return AdminLevelConfig().get_status(), issue

def _panel_users() -> Tuple:
    """
    Get panel users via cpapi, ignoring exceptions like NotSupported, etc.
    """
    try:
        return cpusers()
    except CPAPIException:
        return tuple()

def user_configs() -> Iterator[Tuple[str, UserLevelConfig]]:
    """
    Iterator over all users on the server along with their autotracing configs
    """
    for user in _panel_users():
        try:
            _homedir = ClPwd().get_homedir(user)
        except ClPwd.NoSuchUserException:
            continue
        yield user, UserLevelConfig(_homedir)

def disabled_users() -> List[str]:
    """
    Get list of disabled users.

This is a best-effort daily pass over UNTRUSTED per-tenant config files.
    Each user's status is read independently and ANY per-user config failure
    (refusal, OSError, parse error, non-UTF-8 decode, or any other unexpected
    exception) is logged with the user and the exception, then skipped instead
    of aborting the whole pass. This closes the DoS prong of F-02: a single
    tenant who points ``~/.ssa/autotracing`` at a foreign-owned target, or
    writes a pathological/binary config into their own home dir, can no longer
    break the daily decision-maker pass for every other user. A user whose
    config cannot be read is simply not counted as disabled (safe default) and
    the pass continues; genuine bugs remain diagnosable via the logged
    exception.
    """
    disabled = []
    for username, userconf in user_configs():
        try:
            if userconf.get_status() == Status.DISABLED.value:
                disabled.append(username)
        except Exception as e:  # noqa: BLE001 — best-effort over untrusted per-tenant configs
            logger.warning("Skipping user %s in disabled_users pass: cannot read autotracing config (%s)", username, e)
            continue
    return disabled

def remove_custom_users_configs() -> None:
    """
    Remove custom users configurations.

This is a best-effort pass over UNTRUSTED per-tenant ~/.ssa directories.
    Each user's cleanup is attempted independently and ANY per-user OSError
    (permission error, ENOENT, or ENOTEMPTY from a non-empty ~/.ssa the tenant
    left extra files in) is logged with the user and skipped instead of aborting
    the whole pass. Without this, one low-priv tenant could leave a sibling file
    in ~/.ssa so os.rmdir raises ENOTEMPTY (not a PermissionError) and the
    admin's all-users enable/disable would abort before the global config is
    written (F-04).

The delete itself is symlink-safe (F-05). ~/.ssa and ~/.ssa/autotracing are
    tenant-owned, so a tenant could swap either for a symlink; the old
    os.path.isfile()/os.remove() followed it and let a root-initiated --all run
    unlink a redirected target. Instead we os.open() ~/.ssa with
    O_NOFOLLOW|O_DIRECTORY (a symlinked ~/.ssa is refused with ELOOP) and unlink
    the config relative to that verified dir fd, so unlinkat never dereferences a
    symlinked final component either. A refused symlink surfaces as an OSError
    that is logged and skipped, not one that aborts the pass.
    """
    for user, user_config_path in user_configs():
        pw_data = pwd.getpwnam(user)
        ssa_dir, config_name = os.path.split(user_config_path.config_file)
        try:
            with set_privileges(target_uid=pw_data.pw_uid, target_gid=pw_data.pw_gid):
                try:
                    dir_fd = os.open(ssa_dir, os.O_RDONLY | os.O_NOFOLLOW | os.O_DIRECTORY | os.O_CLOEXEC)
                except FileNotFoundError:
                    # no ~/.ssa for this user — nothing to remove
                    continue
                try:
                    os.unlink(config_name, dir_fd=dir_fd)
                finally:
                    os.close(dir_fd)
                os.rmdir(ssa_dir)
        except FileNotFoundError:
            # ~/.ssa exists but holds no autotracing config — nothing to remove
            continue
        except OSError as e:
            logger.warning(
                "Skipping user %s in remove_custom_users_configs pass: cannot remove autotracing config (%s)", user, e
            )
            continue

def misconfiguration_checks() -> None:
    """Additional checks for known malfunctions"""

def make_error(reason: str, fix: str) -> SSAError:
        message = f"{reason}. You should {fix} in order to get Auto Tracing work"
        return SSAError(message, flag="warning")

# check for edition
    if not is_autotracing_supported():
        raise make_error(
            "Your current server setup is unsupported by Auto Tracing feature",
            "switch Control Panel or CloudLinux edition",
        )

# check of IO throttling availability
    if not is_kernel_version_supported():
        raise make_error(
            "Your kernel does not support throttling detection",
            "update the kernel",
        )

# check if X-Ray is installed
    if xray_version() is None:
        raise make_error("X-Ray is not installed", "install X-Ray")

# check if SSA is enabled
    if not os.path.isfile(flag_file):
        raise make_error("Slow Site Analyzer is disabled", "enable it")

cmd: