CVE-2024-38825

@@ -17,6 +17,7 @@
 import logging
 
 import salt.utils.files
+import salt.utils.versions
 
 # pylint: disable=import-error
 try:
@@ -30,7 +31,7 @@
             from Cryptodome.Util import asn1
         except ImportError:
             from Crypto.Util import asn1  # nosec
-        import OpenSSL
+        import OpenSSL  # pylint: disable=W8410
     HAS_DEPS = True
 except ImportError:
     HAS_DEPS = False
@@ -71,6 +72,10 @@ def auth(username, password, **kwargs):
             your_user:
               - .*
     """
+    salt.utils.versions.warn_until(
+        3008,
+        "This module has been deprecated as it is known to be insecure.",
+    )
     pem = password
     cacert_file = __salt__["config.get"]("external_auth:pki:ca_file")
 

@@ -152,13 +152,40 @@ def crypt(self):
     def ttype(self):
         return self.transport.ttype
 
-    def _package_load(self, load):
+    def _package_load(self, load, nonce=None):
+        """
+        Prepare the load to be sent over the wire.
+
+        For aes channels add a nonce, timestamp and signed token to the load
+        before encrypting it using our aes session key. Then wrap the encrypted
+        load with some meta data. For 'clear' encryption, no extra feilds are
+        added to the load. The unencyrpted load is wrapped with meta data.
+        """
+        if self.crypt == "aes":
+            if nonce is None:
+                nonce = uuid.uuid4().hex
+            try:
+                load["nonce"] = nonce
+                load["ts"] = int(time.time())
+                load["tok"] = self.auth.gen_token(b"salt")
+                load["id"] = self.opts["id"]
+            except TypeError:
+                # Backwards compatability for non dict loads, let the load get
+                # sent and fail to authenticate.
+                log.warning(
+                    "Invalid load passed to request channel. Type is %s should be dict.",
+                    type(load),
+                )
+
+            load = self.auth.session_crypticle.dumps(load)
+
         ret = {
             "enc": self.crypt,
             "load": load,
-            "version": 2,
+            "version": 3,
         }
         if self.crypt == "aes":
+            ret["id"] = self.opts["id"]
             ret["enc_algo"] = self.opts["encryption_algorithm"]
             ret["sig_algo"] = self.opts["signing_algorithm"]
         return ret
@@ -194,12 +221,12 @@ def crypted_transfer_decode_dictentry(
             timeout = self.timeout
         if tries is None:
             tries = self.tries
-        nonce = uuid.uuid4().hex
-        load["nonce"] = nonce
         if not self.auth.authenticated:
             yield self.auth.authenticate()
+
+        nonce = uuid.uuid4().hex
         ret = yield self._send_with_retry(
-            self._package_load(self.auth.crypticle.dumps(load)),
+            self._package_load(load, nonce),
             tries,
             timeout,
         )
@@ -208,7 +235,7 @@ def crypted_transfer_decode_dictentry(
             # Reauth in the case our key is deleted on the master side.
             yield self.auth.authenticate()
             ret = yield self._send_with_retry(
-                self._package_load(self.auth.crypticle.dumps(load)),
+                self._package_load(load, nonce),
                 tries,
                 timeout,
             )
@@ -253,23 +280,21 @@ def _crypted_transfer(self, load, timeout, raw=False):
         :param dict load: A load to send across the wire
         :param int timeout: The number of seconds on a response before failing
         """
-        nonce = uuid.uuid4().hex
-        if load and isinstance(load, dict):
-            load["nonce"] = nonce
 
         @tornado.gen.coroutine
         def _do_transfer():
             # Yield control to the caller. When send() completes, resume by populating data with the Future.result
+            nonce = uuid.uuid4().hex
             data = yield self.transport.send(
-                self._package_load(self.auth.crypticle.dumps(load)),
+                self._package_load(load, nonce),
                 timeout=timeout,
             )
             # we may not have always data
             # as for example for saltcall ret submission, this is a blind
             # communication, we do not subscribe to return events, we just
             # upload the results to the master
             if data:
-                data = self.auth.crypticle.loads(data, raw, nonce=nonce)
+                data = self.auth.session_crypticle.loads(data, raw, nonce=nonce)
             if not raw or self.ttype == "tcp":  # XXX Why is this needed for tcp
                 data = salt.transport.frame.decode_embedded_strs(data)
             raise tornado.gen.Return(data)
@@ -477,7 +502,7 @@ def _package_load(self, load):
         return {
             "enc": self.crypt,
             "load": load,
-            "version": 2,
+            "version": 3,
         }
 
     @tornado.gen.coroutine

@@ -12,6 +12,7 @@
 import os
 import pathlib
 import shutil
+import time
 
 import tornado.gen
 
@@ -57,17 +58,45 @@ def compare_keys(cls, key1, key2):
     def __init__(self, opts, transport):
         self.opts = opts
         self.transport = transport
+        # The event and master_key attributes will be populated after fork.
+        #        self.event = None
+        #        self.master_key = None
         self.event = salt.utils.event.get_master_event(
             self.opts, self.opts["sock_dir"], listen=False
         )
         self.master_key = salt.crypt.MasterKeys(self.opts)
 
+        (pathlib.Path(self.opts["cachedir"]) / "sessions").mkdir(exist_ok=True)
+        self.sessions = {}
+
     @property
     def aes_key(self):
         if self.opts.get("cluster_id", None):
             return salt.master.SMaster.secrets["cluster_aes"]["secret"].value
         return salt.master.SMaster.secrets["aes"]["secret"].value
 
+    def session_key(self, minion):
+        """
+        Returns a session key for the given minion id.
+        """
+        now = time.time()
+        if minion in self.sessions:
+            if now - self.sessions[minion][0] < self.opts["publish_session"]:
+                return self.sessions[minion][1]
+
+        path = pathlib.Path(self.opts["cachedir"]) / "sessions" / minion
+        try:
+            if now - path.stat().st_mtime > self.opts["publish_session"]:
+                salt.crypt.Crypticle.write_key(path)
+        except FileNotFoundError:
+            salt.crypt.Crypticle.write_key(path)
+
+        self.sessions[minion] = (
+            path.stat().st_mtime,
+            salt.crypt.Crypticle.read_key(path),
+        )
+        return self.sessions[minion][1]
+
     def pre_fork(self, process_manager):
         """
         Do anything necessary pre-fork. Since this is on the master side this will
@@ -112,8 +141,16 @@ def post_fork(self, payload_handler, io_loop):
 
     @tornado.gen.coroutine
     def handle_message(self, payload):
+        if (
+            not isinstance(payload, dict)
+            or "enc" not in payload
+            or "load" not in payload
+        ):
+            log.warn("bad load received on socket")
+            raise tornado.gen.Return("bad load")
+        version = payload.get("version", 0)
         try:
-            payload = self._decode_payload(payload)
+            payload = self._decode_payload(payload, version)
         except Exception as exc:  # pylint: disable=broad-except
             exc_type = type(exc).__name__
             if exc_type == "AuthenticationError":
@@ -145,22 +182,53 @@ def handle_message(self, payload):
             log.error("Payload contains non-string id: %s", payload)
             raise tornado.gen.Return(f"bad load: id {id_} is not a string")
 
-        version = 0
-        if "version" in payload:
-            version = payload["version"]
-
         sign_messages = False
         if version > 1:
             sign_messages = True
 
         # intercept the "_auth" commands, since the main daemon shouldn't know
         # anything about our key auth
         if payload["enc"] == "clear" and payload.get("load", {}).get("cmd") == "_auth":
-            raise tornado.gen.Return(self._auth(payload["load"], sign_messages))
+            raise tornado.gen.Return(
+                self._auth(payload["load"], sign_messages, version)
+            )
 
-        nonce = None
-        if version > 1:
-            nonce = payload["load"].pop("nonce", None)
+        if payload["enc"] == "aes":
+            nonce = None
+            if version > 1:
+                nonce = payload["load"].pop("nonce", None)
+
+            # Check validity of message ttl and id's match
+            if version > 2:
+                if self.opts["request_server_ttl"] > 0:
+                    ttl = time.time() - payload["load"]["ts"]
+                    if ttl > self.opts["request_server_ttl"]:
+                        log.warning(
+                            "Received request from %s with expired ttl: %d > %d",
+                            payload["load"]["id"],
+                            ttl,
+                            self.opts["request_server_ttl"],
+                        )
+                        raise tornado.gen.Return("bad load")
+
+                if payload["id"] != payload["load"]["id"]:
+                    log.warning(
+                        "Request id mismatch. Found '%s' but expected '%s'",
+                        payload["load"]["id"],
+                        payload["id"],
+                    )
+                    raise tornado.gen.Return("bad load")
+                if not salt.utils.verify.valid_id(self.opts, payload["load"]["id"]):
+                    log.warning(
+                        "Request contains invalid minion id '%s'", payload["load"]["id"]
+                    )
+                    raise tornado.gen.Return("bad load")
+                if not self.validate_token(payload, required=True):
+                    raise tornado.gen.Return("bad load")
+            # The token won't always be present in the payload for v2 and
+            # below, but if it is we always wanto validate it.
+            elif not self.validate_token(payload, required=False):
+                raise tornado.gen.Return("bad load")
 
         # TODO: test
         try:
@@ -176,7 +244,14 @@ def handle_message(self, payload):
         if req_fun == "send_clear":
             raise tornado.gen.Return(ret)
         elif req_fun == "send":
-            raise tornado.gen.Return(self.crypticle.dumps(ret, nonce))
+            if version > 2:
+                raise tornado.gen.Return(
+                    salt.crypt.Crypticle(self.opts, self.session_key(id_)).dumps(
+                        ret, nonce
+                    )
+                )
+            else:
+                raise tornado.gen.Return(self.crypticle.dumps(ret, nonce))
         elif req_fun == "send_private":
             raise tornado.gen.Return(
                 self._encrypt_private(
@@ -216,7 +291,8 @@ def _encrypt_private(
         try:
             pub = salt.crypt.PublicKey(pubfn)
         except (ValueError, IndexError, TypeError):
-            return self.crypticle.dumps({})
+            log.error("Bad load from minion")
+            return {"error": "bad load"}
         except OSError:
             log.error("AES key not found")
             return {"error": "AES key not found"}
@@ -279,27 +355,56 @@ def _update_aes(self):
             return True
         return False
 
-    def _decode_payload(self, payload):
-        # Sometimes msgpack deserialization of random bytes could be successful,
-        # so we need to ensure payload in good shape to process this function.
-        if (
-            not isinstance(payload, dict)
-            or "enc" not in payload
-            or "load" not in payload
-        ):
-            raise SaltDeserializationError("bad load received on socket!")
-
+    def _decode_payload(self, payload, version):
         # we need to decrypt it
         if payload["enc"] == "aes":
-            try:
-                payload["load"] = self.crypticle.loads(payload["load"])
-            except salt.crypt.AuthenticationError:
-                if not self._update_aes():
-                    raise
-                payload["load"] = self.crypticle.loads(payload["load"])
+            if version > 2:
+                if salt.utils.verify.valid_id(self.opts, payload["id"]):
+                    payload["load"] = salt.crypt.Crypticle(
+                        self.opts,
+                        self.session_key(payload["id"]),
+                    ).loads(payload["load"])
+                else:
+                    raise SaltDeserializationError("Encountered invalid id")
+            else:
+                try:
+                    payload["load"] = self.crypticle.loads(payload["load"])
+                except salt.crypt.AuthenticationError:
+                    if not self._update_aes():
+                        raise
+                    payload["load"] = self.crypticle.loads(payload["load"])
         return payload
 
-    def _auth(self, load, sign_messages=False):
+    def validate_token(self, payload, required=True):
+        if "tok" in payload["load"] and "id" in payload["load"]:
+            if "cluster_id" in self.opts and self.opts["cluster_id"]:
+                pki_dir = self.opts["cluster_pki_dir"]
+            else:
+                pki_dir = self.opts.get("pki_dir", "")
+            id_ = payload["load"]["id"]
+
+            pub_path = os.path.join(pki_dir, "minions", id_)
+            try:
+                pub = salt.crypt.PublicKey(pub_path)
+            except OSError:
+                log.warning(
+                    "Salt minion claiming to be %s attempted to communicate with "
+                    "master, but key could not be read and verification was denied.",
+                    id_,
+                )
+                return False
+            try:
+                if pub.decrypt(payload["load"]["tok"]) != b"salt":
+                    log.error("Minion token did not validate: %s", payload["id"])
+                    return False
+            except ValueError as err:
+                log.error("Unable to decrypt token: %s", err)
+                return False
+        elif required:
+            return False
+        return True
+
+    def _auth(self, load, sign_messages=False, version=0):
         """
         Authenticate the client, use the sent public key to encrypt the AES key
         which was generated at start up.
@@ -711,6 +816,7 @@ def _auth(self, load, sign_messages=False):
                 aes = self.aes_key
 
             ret["aes"] = pub.encrypt(aes, enc_algo)
+            ret["session"] = pub.encrypt(self.session_key(load["id"]), enc_algo)
         else:
             if "token" in load:
                 try:
@@ -730,6 +836,13 @@ def _auth(self, load, sign_messages=False):
 
             aes = self.aes_key
             ret["aes"] = pub.encrypt(aes, enc_algo)
+            ret["session"] = pub.encrypt(self.session_key(load["id"]), enc_algo)
+
+        if version < 3:
+            log.warning(
+                "Minion using legacy request server protocol, please upgrade %s",
+                load["id"],
+            )
 
         # Be aggressive about the signature
         digest = salt.utils.stringutils.to_bytes(hashlib.sha256(aes).hexdigest())

@@ -1013,6 +1013,8 @@ def _gather_buffer_space():
         "signing_algorithm": str,
         # Master publish channel signing
         "publish_signing_algorithm": str,
+        "request_server_ttl": int,
+        "request_server_aes_session": int,
     }
 )
 
@@ -1673,6 +1675,8 @@ def _gather_buffer_space():
         "cluster_pool_port": 4520,
         "features": {},
         "publish_signing_algorithm": "PKCS1v15-SHA1",
+        "request_server_aes_session": 0,
+        "request_server_ttl": 0,
     }
 )
 

@@ -454,7 +454,7 @@ def __init__(self, opts):
         self.cluster_pub_path = None
         self.cluster_rsa_path = None
         self.cluster_key = None
-        if self.opts["cluster_id"]:
+        if self.opts.get("cluster_id"):
             self.cluster_pub_path = os.path.join(
                 self.opts["cluster_pki_dir"], "cluster.pub"
             )
@@ -723,6 +723,7 @@ def __singleton_init__(self, opts, io_loop=None):
             creds = AsyncAuth.creds_map[key]
             self._creds = creds
             self._crypticle = Crypticle(self.opts, creds["aes"])
+            self._session_crypticle = Crypticle(self.opts, creds["session"])
             self._authenticate_future = tornado.concurrent.Future()
             self._authenticate_future.set_result(True)
 
@@ -739,6 +740,10 @@ def __deepcopy__(self, memo):
             setattr(result, key, copy.deepcopy(self.__dict__[key], memo))
         return result
 
+    @property
+    def session_crypticle(self):
+        return self._session_crypticle
+
     @property
     def creds(self):
         return self._creds
@@ -880,11 +885,13 @@ def _authenticate(self):
                     AsyncAuth.creds_map[key] = creds
                     self._creds = creds
                     self._crypticle = Crypticle(self.opts, creds["aes"])
+                    self._session_crypticle = Crypticle(self.opts, creds["session"])
                 elif self._creds["aes"] != creds["aes"]:
                     log.debug("%s The master's aes key has changed.", self)
                     AsyncAuth.creds_map[key] = creds
                     self._creds = creds
                     self._crypticle = Crypticle(self.opts, creds["aes"])
+                    self._session_crypticle = Crypticle(self.opts, creds["session"])
 
                 self._authenticate_future.set_result(
                     True
@@ -972,7 +979,6 @@ def handle_signin_response(self, sign_in_payload, payload):
         clear_signed_data = payload["load"]
         clear_signature = payload["sig"]
         payload = salt.payload.loads(clear_signed_data)
-
         if "pub_key" in payload:
             auth["aes"] = self.verify_master(
                 payload, master_pub="token" in sign_in_payload
@@ -991,6 +997,11 @@ def handle_signin_response(self, sign_in_payload, payload):
                 )
                 raise SaltClientError("Invalid master key")
 
+            key = self.get_keys()
+            auth["session"] = key.decrypt(
+                payload["session"], self.opts["encryption_algorithm"]
+            )
+
         master_pubkey_path = os.path.join(self.opts["pki_dir"], self.mpub)
         if os.path.exists(master_pubkey_path) and not PublicKey(
             master_pubkey_path
@@ -1541,10 +1552,12 @@ def authenticate(self, _=None):  # TODO: remove unused var
                 log.error("%s Got new master aes key.", self)
                 self._creds = creds
                 self._crypticle = Crypticle(self.opts, creds["aes"])
+                self._session_crypticle = Crypticle(self.opts, creds["session"])
             elif self._creds["aes"] != creds["aes"]:
                 log.error("%s The master's aes key has changed.", self)
                 self._creds = creds
                 self._crypticle = Crypticle(self.opts, creds["aes"])
+                self._session_crypticle = Crypticle(self.opts, creds["session"])
 
     def sign_in(self, timeout=60, safe=True, tries=1, channel=None):
         """

@@ -56,6 +56,27 @@
 # Things to do in lower layers:
 # only accept valid minion ids
 
+MINION_EVENT_BLACKLIST = (
+    "salt/job/*/publish",
+    "salt/job/*/new",
+    "salt/job/*/return",
+    "salt/key",
+    "salt/cloud/*",
+    "salt/run/*",
+    "salt/cluster/*",
+    "salt/wheel/*/new",
+    "salt/wheel/*/return",
+    "salt/run/*",
+    "salt/cloud/*",
+)
+
+
+def valid_minion_tag(tag, blacklist=MINION_EVENT_BLACKLIST):
+    for black in blacklist:
+        if fnmatch.fnmatch(tag, black):
+            return False
+    return True
+
 
 def init_git_pillar(opts):
     """
@@ -795,6 +816,9 @@ def _minion_event(self, load):
                     event_data = event["data"]
                 else:
                     event_data = event
+                if not valid_minion_tag(event["tag"]):
+                    log.warning("Filtering blacklisted event tag %s", event["tag"])
+                    continue
                 self.event.fire_event(event_data, event["tag"])  # old dup event
                 if load.get("pretag") is not None:
                     self.event.fire_event(

@@ -362,6 +362,12 @@ class AuthorizationError(SaltException):
     """
 
 
+class SaltValidationError(SaltException):
+    """
+    Thrown when a value fails validation
+    """
+
+
 class UnsupportedAlgorithm(SaltException):
     """
     Thrown when a requested encryption or signing algorithm is un-supported.

@@ -517,7 +517,7 @@ def file_envs(self, load=None):
         if load is None:
             load = {}
         load.pop("cmd", None)
-        return self.envs(**load)
+        return self.envs(back=load.get("back", None), sources=load.get("sources", None))
 
     def init(self, back=None):
         """

@@ -206,6 +206,35 @@ def rotate_cluster_secret(
             log.debug("Pinging all connected minions due to key rotation")
             salt.utils.master.ping_all_connected_minions(opts)
 
+    def populate_secrets(self):
+        if self.opts["cluster_id"]:
+            # Setup the secrets here because the PubServerChannel may need
+            # them as well.
+            SMaster.secrets["cluster_aes"] = {
+                "secret": multiprocessing.Array(
+                    ctypes.c_char,
+                    salt.utils.stringutils.to_bytes(self.read_or_generate_key()),
+                ),
+                "serial": multiprocessing.Value(
+                    ctypes.c_longlong,
+                    lock=False,  # We'll use the lock from 'secret'
+                ),
+                "reload": self.read_or_generate_key,
+            }
+
+        self.secrets["aes"] = {
+            "secret": multiprocessing.Array(
+                ctypes.c_char,
+                salt.utils.stringutils.to_bytes(
+                    salt.crypt.Crypticle.generate_key_string()
+                ),
+            ),
+            "serial": multiprocessing.Value(
+                ctypes.c_longlong, lock=False  # We'll use the lock from 'secret'
+            ),
+            "reload": salt.crypt.Crypticle.generate_key_string,
+        }
+
 
 class Maintenance(salt.utils.process.SignalHandlingProcess):
     """
@@ -782,33 +811,9 @@ def start(self):
         # manager. We don't want the processes being started to inherit those
         # signal handlers
         with salt.utils.process.default_signals(signal.SIGINT, signal.SIGTERM):
-            if self.opts["cluster_id"]:
-                # Setup the secrets here because the PubServerChannel may need
-                # them as well.
-                SMaster.secrets["cluster_aes"] = {
-                    "secret": multiprocessing.Array(
-                        ctypes.c_char,
-                        salt.utils.stringutils.to_bytes(self.read_or_generate_key()),
-                    ),
-                    "serial": multiprocessing.Value(
-                        ctypes.c_longlong,
-                        lock=False,  # We'll use the lock from 'secret'
-                    ),
-                    "reload": self.read_or_generate_key,
-                }
-
-            SMaster.secrets["aes"] = {
-                "secret": multiprocessing.Array(
-                    ctypes.c_char,
-                    salt.utils.stringutils.to_bytes(
-                        salt.crypt.Crypticle.generate_key_string()
-                    ),
-                ),
-                "serial": multiprocessing.Value(
-                    ctypes.c_longlong, lock=False  # We'll use the lock from 'secret'
-                ),
-                "reload": salt.crypt.Crypticle.generate_key_string,
-            }
+            # Setup the secrets here because the PubServerChannel may need
+            # them as well.
+            self.populate_secrets()
 
             log.info("Creating master process manager")
             # Since there are children having their own ProcessManager we should wait for kill more time.
@@ -1367,7 +1372,6 @@ class AESFuncs(TransportMethods):
         "_file_recv",
         "_pillar",
         "_minion_event",
-        "_handle_minion_event",
         "_return",
         "_syndic_return",
         "minion_runner",
@@ -1444,7 +1448,7 @@ def __verify_minion(self, id_, token):
         """
         if not salt.utils.verify.valid_id(self.opts, id_):
             return False
-        pub_path = os.path.join(self.pki_dir, "minions", id_)
+        pub_path = salt.utils.verify.clean_join(self.pki_dir, "minions", id_)
         try:
             pub = salt.crypt.PublicKey(pub_path)
         except OSError:
@@ -1733,11 +1737,10 @@ def _file_recv(self, load):
             # Can overwrite master files!!
             return False
 
-        cpath = os.path.join(
-            self.opts["cachedir"], "minions", load["id"], "files", normpath
-        )
+        rpath = os.path.join(self.opts["cachedir"], "minions", load["id"], "files")
+        cpath = os.path.join(rpath, normpath)
         # One last safety check here
-        if not os.path.normpath(cpath).startswith(self.opts["cachedir"]):
+        if not salt.utils.verify.clean_path(rpath, cpath):
             log.warning(
                 "Attempt to write received file outside of master cache "
                 "directory! Requested path: %s. Access denied.",
@@ -1865,8 +1868,8 @@ def _return(self, load):
         if "sig" in load:
             log.trace("Verifying signed event publish from minion")
             sig = load.pop("sig")
-            this_minion_pubkey = os.path.join(
-                self.pki_dir, "minions/{}".format(load["id"])
+            this_minion_pubkey = salt.utils.verify.clean_join(
+                self.pki_dir, "minions", load["id"]
             )
             serialized_load = salt.serializers.msgpack.serialize(load)
             if not salt.crypt.verify_signature(
@@ -1978,7 +1981,7 @@ def pub_ret(self, load):
         auth_cache = os.path.join(self.opts["cachedir"], "publish_auth")
         if not os.path.isdir(auth_cache):
             os.makedirs(auth_cache)
-        jid_fn = os.path.join(auth_cache, str(load["jid"]))
+        jid_fn = salt.utils.verify.clean_join(auth_cache, str(load["jid"]))
         with salt.utils.files.fopen(jid_fn, "r") as fp_:
             if not load["id"] == fp_.read():
                 return {}

@@ -437,7 +437,7 @@ def ext_pillar(minion_id, pillar, *repos):  # pylint: disable=unused-argument
     opts["__git_pillar"] = True
     git_pillar = salt.utils.gitfs.GitPillar(
         opts,
-        repos,
+        list(repos),
         per_remote_overrides=PER_REMOTE_OVERRIDES,
         per_remote_only=PER_REMOTE_ONLY,
         global_only=GLOBAL_ONLY,

@@ -610,10 +610,15 @@ def __init__(
 
     def __valid_on_demand_ext_pillar(self, opts):
         """
-        Check to see if the on demand external pillar is allowed
+        Check to see if the on demand external pillar is allowed.
+
+        If this check fails self.ext is set to None, this is important to
+        prevent an on-demand pillare from being rendered when it should not be
+        allowed.
         """
         if not isinstance(self.ext, dict):
             log.error("On-demand pillar %s is not formatted as a dictionary", self.ext)
+            self.ext = None
             return False
 
         on_demand = opts.get("on_demand_ext_pillar", [])
@@ -625,6 +630,7 @@ def __valid_on_demand_ext_pillar(self, opts):
                 "The 'on_demand_ext_pillar' configuration option is "
                 "malformed, it should be a list of ext_pillar module names"
             )
+            self.ext = None
             return False
 
         if invalid_on_demand:
@@ -636,6 +642,7 @@ def __valid_on_demand_ext_pillar(self, opts):
                 ", ".join(sorted(invalid_on_demand)),
                 ", ".join(on_demand),
             )
+            self.ext = None
             return False
         return True
 

@@ -8,6 +8,7 @@
 import glob
 import logging
 import os
+import pathlib
 import shutil
 import time
 
@@ -231,6 +232,8 @@ def save_minions(jid, minions, syndic_id=None):
     """
     Save/update the serialized list of minions for a given job
     """
+    import salt.utils.verify
+
     # Ensure we have a list for Python 3 compatibility
     minions = list(minions)
 
@@ -254,10 +257,20 @@ def save_minions(jid, minions, syndic_id=None):
         else:
             raise
 
-    if syndic_id is not None:
-        minions_path = os.path.join(jid_dir, SYNDIC_MINIONS_P.format(syndic_id))
-    else:
-        minions_path = os.path.join(jid_dir, MINIONS_P)
+    try:
+        if syndic_id is not None:
+            name = SYNDIC_MINIONS_P.format(syndic_id)
+        else:
+            name = MINIONS_P
+        minions_path = salt.utils.verify.clean_join(jid_dir, name)
+        target_name = pathlib.Path(minions_path).resolve().name
+        if name != target_name:
+            raise salt.exceptions.SaltValidationError(
+                f"Filenames do not match: {name} != {target_name}"
+            )
+    except salt.exceptions.SaltValidationError as exc:
+        log.error("Error %s", exc)
+        return
 
     try:
         if not os.path.exists(jid_dir):

@@ -169,6 +169,25 @@ def __maybe_string(ptr):
 LIBGIT2_MINVER = Version("0.20.0")
 
 
+def split_name(remote):
+    """
+    Given a string determine if it is a url or a name and url combination.
+
+    Examples:
+
+       None, "https://github.com/saltstack/salt.git" == split_name("https://github.com/saltstack/salt.git")
+
+       "__env__", "https://github.com/saltstack/salt.git" == split_name("__env__ https://github.com/saltstack/salt.git")
+    """
+    parts = remote.split(" ", 1)
+    if len(parts) == 1:
+        return None, remote
+    maybename, maybeurl = parts
+    if not salt.utils.verify.url(maybename):
+        return maybename, maybeurl
+    return None, remote
+
+
 def enforce_types(key, val):
     """
     Force params to be strings unless they should remain a different type
@@ -2625,7 +2644,27 @@ def init_remotes(
             per_remote_defaults[param] = enforce_types(key, self.opts[key])
 
         self.remotes = []
-        for remote in remotes:
+        # In case a tuple is passed.
+        remotes = list(remotes)
+        for remote in list(remotes):
+
+            if isinstance(remote, dict):
+                for key in list(remote):
+                    name, url = split_name(key)
+                    if not salt.utils.verify.url(url):
+                        log.warning("Found bad url data %r", key)
+                        remote.pop(key)
+                        continue
+                # None of the remotes were valid
+                if not remote:
+                    remotes.remove(remote)
+            else:
+                name, url = split_name(remote)
+                if not salt.utils.verify.url(url):
+                    log.warning("Found bad url data %r", remote)
+                    remotes.remove(remote)
+                    continue
+
             repo_obj = self.git_providers[self.provider](
                 self.opts,
                 remote,
@@ -3243,14 +3282,15 @@ def find_file(self, path, tgt_env="base", **kwargs):  # pylint: disable=W0613
         if os.path.isabs(path):
             return fnd
 
-        dest = salt.utils.path.join(self.cache_root, "refs", tgt_env, path)
-        hashes_glob = salt.utils.path.join(
+        # dest = salt.utils.path.join(self.cache_root, "refs", tgt_env, path)
+        dest = salt.utils.verify.clean_join(self.cache_root, "refs", tgt_env, path)
+        hashes_glob = salt.utils.verify.clean_join(
             self.hash_cachedir, tgt_env, f"{path}.hash.*"
         )
-        blobshadest = salt.utils.path.join(
+        blobshadest = salt.utils.verify.clean_join(
             self.hash_cachedir, tgt_env, f"{path}.hash.blob_sha1"
         )
-        lk_fn = salt.utils.path.join(self.hash_cachedir, tgt_env, f"{path}.lk")
+        lk_fn = salt.utils.verify.clean_join(self.hash_cachedir, tgt_env, f"{path}.lk")
         destdir = os.path.dirname(dest)
         hashdir = os.path.dirname(blobshadest)
         if not os.path.isdir(destdir):

@@ -10,14 +10,20 @@
 import socket
 import stat
 import sys
+import urllib.parse
 
 import salt.defaults.exitcodes
 import salt.utils.files
 import salt.utils.path
 import salt.utils.platform
 import salt.utils.user
 from salt._logging import LOG_LEVELS
-from salt.exceptions import CommandExecutionError, SaltClientError, SaltSystemExit
+from salt.exceptions import (
+    CommandExecutionError,
+    SaltClientError,
+    SaltSystemExit,
+    SaltValidationError,
+)
 
 # Original Author: Jeff Schroeder <jeffschroeder@computer.org>
 
@@ -533,22 +539,36 @@ def clean_path(root, path, subdir=False, realpath=True):
     """
     if not os.path.isabs(root):
         root = os.path.join(os.getcwd(), root)
-    root = os.path.normpath(root)
+    normroot = os.path.normpath(root)
     if not os.path.isabs(path):
-        path = os.path.join(root, path)
-    path = os.path.normpath(path)
+        path = os.path.join(normroot, path)
+    normpath = os.path.normpath(path)
     if realpath:
-        root = _realpath(root)
-        path = _realpath(path)
+        normroot = _realpath(normroot)
+        normpath = _realpath(normpath)
     if subdir:
-        if os.path.commonpath([path, root]) == root:
-            return path
+        if os.path.commonpath([normpath, normroot]) == normroot:
+            return normpath
     else:
-        if os.path.dirname(path) == root:
-            return path
+        if os.path.dirname(normpath) == normroot:
+            return normpath
     return ""
 
 
+def clean_join(root, *paths, subdir=False, realpath=True):
+    """
+    Performa a join and then check the result against the clean_path method. If
+    clean_path fails a SaltValidationError is raised.
+    """
+    parent = root
+    for path in paths:
+        child = os.path.join(parent, path)
+        if not clean_path(parent, child, subdir, realpath):
+            raise SaltValidationError(f"Invalid path: {path!r}")
+        parent = child
+    return child
+
+
 def valid_id(opts, id_):
     """
     Returns if the passed id is valid
@@ -777,3 +797,41 @@ def win_verify_env(path, dirs, permissive=False, pki_dir="", skip_extra=False):
     if skip_extra is False:
         # Run the extra verification checks
         zmq_version()
+
+
+SCHEMES = (
+    "http",
+    "https",
+    "ssh",
+    "ftp",
+    "sftp",
+    "file",
+)
+
+
+class URLValidator:
+
+    PCHAR = r"^([a-z0-9\-._~!$&'();=:@,]|%\d\d)+$"
+    ALL_VALID = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~:/?#[]@!$&'()*+,;="
+
+    @classmethod
+    def pchar_matcher(cls):
+        return re.compile(cls.PCHAR, re.IGNORECASE)
+
+    def __init__(self, schemes=SCHEMES):
+        self.schemes = schemes
+
+    def __call__(self, data):
+        if any([x not in self.ALL_VALID for x in data]):
+            return False
+        parsed = urllib.parse.urlparse(data)
+        if parsed.scheme not in self.schemes:
+            return False
+        matcher = self.pchar_matcher()
+        for part in parsed.path.split("/"):
+            if part and not matcher.match(part):
+                return False
+        return True
+
+
+url = URLValidator()

@@ -11,6 +11,7 @@
 import urllib.parse
 
 import salt.utils.files
+import salt.utils.verify
 
 # pylint: disable=E0611
 
@@ -64,10 +65,10 @@ def __init__(self, hyper, id_, opts):
         self.opts = opts
         self.hyper = hyper
         self.id = id_
-        path = os.path.join(self.opts["pki_dir"], "virtkeys", hyper)
+        path = salt.utils.verify.clean_join(self.opts["pki_dir"], "virtkeys", hyper)
         if not os.path.isdir(path):
             os.makedirs(path)
-        self.path = os.path.join(path, id_)
+        self.path = salt.utils.verify.clean_join(path, id_)
 
     def accept(self, pub):
         """
@@ -99,7 +100,7 @@ def accept(self, pub):
             )
             return False
 
-        pubfn = os.path.join(self.opts["pki_dir"], "minions", self.id)
+        pubfn = salt.utils.verify.clean_join(self.opts["pki_dir"], "minions", self.id)
         with salt.utils.files.fopen(pubfn, "w+") as fp_:
             fp_.write(pub)
         self.void()

@@ -997,7 +997,7 @@ def salt_syndic_master_factory(
     config_defaults["transport"] = request.config.getoption("--transport")
 
     config_overrides = {
-        "log_level_logfile": "quiet",
+        "log_level_logfile": "info",
         "fips_mode": FIPS_TESTRUN,
         "publish_signing_algorithm": (
             "PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1"
@@ -1078,7 +1078,7 @@ def salt_syndic_factory(salt_factories, salt_syndic_master_factory):
         opts["aliases.file"] = os.path.join(RUNTIME_VARS.TMP, "aliases")
         opts["transport"] = salt_syndic_master_factory.config["transport"]
         config_defaults["syndic"] = opts
-    config_overrides = {"log_level_logfile": "quiet"}
+    config_overrides = {"log_level_logfile": "info"}
     factory = salt_syndic_master_factory.salt_syndic_daemon(
         "syndic",
         defaults=config_defaults,
@@ -1116,7 +1116,7 @@ def salt_master_factory(
     config_defaults["transport"] = salt_syndic_master_factory.config["transport"]
 
     config_overrides = {
-        "log_level_logfile": "quiet",
+        "log_level_logfile": "info",
         "fips_mode": FIPS_TESTRUN,
         "publish_signing_algorithm": (
             "PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1"
@@ -1226,7 +1226,7 @@ def salt_minion_factory(salt_master_factory):
     config_defaults["transport"] = salt_master_factory.config["transport"]
 
     config_overrides = {
-        "log_level_logfile": "quiet",
+        "log_level_logfile": "info",
         "file_roots": salt_master_factory.config["file_roots"].copy(),
         "pillar_roots": salt_master_factory.config["pillar_roots"].copy(),
         "fips_mode": FIPS_TESTRUN,
@@ -1260,7 +1260,7 @@ def salt_sub_minion_factory(salt_master_factory):
     config_defaults["transport"] = salt_master_factory.config["transport"]
 
     config_overrides = {
-        "log_level_logfile": "quiet",
+        "log_level_logfile": "info",
         "file_roots": salt_master_factory.config["file_roots"].copy(),
         "pillar_roots": salt_master_factory.config["pillar_roots"].copy(),
         "fips_mode": FIPS_TESTRUN,

@@ -1,6 +1,9 @@
 import ctypes
 import logging
 import multiprocessing
+import pathlib
+import shutil
+import time
 
 import pytest
 import tornado.gen
@@ -9,6 +12,7 @@
 import salt.channel.client
 import salt.channel.server
 import salt.config
+import salt.crypt
 import salt.exceptions
 import salt.master
 import salt.utils.platform
@@ -100,6 +104,12 @@ def close(self):
     def _handle_payload(self, payload):
         if self.req_channel_crypt == "clear":
             raise tornado.gen.Return((payload, {"fun": "send_clear"}))
+        for key in (
+            "id",
+            "ts",
+            "tok",
+        ):
+            payload["load"].pop(key, None)
         raise tornado.gen.Return((payload, {"fun": "send"}))
 
 
@@ -117,6 +127,73 @@ def req_server_channel(salt_master, req_channel_crypt):
         )
 
 
+@pytest.fixture
+def req_server_opts(tmp_path):
+    sock_dir = tmp_path / "sock"
+    pki_dir = tmp_path / "pki"
+    cache_dir = tmp_path / "cache"
+    sock_dir.mkdir()
+    pki_dir.mkdir()
+    cache_dir.mkdir()
+    yield {
+        "sock_dir": sock_dir,
+        "pki_dir": pki_dir,
+        "cachedir": cache_dir,
+        "key_pass": "meh",
+        "keysize": 2048,
+        "cluster_id": None,
+        "master_sign_pubkey": False,
+        "pub_server_niceness": None,
+        "con_cache": False,
+        "zmq_monitor": False,
+        "request_server_ttl": 60,
+        "publish_session": 600,
+    }
+
+
+@pytest.fixture
+def req_server(req_server_opts):
+    server = salt.channel.server.ReqServerChannel.factory(req_server_opts)
+    try:
+        yield server
+    finally:
+        server.close()
+
+
+@pytest.fixture
+def minion1_id():
+    yield "minion1"
+
+
+@pytest.fixture
+def minion1_key(minion1_id, tmp_path, req_server_opts):
+    minionpki = tmp_path / minion1_id
+    minionpki.mkdir()
+    key1 = pathlib.Path(salt.crypt.gen_keys(minionpki, minion1_id, 2048))
+
+    pki = pathlib.Path(req_server_opts["pki_dir"])
+    (pki / "minions").mkdir(exist_ok=True)
+    shutil.copy2(key1.with_suffix(".pub"), pki / "minions" / minion1_id)
+    yield salt.crypt.PrivateKey(key1)
+
+
+@pytest.fixture
+def minion2_id():
+    yield "minion2"
+
+
+@pytest.fixture
+def minion2_key(minion2_id, tmp_path, req_server_opts):
+    minionpki = tmp_path / minion2_id
+    minionpki.mkdir()
+    key2 = pathlib.Path(salt.crypt.gen_keys(minionpki, minion2_id, 2048))
+
+    pki = pathlib.Path(req_server_opts["pki_dir"])
+    (pki / "minions").mkdir(exist_ok=True)
+    shutil.copy2(key2.with_suffix(".pub"), pki / "minions" / minion2_id)
+    yield salt.crypt.PrivateKey(key2)
+
+
 def req_channel_crypt_ids(value):
     return f"ReqChannel(crypt='{value}')"
 
@@ -142,6 +219,8 @@ def test_basic(push_channel):
     """
     Test a variety of messages, make sure we get the expected responses
     """
+    if push_channel.crypt == "aes":
+        pytest.skip(reason="test not valid for encrypted channel")
     msgs = [
         {"foo": "bar"},
         {"bar": "baz"},
@@ -156,6 +235,8 @@ def test_normalization(push_channel):
     """
     Since we use msgpack, we need to test that list types are converted to lists
     """
+    if push_channel.crypt == "aes":
+        pytest.skip(reason="test not valid for encrypted channel")
     types = {
         "list": list,
     }
@@ -172,6 +253,8 @@ def test_badload(push_channel, req_channel_crypt):
     """
     Test a variety of bad requests, make sure that we get some sort of error
     """
+    if push_channel.crypt == "aes":
+        pytest.skip(reason="test not valid for encrypted channel")
     msgs = ["", [], tuple()]
     if req_channel_crypt == "clear":
         for msg in msgs:
@@ -181,3 +264,186 @@ def test_badload(push_channel, req_channel_crypt):
         for msg in msgs:
             with pytest.raises(salt.exceptions.AuthenticationError):
                 push_channel.send(msg, timeout=5, tries=1)
+
+
+async def test_req_channel_ttl_v2(req_server, io_loop):
+    req_server.opts["request_server_ttl"] = 60
+
+    async def handler(payload):
+        return payload, {"fun": "send"}
+
+    req_server.post_fork(handler, io_loop)
+    payload = {
+        "enc": "aes",
+        "version": 2,
+        "load": req_server.crypticle.dumps(
+            {
+                "ts": int(time.time() - 61),
+            }
+        ),
+    }
+    ret = await req_server.handle_message(payload)
+    ret = req_server.crypticle.loads(ret)
+    assert ret == payload
+
+
+async def test_req_channel_ttl_valid(req_server, io_loop, minion1_id, minion1_key):
+    req_server.opts["request_server_ttl"] = 60
+    req_server.opts["publish_session"] = 600
+    tok = minion1_key.encrypt(b"salt")
+
+    async def handler(payload):
+        return payload, {"fun": "send"}
+
+    req_server.post_fork(handler, io_loop)
+    key = req_server.session_key(minion1_id)
+
+    crypticle = salt.crypt.Crypticle(req_server.opts, key)
+    payload = {
+        "enc": "aes",
+        "id": minion1_id,
+        "version": 3,
+        "load": crypticle.dumps(
+            {
+                "ts": int(time.time()),
+                "id": minion1_id,
+                "tok": tok,
+            }
+        ),
+    }
+    ret = await req_server.handle_message(payload)
+    ret = crypticle.loads(ret)
+    assert ret == payload
+
+
+async def test_req_channel_ttl_expired(
+    req_server, io_loop, caplog, minion1_id, minion1_key
+):
+    req_server.opts["request_server_ttl"] = 60
+    req_server.opts["publish_session"] = 600
+    tok = minion1_key.encrypt(b"salt")
+
+    async def handler(payload):
+        return payload, {"fun": "send"}
+
+    req_server.post_fork(handler, io_loop)
+    key = req_server.session_key(minion1_id)
+    crypticle = salt.crypt.Crypticle(req_server.opts, key)
+    payload = {
+        "enc": "aes",
+        "id": minion1_id,
+        "version": 3,
+        "load": crypticle.dumps(
+            {
+                "ts": int(time.time() - 61),
+                "id": minion1_id,
+                "tok": tok,
+            }
+        ),
+    }
+    with caplog.at_level(logging.WARNING):
+        ret = await req_server.handle_message(payload)
+        assert f"Received request from {minion1_id} with expired ttl" in caplog.text
+        assert ret == "bad load"
+
+
+async def test_req_channel_id_invalid_chars(
+    req_server, minion1_id, minion1_key, io_loop, caplog
+):
+    req_server.opts["request_server_ttl"] = 60
+    req_server.opts["publish_session"] = 600
+    tok = minion1_key.encrypt(b"salt")
+
+    async def handler(payload):
+        return payload, {"fun": "send"}
+
+    req_server.post_fork(handler, io_loop)
+    key = req_server.session_key(minion1_id)
+    crypticle = salt.crypt.Crypticle(req_server.opts, key)
+    payload = {
+        "enc": "aes",
+        "id": f"{minion1_id}\0",
+        "version": 3,
+        "load": crypticle.dumps(
+            {
+                "ts": int(time.time()),
+                "id": f"{minion1_id}\0",
+                "tok": tok,
+            }
+        ),
+    }
+    with caplog.at_level(logging.WARNING):
+        ret = await req_server.handle_message(payload)
+        assert (
+            "Bad load from minion: SaltDeserializationError: Encountered invalid id"
+            in caplog.text
+        )
+        assert ret == "bad load"
+
+
+async def test_req_channel_id_mismatch(
+    req_server, io_loop, caplog, minion1_id, minion1_key
+):
+
+    id2 = "minion2"
+
+    async def handler(payload):
+        return payload, {"fun": "send"}
+
+    req_server.post_fork(handler, io_loop)
+    key = req_server.session_key(minion1_id)
+    crypticle = salt.crypt.Crypticle(req_server.opts, key)
+    payload = {
+        "enc": "aes",
+        "id": minion1_id,
+        "version": 3,
+        "load": crypticle.dumps(
+            {
+                "ts": int(time.time()),
+                "id": id2,
+            }
+        ),
+    }
+    with caplog.at_level(logging.WARNING):
+        ret = await req_server.handle_message(payload)
+        assert (
+            f"Request id mismatch. Found '{id2}' but expected '{minion1_id}'"
+            in caplog.text
+        )
+        assert ret == "bad load"
+
+
+async def test_req_channel_v2_invalid_token(
+    req_server,
+    io_loop,
+    caplog,
+    tmp_path,
+    minion1_id,
+    minion1_key,
+    minion2_key,
+    minion2_id,
+):
+
+    tok2 = minion2_key.encrypt(b"salt")
+
+    async def handler(payload):
+        return payload, {"fun": "send"}
+
+    req_server.post_fork(handler, io_loop)
+
+    # Minion 1 is trying to impersonate minion2's token.
+    payload = {
+        "enc": "aes",
+        "version": 2,
+        "load": req_server.crypticle.dumps(
+            {
+                "ts": int(time.time()),
+                "id": minion1_id,
+                "tok": tok2,
+            }
+        ),
+    }
+    with caplog.at_level(logging.WARNING):
+        ret = await req_server.handle_message(payload)
+        assert "Unable to decrypt token:" in caplog.text
+        assert ret == "bad load"

@@ -10,7 +10,6 @@
 from pathlib import Path
 
 import pytest
-from pytestshellutils.utils import ports
 from saltfactories.utils import random_string
 
 import salt.channel.client
@@ -69,38 +68,27 @@ def transport(request):
 
 
 @pytest.fixture
-def master_config(root_dir, transport):
-    master_conf = salt.config.master_config("")
-    master_conf.update(
+def master_config(master_opts, transport):
+    master_opts.update(
         transport=transport,
         id="master",
-        root_dir=str(root_dir),
-        sock_dir=str(root_dir),
         interface="127.0.0.1",
-        publish_port=ports.get_unused_localhost_port(),
-        ret_port=ports.get_unused_localhost_port(),
-        pki_dir=str(root_dir / "pki"),
         fips_mode=FIPS_TESTRUN,
         publish_signing_algorithm=(
             "PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1"
         ),
     )
-    os.makedirs(master_conf["pki_dir"])
-    salt.crypt.gen_keys(master_conf["pki_dir"], "master", 4096)
-    minions_keys = os.path.join(master_conf["pki_dir"], "minions")
-    os.makedirs(minions_keys)
-    yield master_conf
+    salt.crypt.gen_keys(master_opts["pki_dir"], "master", 4096)
+    yield master_opts
 
 
 @pytest.fixture
-def minion_config(master_config, channel_minion_id):
-    minion_conf = salt.config.minion_config(
-        "", minion_id=channel_minion_id, cache_minion_id=False
-    )
-    minion_conf.update(
+def minion_config(minion_opts, master_config, channel_minion_id):
+    minion_opts.update(
         transport=master_config["transport"],
         root_dir=master_config["root_dir"],
         id=channel_minion_id,
+        cachedir=master_config["cachedir"],
         sock_dir=master_config["sock_dir"],
         ret_port=master_config["ret_port"],
         interface="127.0.0.1",
@@ -112,12 +100,13 @@ def minion_config(master_config, channel_minion_id):
         encryption_algorithm="OAEP-SHA224" if FIPS_TESTRUN else "OAEP-SHA1",
         signing_algorithm="PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1",
     )
-    os.makedirs(minion_conf["pki_dir"])
-    salt.crypt.gen_keys(minion_conf["pki_dir"], "minion", 4096)
-    minion_pub = os.path.join(minion_conf["pki_dir"], "minion.pub")
+    pathlib.Path(minion_opts["pki_dir"]).mkdir(exist_ok=True)
+    pathlib.Path(master_config["pki_dir"]).mkdir(exist_ok=True)
+    salt.crypt.gen_keys(minion_opts["pki_dir"], "minion", 4096)
+    minion_pub = os.path.join(minion_opts["pki_dir"], "minion.pub")
     pub_on_master = os.path.join(master_config["pki_dir"], "minions", channel_minion_id)
     shutil.copyfile(minion_pub, pub_on_master)
-    return minion_conf
+    return minion_opts
 
 
 @pytest.fixture

@@ -0,0 +1,47 @@
+import logging
+
+import salt.channel.client
+import salt.config
+import salt.crypt
+import salt.utils.args
+import salt.utils.jid
+
+log = logging.getLogger(__name__)
+
+
+def test_minoin_event_blacklist(salt_master, salt_minion, salt_cli, caplog):
+    ret = salt_cli.run("test.ping", minion_tgt=salt_minion.id)
+    assert ret.returncode == 0
+
+    opts = salt.config.minion_config(salt_minion.config_file)
+    opts["master_uri"] = "tcp://{}:{}".format(opts["master"], opts["master_port"])
+
+    jid = salt.utils.jid.gen_jid(opts)
+    auth = salt.crypt.SAuth(opts)
+    tok = auth.gen_token(b"salt")
+
+    load = {
+        "cmd": "_minion_event",
+        "tok": tok,
+        "id": opts["id"],
+        "events": [
+            {
+                "data": {
+                    "fun": "test.ping",
+                    "arg": [],
+                    "jid": jid,
+                    "ret": "",
+                    "tgt": salt_minion.id,
+                    "tgt_type": "glob",
+                    "user": "root",
+                    "__peer_id": "salt",
+                },
+                "tag": f"salt/job/{jid}/publish",
+            }
+        ],
+    }
+    with caplog.at_level(logging.WARNING):
+        with salt.channel.client.ReqChannel.factory(opts) as channel:
+            channel.send(load, tries=1, timeout=10000)
+            log.info("payload sent, jid was %s", jid)
+        assert "Filtering blacklisted" in caplog.text

@@ -0,0 +1,29 @@
+import getpass
+import pathlib
+
+import salt.channel.client
+
+
+def test_file_recv_path(salt_master, salt_minion):
+    config = salt_minion.config.copy()
+    config["master_uri"] = f"tcp://127.0.0.1:{salt_master.config['ret_port']}"
+    keyfile = f".{getpass.getuser()}_key"
+    data = b"asdf"
+    load_path_list = ["..", "..", "..", keyfile]
+    cachedir = salt_master.config["cachedir"]
+    assert (pathlib.Path(cachedir) / keyfile).exists()
+    assert (pathlib.Path(cachedir) / keyfile).read_bytes() != data
+    with salt.channel.client.ReqChannel.factory(config, crypt="aes") as channel:
+        load = {
+            "cmd": "_file_recv",
+            "id": salt_minion.config["id"],
+            "path": load_path_list,
+            "size": len(data),
+            "tok": channel.auth.gen_token(b"salt"),
+            "loc": 0,
+            "data": b"asdf",
+        }
+        ret = channel.send(load)
+    assert ret is False
+    assert (pathlib.Path(cachedir) / keyfile).exists()
+    assert (pathlib.Path(cachedir) / keyfile).read_bytes() != data

@@ -0,0 +1,23 @@
+import pytest
+
+import salt.daemons.masterapi
+
+
+@pytest.mark.parametrize(
+    "tag, valid",
+    [
+        ("salt/job/20160829225914848058/publish", False),
+        ("salt/key", False),
+        ("salt/cluster/fobar", False),
+        ("salt/job/20160829225914848058/return", False),
+        ("salt/job/20160829225914848058/new", False),
+        ("salt/wheel/20160829225914848058/new", False),
+        ("salt/run/20160829225914848058/new", False),
+        ("salt/run/20160829225914848058/ret", False),
+        ("salt/run/20160829225914848058/args", False),
+        ("salt/cloud/20160829225914848058/new", False),
+        ("salt/cloud/20160829225914848058/ret", False),
+    ],
+)
+def test_valid_minion_tag(tag, valid):
+    assert salt.daemons.masterapi.valid_minion_tag(tag) is valid

@@ -127,6 +127,8 @@ def mock_dumps(*args):
             # Crypticle must return bytes to pass to transport.RequestClient.send
             client.auth._crypticle = Mock()
             client.auth._crypticle.dumps = mock_dumps
+            client.auth._session_crypticle = Mock()
+            client.auth._session_crypticle.dumps = mock_dumps
             msg = r"^File client timed out after \d{1,4} seconds$"
             with pytest.raises(salt.exceptions.SaltClientError, match=msg):
                 client.file_list()

@@ -185,7 +185,7 @@ def cache_dir(tmp_path):
 def configure_loader_modules(provider, sock_dir, repo_dir, cache_dir):
     opts = {
         "sock_dir": str(sock_dir),
-        "gitfs_remotes": ["file://" + str(repo_dir)],
+        "gitfs_remotes": [f"file://{repo_dir}"],
         "cachedir": str(cache_dir),
         "gitfs_root": "",
         "fileserver_backend": ["gitfs"],
@@ -346,7 +346,7 @@ def test_ref_types_per_remote(repo_dir, unicode_dirname, tag_name):
     Test the per_remote ref_types config option, using a different
     ref_types setting than the global test.
     """
-    remotes = [{"file://" + repo_dir: [{"ref_types": ["tag"]}]}]
+    remotes = [{f"file://{repo_dir}": [{"ref_types": ["tag"]}]}]
     with patch.dict(gitfs.__opts__, {"gitfs_remotes": remotes}):
         gitfs.update()
         ret = gitfs.envs(ignore_cache=True)
@@ -433,7 +433,7 @@ def test_disable_saltenv_mapping_global_with_mapping_defined_per_remote(repo_dir
     opts = {
         "gitfs_disable_saltenv_mapping": True,
         "gitfs_remotes": [
-            {repo_dir: [{"saltenv": [{"bar": [{"ref": "somebranch"}]}]}]}
+            {f"file://{repo_dir}": [{"saltenv": [{"bar": [{"ref": "somebranch"}]}]}]}
         ],
     }
     with patch.dict(gitfs.__opts__, opts):
@@ -452,7 +452,7 @@ def test_disable_saltenv_mapping_per_remote_with_mapping_defined_globally(repo_d
     option.
     """
     opts = {
-        "gitfs_remotes": [{repo_dir: [{"disable_saltenv_mapping": True}]}],
+        "gitfs_remotes": [{f"file://{repo_dir}": [{"disable_saltenv_mapping": True}]}],
         "gitfs_saltenv": [{"hello": [{"ref": "somebranch"}]}],
     }
 
@@ -474,7 +474,7 @@ def test_disable_saltenv_mapping_per_remote_with_mapping_defined_per_remote(repo
     opts = {
         "gitfs_remotes": [
             {
-                repo_dir: [
+                f"file://{repo_dir}": [
                     {"disable_saltenv_mapping": True},
                     {"saltenv": [{"world": [{"ref": "somebranch"}]}]},
                 ]

@@ -6,9 +6,13 @@
 
 import pytest
 
+import salt.config
+import salt.crypt
 import salt.master
+import salt.utils.files
 import salt.utils.platform
 from tests.support.mock import MagicMock, patch
+from tests.support.runtests import RUNTIME_VARS
 
 
 @pytest.fixture
@@ -67,12 +71,14 @@ def cluster_maintenance(cluster_maintenance_opts):
 @pytest.fixture
 def encrypted_requests(tmp_path):
     # To honor the comment on AESFuncs
+    (tmp_path / "pki").mkdir()
     return salt.master.AESFuncs(
         opts={
+            "pki_dir": str(tmp_path / "pki"),
             "cachedir": str(tmp_path / "cache"),
             "sock_dir": str(tmp_path / "sock_drawer"),
             "conf_file": str(tmp_path / "config.conf"),
-            "fileserver_backend": "local",
+            "fileserver_backend": ["local"],
             "master_job_cache": False,
         }
     )
@@ -310,6 +316,7 @@ def test_aes_funcs_black(master_opts):
         "destroy",
         "get_method",
         "run_func",
+        "_handle_minion_event",
     ]
     try:
         for name in dir(aes_funcs):
@@ -1078,3 +1085,154 @@ def test_syndic_return_cache_dir_creation_traversal(encrypted_requests):
     )
     assert not (cachedir / "syndics").exists()
     assert not (cachedir / "mamajama").exists()
+
+
+def test_pub_ret_traversal(encrypted_requests, tmp_path):
+    """
+    master's  AESFuncs._syndic_return method cachdir creation is not vulnerable to a directory traversal
+    """
+    salt.crypt.gen_keys(tmp_path, "minion", 2048)
+
+    minions = pathlib.Path(encrypted_requests.opts["pki_dir"]) / "minions"
+    minions.mkdir()
+
+    with salt.utils.files.fopen(minions / "minion", "wb") as wfp:
+        with salt.utils.files.fopen(tmp_path / "minion.pub", "rb") as rfp:
+            wfp.write(rfp.read())
+
+    priv = salt.crypt.PrivateKey(tmp_path / "minion.pem")
+    with pytest.raises(salt.exceptions.SaltValidationError):
+        encrypted_requests.pub_ret(
+            {
+                "tok": priv.encrypt(b"salt"),
+                "id": "minion",
+                "jid": "asdf/../../../sdf",
+                "return": {},
+            }
+        )
+
+
+def _git_pillar_base_config(tmp_path):
+    return {
+        "__role": "master",
+        "pki_dir": str(tmp_path / "pki"),
+        "cachedir": str(tmp_path / "cache"),
+        "sock_dir": str(tmp_path / "sock_drawer"),
+        "conf_file": str(tmp_path / "config.conf"),
+        "fileserver_backend": ["local"],
+        "master_job_cache": False,
+        "file_client": "local",
+        "pillar_cache": False,
+        "state_top": "top.sls",
+        "pillar_roots": {
+            "base": [str(tmp_path / "pillar")],
+        },
+        "render_dirs": [str(pathlib.Path(RUNTIME_VARS.SALT_CODE_DIR) / "renderer")],
+        "renderer": "jinja|yaml",
+        "renderer_blacklist": [],
+        "renderer_whitelist": [],
+        "optimization_order": [0, 1, 2],
+        "on_demand_ext_pillar": [],
+        "git_pillar_user": "",
+        "git_pillar_password": "",
+        "git_pillar_pubkey": "",
+        "git_pillar_privkey": "",
+        "git_pillar_passphrase": "",
+        "git_pillar_insecure_auth": False,
+        "git_pillar_refspecs": salt.config._DFLT_REFSPECS,
+        "git_pillar_ssl_verify": True,
+        "git_pillar_branch": "master",
+        "git_pillar_base": "master",
+        "git_pillar_root": "",
+        "git_pillar_env": "",
+        "git_pillar_fallback": "",
+    }
+
+
+@pytest.fixture
+def allowed_funcs(tmp_path):
+    """
+    Configuration with git on demand pillar allowed
+    """
+    opts = _git_pillar_base_config(tmp_path)
+    opts["on_demand_ext_pillar"] = ["git"]
+    salt.crypt.gen_keys(str(tmp_path), "minion", 2048)
+    master_pki = tmp_path / "pki"
+    master_pki.mkdir()
+    accepted_pki = master_pki / "minions"
+    accepted_pki.mkdir()
+    (accepted_pki / "minion.pub").write_text((tmp_path / "minion.pub").read_text())
+
+    return salt.master.AESFuncs(opts=opts)
+
+
+def test_on_demand_allowed_command_injection(allowed_funcs, tmp_path, caplog):
+    """
+    Verify on demand pillars validate remote urls
+    """
+    pwnpath = tmp_path / "pwn"
+    assert not pwnpath.exists()
+    load = {
+        "cmd": "_pillar",
+        "saltenv": "base",
+        "pillarenv": "base",
+        "id": "carbon",
+        "grains": {},
+        "ver": 2,
+        "ext": {
+            "git": [
+                f'base ssh://fake@git/repo\n[core]\nsshCommand = touch {pwnpath}\n[remote "origin"]\n'
+            ]
+        },
+        "clean_cache": True,
+    }
+    with caplog.at_level(level="WARNING"):
+        ret = allowed_funcs._pillar(load)
+    assert not pwnpath.exists()
+    assert "Found bad url data" in caplog.text
+
+
+@pytest.fixture
+def not_allowed_funcs(tmp_path):
+    """
+    Configuration with no on demand pillars allowed
+    """
+    opts = _git_pillar_base_config(tmp_path)
+    opts["on_demand_ext_pillar"] = []
+    salt.crypt.gen_keys(str(tmp_path), "minion", 2048)
+    master_pki = tmp_path / "pki"
+    master_pki.mkdir()
+    accepted_pki = master_pki / "minions"
+    accepted_pki.mkdir()
+    (accepted_pki / "minion.pub").write_text((tmp_path / "minion.pub").read_text())
+
+    return salt.master.AESFuncs(opts=opts)
+
+
+def test_on_demand_not_allowed(not_allowed_funcs, tmp_path, caplog):
+    """
+    Verify on demand pillars do not render when not allowed
+    """
+    pwnpath = tmp_path / "pwn"
+    assert not pwnpath.exists()
+    load = {
+        "cmd": "_pillar",
+        "saltenv": "base",
+        "pillarenv": "base",
+        "id": "carbon",
+        "grains": {},
+        "ver": 2,
+        "ext": {
+            "git": [
+                f'base ssh://fake@git/repo\n[core]\nsshCommand = touch {pwnpath}\n[remote "origin"]\n'
+            ]
+        },
+        "clean_cache": True,
+    }
+    with caplog.at_level(level="WARNING"):
+        ret = not_allowed_funcs._pillar(load)
+    assert not pwnpath.exists()
+    assert (
+        "The following ext_pillar modules are not allowed for on-demand pillar data: git."
+        in caplog.text
+    )

@@ -1,1701 +0,0 @@
-"""
-    :codeauthor: Thomas Jackson <jacksontj.89@gmail.com>
-"""
-
-import asyncio
-import ctypes
-import logging
-import multiprocessing
-import os
-import threading
-import time
-import uuid
-
-import pytest
-import tornado.gen
-import tornado.ioloop
-
-import salt.channel.client
-import salt.channel.server
-import salt.crypt
-import salt.transport.zeromq
-import salt.utils.process
-import salt.utils.stringutils
-from salt.master import SMaster
-from tests.conftest import FIPS_TESTRUN
-from tests.support.mock import MagicMock, patch
-
-log = logging.getLogger(__name__)
-
-
-pytestmark = [
-    pytest.mark.core_test,
-]
-
-
-MASTER_PRIV_KEY = """
------BEGIN RSA PRIVATE KEY-----
-MIIEogIBAAKCAQEAoAsMPt+4kuIG6vKyw9r3+OuZrVBee/2vDdVetW+Js5dTlgrJ
-aghWWn3doGmKlEjqh7E4UTa+t2Jd6w8RSLnyHNJ/HpVhMG0M07MF6FMfILtDrrt8
-ZX7eDVt8sx5gCEpYI+XG8Y07Ga9i3Hiczt+fu6HYwu96HggmG2pqkOrn3iGfqBvV
-YVFJzSZYe7e4c1PeEs0xYcrA4k+apyGsMtpef8vRUrNicRLc7dAcvfhtgt2DXEZ2
-d72t/CR4ygtUvPXzisaTPW0G7OWAheCloqvTIIPQIjR8htFxGTz02STVXfnhnJ0Z
-k8KhqKF2v1SQvIYxsZU7jaDgl5i3zpeh58cYOwIDAQABAoIBABZUJEO7Y91+UnfC
-H6XKrZEZkcnH7j6/UIaOD9YhdyVKxhsnax1zh1S9vceNIgv5NltzIsfV6vrb6v2K
-Dx/F7Z0O0zR5o+MlO8ZncjoNKskex10gBEWG00Uqz/WPlddiQ/TSMJTv3uCBAzp+
-S2Zjdb4wYPUlgzSgb2ygxrhsRahMcSMG9PoX6klxMXFKMD1JxiY8QfAHahPzQXy9
-F7COZ0fCVo6BE+MqNuQ8tZeIxu8mOULQCCkLFwXmkz1FpfK/kNRmhIyhxwvCS+z4
-JuErW3uXfE64RLERiLp1bSxlDdpvRO2R41HAoNELTsKXJOEt4JANRHm/CeyA5wsh
-NpscufUCgYEAxhgPfcMDy2v3nL6KtkgYjdcOyRvsAF50QRbEa8ldO+87IoMDD/Oe
-osFERJ5hhyyEO78QnaLVegnykiw5DWEF02RKMhD/4XU+1UYVhY0wJjKQIBadsufB
-2dnaKjvwzUhPh5BrBqNHl/FXwNCRDiYqXa79eWCPC9OFbZcUWWq70s8CgYEAztOI
-61zRfmXJ7f70GgYbHg+GA7IrsAcsGRITsFR82Ho0lqdFFCxz7oK8QfL6bwMCGKyk
-nzk+twh6hhj5UNp18KN8wktlo02zTgzgemHwaLa2cd6xKgmAyuPiTgcgnzt5LVNG
-FOjIWkLwSlpkDTl7ZzY2QSy7t+mq5d750fpIrtUCgYBWXZUbcpPL88WgDB7z/Bjg
-dlvW6JqLSqMK4b8/cyp4AARbNp12LfQC55o5BIhm48y/M70tzRmfvIiKnEc/gwaE
-NJx4mZrGFFURrR2i/Xx5mt/lbZbRsmN89JM+iKWjCpzJ8PgIi9Wh9DIbOZOUhKVB
-9RJEAgo70LvCnPTdS0CaVwKBgDJW3BllAvw/rBFIH4OB/vGnF5gosmdqp3oGo1Ik
-jipmPAx6895AH4tquIVYrUl9svHsezjhxvjnkGK5C115foEuWXw0u60uiTiy+6Pt
-2IS0C93VNMulenpnUrppE7CN2iWFAiaura0CY9fE/lsVpYpucHAWgi32Kok+ZxGL
-WEttAoGAN9Ehsz4LeQxEj3x8wVeEMHF6OsznpwYsI2oVh6VxpS4AjgKYqeLVcnNi
-TlZFsuQcqgod8OgzA91tdB+Rp86NygmWD5WzeKXpCOg9uA+y/YL+0sgZZHsuvbK6
-PllUgXdYxqClk/hdBFB7v9AQoaj7K9Ga22v32msftYDQRJ94xOI=
------END RSA PRIVATE KEY-----
-"""
-
-
-MASTER_PUB_KEY = """
------BEGIN PUBLIC KEY-----
-MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAoAsMPt+4kuIG6vKyw9r3
-+OuZrVBee/2vDdVetW+Js5dTlgrJaghWWn3doGmKlEjqh7E4UTa+t2Jd6w8RSLny
-HNJ/HpVhMG0M07MF6FMfILtDrrt8ZX7eDVt8sx5gCEpYI+XG8Y07Ga9i3Hiczt+f
-u6HYwu96HggmG2pqkOrn3iGfqBvVYVFJzSZYe7e4c1PeEs0xYcrA4k+apyGsMtpe
-f8vRUrNicRLc7dAcvfhtgt2DXEZ2d72t/CR4ygtUvPXzisaTPW0G7OWAheCloqvT
-IIPQIjR8htFxGTz02STVXfnhnJ0Zk8KhqKF2v1SQvIYxsZU7jaDgl5i3zpeh58cY
-OwIDAQAB
------END PUBLIC KEY-----
-"""
-
-MASTER2_PRIV_KEY = """
------BEGIN RSA PRIVATE KEY-----
-MIIEogIBAAKCAQEAp+8cTxguO6Vg+YO92VfHgNld3Zy8aM3JbZvpJcjTnis+YFJ7
-Zlkcc647yPRRwY9nYBNywahnt5kIeuT1rTvTsMBZWvmUoEVUj1Xg8XXQkBvb9Ozy
-Gqy/G/p8KDDpzMP/U+XCnUeHiXTZrgnqgBIc2cKeCVvWFqDi0GRFGzyaXLaX3PPm
-M7DJ0MIPL1qgmcDq6+7Ze0gJ9SrDYFAeLmbuT1OqDfufXWQl/82JXeiwU2cOpqWq
-7n5fvPOWim7l1tzQ+dSiMRRm0xa6uNexCJww3oJSwvMbAmgzvOhqqhlqv+K7u0u7
-FrFFojESsL36Gq4GBrISnvu2tk7u4GGNTYYQbQIDAQABAoIBAADrqWDQnd5DVZEA
-lR+WINiWuHJAy/KaIC7K4kAMBgbxrz2ZbiY9Ok/zBk5fcnxIZDVtXd1sZicmPlro
-GuWodIxdPZAnWpZ3UtOXUayZK/vCP1YsH1agmEqXuKsCu6Fc+K8VzReOHxLUkmXn
-FYM+tixGahXcjEOi/aNNTWitEB6OemRM1UeLJFzRcfyXiqzHpHCIZwBpTUAsmzcG
-QiVDkMTKubwo/m+PVXburX2CGibUydctgbrYIc7EJvyx/cpRiPZXo1PhHQWdu4Y1
-SOaC66WLsP/wqvtHo58JQ6EN/gjSsbAgGGVkZ1xMo66nR+pLpR27coS7o03xCks6
-DY/0mukCgYEAuLIGgBnqoh7YsOBLd/Bc1UTfDMxJhNseo+hZemtkSXz2Jn51322F
-Zw/FVN4ArXgluH+XsOhvG/MFFpojwZSrb0Qq5b1MRdo9qycq8lGqNtlN1WHqosDQ
-zW29kpL0tlRrSDpww3wRESsN9rH5XIrJ1b3ZXuO7asR+KBVQMy/+NcUCgYEA6MSC
-c+fywltKPgmPl5j0DPoDe5SXE/6JQy7w/vVGrGfWGf/zEJmhzS2R+CcfTTEqaT0T
-Yw8+XbFgKAqsxwtE9MUXLTVLI3sSUyE4g7blCYscOqhZ8ItCUKDXWkSpt++rG0Um
-1+cEJP/0oCazG6MWqvBC4NpQ1nzh46QpjWqMwokCgYAKDLXJ1p8rvx3vUeUJW6zR
-dfPlEGCXuAyMwqHLxXgpf4EtSwhC5gSyPOtx2LqUtcrnpRmt6JfTH4ARYMW9TMef
-QEhNQ+WYj213mKP/l235mg1gJPnNbUxvQR9lkFV8bk+AGJ32JRQQqRUTbU+yN2MQ
-HEptnVqfTp3GtJIultfwOQKBgG+RyYmu8wBP650izg33BXu21raEeYne5oIqXN+I
-R5DZ0JjzwtkBGroTDrVoYyuH1nFNEh7YLqeQHqvyufBKKYo9cid8NQDTu+vWr5UK
-tGvHnwdKrJmM1oN5JOAiq0r7+QMAOWchVy449VNSWWV03aeftB685iR5BXkstbIQ
-EVopAoGAfcGBTAhmceK/4Q83H/FXBWy0PAa1kZGg/q8+Z0KY76AqyxOVl0/CU/rB
-3tO3sKhaMTHPME/MiQjQQGoaK1JgPY6JHYvly2KomrJ8QTugqNGyMzdVJkXAK2AM
-GAwC8ivAkHf8CHrHa1W7l8t2IqBjW1aRt7mOW92nfG88Hck0Mbo=
------END RSA PRIVATE KEY-----
-"""
-
-
-MASTER2_PUB_KEY = """
------BEGIN PUBLIC KEY-----
-MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAp+8cTxguO6Vg+YO92VfH
-gNld3Zy8aM3JbZvpJcjTnis+YFJ7Zlkcc647yPRRwY9nYBNywahnt5kIeuT1rTvT
-sMBZWvmUoEVUj1Xg8XXQkBvb9OzyGqy/G/p8KDDpzMP/U+XCnUeHiXTZrgnqgBIc
-2cKeCVvWFqDi0GRFGzyaXLaX3PPmM7DJ0MIPL1qgmcDq6+7Ze0gJ9SrDYFAeLmbu
-T1OqDfufXWQl/82JXeiwU2cOpqWq7n5fvPOWim7l1tzQ+dSiMRRm0xa6uNexCJww
-3oJSwvMbAmgzvOhqqhlqv+K7u0u7FrFFojESsL36Gq4GBrISnvu2tk7u4GGNTYYQ
-bQIDAQAB
------END PUBLIC KEY-----
-"""
-
-
-MASTER_SIGNING_PRIV = """
------BEGIN RSA PRIVATE KEY-----
-MIIEpAIBAAKCAQEAtieqrBMTM0MSIbhPKkDcozHqyXKyL/+bXYYw+iVPsns7c7bJ
-zBqenLQlWoRVyrVyBFrrwQSrKu/0Mqn3l639iOGPlUoR3I7aZKIpyEdDkqd3xGIC
-e+BtNNDqhUai67L63hEdG+iYAchi8UZw3LZGtcGpJ3FkBH4cYFX9EOam2QjbD7WY
-EO7m1+j6XEYIOTCmAP9dGAvBbU0Jblc+wYxG3qNr+2dBWsK76QXWEqib2VSOGP+z
-gjJa8tqY7PXXdOJpalQXNphmD/4o4pHKR4Euy0yL/1oMkpacmrV61LWB8Trnx9nS
-9gdVrUteQF/cL1KAGwOsdVmiLpHfvqLLRqSAAQIDAQABAoIBABjB+HEN4Kixf4fk
-wKHKEhL+SF6b/7sFX00NXZ/KLXRhSnnWSMQ8g/1hgMg2P2DfW4FbCDsCUu9xkLvI
-HTZY+CJAIh9U42uaYPWXkt09TmJi76TZ+2Nx4/XvRUjbCm7Fs1I2ekHeUbbAUS5g
-+BsPjTnL+h05zLHNoDa5yT0gVGIgFsQcX/w38arZCe8Rjp9le7PXUB5IIqASsDiw
-t8zJvdyWToeXd0WswCHTQu5coHvKo5MCjIZZ1Ink1yJcCCc3rKDc+q3jB2z9T9oW
-cUsKzJ4VuleiYj1eRxFITBmXbjKrb/GPRRUkeqCQbs68Hyj2d3UtOFDPeF4vng/3
-jGsHPq8CgYEA0AHAbwykVC6NMa37BTvEqcKoxbjTtErxR+yczlmVDfma9vkwtZvx
-FJdbS/+WGA/ucDby5x5b2T5k1J9ueMR86xukb+HnyS0WKsZ94Ie8WnJAcbp+38M6
-7LD0u74Cgk93oagDAzUHqdLq9cXxv/ppBpxVB1Uvu8DfVMHj+wt6ie8CgYEA4C7u
-u+6b8EmbGqEdtlPpScKG0WFstJEDGXRARDCRiVP2w6wm25v8UssCPvWcwf8U1Hoq
-lhMY+H6a5dnRRiNYql1MGQAsqMi7VeJNYb0B1uxi7X8MPM+SvXoAglX7wm1z0cVy
-O4CE5sEKbBg6aQabx1x9tzdrm80SKuSsLc5HRQ8CgYEAp/mCKSuQWNru8ruJBwTp
-IB4upN1JOUN77ZVKW+lD0XFMjz1U9JPl77b65ziTQQM8jioRpkqB6cHVM088qxIh
-vssn06Iex/s893YrmPKETJYPLMhqRNEn+JQ+To53ADykY0uGg0SD18SYMbmULHBP
-+CKvF6jXT0vGDnA1ZzoxzskCgYEA2nQhYrRS9EVlhP93KpJ+A8gxA5tCCHo+YPFt
-JoWFbCKLlYUNoHZR3IPCPoOsK0Zbj+kz0mXtsUf9vPkR+py669haLQqEejyQgFIz
-QYiiYEKc6/0feapzvXtDP751w7JQaBtVAzJrT0jQ1SCO2oT8C7rPLlgs3fdpOq72
-MPSPcnUCgYBWHm6bn4HvaoUSr0v2hyD9fHZS/wDTnlXVe5c1XXgyKlJemo5dvycf
-HUCmN/xIuO6AsiMdqIzv+arNJdboz+O+bNtS43LkTJfEH3xj2/DdUogdvOgG/iPM
-u9KBT1h+euws7PqC5qt4vqLwCTTCZXmUS8Riv+62RCC3kZ5AbpT3ZA==
------END RSA PRIVATE KEY-----
-"""
-
-MASTER_SIGNING_PUB = """
------BEGIN PUBLIC KEY-----
-MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAtieqrBMTM0MSIbhPKkDc
-ozHqyXKyL/+bXYYw+iVPsns7c7bJzBqenLQlWoRVyrVyBFrrwQSrKu/0Mqn3l639
-iOGPlUoR3I7aZKIpyEdDkqd3xGICe+BtNNDqhUai67L63hEdG+iYAchi8UZw3LZG
-tcGpJ3FkBH4cYFX9EOam2QjbD7WYEO7m1+j6XEYIOTCmAP9dGAvBbU0Jblc+wYxG
-3qNr+2dBWsK76QXWEqib2VSOGP+zgjJa8tqY7PXXdOJpalQXNphmD/4o4pHKR4Eu
-y0yL/1oMkpacmrV61LWB8Trnx9nS9gdVrUteQF/cL1KAGwOsdVmiLpHfvqLLRqSA
-AQIDAQAB
------END PUBLIC KEY-----
-"""
-
-MINION_PRIV_KEY = """
------BEGIN RSA PRIVATE KEY-----
-MIIEowIBAAKCAQEAsT6TwnlI0L7urjXu6D5E11tFJ/NglQ45jW/WN9tAUNvphq6Q
-cjJCd/aWmdqlqe7ix8y9M/8rgwghRQsnPXblVBvPwFcUEXhMRnOGzqbq/0zyQX01
-KecT0plBhlDt2lTyCLU6E4XCqyLbPfOxgXzsVqM0/TnzRtpVvGNy+5N4eFGylrjb
-cJhPxKt2G9TDOCM/hYacDs5RVIYQQmcYb8LJq7G3++FfWpYRDaxdKoHNFDspEynd
-jzr67hgThnwzc388OKNJx/7B2atwPTunPb3YBjgwDyRO/01OKK4gUHdw5KoctFgp
-kDCDjwjemlyXV+MYODRTIdtOlAP83ZkntEuLoQIDAQABAoIBAAJOKNtvFGfF2l9H
-S4CXZSUGU0a+JaCkR+wmnjsPwPn/dXDpAe8nGpidpNicPWqRm6WABjeQHaxda+fB
-lpSrRtEdo3zoi2957xQJ5wddDtI1pmXJQrdbm0H/K39oIg/Xtv/IZT769TM6OtVg
-paUxG/aftmeGXDtGfIL8w1jkuPABRBLOakWQA9uVdeG19KTU0Ag8ilpJdEX64uFJ
-W75bpVjT+KO/6aV1inuCntQSP097aYvUWajRwuiYVJOxoBZHme3IObcE6mdnYXeQ
-wblyWBpJUHrOS4MP4HCODV2pHKZ2rr7Nwhh8lMNw/eY9OP0ifz2AcAqe3sUMQOKP
-T0qRC6ECgYEAyeU5JvUPOpxXvvChYh6gJ8pYTIh1ueDP0O5e4t3vhz6lfy9DKtRN
-ROJLUorHvw/yVXMR72nT07a0z2VswcrUSw8ov3sI53F0NkLGEafQ35lVhTGs4vTl
-CFoQCuAKPsxeUl4AIbfbpkDsLGQqzW1diFArK7YeQkpGuGaGodXl480CgYEA4L40
-x5cUXnAhTPsybo7sbcpiwFHoGblmdkvpYvHA2QxtNSi2iHHdqGo8qP1YsZjKQn58
-371NhtqidrJ6i/8EBFP1dy+y/jr9qYlZNNGcQeBi+lshrEOIf1ct56KePG79s8lm
-DmD1OY8tO2R37+Py46Nq1n6viT/ST4NjLQI3GyUCgYEAiOswSDA3ZLs0cqRD/gPg
-/zsliLmehTFmHj4aEWcLkz+0Ar3tojUaNdX12QOPFQ7efH6uMhwl8NVeZ6xUBlTk
-hgbAzqLE1hjGBCpiowSZDZqyOcMHiV8ll/VkHcv0hsQYT2m6UyOaDXTH9g70TB6Y
-KOKddGZsvO4cad/1+/jQkB0CgYAzDEEkzLY9tS57M9uCrUgasAu6L2CO50PUvu1m
-Ig9xvZbYqkS7vVFhva/FmrYYsOHQNLbcgz0m0mZwm52mSuh4qzFoPxdjE7cmWSJA
-ExRxCiyxPR3q6PQKKJ0urgtPIs7RlX9u6KsKxfC6OtnbTWWQO0A7NE9e13ZHxUoz
-oPsvWQKBgCa0+Fb2lzUeiQz9bV1CBkWneDZUXuZHmabAZomokX+h/bq+GcJFzZjW
-3kAHwYkIy9IAy3SyO/6CP0V3vAye1p+XbotiwsQ/XZnr0pflSQL3J1l1CyN3aopg
-Niv7k/zBn15B72aK73R/CpUSk9W/eJGqk1NcNwf8hJHsboRYx6BR
------END RSA PRIVATE KEY-----
-"""
-
-
-MINION_PUB_KEY = """
------BEGIN PUBLIC KEY-----
-MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAsT6TwnlI0L7urjXu6D5E
-11tFJ/NglQ45jW/WN9tAUNvphq6QcjJCd/aWmdqlqe7ix8y9M/8rgwghRQsnPXbl
-VBvPwFcUEXhMRnOGzqbq/0zyQX01KecT0plBhlDt2lTyCLU6E4XCqyLbPfOxgXzs
-VqM0/TnzRtpVvGNy+5N4eFGylrjbcJhPxKt2G9TDOCM/hYacDs5RVIYQQmcYb8LJ
-q7G3++FfWpYRDaxdKoHNFDspEyndjzr67hgThnwzc388OKNJx/7B2atwPTunPb3Y
-BjgwDyRO/01OKK4gUHdw5KoctFgpkDCDjwjemlyXV+MYODRTIdtOlAP83ZkntEuL
-oQIDAQAB
------END PUBLIC KEY-----
-"""
-
-AES_KEY = "8wxWlOaMMQ4d3yT74LL4+hGrGTf65w8VgrcNjLJeLRQ2Q6zMa8ItY2EQUgMKKDb7JY+RnPUxbB0="
-
-
-@pytest.fixture
-def signing_algorithm():
-    if FIPS_TESTRUN:
-        return salt.crypt.PKCS1v15_SHA224
-    return salt.crypt.PKCS1v15_SHA1
-
-
-@pytest.fixture
-def encryption_algorithm():
-    if FIPS_TESTRUN:
-        return salt.crypt.OAEP_SHA224
-    return salt.crypt.OAEP_SHA1
-
-
-@pytest.fixture
-def pki_dir(tmp_path):
-    _pki_dir = tmp_path / "pki"
-    _pki_dir.mkdir()
-    madir = _pki_dir / "master"
-    madir.mkdir()
-
-    mapriv = madir / "master.pem"
-    mapriv.write_text(MASTER_PRIV_KEY.strip())
-    mapub = madir / "master.pub"
-    mapub.write_text(MASTER_PUB_KEY.strip())
-
-    maspriv = madir / "master_sign.pem"
-    maspriv.write_text(MASTER_SIGNING_PRIV.strip())
-    maspub = madir / "master_sign.pub"
-    maspub.write_text(MASTER_SIGNING_PUB.strip())
-
-    misdir = madir / "minions"
-    misdir.mkdir()
-    misdir.joinpath("minion").write_text(MINION_PUB_KEY.strip())
-    for sdir in [
-        "minions_autosign",
-        "minions_denied",
-        "minions_pre",
-        "minions_rejected",
-    ]:
-        madir.joinpath(sdir).mkdir()
-
-    midir = _pki_dir / "minion"
-    midir.mkdir()
-    mipub = midir / "minion.pub"
-    mipub.write_text(MINION_PUB_KEY.strip())
-    mipriv = midir / "minion.pem"
-    mipriv.write_text(MINION_PRIV_KEY.strip())
-    mimapriv = midir / "minion_master.pub"
-    mimapriv.write_text(MASTER_PUB_KEY.strip())
-    mimaspriv = midir / "master_sign.pub"
-    mimaspriv.write_text(MASTER_SIGNING_PUB.strip())
-    yield _pki_dir
-
-
-def test_master_uri():
-    """
-    test _get_master_uri method
-    """
-
-    m_ip = "127.0.0.1"
-    m_port = 4505
-    s_ip = "111.1.0.1"
-    s_port = 4058
-
-    m_ip6 = "1234:5678::9abc"
-    s_ip6 = "1234:5678::1:9abc"
-
-    with patch("salt.transport.zeromq.LIBZMQ_VERSION_INFO", (4, 1, 6)), patch(
-        "salt.transport.zeromq.ZMQ_VERSION_INFO", (16, 0, 1)
-    ):
-        # pass in both source_ip and source_port
-        assert (
-            salt.transport.zeromq._get_master_uri(
-                master_ip=m_ip, master_port=m_port, source_ip=s_ip, source_port=s_port
-            )
-            == f"tcp://{s_ip}:{s_port};{m_ip}:{m_port}"
-        )
-
-        assert (
-            salt.transport.zeromq._get_master_uri(
-                master_ip=m_ip6, master_port=m_port, source_ip=s_ip6, source_port=s_port
-            )
-            == f"tcp://[{s_ip6}]:{s_port};[{m_ip6}]:{m_port}"
-        )
-
-        # source ip and source_port empty
-        assert (
-            salt.transport.zeromq._get_master_uri(master_ip=m_ip, master_port=m_port)
-            == f"tcp://{m_ip}:{m_port}"
-        )
-
-        assert (
-            salt.transport.zeromq._get_master_uri(master_ip=m_ip6, master_port=m_port)
-            == f"tcp://[{m_ip6}]:{m_port}"
-        )
-
-        # pass in only source_ip
-        assert (
-            salt.transport.zeromq._get_master_uri(
-                master_ip=m_ip, master_port=m_port, source_ip=s_ip
-            )
-            == f"tcp://{s_ip}:0;{m_ip}:{m_port}"
-        )
-
-        assert (
-            salt.transport.zeromq._get_master_uri(
-                master_ip=m_ip6, master_port=m_port, source_ip=s_ip6
-            )
-            == f"tcp://[{s_ip6}]:0;[{m_ip6}]:{m_port}"
-        )
-
-        # pass in only source_port
-        assert (
-            salt.transport.zeromq._get_master_uri(
-                master_ip=m_ip, master_port=m_port, source_port=s_port
-            )
-            == f"tcp://0.0.0.0:{s_port};{m_ip}:{m_port}"
-        )
-
-
-def test_clear_req_channel_master_uri_override(temp_salt_minion, temp_salt_master):
-    """
-    ensure master_uri kwarg is respected
-    """
-    opts = temp_salt_minion.config.copy()
-    # minion_config should be 127.0.0.1, we want a different uri that still connects
-    opts.update(
-        {
-            "id": "root",
-            "transport": "zeromq",
-            "auth_tries": 1,
-            "auth_timeout": 5,
-            "master_ip": "127.0.0.1",
-            "master_port": temp_salt_master.config["ret_port"],
-            "master_uri": "tcp://127.0.0.1:{}".format(
-                temp_salt_master.config["ret_port"]
-            ),
-        }
-    )
-    master_uri = "tcp://{master_ip}:{master_port}".format(
-        master_ip="localhost", master_port=opts["master_port"]
-    )
-    with salt.channel.client.ReqChannel.factory(opts, master_uri=master_uri) as channel:
-        assert "127.0.0.1" in channel.transport.master_uri
-
-
-def run_loop_in_thread(loop, evt):
-    """
-    Run the provided loop until an event is set
-    """
-    asyncio.set_event_loop(loop.asyncio_loop)
-
-    async def stopper():
-        await asyncio.sleep(0.1)
-        while True:
-            if not evt.is_set():
-                loop.stop()
-                break
-            await asyncio.sleep(0.3)
-
-    loop.add_callback(evt.set)
-    loop.add_callback(stopper)
-    try:
-        loop.start()
-    finally:
-        loop.close()
-
-
-class MockSaltMinionMaster:
-    mock = MagicMock()
-
-    def __init__(self, temp_salt_minion, temp_salt_master):
-        SMaster.secrets["aes"] = {
-            "secret": multiprocessing.Array(
-                ctypes.c_char,
-                salt.utils.stringutils.to_bytes(
-                    salt.crypt.Crypticle.generate_key_string()
-                ),
-            ),
-            "reload": salt.crypt.Crypticle.generate_key_string,
-        }
-        self.process_manager = salt.utils.process.ProcessManager(
-            name="ReqServer_ProcessManager"
-        )
-
-        master_opts = temp_salt_master.config.copy()
-        master_opts.update({"transport": "zeromq"})
-        self.server_channel = salt.channel.server.ReqServerChannel.factory(master_opts)
-        self.server_channel.pre_fork(self.process_manager)
-        self.io_loop = tornado.ioloop.IOLoop(make_current=False)
-        self.evt = threading.Event()
-        self.server_channel.post_fork(self._handle_payload, io_loop=self.io_loop)
-        self.server_thread = threading.Thread(
-            target=run_loop_in_thread, args=(self.io_loop, self.evt)
-        )
-        self.server_thread.start()
-        minion_opts = temp_salt_minion.config.copy()
-        minion_opts.update(
-            {
-                "master_ip": "127.0.0.1",
-                "transport": "zeromq",
-            }
-        )
-        self.channel = salt.channel.client.ReqChannel.factory(
-            minion_opts, crypt="clear"
-        )
-
-    def __enter__(self):
-        self.channel.__enter__()
-        self.evt.wait()
-        return self
-
-    def __exit__(self, *args, **kwargs):
-        self.channel.__exit__(*args, **kwargs)
-        del self.channel
-        self.server_channel.close()
-        # Attempting to kill the children hangs the test suite.
-        # Let the test suite handle this instead.
-        self.process_manager.stop_restarting()
-        self.process_manager.kill_children()
-        self.evt.clear()
-        self.server_thread.join()
-        # Give the procs a chance to fully close before we stop the io_loop
-        time.sleep(2)
-        SMaster.secrets.pop("aes")
-        del self.server_channel
-        del self.io_loop
-        del self.process_manager
-        del self.server_thread
-
-    # pylint: enable=W1701
-    @classmethod
-    @tornado.gen.coroutine
-    def _handle_payload(cls, payload):
-        """
-        TODO: something besides echo
-        """
-        cls.mock._handle_payload_hook()
-        raise tornado.gen.Return((payload, {"fun": "send_clear"}))
-
-
-@pytest.mark.parametrize("message", ["", [], ()])
-def test_badload(temp_salt_minion, temp_salt_master, message):
-    """
-    Test a variety of bad requests, make sure that we get some sort of error
-    """
-    with MockSaltMinionMaster(temp_salt_minion, temp_salt_master) as minion_master:
-        ret = minion_master.channel.send(message, timeout=5, tries=1)
-        assert ret == "payload and load must be a dict"
-
-
-def test_payload_handling_exception(temp_salt_minion, temp_salt_master):
-    """
-    test of getting exception on payload handling
-    """
-    with MockSaltMinionMaster(temp_salt_minion, temp_salt_master) as minion_master:
-        with patch.object(minion_master.mock, "_handle_payload_hook") as _mock:
-            _mock.side_effect = Exception()
-            ret = minion_master.channel.send({}, timeout=15, tries=1)
-            assert ret == "Some exception handling minion payload"
-
-
-def test_serverside_exception(temp_salt_minion, temp_salt_master):
-    """
-    test of getting server side exception on payload handling
-    """
-    with MockSaltMinionMaster(temp_salt_minion, temp_salt_master) as minion_master:
-        with patch.object(minion_master.mock, "_handle_payload_hook") as _mock:
-            _mock.side_effect = tornado.gen.Return(({}, {"fun": "madeup-fun"}))
-            ret = minion_master.channel.send({}, timeout=5, tries=1)
-            assert ret == "Server-side exception handling payload"
-
-
-def test_req_server_chan_encrypt_v2(
-    pki_dir, encryption_algorithm, signing_algorithm, master_opts
-):
-    loop = tornado.ioloop.IOLoop.current()
-    master_opts.update(
-        {
-            "worker_threads": 1,
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "zmq_monitor": False,
-            "mworker_queue_niceness": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("master")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-        }
-    )
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    dictkey = "pillar"
-    nonce = "abcdefg"
-    pillar_data = {"pillar1": "meh"}
-    try:
-        ret = server._encrypt_private(
-            pillar_data,
-            dictkey,
-            "minion",
-            nonce,
-            encryption_algorithm=encryption_algorithm,
-            signing_algorithm=signing_algorithm,
-        )
-        assert "key" in ret
-        assert dictkey in ret
-
-        key = salt.crypt.PrivateKey(str(pki_dir.joinpath("minion", "minion.pem")))
-        aes = key.decrypt(ret["key"], encryption_algorithm)
-        pcrypt = salt.crypt.Crypticle(master_opts, aes)
-        signed_msg = pcrypt.loads(ret[dictkey])
-
-        assert "sig" in signed_msg
-        assert "data" in signed_msg
-        data = salt.payload.loads(signed_msg["data"])
-        assert "key" in data
-        assert data["key"] == ret["key"]
-        assert "key" in data
-        assert data["nonce"] == nonce
-        assert "pillar" in data
-        assert data["pillar"] == pillar_data
-    finally:
-        server.close()
-
-
-def test_req_server_chan_encrypt_v1(pki_dir, encryption_algorithm, master_opts):
-    loop = tornado.ioloop.IOLoop.current()
-    master_opts.update(
-        {
-            "worker_threads": 1,
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "zmq_monitor": False,
-            "mworker_queue_niceness": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("master")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-        }
-    )
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    dictkey = "pillar"
-    nonce = "abcdefg"
-    pillar_data = {"pillar1": "meh"}
-    try:
-        ret = server._encrypt_private(
-            pillar_data,
-            dictkey,
-            "minion",
-            sign_messages=False,
-            encryption_algorithm=encryption_algorithm,
-        )
-
-        assert "key" in ret
-        assert dictkey in ret
-
-        key = salt.crypt.PrivateKey(str(pki_dir.joinpath("minion", "minion.pem")))
-        aes = key.decrypt(ret["key"], encryption_algorithm)
-        pcrypt = salt.crypt.Crypticle(master_opts, aes)
-        data = pcrypt.loads(ret[dictkey])
-        assert data == pillar_data
-    finally:
-        server.close()
-
-
-def test_req_chan_decode_data_dict_entry_v1(
-    pki_dir, encryption_algorithm, minion_opts, master_opts
-):
-    mockloop = MagicMock()
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    master_opts = dict(master_opts, pki_dir=str(pki_dir.joinpath("master")))
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    client = salt.channel.client.ReqChannel.factory(minion_opts, io_loop=mockloop)
-    try:
-        dictkey = "pillar"
-        target = "minion"
-        pillar_data = {"pillar1": "meh"}
-        ret = server._encrypt_private(
-            pillar_data,
-            dictkey,
-            target,
-            sign_messages=False,
-            encryption_algorithm=encryption_algorithm,
-        )
-        key = client.auth.get_keys()
-        aes = key.decrypt(ret["key"], encryption_algorithm)
-        pcrypt = salt.crypt.Crypticle(client.opts, aes)
-        ret_pillar_data = pcrypt.loads(ret[dictkey])
-        assert ret_pillar_data == pillar_data
-    finally:
-        client.close()
-        server.close()
-
-
-async def test_req_chan_decode_data_dict_entry_v2(minion_opts, master_opts, pki_dir):
-    mockloop = MagicMock()
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=mockloop)
-
-    dictkey = "pillar"
-    target = "minion"
-    pillar_data = {"pillar1": "meh"}
-
-    # Mock auth and message client.
-    auth = client.auth
-    auth._crypticle = salt.crypt.Crypticle(minion_opts, AES_KEY)
-    client.auth = MagicMock()
-    client.auth.mpub = auth.mpub
-    client.auth.authenticated = True
-    client.auth.get_keys = auth.get_keys
-    client.auth.crypticle.dumps = auth.crypticle.dumps
-    client.auth.crypticle.loads = auth.crypticle.loads
-    client.transport = MagicMock()
-
-    print(minion_opts["encryption_algorithm"])
-
-    @tornado.gen.coroutine
-    def mocksend(msg, timeout=60, tries=3):
-        client.transport.msg = msg
-        load = client.auth.crypticle.loads(msg["load"])
-        ret = server._encrypt_private(
-            pillar_data,
-            dictkey,
-            target,
-            nonce=load["nonce"],
-            sign_messages=True,
-            encryption_algorithm=minion_opts["encryption_algorithm"],
-            signing_algorithm=minion_opts["signing_algorithm"],
-        )
-        raise tornado.gen.Return(ret)
-
-    client.transport.send = mocksend
-
-    # Note the 'ver' value in 'load' does not represent the the 'version' sent
-    # in the top level of the transport's message.
-    load = {
-        "id": target,
-        "grains": {},
-        "saltenv": "base",
-        "pillarenv": "base",
-        "pillar_override": True,
-        "extra_minion_data": {},
-        "ver": "2",
-        "cmd": "_pillar",
-    }
-    try:
-        ret = await client.crypted_transfer_decode_dictentry(  # pylint: disable=E1121,E1123
-            load,
-            dictkey="pillar",
-        )
-        assert "version" in client.transport.msg
-        assert client.transport.msg["version"] == 2
-        assert ret == {"pillar1": "meh"}
-    finally:
-        client.close()
-        server.close()
-
-
-async def test_req_chan_decode_data_dict_entry_v2_bad_nonce(
-    minion_opts, master_opts, pki_dir
-):
-    mockloop = MagicMock()
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=mockloop)
-
-    dictkey = "pillar"
-    badnonce = "abcdefg"
-    target = "minion"
-    pillar_data = {"pillar1": "meh"}
-
-    # Mock auth and message client.
-    auth = client.auth
-    auth._crypticle = salt.crypt.Crypticle(minion_opts, AES_KEY)
-    client.auth = MagicMock()
-    client.auth.mpub = auth.mpub
-    client.auth.authenticated = True
-    client.auth.get_keys = auth.get_keys
-    client.auth.crypticle.dumps = auth.crypticle.dumps
-    client.auth.crypticle.loads = auth.crypticle.loads
-    real_transport = client.transport
-    client.transport = MagicMock()
-    real_transport.close()
-    ret = server._encrypt_private(
-        pillar_data,
-        dictkey,
-        target,
-        nonce=badnonce,
-        sign_messages=True,
-        encryption_algorithm=minion_opts["encryption_algorithm"],
-        signing_algorithm=minion_opts["signing_algorithm"],
-    )
-
-    @tornado.gen.coroutine
-    def mocksend(msg, timeout=60, tries=3):
-        client.transport.msg = msg
-        raise tornado.gen.Return(ret)
-
-    client.transport.send = mocksend
-
-    # Note the 'ver' value in 'load' does not represent the the 'version' sent
-    # in the top level of the transport's message.
-    load = {
-        "id": target,
-        "grains": {},
-        "saltenv": "base",
-        "pillarenv": "base",
-        "pillar_override": True,
-        "extra_minion_data": {},
-        "ver": "2",
-        "cmd": "_pillar",
-    }
-
-    try:
-        with pytest.raises(salt.crypt.AuthenticationError) as excinfo:
-            ret = await client.crypted_transfer_decode_dictentry(  # pylint: disable=E1121,E1123
-                load,
-                dictkey="pillar",
-            )
-        assert "Pillar nonce verification failed." == excinfo.value.message
-    finally:
-        client.close()
-        server.close()
-
-
-async def test_req_chan_decode_data_dict_entry_v2_bad_signature(
-    pki_dir, minion_opts, master_opts
-):
-    mockloop = MagicMock()
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=mockloop)
-
-    dictkey = "pillar"
-    badnonce = "abcdefg"
-    target = "minion"
-    pillar_data = {"pillar1": "meh"}
-
-    # Mock auth and message client.
-    auth = client.auth
-    auth._crypticle = salt.crypt.Crypticle(minion_opts, AES_KEY)
-    client.auth = MagicMock()
-    client.auth.mpub = auth.mpub
-    client.auth.authenticated = True
-    client.auth.get_keys = auth.get_keys
-    client.auth.crypticle.dumps = auth.crypticle.dumps
-    client.auth.crypticle.loads = auth.crypticle.loads
-    client.transport = MagicMock()
-
-    @tornado.gen.coroutine
-    def mocksend(msg, timeout=60, tries=3):
-        client.transport.msg = msg
-        load = client.auth.crypticle.loads(msg["load"])
-        ret = server._encrypt_private(
-            pillar_data,
-            dictkey,
-            target,
-            nonce=load["nonce"],
-            sign_messages=True,
-            encryption_algorithm=minion_opts["encryption_algorithm"],
-            signing_algorithm=minion_opts["signing_algorithm"],
-        )
-
-        key = client.auth.get_keys()
-        aes = key.decrypt(ret["key"], minion_opts["encryption_algorithm"])
-        pcrypt = salt.crypt.Crypticle(client.opts, aes)
-        signed_msg = pcrypt.loads(ret[dictkey])
-        # Changing the pillar data will cause the signature verification to
-        # fail.
-        data = salt.payload.loads(signed_msg["data"])
-        data["pillar"] = {"pillar1": "bar"}
-        signed_msg["data"] = salt.payload.dumps(data)
-        ret[dictkey] = pcrypt.dumps(signed_msg)
-        raise tornado.gen.Return(ret)
-
-    client.transport.send = mocksend
-
-    # Note the 'ver' value in 'load' does not represent the the 'version' sent
-    # in the top level of the transport's message.
-    load = {
-        "id": target,
-        "grains": {},
-        "saltenv": "base",
-        "pillarenv": "base",
-        "pillar_override": True,
-        "extra_minion_data": {},
-        "ver": "2",
-        "cmd": "_pillar",
-    }
-
-    try:
-        with pytest.raises(salt.crypt.AuthenticationError) as excinfo:
-            ret = await client.crypted_transfer_decode_dictentry(  # pylint: disable=E1121,E1123
-                load,
-                dictkey="pillar",
-            )
-        assert "Pillar payload signature failed to validate." == excinfo.value.message
-    finally:
-        client.close()
-        server.close()
-
-
-async def test_req_chan_decode_data_dict_entry_v2_bad_key(
-    pki_dir, minion_opts, master_opts
-):
-    mockloop = MagicMock()
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=mockloop)
-
-    dictkey = "pillar"
-    badnonce = "abcdefg"
-    target = "minion"
-    pillar_data = {"pillar1": "meh"}
-
-    # Mock auth and message client.
-    auth = client.auth
-    auth._crypticle = salt.crypt.Crypticle(master_opts, AES_KEY)
-    client.auth = MagicMock()
-    client.auth.mpub = auth.mpub
-    client.auth.authenticated = True
-    client.auth.get_keys = auth.get_keys
-    client.auth.crypticle.dumps = auth.crypticle.dumps
-    client.auth.crypticle.loads = auth.crypticle.loads
-    client.transport = MagicMock()
-
-    @tornado.gen.coroutine
-    def mocksend(msg, timeout=60, tries=3):
-        client.transport.msg = msg
-        load = client.auth.crypticle.loads(msg["load"])
-        ret = server._encrypt_private(
-            pillar_data,
-            dictkey,
-            target,
-            nonce=load["nonce"],
-            sign_messages=True,
-            encryption_algorithm=minion_opts["encryption_algorithm"],
-            signing_algorithm=minion_opts["signing_algorithm"],
-        )
-
-        mkey = client.auth.get_keys()
-        aes = mkey.decrypt(ret["key"], minion_opts["encryption_algorithm"])
-        pcrypt = salt.crypt.Crypticle(client.opts, aes)
-        signed_msg = pcrypt.loads(ret[dictkey])
-
-        # Now encrypt with a different key
-        key = salt.crypt.Crypticle.generate_key_string()
-        pcrypt = salt.crypt.Crypticle(master_opts, key)
-        pubfn = os.path.join(master_opts["pki_dir"], "minions", "minion")
-        pub = salt.crypt.PublicKey(pubfn)
-        ret[dictkey] = pcrypt.dumps(signed_msg)
-        key = salt.utils.stringutils.to_bytes(key)
-        ret["key"] = pub.encrypt(key, minion_opts["encryption_algorithm"])
-        raise tornado.gen.Return(ret)
-
-    client.transport.send = mocksend
-
-    # Note the 'ver' value in 'load' does not represent the the 'version' sent
-    # in the top level of the transport's message.
-    load = {
-        "id": target,
-        "grains": {},
-        "saltenv": "base",
-        "pillarenv": "base",
-        "pillar_override": True,
-        "extra_minion_data": {},
-        "ver": "2",
-        "cmd": "_pillar",
-    }
-    try:
-        with pytest.raises(salt.crypt.AuthenticationError) as excinfo:
-            await client.crypted_transfer_decode_dictentry(  # pylint: disable=E1121,E1123
-                load,
-                dictkey="pillar",
-            )
-        assert "Key verification failed." == excinfo.value.message
-    finally:
-        client.close()
-        server.close()
-
-
-async def test_req_serv_auth_v1(minion_opts, master_opts, pki_dir):
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "master_sign_pubkey": False,
-            "publish_port": 4505,
-            "auth_mode": 1,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-
-    pub = salt.crypt.get_rsa_pub_key(str(pki_dir.joinpath("minion", "minion.pub")))
-    token = salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string())
-    nonce = uuid.uuid4().hex
-
-    # We need to read the public key with fopen otherwise the newlines might
-    # not match on windows.
-    with salt.utils.files.fopen(
-        str(pki_dir.joinpath("minion", "minion.pub")), "r"
-    ) as fp:
-        pub_key = salt.crypt.clean_key(fp.read())
-
-    load = {
-        "cmd": "_auth",
-        "id": "minion",
-        "token": token,
-        "pub": pub_key,
-        "enc_algo": minion_opts["encryption_algorithm"],
-        "sig_algo": minion_opts["signing_algorithm"],
-    }
-    try:
-        ret = server._auth(load, sign_messages=False)
-        assert "load" not in ret
-    finally:
-        server.close()
-
-
-async def test_req_serv_auth_v2(minion_opts, master_opts, pki_dir):
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "master_sign_pubkey": False,
-            "publish_port": 4505,
-            "auth_mode": 1,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-
-    pub = salt.crypt.get_rsa_pub_key(str(pki_dir.joinpath("minion", "minion.pub")))
-    token = salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string())
-    nonce = uuid.uuid4().hex
-
-    # We need to read the public key with fopen otherwise the newlines might
-    # not match on windows.
-    with salt.utils.files.fopen(
-        str(pki_dir.joinpath("minion", "minion.pub")), "r"
-    ) as fp:
-        pub_key = salt.crypt.clean_key(fp.read())
-
-    load = {
-        "cmd": "_auth",
-        "id": "minion",
-        "nonce": nonce,
-        "token": token,
-        "pub": pub_key,
-        "enc_algo": minion_opts["encryption_algorithm"],
-        "sig_algo": minion_opts["signing_algorithm"],
-    }
-    try:
-        ret = server._auth(load, sign_messages=True)
-        assert "sig" in ret
-        assert "load" in ret
-    finally:
-        server.close()
-
-
-async def test_req_chan_auth_v2(minion_opts, master_opts, pki_dir, io_loop):
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "publish_port": 4505,
-            "auth_mode": 1,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    master_opts["master_sign_pubkey"] = False
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-    minion_opts["verify_master_pubkey_sign"] = False
-    minion_opts["always_verify_signature"] = False
-    client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
-    signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
-    try:
-        assert "version" in pload
-        assert pload["version"] == 2
-
-        ret = server._auth(pload["load"], sign_messages=True)
-        assert "sig" in ret
-        ret = client.auth.handle_signin_response(signin_payload, ret)
-        assert "aes" in ret
-        assert "master_uri" in ret
-        assert "publish_port" in ret
-    finally:
-        client.close()
-        server.close()
-
-
-async def test_req_chan_auth_v2_with_master_signing(
-    pki_dir, io_loop, minion_opts, master_opts
-):
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "publish_port": 4505,
-            "auth_mode": 1,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts = dict(master_opts, pki_dir=str(pki_dir.joinpath("master")))
-    master_opts["master_sign_pubkey"] = True
-    master_opts["master_use_pubkey_signature"] = False
-    master_opts["signing_key_pass"] = ""
-    master_opts["master_sign_key_name"] = "master_sign"
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.event = salt.utils.event.get_master_event(
-        master_opts, master_opts["sock_dir"], listen=False
-    )
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-    minion_opts["verify_master_pubkey_sign"] = True
-    minion_opts["always_verify_signature"] = True
-    minion_opts["master_sign_key_name"] = "master_sign"
-    minion_opts["master"] = "master"
-
-    assert (
-        pki_dir.joinpath("minion", "minion_master.pub").read_text()
-        == pki_dir.joinpath("master", "master.pub").read_text()
-    )
-
-    client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
-    signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
-    assert "version" in pload
-    assert pload["version"] == 2
-
-    server_reply = server._auth(pload["load"], sign_messages=True)
-    # With version 2 we always get a clear signed response
-    assert "enc" in server_reply
-    assert server_reply["enc"] == "clear"
-    assert "sig" in server_reply
-    assert "load" in server_reply
-    ret = client.auth.handle_signin_response(signin_payload, server_reply)
-    assert "aes" in ret
-    assert "master_uri" in ret
-    assert "publish_port" in ret
-
-    # Now create a new master key pair and try auth with it.
-    mapriv = pki_dir.joinpath("master", "master.pem")
-    mapriv.unlink()
-    mapriv.write_text(MASTER2_PRIV_KEY.strip())
-    mapub = pki_dir.joinpath("master", "master.pub")
-    mapub.unlink()
-    mapub.write_text(MASTER2_PUB_KEY.strip())
-
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.event = salt.utils.event.get_master_event(
-        master_opts, master_opts["sock_dir"], listen=False
-    )
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-
-    signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
-    server_reply = server._auth(pload["load"], sign_messages=True)
-    try:
-        ret = client.auth.handle_signin_response(signin_payload, server_reply)
-
-        assert "aes" in ret
-        assert "master_uri" in ret
-        assert "publish_port" in ret
-
-        assert (
-            pki_dir.joinpath("minion", "minion_master.pub").read_text()
-            == pki_dir.joinpath("master", "master.pub").read_text()
-        )
-    finally:
-        client.close()
-        server.close()
-
-
-async def test_req_chan_auth_v2_new_minion_with_master_pub(
-    minion_opts, master_opts, pki_dir, io_loop
-):
-
-    pki_dir.joinpath("master", "minions", "minion").unlink()
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "publish_port": 4505,
-            "auth_mode": 1,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    master_opts["master_sign_pubkey"] = False
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-    minion_opts["verify_master_pubkey_sign"] = False
-    minion_opts["always_verify_signature"] = False
-    client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
-    signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
-    try:
-        assert "version" in pload
-        assert pload["version"] == 2
-
-        ret = server._auth(pload["load"], sign_messages=True)
-        assert "sig" in ret
-        ret = client.auth.handle_signin_response(signin_payload, ret)
-        assert ret == "retry"
-    finally:
-        client.close()
-        server.close()
-
-
-async def test_req_chan_auth_v2_new_minion_with_master_pub_bad_sig(
-    minion_opts, master_opts, pki_dir, io_loop
-):
-
-    pki_dir.joinpath("master", "minions", "minion").unlink()
-
-    # Give the master a different key than the minion has.
-    mapriv = pki_dir.joinpath("master", "master.pem")
-    mapriv.unlink()
-    mapriv.write_text(MASTER2_PRIV_KEY.strip())
-    mapub = pki_dir.joinpath("master", "master.pub")
-    mapub.unlink()
-    mapub.write_text(MASTER2_PUB_KEY.strip())
-
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "publish_port": 4505,
-            "auth_mode": 1,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    master_opts["master_sign_pubkey"] = False
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-    minion_opts["verify_master_pubkey_sign"] = False
-    minion_opts["always_verify_signature"] = False
-    client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
-    signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
-    try:
-        assert "version" in pload
-        assert pload["version"] == 2
-
-        ret = server._auth(pload["load"], sign_messages=True)
-        assert "sig" in ret
-        with pytest.raises(salt.crypt.SaltClientError, match="Invalid signature"):
-            ret = client.auth.handle_signin_response(signin_payload, ret)
-    finally:
-        client.close()
-        server.close()
-
-
-async def test_req_chan_auth_v2_new_minion_without_master_pub(
-    minion_opts, master_opts, pki_dir, io_loop
-):
-
-    pki_dir.joinpath("master", "minions", "minion").unlink()
-    pki_dir.joinpath("minion", "minion_master.pub").unlink()
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "publish_port": 4505,
-            "auth_mode": 1,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    master_opts["master_sign_pubkey"] = False
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-    minion_opts["verify_master_pubkey_sign"] = False
-    minion_opts["always_verify_signature"] = False
-    client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
-    signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
-    try:
-        assert "version" in pload
-        assert pload["version"] == 2
-
-        ret = server._auth(pload["load"], sign_messages=True)
-        assert "sig" in ret
-        ret = client.auth.handle_signin_response(signin_payload, ret)
-        assert ret == "retry"
-    finally:
-        client.close()
-        server.close()
-
-
-async def test_req_chan_bad_payload_to_decode(
-    minion_opts, master_opts, pki_dir, io_loop
-):
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "publish_port": 4505,
-            "auth_mode": 1,
-            "acceptance_wait_time": 3,
-            "acceptance_wait_time_max": 3,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts.update(dict(minion_opts, pki_dir=str(pki_dir.joinpath("master"))))
-    master_opts["master_sign_pubkey"] = False
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-
-    with pytest.raises(salt.exceptions.SaltDeserializationError):
-        server._decode_payload(None)
-    with pytest.raises(salt.exceptions.SaltDeserializationError):
-        server._decode_payload({})
-    with pytest.raises(salt.exceptions.SaltDeserializationError):
-        server._decode_payload(12345)
-
-
-def test_req_server_auth_garbage_sig_algo(pki_dir, minion_opts, master_opts, caplog):
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "master_sign_pubkey": False,
-            "publish_port": 4505,
-            "auth_mode": 1,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.event = salt.utils.event.get_master_event(
-        master_opts, master_opts["sock_dir"], listen=False
-    )
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-    pub = salt.crypt.PublicKey(str(pki_dir.joinpath("master", "master.pub")))
-    token = pub.encrypt(
-        salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        algorithm=minion_opts["encryption_algorithm"],
-    )
-    nonce = uuid.uuid4().hex
-
-    # We need to read the public key with fopen otherwise the newlines might
-    # not match on windows.
-    with salt.utils.files.fopen(
-        str(pki_dir.joinpath("minion", "minion.pub")), "r"
-    ) as fp:
-        pub_key = salt.crypt.clean_key(fp.read())
-
-    load = {
-        "version": 2,
-        "cmd": "_auth",
-        "id": "minion",
-        "token": token,
-        "pub": pub_key,
-        "nonce": "asdfse",
-        "enc_algo": minion_opts["encryption_algorithm"],
-        "sig_algo": "IAMNOTANALGO",
-    }
-    with caplog.at_level(logging.INFO):
-        ret = server._auth(load, sign_messages=True)
-        assert (
-            "Minion tried to authenticate with unsupported signing algorithm: IAMNOTANALGO"
-            in caplog.text
-        )
-        assert "load" in ret
-        assert "ret" in ret["load"]
-        assert ret["load"]["ret"] == "bad sig algo"
-
-
-@pytest.mark.skipif(not FIPS_TESTRUN, reason="Only run on fips enabled platforms")
-def test_req_server_auth_unsupported_enc_algo(
-    pki_dir, minion_opts, master_opts, caplog
-):
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "master_sign_pubkey": False,
-            "publish_port": 4505,
-            "auth_mode": 1,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.event = salt.utils.event.get_master_event(
-        master_opts, master_opts["sock_dir"], listen=False
-    )
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-    import tests.pytests.unit.crypt
-
-    pub = tests.pytests.unit.crypt.LegacyPublicKey(
-        str(pki_dir.joinpath("master", "master.pub"))
-    )
-    token = pub.encrypt(
-        salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-    )
-    nonce = uuid.uuid4().hex
-
-    # We need to read the public key with fopen otherwise the newlines might
-    # not match on windows.
-    with salt.utils.files.fopen(
-        str(pki_dir.joinpath("minion", "minion.pub")), "r"
-    ) as fp:
-        pub_key = salt.crypt.clean_key(fp.read())
-
-    load = {
-        "version": 2,
-        "cmd": "_auth",
-        "id": "minion",
-        "token": token,
-        "pub": pub_key,
-        "nonce": "asdfse",
-        "enc_algo": "OAEP-SHA1",
-        "sig_algo": minion_opts["signing_algorithm"],
-    }
-    with caplog.at_level(logging.INFO):
-        ret = server._auth(load, sign_messages=True)
-        assert (
-            "Minion minion tried to authenticate with unsupported encryption algorithm: OAEP-SHA1"
-            in caplog.text
-        )
-        assert "load" in ret
-        assert "ret" in ret["load"]
-        assert ret["load"]["ret"] == "bad enc algo"
-
-
-def test_req_server_auth_garbage_enc_algo(pki_dir, minion_opts, master_opts, caplog):
-    minion_opts.update(
-        {
-            "master_uri": "tcp://127.0.0.1:4506",
-            "interface": "127.0.0.1",
-            "ret_port": 4506,
-            "ipv6": False,
-            "sock_dir": ".",
-            "pki_dir": str(pki_dir.joinpath("minion")),
-            "id": "minion",
-            "__role": "minion",
-            "keysize": 4096,
-            "max_minions": 0,
-            "auto_accept": False,
-            "open_mode": False,
-            "key_pass": None,
-            "master_sign_pubkey": False,
-            "publish_port": 4505,
-            "auth_mode": 1,
-        }
-    )
-    SMaster.secrets["aes"] = {
-        "secret": multiprocessing.Array(
-            ctypes.c_char,
-            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-        ),
-        "reload": salt.crypt.Crypticle.generate_key_string,
-    }
-    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
-    server = salt.channel.server.ReqServerChannel.factory(master_opts)
-
-    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
-    server.cache_cli = False
-    server.event = salt.utils.event.get_master_event(
-        master_opts, master_opts["sock_dir"], listen=False
-    )
-    server.master_key = salt.crypt.MasterKeys(server.opts)
-    import tests.pytests.unit.crypt
-
-    pub = tests.pytests.unit.crypt.LegacyPublicKey(
-        str(pki_dir.joinpath("master", "master.pub"))
-    )
-    token = pub.encrypt(
-        salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
-    )
-    nonce = uuid.uuid4().hex
-
-    # We need to read the public key with fopen otherwise the newlines might
-    # not match on windows.
-    with salt.utils.files.fopen(
-        str(pki_dir.joinpath("minion", "minion.pub")), "r"
-    ) as fp:
-        pub_key = salt.crypt.clean_key(fp.read())
-
-    load = {
-        "version": 2,
-        "cmd": "_auth",
-        "id": "minion",
-        "token": token,
-        "pub": pub_key,
-        "nonce": "asdfse",
-        "enc_algo": "IAMNOTAENCALGO",
-        "sig_algo": minion_opts["signing_algorithm"],
-    }
-    with caplog.at_level(logging.INFO):
-        ret = server._auth(load, sign_messages=True)
-        assert (
-            "Minion minion tried to authenticate with unsupported encryption algorithm: IAMNOTAENCALGO"
-            in caplog.text
-        )
-        assert "load" in ret
-        assert "ret" in ret["load"]
-        assert ret["load"]["ret"] == "bad enc algo"

@@ -357,6 +357,98 @@ def _handle_payload(cls, payload):
         raise tornado.gen.Return((payload, {"fun": "send_clear"}))
 
 
+def test_master_uri():
+    """
+    test _get_master_uri method
+    """
+
+    m_ip = "127.0.0.1"
+    m_port = 4505
+    s_ip = "111.1.0.1"
+    s_port = 4058
+
+    m_ip6 = "1234:5678::9abc"
+    s_ip6 = "1234:5678::1:9abc"
+
+    with patch("salt.transport.zeromq.LIBZMQ_VERSION_INFO", (4, 1, 6)), patch(
+        "salt.transport.zeromq.ZMQ_VERSION_INFO", (16, 0, 1)
+    ):
+        # pass in both source_ip and source_port
+        assert (
+            salt.transport.zeromq._get_master_uri(
+                master_ip=m_ip, master_port=m_port, source_ip=s_ip, source_port=s_port
+            )
+            == f"tcp://{s_ip}:{s_port};{m_ip}:{m_port}"
+        )
+
+        assert (
+            salt.transport.zeromq._get_master_uri(
+                master_ip=m_ip6, master_port=m_port, source_ip=s_ip6, source_port=s_port
+            )
+            == f"tcp://[{s_ip6}]:{s_port};[{m_ip6}]:{m_port}"
+        )
+
+        # source ip and source_port empty
+        assert (
+            salt.transport.zeromq._get_master_uri(master_ip=m_ip, master_port=m_port)
+            == f"tcp://{m_ip}:{m_port}"
+        )
+
+        assert (
+            salt.transport.zeromq._get_master_uri(master_ip=m_ip6, master_port=m_port)
+            == f"tcp://[{m_ip6}]:{m_port}"
+        )
+
+        # pass in only source_ip
+        assert (
+            salt.transport.zeromq._get_master_uri(
+                master_ip=m_ip, master_port=m_port, source_ip=s_ip
+            )
+            == f"tcp://{s_ip}:0;{m_ip}:{m_port}"
+        )
+
+        assert (
+            salt.transport.zeromq._get_master_uri(
+                master_ip=m_ip6, master_port=m_port, source_ip=s_ip6
+            )
+            == f"tcp://[{s_ip6}]:0;[{m_ip6}]:{m_port}"
+        )
+
+        # pass in only source_port
+        assert (
+            salt.transport.zeromq._get_master_uri(
+                master_ip=m_ip, master_port=m_port, source_port=s_port
+            )
+            == f"tcp://0.0.0.0:{s_port};{m_ip}:{m_port}"
+        )
+
+
+def test_clear_req_channel_master_uri_override(temp_salt_minion, temp_salt_master):
+    """
+    ensure master_uri kwarg is respected
+    """
+    opts = temp_salt_minion.config.copy()
+    # minion_config should be 127.0.0.1, we want a different uri that still connects
+    opts.update(
+        {
+            "id": "root",
+            "transport": "zeromq",
+            "auth_tries": 1,
+            "auth_timeout": 5,
+            "master_ip": "127.0.0.1",
+            "master_port": temp_salt_master.config["ret_port"],
+            "master_uri": "tcp://127.0.0.1:{}".format(
+                temp_salt_master.config["ret_port"]
+            ),
+        }
+    )
+    master_uri = "tcp://{master_ip}:{master_port}".format(
+        master_ip="localhost", master_port=opts["master_port"]
+    )
+    with salt.channel.client.ReqChannel.factory(opts, master_uri=master_uri) as channel:
+        assert "127.0.0.1" in channel.transport.master_uri
+
+
 @pytest.mark.parametrize("message", ["", [], ()])
 def test_badload(temp_salt_minion, temp_salt_master, message):
     """
@@ -672,20 +764,24 @@ async def test_req_chan_decode_data_dict_entry_v2(minion_opts, master_opts, pki_
     # Mock auth and message client.
     auth = client.auth
     auth._crypticle = salt.crypt.Crypticle(minion_opts, AES_KEY)
+    auth._session_crypticle = salt.crypt.Crypticle(
+        minion_opts, server.session_key(target)
+    )
     client.auth = MagicMock()
     client.auth.mpub = auth.mpub
     client.auth.authenticated = True
     client.auth.get_keys = auth.get_keys
+    client.auth.gen_token = auth.gen_token
     client.auth.crypticle.dumps = auth.crypticle.dumps
     client.auth.crypticle.loads = auth.crypticle.loads
+    client.auth.session_crypticle.dumps = auth.session_crypticle.dumps
+    client.auth.session_crypticle.loads = auth.session_crypticle.loads
     client.transport = MagicMock()
 
-    print(minion_opts["encryption_algorithm"])
-
     @tornado.gen.coroutine
     def mocksend(msg, timeout=60, tries=3):
         client.transport.msg = msg
-        load = client.auth.crypticle.loads(msg["load"])
+        load = client.auth.session_crypticle.loads(msg["load"])
         ret = server._encrypt_private(
             pillar_data,
             dictkey,
@@ -708,15 +804,15 @@ def mocksend(msg, timeout=60, tries=3):
         "pillarenv": "base",
         "pillar_override": True,
         "extra_minion_data": {},
-        "ver": "2",
+        "ver": "3",
         "cmd": "_pillar",
     }
     ret = await client.crypted_transfer_decode_dictentry(  # pylint: disable=E1121,E1123
         load,
         dictkey="pillar",
     )
     assert "version" in client.transport.msg
-    assert client.transport.msg["version"] == 2
+    assert client.transport.msg["version"] == 3
     assert ret == {"pillar1": "meh"}
 
 
@@ -827,18 +923,24 @@ async def test_req_chan_decode_data_dict_entry_v2_bad_signature(
     # Mock auth and message client.
     auth = client.auth
     auth._crypticle = salt.crypt.Crypticle(minion_opts, AES_KEY)
+    auth._session_crypticle = salt.crypt.Crypticle(
+        minion_opts, server.session_key(target)
+    )
     client.auth = MagicMock()
     client.auth.mpub = auth.mpub
     client.auth.authenticated = True
     client.auth.get_keys = auth.get_keys
+    client.auth.gen_token = auth.gen_token
     client.auth.crypticle.dumps = auth.crypticle.dumps
     client.auth.crypticle.loads = auth.crypticle.loads
+    client.auth.session_crypticle.dumps = auth.session_crypticle.dumps
+    client.auth.session_crypticle.loads = auth.session_crypticle.loads
     client.transport = MagicMock()
 
     @tornado.gen.coroutine
     def mocksend(msg, timeout=60, tries=3):
         client.transport.msg = msg
-        load = client.auth.crypticle.loads(msg["load"])
+        load = client.auth.session_crypticle.loads(msg["load"])
         ret = server._encrypt_private(
             pillar_data,
             dictkey,
@@ -872,7 +974,7 @@ def mocksend(msg, timeout=60, tries=3):
         "pillarenv": "base",
         "pillar_override": True,
         "extra_minion_data": {},
-        "ver": "2",
+        "ver": "3",
         "cmd": "_pillar",
     }
 
@@ -915,18 +1017,24 @@ async def test_req_chan_decode_data_dict_entry_v2_bad_key(
     # Mock auth and message client.
     auth = client.auth
     auth._crypticle = salt.crypt.Crypticle(master_opts, AES_KEY)
+    auth._session_crypticle = salt.crypt.Crypticle(
+        minion_opts, server.session_key(target)
+    )
     client.auth = MagicMock()
     client.auth.mpub = auth.mpub
     client.auth.authenticated = True
     client.auth.get_keys = auth.get_keys
+    client.auth.gen_token = auth.gen_token
     client.auth.crypticle.dumps = auth.crypticle.dumps
     client.auth.crypticle.loads = auth.crypticle.loads
+    client.auth.session_crypticle.dumps = auth.session_crypticle.dumps
+    client.auth.session_crypticle.loads = auth.session_crypticle.loads
     client.transport = MagicMock()
 
     @tornado.gen.coroutine
     def mocksend(msg, timeout=60, tries=3):
         client.transport.msg = msg
-        load = client.auth.crypticle.loads(msg["load"])
+        load = client.auth.session_crypticle.loads(msg["load"])
         ret = server._encrypt_private(
             pillar_data,
             dictkey,
@@ -1142,10 +1250,14 @@ async def test_req_chan_auth_v2(pki_dir, io_loop, minion_opts, master_opts):
     minion_opts["verify_master_pubkey_sign"] = False
     minion_opts["always_verify_signature"] = False
     client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
+    client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
+    auth_client = salt.channel.client.AsyncReqChannel.factory(
+        minion_opts, io_loop=io_loop, crypt="clear"
+    )
     signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
+    pload = auth_client._package_load(signin_payload)
     assert "version" in pload
-    assert pload["version"] == 2
+    assert pload["version"] == 3
 
     ret = server._auth(pload["load"], sign_messages=True)
     assert "sig" in ret
@@ -1209,10 +1321,13 @@ async def test_req_chan_auth_v2_with_master_signing(
     )
 
     client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
+    auth_client = salt.channel.client.AsyncReqChannel.factory(
+        minion_opts, io_loop=io_loop, crypt="clear"
+    )
     signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
+    pload = auth_client._package_load(signin_payload)
     assert "version" in pload
-    assert pload["version"] == 2
+    assert pload["version"] == 3
 
     server_reply = server._auth(pload["load"], sign_messages=True)
     # With version 2 we always get a clear signed response
@@ -1242,7 +1357,8 @@ async def test_req_chan_auth_v2_with_master_signing(
     server.master_key = salt.crypt.MasterKeys(server.opts)
 
     signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
+
+    pload = auth_client._package_load(signin_payload)
     server_reply = server._auth(pload["load"], sign_messages=True)
     ret = client.auth.handle_signin_response(signin_payload, server_reply)
 
@@ -1301,10 +1417,13 @@ async def test_req_chan_auth_v2_new_minion_with_master_pub(
     minion_opts["verify_master_pubkey_sign"] = False
     minion_opts["always_verify_signature"] = False
     client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
+    auth_client = salt.channel.client.AsyncReqChannel.factory(
+        minion_opts, io_loop=io_loop, crypt="clear"
+    )
     signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
+    pload = auth_client._package_load(signin_payload)
     assert "version" in pload
-    assert pload["version"] == 2
+    assert pload["version"] == 3
 
     ret = server._auth(pload["load"], sign_messages=True)
     assert "sig" in ret
@@ -1368,9 +1487,12 @@ async def test_req_chan_auth_v2_new_minion_with_master_pub_bad_sig(
     minion_opts["always_verify_signature"] = False
     client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
     signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
+    auth_client = salt.channel.client.AsyncReqChannel.factory(
+        minion_opts, io_loop=io_loop, crypt="clear"
+    )
+    pload = auth_client._package_load(signin_payload)
     assert "version" in pload
-    assert pload["version"] == 2
+    assert pload["version"] == 3
 
     ret = server._auth(pload["load"], sign_messages=True)
     assert "sig" in ret
@@ -1427,11 +1549,14 @@ async def test_req_chan_auth_v2_new_minion_without_master_pub(
     minion_opts["verify_master_pubkey_sign"] = False
     minion_opts["always_verify_signature"] = False
     client = salt.channel.client.AsyncReqChannel.factory(minion_opts, io_loop=io_loop)
+    auth_client = salt.channel.client.AsyncReqChannel.factory(
+        minion_opts, io_loop=io_loop, crypt="clear"
+    )
     signin_payload = client.auth.minion_sign_in_payload()
-    pload = client._package_load(signin_payload)
+    pload = auth_client._package_load(signin_payload)
     try:
         assert "version" in pload
-        assert pload["version"] == 2
+        assert pload["version"] == 3
 
         ret = server._auth(pload["load"], sign_messages=True)
         assert "sig" in ret
@@ -1468,6 +1593,53 @@ def message_handler(payload):
     assert ret == {"msg": "bad load"}
 
 
+async def test_req_chan_bad_payload_to_decode(pki_dir, io_loop, caplog):
+    opts = {
+        "master_uri": "tcp://127.0.0.1:4506",
+        "interface": "127.0.0.1",
+        "ret_port": 4506,
+        "ipv6": False,
+        "sock_dir": ".",
+        "cachedir": "",
+        "pki_dir": str(pki_dir.joinpath("minion")),
+        "id": "minion",
+        "__role": "minion",
+        "keysize": 4096,
+        "max_minions": 0,
+        "auto_accept": False,
+        "open_mode": False,
+        "key_pass": None,
+        "publish_port": 4505,
+        "auth_mode": 1,
+        "acceptance_wait_time": 3,
+        "acceptance_wait_time_max": 3,
+    }
+    SMaster.secrets["aes"] = {
+        "secret": multiprocessing.Array(
+            ctypes.c_char,
+            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
+        ),
+        "reload": salt.crypt.Crypticle.generate_key_string,
+    }
+    master_opts = dict(opts, pki_dir=str(pki_dir.joinpath("master")))
+    master_opts["master_sign_pubkey"] = False
+    server = salt.channel.server.ReqServerChannel.factory(master_opts)
+
+    with caplog.at_level(logging.WARNING):
+        await server.handle_message(None)
+        assert "bad load received on socket" in caplog.text
+    caplog.clear()
+
+    with caplog.at_level(logging.WARNING):
+        await server.handle_message({})
+        assert "bad load received on socket" in caplog.text
+    caplog.clear()
+
+    with caplog.at_level(logging.WARNING):
+        await server.handle_message(12345)
+        assert "bad load received on socket" in caplog.text
+
+
 async def test_client_timeout_msg(minion_opts):
     client = salt.transport.zeromq.AsyncReqMessageClient(
         minion_opts, "tcp://127.0.0.1:4506"
@@ -1553,3 +1725,228 @@ async def test_unclosed_publish_client(minion_opts, io_loop):
             client.__del__()  # pylint: disable=unnecessary-dunder-call
     finally:
         client.close()
+
+
+def test_req_server_auth_garbage_sig_algo(pki_dir, minion_opts, master_opts, caplog):
+    minion_opts.update(
+        {
+            "master_uri": "tcp://127.0.0.1:4506",
+            "interface": "127.0.0.1",
+            "ret_port": 4506,
+            "ipv6": False,
+            "sock_dir": ".",
+            "pki_dir": str(pki_dir.joinpath("minion")),
+            "id": "minion",
+            "__role": "minion",
+            "keysize": 4096,
+            "max_minions": 0,
+            "auto_accept": False,
+            "open_mode": False,
+            "key_pass": None,
+            "master_sign_pubkey": False,
+            "publish_port": 4505,
+            "auth_mode": 1,
+        }
+    )
+    SMaster.secrets["aes"] = {
+        "secret": multiprocessing.Array(
+            ctypes.c_char,
+            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
+        ),
+        "reload": salt.crypt.Crypticle.generate_key_string,
+    }
+    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
+    server = salt.channel.server.ReqServerChannel.factory(master_opts)
+
+    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
+    server.cache_cli = False
+    server.event = salt.utils.event.get_master_event(
+        master_opts, master_opts["sock_dir"], listen=False
+    )
+    server.master_key = salt.crypt.MasterKeys(server.opts)
+    pub = salt.crypt.PublicKey(str(pki_dir.joinpath("master", "master.pub")))
+    token = pub.encrypt(
+        salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
+        algorithm=minion_opts["encryption_algorithm"],
+    )
+    nonce = uuid.uuid4().hex
+
+    # We need to read the public key with fopen otherwise the newlines might
+    # not match on windows.
+    with salt.utils.files.fopen(
+        str(pki_dir.joinpath("minion", "minion.pub")), "r"
+    ) as fp:
+        pub_key = salt.crypt.clean_key(fp.read())
+
+    load = {
+        "version": 2,
+        "cmd": "_auth",
+        "id": "minion",
+        "token": token,
+        "pub": pub_key,
+        "nonce": "asdfse",
+        "enc_algo": minion_opts["encryption_algorithm"],
+        "sig_algo": "IAMNOTANALGO",
+    }
+    with caplog.at_level(logging.INFO):
+        ret = server._auth(load, sign_messages=True)
+        assert (
+            "Minion tried to authenticate with unsupported signing algorithm: IAMNOTANALGO"
+            in caplog.text
+        )
+        assert "load" in ret
+        assert "ret" in ret["load"]
+        assert ret["load"]["ret"] == "bad sig algo"
+
+
+@pytest.mark.skipif(not FIPS_TESTRUN, reason="Only run on fips enabled platforms")
+def test_req_server_auth_unsupported_enc_algo(
+    pki_dir, minion_opts, master_opts, caplog
+):
+    minion_opts.update(
+        {
+            "master_uri": "tcp://127.0.0.1:4506",
+            "interface": "127.0.0.1",
+            "ret_port": 4506,
+            "ipv6": False,
+            "sock_dir": ".",
+            "pki_dir": str(pki_dir.joinpath("minion")),
+            "id": "minion",
+            "__role": "minion",
+            "keysize": 4096,
+            "max_minions": 0,
+            "auto_accept": False,
+            "open_mode": False,
+            "key_pass": None,
+            "master_sign_pubkey": False,
+            "publish_port": 4505,
+            "auth_mode": 1,
+        }
+    )
+    SMaster.secrets["aes"] = {
+        "secret": multiprocessing.Array(
+            ctypes.c_char,
+            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
+        ),
+        "reload": salt.crypt.Crypticle.generate_key_string,
+    }
+    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
+    server = salt.channel.server.ReqServerChannel.factory(master_opts)
+
+    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
+    server.cache_cli = False
+    server.event = salt.utils.event.get_master_event(
+        master_opts, master_opts["sock_dir"], listen=False
+    )
+    server.master_key = salt.crypt.MasterKeys(server.opts)
+    import tests.pytests.unit.crypt
+
+    pub = tests.pytests.unit.crypt.LegacyPublicKey(
+        str(pki_dir.joinpath("master", "master.pub"))
+    )
+    token = pub.encrypt(
+        salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
+    )
+    nonce = uuid.uuid4().hex
+
+    # We need to read the public key with fopen otherwise the newlines might
+    # not match on windows.
+    with salt.utils.files.fopen(
+        str(pki_dir.joinpath("minion", "minion.pub")), "r"
+    ) as fp:
+        pub_key = salt.crypt.clean_key(fp.read())
+
+    load = {
+        "version": 2,
+        "cmd": "_auth",
+        "id": "minion",
+        "token": token,
+        "pub": pub_key,
+        "nonce": "asdfse",
+        "enc_algo": "OAEP-SHA1",
+        "sig_algo": minion_opts["signing_algorithm"],
+    }
+    with caplog.at_level(logging.INFO):
+        ret = server._auth(load, sign_messages=True)
+        assert (
+            "Minion minion tried to authenticate with unsupported encryption algorithm: OAEP-SHA1"
+            in caplog.text
+        )
+        assert "load" in ret
+        assert "ret" in ret["load"]
+        assert ret["load"]["ret"] == "bad enc algo"
+
+
+def test_req_server_auth_garbage_enc_algo(pki_dir, minion_opts, master_opts, caplog):
+    minion_opts.update(
+        {
+            "master_uri": "tcp://127.0.0.1:4506",
+            "interface": "127.0.0.1",
+            "ret_port": 4506,
+            "ipv6": False,
+            "sock_dir": ".",
+            "pki_dir": str(pki_dir.joinpath("minion")),
+            "id": "minion",
+            "__role": "minion",
+            "keysize": 4096,
+            "max_minions": 0,
+            "auto_accept": False,
+            "open_mode": False,
+            "key_pass": None,
+            "master_sign_pubkey": False,
+            "publish_port": 4505,
+            "auth_mode": 1,
+        }
+    )
+    SMaster.secrets["aes"] = {
+        "secret": multiprocessing.Array(
+            ctypes.c_char,
+            salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
+        ),
+        "reload": salt.crypt.Crypticle.generate_key_string,
+    }
+    master_opts.update(pki_dir=str(pki_dir.joinpath("master")))
+    server = salt.channel.server.ReqServerChannel.factory(master_opts)
+
+    server.auto_key = salt.daemons.masterapi.AutoKey(server.opts)
+    server.cache_cli = False
+    server.event = salt.utils.event.get_master_event(
+        master_opts, master_opts["sock_dir"], listen=False
+    )
+    server.master_key = salt.crypt.MasterKeys(server.opts)
+    import tests.pytests.unit.crypt
+
+    pub = tests.pytests.unit.crypt.LegacyPublicKey(
+        str(pki_dir.joinpath("master", "master.pub"))
+    )
+    token = pub.encrypt(
+        salt.utils.stringutils.to_bytes(salt.crypt.Crypticle.generate_key_string()),
+    )
+    nonce = uuid.uuid4().hex
+
+    # We need to read the public key with fopen otherwise the newlines might
+    # not match on windows.
+    with salt.utils.files.fopen(
+        str(pki_dir.joinpath("minion", "minion.pub")), "r"
+    ) as fp:
+        pub_key = salt.crypt.clean_key(fp.read())
+
+    load = {
+        "version": 2,
+        "cmd": "_auth",
+        "id": "minion",
+        "token": token,
+        "pub": pub_key,
+        "nonce": "asdfse",
+        "enc_algo": "IAMNOTAENCALGO",
+        "sig_algo": minion_opts["signing_algorithm"],
+    }
+    with caplog.at_level(logging.INFO):
+        ret = server._auth(load, sign_messages=True)
+        assert (
+            "Minion minion tried to authenticate with unsupported encryption algorithm: IAMNOTAENCALGO"
+            in caplog.text
+        )
+        assert "load" in ret
+        assert "ret" in ret["load"]
+        assert ret["load"]["ret"] == "bad enc algo"

@@ -4,6 +4,7 @@
 import pytest
 
 import salt.config
+import salt.exceptions
 import salt.fileserver.gitfs
 import salt.utils.gitfs
 from salt.exceptions import FileserverConfigError
@@ -262,3 +263,74 @@ def test_checkout_pygit2(_prepare_provider):
 )
 def test_get_cachedir_basename_pygit2(_prepare_provider):
     assert "_" == _prepare_provider.get_cache_basename()
+
+
+def test_find_file(tmp_path):
+    opts = {
+        "cachedir": f"{tmp_path / 'cache'}",
+        "gitfs_user": "",
+        "gitfs_password": "",
+        "gitfs_pubkey": "",
+        "gitfs_privkey": "",
+        "gitfs_passphrase": "",
+        "gitfs_insecure_auth": False,
+        "gitfs_refspecs": salt.config._DFLT_REFSPECS,
+        "gitfs_ssl_verify": True,
+        "gitfs_branch": "master",
+        "gitfs_base": "master",
+        "gitfs_root": "",
+        "gitfs_env": "",
+        "gitfs_fallback": "",
+    }
+    remotes = []
+
+    gitfs = salt.utils.gitfs.GitFS(opts, remotes)
+    assert gitfs.find_file("asdf") == {"path": "", "rel": ""}
+
+
+def test_find_file_bad_path(tmp_path):
+    opts = {
+        "cachedir": f"{tmp_path / 'cache'}",
+        "gitfs_user": "",
+        "gitfs_password": "",
+        "gitfs_pubkey": "",
+        "gitfs_privkey": "",
+        "gitfs_passphrase": "",
+        "gitfs_insecure_auth": False,
+        "gitfs_refspecs": salt.config._DFLT_REFSPECS,
+        "gitfs_ssl_verify": True,
+        "gitfs_branch": "master",
+        "gitfs_base": "master",
+        "gitfs_root": "",
+        "gitfs_env": "",
+        "gitfs_fallback": "",
+    }
+    remotes = []
+
+    gitfs = salt.utils.gitfs.GitFS(opts, remotes)
+    with pytest.raises(salt.exceptions.SaltValidationError):
+        gitfs.find_file("sdf/../../../asdf")
+
+
+def test_find_file_bad_env(tmp_path):
+    opts = {
+        "cachedir": f"{tmp_path / 'cache'}",
+        "gitfs_user": "",
+        "gitfs_password": "",
+        "gitfs_pubkey": "",
+        "gitfs_privkey": "",
+        "gitfs_passphrase": "",
+        "gitfs_insecure_auth": False,
+        "gitfs_refspecs": salt.config._DFLT_REFSPECS,
+        "gitfs_ssl_verify": True,
+        "gitfs_branch": "master",
+        "gitfs_base": "master",
+        "gitfs_root": "",
+        "gitfs_env": "",
+        "gitfs_fallback": "",
+    }
+    remotes = []
+
+    gitfs = salt.utils.gitfs.GitFS(opts, remotes)
+    with pytest.raises(salt.exceptions.SaltValidationError):
+        gitfs.find_file("asdf", tgt_env="asd/../../../sdf")

@@ -0,0 +1,21 @@
+import pytest
+
+import salt.exceptions
+import salt.utils.virt
+
+
+def test_virt_key(tmp_path):
+    opts = {"pki_dir": f"{tmp_path / 'pki'}"}
+    salt.utils.virt.VirtKey("asdf", "minion", opts)
+
+
+def test_virt_key_bad_hyper(tmp_path):
+    opts = {"pki_dir": f"{tmp_path / 'pki'}"}
+    with pytest.raises(salt.exceptions.SaltValidationError):
+        salt.utils.virt.VirtKey("asdf/../../../sdf", "minion", opts)
+
+
+def test_virt_key_bad_id_(tmp_path):
+    opts = {"pki_dir": f"{tmp_path / 'pki'}"}
+    with pytest.raises(salt.exceptions.SaltValidationError):
+        salt.utils.virt.VirtKey("hyper", "minion/../../", opts)

@@ -1,75 +0,0 @@
-"""
-Ensure salt.utils.clean_path works with symlinked directories and files
-"""
-
-import ctypes
-
-import pytest
-
-import salt.utils.verify
-
-
-class Symlink:
-    """
-    symlink(source, link_name) Creates a symbolic link pointing to source named
-    link_name
-    """
-
-    def __init__(self):
-        self._csl = None
-
-    def __call__(self, source, link_name):
-        if self._csl is None:
-            self._csl = ctypes.windll.kernel32.CreateSymbolicLinkW
-            self._csl.argtypes = (ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.c_uint32)
-            self._csl.restype = ctypes.c_ubyte
-        flags = 0
-        if source is not None and source.is_dir():
-            flags = 1
-
-        if self._csl(str(link_name), str(source), flags) == 0:
-            raise ctypes.WinError()
-
-
-@pytest.fixture(scope="module")
-def symlink():
-    return Symlink()
-
-
-@pytest.fixture
-def setup_links(tmp_path, symlink):
-    to_path = tmp_path / "linkto"
-    from_path = tmp_path / "linkfrom"
-    if salt.utils.platform.is_windows():
-        kwargs = {}
-    else:
-        kwargs = {"target_is_directory": True}
-    if salt.utils.platform.is_windows():
-        symlink(to_path, from_path, **kwargs)
-    else:
-        from_path.symlink_to(to_path, **kwargs)
-    return to_path, from_path
-
-
-def test_clean_path_symlinked_src(setup_links):
-    to_path, from_path = setup_links
-    test_path = from_path / "test"
-    expect_path = str(to_path / "test")
-    ret = salt.utils.verify.clean_path(str(from_path), str(test_path))
-    assert ret == expect_path, f"{ret} is not {expect_path}"
-
-
-def test_clean_path_symlinked_tgt(setup_links):
-    to_path, from_path = setup_links
-    test_path = to_path / "test"
-    expect_path = str(to_path / "test")
-    ret = salt.utils.verify.clean_path(str(from_path), str(test_path))
-    assert ret == expect_path, f"{ret} is not {expect_path}"
-
-
-def test_clean_path_symlinked_src_unresolved(setup_links):
-    to_path, from_path = setup_links
-    test_path = from_path / "test"
-    expect_path = str(from_path / "test")
-    ret = salt.utils.verify.clean_path(str(from_path), str(test_path), realpath=False)
-    assert ret == expect_path, f"{ret} is not {expect_path}"

@@ -2,10 +2,81 @@
 salt.utils.clean_path works as expected
 """
 
+import ctypes
+import os
+
+import pytest
+
 import salt.utils.verify
 from tests.support.mock import patch
 
 
+class Symlink:
+    """
+    symlink(source, link_name) Creates a symbolic link pointing to source named
+    link_name
+    """
+
+    def __init__(self):
+        self._csl = None
+
+    def __call__(self, source, link_name):
+        if self._csl is None:
+            self._csl = ctypes.windll.kernel32.CreateSymbolicLinkW
+            self._csl.argtypes = (ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.c_uint32)
+            self._csl.restype = ctypes.c_ubyte
+        flags = 0
+        if source is not None and source.is_dir():
+            flags = 1
+
+        if self._csl(str(link_name), str(source), flags) == 0:
+            raise ctypes.WinError()
+
+
+@pytest.fixture(scope="module")
+def symlink():
+    return Symlink()
+
+
+@pytest.fixture
+def setup_links(tmp_path, symlink):
+    to_path = tmp_path / "linkto"
+    from_path = tmp_path / "linkfrom"
+    if salt.utils.platform.is_windows():
+        kwargs = {}
+    else:
+        kwargs = {"target_is_directory": True}
+    if salt.utils.platform.is_windows():
+        symlink(to_path, from_path, **kwargs)
+    else:
+        from_path.symlink_to(to_path, **kwargs)
+    return to_path, from_path
+
+
+def test_clean_path_symlinked_src(setup_links):
+    to_path, from_path = setup_links
+    test_path = from_path / "test"
+    expect_path = str(to_path / "test")
+    ret = salt.utils.verify.clean_path(str(from_path), str(test_path))
+    assert ret == expect_path, f"{ret} is not {expect_path}"
+
+
+def test_clean_path_symlinked_tgt(setup_links):
+    to_path, from_path = setup_links
+    test_path = to_path / "test"
+    expect_path = str(to_path / "test")
+    ret = salt.utils.verify.clean_path(str(from_path), str(test_path))
+    assert ret == expect_path, f"{ret} is not {expect_path}"
+
+
+def test_clean_path_symlinked_src_unresolved(setup_links):
+    to_path, from_path = setup_links
+    test_path = from_path / "test"
+    expect_path = str(from_path / "test")
+    ret = salt.utils.verify.clean_path(str(from_path), str(test_path), realpath=False)
+    assert ret == expect_path, f"{ret} is not {expect_path}"
+
+
 def test_clean_path_valid(tmp_path):
     path_a = str(tmp_path / "foo")
     path_b = str(tmp_path / "foo" / "bar")
@@ -23,3 +94,24 @@ def test_clean_path_relative_root(tmp_path):
         path_a = "foo"
         path_b = str(tmp_path / "foo" / "bar")
         assert salt.utils.verify.clean_path(path_a, path_b) == path_b
+
+
+def test_clean_traverse_in_path_a(tmp_path):
+    path_a = str(tmp_path)
+    path_b = str(tmp_path / "foo" / ".." / "bar")
+    assert salt.utils.verify.clean_path(path_a, path_b) == os.path.normpath(path_b)
+
+
+def test_clean_traverse_in_path_b(tmp_path):
+    path_a = str(tmp_path)
+    path_b = str(tmp_path / "foo.foo/../bar")
+    assert salt.utils.verify.clean_path(path_a, path_b) == os.path.normpath(path_b)
+
+
+def test_clean_traverse_in_path_c(tmp_path):
+    path_a = str(tmp_path)
+    path_b = str(tmp_path / "foo/../bar/bang")
+    assert salt.utils.verify.clean_path(path_a, path_b) == ""
+    assert salt.utils.verify.clean_path(
+        path_a, path_b, subdir=True
+    ) == os.path.normpath(path_b)

@@ -0,0 +1,44 @@
+import pytest
+
+import salt.utils.verify
+
+
+@pytest.mark.parametrize(
+    "data, result",
+    [
+        ("https://saltproject.io", True),
+        (
+            "https://mail.google.com/mail/u/0/#inbox/FMfcgzQbdrMwJwbwbPfCFLjMRQvWVcJK",
+            True,
+        ),
+        ("http://parts.org/foo/bar=/bat", True),
+        ("foobar://saltproject.io", False),
+        ("http://parts.org/foo/b\nar=/bat", False),
+        (
+            'base ssh://fake@git/repo\n[core]\nsshCommand = touch /tmp/pwn\n[remote "origin"]\n',
+            False,
+        ),
+        (
+            'ssh://fake@git/repo\n[core]\nsshCommand = touch /tmp/pwn\n[remote "origin"]\n',
+            False,
+        ),
+        ("https://github.com/saltstack/salt-test-pillar-gitfs.git", True),
+    ],
+)
+def test_url_validator(data, result):
+    assert salt.utils.verify.URLValidator()(data) is result
+
+
+@pytest.mark.parametrize(
+    "data, result",
+    [
+        ("asdf", True),
+        ("asdf-", True),
+        ("0123456789abcdefghijklmnopqrstuv-._~!$&'():@,", True),
+        ("0123456789ABCDEFGHIJKLMNOPQRSTUV-._~!$&'():@,", True),
+        ("abcd\\efg", False),
+    ],
+)
+def test_pchar_validator(data, result):
+    matcher = salt.utils.verify.URLValidator.pchar_matcher()
+    assert bool(matcher.match(data)) == result

@@ -17,6 +17,7 @@
 import logging
 
 import salt.utils.files
+import salt.utils.versions
 
 # pylint: disable=import-error
 try:
@@ -30,7 +31,7 @@
             from Cryptodome.Util import asn1
         except ImportError:
             from Crypto.Util import asn1  # nosec
-        import OpenSSL
+        import OpenSSL  # pylint: disable=W8410
     HAS_DEPS = True
 except ImportError:
     HAS_DEPS = False
@@ -71,6 +72,10 @@ def auth(username, password, **kwargs):
             your_user:
               - .*
     """
+    salt.utils.versions.warn_until(
+        "Argon",
+        "This module has been deprecated as it is known to be insecure.",
+    )
     pem = password
     cacert_file = __salt__["config.get"]("external_auth:pki:ca_file")