Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def kill_plugin(engine, pid):
"""Kill plugin."""
logger = engine.logger
plugins = engine.store.plugins
plugin_sids = engine.store.plugin_sids
plugin_signatures = engine.store.plugin_signatures
if pid in plugins:
try:
plugins[pid]["abort"].set()
plugins[pid]["aborting"] = asyncio.get_event_loop().create_future()
if plugins[pid]["process_id"] is not None:
kill_process(logger, plugins[pid]["process_id"])
except Exception as exc: # pylint: disable=broad-except
logger.error("Failed to kill plugin %s, error: %s", pid, exc)
if "sid" in plugins[pid]:
if plugins[pid]["sid"] in plugin_sids:
del plugin_sids[plugins[pid]["sid"]]
if plugins[pid]["signature"] in plugin_signatures:
logger.info(
"Cleaning up killed plugin signature %s", plugins[pid]["signature"]
)
del plugin_signatures[plugins[pid]["signature"]]
logger.info("Cleaning up killed plugin %s", pid)
del plugins[pid]
progress = 0
logging_callback(progress, type="progress")
while True:
out = process.stdout.read(1)
if out == "" and process.poll() is not None:
break
os.write(stdfn, out)
sys.stdout.flush()
if abort.is_set() or process.poll() is not None:
break
time.sleep(0)
logger.info("Plugin aborting")
kill_process(logger, process.pid)
outputs, errors = process.communicate()
if outputs is not None:
outputs = str(outputs, "utf-8")
if errors is not None:
errors = str(errors, "utf-8")
exit_code = process.returncode
except Exception as exc: # pylint: disable=broad-except
logger.error(traceback.format_exc())
outputs, errors = "", str(exc)
exit_code = 100
if exit_code == 0:
logging_callback(f"Plugin process exited with code {exit_code}")
stop_callback(True, outputs)
return True