fix(outputs): parse resource metadata once per batch (#13149)

api/changelog.d/report-metadata-memory.fixed.md

@@ -0,0 +1 @@
+Report generation shares the parsed resource metadata between the findings of a batch and releases each batch before building the next one

api/src/backend/tasks/tasks.py

@@ -897,8 +897,12 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
     )
     with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
         for batch, is_last in batched(qs, DJANGO_FINDINGS_BATCH_SIZE):
+            metadata_cache = {}
             fos = [
-                FindingOutput.transform_api_finding(f, prowler_provider) for f in batch
+                FindingOutput.transform_api_finding(
+                    f, prowler_provider, metadata_cache=metadata_cache
+                )
+                for f in batch
             ]
 
             # Outputs
@@ -979,6 +983,8 @@ def generate_outputs_task(scan_id: str, provider_id: str, tenant_id: str):
                 writer.batch_write_data_to_file()
                 writer._data.clear()
 
+            del fos, metadata_cache
+
     compressed = _compress_output_files(out_dir)
 
     upload_uri = _upload_to_s3(

api/src/backend/tasks/tests/test_tasks.py

@@ -742,7 +742,7 @@ class TestGenerateOutputs:
             patch("tasks.tasks.FindingOutput._transform_findings_stats"),
             patch(
                 "tasks.tasks.FindingOutput.transform_api_finding",
-                side_effect=lambda f, _prov: f,
+                side_effect=lambda f, _prov, **_kw: f,
             ),
             patch("tasks.tasks._compress_output_files", return_value="outdir.zip"),
             patch("tasks.tasks._upload_to_s3", return_value="s3://bucket/outdir.zip"),

docs/developer-guide/provider.mdx

@@ -3251,7 +3251,7 @@ If your provider has specific fields, add them to the finding transformation:
 **File:** `prowler/lib/outputs/finding.py`
 
 ```python
-def transform_api_finding(cls, finding, provider) -> "Finding":
+def transform_api_finding(cls, finding, provider, metadata_cache=None) -> "Finding":
     # ... existing code ...
 
     # Your provider specific field

prowler/changelog.d/report-metadata-memory.fixed.md

@@ -0,0 +1 @@
+Scan report generation parses each resource's metadata once per batch and sanitizes it once per finding, so providers with very large resource metadata no longer exhaust the worker memory

prowler/lib/outputs/compliance/universal/ocsf_compliance.py

@@ -25,7 +25,7 @@ from prowler.lib.check.compliance_config_eval import (
 )
 from prowler.lib.check.compliance_models import ComplianceFramework
 from prowler.lib.logger import logger
-from prowler.lib.outputs.utils import unroll_dict_to_list
+from prowler.lib.outputs.utils import is_plain_json, unroll_dict_to_list
 from prowler.lib.utils.utils import open_file
 
 if TYPE_CHECKING:
@@ -58,6 +58,12 @@ def _sanitize_resource_data(resource_details, resource_metadata) -> dict:
             return [_make_serializable(v) for v in obj]
         return obj
 
+    if is_plain_json(resource_metadata):
+        return {
+            "details": resource_details,
+            "metadata": resource_metadata,
+        }
+
     try:
         converted = _make_serializable(resource_metadata)
         sanitized_metadata = json.loads(json.dumps(converted, default=str))
@@ -193,6 +199,15 @@ class OCSFComplianceOutput:
 
         for finding in findings:
             if finding.check_id in check_req_map:
+                try:
+                    resource_data = _sanitize_resource_data(
+                        finding.resource_details, finding.resource_metadata
+                    )
+                except Exception as error:
+                    logger.error(
+                        f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                    )
+                    continue
                 for req in check_req_map[finding.check_id]:
                     cf = self._build_compliance_finding(
                         finding,
@@ -200,6 +215,7 @@ class OCSFComplianceOutput:
                         req,
                         compliance_name,
                         requirement_config_status.get(req.id, (True, "")),
+                        resource_data=resource_data,
                     )
                     if cf:
                         self._data.append(cf)
@@ -255,8 +271,13 @@ class OCSFComplianceOutput:
         requirement,
         compliance_name: str,
         config_status: tuple = (True, ""),
+        resource_data: dict | None = None,
     ) -> ComplianceFinding:
         try:
+            if resource_data is None:
+                resource_data = _sanitize_resource_data(
+                    finding.resource_details, finding.resource_metadata
+                )
             effective_status, message = apply_config_status(
                 finding.status, finding.status_extended, config_status
             )
@@ -348,10 +369,7 @@ class OCSFComplianceOutput:
                             if finding.provider == "kubernetes"
                             else None
                         ),
-                        data=_sanitize_resource_data(
-                            finding.resource_details,
-                            finding.resource_metadata,
-                        ),
+                        data=resource_data,
                     )
                 ],
                 severity_id=finding_severity.value,

prowler/lib/outputs/finding.py

@@ -21,6 +21,8 @@ from prowler.lib.utils.utils import dict_to_lowercase, get_nested_attribute
 from prowler.providers.common.provider import Provider
 from prowler.providers.github.models import GithubAppIdentityInfo, GithubIdentityInfo
 
+RESOURCE_METADATA_WARNING_SIZE = 10 * 1024 * 1024
+
 
 class Finding(BaseModel):
     """
@@ -601,8 +603,28 @@ class Finding(BaseModel):
             )
             raise error
 
+    @staticmethod
+    def _load_resource_metadata(resource, metadata_cache: dict | None) -> dict:
+        """Parse the stored metadata once per resource when a cache is given."""
+        if metadata_cache is None:
+            return json.loads(resource.metadata)
+        if resource.id in metadata_cache:
+            return metadata_cache[resource.id]
+
+        metadata_size = len(resource.metadata)
+        if metadata_size > RESOURCE_METADATA_WARNING_SIZE:
+            logger.warning(
+                f"Resource {resource.id} ({resource.uid}) has {metadata_size} characters of metadata"
+            )
+
+        metadata = json.loads(resource.metadata)
+        metadata_cache[resource.id] = metadata
+        return metadata
+
     @classmethod
-    def transform_api_finding(cls, finding, provider) -> "Finding":
+    def transform_api_finding(
+        cls, finding, provider, metadata_cache: dict | None = None
+    ) -> "Finding":
         """
         Transform a FindingModel instance into an API-friendly Finding object.
 
@@ -616,6 +638,8 @@ class Finding(BaseModel):
         Args:
             finding (API Finding): An API Finding instance containing data from the database.
             provider (Provider): the provider object.
+            metadata_cache (dict | None): parsed resource metadata keyed by resource id,
+                shared by the findings of a batch so each resource is parsed once.
 
         Returns:
             Finding: A new Finding instance populated with data from the provided model.
@@ -624,7 +648,7 @@ class Finding(BaseModel):
         resource = finding.resources.first()
         finding.resource_arn = resource.uid
         finding.resource_name = resource.name
-        finding.resource = json.loads(resource.metadata)
+        finding.resource = cls._load_resource_metadata(resource, metadata_cache)
         finding.resource_details = resource.details
 
         finding.resource_id = resource.name if provider.type == "aws" else resource.uid

prowler/lib/outputs/ocsf/ocsf.py

@@ -27,7 +27,7 @@ from uuid6 import UUID
 from prowler.lib.logger import logger
 from prowler.lib.outputs.finding import Finding
 from prowler.lib.outputs.output import Output
-from prowler.lib.outputs.utils import unroll_dict_to_list
+from prowler.lib.outputs.utils import is_plain_json, unroll_dict_to_list
 
 
 class OCSF(Output):
@@ -279,6 +279,12 @@ class OCSF(Output):
                 return [_make_serializable(v) for v in obj]
             return obj
 
+        if is_plain_json(resource_metadata):
+            return {
+                "details": resource_details,
+                "metadata": resource_metadata,
+            }
+
         try:
             converted = _make_serializable(resource_metadata)
             sanitized_metadata = json.loads(json.dumps(converted, default=str))

prowler/lib/outputs/utils.py

@@ -220,3 +220,20 @@ def sanitize_csv_value(value):
     if isfinite(number) and value == value.strip():
         return value
     return f"'{value}"
+
+
+def is_plain_json(value) -> bool:
+    """Return True when value is made only of JSON types with str keys."""
+    try:
+        if value is None or isinstance(value, (str, int, float, bool)):
+            return True
+        if isinstance(value, dict):
+            return all(
+                isinstance(key, str) and is_plain_json(item)
+                for key, item in value.items()
+            )
+        if isinstance(value, list):
+            return all(is_plain_json(item) for item in value)
+        return False
+    except RecursionError:
+        return False

tests/lib/outputs/compliance/universal/ocsf_compliance_test.py

@@ -1,6 +1,7 @@
 import json
 from datetime import datetime, timezone
 from types import SimpleNamespace
+from unittest.mock import patch
 
 from py_ocsf_models.events.base_event import StatusID as EventStatusID
 from py_ocsf_models.events.findings.compliance_finding import ComplianceFinding
@@ -301,6 +302,56 @@ class TestOCSFComplianceOutput:
         assert ComplianceStatusID.Pass in statuses
         assert ComplianceStatusID.Fail in statuses
 
+    def test_resource_data_sanitized_once_per_finding(self):
+        fw = _make_framework(
+            [
+                _simple_requirement("REQ-1", ["check_a"]),
+                _simple_requirement("REQ-2", ["check_a"]),
+                _simple_requirement("REQ-3", ["check_a"]),
+            ]
+        )
+        finding = _make_finding("check_a")
+        finding.resource_metadata = {"big": ["payload"]}
+
+        with patch(
+            "prowler.lib.outputs.compliance.universal.ocsf_compliance._sanitize_resource_data",
+            wraps=_sanitize_resource_data,
+        ) as sanitize:
+            output = OCSFComplianceOutput(
+                findings=[finding], framework=fw, provider="aws"
+            )
+
+        assert len(output.data) == 3
+        sanitize.assert_called_once_with(
+            finding.resource_details, finding.resource_metadata
+        )
+        assert all(
+            cf.resources[0].data["metadata"] is finding.resource_metadata
+            for cf in output.data
+        )
+
+    def test_resource_data_error_skips_only_that_finding(self):
+        fw = _make_framework([_simple_requirement("REQ-1", ["check_a"])])
+        broken = _make_finding("check_a")
+        healthy = _make_finding("check_a")
+
+        def sanitize(details, metadata):
+            if metadata is broken.resource_metadata:
+                raise RuntimeError("boom")
+            return _sanitize_resource_data(details, metadata)
+
+        broken.resource_metadata = {"broken": True}
+        with patch(
+            "prowler.lib.outputs.compliance.universal.ocsf_compliance._sanitize_resource_data",
+            side_effect=sanitize,
+        ):
+            output = OCSFComplianceOutput(
+                findings=[broken, healthy], framework=fw, provider="aws"
+            )
+
+        assert len(output.data) == 1
+        assert output.data[0].resources[0].data["metadata"] == {}
+
     def test_requirement_attributes_in_unmapped(self):
         req = UniversalComplianceRequirement(
             id="REQ-1",
@@ -509,6 +560,11 @@ class TestSanitizeResourceData:
         result = _sanitize_resource_data("details", {"a": 1, "b": "two"})
         assert result == {"details": "details", "metadata": {"a": 1, "b": "two"}}
 
+    def test_plain_json_is_not_copied(self):
+        metadata = {"a": [1, {"b": "two"}]}
+        result = _sanitize_resource_data("details", metadata)
+        assert result["metadata"] is metadata
+
     def test_none_metadata(self):
         result = _sanitize_resource_data("details", None)
         assert result == {"details": "details", "metadata": None}

tests/lib/outputs/finding_test.py

@@ -93,7 +93,9 @@ class DummyResource:
         details=None,
         metadata=None,
         partition=None,
+        id=None,
     ):
+        self.id = id or uid
         self.uid = uid
         self.name = name
         self.resource_arn = resource_arn
@@ -138,6 +140,51 @@ class DummyAPIFinding:
     """
 
 
+def _aws_api_finding_fixture():
+    provider = DummyProvider(uid="account123")
+    provider.type = "aws"
+    provider.organizations_metadata = SimpleNamespace(
+        account_name="test-account",
+        account_email="test@example.com",
+        organization_arn="arn:aws:organizations::123456789012:organization/o-abcdef123456",
+        organization_id="o-abcdef123456",
+        account_tags={},
+    )
+    provider.identity = SimpleNamespace(
+        account="123456789012", partition="aws", profile="default"
+    )
+    api_finding = DummyAPIFinding()
+    api_finding.inserted_at = 1234567890
+    api_finding.scan = DummyScan(provider=provider)
+    api_finding.uid = "finding-uid-1"
+    api_finding.status = "FAIL"
+    api_finding.status_extended = "extended"
+    api_finding.check_metadata = {
+        "provider": "aws",
+        "checkid": "service_check_001",
+        "checktitle": "Test Check",
+        "checktype": [],
+        "servicename": "service",
+        "subservicename": "",
+        "severity": "high",
+        "resourcetype": "TestResource",
+        "description": "A test check",
+        "risk": "High risk",
+        "relatedurl": "",
+        "remediation": {
+            "recommendation": {"text": "Fix it", "url": ""},
+            "code": {"nativeiac": "", "terraform": "", "cli": "", "other": ""},
+        },
+        "resourceidtemplate": "",
+        "categories": [],
+        "dependson": [],
+        "relatedto": [],
+        "notes": "",
+    }
+    api_finding.muted = False
+    return provider, api_finding
+
+
 class TestFinding:
     def test_generate_output_aws(self):
         # Mock provider
@@ -1115,6 +1162,93 @@ class TestFinding:
             "MITRE-ATTACK": ["T1098"],
         }
 
+    @patch(
+        "prowler.lib.outputs.finding.get_check_compliance",
+        new=mock_get_check_compliance,
+    )
+    def test_transform_api_finding_shares_parsed_metadata_within_cache(self):
+        provider, api_finding = _aws_api_finding_fixture()
+        resource = DummyResource(
+            uid="res-uid-1",
+            name="ResourceName1",
+            resource_arn="arn",
+            region="us-east-1",
+            tags=[],
+            metadata='{"nested": {"items": [1, 2, 3]}}',
+        )
+        api_finding.resources = DummyResources(resource)
+        _, other_api_finding = _aws_api_finding_fixture()
+        other_api_finding.resources = DummyResources(resource)
+        metadata_cache = {}
+
+        first = Finding.transform_api_finding(
+            api_finding, provider, metadata_cache=metadata_cache
+        )
+        second = Finding.transform_api_finding(
+            other_api_finding, provider, metadata_cache=metadata_cache
+        )
+
+        assert first.resource_metadata == {"nested": {"items": [1, 2, 3]}}
+        assert second.resource_metadata is first.resource_metadata
+        assert metadata_cache == {resource.id: first.resource_metadata}
+
+    @patch(
+        "prowler.lib.outputs.finding.get_check_compliance",
+        new=mock_get_check_compliance,
+    )
+    def test_transform_api_finding_without_cache_parses_each_time(self):
+        provider, api_finding = _aws_api_finding_fixture()
+        resource = DummyResource(
+            uid="res-uid-1",
+            name="ResourceName1",
+            resource_arn="arn",
+            region="us-east-1",
+            tags=[],
+            metadata='{"a": 1}',
+        )
+        api_finding.resources = DummyResources(resource)
+        _, other_api_finding = _aws_api_finding_fixture()
+        other_api_finding.resources = DummyResources(resource)
+
+        first = Finding.transform_api_finding(api_finding, provider)
+        second = Finding.transform_api_finding(other_api_finding, provider)
+
+        assert first.resource_metadata == second.resource_metadata == {"a": 1}
+        assert first.resource_metadata is not second.resource_metadata
+
+    @patch(
+        "prowler.lib.outputs.finding.get_check_compliance",
+        new=mock_get_check_compliance,
+    )
+    @patch("prowler.lib.outputs.finding.RESOURCE_METADATA_WARNING_SIZE", 16)
+    @patch("prowler.lib.outputs.finding.logger")
+    def test_transform_api_finding_warns_once_on_large_metadata(self, logger):
+        provider, api_finding = _aws_api_finding_fixture()
+        resource = DummyResource(
+            uid="res-uid-1",
+            name="ResourceName1",
+            resource_arn="arn",
+            region="us-east-1",
+            tags=[],
+            metadata='{"payload": "' + "x" * 32 + '"}',
+        )
+        api_finding.resources = DummyResources(resource)
+        _, other_api_finding = _aws_api_finding_fixture()
+        other_api_finding.resources = DummyResources(resource)
+        metadata_cache = {}
+
+        Finding.transform_api_finding(
+            api_finding, provider, metadata_cache=metadata_cache
+        )
+        Finding.transform_api_finding(
+            other_api_finding, provider, metadata_cache=metadata_cache
+        )
+
+        logger.warning.assert_called_once()
+        message = logger.warning.call_args.args[0]
+        assert "res-uid-1" in message
+        assert str(len(resource.metadata)) in message
+
     @patch(
         "prowler.lib.outputs.finding.get_check_compliance",
         new=mock_get_check_compliance,

tests/lib/outputs/ocsf/ocsf_test.py

@@ -771,6 +771,11 @@ class TestOCSF:
             "metadata": {"key": "value"},
         }
 
+    def test_sanitize_resource_data_plain_json_is_not_copied(self):
+        metadata = {"key": ["value", {"nested": 1}]}
+        result = OCSF._sanitize_resource_data("details", metadata)
+        assert result["metadata"] is metadata
+
     def test_sanitize_resource_data_empty_dict(self):
         result = OCSF._sanitize_resource_data("details", {})
         assert result == {

tests/lib/outputs/outputs_test.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from unittest import mock
 
 import pytest
@@ -11,6 +12,7 @@ from prowler.lib.outputs.outputs import (
     set_report_color,
 )
 from prowler.lib.outputs.utils import (
+    is_plain_json,
     parse_html_string,
     parse_json_tags,
     sanitize_csv_value,
@@ -1362,3 +1364,21 @@ class TestSanitizeCSVValue:
     @pytest.mark.parametrize("value", [" ", "\n", "  text"])
     def test_whitespace_without_initiator_is_untouched(self, value):
         assert sanitize_csv_value(value) == value
+
+    def test_is_plain_json_accepts_json_types(self):
+        assert is_plain_json({"a": [1, 2.5, "x", True, None, {"b": {}}]})
+        assert is_plain_json(None)
+        assert is_plain_json([])
+
+    def test_is_plain_json_rejects_tuples(self):
+        assert not is_plain_json({"t": (1, 2)})
+
+    def test_is_plain_json_rejects_non_str_keys(self):
+        assert not is_plain_json({1: "a"})
+
+    def test_is_plain_json_rejects_objects(self):
+        class Opaque:
+            pass
+
+        assert not is_plain_json({"o": Opaque()})
+        assert not is_plain_json({"d": datetime(2024, 1, 1)})