new feature: refactor server-side PKI

the end goal of this work is to support a non file/disk based interface
for pub key management. Using the existing salt.cache interface allows
us to leverage existing implementations of other storage backends (ie
mysql, postgres, redis, etc). Of note, this will allow a common shared
view of pub keys for multi-master setups.

To avoid a break-the-world scenario in this change, a fully backward
compatible salt.cache.localfs_key_backcompat is provided as the new
default that emulates the disk operations salt.key was doing before.

open discussion items:
- as is i've left master side pub/priv key as is to live in
  etc/salt/pki; it COULD be moved to the same interface if thats desired

Author: mattp-

changelog/67799.added.md

@@ -0,0 +1 @@
+refactored server-side PKI to support cache interface

doc/ref/cache/all/index.rst

@@ -15,5 +15,6 @@ For understanding and usage of the cache modules see the :ref:`cache` topic.
     consul
     etcd_cache
     localfs
+    localfs_key_backcompat
     mysql_cache
     redis_cache

doc/ref/cache/all/salt.cache.localfs_key_backcompat.rst

@@ -0,0 +1,5 @@
+salt.cache.localfs_key_backcompat
+=================================
+
+.. automodule:: salt.cache.localfs_key_backcompat
+    :members:

salt/cache/__init__.py

@@ -62,7 +62,9 @@ def __init__(self, opts, cachedir=None, **kwargs):
             self.cachedir = opts.get("cachedir", salt.syspaths.CACHE_DIR)
         else:
             self.cachedir = cachedir
-        self.driver = opts.get("cache", salt.config.DEFAULT_MASTER_OPTS["cache"])
+        self.driver = kwargs.get(
+            "driver", opts.get("cache", salt.config.DEFAULT_MASTER_OPTS["cache"])
+        )
         self._modules = None
         self._kwargs = kwargs
         self._kwargs["cachedir"] = self.cachedir

salt/cache/localfs_key_backcompat.py

@@ -0,0 +1,287 @@
+"""
+Backward compatible shim layer for pki interaction
+
+.. versionadded:: 3008.0
+
+The ``localfs_keys_backcompat`` is a shim driver meant to allow the salt.cache
+subsystem to interact with the existing master pki folder/file structure
+without any migration from previous versions of salt.  It is not meant for
+general purpose use and should not be used outside of the master auth system.
+
+The main difference from before is the 'state' of the key, ie accepted/rejected
+is now stored in the data itself, as opposed to the cache equivalent of a bank
+previously.
+
+store and fetch handle ETL from new style, where data itself contains key
+state, to old style, where folder and/or bank contain state.
+flush/list/contains/updated are left as nearly equivalent to localfs, without
+the .p file extension to work with legacy keys via banks.
+"""
+
+import errno
+import logging
+import os
+import os.path
+import shutil
+import tempfile
+
+import salt.utils.atomicfile
+import salt.utils.files
+from salt.exceptions import SaltCacheError
+from salt.utils.verify import valid_id
+
+log = logging.getLogger(__name__)
+
+__func_alias__ = {"list_": "list"}
+
+BASE_MAPPING = {
+    "minions_pre": "pending",
+    "minions_rejected": "rejected",
+    "minions": "accepted",
+    "minions_denied": "denied",
+}
+
+
+# we explicitly override cache dir to point to pki here
+def init_kwargs(kwargs):
+    """
+    setup kwargs for cache functions
+    """
+    if __opts__.get("cluster_id"):
+        pki_dir = __opts__["cluster_pki_dir"]
+    else:
+        pki_dir = __opts__["pki_dir"]
+
+    return {"cachedir": pki_dir, "user": kwargs.get("user")}
+
+
+def store(bank, key, data, cachedir, user, **kwargs):
+    """
+    Store key state information. storing a accepted/pending/rejected state
+    means clearing it from the other 2. denied is handled separately
+    """
+    if not valid_id(__opts__, key):
+        raise SaltCacheError(f"key {key} is not a valid minion_id")
+
+    if bank not in ["keys", "denied_keys"]:
+        raise SaltCacheError(f"Unrecognized bank: {bank}")
+
+    if bank == "keys":
+        if data["state"] == "rejected":
+            base = "minions_rejected"
+        elif data["state"] == "pending":
+            base = "minions_pre"
+        elif data["state"] == "accepted":
+            base = "minions"
+        else:
+            raise SaltCacheError("Unrecognized data/bank: {}".format(data["state"]))
+        data = data["pub"]
+    elif bank == "denied_keys":
+        # denied keys is a list post migration, but is a single key in legacy
+        data = data[0]
+        base = "minions_denied"
+
+    base = os.path.join(cachedir, base)
+    savefn = os.path.join(base, key)
+
+    try:
+        os.makedirs(base)
+    except OSError as exc:
+        if exc.errno != errno.EEXIST:
+            raise SaltCacheError(
+                f"The cache directory, {base}, could not be created: {exc}"
+            )
+
+    # delete current state before re-serializing new state
+    flush(bank, key, cachedir, **kwargs)
+
+    if __opts__["permissive_pki_access"]:
+        umask = 0o0700
+    else:
+        umask = 0o0750
+
+    tmpfh, tmpfname = tempfile.mkstemp(dir=base)
+    os.close(tmpfh)
+
+    if user:
+        try:
+            import pwd
+
+            uid = pwd.getpwnam(user).pw_uid
+            os.chown(tmpfname, uid, -1)
+        except (KeyError, ImportError, OSError):
+            # The specified user was not found, allow the backup systems to
+            # report the error
+            pass
+
+    try:
+        with salt.utils.files.set_umask(umask):
+            with salt.utils.files.fopen(tmpfname, "w+b") as fh_:
+                fh_.write(data.encode("utf-8"))
+        # On Windows, os.rename will fail if the destination file exists.
+        salt.utils.atomicfile.atomic_rename(tmpfname, savefn)
+    except OSError as exc:
+        raise SaltCacheError(
+            f"There was an error writing the cache file, {base}: {exc}"
+        )
+
+
+def fetch(bank, key, cachedir, **kwargs):
+    """
+    Fetch and construct state data for a given minion based on the bank and id
+    """
+    if not valid_id(__opts__, key):
+        raise SaltCacheError(f"key {key} is not a valid minion_id")
+
+    if bank not in ["keys", "denied_keys"]:
+        raise SaltCacheError(f"Unrecognized bank: {bank}")
+
+    if key == ".key_cache":
+        raise SaltCacheError("trying to read key_cache, there is a bug at call-site")
+    try:
+        if bank == "keys":
+            for state, bank in [
+                ("rejected", "minions_rejected"),
+                ("pending", "minions_pre"),
+                ("accepted", "minions"),
+            ]:
+                pubfn = os.path.join(cachedir, bank, key)
+                if os.path.isfile(pubfn):
+                    with salt.utils.files.fopen(pubfn, "r") as fh_:
+                        return {"state": state, "pub": fh_.read()}
+            return None
+        elif bank == "denied_keys":
+            # there can be many denied keys per minion post refactor, but only 1
+            # with the filesystem, so return a list of 1
+            pubfn_denied = os.path.join(cachedir, "minions_denied", key)
+
+            if os.path.isfile(pubfn_denied):
+                with salt.utils.files.fopen(pubfn_denied, "r") as fh_:
+                    return [fh_.read()]
+        else:
+            raise SaltCacheError(f'unrecognized bank "{bank}"')
+    except OSError as exc:
+        raise SaltCacheError(
+            'There was an error reading the cache bank "{}", key "{}": {}'.format(
+                bank, key, exc
+            )
+        )
+
+
+def updated(bank, key, cachedir, **kwargs):
+    """
+    Return the epoch of the mtime for this cache file
+    """
+    if not valid_id(__opts__, key):
+        raise SaltCacheError(f"key {key} is not a valid minion_id")
+
+    if bank == "keys":
+        bases = [base for base in BASE_MAPPING if base != "minions_denied"]
+    elif bank == "denied_keys":
+        bases = ["minions_denied"]
+    else:
+        raise SaltCacheError(f"Unrecognized bank: {bank}")
+
+    for dir in bases:
+        key_file = os.path.join(cachedir, dir, key)
+        if os.path.isfile(key_file):
+            try:
+                return int(os.path.getmtime(key_file))
+            except OSError as exc:
+                raise SaltCacheError(
+                    'There was an error reading the mtime for "{}": {}'.format(
+                        key_file, exc
+                    )
+                )
+    log.debug('pki file "%s" does not exist in accepted/rejected/pending', key)
+    return
+
+
+def flush(bank, key=None, cachedir=None, **kwargs):
+    """
+    Remove the key from the cache bank with all the key content.
+    flush can take a legacy bank or a keys/denied_keys modern bank
+    """
+    if not valid_id(__opts__, key):
+        raise SaltCacheError(f"key {key} is not a valid minion_id")
+
+    if cachedir is None:
+        raise SaltCacheError("cachedir missing")
+
+    if bank == "keys":
+        bases = [base for base in BASE_MAPPING if base != "minions_denied"]
+    elif bank == "denied_keys":
+        bases = ["minions_denied"]
+    else:
+        raise SaltCacheError(f"Unrecognized bank: {bank}")
+
+    flushed = False
+
+    for base in bases:
+        try:
+            if key is None:
+                target = os.path.join(cachedir, base)
+                if not os.path.isdir(target):
+                    return False
+                shutil.rmtree(target)
+            else:
+                target = os.path.join(cachedir, base, key)
+                if not os.path.isfile(target):
+                    continue
+                os.remove(target)
+                flushed = True
+        except OSError as exc:
+            raise SaltCacheError(f'There was an error removing "{target}": {exc}')
+    return flushed
+
+
+def list_(bank, cachedir, **kwargs):
+    """
+    Return an iterable object containing all entries stored in the specified bank.
+    """
+    if bank == "keys":
+        bases = [base for base in BASE_MAPPING if base != "minions_denied"]
+    elif bank == "denied_keys":
+        bases = ["minions_denied"]
+    else:
+        raise SaltCacheError(f"Unrecognized bank: {bank}")
+
+    ret = []
+    for base in bases:
+        base = os.path.join(cachedir, os.path.normpath(base))
+        if not os.path.isdir(base):
+            continue
+        try:
+            items = os.listdir(base)
+        except OSError as exc:
+            raise SaltCacheError(
+                f'There was an error accessing directory "{base}": {exc}'
+            )
+        for item in items:
+            # salt foolishly dumps a file here for key cache, ignore it
+            if valid_id(__opts__, item):
+                ret.append(item)
+            else:
+                log.error("saw invalid id %s, discarding", item)
+    return ret
+
+
+def contains(bank, key, cachedir, **kwargs):
+    """
+    Checks if the specified bank contains the specified key.
+    """
+    if not valid_id(__opts__, key):
+        raise SaltCacheError(f"key {key} is not a valid minion_id")
+
+    if bank == "keys":
+        bases = [base for base in BASE_MAPPING if base != "minions_denied"]
+    elif bank == "denied_keys":
+        bases = ["minions_denied"]
+    else:
+        raise SaltCacheError(f"Unrecognized bank: {bank}")
+
+    for base in bases:
+        keyfile = os.path.join(cachedir, base, key)
+        if os.path.isfile(keyfile):
+            return True
+    return False

salt/channel/client.py

@@ -18,7 +18,6 @@
 import salt.transport.frame
 import salt.utils.event
 import salt.utils.files
-import salt.utils.minions
 import salt.utils.stringutils
 import salt.utils.verify
 import salt.utils.versions
@@ -204,7 +203,7 @@ def crypted_transfer_decode_dictentry(
             timeout,
         )
         key = self.auth.get_keys()
-        if "key" not in ret:
+        if not isinstance(ret, dict) or "key" not in ret:
             # Reauth in the case our key is deleted on the master side.
             yield self.auth.authenticate()
             ret = yield self._send_with_retry(
@@ -235,7 +234,7 @@ def crypted_transfer_decode_dictentry(
         raise tornado.gen.Return(data["pillar"])
 
     def verify_signature(self, data, sig):
-        return salt.crypt.PublicKey(self.master_pubkey_path).verify(
+        return salt.crypt.PublicKey.from_file(self.master_pubkey_path).verify(
             data, sig, self.opts["signing_algorithm"]
         )