types.py

# 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 classes for X-Ray internal objects
"""

# PEP 563: keep annotations unevaluated. The X | None / list[...] forms below
# are PEP 604/585 syntax, which is only valid at runtime on Python 3.10+/3.9+.
# This package installs into the cloudlinux-venv interpreter, and the spec only
# floors that at `cloudlinux-venv >= 1.0.2` -- it does not pin a Python version,
# and the package still builds for CL7. Deferring annotation evaluation keeps the
# modern syntax readable without raising the runtime floor.
from __future__ import annotations

import errno
import hashlib
import logging
import os
import pwd
import re
import stat
import subprocess
from collections.abc import Iterator
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from fnmatch import fnmatchcase
from typing import NamedTuple
from urllib.parse import urlparse

from clcommon.const import Feature
from clcommon.cpapi import is_panel_feature_supported

from xray import gettext as _

from ..reconfiguration.global_ini import is_global_ini_mode
from ..reconfiguration.website_isolation import (
    remove_website_isolation_ini,
    update_website_isolation_ini,
)
from .constants import (
    api_server,
    is_allowed_ini_path,
    part_delimiter,
    request_data_storage,
    task_delimiter,
    tasks_base_storage,
)
from .exceptions import XRayManagerError, XRayManagerExit
from .utils import (
    _cagefsctl_remount,
    _is_cagefs_enabled,
    _is_selector_phpd_location_set,
    _selectorctl_get_version,
    cagefsctl_get_prefix,
    chmod_created_dir,
    dbm_storage_update,
    set_privileges,
    skeleton_update,
    timestamp,
)

logger = logging.getLogger("types")

# Upper bound for an xray.ini / xray.tasks read. A legitimate file is a few
# lines; the cap defends a TOCTOU race where the inode grows between fstat and
# read. Matches website_isolation.MAX_INI_SIZE.
_MAX_READ_SIZE = 64 * 1024

# Roots the daemon itself writes to besides the PHP ini trees: the per-uid
# tasks storage (tasks_base_storage=/usr/share/alt-php-xray-tasks/<uid>, i.e.
# xray.tasks and fake-id/real-id files) and the daemon-local state under
# /usr/share/alt-php-xray (local_tasks_storage, continuous_storage,
# request_data_storage, ...).
_DAEMON_STORAGE_ROOTS = (tasks_base_storage, "/usr/share/alt-php-xray")

def _is_allowed_write_dir(path: str) -> bool:
    """True if the daemon may write into ``path``: an allowlisted PHP ini
    directory (is_allowed_ini_path) or one of the daemon's own root-owned
    storage roots (xray.tasks / fake-id files, continuous-tracing and
    request-data state).

``path`` is the CANONICAL path of an already-pinned directory fd (read
    back from /proc/self/fd by unified_write/unified_erase), so the prefix
    checks below are race-free: they describe exactly the inode the
    subsequent openat/renameat/unlinkat operate on.
    """
    if is_allowed_ini_path(path):
        return True
    for root in _DAEMON_STORAGE_ROOTS:
        root = root.rstrip("/")
        if path == root or path.startswith(root + os.sep):
            return True
    return False

class URL(NamedTuple):
    """
    Representation of an URL
    """

domain_name: str
    uri_path: str

class ContinuousTask(NamedTuple):
    """
    Representation of a continuous tracing task
    """

creation_time: int
    domain: str
    original_url: str
    email: str
    execution_count: int
    status: int

@dataclass
class User:
    """
    Representation of a user entity
    """

name: str
    uid: int
    gid: int
    path: str = field(init=False)

def __post_init__(self):
        self.path = os.path.join(tasks_base_storage, str(self.uid))

def url_split(url: str) -> URL:
    """
    Split URL into domain_name and uripath including query string
    :param url: URL of format protocol://domain/path;parameters?query#fragment
    :return: namedtuple URL(domain_name, uripath)
    """
    fragments = urlparse(url)
    qs = f"?{fragments.query}" if fragments.query else ""
    uri = f"{fragments.path}{qs}" if fragments.path else "/"
    if fragments.netloc.startswith("www."):
        _no_www_netloc = fragments.netloc[4:]
    else:
        _no_www_netloc = fragments.netloc
    _no_port_netloc = _no_www_netloc.split(":")[0]
    logger.info("Parsed %s into %s:%s", url, _no_port_netloc, uri)
    return URL(_no_port_netloc, uri)

class DomainInfo:
    """
    Simple container class for storing information about domain.
    Allows to add whatever attributes.
    Implements two properties:
     - version set in CL selector
     - direct ini path in CageFS
    """

def __init__(self, **kwargs):
        self.name = None
        self.user = None
        self.panel_php_version = None
        self.saved_selector_php_version = "native"
        self.panel_fpm = None

self.is_selector_applied: bool = False
        self.php_ini_scan_dir: str | None = None

for k, v in kwargs.items():
            self.__setattr__(k, v)

def __repr__(self):
        return str(self.__dict__)

def __str__(self):
        return f"Domain {self.name}"

def add_cagefs_dirpath(self, basename: str) -> str | None:
        """
        Get path to ini location directory in CageFS
        """
        prefix = cagefsctl_get_prefix(self.user)
        if prefix is None:
            raise ValueError(_("CageFS prefix resolved as None, but should be a number"))
        return f"/var/cagefs/{prefix}/{self.user}/etc/cl.php.d/{basename}"

@property
    def selector_php_version(self) -> str | None:
        """
        Get PHP version set in selector
        :param domain_user: user of domain
        :return: PHP version in format of 'alt-phpXY'
        """
        if self.saved_selector_php_version == "native":
            result = _selectorctl_get_version(self.user)
            if result is not None:
                out, err = result
                not_err_result = "ERROR" not in err and "ERROR" not in out
                if not_err_result and "native" not in out:
                    short_version = out.split()[0].strip()
                    self.saved_selector_php_version = f"alt-php{''.join(short_version.split('.'))}"
                else:
                    # selectorctl success, but returned ERROR or native version
                    self.saved_selector_php_version = None
            else:
                # selectorctl command failed or just returned non-zero retcode
                self.saved_selector_php_version = None
        return self.saved_selector_php_version

@property
    def selector_ini_path(self) -> str | None:
        """
        Resolve direct ini path in CageFS
        """
        return self.add_cagefs_dirpath(self.selector_php_version)

@property
    def phpd_location_ini_path(self) -> str | None:
        """
        Resolve direct ini path for the case of php.d.location = selector.
        This, if set, affects only alt-php versions with enabled CageFS.
        Also, consider that FPM is not compatible with selector behaviour
        """
        alt_php_in_cage = "alt-php" in self.panel_php_version and _is_cagefs_enabled(self.user)
        if _is_selector_phpd_location_set() and alt_php_in_cage and not self.panel_fpm:
            return self.add_cagefs_dirpath(self.panel_php_version)

class Task:
    """
    Container class for storing task data
    """

cron_file = "/etc/cron.d/xray-manager"

def __init__(
        self,
        *,
        url: str,
        client_ip: str,
        tracing_by: str,
        tracing_count: int,
        ini_location: str = "unknown",
        task_id: str = "unknown",
        status: str = "unknown",
        starttime: int = -1,
        initial_count: int = 0,
        request_count: int = 0,
        auto_task: bool = False,
        user: str | None = None,
        domain_owner: str | None = None,
    ):
        self.url = url
        self.client_ip = client_ip
        self.tracing_by = tracing_by
        self.tracing_count = tracing_count
        self.ini_location = ini_location
        self.task_id = task_id
        self.status = status
        self.starttime = starttime
        self.request_count = request_count
        self.initial_count = tracing_count if initial_count == 0 else initial_count
        self.auto_task = auto_task
        self.user = user
        self.domain_owner = domain_owner
        self.owner_data = None
        self._add_ini_updated = False
        logger.debug("Instantiate Task %s: %s", self.task_id, self.as_dict())

def update_with_local_data(self, *, next_request_id):
        self.request_count = next_request_id - 1

if self.tracing_by != "time":
            self.tracing_count = self.initial_count - self.request_count

def __repr__(self):
        return str(self.__dict__)

def __str__(self):
        return f"Task {self.task_id}"

@property
    def _owner(self) -> User:
        """Resolve (and memoise) the task owner's name/uid/gid."""
        if self.owner_data is None:
            if self.domain_owner is not None:
                user_data = pwd.getpwnam(self.domain_owner)
                self.owner_data = User(name=self.domain_owner, uid=user_data.pw_uid, gid=user_data.pw_gid)
            else:
                raise XRayManagerError(_("Unable to operate with tracing task, because domain owner is not set"))
        return self.owner_data

@property
    def fake_id(self) -> str:
        return hashlib.blake2b(self.task_id.encode(), digest_size=10).hexdigest()

@property
    def shared_link(self) -> str:
        if self.auto_task:
            return f"https://{api_server}/static/xray/reports/?tracing_task_id={self.task_id}"
        raise XRayManagerError(_("No shared link could be obtained for manual task"))

@property
    def is_manual(self) -> bool:
        """If task is a manual one"""
        return not self.auto_task

@property
    def is_continuous(self) -> bool:
        """If task is a continuous one"""
        return self.auto_task and self.user != "*autotracing*"

@property
    def is_autotracing(self) -> bool:
        """If task is an autotracing one"""
        return self.auto_task and self.user == "*autotracing*"

def as_dict(self) -> dict:
        """
        Represent task as a regular dictionary without task_id field
        :return: dict
        """
        return {k: v for k, v in self.__dict__.items() if k not in ("task_id", "owner_data", "_add_ini_updated")}

def set_domain_owner(self, username: str) -> None:
        """
        Update domain_owner attribute if it is not already set
        """
        if self.domain_owner is None:
            self.domain_owner = username

def generate(self) -> str:
        """
        Generate task for writing into xray.ini file.
        :return: domain_name:uripath:client_ip:tracing_task_id
        """
        # encode only " chars within URL
        # since we wrap the value of xray.tasks into them
        url_data = part_delimiter.join(url_split(self.url)).replace('"', "%22")
        task_view = part_delimiter.join((url_data, self.client_ip, self.fake_id))
        logger.info("Generated task %s", task_view)
        return task_view

def is_path_available(self, p: str) -> bool:
        """
        Check if given path is available in user environment
        """
        if not is_panel_feature_supported(Feature.CAGEFS):
            # this check is skipped in case if CageFS is not available
            # since in such a case cagefs_enter_user is available, but throws
            # 'error while loading shared libraries: liblve.so.0'
            return True

cmd = ["/sbin/cagefs_enter_user", self._owner.name, "/usr/bin/stat", p]
        try:
            subprocess.run(cmd, capture_output=True, check=True)
        except subprocess.CalledProcessError:
            logger.error("Mount %s missing", p)
            return False
        except (OSError, ValueError, subprocess.SubprocessError) as e:
            logger.error("External command `%s` failed: %s", cmd, str(e))
            return False
        return True

def ini_file(self) -> str:
        """
        Full path to xray.ini file
        :return: xray.ini full path
        """
        return os.path.join(self.ini_location, "xray.ini")

def check_xray_dir(self):
        if not os.path.isdir(self._owner.path):
            with set_privileges(target_uid=0, target_gid=0, mask=0o066):
                os.makedirs(self._owner.path)
                _cagefsctl_remount(self._owner.name)

def tasks_file(self) -> str:
        """
        Full path to xray.tasks file
        Create location on the fly
        :return: xray.tasks full path
        """
        self.check_xray_dir()
        return os.path.join(self._owner.path, "xray.tasks")

def fake_id_real_id_file(self) -> str:
        self.check_xray_dir()
        return os.path.join(self._owner.path, self.fake_id)

@staticmethod
    def _sanitize_task_id(task_id: str) -> str:
        """Validate task_id contains only safe characters to prevent
        path traversal and shell injection in cron files"""
        from xray.internal.constants import safe_id_pattern

if not safe_id_pattern.match(str(task_id)):
            raise XRayManagerError(f'Invalid task_id format: {task_id}')
        return str(task_id)

def set_cronjob(self, system_id: str) -> None:
        """
        Set a cron job for stop task by time
        :param system_id: a unique system ID
        """
        if self.tracing_by == "time":
            safe_task_id = self._sanitize_task_id(self.task_id)
            safe_system_id = self._sanitize_task_id(system_id)
            # Naive LOCAL time on purpose: the fields below are written straight
            # into a cron schedule, and cron evaluates them in the server's local
            # timezone. A tz-aware (UTC) value here would schedule the stop at the
            # wrong wall-clock time on every host that is not on UTC.
            stop_at = datetime.fromtimestamp(self.starttime) + timedelta(  # noqa: DTZ006
                minutes=self.tracing_count + 1
            )
            shceduled = f"{stop_at.minute} {stop_at.hour} {stop_at.day} {stop_at.month} *"
            cmd = "root /usr/sbin/cloudlinux-xray-manager stop"
            params = f"--system_id {safe_system_id} --tracing_task_id {safe_task_id}"
            cron_line = f"{shceduled}\t{cmd} {params}\n"
            individual_cron_file = self.cron_file + f"_{safe_task_id}"
            logger.info(
                "Generating cron file %s with command %s",
                individual_cron_file,
                cron_line,
            )
            with open(individual_cron_file, "w") as cron:
                cron.write(cron_line)

def drop_cronjob(self) -> None:
        """
        Drop existing cron job file for current task
        """
        safe_task_id = self._sanitize_task_id(self.task_id)
        job = self.cron_file + f"_{safe_task_id}"
        logger.info("Removing cron job %s", job)
        try:
            os.unlink(job)
        except FileNotFoundError:
            logger.info("No such job")

@staticmethod
    def read_file(filepath, expected_uid: int | None = None) -> str:
        """
        Read contents of given file.

The xray.ini parent directory may live inside the user's CageFS jail, so
        the user can replace xray.ini with a symlink to a root-readable file
        (e.g. /etc/shadow) and have the root daemon slurp its bytes. The read
        is therefore opened with O_NOFOLLOW (the symlink is never followed) and
        O_NONBLOCK (so a tenant-planted FIFO/device at the path returns from
        open() at once instead of blocking the root daemon forever waiting for a
        writer  the open must not hang before the inode-type check can reject
        it), and the opened inode is fstat-checked: it must be a regular file
        and, when the caller passes ``expected_uid``, owned by that uid. The caller pins
        ``expected_uid`` to the owner of the ini_location DIRECTORY (the same
        uid unified_write writes the file as), so a legitimately-written ini is
        always accepted  user-owned in per-user/CageFS mode, root-owned in
        default mode  while a file whose owner differs from its dir's owner is
        an anomaly and rejected. Privilege is NOT derived from the path's stat()
        target. Mirrors the write side (unified_write's _nofollow_opener) and the
        per-website read in website_isolation.regenerate_ini_for_website_isolation.

Raise XRayManagerError if the file does not exist, is a symlink, is not
        a regular file, or is not owned by the expected account.
        :return: contents of the file with trailing newlines stripped
        """
        if not os.path.exists(filepath):
            raise XRayManagerError(f"Failed to find {os.path.basename(filepath)}")

def _nofollow_opener(path, flags):
            # O_NONBLOCK: a FIFO/device opened read-only without it blocks until a
            # writer appears, so a tenant who plants a FIFO at the ini path would
            # hang the root daemon indefinitely (a DoS) before the S_ISREG check
            # below ever runs. With O_NONBLOCK the open returns at once and the
            # fstat check rejects the non-regular inode. On a regular file
            # O_NONBLOCK is a no-op for the subsequent read.
            return os.open(path, flags | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC)

try:
            with open(filepath, opener=_nofollow_opener) as existing:
                st = os.fstat(existing.fileno())
                if not stat.S_ISREG(st.st_mode):
                    raise XRayManagerError(f"Refusing to read non-regular file {os.path.basename(filepath)}")
                if expected_uid is not None and st.st_uid != expected_uid:
                    raise XRayManagerError(
                        f"Refusing to read {os.path.basename(filepath)}: "
                        f"owner uid {st.st_uid} != expected uid {expected_uid}"
                    )
                # Bound the read explicitly against a TOCTOU grow-after-fstat.
                return existing.read(_MAX_READ_SIZE).strip("\n")
        except OSError as e:
            if e.errno == errno.ELOOP:
                logger.warning("Refusing to follow symlink at %s", filepath)
            raise XRayManagerError(f"Failed to read {os.path.basename(filepath)}") from e

def read_ini(self) -> list[str]:
        """
        Read contents of xray.ini
        Raise XRayError in case if xray.ini does not exists
        :return: list of lines contained in xray.ini
        """
        # xray.ini is written by unified_write(target_path=self.ini_location):
        # with no explicit target_uid, set_privileges adopts the ini_location
        # DIRECTORY's owner, so the legit file owner == the dir owner. That is
        # the user in CageFS/per-user mode, but root in default mode where the
        # ini_location is a root-owned dir (e.g. /opt/alt/phpXX/link/conf).
        # Pin the read to that same dir owner: a file whose owner differs from
        # its containing dir's owner is an anomaly and is rejected. O_NOFOLLOW
        # in read_file is what closes the symlink-follow read; this owner pin is
        # a consistency / defense-in-depth check that holds for both modes.
        try:
            expected_uid = os.stat(self.ini_location).st_uid
        except OSError:
            expected_uid = None
        return self.read_file(self.ini_file(), expected_uid=expected_uid).split("\n")

@staticmethod
    def read_tasks_file(file) -> list[str]:
        """
        Read contents of xray.tasks
        :return: list tasks records
        """
        try:
            tasks = Task.read_file(file)
            split_pattern = r"(?<=:[0-9A-Za-z]{20}),(?=[a-zA-Z0-9\.-]+:)"
            tasks_list = re.split(split_pattern, tasks)
            return tasks_list
        except XRayManagerError:
            return []

@staticmethod
    def ini_tasks_count(xray_tasks: str) -> int:
        """
        Retrieve the list of existing tracing tasks from given xray.tasks line
        """
        return int(xray_tasks[12:].strip('"\n'))

@staticmethod
    def generate_tasks_line(*args: str) -> str:
        """
        Create an xray.tasks line with given args like:
        args0,args1,args2,...argsN\n
        """
        return task_delimiter.join(args)

def _tasks(self) -> list[str]:
        """
        Retrieve the list of existing tracing tasks from xray.tasks file
        """
        return self.read_tasks_file(self.tasks_file())

def tasks_parsed(self) -> Iterator[URL]:
        """
        Iterate over URL of each task in xray.tasks file
        :return: list of URLs
        """
        for task in self._tasks():
            info = task.split(":")
            yield URL(info[0], info[1])

def count_tasks_for_domain(self, domain: str) -> int:
        """
        Count the number of tasks in xray.tasks file that belong to a specific domain.
        This is used for per-website xray.ini to track domain-specific task count.

:param domain: Domain name to count tasks for
        :return: Number of tasks for the specified domain
        """
        count = 0
        for task_url in self.tasks_parsed():
            if task_url.domain_name == domain:
                count += 1
        return count

def is_a_duplicate(self):
        """
        Check if task is duplicated by another tasks
        already present in current xray.ini
        :return: True if task is a duplicate, False otherwise
        """
        domain, uri = url_split(self.url)

def check(t):
            """
            Match domain name and uri separately, but return joined result
            :param t: task
            :return: True if both domain and uri matches task, False otherwise
            """
            direct_match = domain == t.domain_name and uri == t.uri_path
            wildcard_match = fnmatchcase(domain, t.domain_name) and fnmatchcase(uri, t.uri_path)
            return direct_match or wildcard_match

return any(check(task) for task in self.tasks_parsed())

@staticmethod
    def so_path(php_ver: str | None = None) -> str:
        """
        Ensure full path for versions, which are not among known.6
        In such a case version is expected as two numbers, e.g. 56 or 74
        """
        if php_ver is None or len(php_ver) > 2:
            return "xray.so"
        return f"/opt/alt/php{php_ver}/usr/lib64/php/modules/xray.so"

def _ini_dir_validator(self):
        """Build the pinned-directory binding for the xray.ini write/erase.

Returns ``None`` when no domain owner is known (no binding available).
        Otherwise returns ``validate(pinned_dir, st)`` that unified_write /
        unified_erase run against the ALREADY-PINNED inode.

The binding preserves the identity of the SPECIFIC directory the
        ini_location was authorized as (``self.ini_location`` is canonical --
        phpinfo resolves absolute_ini_scan_dir, custom._validate_ini_path
        returns realpath):

* If the authorized location is a PER-TENANT dir (the owner's own CageFS
          subtree or a per-domain ini dir <root>/<owner>), the pinned inode MUST stay inside that
          owner's own subtree. A tenant symlink-swap of the authorized dir lands
          the pinned inode on a root-owned global (e.g. /etc/php.d) or a foreign
          tenant's dir -- outside the owner subtree -- so it is rejected. This is
          what a bare ``st_uid in (0, owner)`` check missed: it accepted any
          root-owned allowlisted dir.
        * Otherwise the authorized location is an operator-controlled GLOBAL dir
          whose path components are root-owned (a tenant cannot swap them); the
          pinned inode must be root-owned -- but a PER-TENANT tree is never a
          legitimate landing site for that branch, so one is rejected outright
          rather than being allowed to satisfy the root-ownership test. Without
          that, an operator-created and therefore root-owned <root>/<other>/php/
          <vhost> passes `st.st_uid == 0`, which is exactly the check the
          per-tenant branch exists to be stronger than. Not reachable today (the
          phpinfo gate owner-binds via _ini_scan_dir_belongs_to_user and
          custom._validate_ini_path rejects foreign roots first), but this
          validator is the LAST check before a root-performed write, so it does
          not lean on those callers staying correct. Review point 1 on MR !171.
        """
        if not self.domain_owner:
            return None
        # Local import avoids any import-time coupling between the internal
        # modules; phpinfo_utils does not import types, so this is one-way.
        from . import phpinfo_utils

owner = self.domain_owner
        authorized = self.ini_location

def _own_subtree(path):
            return phpinfo_utils._is_own_cagefs_subtree(owner, path) or phpinfo_utils._is_own_per_domain_ini_dir(
                owner, path
            )

authorized_is_per_tenant = _own_subtree(authorized)

def validate(pinned_dir, st):
            if authorized_is_per_tenant:
                return _own_subtree(pinned_dir)
            # The authorized location is a global dir, so the pinned inode must be
            # one too. A per-tenant tree that is not this owner's belongs to
            # somebody else whatever its uid says -- fail closed instead of
            # falling through to the root-ownership test below.
            if phpinfo_utils._is_under_per_domain_ini_root(pinned_dir) or phpinfo_utils._is_within(
                phpinfo_utils._CAGEFS_ROOT, pinned_dir
            ):
                return False
            return st.st_uid == 0

return validate

def update_ini(self, version: str | None = None, with_decrement: bool = False) -> None:
        """
        Update xray.ini tasks counter.

The per-user xray.ini tracks total tasks for the user.
        The per-website xray.ini (for website isolation) tracks tasks for that specific domain only.
        """
        # Re-check the allowlist on every read/write of ini_location. The remove
        # path reaches this without going through create_ini_location(), so this
        # is the only gate guarding ini_location on stop/decrement.
        if not is_allowed_ini_path(self.ini_location):
            raise XRayManagerError(_('ini_location outside allowed paths: %s') % self.ini_location)
        # Write-time binding to the authorized ini directory (see unified_write /
        # _ini_dir_validator). Carries the authorized location's identity so a
        # post-check directory rename cannot redirect the root xray.ini write.
        ini_validator = self._ini_dir_validator()
        new_counter = 0
        existing_contents = None
        try:
            existing_contents = self.read_ini()
        except XRayManagerError:
            ini_contents = f"""extension={self.so_path(version)}
;xray.tasks=1\n"""
            new_counter = 1
        else:

def update():
                """
                A generator function aimed to update xray.task field
                """
                nonlocal new_counter
                for line in existing_contents:
                    if "xray.tasks" in line:
                        current = self.ini_tasks_count(line)
                        new_counter = current - 1 if with_decrement else current + 1
                        yield f";xray.tasks={new_counter}\n"
                    else:
                        yield line + "\n"

ini_contents = "".join(list(update()))

# Extract domain from task URL for website isolation
        domain = url_split(self.url).domain_name

# Calculate per-domain task count for website isolation
        # This is different from the per-user counter (new_counter)
        # The xray.tasks file is already updated by add_task/remove_task before update_ini is called,
        # so it reflects the current state after the task change
        # Note: we can only count domain tasks if domain_owner is set (needed to access xray.tasks file)
        domain_task_count = 0
        if domain and self.domain_owner:
            domain_task_count = self.count_tasks_for_domain(domain)

if new_counter <= 0 and not is_global_ini_mode():
            # this indicates that there are no tasks left for this user,
            # thus the whole xray.ini should be deleted
            if not self.unified_erase(self.ini_file(), dir_validator=ini_validator):
                raise XRayManagerError(_('Failed to remove xray.ini'))
            # Also remove per-website copy if website isolation is enabled
            if domain and self.domain_owner:
                remove_website_isolation_ini(self._owner.name, domain)
        else:
            # in case when global ini mode is enabled
            # or counter still reports some tasks
            # keep the file and update counter
            # mask is explicit on purpose. Every PHP process scanning
            # ini_location has to read this file, so it must be 0644 no matter
            # what umask the caller happened to run under -- the daily
            # autocomplete-tasks cron used to leak 077 in here and left xray.ini
            # at 0600 (ZD 290458). 0o022 reproduces the mode this write has
            # always produced under the normal umask; it is not a relaxation.
            self.unified_write(
                self.ini_file(),
                ini_contents,
                target_path=self.ini_location,
                mask=0o022,
                dir_validator=ini_validator,
            )
            # The shared counter is durable at this point. Website-isolation
            # work below may still fail, so record the increment before that
            # work starts; rollback_add must then apply the matching decrement.
            if not with_decrement:
                self._add_ini_updated = True
            # Handle per-website ini separately based on domain-specific task count
            if domain and self.domain_owner:
                if domain_task_count <= 0:
                    # No tasks left for this specific domain - remove per-website ini
                    remove_website_isolation_ini(self._owner.name, domain)
                else:
                    # Update per-website ini with domain-specific task counter
                    update_website_isolation_ini(
                        self._owner.name,
                        self._owner.uid,
                        self._owner.gid,
                        domain,
                        domain_task_count,
                        existing_contents,
                        version,
                    )

def add_task(self) -> None:
        """
        Add a task into xray.tasks
        """
        _file = self.tasks_file()
        if not self.is_path_available(self._owner.path):
            # check to see if mount is missing (generally cagefs related check)
            _cagefsctl_remount(self._owner.name)
            raise XRayManagerError(
                _(
                    "Creation of tracing task failed due to filesystem problem. Please, try to recreate the task. If the problem persists, contact support for resolution"
                )
            )
        tasks_contents = self.generate_tasks_line(*self._tasks(), self.generate())
        self.unified_write(_file, tasks_contents, target_uid=0, target_gid=self._owner.gid, mask=0o137)
        self.generate_fake_real_id_file()

def remove_task(self) -> None:
        """
        Remove a task from xray.tasks
        """
        if os.path.exists(self.fake_id_real_id_file()):
            self.unified_erase(self.fake_id_real_id_file())
        tasks_contents = self.generate_tasks_line(*[task for task in self._tasks() if self.generate() != task])

if not tasks_contents.strip():
            # this indicates that there are no tasks left,
            # thus the xray.tasks file should be deleted
            self.unified_erase(self.tasks_file())
        else:
            self.unified_write(
                self.tasks_file(),
                tasks_contents,
                target_uid=0,
                target_gid=self._owner.gid,
                mask=0o137,
            )

def generate_fake_real_id_file(self):
        """
        /usr/share/alt-php-xray-tasks/1004/<fake_id>
        used by x-ray extension to obtain real task id
        """
        self.unified_write(
            self.fake_id_real_id_file(),
            self.task_id,
            target_uid=0,
            target_gid=self._owner.gid,
            mask=0o137,
        )

@staticmethod
    def unified_write(
        filepath: str,
        contents: str,
        target_uid: int | None = None,
        target_gid: int | None = None,
        target_path=".",
        mask: int | None = None,
        dir_validator=None,
    ) -> None:
        """
        Unified writing files method.
        Includes writing into temporary file and
         then atomically renaming temporary to original path.

O_NOFOLLOW on the leaf alone is not enough: the DIRECTORY component of
        filepath may live inside the user's CageFS jail, so a tenant can swap it
        (or any ancestor) for a symlink to a root-owned target (e.g.
        /etc/cron.d) and win a TOCTOU race against the allowlist check,
        redirecting the root write. To close that, the parent directory is
        opened ONCE into a pinned file descriptor -- FOLLOWING symlinks, because
        legit root-owned links such as /opt/alt/phpNN/link/conf ->
        /opt/alt/phpNN/etc/php.d must keep resolving -- and the pinned inode's
        canonical path is then read back via /proc/self/fd and validated with
        _is_allowed_write_dir. That post-pin check is race-free (it reflects
        exactly the inode the subsequent openat/renameat use, no matter how any
        path component is swapped afterwards) and rejects BOTH final- and
        ancestor-component symlink escapes, while legit ini dirs and the
        daemon's own storage roots pass. The tmp write and the rename are then
        performed relative to that fd -- the pathname is never re-resolved
        after the pin.
        """
        working_path = filepath + ".tmp"
        target_dir = os.path.dirname(filepath)
        base = os.path.basename(filepath)
        tmp_base = base + ".tmp"
        logger.info("Writing %s with contents %s", working_path, contents)

try:
            dir_fd = os.open(
                target_dir,
                os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC,
            )
        except OSError as e:
            logger.error(
                "Failed to generate %s %s",
                base,
                str(e),
                extra={"reason": str(e)},
            )
            raise XRayManagerExit(_("Failed to generate {}. {}").format(base, str(e.strerror))) from e

# dir_fd is pinned to a concrete inode; its canonical path (read back
        # via /proc/self/fd) reflects exactly where the subsequent
        # openat/renameat operate, regardless of any post-open symlink swap of
        # target_dir or of any ancestor. Reject the fd unless it landed in an
        # allowlisted ini tree or the daemon's own storage roots: a tenant
        # symlink resolving to e.g. /etc/cron.d is refused, while legit
        # root-owned symlinks (alt-php link/conf) resolve into an allowlisted
        # dir and pass. Close the fd exactly once on the reject path -- the
        # raise happens BEFORE the try/finally below takes ownership of it.
        pinned_dir = os.readlink(f"/proc/self/fd/{dir_fd}")
        if not _is_allowed_write_dir(pinned_dir):
            os.close(dir_fd)
            raise XRayManagerError(_('ini_location outside allowed paths: %s') % pinned_dir)

try:
            st = os.fstat(dir_fd)
            # Bind the write to the SPECIFIC authorized directory (race-free: the
            # validator runs against the PINNED inode, the exact one the
            # openat/renameat below use). _is_allowed_write_dir above only
            # confirms the pinned dir is SOME allowlisted ini tree or daemon
            # storage root -- and that allowlist deliberately spans this owner's
            # own trees, OTHER tenants' trees, and root-owned global dirs. A uid
            # class check is not enough: a tenant can symlink their authorized
            # per-tenant ini dir onto a root-owned GLOBAL dir (e.g. /etc/php.d),
            # which is both allowlisted and root-owned. dir_validator carries the
            # authorized location's identity (per-tenant vs global) and rejects a
            # pinned inode that escapes it -- see Task._ini_dir_validator.
            if dir_validator is not None and not dir_validator(pinned_dir, st):
                raise XRayManagerError(
                    _('ini_location write target not bound to the authorized directory: %s') % pinned_dir
                )
            # Derive the write identity from the PINNED inode, not a re-resolved
            # pathname. Explicit caller uid/gid win; otherwise adopt the pinned
            # dir owner -- equivalent to the previous stat(ini_location) result
            # because target_dir is exactly that directory.
            eff_uid = target_uid if target_uid is not None else st.st_uid
            eff_gid = target_gid if target_gid is not None else st.st_gid

def _dirfd_opener(path, flags):
                # Create the tmp file relative to the pinned dir_fd; O_NOFOLLOW
                # still guards the leaf against a pre-planted symlink.
                return os.open(
                    tmp_base,
                    flags | os.O_NOFOLLOW | os.O_CLOEXEC,
                    0o644,
                    dir_fd=dir_fd,
                )

with set_privileges(eff_uid, eff_gid, target_path, mask):
                try:
                    with open(working_path, "w", opener=_dirfd_opener) as _file:
                        _file.write(contents)
                except OSError as e:
                    logger.error(
                        "Failed to generate %s %s",
                        os.path.basename(filepath),
                        str(e),
                        extra={"reason": str(e)},
                    )
                    base_path = os.path.basename(filepath)
                    raise XRayManagerExit(_("Failed to generate {}. {}").format(base_path, str(e.strerror))) from e
                else:
                    # renameat within the pinned dir -- never re-resolves a
                    # pathname, so the swapped-directory attack cannot redirect
                    # it. src and dst share dir_fd, so the rename stays atomic.
                    os.replace(tmp_base, base, src_dir_fd=dir_fd, dst_dir_fd=dir_fd)
        finally:
            os.close(dir_fd)

@staticmethod
    def unified_erase(filepath: str, dir_validator=None) -> bool:
        """
        Unified erase method.
        Unlinks target file.

Like unified_write, the parent directory is opened ONCE into a pinned
        file descriptor (following symlinks, so legit root-owned links such as
        alt-php's link/conf resolve), the pinned inode's canonical path is
        validated with _is_allowed_write_dir, and the unlink is performed
        relative to the fd -- so a swapped symlinked directory component (final
        or ancestor) cannot redirect the root unlink outside the allowed trees.
        """
        logger.info("Remove %s: no tasks left", os.path.basename(filepath))
        target_dir = os.path.dirname(filepath)
        base = os.path.basename(filepath)
        try:
            dir_fd = os.open(
                target_dir,
                os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC,
            )
        except FileNotFoundError:
            return True
        except OSError as e:
            logger.warning(
                "Failed to unlink %s file",
                filepath,
                extra={"file": filepath, "err": str(e)},
            )
            return False

# dir_fd is pinned to a concrete inode; validate its canonical path
        # (via /proc/self/fd) against the union of the ini allowlist and the
        # daemon's storage roots. Race-free: it reflects exactly the inode the
        # unlinkat below operates on. Close the fd exactly once on the reject
        # path -- the raise happens BEFORE the try/finally takes ownership.
        pinned_dir = os.readlink(f"/proc/self/fd/{dir_fd}")
        if not _is_allowed_write_dir(pinned_dir):
            os.close(dir_fd)
            raise XRayManagerError(_('ini_location outside allowed paths: %s') % pinned_dir)

try:
            st = os.fstat(dir_fd)
            # Same directory binding as unified_write: reject a pinned inode that
            # escapes the authorized location (e.g. a symlink redirect from a
            # per-tenant dir onto a root-owned global or a foreign tenant), so a
            # post-check rename cannot redirect the root unlink.
            if dir_validator is not None and not dir_validator(pinned_dir, st):
                raise XRayManagerError(
                    _('ini_location write target not bound to the authorized directory: %s') % pinned_dir
                )
            # Unlink requires write on the directory; adopt the pinned dir
            # owner as the identity (matches the file owner in per-user/CageFS
            # mode, root in default mode).
            with set_privileges(st.st_uid, st.st_gid, target_path=target_dir):
                try:
                    os.unlink(base, dir_fd=dir_fd)
                except FileNotFoundError:
                    return True
                except OSError as e:
                    logger.warning(
                        "Failed to unlink %s file",
                        filepath,
                        extra={"file": filepath, "err": str(e)},
                    )
                    return False
                return True
        finally:
            os.close(dir_fd)

def create_ini_location(self) -> None:
        """
        Create ini location dir if it does not exist.
        Due to troubles with additional ini scan dir on DirectAdmin
        we are to create ini_location path before generating xray.ini
        """
        resolved = os.path.realpath(self.ini_location)
        if not is_allowed_ini_path(resolved):
            raise XRayManagerError(_('ini_location outside allowed paths: %s') % self.ini_location)
        if not os.path.exists(resolved):
            try:
                os.mkdir(resolved)
            except FileExistsError:
                pass
            else:
                # mkdir()'s mode argument is masked by the ambient umask, so it
                # cannot guarantee a traversable ini scan dir on its own. PHP
                # must be able to enter this directory as an unprivileged user;
                # setting the mode after the fact pins 0755 regardless of the
                # caller's umask (see the xray.ini write above, ZD 290458).
                #
                # This runs as root with no privilege drop, and ini_location can
                # resolve inside a tenant-writable subtree, so it must NOT be a
                # path-based os.chmod: the tenant could win the mkdir -> chmod
                # window and swap in a symlink to a root-owned directory.
                # chmod_created_dir pins the inode we just created.
                chmod_created_dir(resolved, 0o755)

@skeleton_update
    @dbm_storage_update
    def add(self, php_version: str | None = None) -> None:
        """
        Add task with primary check for duplicates
        Adding task includes modifying two files:
         - xray.ini file with xray extension and task count
         - xray.tasks file per each user with list of tasks
        """
        if self.is_a_duplicate():
            raise XRayManagerError(_("Task is duplicated by URL"), flag="warning")
        self._add_ini_updated = False
        self.create_ini_location()
        self.add_task()
        self.update_ini(version=php_version)
        self.starttime = timestamp()
        logger.info("Starttime set %s", self.starttime)

@skeleton_update
    @dbm_storage_update
    def remove(self) -> bool:
        """
        If task is present in xray.tasks file,
        remove task from xray.tasks file and
        decrement tasks counter in ini.
        :return: True if the task entry was actually removed,
                 False if there was nothing to remove (already drained)
        """
        if self.generate() not in self._tasks():
            return False

self.remove_task()
        self.update_ini(with_decrement=True)
        return True

def rollback_add(self) -> bool:
        """Undo the durable changes made by a failed ``add``.

``add_task`` completes before ``update_ini`` starts. If the failure was
        earlier than the durable shared-counter write, only the task record must
        be removed; decrementing would disable an older active task. Once that
        write completes, the increment belongs to this failed add and must be
        reverted even if later website-isolation work fails.
        """
        task_present = self.generate() in self._tasks()
        if not task_present and not self._add_ini_updated:
            return False
        if task_present:
            self.remove_task()
            if self.generate() in self._tasks():
                raise XRayManagerError(_('Failed to remove task record during rollback'))
        if self._add_ini_updated:
            self.update_ini(with_decrement=True)
            self._add_ini_updated = False
        return True

def recalculate_counts(self) -> int:
        """
        Recalculates remaining number of time|request_qty counts
        :return: remaining number of counts
        """
        if self.tracing_by == "time":
            current = timestamp()
            if current < self.starttime:
                logger.error(
                    "Failed to recalculate time count: current timestamp appears to be in the past",
                    extra={"currenttime": current, "starttime": self.starttime},
                )
                raise XRayManagerError(
                    _(
                        "Failed to recalculate time count: current timestamp appears to be in the past relatively to task start time"
                    )
                )
            remaining = (current - self.starttime) // 60
            recalculated = self.tracing_count - remaining
            logger.info("Remaining time recalculated: %s left", recalculated)
            return recalculated
        elif self.tracing_by == "request_qty":
            return self.initial_count - self.request_count
        raise XRayManagerError(_("Unknown tracing marker: %s") % self.tracing_by)

def erase_request_id_storage(self) -> None:
        """
        Unlink request ID file (by fake ID of tracing_task)
        """
        req_id_file = os.path.join(request_data_storage, self.fake_id)
        logger.info("Erasing storage %s", req_id_file)
        try:
            os.unlink(req_id_file)
        except OSError as e:
            logger.warning(
                "Failed to unlink request_id file",
                extra={"file": req_id_file, "err": str(e)},
            )

docroot: /home/induvedam/public_html

curdir: /opt/cloudlinux/venv/lib/python3.11/site-packages/xray/internal

	.	4096
	..	4096
X	__init__.py	550
	__pycache__	4096
X	advanced_metrics.py	1566
X	clwpos_safe_imports.py	2266
X	constants.py	6538
X	exceptions.py	3782
X	fpm_utils.py	4709
X	local_counters.py	6097
X	nginx_utils.py	7706
X	phpinfo_utils.py	47796
X	probe_targets.py	18302
X	types.py	47918
X	user_limits.py	4622
X	user_manager.py	5995
X	user_plugin_utils.py	13070
X	utils.py	36818