# # Copyright (c) nexB Inc. and others. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # See http://www.apache.org/licenses/LICENSE-2.0 for the license text. # See https://github.com/nexB/commoncode for support or download. # See https://aboutcode.org for more information about nexB OSS projects. # import ctypes import contextlib import io import os from os import path import logging import signal import subprocess from commoncode.fileutils import get_temp_dir from commoncode.system import on_posix from commoncode.system import on_windows from commoncode import text """ Wrapper for executing external commands in sub-processes which works consistently the same way on POSIX and Windows OSes and can cope with the capture of large command standard outputs without exhausting memory. """ logger = logging.getLogger(__name__) TRACE = False if TRACE: import sys logging.basicConfig(stream=sys.stdout) logger.setLevel(logging.DEBUG) # current directory is the root dir of this library curr_dir = path.dirname(path.dirname(path.abspath(__file__))) PATH_ENV_VAR = 'PATH' LD_LIBRARY_PATH = 'LD_LIBRARY_PATH' DYLD_LIBRARY_PATH = 'DYLD_LIBRARY_PATH' def execute(cmd_loc, args, cwd=None, env=None, to_files=False, log=TRACE): """ Run a `cmd_loc` command with the `args` arguments list and return the return code, the stdout and stderr. To avoid RAM exhaustion, always write stdout and stderr streams to files. If `to_files` is False, return the content of stderr and stdout as ASCII strings. Otherwise, return the locations to the stderr and stdout temporary files. Run the command using the `cwd` current working directory with an `env` dict of environment variables. """ assert cmd_loc full_cmd = [cmd_loc] + (args or []) # any shared object should be either in the PATH, the rpath or # side-by-side with the exceutable cmd_dir = os.path.dirname(cmd_loc) env = get_env(env, lib_dir=cmd_dir) or None cwd = cwd or curr_dir # temp files for stderr and stdout tmp_dir = get_temp_dir(prefix='cmd-') sop = path.join(tmp_dir, 'stdout') sep = path.join(tmp_dir, 'stderr') # shell==True is DANGEROUS but we are not running arbitrary commands # though we can execute commands that just happen to be in the path # See why we need it on Windows https://bugs.python.org/issue8557 shell = True if on_windows else False if log: printer = logger.debug if TRACE else lambda x: print(x) printer( 'Executing command %(cmd_loc)r as:\n%(full_cmd)r\nwith: env=%(env)r\n' 'shell=%(shell)r\ncwd=%(cwd)r\nstdout=%(sop)r\nstderr=%(sep)r' % locals()) proc = None rc = 100 try: with io.open(sop, 'wb') as stdout, io.open(sep, 'wb') as stderr, pushd(cmd_dir): proc = subprocess.Popen( full_cmd, cwd=cwd, env=env, stdout=stdout, stderr=stderr, shell=shell, # -1 defaults bufsize to system bufsize bufsize=-1, universal_newlines=True, ) stdout, stderr = proc.communicate() rc = proc.returncode if proc else 0 finally: close(proc) if not to_files: # return output as ASCII string loaded from the output files with open(sop, 'rb') as so: sor = so.read() sop = text.toascii(sor).strip() with open(sep, 'rb') as se: ser = se.read() sep = text.toascii(ser).strip() return rc, sop, sep def execute2( cmd_loc, args, lib_dir=None, cwd=None, env=None, to_files=False, log=TRACE, ): """ DEPRECATED: DO NOT USE. Use execute() instead Run a `cmd_loc` command with the `args` arguments list and return the return code, the stdout and stderr. To avoid RAM exhaustion, always write stdout and stderr streams to files. If `to_files` is False, return the content of stderr and stdout as ASCII strings. Otherwise, return the locations to the stderr and stdout temporary files. Run the command using the `cwd` current working directory with an `env` dict of environment variables. """ import warnings warnings.warn( "commoncode.command.execute2 is deprecated. Use execute() instead.", DeprecationWarning, stacklevel=2, ) return execute(cmd_loc, args, cwd, env, to_files, log) def get_env(base_vars=None, lib_dir=None): """ Return a dictionary of environment variables for command execution with appropriate DY/LD_LIBRARY_PATH path variables. Use the optional `base_vars` environment variables dictionary as a base if provided. Note: if `base_vars` contains DY/LD_LIBRARY_PATH variables these will be overwritten. On POSIX, add `lib_dir` as DY/LD_LIBRARY_PATH-like path if provided. """ env_vars = {} if base_vars: env_vars.update(base_vars) # Create and add LD environment variables if lib_dir and on_posix: new_path = f'{lib_dir}' # on Linux/posix ld_lib_path = os.environ.get(LD_LIBRARY_PATH) env_vars.update({LD_LIBRARY_PATH: update_path_var(ld_lib_path, new_path)}) # on Mac, though LD_LIBRARY_PATH should work too dyld_lib_path = os.environ.get(DYLD_LIBRARY_PATH) env_vars.update({DYLD_LIBRARY_PATH: update_path_var(dyld_lib_path, new_path)}) env_vars = {text.as_unicode(k): text.as_unicode(v) for k, v in env_vars.items()} return env_vars def close(proc): """ Close a `proc` process opened pipes and kill the process. """ if not proc: return def close_pipe(p): if not p: return try: p.close() except IOError: pass close_pipe(getattr(proc, 'stdin', None)) close_pipe(getattr(proc, 'stdout', None)) close_pipe(getattr(proc, 'stderr', None)) try: # Ensure process death otherwise proc.wait may hang in some cases # NB: this will run only on POSIX OSes supporting signals os.kill(proc.pid, signal.SIGKILL) # NOQA except: pass # This may slow things down a tad on non-POSIX Oses but is safe: # this calls os.waitpid() to make sure the process is dead proc.wait() def load_shared_library(dll_loc, *args): """ Return the loaded shared library object from the ``dll_loc`` location. """ if not dll_loc or not path.exists(dll_loc): raise ImportError(f'Shared library does not exists: dll_loc: {dll_loc}') if not isinstance(dll_loc, str): dll_loc = os.fsdecode(dll_loc) lib = None dll_dir = os.path.dirname(dll_loc) try: with pushd(dll_dir): lib = ctypes.CDLL(dll_loc) except OSError as e: from pprint import pformat import traceback msgs = tuple([ f'ctypes.CDLL("{dll_loc}")', 'os.environ:\n{}'.format(pformat(dict(os.environ))), traceback.format_exc(), ]) raise Exception(msgs) from e if lib and lib._name: return lib raise Exception(f'Failed to load shared library with ctypes: {dll_loc}') @contextlib.contextmanager def pushd(path): """ Context manager to change the current working directory to `path`. """ original_cwd = os.getcwd() if not path: path = original_cwd try: os.chdir(path) yield os.getcwd() finally: os.chdir(original_cwd) def update_path_var(existing_path_var, new_path): """ Return an updated value for the `existing_path_var` PATH-like environment variable value by adding `new_path` to the front of that variable if `new_path` is not already part of this PATH-like variable. """ if not new_path: return existing_path_var existing_path_var = existing_path_var or '' existing_path_var = os.fsdecode(existing_path_var) new_path = os.fsdecode(new_path) path_elements = existing_path_var.split(os.pathsep) if not path_elements: updated_path_var = new_path elif new_path not in path_elements: # add new path to the front of the PATH env var path_elements.insert(0, new_path) updated_path_var = os.pathsep.join(path_elements) else: # new path is already in PATH, change nothing updated_path_var = existing_path_var if not isinstance(updated_path_var, str): updated_path_var = os.fsdecode(updated_path_var) return updated_path_var PATH_VARS = DYLD_LIBRARY_PATH, LD_LIBRARY_PATH, 'PATH', def searchable_paths(env_vars=PATH_VARS): """ Return a list of directories where to search "in the PATH" in the provided ``env_vars`` list of PATH-like environment variables. """ dirs = [] for env_var in env_vars: value = os.environ.get(env_var, '') or '' dirs.extend(value.split(os.pathsep)) dirs = [os.path.realpath(d.strip()) for d in dirs if d.strip()] return tuple(d for d in dirs if os.path.isdir(d)) def find_in_path(filename, searchable_paths=searchable_paths()): """ Return the location of a ``filename`` found in the ``searchable_paths`` list of directory or None. """ for path in searchable_paths: location = os.path.join(path, filename) if os.path.exists(location): return location