def kill(self, sig):
'''Sends a Unix signal to the subprocess.
Use constants from the :mod:`signal` module to specify which signal.
'''
if sys.platform == 'win32':
if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
sig = signal.CTRL_C_EVENT
elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
sig = signal.CTRL_BREAK_EVENT
else:
sig = signal.SIGTERM
os.kill(self.proc.pid, sig)
python类SIGBREAK的实例源码
def __init__(self, master=False):
# Setup signal fd, this allows signal to behave correctly
if os.name == 'posix':
self.signal_pipe_r, self.signal_pipe_w = os.pipe()
self._set_nonblock(self.signal_pipe_r)
self._set_nonblock(self.signal_pipe_w)
signal.set_wakeup_fd(self.signal_pipe_w)
self._signals_received = collections.deque()
signal.signal(signal.SIGINT, signal.SIG_DFL)
if os.name == 'posix':
signal.signal(signal.SIGCHLD, signal.SIG_DFL)
signal.signal(signal.SIGTERM, self._signal_catcher)
signal.signal(signal.SIGALRM, self._signal_catcher)
signal.signal(signal.SIGHUP, self._signal_catcher)
else:
# currently a noop on window...
signal.signal(signal.SIGTERM, self._signal_catcher)
# FIXME(sileht): should allow to catch signal CTRL_BREAK_EVENT,
# but we to create the child process with CREATE_NEW_PROCESS_GROUP
# to make this work, so current this is a noop for later fix
signal.signal(signal.SIGBREAK, self._signal_catcher)
def test_issue9324(self):
# Updated for issue #10003, adding SIGBREAK
handler = lambda x, y: None
checked = set()
for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
signal.SIGTERM):
# Set and then reset a handler for signals that work on windows.
# Issue #18396, only for signals without a C-level handler.
if signal.getsignal(sig) is not None:
signal.signal(sig, signal.signal(sig, handler))
checked.add(sig)
# Issue #18396: Ensure the above loop at least tested *something*
self.assertTrue(checked)
with self.assertRaises(ValueError):
signal.signal(-1, handler)
with self.assertRaises(ValueError):
signal.signal(7, handler)
def test_issue9324(self):
# Updated for issue #10003, adding SIGBREAK
handler = lambda x, y: None
checked = set()
for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
signal.SIGTERM):
# Set and then reset a handler for signals that work on windows.
# Issue #18396, only for signals without a C-level handler.
if signal.getsignal(sig) is not None:
signal.signal(sig, signal.signal(sig, handler))
checked.add(sig)
# Issue #18396: Ensure the above loop at least tested *something*
self.assertTrue(checked)
with self.assertRaises(ValueError):
signal.signal(-1, handler)
with self.assertRaises(ValueError):
signal.signal(7, handler)
def test_issue9324(self):
# Updated for issue #10003, adding SIGBREAK
handler = lambda x, y: None
checked = set()
for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
signal.SIGTERM):
# Set and then reset a handler for signals that work on windows.
# Issue #18396, only for signals without a C-level handler.
if signal.getsignal(sig) is not None:
signal.signal(sig, signal.signal(sig, handler))
checked.add(sig)
# Issue #18396: Ensure the above loop at least tested *something*
self.assertTrue(checked)
with self.assertRaises(ValueError):
signal.signal(-1, handler)
with self.assertRaises(ValueError):
signal.signal(7, handler)
def kill(self, sig):
'''Sends a Unix signal to the subprocess.
Use constants from the :mod:`signal` module to specify which signal.
'''
if sys.platform == 'win32':
if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
sig = signal.CTRL_C_EVENT
elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
sig = signal.CTRL_BREAK_EVENT
else:
sig = signal.SIGTERM
os.kill(self.proc.pid, sig)
def test_issue9324(self):
# Updated for issue #10003, adding SIGBREAK
handler = lambda x, y: None
for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
signal.SIGTERM):
# Set and then reset a handler for signals that work on windows
signal.signal(sig, signal.signal(sig, handler))
with self.assertRaises(ValueError):
signal.signal(-1, handler)
with self.assertRaises(ValueError):
signal.signal(7, handler)
def test_issue9324(self):
# Updated for issue #10003, adding SIGBREAK
handler = lambda x, y: None
for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
signal.SIGTERM):
# Set and then reset a handler for signals that work on windows
signal.signal(sig, signal.signal(sig, handler))
with self.assertRaises(ValueError):
signal.signal(-1, handler)
with self.assertRaises(ValueError):
signal.signal(7, handler)
def test_issue9324(self):
# Updated for issue #10003, adding SIGBREAK
handler = lambda x, y: None
for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
signal.SIGTERM):
# Set and then reset a handler for signals that work on windows
signal.signal(sig, signal.signal(sig, handler))
with self.assertRaises(ValueError):
signal.signal(-1, handler)
with self.assertRaises(ValueError):
signal.signal(7, handler)
def term_signal():
"""Gets the term signal for the os."""
if os.name == 'nt':
return signal.SIGBREAK
else:
return signal.SIGTERM
def kill(self, sig):
'''Sends a Unix signal to the subprocess.
Use constants from the :mod:`signal` module to specify which signal.
'''
if sys.platform == 'win32':
if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
sig = signal.CTRL_C_EVENT
elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
sig = signal.CTRL_BREAK_EVENT
else:
sig = signal.SIGTERM
os.kill(self.proc.pid, sig)
def test_issue9324(self):
# Updated for issue #10003, adding SIGBREAK
handler = lambda x, y: None
for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
signal.SIGTERM):
# Set and then reset a handler for signals that work on windows
signal.signal(sig, signal.signal(sig, handler))
with self.assertRaises(ValueError):
signal.signal(-1, handler)
with self.assertRaises(ValueError):
signal.signal(7, handler)
def kill(self, sig):
'''Sends a Unix signal to the subprocess.
Use constants from the :mod:`signal` module to specify which signal.
'''
if sys.platform == 'win32':
if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
sig = signal.CTRL_C_EVENT
elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
sig = signal.CTRL_BREAK_EVENT
else:
sig = signal.SIGTERM
os.kill(self.proc.pid, sig)
def test_issue9324(self):
# Updated for issue #10003, adding SIGBREAK
handler = lambda x, y: None
for sig in (signal.SIGABRT, signal.SIGBREAK, signal.SIGFPE,
signal.SIGILL, signal.SIGINT, signal.SIGSEGV,
signal.SIGTERM):
# Set and then reset a handler for signals that work on windows
signal.signal(sig, signal.signal(sig, handler))
with self.assertRaises(ValueError):
signal.signal(-1, handler)
with self.assertRaises(ValueError):
signal.signal(7, handler)
def kill(self, sig):
'''Sends a Unix signal to the subprocess.
Use constants from the :mod:`signal` module to specify which signal.
'''
if sys.platform == 'win32':
if sig in [signal.SIGINT, signal.CTRL_C_EVENT]:
sig = signal.CTRL_C_EVENT
elif sig in [signal.SIGBREAK, signal.CTRL_BREAK_EVENT]:
sig = signal.CTRL_BREAK_EVENT
else:
sig = signal.SIGTERM
os.kill(self.proc.pid, sig)
def kill_windows(pid, signum):
"""
Adapt os.kill for Windows platform.
Reference: <https://stackoverflow.com/questions/35772001/how-to-handle-the-signal-in-python-on-windows-machine>
"""
sigmap = {signal.SIGINT: signal.CTRL_C_EVENT, signal.SIGBREAK: signal.CTRL_BREAK_EVENT}
if signum in sigmap and pid == os.getpid():
# we don't know if the current process is a
# process group leader, so just broadcast
# to all processes attached to this console.
pid = 0
thread = threading.current_thread()
handler = signal.getsignal(signum)
# work around the synchronization problem when calling
# kill from the main thread.
if (signum in sigmap and
thread.name == 'MainThread' and
callable(handler) and
pid == 0):
event = threading.Event()
def handler_set_event(signum, frame):
event.set()
return handler(signum, frame)
signal.signal(signum, handler_set_event)
try:
os.kill(pid, sigmap[signum])
# busy wait because we can't block in the main
# thread, else the signal handler can't execute.
while not event.is_set():
pass
finally:
signal.signal(signum, handler)
else:
os.kill(pid, sigmap.get(signum, signum))
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
if p.returncode != 0:
sys.exit(p.returncode)
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
external_search_command.py 文件源码
项目:splunk_ta_ps4_f1_2016
作者: jonathanvarley
项目源码
文件源码
阅读 26
收藏 0
点赞 0
评论 0
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
external_search_command.py 文件源码
项目:splunk_ta_ps4_f1_2016
作者: jonathanvarley
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
external_search_command.py 文件源码
项目:cb-defense-splunk-app
作者: carbonblack
项目源码
文件源码
阅读 31
收藏 0
点赞 0
评论 0
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
external_search_command.py 文件源码
项目:cb-defense-splunk-app
作者: carbonblack
项目源码
文件源码
阅读 33
收藏 0
点赞 0
评论 0
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
if p.returncode != 0:
sys.exit(p.returncode)
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
sys.exit(p.returncode)
def _execute(path, argv=None, environ=None):
""" Executes an external search command.
:param path: Path to the external search command.
:type path: unicode
:param argv: Argument list.
:type argv: list or tuple
The arguments to the child process should start with the name of the command being run, but this is not
enforced. A value of :const:`None` specifies that the base name of path name :param:`path` should be used.
:param environ: A mapping which is used to define the environment variables for the new process.
:type environ: dict or None.
This mapping is used instead of the current process’s environment. A value of :const:`None` specifies that
the :data:`os.environ` mapping should be used.
:return: None
"""
search_path = os.getenv('PATH') if environ is None else environ.get('PATH')
found = ExternalSearchCommand._search_path(path, search_path)
if found is None:
raise ValueError('Cannot find command on path: {}'.format(path))
path = found
logger.debug('starting command="%s", arguments=%s', path, argv)
def terminate(signal_number, frame):
sys.exit('External search command is terminating on receipt of signal={}.'.format(signal_number))
def terminate_child():
if p.pid is not None and p.returncode is None:
logger.debug('terminating command="%s", arguments=%d, pid=%d', path, argv, p.pid)
os.kill(p.pid, CTRL_BREAK_EVENT)
p = Popen(argv, executable=path, env=environ, stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr)
atexit.register(terminate_child)
signal(SIGBREAK, terminate)
signal(SIGINT, terminate)
signal(SIGTERM, terminate)
logger.debug('started command="%s", arguments=%s, pid=%d', path, argv, p.pid)
p.wait()
logger.debug('finished command="%s", arguments=%s, pid=%d, returncode=%d', path, argv, p.pid, p.returncode)
if p.returncode != 0:
sys.exit(p.returncode)