1#@!#!123s

: / 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 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