conductor-worker-python/pyexecplugins/Shell.py

39 lines
1.8 KiB
Python

import subprocess
from subprocess import PIPE
import logging
from pyexecplugins.pyexecplugins import PyExecPlugin
class Plugin(PyExecPlugin):
name = "Shell"
taskdef = {
"name" : "pyshell",
"description" : "Execute an Shell commands on target machine. Commands are in the form of an array of objects like { 'line': 'ls -l', 'expect' : 0, 'withshell' : False}. For expect 0 is the default, for withshell False is the default thus they can be omitted.",
"inputKeys" : ["commands"],
"outputKeys" : ["results"],
"ownerEmail" : "m.lettere@gmail.com"
}
def __init__(self, data=None, config=None):
super().__init__(data, config)
def execute(self):
results = []
for cmdentry in self.data.get("commands"):
expect = cmdentry.get("expect", 0)
cmd = cmdentry.get("line")
#Use shell=True for very complex cmdlines that do have quotes and other amenities
withshell = cmdentry.get("withshell", self.data.get("withshell", False))
if cmd == None:
continue
else:
cmd = cmd.split() if not withshell else cmd
logging.getLogger("pyexec").debug("Going to execute (withshell=%b) %s", withshell, cmd)
try:
completed = subprocess.run(cmd, shell=withshell, capture_output=True, text=True)
except Exception as e:
completed = subprocess.run(cmd, shell=withshell, stdout=PIPE, stderr=PIPE)
results.append({ 'returncode' : completed.returncode, 'stdout' : str(completed.stdout), 'stderr': str(completed.stderr)})
if completed.returncode != expect:
raise Exception("Intermediate command failed with status code " + str(completed.returncode) )
return { "results": results }