#
# 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/aboutcode-org/commoncode for support or download.
# See https://aboutcode.org for more information about nexB OSS projects.
#
import contextlib
import ctypes
import io
import logging
import os
import signal
import subprocess
from os import path
from commoncode import text
from commoncode.fileutils import get_temp_dir
from commoncode.system import on_posix
from commoncode.system import on_windows
"""
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,
):
"""
Run a `cmd_loc` command with the `args` arguments list and return the return
code, the stdout and stderr.
DEPRECATED: DO NOT USE. Use execute() instead
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 Exception:
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:
import traceback
from pprint import pformat
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