diff --git a/prowler/lib/outputs/ocsf/ocsf.py b/prowler/lib/outputs/ocsf/ocsf.py index 29898ce813..6d271d9bce 100644 --- a/prowler/lib/outputs/ocsf/ocsf.py +++ b/prowler/lib/outputs/ocsf/ocsf.py @@ -1,8 +1,9 @@ import json import os from datetime import datetime, timezone +from functools import lru_cache from random import getrandbits -from typing import List, Optional +from typing import Dict, List, Optional from py_ocsf_models.events.base_event import SeverityID, StatusID from py_ocsf_models.events.findings.detection_finding import ( @@ -345,36 +346,68 @@ def _build_analytic(finding: Finding) -> Analytic: ) +@lru_cache(maxsize=None) +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", + ) + with open(mitre_file) as file: + data = json.load(file) + return { + requirement["Id"]: requirement + for requirement in data.get("Requirements", []) + } + except Exception as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return {} + + def _build_mitre_attacks(finding: Finding) -> Optional[List[MITREAttack]]: - """Build OCSF MITREAttack objects from MITRE-ATTACK metadata. + """Build OCSF MITREAttack objects from finding compliance technique IDs. Args: - finding (Finding): Finding with compliance metadata attached to its check metadata. + finding (Finding): Finding with MITRE ATT&CK compliance technique IDs. Returns: - Optional[List[MITREAttack]]: MITRE attacks derived from metadata, or None - when the finding has no MITRE-ATTACK metadata. + Optional[List[MITREAttack]]: MITRE attacks for known provider techniques, + or None when none can be built. """ + technique_map = _load_mitre_technique_map(finding.provider) attacks = [] - for compliance in finding.metadata.Compliance or []: - if compliance.Framework.upper() != "MITRE-ATTACK": + for technique_id in finding.compliance.get("MITRE-ATTACK", []): + requirement = technique_map.get(technique_id) + if not requirement: continue - - for requirement in compliance.Requirements: - technique = Technique( - uid=requirement.Id, - name=requirement.Name, - src_url=requirement.TechniqueURL, + technique_name = requirement.get("Name") + if not technique_name: + logger.warning( + f"Skipping MITRE ATT&CK technique {technique_id} for provider {finding.provider}: missing Name" ) - for tactic_name in requirement.Tactics: - attacks.append( - MITREAttack( - technique=technique, - tactic=Tactic(name=tactic_name), - ) + continue + technique = Technique( + uid=technique_id, + name=technique_name, + src_url=requirement.get("TechniqueURL"), + ) + for tactic_name in requirement.get("Tactics", []): + attacks.append( + MITREAttack( + technique=technique, + tactic=Tactic(name=tactic_name), ) + ) - return attacks if attacks else None + return attacks or None # NOTE: Copied from api/src/backend/api/uuid_utils.py (datetime_to_uuid7) diff --git a/tests/lib/outputs/ocsf/ocsf_test.py b/tests/lib/outputs/ocsf/ocsf_test.py index cd1ba289ea..359e60957c 100644 --- a/tests/lib/outputs/ocsf/ocsf_test.py +++ b/tests/lib/outputs/ocsf/ocsf_test.py @@ -1,4 +1,5 @@ import json +import re from datetime import datetime, timezone from io import StringIO from typing import Optional @@ -28,11 +29,6 @@ from py_ocsf_models.objects.resource_details import ResourceDetails from pydantic.v1 import BaseModel as V1BaseModel from prowler.config.config import prowler_version -from prowler.lib.check.compliance_models import ( - Compliance, - Mitre_Requirement, - Mitre_Requirement_Attribute_AWS, -) from prowler.lib.outputs.ocsf.ocsf import OCSF from tests.lib.outputs.fixtures.fixtures import generate_finding_output from tests.providers.aws.utils import AWS_REGION_EU_WEST_1 @@ -148,89 +144,150 @@ class TestOCSF: 1619600000, tz=timezone.utc ) - def test_transform_mitre_attacks_populated(self): + def test_transform_mitre_attacks_from_multiple_finding_compliance_ids(self): finding = generate_finding_output( provider="aws", - compliance={"MITRE-ATTACK": ["T4242"]}, + compliance={"MITRE-ATTACK": ["T1078", "T1098"]}, check_id="iam_user_mfa_enabled_console_access", check_title="IAM users with console access have MFA enabled", service_name="iam", ) - finding.metadata.Compliance = [ - Compliance( - Framework="MITRE-ATTACK", - Name="MITRE ATT&CK compliance framework", - Provider="AWS", - Version="", - Description="MITRE ATT&CK test framework", - Requirements=[ - Mitre_Requirement( - Name="Synthetic Valid Accounts", - Id="T4242", - Tactics=["Persistence", "Privilege Escalation"], - SubTechniques=[], - Description="Synthetic MITRE technique for OCSF tests.", - Platforms=["IaaS"], - TechniqueURL="https://attack.mitre.org/techniques/T4242/", - Attributes=[ - Mitre_Requirement_Attribute_AWS( - AWSService="AWS IAM", - Category="Protect", - Value="Significant", - Comment="Test mapping", - ) - ], - Checks=["iam_user_mfa_enabled_console_access"], - ) - ], - ) - ] - ocsf = OCSF([finding]) + technique_map = { + "T1078": { + "Name": "Valid Accounts", + "TechniqueURL": "https://attack.mitre.org/techniques/T1078/", + "Tactics": [ + "Defense Evasion", + "Persistence", + "Privilege Escalation", + "Initial Access", + ], + }, + "T1098": { + "Name": "Account Manipulation", + "TechniqueURL": "https://attack.mitre.org/techniques/T1098/", + "Tactics": ["Persistence"], + }, + } + + with patch( + "prowler.lib.outputs.ocsf.ocsf._load_mitre_technique_map", + return_value=technique_map, + ): + ocsf = OCSF([finding]) output_data = ocsf.data[0] assert output_data.finding_info.attacks is not None - assert len(output_data.finding_info.attacks) == 2 + assert len(output_data.finding_info.attacks) == 5 attack = output_data.finding_info.attacks[0] assert isinstance(attack, MITREAttack) - assert attack.technique.uid == "T4242" - assert attack.technique.name == "Synthetic Valid Accounts" - assert attack.technique.src_url == "https://attack.mitre.org/techniques/T4242/" + assert attack.technique.uid == "T1078" + assert attack.technique.name == "Valid Accounts" + assert attack.technique.src_url == "https://attack.mitre.org/techniques/T1078/" assert attack.tactic is not None assert [attack.tactic.name for attack in output_data.finding_info.attacks] == [ + "Defense Evasion", "Persistence", "Privilege Escalation", + "Initial Access", + "Persistence", + ] + assert [ + attack.technique.uid for attack in output_data.finding_info.attacks + ] == [ + "T1078", + "T1078", + "T1078", + "T1078", + "T1098", ] - def test_transform_mitre_attacks_unknown_technique(self): + def test_transform_mitre_attacks_ignores_unknown_technique(self): finding = generate_finding_output( provider="aws", - compliance={"MITRE-ATTACK": ["T9999"]}, + compliance={"MITRE-ATTACK": ["T9999", "T1098"]}, ) - finding.metadata.Compliance = [ - Compliance( - Framework="MITRE-ATTACK", - Name="MITRE ATT&CK compliance framework", - Provider="AWS", - Version="", - Description="MITRE ATT&CK test framework", - Requirements=[], - ) - ] - ocsf = OCSF([finding]) - assert ocsf.data[0].finding_info.attacks is None + with patch( + "prowler.lib.outputs.ocsf.ocsf._load_mitre_technique_map", + return_value={ + "T1098": { + "Name": "Account Manipulation", + "Tactics": ["Persistence"], + } + }, + ): + ocsf = OCSF([finding]) + attacks = ocsf.data[0].finding_info.attacks - def test_transform_mitre_attacks_without_mitre_metadata(self): + assert attacks is not None + assert len(attacks) == 1 + assert attacks[0].technique.uid == "T1098" + + def test_transform_mitre_attacks_provider_without_catalog(self): finding = generate_finding_output( provider="kubernetes", compliance={"MITRE-ATTACK": ["T1078"]}, check_type=[], ) - ocsf = OCSF([finding]) + with patch( + "prowler.lib.outputs.ocsf.ocsf._load_mitre_technique_map", + return_value={}, + ): + ocsf = OCSF([finding]) assert ocsf.data[0].finding_info.attacks is None + def test_load_mitre_technique_map_logs_failure(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.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) + finally: + _load_mitre_technique_map.cache_clear() + + def test_transform_mitre_attacks_skips_technique_without_name(self): + findings = [ + generate_finding_output( + provider="aws", + compliance={"MITRE-ATTACK": ["T1078", "T1098"]}, + ), + generate_finding_output(provider="aws"), + ] + technique_map = { + "T1078": {"Tactics": ["Initial Access"]}, + "T1098": { + "Name": "Account Manipulation", + "Tactics": ["Persistence"], + }, + } + + with ( + patch( + "prowler.lib.outputs.ocsf.ocsf._load_mitre_technique_map", + return_value=technique_map, + ), + patch("prowler.lib.outputs.ocsf.ocsf.logger.warning") as mock_warning, + ): + ocsf = OCSF(findings) + + assert len(ocsf.data) == 2 + attacks = ocsf.data[0].finding_info.attacks + assert attacks is not None + assert [attack.technique.uid for attack in attacks] == ["T1098"] + mock_warning.assert_called_once_with( + "Skipping MITRE ATT&CK technique T1078 for provider aws: missing Name" + ) + def test_scan_id_is_unique_per_provider_and_account(self): findings = [ generate_finding_output(provider="aws", account_uid="111111111111"),