Create GNUPGHOME if it does not exist

changelog/66312.fixed.md

@@ -0,0 +1 @@
+Made gpg modules create GNUPGHOME if it does not exist

salt/modules/gpg.py

@@ -173,32 +173,35 @@ def _restore_ownership(func):
         userinfo = _get_user_info(user)
         run_user = _get_user_info()
 
+        if not os.path.exists(gnupghome):
+            _create_gnupghome(user, gnupghome)
+
         if userinfo["uid"] != run_user["uid"]:
-            group = None
-            if os.path.exists(gnupghome):
             # Given user is different from one who runs Salt process,
             # need to fix ownership permissions for GnuPG home dir
-                group = __salt__["file.gid_to_group"](run_user["gid"])
             for path in [gnupghome] + __salt__["file.find"](gnupghome):
-                    __salt__["file.chown"](path, run_user["name"], group)
+                __salt__["file.chown"](
+                    path, user=run_user["uid"], group=run_user["gid"]
+                )
             if keyring and os.path.exists(keyring):
-                if group is None:
-                    group = __salt__["file.gid_to_group"](run_user["gid"])
-                __salt__["file.chown"](keyring, run_user["name"], group)
+                __salt__["file.chown"](
+                    keyring, user=run_user["uid"], group=run_user["gid"]
+                )
 
         # Filter special kwargs
-        for key in list(kwargs):
-            if key.startswith("__"):
-                del kwargs[key]
+        filtered_kwargs = {k: v for k, v in kwargs.items() if not k.startswith("__")}
 
-        ret = func(*args, **kwargs)
+        ret = func(*args, **filtered_kwargs)
 
         if userinfo["uid"] != run_user["uid"]:
-            group = __salt__["file.gid_to_group"](userinfo["gid"])
             for path in [gnupghome] + __salt__["file.find"](gnupghome):
-                __salt__["file.chown"](path, user, group)
+                __salt__["file.chown"](
+                    path, user=userinfo["uid"], group=userinfo["gid"]
+                )
             if keyring and os.path.exists(keyring):
-                __salt__["file.chown"](keyring, user, group)
+                __salt__["file.chown"](
+                    keyring, user=userinfo["uid"], group=userinfo["gid"]
+                )
         return ret
 
     return func_wrapper
@@ -216,11 +219,24 @@ def _create_gpg(user=None, gnupghome=None, keyring=None):
             "Please pass keyring as a string. Multiple keyrings are not allowed"
         )
 
+    try:
+        gpg = gnupg.GPG(gnupghome=gnupghome, keyring=keyring)
+    except ValueError as err:
+        if not str(err).startswith("gnupghome should be a directory"):
+            raise
+        _create_gnupghome(user, gnupghome)
         gpg = gnupg.GPG(gnupghome=gnupghome, keyring=keyring)
 
     return gpg
 
 
+def _create_gnupghome(user, gnupghome):
+    user_info = _get_user_info(user)
+    __salt__["file.mkdir"](
+        gnupghome, user=user_info["uid"], group=user_info["gid"], mode="0700"
+    )
+
+
 def _list_keys(secret=False, user=None, gnupghome=None, keyring=None):
     """
     Helper function for listing keys

tests/pytests/functional/modules/test_gpg.py

@@ -12,14 +12,7 @@ pytestmark = [
 ]
 
 
-@pytest.fixture
-def gpghome(tmp_path):
-    root = tmp_path / "gpghome"
-    root.mkdir(mode=0o0700)
-    try:
-        yield root
-    finally:
-        # Make sure we don't leave any gpg-agents running behind
+def _kill_gpg_agent(root):
     gpg_connect_agent = shutil.which("gpg-connect-agent")
     if gpg_connect_agent:
         gnupghome = root / ".gnupg"
@@ -50,6 +43,17 @@ def gpghome(tmp_path):
             pass
 
 
+@pytest.fixture
+def gpghome(tmp_path):
+    root = tmp_path / "gpghome"
+    root.mkdir(mode=0o0700)
+    try:
+        yield root
+    finally:
+        # Make sure we don't leave any gpg-agents running behind
+        _kill_gpg_agent(root)
+
+
 @pytest.fixture
 def gpg(loaders, modules, gpghome):
     yield modules.gpg
@@ -931,3 +935,17 @@ def test_read_key_multiple(
     else:
         assert len(res) == 1
     assert any(key["fingerprint"] == key_a_fp for key in res)
+
+
+def test_missing_gnupghome(gpg, tmp_path):
+    """
+    Ensure the directory passed as `gnupghome` is created before
+    python-gnupg is invoked. Issue #66312.
+    """
+    gnupghome = tmp_path / "gnupghome"
+    try:
+        res = gpg.list_keys(gnupghome=tmp_path / "gnupghome")
+        assert res == []
+    finally:
+        # Make sure we don't leave any gpg-agents running behind
+        _kill_gpg_agent(gnupghome)