conductor-worker-python/pyexecplugins/Shell.py

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").info("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 }