fix(outputs): redact detected secrets from finding resource metadata

Secret-scanning checks embed the scanned resource verbatim in their finding,
and that resource is serialized into every output (OCSF, CSV, JSON) and any
uploaded findings, so a plaintext secret was written out even though the finding
message only names the offending field.

Add a central redactor in the output pipeline: for FAIL findings from a
secrets-category check, Finding.generate_output re-scans the resource metadata's
string values and masks any value flagged as a secret, preserving non-sensitive
context. This covers the whole secrets-check family in one place, so no
individual check has to sanitize its own resource. Best-effort and offline: on
scan error the metadata is left unchanged.

prowler/changelog.d/redact-secrets-in-findings-output.fixed.md

@@ -0,0 +1 @@
+Detected secrets are now redacted from the resource metadata of secret-scanning check findings, so raw credentials are no longer written to output (OCSF, CSV, JSON) or uploaded findings

prowler/lib/outputs/finding.py

@@ -17,7 +17,11 @@ from prowler.lib.logger import logger
 from prowler.lib.outputs.common import Status, fill_common_finding_data
 from prowler.lib.outputs.compliance.compliance_check import get_check_compliance
 from prowler.lib.outputs.utils import unroll_tags
-from prowler.lib.utils.utils import dict_to_lowercase, get_nested_attribute
+from prowler.lib.utils.utils import (
+    dict_to_lowercase,
+    get_nested_attribute,
+    redact_scanned_secrets,
+)
 from prowler.providers.common.provider import Provider
 from prowler.providers.github.models import GithubAppIdentityInfo, GithubIdentityInfo
 
@@ -136,7 +140,22 @@ class Finding(BaseModel):
             )
         try:
             output_data["provider"] = provider.type
-            output_data["resource_metadata"] = check_output.resource
+            resource_metadata = check_output.resource
+            # Secret-scanning checks embed the scanned resource verbatim, which
+            # would write the raw secret to every output. Redact detected secrets
+            # from the resource metadata of failing secrets-category findings,
+            # centrally, so no individual check has to sanitize its own resource.
+            try:
+                categories = (
+                    getattr(check_output.check_metadata, "Categories", []) or []
+                )
+                if check_output.status == "FAIL" and "secrets" in categories:
+                    resource_metadata = redact_scanned_secrets(resource_metadata)
+            except Exception as error:
+                logger.warning(
+                    f"Could not redact secrets from resource metadata: {error}"
+                )
+            output_data["resource_metadata"] = resource_metadata
 
             if provider.type == "aws":
                 output_data["account_uid"] = get_nested_attribute(

prowler/lib/utils/utils.py

@@ -1,3 +1,4 @@
+import copy
 import json
 import os
 from operator import attrgetter
@@ -362,6 +363,75 @@ def annotate_verified_secrets(report, secrets: list) -> None:
         )
 
 
+SECRET_REDACTED_PLACEHOLDER = "<REDACTED:potential-secret>"
+
+
+def _iter_secret_scan_leaves(obj, path=()):
+    """Yield ``(path, payload)`` for each scannable string leaf of ``obj``.
+
+    ``path`` is the tuple of dict keys / list indices locating the leaf. Each
+    dict value is emitted as a ``{"key": value}`` JSON payload so keyword-based
+    rules (Generic Password, etc.) fire exactly as they do in the secret checks,
+    which scan the same ``{key: value}`` shape; list items and their nested
+    values are scanned as bare strings.
+    """
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            if isinstance(value, (dict, list)):
+                yield from _iter_secret_scan_leaves(value, path + (key,))
+            elif isinstance(value, str) and value:
+                yield path + (key,), json.dumps({str(key): value})
+    elif isinstance(obj, list):
+        for index, value in enumerate(obj):
+            if isinstance(value, (dict, list)):
+                yield from _iter_secret_scan_leaves(value, path + (index,))
+            elif isinstance(value, str) and value:
+                yield path + (index,), json.dumps(value)
+
+
+def _set_at_path(obj, path, value):
+    """Set ``value`` at the ``(key/index, ...)`` ``path`` inside ``obj``."""
+    for key in path[:-1]:
+        obj = obj[key]
+    obj[path[-1]] = value
+
+
+def redact_scanned_secrets(resource_metadata):
+    """Return a copy of ``resource_metadata`` with detected secrets masked.
+
+    Secret-scanning checks embed the scanned resource verbatim in their finding,
+    and that resource is serialized into every output (OCSF, CSV, JSON) and any
+    uploaded findings, so a plaintext secret would be written out even though the
+    finding message only names the offending field. This re-scans the resource's
+    string values and replaces any value flagged as a secret with a placeholder,
+    centralizing redaction for the whole secrets-check family in the output path
+    instead of requiring each check to sanitize its own resource.
+
+    The scan runs fully offline (no live validation, no exclude patterns: masking
+    an excluded token in the output is harmless). It is best-effort: on any scan
+    error the metadata is returned unchanged, since the finding is already
+    reported as containing a secret.
+    """
+    if not isinstance(resource_metadata, (dict, list)):
+        return resource_metadata
+    redacted = copy.deepcopy(resource_metadata)
+    leaves = list(_iter_secret_scan_leaves(redacted))
+    if not leaves:
+        return redacted
+    try:
+        batch_results = detect_secrets_scan_batch(
+            ((path, payload) for path, payload in leaves),
+            excluded_secrets=[],
+            validate=False,
+        )
+    except SecretsScanError:
+        return redacted
+    for path, _ in leaves:
+        if batch_results.get(path):
+            _set_at_path(redacted, path, SECRET_REDACTED_PLACEHOLDER)
+    return redacted
+
+
 def validate_ip_address(ip_string):
     """validate_ip_address return True if the IP is valid, otherwise returns False."""
     try:

tests/lib/outputs/finding_test.py

@@ -1,3 +1,4 @@
+import json
 from datetime import datetime
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
@@ -245,6 +246,92 @@ class TestFinding:
         assert finding_output.service_name == "service"
         assert finding_output.raw == {}
 
+    def _secrets_check_output(self, provider, status, categories):
+        check_metadata = mock_check_metadata(provider="aws")
+        check_metadata.Categories = categories
+        check_output = MagicMock()
+        check_output.resource_id = "jdbc-secret"
+        check_output.resource_arn = "test_resource_arn"
+        check_output.resource_details = ""
+        check_output.resource_tags = {}
+        check_output.region = "us-east-1"
+        check_output.partition = "aws"
+        check_output.status = status
+        check_output.status_extended = "mock_status_extended"
+        check_output.muted = False
+        check_output.check_metadata = check_metadata
+        check_output.resource = {
+            "name": "jdbc-secret",
+            "properties": {
+                "JDBC_CONNECTION_URL": "jdbc:mysql://db.example.com:3306/app",
+                "PASSWORD": "AKIAsupersecretkey1234",
+            },
+        }
+        check_output.compliance = {}
+        return check_output
+
+    def test_generate_output_redacts_secret_in_failed_secrets_finding(self):
+        provider = MagicMock()
+        provider.type = "aws"
+        provider.identity.profile = "mock_auth"
+        provider.identity.account = "mock_account_uid"
+        provider.identity.partition = "aws"
+        provider.organizations_metadata.account_name = "mock_account_name"
+        provider.organizations_metadata.account_email = "mock_account_email"
+        provider.organizations_metadata.organization_arn = "mock_account_org_uid"
+        provider.organizations_metadata.organization_id = "mock_account_org_name"
+        provider.organizations_metadata.account_tags = {"tag1": "value1"}
+        provider.organizations_metadata.account_ou_id = "ou-test-12345678"
+        provider.organizations_metadata.account_ou_name = "TestOU/SubOU"
+        output_options = MagicMock()
+        output_options.unix_timestamp = False
+
+        check_output = self._secrets_check_output(provider, Status.FAIL, ["secrets"])
+
+        finding_output = Finding.generate_output(provider, check_output, output_options)
+
+        # The flagged secret value is masked in the serialized resource metadata.
+        assert (
+            finding_output.resource_metadata["properties"]["PASSWORD"]
+            == "<REDACTED:potential-secret>"
+        )
+        # Non-secret context is preserved.
+        assert (
+            finding_output.resource_metadata["properties"]["JDBC_CONNECTION_URL"]
+            == "jdbc:mysql://db.example.com:3306/app"
+        )
+        # The raw secret is nowhere in the finding's resource metadata.
+        assert "AKIAsupersecretkey1234" not in json.dumps(
+            finding_output.resource_metadata
+        )
+
+    def test_generate_output_does_not_redact_non_secrets_finding(self):
+        # A FAIL from a non-secrets check is not re-scanned, so its metadata is
+        # left untouched (redaction is gated to the secrets-check family).
+        provider = MagicMock()
+        provider.type = "aws"
+        provider.identity.profile = "mock_auth"
+        provider.identity.account = "mock_account_uid"
+        provider.identity.partition = "aws"
+        provider.organizations_metadata.account_name = "mock_account_name"
+        provider.organizations_metadata.account_email = "mock_account_email"
+        provider.organizations_metadata.organization_arn = "mock_account_org_uid"
+        provider.organizations_metadata.organization_id = "mock_account_org_name"
+        provider.organizations_metadata.account_tags = {"tag1": "value1"}
+        provider.organizations_metadata.account_ou_id = "ou-test-12345678"
+        provider.organizations_metadata.account_ou_name = "TestOU/SubOU"
+        output_options = MagicMock()
+        output_options.unix_timestamp = False
+
+        check_output = self._secrets_check_output(provider, Status.FAIL, [])
+
+        finding_output = Finding.generate_output(provider, check_output, output_options)
+
+        assert (
+            finding_output.resource_metadata["properties"]["PASSWORD"]
+            == "AKIAsupersecretkey1234"
+        )
+
     def test_generate_output_aws_without_organizations_metadata(self):
         # Simulates running without --organizations-role
         provider = MagicMock()

tests/lib/utils/utils_test.py

@@ -1,3 +1,4 @@
+import json
 import os
 import subprocess
 import tempfile
@@ -8,6 +9,7 @@ import pytest
 from mock import patch
 
 from prowler.lib.utils.utils import (
+    SECRET_REDACTED_PLACEHOLDER,
     SecretsScanError,
     detect_secrets_scan_batch,
     file_exists,
@@ -17,11 +19,54 @@ from prowler.lib.utils.utils import (
     open_file,
     outputs_unix_timestamp,
     parse_json_file,
+    redact_scanned_secrets,
     strip_ansi_codes,
     validate_ip_address,
 )
 
 
+class Test_redact_scanned_secrets:
+    def test_masks_flagged_values_preserves_context(self):
+        resource = {
+            "name": "c1",
+            "properties": {
+                "JDBC_CONNECTION_URL": "jdbc:mysql://db.example.com:3306/app",
+                "USERNAME": "app_user",
+                "PASSWORD": "AKIAsupersecretkey1234",
+            },
+        }
+
+        out = redact_scanned_secrets(resource)
+
+        # The flagged secret value is masked...
+        assert out["properties"]["PASSWORD"] == SECRET_REDACTED_PLACEHOLDER
+        # ...while non-secret context is preserved.
+        assert out["properties"]["USERNAME"] == "app_user"
+        assert (
+            out["properties"]["JDBC_CONNECTION_URL"]
+            == "jdbc:mysql://db.example.com:3306/app"
+        )
+        # The raw secret is absent from the serialized output.
+        assert "AKIAsupersecretkey1234" not in json.dumps(out)
+        # The input is never mutated (only a copy is redacted).
+        assert resource["properties"]["PASSWORD"] == "AKIAsupersecretkey1234"
+
+    def test_non_mapping_input_returned_as_is(self):
+        assert redact_scanned_secrets("plain string") == "plain string"
+        assert redact_scanned_secrets(None) is None
+
+    def test_scan_error_returns_metadata_unchanged(self):
+        resource = {"properties": {"PASSWORD": "AKIAsupersecretkey1234"}}
+
+        with patch(
+            "prowler.lib.utils.utils.detect_secrets_scan_batch",
+            side_effect=SecretsScanError("scanner failed"),
+        ):
+            out = redact_scanned_secrets(resource)
+
+        assert out == resource
+
+
 def _fake_kingfisher_run(output_content=None, returncode=0, stderr=""):
     """Build a ``subprocess.run`` replacement that mimics a Kingfisher call.