[master] Acme fixes

changelog/63700.fixed.md

@@ -0,0 +1 @@
+Fix missing arg in acme state/module "dns_plugin_propagate_seconds" that only existed in docstrings.

changelog/64686.added.md

@@ -0,0 +1,3 @@
+* Added `replace_staging` to Acme state.
+* Added `revoke` to Acme module.
+* Added `delete` to Acme module

salt/modules/acme.py

@@ -132,6 +132,7 @@ def cert(
     http_01_address=None,
     dns_plugin=None,
     dns_plugin_credentials=None,
+    dns_plugin_propagate_seconds=None,
 ):
     """
     Obtain/renew a certificate from an ACME CA, probably Let's Encrypt.
@@ -216,6 +217,10 @@ def cert(
         if dns_plugin == "cloudflare":
             cmd.append("--dns-cloudflare")
             cmd.append(f"--dns-cloudflare-credentials {dns_plugin_credentials}")
+            if dns_plugin_propagate_seconds:
+                cmd.append(
+                    f"--dns-cloudflare-propagation-seconds {dns_plugin_propagate_seconds}"
+                )
         else:
             return {
                 "result": False,
@@ -421,3 +426,93 @@ def needs_renewal(name, window=None):
         window = int(window)
 
     return _renew_by(name, window) <= datetime.datetime.today()
+
+
+def revoke(name, reason=None):
+    """
+    .. versionadded:: 3007.0
+
+    Revoke certificate
+    The revoked certificate will also be deleted due to the `--non-interactive` flag.
+    :param str name: Name of certificate
+    :param str reason: Optional - why you are revoking the certificate; default is 'unspecified'.
+    :rtype: bool
+    :return: Whether or not the certificate was successfully revoked.
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt 'gitlab.example.com' acme.revoke dev.example.com
+
+
+    Code example:
+    .. code-block:: python
+
+         __salt__['acme.revoke']('dev.example.com'):
+
+    """
+    cmd = [LEA, "revoke", "--non-interactive", f"--cert-name {name}"]
+    if reason:
+        cmd.append(f"--reason {reason}")
+
+    res = __salt__["cmd.run_all"](" ".join(cmd))
+
+    if res["retcode"] != 0:
+        if "expand" in res["stderr"]:
+            cmd.append("--expand")
+            res = __salt__["cmd.run_all"](" ".join(cmd))
+            if res["retcode"] != 0:
+                return {
+                    "result": False,
+                    "comment": f"Certificate {name} revocation failed with:\n{res['stderr']}",
+                }
+        else:
+            return {
+                "result": False,
+                "comment": f"Certificate {name} revocation failed with:\n{res['stderr']}",
+            }
+    return True
+
+
+def delete(name):
+    """
+    .. versionadded:: 3007.0
+
+    Delete certificate
+    :param str name: Name of certificate
+    :rtype: bool
+    :return: Whether or not the certificate was successfully deleted.
+
+    CLI Example:
+
+    .. code-block:: bash
+
+        salt 'gitlab.example.com' acme.delete dev.example.com
+
+
+    Code example:
+    .. code-block:: python
+
+         __salt__['acme.delete']('dev.example.com'):
+
+    """
+    cmd = [LEA, "delete", "--non-interactive", f"--cert-name {name}"]
+
+    res = __salt__["cmd.run_all"](" ".join(cmd))
+
+    if res["retcode"] != 0:
+        if "expand" in res["stderr"]:
+            cmd.append("--expand")
+            res = __salt__["cmd.run_all"](" ".join(cmd))
+            if res["retcode"] != 0:
+                return {
+                    "result": False,
+                    "comment": f"Certificate {name} deletion failed with:\n{res['stderr']}",
+                }
+        else:
+            return {
+                "result": False,
+                "comment": f"Certificate {name} deletion failed with:\n{res['stderr']}",
+            }
+    return True

salt/states/acme.py

@@ -61,6 +61,8 @@ def cert(
     http_01_address=None,
     dns_plugin=None,
     dns_plugin_credentials=None,
+    dns_plugin_propagate_seconds=None,
+    replace_staging=None,
 ):
     """
     Obtain/renew a certificate from an ACME CA, probably Let's Encrypt.
@@ -91,6 +93,8 @@ def cert(
     :param https_01_address: The address the server listens to during http-01 challenge.
     :param dns_plugin: Name of a DNS plugin to use (currently only 'cloudflare')
     :param dns_plugin_credentials: Path to the credentials file if required by the specified DNS plugin
+    :param dns_plugin_propagate_seconds: How many seconds to wait for the DNS change to be live
+    :param bool replace_staging: Whether or not to replace the staging (test) certificate of the same name with production. The revoked certificate will also be deleted.
     """
 
     if certname is None:
@@ -101,6 +105,14 @@ def cert(
 
     current_certificate = {}
     new_certificate = {}
+
+    if replace_staging and not test_cert:
+        if (
+            __salt__["acme.has"](certname)
+            and __salt__["acme.info"](certname)["issuer"]["O"].find("STAGING") != -1
+        ):
+            __salt__["acme.revoke"](certname)
+
     if not __salt__["acme.has"](certname):
         action = "obtain"
     elif __salt__["acme.needs_renewal"](certname, renew):
@@ -138,6 +150,7 @@ def cert(
                 http_01_address=http_01_address,
                 dns_plugin=dns_plugin,
                 dns_plugin_credentials=dns_plugin_credentials,
+                dns_plugin_propagate_seconds=dns_plugin_propagate_seconds,
             )
             ret["result"] = res["result"]
             ret["comment"].append(res["comment"])

tests/pytests/unit/modules/test_acme.py

@@ -338,3 +338,83 @@ def test_cert():
     ):
         assert acme.cert("test") == result_renew
         assert acme.cert("testing.example.com", certname="test") == result_renew
+
+
+def test_revoke():
+    """
+    Test certificate revocation
+    """
+    # Test successful revocation
+    with patch.dict(
+        acme.__salt__, {"cmd.run_all": MagicMock(return_value={"retcode": 0})}
+    ):
+        assert acme.revoke("test_cert") is True
+
+    # Test failure with non-zero retcode and without 'expand' in stderr
+    with patch.dict(
+        acme.__salt__,
+        {
+            "cmd.run_all": MagicMock(
+                return_value={"retcode": 1, "stderr": "Error message"}
+            )
+        },
+    ):
+        result = acme.revoke("test_cert")
+        assert result["result"] is False
+        assert "revocation failed" in result["comment"]
+
+    # Test failure with non-zero retcode and 'expand' in stderr
+    with patch.dict(
+        acme.__salt__,
+        {
+            "cmd.run_all": MagicMock(
+                side_effect=[
+                    {"retcode": 1, "stderr": "Error message with expand"},
+                    {"retcode": 1, "stderr": "Final error message"},
+                ]
+            )
+        },
+    ):
+        result = acme.revoke("test_cert")
+        assert result["result"] is False
+        assert "Final error message" in result["comment"]
+
+    # Test with reason parameter
+    with patch.dict(
+        acme.__salt__, {"cmd.run_all": MagicMock(return_value={"retcode": 0})}
+    ):
+        assert acme.revoke("test_cert", reason="keyCompromise") is True
+
+
+def test_delete():
+    """
+    Test certificate deletion.
+    """
+    # Mocking successful deletion
+    mock_successful_delete = MagicMock(return_value={"retcode": 0})
+    with patch.dict(acme.__salt__, {"cmd.run_all": mock_successful_delete}):
+        assert acme.delete("test_cert") is True
+
+    # Mocking failed deletion with non-zero return code
+    mock_failed_delete = MagicMock(
+        return_value={"retcode": 1, "stderr": "Error message"}
+    )
+    with patch.dict(acme.__salt__, {"cmd.run_all": mock_failed_delete}):
+        result = acme.delete("test_cert")
+        assert not result["result"]
+        assert (
+            "Certificate test_cert deletion failed with:\nError message"
+            in result["comment"]
+        )
+
+    # Mocking failed deletion with 'expand' in stderr
+    mock_failed_delete_expand = MagicMock(
+        return_value={"retcode": 1, "stderr": "Error message with expand"}
+    )
+    with patch.dict(acme.__salt__, {"cmd.run_all": mock_failed_delete_expand}):
+        result = acme.delete("test_cert")
+        assert not result["result"]
+        assert (
+            "Certificate test_cert deletion failed with:\nError message with expand"
+            in result["comment"]
+        )

tests/pytests/unit/states/test_acme.py

@@ -145,3 +145,21 @@ def test_cert_renew_certificate():
         }
         assert acme.cert("test") == match
         assert acme.cert("testing.example.com", certname="test") == match
+
+
+def test_cert_replace_staging():
+    """
+    Test cert state replacing staging certificate when test_cert is False.
+    """
+    with patch.dict(
+        acme.__salt__,
+        {  # pylint: disable=no-member
+            "acme.has": MagicMock(return_value=True),
+            "acme.info": MagicMock(return_value={"issuer": {"O": "STAGING"}}),
+            "acme.revoke": MagicMock(),
+            "acme.cert": MagicMock(return_value={"result": True, "comment": "Mockery"}),
+            "acme.needs_renewal": MagicMock(return_value=False),
+        },
+    ):
+        acme.cert("test", replace_staging=True, test_cert=False)
+        acme.__salt__["acme.revoke"].assert_called_with("test")