[3002.4] Fix Runners in SSE (#175)

* Runners fix

* Rework runner fix

changelog/165.fixed

@@ -0,0 +1 @@
+Fix runners that broke when patching for CVE-2021-25281

salt/client/mixins.py

@@ -479,7 +479,7 @@ class AsyncClientMixin:
     client = None
     tag_prefix = None
 
-    def _proc_function(self, fun, low, user, tag, jid, daemonize=True):
+    def _proc_function_remote(self, fun, low, user, tag, jid, daemonize=True):
         """
         Run this method in a multiprocess target to execute the function on the
         master and fire the return data on the event bus
@@ -498,7 +498,31 @@ def _proc_function(self, fun, low, user, tag, jid, daemonize=True):
         low["__user__"] = user
         low["__tag__"] = tag
 
-        return self.low(fun, low, full_return=False)
+        try:
+            return self.cmd_sync(low)
+        except salt.exceptions.EauthAuthenticationError as exc:
+            log.error(exc)
+
+    def _proc_function(self, fun, low, user, tag, jid, daemonize=True):
+        """
+        Run this method in a multiprocess target to execute the function
+        locally and fire the return data on the event bus
+        """
+        if daemonize and not salt.utils.platform.is_windows():
+            # Shutdown the multiprocessing before daemonizing
+            salt.log.setup.shutdown_multiprocessing_logging()
+
+            salt.utils.process.daemonize()
+
+            # Reconfigure multiprocessing logging after daemonizing
+            salt.log.setup.setup_multiprocessing_logging()
+
+        # pack a few things into low
+        low["__jid__"] = jid
+        low["__user__"] = user
+        low["__tag__"] = tag
+
+        return self.low(fun, low)
 
     def _proc_function_local(self, fun, low, user, tag, jid, daemonize=True):
         """
@@ -547,15 +571,15 @@ def _gen_async_pub(self, jid=None):
         tag = salt.utils.event.tagify(jid, prefix=self.tag_prefix)
         return {"tag": tag, "jid": jid}
 
-    def asynchronous(self, fun, low, user="UNKNOWN", pub=None, local=False):
+    def asynchronous(self, fun, low, user="UNKNOWN", pub=None, local=True):
         """
         Execute the function in a multiprocess and return the event tag to use
         to watch for the return
         """
         if local:
-            proc_func = self._proc_function_local
-        else:
             proc_func = self._proc_function
+        else:
+            proc_func = self._proc_function_remote
         async_pub = pub if pub is not None else self._gen_async_pub()
         proc = salt.utils.process.SignalHandlingProcess(
             target=proc_func,

salt/wheel/__init__.py

@@ -123,8 +123,8 @@ def cmd_async(self, low):
             })
             {'jid': '20131219224744416681', 'tag': 'salt/wheel/20131219224744416681'}
         """
-        fun = low.pop("fun")
-        return self.asynchronous(fun, low)
+        fun = low.get("fun")
+        return self.asynchronous(fun, low, local=False)
 
     def cmd(
         self,