Make security and filesystem checks return predicate results

Author: marcofortina

docs/plugins/generated.md

@@ -7927,7 +7927,7 @@ with:
 
 ### `security.authselect.check`
 
-Assert the current authselect profile and enabled features on RHEL-like systems.
+Check the current authselect profile and enabled features on RHEL-like systems.
 
 - Remote session: `true`
 - Dry-run support: `true`
@@ -8511,7 +8511,7 @@ with:
 
 ### `security.pki.cert.expiry_check`
 
-Assert that a certificate remains valid for at least min_days.
+Check whether a certificate remains valid for at least min_days.
 
 - Remote session: `true`
 - Dry-run support: `true`
@@ -8645,7 +8645,7 @@ with:
 
 ### `security.pki.cert.issuer_check`
 
-Assert that a certificate issuer contains an expected string.
+Check whether a certificate issuer contains an expected string.
 
 - Remote session: `true`
 - Dry-run support: `true`
@@ -8677,7 +8677,7 @@ with:
 
 ### `security.pki.cert.key_match_check`
 
-Assert that a certificate public key matches a private key.
+Check whether a certificate public key matches a private key.
 
 - Remote session: `true`
 - Dry-run support: `true`
@@ -8709,7 +8709,7 @@ with:
 
 ### `security.pki.cert.san_check`
 
-Assert that a certificate contains required Subject Alternative Names.
+Check whether a certificate contains required Subject Alternative Names.
 
 - Remote session: `true`
 - Dry-run support: `true`
@@ -8779,7 +8779,7 @@ with:
 
 ### `security.pki.cert.subject_check`
 
-Assert that a certificate subject contains an expected string.
+Check whether a certificate subject contains an expected string.
 
 - Remote session: `true`
 - Dry-run support: `true`
@@ -8959,7 +8959,7 @@ with:
 
 ### `security.secret.redact_check`
 
-Assert that a payload contains no declared secret values after redaction policy is applied.
+Check whether a payload contains no declared secret values after redaction policy is applied.
 
 - Remote session: `false`
 - Dry-run support: `true`

src/automax/plugins/cert_ops.py

@@ -9,7 +9,7 @@
 
 from automax.core.models import ExecutionContext, PluginResult
 from automax.plugins.base import BasePlugin
-from automax.plugins.remote_utils import CHANGE_MARKER, exec_remote, quote, result_from_remote, sudo_prefix
+from automax.plugins.remote_utils import CHANGE_MARKER, exec_remote, predicate_result_from_remote, quote, result_from_remote, sudo_prefix
 
 
 
@@ -70,9 +70,15 @@ def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) ->
 
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         rc, out, err = exec_remote(context, self.manual_commands(params, context)[0])
-        if rc != 0:
-            return PluginResult.failure(rc=rc, stdout=out, stderr=err, message="security.pki.cert.chain_check failed")
-        return PluginResult.success(changed=False, rc=rc, stdout=out, stderr=err)
+        return predicate_result_from_remote(
+            rc=rc,
+            stdout=out,
+            stderr=err,
+            message="security.pki.cert.chain_check failed",
+            data_key="valid",
+            data={"cert": str(params["cert"]), "ca_file": str(params["ca_file"])},
+            false_rcs=(1, 2),
+        )
 
 class CertInstallKeypairPlugin(BasePlugin):
     name = "security.pki.cert.install_keypair"
@@ -157,14 +163,14 @@ def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginRe
 
 class CertMatchesKeyPlugin(BasePlugin):
     name = "security.pki.cert.key_match_check"
-    description = "Assert that a certificate public key matches a private key."
+    description = "Check whether a certificate public key matches a private key."
     required_params = ("cert", "key")
     optional_params = ("sudo",)
     opens_remote_session = True
     supports_check_mode = True
 
     def diff_preview_reason(self, params: Dict[str, Any], context: ExecutionContext) -> str:
-        return "security.pki.cert.key_match_check is a read-only certificate/private-key assertion"
+        return "security.pki.cert.key_match_check is a read-only certificate/private-key predicate"
 
     def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) -> list[str]:
         sudo = sudo_prefix(params, default=True)
@@ -174,21 +180,27 @@ def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) ->
 
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         rc, out, err = exec_remote(context, self.manual_commands(params, context)[0])
-        if rc != 0:
-            return PluginResult.failure(rc=rc, stdout=out, stderr=err, message="security.pki.cert.key_match_check failed")
-        return PluginResult.success(changed=False, rc=rc, stdout=out, stderr=err)
+        return predicate_result_from_remote(
+            rc=rc,
+            stdout=out,
+            stderr=err,
+            message="security.pki.cert.key_match_check failed",
+            data_key="matches",
+            data={"cert": str(params["cert"]), "key": str(params["key"])},
+            false_rcs=(1, 2),
+        )
 
 
 class CertSanAssertPlugin(BasePlugin):
     name = "security.pki.cert.san_check"
-    description = "Assert that a certificate contains required Subject Alternative Names."
+    description = "Check whether a certificate contains required Subject Alternative Names."
     required_params = ("cert", "names")
     optional_params = ("sudo",)
     opens_remote_session = True
     supports_check_mode = True
 
     def diff_preview_reason(self, params: Dict[str, Any], context: ExecutionContext) -> str:
-        return "security.pki.cert.san_check is a read-only certificate SAN assertion"
+        return "security.pki.cert.san_check is a read-only certificate SAN predicate"
 
     def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) -> list[str]:
         self.validate(params)
@@ -201,51 +213,69 @@ def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) ->
 
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         rc, out, err = exec_remote(context, " && ".join(self.manual_commands(params, context)))
-        if rc != 0:
-            return PluginResult.failure(rc=rc, stdout=out, stderr=err, message="security.pki.cert.san_check failed")
-        return PluginResult.success(changed=False, rc=rc, stdout=out, stderr=err)
+        return predicate_result_from_remote(
+            rc=rc,
+            stdout=out,
+            stderr=err,
+            message="security.pki.cert.san_check failed",
+            data_key="matches",
+            data={"cert": str(params["cert"])},
+            false_rcs=(1, 2),
+        )
 
 
 class CertSubjectAssertPlugin(BasePlugin):
     name = "security.pki.cert.subject_check"
-    description = "Assert that a certificate subject contains an expected string."
+    description = "Check whether a certificate subject contains an expected string."
     required_params = ("cert", "subject")
     optional_params = ("sudo",)
     opens_remote_session = True
     supports_check_mode = True
 
     def diff_preview_reason(self, params: Dict[str, Any], context: ExecutionContext) -> str:
-        return "security.pki.cert.subject_check is a read-only certificate subject assertion"
+        return "security.pki.cert.subject_check is a read-only certificate subject predicate"
 
     def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) -> list[str]:
         return [f"{sudo_prefix(params, default=True)}openssl x509 -in {quote(params['cert'])} -noout -subject | grep -F -- {quote(params['subject'])}"]
 
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         rc, out, err = exec_remote(context, self.manual_commands(params, context)[0])
-        if rc != 0:
-            return PluginResult.failure(rc=rc, stdout=out, stderr=err, message="security.pki.cert.subject_check failed")
-        return PluginResult.success(changed=False, rc=rc, stdout=out, stderr=err)
+        return predicate_result_from_remote(
+            rc=rc,
+            stdout=out,
+            stderr=err,
+            message="security.pki.cert.subject_check failed",
+            data_key="matches",
+            data={"cert": str(params["cert"]), "subject": str(params["subject"])},
+            false_rcs=(1, 2),
+        )
 
 
 class CertIssuerAssertPlugin(BasePlugin):
     name = "security.pki.cert.issuer_check"
-    description = "Assert that a certificate issuer contains an expected string."
+    description = "Check whether a certificate issuer contains an expected string."
     required_params = ("cert", "issuer")
     optional_params = ("sudo",)
     opens_remote_session = True
     supports_check_mode = True
 
     def diff_preview_reason(self, params: Dict[str, Any], context: ExecutionContext) -> str:
-        return "security.pki.cert.issuer_check is a read-only certificate issuer assertion"
+        return "security.pki.cert.issuer_check is a read-only certificate issuer predicate"
 
     def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) -> list[str]:
         return [f"{sudo_prefix(params, default=True)}openssl x509 -in {quote(params['cert'])} -noout -issuer | grep -F -- {quote(params['issuer'])}"]
 
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         rc, out, err = exec_remote(context, self.manual_commands(params, context)[0])
-        if rc != 0:
-            return PluginResult.failure(rc=rc, stdout=out, stderr=err, message="security.pki.cert.issuer_check failed")
-        return PluginResult.success(changed=False, rc=rc, stdout=out, stderr=err)
+        return predicate_result_from_remote(
+            rc=rc,
+            stdout=out,
+            stderr=err,
+            message="security.pki.cert.issuer_check failed",
+            data_key="matches",
+            data={"cert": str(params["cert"]), "issuer": str(params["issuer"])},
+            false_rcs=(1, 2),
+        )
 
 
 class CertInstallCaBundlePlugin(BasePlugin):

src/automax/plugins/fs_system.py

@@ -130,9 +130,14 @@ def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) ->
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         command = " && ".join(self.manual_commands(params, context))
         rc, out, err = exec_remote(context, command)
-        if rc != 0:
-            return PluginResult.failure(rc=rc, stdout=out, stderr=err, message="fs.attr.check failed")
-        return PluginResult.success(changed=False, rc=rc, stdout=out, stderr=err, data={"path": params["path"], "attrs": _attr_entries(params["attrs"])})
+        return predicate_result_from_remote(
+            rc=rc,
+            stdout=out,
+            stderr=err,
+            message="fs.attr.check failed",
+            data_key="matches",
+            data={"path": params["path"], "attrs": _attr_entries(params["attrs"])},
+        )
 
 
 class FsQuotaPlugin(BasePlugin):
@@ -219,9 +224,14 @@ def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) ->
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         command = " && ".join(self.manual_commands(params, context))
         rc, out, err = exec_remote(context, command)
-        if rc != 0:
-            return PluginResult.failure(rc=rc, stdout=out, stderr=err, message="fs.acl.check failed")
-        return PluginResult.success(changed=False, rc=rc, stdout=out, stderr=err, data={"path": params["path"], "acl": _acl_entries(params["acl"])})
+        return predicate_result_from_remote(
+            rc=rc,
+            stdout=out,
+            stderr=err,
+            message="fs.acl.check failed",
+            data_key="matches",
+            data={"path": params["path"], "acl": _acl_entries(params["acl"])},
+        )
 
 
 class FsAclRestorePlugin(BasePlugin):

src/automax/plugins/pam_ops.py

@@ -10,7 +10,7 @@
 
 from automax.core.models import ExecutionContext, PluginResult
 from automax.plugins.base import BasePlugin, PluginValidationError
-from automax.plugins.remote_utils import cleanup_trap_command, CHANGE_MARKER, exec_remote, heredoc_to_file_expr, heredoc_to_stdin, shell_var_ref, tempfile_command, quote, result_from_remote, sudo_prefix
+from automax.plugins.remote_utils import cleanup_trap_command, CHANGE_MARKER, exec_remote, heredoc_to_file_expr, heredoc_to_stdin, predicate_result_from_remote, shell_var_ref, tempfile_command, quote, result_from_remote, sudo_prefix
 
 
 
@@ -362,7 +362,7 @@ def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginRe
 
 class PamAuthselectPlugin(BasePlugin):
     name = "security.authselect.check"
-    description = "Assert the current authselect profile and enabled features on RHEL-like systems."
+    description = "Check the current authselect profile and enabled features on RHEL-like systems."
     optional_params = ("profile", "features", "sudo")
     opens_remote_session = True
     supports_check_mode = True
@@ -371,7 +371,7 @@ def _features(self, params: Dict[str, Any]) -> list[str]:
         return _as_list(params.get("features"))
 
     def diff_preview_reason(self, params: Dict[str, Any], context: ExecutionContext) -> str:
-        return "security.authselect.check is a read-only authselect profile assertion"
+        return "security.authselect.check is a read-only authselect profile predicate"
 
     def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) -> list[str]:
         command = f"{sudo_prefix(params, default=True)}authselect current"
@@ -383,6 +383,11 @@ def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) ->
 
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         rc, out, err = exec_remote(context, self.manual_commands(params, context)[0])
-        if rc != 0:
-            return PluginResult.failure(rc=rc, stdout=out, stderr=err, message="security.authselect.check failed")
-        return PluginResult.success(changed=False, rc=rc, stdout=out, stderr=err, data={"authselect": out})
+        return predicate_result_from_remote(
+            rc=rc,
+            stdout=out,
+            stderr=err,
+            message="security.authselect.check failed",
+            data_key="matches",
+            data={"authselect": out},
+        )

src/automax/plugins/pki.py

@@ -11,7 +11,7 @@
 
 from automax.core.models import ExecutionContext, PluginResult
 from automax.plugins.base import BasePlugin, PluginValidationError
-from automax.plugins.remote_utils import cleanup_trap_command, CHANGE_MARKER, exec_remote, heredoc_to_file_expr, shell_var_ref, tempfile_command, quote, result_from_remote, sudo_prefix
+from automax.plugins.remote_utils import cleanup_trap_command, CHANGE_MARKER, exec_remote, heredoc_to_file_expr, predicate_result_from_remote, shell_var_ref, tempfile_command, quote, result_from_remote, sudo_prefix
 
 
 
@@ -118,13 +118,13 @@ def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginRe
 
 class PkiCertExpiryAssertPlugin(BasePlugin):
     name = "security.pki.cert.expiry_check"
-    description = "Assert that a certificate remains valid for at least min_days."
+    description = "Check whether a certificate remains valid for at least min_days."
     required_params = ("path",)
     optional_params = ("min_days", "sudo")
     opens_remote_session = True
 
     def diff_preview_reason(self, params: Dict[str, Any], context: ExecutionContext) -> str:
-        return "security.pki.cert.expiry_check is a read-only certificate assertion with no file diff"
+        return "security.pki.cert.expiry_check is a read-only certificate expiry predicate with no file diff"
 
     def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) -> list[str]:
         self.validate(params)
@@ -133,6 +133,12 @@ def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) ->
 
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         rc, out, err = exec_remote(context, self.manual_commands(params, context)[0])
-        if rc != 0:
-            return PluginResult.failure(rc=rc, stdout=out, stderr=err, message="certificate expiry assertion failed")
-        return PluginResult.success(stdout=out, stderr=err, data={"path": params["path"], "min_days": int(params.get("min_days", 30))})
+        return predicate_result_from_remote(
+            rc=rc,
+            stdout=out,
+            stderr=err,
+            message="security.pki.cert.expiry_check failed",
+            data_key="valid",
+            data={"path": params["path"], "min_days": int(params.get("min_days", 30))},
+            false_rcs=(1, 2),
+        )

src/automax/plugins/redaction.py

@@ -1,7 +1,7 @@
 # Copyright (C) 2026 Marco Fortina
 # SPDX-License-Identifier: AGPL-3.0-or-later
 
-"""Secret redaction policy assertion plugins."""
+"""Secret redaction policy check plugins."""
 
 from __future__ import annotations
 
@@ -24,7 +24,7 @@ def _payload(params: Dict[str, Any], context: ExecutionContext) -> str:
 
 class SecretRedactAssertPlugin(BasePlugin):
     name = "security.secret.redact_check"
-    description = "Assert that a payload contains no declared secret values after redaction policy is applied."
+    description = "Check whether a payload contains no declared secret values after redaction policy is applied."
     optional_params = ("text", "value", "source")
     supports_check_mode = True
 
@@ -38,14 +38,19 @@ def manual_commands(self, params: Dict[str, Any], context: ExecutionContext) ->
         return ["# security.secret.redact_check is evaluated inside Automax before output persistence"]
 
     def diff_preview_reason(self, params: Dict[str, Any], context: ExecutionContext) -> str:
-        return "security.secret.redact_check is an in-process redaction policy assertion"
+        return "security.secret.redact_check is an in-process redaction policy predicate"
 
     def execute(self, params: Dict[str, Any], context: ExecutionContext) -> PluginResult:
         text = _payload(params, context)
         leaked = find_unredacted_secrets(text, context.secrets)
-        if leaked:
-            return PluginResult.failure(message="unredacted secret values detected", data={"leaks": len(leaked)})
-        return PluginResult.success(changed=False, data={"redacted": redact_text(text, context.secrets)})
+        return PluginResult.success(
+            changed=False,
+            data={
+                "clean": not leaked,
+                "leaks": len(leaked),
+                "redacted": redact_text(text, context.secrets),
+            },
+        )
 
 
 class SecretScanOutputPlugin(BasePlugin):