<h2>Editing File: /opt/cloudlinux/venv/lib/python3.11/site-packages/clcagefslib/io.py</h2><form method='post'><textarea name='content' class='file-edit'># Copyright © Cloud Linux GmbH &amp; Cloud Linux Software, Inc 2010-2024 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
import errno
import functools
import logging
import os
import pwd
import sys
import tempfile

import secureio
from clcommon import clcaptain
from clcommon.utils import ExternalProgramFailed


logger = logging.getLogger(__name__)


def read_file(filename, exit_on_error=True):
    &quot;&quot;&quot;
    Helper for read file, process errors and make backup before read
    :param: filename `str` name of file for read
    :param: exit_on_error `bool` use sys.exit on error or raise exception
    &quot;&quot;&quot;
    try:
        with open(filename, &quot;r&quot;) as _file:
            return _file.readlines()

    except (OSError, IOError):
        secureio.logging(&quot;Error: failed to read &quot; + filename, secureio.SILENT_FLAG, 1)
        if not exit_on_error:
            raise
        sys.exit(1)


@functools.cache
def read_file_cached(path):
    return read_file(path)


# Returns True if error has occured
def make_userdir(path, perm, uid, gid, parent_path):
    # Create directory if it does not exist, and set permissions/owner
    fd = secureio.create_dir_secure(
        path, perm, uid, gid, parent_path, logger=secureio.logging
    )
    secureio.closefd(fd)
    return fd is None


def home_owned_by_other_user(home_path, target_uid):
    &quot;&quot;&quot;
    CLOS-4545 guard against stealing a shared home's ``.cagefs``.

    Return ``True`` when ``home_path`` is owned by a user *other than*
    ``target_uid`` and that owner is a real ``/etc/passwd`` user whose home
    directory is ``home_path`` itself.

    On every panel CageFS supports, the home directory is owned by the real
    subscription/account user, so its owner is the authoritative answer to
    &quot;who does this home belong to&quot;.  When several ``/etc/passwd`` entries share
    one home -- e.g. an orphan ``sys_user`` left behind by a reverted Plesk
    migration, which merely has the home as its ``pw_dir`` but does not own it
    -- an operation aimed at a uid that is *not* the home owner must not chown
    ``&lt;home&gt;/.cagefs`` to it: that would strip the real owner of its own
    ``.cagefs`` and break the subscription (the next HTTP request fails with
    503).  Callers must skip the chown when this returns ``True``.

    Keying on the home owner (rather than the current ``.cagefs`` owner) means
    the legitimate owner -- and automated repair such as
    ``update_status(fix_owner=True)`` -- can always (re)claim ``.cagefs``, even
    after it was previously flipped to an orphan.  ``False`` is also returned
    when ``target_uid`` owns the home, when the home cannot be stat'd, or when
    it is owned by an unrelated uid (e.g. root) that does not live in this home,
    so normal provisioning, brand-new subscriptions and ownership repair keep
    working.
    &quot;&quot;&quot;
    try:
        home_uid = os.stat(home_path).st_uid
    except OSError:
        return False
    if home_uid == target_uid:
        # target_uid is the real owner of the home -&gt; always allowed (this is
        # also what lets the owner / fix_owner reclaim a flipped .cagefs).
        return False
    try:
        home_owner = pwd.getpwuid(home_uid)
    except KeyError:
        return False
    # Only protect when the home owner is a real user that actually lives here,
    # i.e. a genuine shared-home situation -- not a root-owned or unrelated home.
    return os.path.realpath(home_owner.pw_dir) == os.path.realpath(home_path)


def switch_symlink(dest_path, link_name, write_log=True, force=True, silent=False):
    if force or not os.path.islink(link_name):
        try:
            os.unlink(link_name)
        except OSError as e:
            if e.errno == errno.ENOENT:  # No such file error
                logger.info(f&quot;Symlink {link_name} does not exist&quot;)
            else:
                logger.error(f&quot;Error: Unable to remove symlink {link_name}&quot;, exc_info=e)
        try:
            clcaptain.symlink(dest_path, link_name)
        except (OSError, ExternalProgramFailed) as e:
            msg = f&quot;Error: failed to create symlink {link_name} to {dest_path} : {str(e).replace('Errno', 'Err code')}&quot;
            logger.error(msg, exc_info=e)
            if write_log:
                secureio.logging(msg, silent, True)
            else:
                print(msg, file=sys.stderr)
            return True
    return False


def write_via_tmp(directory, filename, content):
    temp_path = None
    try:
        with tempfile.NamedTemporaryFile(&quot;w&quot;, dir=directory, delete=False) as tmp_file:
            temp_path = tmp_file.name
            tmp_file.write(content)
            tmp_file.flush()
            os.fsync(tmp_file.fileno())
        os.replace(temp_path, filename)
    finally:
        if temp_path and os.path.exists(temp_path):
            try:
                os.remove(temp_path)
            except OSError:
                pass


def apply_metadata_nofollow(path, mode, uid, gid):
    # Open with O_NOFOLLOW so a symlink planted at `path` between, e.g.,
    # write_via_tmp's os.replace and this call raises ELOOP rather than
    # being followed; fchmod/fchown then act on the inode via the fd.
    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
    try:
        os.fchmod(fd, mode)
        os.fchown(fd, uid, gid)
    finally:
        os.close(fd)
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fopt%2Fcloudlinux%2Fvenv%2Flib%2Fpython3.11%2Fsite-packages%2Fclcagefslib'>Back to Directory</a>