File Manager

Editing File: /opt/cloudlinux/venv/lib/python3.11/site-packages/clcagefslib/io.py

# 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

Back to Directory File Manager