diff --git a/prowler/lib/outputs/ocsf/ocsf.py b/prowler/lib/outputs/ocsf/ocsf.py index 6d271d9bce..1c679d117f 100644 --- a/prowler/lib/outputs/ocsf/ocsf.py +++ b/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 diff --git a/tests/lib/outputs/ocsf/ocsf_test.py b/tests/lib/outputs/ocsf/ocsf_test.py index 359e60957c..ecd8d345f6 100644 --- a/tests/lib/outputs/ocsf/ocsf_test.py +++ b/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()