1#@!#!123s

: / proc / self / root / opt / cloudlinux / venv / lib / python3.11 / site-packages / clcagefslib /
Filename : io.py
back
# Copyright © Cloud Linux GmbH & 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 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):
    """
    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
    """
    try:
        with open(filename, "r") as _file:
            return _file.readlines()

    except (OSError, IOError):
        secureio.logging("Error: failed to read " + 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 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"Symlink {link_name} does not exist")
            else:
                logger.error(f"Error: Unable to remove symlink {link_name}", exc_info=e)
        try:
            clcaptain.symlink(dest_path, link_name)
        except (OSError, ExternalProgramFailed) as e:
            msg = f"Error: failed to create symlink {link_name} to {dest_path} : {str(e).replace('Errno', 'Err code')}"
            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("w", 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