usr
/
libexec
/
kcare
/
python
/
kcarectl
➕ New
📤 Upload
✎ Editing:
process_utils.py
← Back
# Copyright (c) Cloud Linux Software, Inc # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENCE.TXT import functools import os import subprocess import textwrap from . import log_utils, utils if False: # pragma: no cover from typing import Any, List, Optional, Tuple, Union # noqa: F401 @utils.cached def find_cmd(name, paths=None, raise_exc=True): # type: (str, Optional[tuple[str, ...]], bool) -> Optional[str] paths = paths or ('/usr/sbin', '/sbin', '/usr/bin', '/bin') for it in paths: fname = os.path.join(it, name) if os.path.isfile(fname): return fname if raise_exc: raise Exception('{0} could not be found at {1}'.format(name, paths)) else: return None def run_command(command, catch_stdout=False, catch_stderr=False, shell=False, check=False): # mocked: tests/unit/conftest.py stdout = subprocess.PIPE if catch_stdout else None stderr = subprocess.PIPE if catch_stderr else None # We need to eventually keep shell=True as it might break customer's hooks, skip this bandit check. p = subprocess.Popen(command, stdout=stdout, stderr=stderr, shell=shell) # nosec B602 stdout_captured, stderr_captured = p.communicate() # type: Union[bytes, str, None], Union[bytes, str, None] code = p.returncode if stdout_captured is not None: stdout_captured = utils.nstr(stdout_captured) if stderr is not None: stderr_captured = utils.nstr(stderr_captured) log_utils.logdebug( textwrap.dedent( """ Call result for `{cmd}`: exit code {exit_code} === STDOUT === {stdout} === STDERR === {stderr} === END === """ ).format(exit_code=p.returncode, stdout=stdout_captured, stderr=stderr_captured, cmd=' '.join(command)) ) if check and code: exc = subprocess.CalledProcessError(code, command) exc.output = stdout_captured exc.stderr = stderr_captured raise exc return code, stdout_captured, stderr_captured def check_output(args, check=False): # type: (list[str], bool) -> Any _, stdout, _ = run_command(args, catch_stdout=True, check=check) return stdout def _get_parent_pid_and_process_name(pid): # type: (int) -> tuple[Optional[int], Optional[str]] try: # use two subprocesses for ppid and comm to prevent parsing problems (there could be any symbols in comm) cmd_ppid = ['ps', '--no-headers', '-o', 'ppid', '-p', str(pid)] code, stdout, _ = run_command(cmd_ppid, catch_stdout=True) if code: log_utils.loginfo("Could not retrieve process parent PID for PID {pid}".format(pid=pid), print_msg=False) return None, None ppid = stdout.strip() cmd_comm = ['ps', '--no-headers', '-o', 'comm', '-p', str(pid)] code, stdout, _ = run_command(cmd_comm, catch_stdout=True) if code: log_utils.loginfo("Could not retrieve process name for PID {pid}".format(pid=pid), print_msg=False) return None, None name = stdout.strip() return int(ppid), name except Exception as e: log_utils.loginfo( "Could not retrieve process name and parent PID for PID {pid}, error: {err}".format(pid=pid, err=e), print_msg=False ) return None, None def log_all_parent_processes(func): """Decorator that logs parent process chain before calling the wrapped function.""" @functools.wraps(func) def wrapper(*args, **kwargs): _log_all_parent_processes() return func(*args, **kwargs) return wrapper def _log_all_parent_processes(): # type: () -> None process_chain = [] # type: list[tuple[int, Optional[str]]] current_pid = os.getpid() while current_pid != 1 and current_pid != 0: ppid, process_name = _get_parent_pid_and_process_name(current_pid) process_chain.append((current_pid, process_name)) if ppid is None: break current_pid = ppid log_utils.loginfo("Agent parent processes chain:", print_msg=False) for level, (pid, name) in enumerate(reversed(process_chain)): prefix = "-" * level + "->" log_utils.loginfo( '{prefix} "{name}" (pid: {pid})'.format(prefix=prefix, name=name or 'unknown', pid=pid or 'unknown'), print_msg=False )
💾 Save Changes
Cancel
📤 Upload File
×
Select File
Upload
Cancel
➕ Create New
×
Type
📄 File
📁 Folder
Name
Create
Cancel
✎ Rename Item
×
Current Name
New Name
Rename
Cancel
🔐 Change Permissions
×
Target File
Permission (e.g., 0755, 0644)
0755
0644
0777
Apply
Cancel