diff --git a/prowler/changelog.d/redact-secrets-in-findings-output.fixed.md b/prowler/changelog.d/redact-secrets-in-findings-output.fixed.md new file mode 100644 index 0000000000..a4687383bd --- /dev/null +++ b/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 diff --git a/prowler/lib/outputs/finding.py b/prowler/lib/outputs/finding.py index dc88044692..93bc126429 100644 --- a/prowler/lib/outputs/finding.py +++ b/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( diff --git a/prowler/lib/utils/utils.py b/prowler/lib/utils/utils.py index fdec495c25..b2d38dd1e1 100644 --- a/prowler/lib/utils/utils.py +++ b/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 = "" + + +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: diff --git a/tests/lib/outputs/finding_test.py b/tests/lib/outputs/finding_test.py index 1315970201..adc8d0e6dd 100644 --- a/tests/lib/outputs/finding_test.py +++ b/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"] + == "" + ) + # 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() diff --git a/tests/lib/utils/utils_test.py b/tests/lib/utils/utils_test.py index 342f5d4bd5..c3bcb00ad7 100644 --- a/tests/lib/utils/utils_test.py +++ b/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.