Make salt-ssh more strict when handling unexpected situations

This commit ensures that output indicating failure is not
passed off as the output of the command when it is called
via `__salt__`, e.g. during template rendering or in wrapper
modules. It does this by raising exceptions when issues
are detected and thus simulates the remote error locally.
See issue saltstack#64531 for a detailed description.

changelog/52452.fixed.md

@@ -0,0 +1 @@
+Fixed salt-ssh continues state/pillar rendering with incorrect data when an exception is raised by a module on the target

salt/client/ssh/__init__.py

@@ -555,41 +555,37 @@ def handle_routine(self, que, opts, host, target, mine=False):
             **target,
         )
         ret = {"id": single.id}
-        stdout, stderr, retcode = single.run()
+        stdout = stderr = ""
+        retcode = 0
         try:
-            retcode = int(retcode)
-        except (TypeError, ValueError):
-            log.warning(f"Got an invalid retcode for host '{host}': '{retcode}'")
-            retcode = 1
-        # This job is done, yield
-        try:
-            data = salt.utils.json.find_json(stdout)
-            if len(data) < 2 and "local" in data:
-                ret["ret"] = data["local"]
-                try:
-                    # Ensure a reported local retcode is kept
-                    remote_retcode = data["local"]["retcode"]
-                    try:
-                        retcode = int(remote_retcode)
-                    except (TypeError, ValueError):
-                        log.warning(
-                            f"Host '{host}' reported an invalid retcode: '{remote_retcode}'"
-                        )
-                        retcode = max(retcode, 1)
-                except (KeyError, TypeError):
-                    pass
-            else:
-                ret["ret"] = {
-                    "stdout": stdout,
-                    "stderr": stderr,
-                    "retcode": retcode,
-                }
-        except Exception:  # pylint: disable=broad-except
+            stdout, stderr, retcode = single.run()
+            ret["ret"] = salt.client.ssh.wrapper.parse_ret(stdout, stderr, retcode)
+        except (
+            salt.client.ssh.wrapper.SSHPermissionDeniedError,
+            salt.client.ssh.wrapper.SSHCommandExecutionError,
+        ) as err:
+            ret["ret"] = err.to_ret()
+            # All caught errors always indicate the retcode is/should be > 0
+            retcode = max(retcode, err.retcode, 1)
+        except salt.client.ssh.wrapper.SSHException as err:
+            ret["ret"] = err.to_ret()
+            if not self.opts.get("raw_shell"):
+                # We only expect valid JSON output from Salt
+                retcode = max(retcode, err.retcode, 1)
+                ret["ret"].pop("_error", None)
+        except Exception as err:  # pylint: disable=broad-except
+            log.error(
+                f"Error while parsing the command output: {err}",
+                exc_info_on_loglevel=logging.DEBUG,
+            )
             ret["ret"] = {
+                "_error": f"Internal error while parsing the command output: {err}",
                 "stdout": stdout,
                 "stderr": stderr,
                 "retcode": retcode,
+                "data": None,
             }
+            retcode = max(retcode, 1)
         que.put((ret, retcode))
 
     def handle_ssh(self, mine=False):
@@ -737,7 +733,7 @@ def run_iter(self, mine=False, jid=None):
             self.cache_job(jid, host, ret[host], fun)
             if self.event:
                 id_, data = next(iter(ret.items()))
-                if isinstance(data, str):
+                if not isinstance(data, dict):
                     data = {"return": data}
                 if "id" not in data:
                     data["id"] = id_
@@ -854,7 +850,7 @@ def run(self, jid=None):
                 salt.output.display_output(p_data, outputter, self.opts)
             if self.event:
                 id_, data = next(iter(ret.items()))
-                if isinstance(data, str):
+                if not isinstance(data, dict):
                     data = {"return": data}
                 if "id" not in data:
                     data["id"] = id_
@@ -1037,7 +1033,7 @@ def run_ssh_pre_flight(self):
             # NamedTemporaryFile
             try:
                 shutil.copyfile(self.ssh_pre_flight, temp.name)
-            except OSError as err:
+            except OSError:
                 return (
                     "",
                     "Could not copy pre flight script to temporary path",
@@ -1099,7 +1095,8 @@ def run(self, deploy_attempted=False):
 
         Returns tuple of (stdout, stderr, retcode)
         """
-        stdout = stderr = retcode = None
+        stdout = stderr = ""
+        retcode = 0
 
         if self.ssh_pre_flight:
             if not self.opts.get("ssh_run_pre_flight", False) and self.check_thin_dir():
@@ -1176,11 +1173,6 @@ def run_wfunc(self):
             )
 
             opts_pkg = pre_wrapper["test.opts_pkg"]()  # pylint: disable=E1102
-            if "_error" in opts_pkg:
-                # Refresh failed
-                retcode = opts_pkg["retcode"]
-                ret = salt.utils.json.dumps({"local": opts_pkg})
-                return ret, retcode
 
             opts_pkg["file_roots"] = self.opts["file_roots"]
             opts_pkg["pillar_roots"] = self.opts["pillar_roots"]
@@ -1307,14 +1299,18 @@ def run_wfunc(self):
                 result = wrapper[mine_fun](*self.args, **self.kwargs)
             else:
                 result = self.wfuncs[self.fun](*self.args, **self.kwargs)
+        except salt.client.ssh.wrapper.SSHException:
+            # SSHExceptions indicating remote command failure or
+            # parsing issues are handled centrally in SSH.handle_routine
+            raise
         except TypeError as exc:
-            result = f"TypeError encountered executing {self.fun}: {exc}"
+            result = {"local": f"TypeError encountered executing {self.fun}: {exc}"}
             log.error(result, exc_info_on_loglevel=logging.DEBUG)
             retcode = 1
         except Exception as exc:  # pylint: disable=broad-except
-            result = "An Exception occurred while executing {}: {}".format(
-                self.fun, exc
-            )
+            result = {
+                "local": f"An Exception occurred while executing {self.fun}: {exc}"
+            }
             log.error(result, exc_info_on_loglevel=logging.DEBUG)
             retcode = 1
 
@@ -1325,6 +1321,10 @@ def run_wfunc(self):
         # emit (as seen in ssh_py_shim.py)
         if isinstance(result, dict) and "local" in result:
             ret = salt.utils.json.dumps({"local": result["local"]})
+        elif self.context.get("retcode"):
+            # The wrapped command failed, the usual behavior is that
+            # the return is dumped as-is without declaring it as a result.
+            ret = salt.utils.json.dumps({"local": result})
         else:
             ret = salt.utils.json.dumps({"local": {"return": result}})
         return ret, retcode

salt/client/ssh/wrapper/__init__.py

@@ -7,11 +7,94 @@
 
 
 import copy
+import logging
 
 import salt.client.ssh
 import salt.loader
 import salt.utils.data
 import salt.utils.json
+from salt.defaults import NOT_SET
+from salt.exceptions import CommandExecutionError, SaltException
+
+log = logging.getLogger(__name__)
+
+
+class SSHException(SaltException):
+    """
+    Indicates general command failure via salt-ssh.
+    """
+
+    _error = ""
+
+    def __init__(
+        self, stdout, stderr, retcode, result=NOT_SET, data=None, *args, **kwargs
+    ):
+        super().__init__(stderr, *args, **kwargs)
+        self.stdout = stdout
+        self.stderr = stderr
+        self.result = result
+        self.data = data
+        self.retcode = retcode
+        if args:
+            self._error = args.pop(0)
+
+    def to_ret(self):
+        ret = {
+            "stdout": self.stdout,
+            "stderr": self.stderr,
+            "retcode": self.retcode,
+            "data": self.data,
+        }
+        if self._error:
+            ret["_error"] = self._error
+        if self.result is not NOT_SET:
+            ret["return"] = self.result
+        return ret
+
+
+class SSHCommandExecutionError(SSHException, CommandExecutionError):
+    """
+    Thrown whenever a non-zero exit code is returned.
+    This was introduced to make the salt-ssh FunctionWrapper behave
+    more like the usual one, in particular to force template rendering
+    to stop when a function call results in an exception.
+    """
+
+    _error = "The command resulted in a non-zero exit code"
+
+    def to_ret(self):
+        if self.data and "local" in self.data:
+            # Wrapped commands that indicate a non-zero retcode
+            return self.data["local"]
+        elif self.stderr:
+            # Remote executions that indicate a non-zero retcode
+            return self.stderr
+        return super().to_ret()
+
+
+class SSHPermissionDeniedError(SSHException):
+    """
+    Thrown when "Permission denied" is found in stderr
+    """
+
+    _error = "Permission Denied"
+
+
+class SSHReturnDecodeError(SSHException):
+    """
+    Thrown when JSON-decoding stdout fails and the retcode is 0 otherwise
+    """
+
+    _error = "Failed to return clean data"
+
+
+class SSHMalformedReturnError(SSHException):
+    """
+    Thrown when a decoded return dict is not formed as
+    {"local": {"return": ...}}
+    """
+
+    _error = "Return dict was malformed"
 
 
 class FunctionWrapper:
@@ -119,26 +202,7 @@ def caller(*args, **kwargs):
                 **self.kwargs
             )
             stdout, stderr, retcode = single.cmd_block()
-            if stderr.count("Permission Denied"):
-                return {
-                    "_error": "Permission Denied",
-                    "stdout": stdout,
-                    "stderr": stderr,
-                    "retcode": retcode,
-                }
-            try:
-                ret = salt.utils.json.loads(stdout)
-                if len(ret) < 2 and "local" in ret:
-                    ret = ret["local"]
-                ret = ret.get("return", {})
-            except ValueError:
-                ret = {
-                    "_error": "Failed to return clean data",
-                    "stderr": stderr,
-                    "stdout": stdout,
-                    "retcode": retcode,
-                }
-            return ret
+            return parse_ret(stdout, stderr, retcode, result_only=True)
 
         return caller
 
@@ -176,3 +240,67 @@ def get(self, cmd, default):
             return self[cmd]
         else:
             return default
+
+
+def parse_ret(stdout, stderr, retcode, result_only=False):
+    """
+    Parse the output of a remote or local command and return its
+    result. Raise exceptions if the command has a non-zero exitcode
+    or its output is not valid JSON or is not in the expected format,
+    usually ``{"local": {"return": value}}`` (+ optional keys in the "local" dict).
+    """
+    try:
+        retcode = int(retcode)
+    except (TypeError, ValueError):
+        log.warning(f"Got an invalid retcode for host: '{retcode}'")
+        retcode = 1
+
+    if retcode and stderr.count("Permission Denied"):
+        raise SSHPermissionDeniedError(stdout=stdout, stderr=stderr, retcode=retcode)
+
+    result = NOT_SET
+    error = None
+    data = None
+    local_data = None
+
+    try:
+        data = salt.utils.json.loads(stdout)
+        if len(data) < 2 and "local" in data:
+            try:
+                result = data["local"]
+                try:
+                    # Ensure a reported local retcode is kept (at least)
+                    retcode = max(retcode, result["retcode"])
+                except (KeyError, TypeError):
+                    pass
+                if not isinstance(data["local"], dict):
+                    # When a command has failed, the return is dumped as-is
+                    # without declaring it as a result, usually a string or list.
+                    error = SSHCommandExecutionError
+                elif result_only:
+                    result = result["return"]
+            except KeyError:
+                error = SSHMalformedReturnError
+        else:
+            error = SSHMalformedReturnError
+
+    except ValueError:
+        # No valid JSON output was found
+        error = SSHReturnDecodeError
+    if retcode:
+        raise SSHCommandExecutionError(
+            stdout=stdout,
+            stderr=stderr,
+            retcode=retcode,
+            result=result,
+            data=local_data if local_data is not None else data,
+        )
+    if error is not None:
+        raise error(
+            stdout=stdout,
+            stderr=stderr,
+            retcode=retcode,
+            result=result,
+            data=local_data if local_data is not None else data,
+        )
+    return result