fixed gpg.decrypt function

changelog/62977.fixed

@@ -0,0 +1 @@
+Fixed gpg_passphrase issue with gpg decrypt/encrypt functions

salt/modules/gpg.py

@@ -1158,6 +1158,7 @@ def encrypt(
     output=None,
     sign=None,
     use_passphrase=False,
+    always_trust=False,
     gnupghome=None,
     bare=False,
 ):
@@ -1170,7 +1171,8 @@ def encrypt(
         ``/etc/salt/gpgkeys``.
 
     recipients
-        The fingerprints for those recipient whom the data is being encrypted for.
+        The key ID, fingerprint, user ID or email address associated with the recipients
+        key can be used.
 
     text
         The text to encrypt.
@@ -1186,8 +1188,11 @@ def encrypt(
         default key or fingerprint to specify a different key to sign with.
 
     use_passphrase
-        Whether to use a passphrase with the signing key. Passphrase is received
-        from Pillar.
+        Whether to use a passphrase with the signing key.
+        Passphrase is received from Pillar.
+
+    always_trust
+        Skip key validation and assume that used keys are fully trusted.
 
     gnupghome
         Specify the location where GPG keyring and related files are stored.
@@ -1211,30 +1216,32 @@ def encrypt(
     ret = {"res": True, "comment": ""}
     gpg = _create_gpg(user, gnupghome)
 
-    if use_passphrase:
+    if sign and use_passphrase:
         gpg_passphrase = __salt__["pillar.get"]("gpg_passphrase")
         if not gpg_passphrase:
             raise SaltInvocationError("gpg_passphrase not available in pillar.")
-        gpg_passphrase = gpg_passphrase["gpg_passphrase"]
     else:
         gpg_passphrase = None
 
     if text:
-        result = gpg.encrypt(text, recipients, passphrase=gpg_passphrase)
+        result = gpg.encrypt(
+            text,
+            recipients,
+            sign=sign,
+            passphrase=gpg_passphrase,
+            always_trust=always_trust,
+            output=output,
+        )
     elif filename:
         with salt.utils.files.flopen(filename, "rb") as _fp:
-            if output:
-                result = gpg.encrypt_file(
-                    _fp,
-                    recipients,
-                    passphrase=gpg_passphrase,
-                    output=output,
-                    sign=sign,
-                )
-            else:
-                result = gpg.encrypt_file(
-                    _fp, recipients, passphrase=gpg_passphrase, sign=sign
-                )
+            result = gpg.encrypt_file(
+                _fp,
+                recipients,
+                sign=sign,
+                passphrase=gpg_passphrase,
+                always_trust=always_trust,
+                output=output,
+            )
     else:
         raise SaltInvocationError("filename or text must be passed.")
 
@@ -1254,7 +1261,9 @@ def encrypt(
             )
         else:
             ret = False
+
         log.error(result.stderr)
+
     return ret
 
 
@@ -1310,7 +1319,6 @@ def decrypt(
         gpg_passphrase = __salt__["pillar.get"]("gpg_passphrase")
         if not gpg_passphrase:
             raise SaltInvocationError("gpg_passphrase not available in pillar.")
-        gpg_passphrase = gpg_passphrase["gpg_passphrase"]
     else:
         gpg_passphrase = None
 

tests/pytests/unit/modules/test_gpg.py

@@ -865,13 +865,96 @@ def test_gpg_sign(gpghome):
     with patch.dict(gpg.__salt__, {"config.option": config_user}):
         with patch.dict(gpg.__salt__, {"user.info": user_info}):
             with patch.dict(gpg.__salt__, {"pillar.get": pillar_mock}):
-                ret = gpg.import_key(None, str(gpghome.priv), "salt", str(gpghome.path))
-                assert ret["res"] is True
+                key = gpg.import_key(None, str(gpghome.priv), "salt", str(gpghome.path))
+                assert key["res"] is True
                 gpg_text_input = "The quick brown fox jumped over the lazy dog"
-                gpg_sign_output = gpg.sign(
+                ret = gpg.sign(
                     keyid=GPG_TEST_KEY_ID,
                     text=gpg_text_input,
                     use_passphrase=True,
                     gnupghome=str(gpghome.path),
                 )
-                assert "-----BEGIN PGP SIGNATURE-----" in str(gpg_sign_output)
+                assert "-----BEGIN PGP SIGNATURE-----" in str(ret)
+
+
+def test_gpg_encrypt_message(gpghome):
+    config_user = MagicMock(return_value="salt")
+
+    user_info = MagicMock(
+        return_value={"name": "salt", "home": str(gpghome.path), "uid": 1000}
+    )
+    with patch.dict(gpg.__salt__, {"config.option": config_user}):
+        with patch.dict(gpg.__salt__, {"user.info": user_info}):
+            key = gpg.import_key(None, str(gpghome.priv), "salt", str(gpghome.path))
+            assert key["res"] is True
+            gpg_text_input = "The quick brown fox jumped over the lazy dog"
+            ret = gpg.encrypt(
+                recipients=GPG_TEST_KEY_ID,
+                text=gpg_text_input,
+                use_passphrase=True,
+                always_trust=True,
+                gnupghome=str(gpghome.path),
+            )
+            assert ret["res"] is True
+
+
+def test_gpg_encrypt_and_sign_message_with_gpg_passphrase_in_pillar(gpghome):
+    config_user = MagicMock(return_value="salt")
+
+    user_info = MagicMock(
+        return_value={"name": "salt", "home": str(gpghome.path), "uid": 1000}
+    )
+    pillar_mock = MagicMock(return_value=GPG_TEST_KEY_PASSPHRASE)
+    with patch.dict(gpg.__salt__, {"config.option": config_user}):
+        with patch.dict(gpg.__salt__, {"user.info": user_info}):
+            with patch.dict(gpg.__salt__, {"pillar.get": pillar_mock}):
+                key = gpg.import_key(None, str(gpghome.priv), "salt", str(gpghome.path))
+                assert key["res"] is True
+                gpg_text_input = "The quick brown fox jumped over the lazy dog"
+                ret = gpg.encrypt(
+                    recipients=GPG_TEST_KEY_ID,
+                    text=gpg_text_input,
+                    sign=True,
+                    use_passphrase=True,
+                    always_trust=True,
+                    gnupghome=str(gpghome.path),
+                )
+                assert ret["res"] is True
+
+
+def test_gpg_decrypt_message_with_gpg_passphrase_in_pillar(gpghome):
+    """
+    Test gpg.decrypt with passphrase and gpg_passphrase pillar
+    """
+    gpg_encrypted_message = """-----BEGIN PGP MESSAGE-----
+    hQGMA7z9rKs9ZvTOAQwAnMbwchCm1VXOD+Ml0rnNrhDhsRm+6O96FOq5lWY0ntkj
+    vnXeFOgUf0wzK4hkQT/Yo4/ZpDkV3iwwSIjesqNDS1U/KWfbe2pFeph6w9fHFnXf
+    e8RJY8OGJHN8A9TlCIyKDVKGoXZEcSDZ4K0fNL/OxF3LxsSjJ894F+e2MCnoQZgq
+    3uxQFYb9Gec8iPpBcd7s74J8cX1To9rBKRNSP73NRwVvKbPgB0NjzrLVPmqHGO7Q
+    wFf9tZdGYmxV/AB9H0gCUITBCwnndissVqFpjUpVEIL1jOBus73oiL9Grn5QlZic
+    wDOZyKDlWR15jiiDaWvq4ES6O0j2Oj/FodItJhTbzeXEFOxn7cNbW2kAAc08Hs1u
+    zPI2LywaCSM/IOxrRI8n/ExXxtU0k48827AWTLYmhm+vDct7gQHEJ8Qu3/d5FsP/
+    f1qWrtl+E7TpIIlUIgtPL596LYiwQLisqh1JeqqleSnBESIzgg7teolYdYED606z
+    j9a1oTUHvSnKz6Q5Yh850mYBCBc/iEn0gJBgseGK6rXJF54f5duybkuTKrV+7hUd
+    euy/+K0mnQWPS0G9NhCdQAaydNBZywOyQU6Yv1gPHapRrHOkhZq51sBKmZRGdqH/
+    llX+Wd7pUetBV+vmZp75l8q9CNVpefA=
+    =WXYF
+    -----END PGP MESSAGE-----
+    """
+
+    config_user = MagicMock(return_value="salt")
+    user_info = MagicMock(
+        return_value={"name": "salt", "home": str(gpghome.path), "uid": 1000}
+    )
+    pillar_mock = MagicMock(return_value=GPG_TEST_KEY_PASSPHRASE)
+    with patch.dict(gpg.__salt__, {"config.option": config_user}):
+        with patch.dict(gpg.__salt__, {"user.info": user_info}):
+            with patch.dict(gpg.__salt__, {"pillar.get": pillar_mock}):
+                key = gpg.import_key(None, str(gpghome.priv), "salt", str(gpghome.path))
+                assert key["res"] is True
+                ret = gpg.decrypt(
+                    text=gpg_encrypted_message,
+                    use_passphrase=True,
+                    gnupghome=str(gpghome.path),
+                )
+                assert ret["res"] is True