FM / clcommon

/opt/imunify360/venv/lib/python3.11/site-packages/clcommon/

/opt/imunify360/venv/lib/python3.11/site-packages/clcommon

Name	Size	Mode	Actions
cpapi/	-	0755	rm
lib/	-	0755	rm
public_hooks/	-	0755	rm
__pycache__/	-	0755	rm
clcagefs.py	11255	0644	edit dl rm
clcaptain.py	2003	0644	edit dl rm
clconfig.py	1724	0644	edit dl rm
clconfpars.py	12376	0644	edit dl rm
clcustomscript.py	1184	0644	edit dl rm
cldebug.py	905	0644	edit dl rm
clemail.py	1693	0644	edit dl rm
clexception.py	1165	0644	edit dl rm
clfunc.py	6623	0644	edit dl rm
clhook.py	8869	0644	edit dl rm
cllog.py	1488	0644	edit dl rm
cloutput.py	471	0644	edit dl rm
clproc.py	4146	0644	edit dl rm
clpwd.py	7929	0644	edit dl rm
clquota.py	1296	0644	edit dl rm
clsec.py	657	0644	edit dl rm
clwpos_lib.py	17079	0644	edit dl rm
const.py	277	0644	edit dl rm
evr_utils.py	3638	0644	edit dl rm
features.py	5130	0644	edit dl rm
group_info_reader.py	5413	0644	edit dl rm
lock.py	1041	0644	edit dl rm
mail_helper.py	4557	0644	edit dl rm
mysql_lib.py	5984	0644	edit dl rm
php_conf_reader.py	10000	0644	edit dl rm
sysctl.py	7792	0644	edit dl rm
ui_config.py	3198	0644	edit dl rm
utils.py	34918	0644	edit dl rm
utils_cmd.py	2771	0644	edit dl rm
__init__.py	1406	0644	edit dl rm

Edit: /opt/imunify360/venv/lib/python3.11/site-packages/clcommon/clcaptain.py (2003B)

# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
#

import os
import errno
from subprocess import Popen, PIPE
from clcommon.utils import ExternalProgramFailed

CAPTAIN_PATH = '/usr/share/python-cllib/cl.captain'

def captain(do_try, cmd, stdin=None):
    try:
        do_try()
    except (OSError, IOError) as e:
        if e.errno == errno.EDQUOT and os.geteuid() > 0:
            command = [CAPTAIN_PATH] + [str(s) for s in cmd]
            with Popen(
                command,
                stdin=None if stdin is None else PIPE,
                stdout=PIPE,
                stderr=PIPE,
                text=True,
            ) as proc:
                if stdin is None:
                    std_out, std_err = proc.communicate()
                else:
                    std_out, std_err = proc.communicate(stdin)

if proc.returncode != 0:
                    raise ExternalProgramFailed(
                        std_err or f'output of "{command}": {std_out}'
                    ) from e
        else:
            raise

def mkdir(path, mode=0o755, recursive=False):
    umask = os.umask(0)
    if recursive:
        collector = "/"
        for subdir in path.split("/"):
            collector = os.path.join(collector, subdir)
            if os.path.exists(collector):
                continue
            captain(lambda: os.mkdir(collector, mode), ['-d', collector, '-m', f'{mode:o}'])
    else:
        captain(lambda: os.mkdir(path, mode), ['-d', path, '-m', f'{mode:o}'])
    os.umask(umask)

def write(path, content=''):
    def do_try():
        with open(path, 'w', encoding='utf-8') as fp:
            fp.write(content)
            fp.flush()
            os.fsync(fp.fileno())

captain(do_try, ['-f', path], content)

def symlink(src, dst):
    captain(lambda: os.symlink(src, dst), ['-l', dst, '-s', src])

# -*- coding: utf-8 -*- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import os import errno from subprocess import Popen, PIPE from clcommon.utils import ExternalProgramFailed CAPTAIN_PATH = '/usr/share/python-cllib/cl.captain' def captain(do_try, cmd, stdin=None): try: do_try() except (OSError, IOError) as e: if e.errno == errno.EDQUOT and os.geteuid() > 0: command = [CAPTAIN_PATH] + [str(s) for s in cmd] with Popen( command, stdin=None if stdin is None else PIPE, stdout=PIPE, stderr=PIPE, text=True, ) as proc: if stdin is None: std_out, std_err = proc.communicate() else: std_out, std_err = proc.communicate(stdin) if proc.returncode != 0: raise ExternalProgramFailed( std_err or f'output of "{command}": {std_out}' ) from e else: raise def mkdir(path, mode=0o755, recursive=False): umask = os.umask(0) if recursive: collector = "/" for subdir in path.split("/"): collector = os.path.join(collector, subdir) if os.path.exists(collector): continue captain(lambda: os.mkdir(collector, mode), ['-d', collector, '-m', f'{mode:o}']) else: captain(lambda: os.mkdir(path, mode), ['-d', path, '-m', f'{mode:o}']) os.umask(umask) def write(path, content=''): def do_try(): with open(path, 'w', encoding='utf-8') as fp: fp.write(content) fp.flush() os.fsync(fp.fileno()) captain(do_try, ['-f', path], content) def symlink(src, dst): captain(lambda: os.symlink(src, dst), ['-l', dst, '-s', src])