fix(ocsf): avoid missing MITRE catalog errors (#12222)

prowler/lib/outputs/ocsf/ocsf.py

@@ -2,6 +2,7 @@ import json
 import os
 from datetime import datetime, timezone
 from functools import lru_cache
+from importlib import resources
 from random import getrandbits
 from typing import Dict, List, Optional
 
@@ -350,16 +351,18 @@ def _build_analytic(finding: Finding) -> Analytic:
 def _load_mitre_technique_map(provider: str) -> Dict[str, dict]:
     """Load and cache MITRE ATT&CK techniques for a provider."""
     try:
-        mitre_file = os.path.join(
-            os.path.dirname(__file__),
-            "..",
-            "..",
-            "..",
-            "compliance",
-            provider,
-            f"mitre_attack_{provider}.json",
+        mitre_file = (
+            resources.files("prowler.compliance")
+            .joinpath(provider)
+            .joinpath(f"mitre_attack_{provider}.json")
         )
-        with open(mitre_file) as file:
+        if not mitre_file.is_file():
+            logger.debug(
+                f"MITRE ATT&CK catalog is not available for provider {provider}"
+            )
+            return {}
+
+        with mitre_file.open(encoding="utf-8") as file:
             data = json.load(file)
         return {
             requirement["Id"]: requirement
@@ -382,9 +385,13 @@ def _build_mitre_attacks(finding: Finding) -> Optional[List[MITREAttack]]:
         Optional[List[MITREAttack]]: MITRE attacks for known provider techniques,
             or None when none can be built.
     """
+    technique_ids = finding.compliance.get("MITRE-ATTACK", [])
+    if not technique_ids:
+        return None
+
     technique_map = _load_mitre_technique_map(finding.provider)
     attacks = []
-    for technique_id in finding.compliance.get("MITRE-ATTACK", []):
+    for technique_id in technique_ids:
         requirement = technique_map.get(technique_id)
         if not requirement:
             continue

tests/lib/outputs/ocsf/ocsf_test.py

@@ -239,19 +239,76 @@ class TestOCSF:
             ocsf = OCSF([finding])
         assert ocsf.data[0].finding_info.attacks is None
 
-    def test_load_mitre_technique_map_logs_failure(self):
+    def test_transform_without_mitre_ids_does_not_load_catalog(self):
+        finding = generate_finding_output(provider="kubernetes")
+
+        with patch(
+            "prowler.lib.outputs.ocsf.ocsf._load_mitre_technique_map"
+        ) as mock_load_catalog:
+            ocsf = OCSF([finding])
+
+        assert ocsf.data[0].finding_info.attacks is None
+        mock_load_catalog.assert_not_called()
+
+    def test_load_mitre_technique_map_missing_catalog_is_expected(self):
         from prowler.lib.outputs.ocsf.ocsf import _load_mitre_technique_map
 
         _load_mitre_technique_map.cache_clear()
         try:
             with (
-                patch("builtins.open", side_effect=OSError("catalog unavailable")),
+                patch("prowler.lib.outputs.ocsf.ocsf.logger.debug") as mock_debug,
+                patch("prowler.lib.outputs.ocsf.ocsf.logger.error") as mock_error,
+            ):
+                assert _load_mitre_technique_map("unsupported") == {}
+
+            mock_debug.assert_called_once_with(
+                "MITRE ATT&CK catalog is not available for provider unsupported"
+            )
+            mock_error.assert_not_called()
+        finally:
+            _load_mitre_technique_map.cache_clear()
+
+    def test_load_mitre_technique_map_existing_catalog(self):
+        from prowler.lib.outputs.ocsf.ocsf import _load_mitre_technique_map
+
+        _load_mitre_technique_map.cache_clear()
+        try:
+            technique_map = _load_mitre_technique_map("aws")
+
+            assert technique_map
+            assert all(
+                technique_id == requirement["Id"]
+                for technique_id, requirement in technique_map.items()
+            )
+        finally:
+            _load_mitre_technique_map.cache_clear()
+
+    def test_load_mitre_technique_map_logs_failure(self):
+        from prowler.lib.outputs.ocsf.ocsf import _load_mitre_technique_map
+
+        mitre_file = MagicMock()
+        mitre_file.is_file.return_value = True
+        mitre_file.open.side_effect = OSError("catalog unavailable")
+        provider_directory = MagicMock()
+        provider_directory.joinpath.return_value = mitre_file
+        compliance_package = MagicMock()
+        compliance_package.joinpath.return_value = provider_directory
+
+        _load_mitre_technique_map.cache_clear()
+        try:
+            with (
+                patch(
+                    "prowler.lib.outputs.ocsf.ocsf.resources.files",
+                    return_value=compliance_package,
+                ),
                 patch("prowler.lib.outputs.ocsf.ocsf.logger.error") as mock_error,
             ):
                 assert _load_mitre_technique_map("aws") == {}
 
             message = mock_error.call_args.args[0]
             assert re.fullmatch(r"OSError\[\d+\]: catalog unavailable", message)
+            compliance_package.joinpath.assert_called_once_with("aws")
+            provider_directory.joinpath.assert_called_once_with("mitre_attack_aws.json")
         finally:
             _load_mitre_technique_map.cache_clear()