chore(mitre azure): add mapping to mitre for azure provider

prowler/__main__.py

@@ -124,6 +124,7 @@ def prowler():
  bulk_checks_metadata = update_checks_metadata_with_compliance(
  bulk_compliance_frameworks, bulk_checks_metadata
  )
+
  # Update checks metadata if the --custom-checks-metadata-file is present
  custom_checks_metadata = None
  if custom_checks_metadata_file:

prowler/lib/check/compliance_models.py

@@ -123,7 +123,7 @@ class ISO27001_2013_Requirement_Attribute(BaseModel):
 class Mitre_Requirement_Attribute(BaseModel):
  """MITRE Requirement Attribute"""
 
- AWSService: str
+ Service: str
  Category: str
  Value: str
  Comment: str

prowler/lib/outputs/compliance/compliance.py

@@ -17,9 +17,9 @@
 from prowler.lib.outputs.compliance.iso27001_2013_aws import (
  write_compliance_row_iso27001_2013_aws,
 )
-from prowler.lib.outputs.compliance.mitre_attack_aws import (
+from prowler.lib.outputs.compliance.mittre_attack import (
  get_mitre_attack_table,
- write_compliance_row_mitre_attack_aws,
+ write_compliance_row_mitre_attack,
 )
 
 
@@ -77,7 +77,6 @@
  )
  if compliance_name.replace("-", "_") in input_compliance_frameworks:
  check_compliances.append(compliance)
-
  return check_compliances
 
 
@@ -125,12 +124,9 @@
  file_descriptors, finding, compliance, output_options, provider
  )
 
- elif (
- compliance.Framework == "MITRE-ATTACK"
- and compliance.Version == ""
- and compliance.Provider == "AWS"
- ):
- write_compliance_row_mitre_attack_aws(
+ elif compliance.Framework == "MITRE-ATTACK" and compliance.Version == "":
+ # print(file_descriptors)
+ write_compliance_row_mitre_attack(
  file_descriptors, finding, compliance, output_options, provider
  )
 

prowler/lib/outputs/compliance/mitre_attack_aws.py → prowler/lib/outputs/compliance/mittre_attack.py

@@ -1,74 +1,91 @@
 from csv import DictWriter
+from importlib import import_module
 
 from colorama import Fore, Style
 from tabulate import tabulate
 
 from prowler.config.config import orange_color, timestamp
-from prowler.lib.outputs.compliance.models import Check_Output_MITRE_ATTACK
+from prowler.lib.logger import logger
 from prowler.lib.outputs.csv.csv import generate_csv_fields
 from prowler.lib.outputs.utils import unroll_list
 from prowler.lib.utils.utils import outputs_unix_timestamp
 
 
-def write_compliance_row_mitre_attack_aws(
+def write_compliance_row_mitre_attack(
  file_descriptors, finding, compliance, output_options, provider
 ):
- compliance_output = compliance.Framework
- if compliance.Version != "":
- compliance_output += "_" + compliance.Version
- if compliance.Provider != "":
- compliance_output += "_" + compliance.Provider
+ try:
+ compliance_output = compliance.Framework
+ if compliance.Version != "":
+ compliance_output += "_" + compliance.Version
+ if compliance.Provider != "":
+ compliance_output += "_" + compliance.Provider
 
- compliance_output = compliance_output.lower().replace("-", "_")
- csv_header = generate_csv_fields(Check_Output_MITRE_ATTACK)
- csv_writer = DictWriter(
- file_descriptors[compliance_output],
- fieldnames=csv_header,
- delimiter=";",
- )
- for requirement in compliance.Requirements:
- requirement_description = requirement.Description
- requirement_id = requirement.Id
- requirement_name = requirement.Name
- attributes_aws_services = ", ".join(
- attribute.AWSService for attribute in requirement.Attributes
- )
- attributes_categories = ", ".join(
- attribute.Category for attribute in requirement.Attributes
- )
- attributes_values = ", ".join(
- attribute.Value for attribute in requirement.Attributes
- )
- attributes_comments = ", ".join(
- attribute.Comment for attribute in requirement.Attributes
- )
- compliance_row = Check_Output_MITRE_ATTACK(
- Provider=finding.check_metadata.Provider,
- Description=compliance.Description,
- AccountId=provider.identity.account,
- Region=finding.region,
- AssessmentDate=outputs_unix_timestamp(
- output_options.unix_timestamp, timestamp
- ),
- Requirements_Id=requirement_id,
- Requirements_Description=requirement_description,
- Requirements_Name=requirement_name,
- Requirements_Tactics=unroll_list(requirement.Tactics),
- Requirements_SubTechniques=unroll_list(requirement.SubTechniques),
- Requirements_Platforms=unroll_list(requirement.Platforms),
- Requirements_TechniqueURL=requirement.TechniqueURL,
- Requirements_Attributes_AWSServices=attributes_aws_services,
- Requirements_Attributes_Categories=attributes_categories,
- Requirements_Attributes_Values=attributes_values,
- Requirements_Attributes_Comments=attributes_comments,
- Status=finding.status,
- StatusExtended=finding.status_extended,
- ResourceId=finding.resource_id,
- CheckId=finding.check_metadata.CheckID,
- Muted=finding.muted,
+ mitre_attack_model_name = "Check_Output_MITRE_ATTACK_" + compliance.Provider
+ module = import_module("prowler.lib.outputs.compliance.models")
+ mitre_attack_model = getattr(module, mitre_attack_model_name)
+ compliance_output = compliance_output.lower().replace("-", "_")
+ csv_header = generate_csv_fields(mitre_attack_model)
+ csv_writer = DictWriter(
+ file_descriptors[compliance_output],
+ fieldnames=csv_header,
+ delimiter=";",
  )
+ for requirement in compliance.Requirements:
+ requirement_description = requirement.Description
+ requirement_id = requirement.Id
+ requirement_name = requirement.Name
+ attributes_services = ", ".join(
+ attribute.Service for attribute in requirement.Attributes
+ )
+ attributes_categories = ", ".join(
+ attribute.Category for attribute in requirement.Attributes
+ )
+ attributes_values = ", ".join(
+ attribute.Value for attribute in requirement.Attributes
+ )
+ attributes_comments = ", ".join(
+ attribute.Comment for attribute in requirement.Attributes
+ )
+
+ common_data = {
+ "Provider": finding.check_metadata.Provider,
+ "Description": compliance.Description,
+ "AssessmentDate": outputs_unix_timestamp(
+ output_options.unix_timestamp, timestamp
+ ),
+ "Requirements_Id": requirement_id,
+ "Requirements_Name": requirement_name,
+ "Requirements_Description": requirement_description,
+ "Requirements_Tactics": unroll_list(requirement.Tactics),
+ "Requirements_SubTechniques": unroll_list(requirement.SubTechniques),
+ "Requirements_Platforms": unroll_list(requirement.Platforms),
+ "Requirements_TechniqueURL": requirement.TechniqueURL,
+ "Requirements_Attributes_Services": attributes_services,
+ "Requirements_Attributes_Categories": attributes_categories,
+ "Requirements_Attributes_Values": attributes_values,
+ "Requirements_Attributes_Comments": attributes_comments,
+ "Status": finding.status,
+ "StatusExtended": finding.status_extended,
+ "ResourceId": finding.resource_id,
+ "CheckId": finding.check_metadata.CheckID,
+ "Muted": finding.muted,
+ }
+ if compliance.Provider == "AWS":
+ common_data["AccountId"] = provider.identity.account
+ common_data["Region"] = finding.region
+ elif compliance.Provider == "Azure":
+ common_data["SubscriptionId"] = unroll_list(
+ provider.identity.subscriptions
+ )
+
+ compliance_row = mitre_attack_model(**common_data)
 
- csv_writer.writerow(compliance_row.__dict__)
+ csv_writer.writerow(compliance_row.__dict__)
+ except Exception as error:
+ logger.critical(
+ f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+ )
 
 
 def get_mitre_attack_table(
@@ -92,12 +109,14 @@
  for index, finding in enumerate(findings):
  check = bulk_checks_metadata[finding.check_metadata.CheckID]
  check_compliances = check.Compliance
+ # print(check_compliances)
  for compliance in check_compliances:
  if (
  "MITRE-ATTACK" in compliance.Framework
  and compliance.Version in compliance_framework
  ):
  for requirement in compliance.Requirements:
+ # print(requirement)
  for tactic in requirement.Tactics:
  if tactic not in tactics:
  tactics[tactic] = {"FAIL": 0, "PASS": 0, "Muted": 0}

prowler/lib/outputs/compliance/models.py

@@ -4,9 +4,9 @@
 
 
 # TODO: move this to outputs/<compliance>/models.py
-class Check_Output_MITRE_ATTACK(BaseModel):
+class Check_Output_MITRE_ATTACK_AWS(BaseModel):
  """
- Check_Output_MITRE_ATTACK generates a finding's output in CSV MITRE ATTACK format.
+ Check_Output_MITRE_ATTACK generates a finding's output in CSV MITRE ATTACK format for AWS.
  """
 
  Provider: str
@@ -21,7 +21,34 @@ class Check_Output_MITRE_ATTACK(BaseModel):
  Requirements_SubTechniques: str
  Requirements_Platforms: str
  Requirements_TechniqueURL: str
- Requirements_Attributes_AWSServices: str
+ Requirements_Attributes_Services: str
+ Requirements_Attributes_Categories: str
+ Requirements_Attributes_Values: str
+ Requirements_Attributes_Comments: str
+ Status: str
+ StatusExtended: str
+ ResourceId: str
+ CheckId: str
+ Muted: bool
+
+
+class Check_Output_MITRE_ATTACK_Azure(BaseModel):
+ """
+ Check_Output_MITRE_ATTACK generates a finding's output in CSV MITRE ATTACK format for AWS.
+ """
+
+ Provider: str
+ Description: str
+ SubscriptionId: str
+ AssessmentDate: str
+ Requirements_Id: str
+ Requirements_Name: str
+ Requirements_Description: str
+ Requirements_Tactics: str
+ Requirements_SubTechniques: str
+ Requirements_Platforms: str
+ Requirements_TechniqueURL: str
+ Requirements_Attributes_Services: str
  Requirements_Attributes_Categories: str
  Requirements_Attributes_Values: str
  Requirements_Attributes_Comments: str

prowler/lib/outputs/file_descriptors.py

@@ -18,7 +18,8 @@
  Check_Output_CSV_GCP_CIS,
  Check_Output_CSV_Generic_Compliance,
  Check_Output_CSV_KUBERNETES_CIS,
- Check_Output_MITRE_ATTACK,
+ Check_Output_MITRE_ATTACK_AWS,
+ Check_Output_MITRE_ATTACK_Azure,
 )
 from prowler.lib.outputs.csv.csv import generate_csv_fields
 from prowler.lib.utils.utils import file_exists, open_file
@@ -117,6 +118,13 @@
  Check_Output_CSV_AZURE_CIS,
  )
  file_descriptors.update({output_mode: file_descriptor})
+ elif output_mode == "mitre_attack_azure":
+ file_descriptor = initialize_file_descriptor(
+ filename,
+ output_mode,
+ Check_Output_MITRE_ATTACK_Azure,
+ )
+ file_descriptors.update({output_mode: file_descriptor})
  else:
  file_descriptor = initialize_file_descriptor(
  filename,
@@ -170,7 +178,7 @@
  file_descriptor = initialize_file_descriptor(
  filename,
  output_mode,
- Check_Output_MITRE_ATTACK,
+ Check_Output_MITRE_ATTACK_AWS,
  )
  file_descriptors.update({output_mode: file_descriptor})
 

prowler/lib/outputs/outputs.py

@@ -88,7 +88,6 @@ def report(check_findings, provider):
  available_compliance_frameworks
  )
  )
-
  fill_compliance(
  output_options,
  finding,