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

Co-authored-by: Sergio Garcia <[email protected]>

dashboard/compliance/mitre_attack_azure.py

@@ -0,0 +1,23 @@
+import warnings
+
+from dashboard.common_methods import get_section_containers_format2
+
+warnings.filterwarnings("ignore")
+
+
+def get_table(data):
+ aux = data[
+ [
+ "REQUIREMENTS_ID",
+ "REQUIREMENTS_SUBTECHNIQUES",
+ "CHECKID",
+ "STATUS",
+ "REGION",
+ "ACCOUNTID",
+ "RESOURCEID",
+ ]
+ ].copy()
+
+ return get_section_containers_format2(
+ aux, "REQUIREMENTS_ID", "REQUIREMENTS_SUBTECHNIQUES"
+ )

dashboard/pages/compliance.py

@@ -267,6 +267,7 @@ def load_csv_files(files):
  # Rename the column SUBSCRIPTIONID to ACCOUNTID for Azure
  if data.columns.str.contains("SUBSCRIPTIONID").any():
  data.rename(columns={"SUBSCRIPTIONID": "ACCOUNTID"}, inplace=True)
+ data["REGION"] = "-"
  # Handle v3 azure cis compliance
  if data.columns.str.contains("SUBSCRIPTION").any():
  data.rename(columns={"SUBSCRIPTION": "ACCOUNTID"}, inplace=True)
@@ -433,6 +434,12 @@ def load_csv_files(files):
  ):
  pie_2 = get_bar_graph(df, "REQUIREMENTS_ATTRIBUTES_SERVICE")
  current_filter = "services"
+ elif (
+ "REQUIREMENTS_ID" in df.columns
+ and not df["REQUIREMENTS_ID"].isnull().values.any()
+ ):
+ pie_2 = get_bar_graph(df, "REQUIREMENTS_ID")
+ current_filter = "techniques"
  else:
  fig = px.pie()
  fig.update_layout(

prowler/lib/check/compliance_models.py

@@ -120,7 +120,7 @@ class ISO27001_2013_Requirement_Attribute(BaseModel):
 
 
 # MITRE Requirement Attribute
-class Mitre_Requirement_Attribute(BaseModel):
+class Mitre_Requirement_Attribute_AWS(BaseModel):
  """MITRE Requirement Attribute"""
 
  AWSService: str
@@ -129,6 +129,16 @@ class Mitre_Requirement_Attribute(BaseModel):
  Comment: str
 
 
+# MITRE Requirement Attribute
+class Mitre_Requirement_Attribute_Azure(BaseModel):
+ """MITRE Requirement Attribute"""
+
+ AzureService: str
+ Category: str
+ Value: str
+ Comment: str
+
+
 # MITRE Requirement
 class Mitre_Requirement(BaseModel):
  """Mitre_Requirement holds the model for every MITRE requirement"""
@@ -140,7 +150,9 @@ class Mitre_Requirement(BaseModel):
  Description: str
  Platforms: list[str]
  TechniqueURL: str
- Attributes: list[Mitre_Requirement_Attribute]
+ Attributes: Union[
+ list[Mitre_Requirement_Attribute_AWS], list[Mitre_Requirement_Attribute_Azure]
+ ]
  Checks: list[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.mitre_attack.mitre_attack import (
  get_mitre_attack_table,
- write_compliance_row_mitre_attack_aws,
+ write_compliance_row_mitre_attack,
 )
 
 
@@ -77,7 +77,6 @@ def get_check_compliance_frameworks_in_input(
  )
  if compliance_name.replace("-", "_") in input_compliance_frameworks:
  check_compliances.append(compliance)
-
  return check_compliances
 
 
@@ -125,13 +124,9 @@ def fill_compliance(
  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(
- file_descriptors, finding, compliance, output_options, provider
+ elif compliance.Framework == "MITRE-ATTACK" and compliance.Version == "":
+ write_compliance_row_mitre_attack(
+ file_descriptors, finding, compliance, provider
  )
 
  else:

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

@@ -1,74 +1,95 @@
 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(
- 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
+def write_compliance_row_mitre_attack(file_descriptors, finding, 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 = "MitreAttack" + compliance.Provider
+ module = import_module("prowler.lib.outputs.compliance.mitre_attack.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:
 
- csv_writer.writerow(compliance_row.__dict__)
+ if compliance.Provider == "AWS":
+ attributes_services = ", ".join(
+ attribute.AWSService for attribute in requirement.Attributes
+ )
+ elif compliance.Provider == "Azure":
+ attributes_services = ", ".join(
+ attribute.AzureService for attribute in requirement.Attributes
+ )
+ requirement_description = requirement.Description
+ requirement_id = requirement.Id
+ requirement_name = requirement.Name
+ 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(
+ provider.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__)
+ except Exception as error:
+ logger.error(
+ f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+ )
 
 
 def get_mitre_attack_table(

prowler/lib/outputs/compliance/mitre_attack/models.py

@@ -0,0 +1,56 @@
+from pydantic import BaseModel
+
+
+class MitreAttackAWS(BaseModel):
+ """
+ MitreAttackAWS generates a finding's output in CSV MITRE ATTACK format for AWS.
+ """
+
+ Provider: str
+ Description: str
+ AccountId: str
+ Region: 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
+ Status: str
+ StatusExtended: str
+ ResourceId: str
+ CheckId: str
+ Muted: bool
+
+
+class MitreAttackAzure(BaseModel):
+ """
+ MitreAttackAzure generates a finding's output in CSV MITRE ATTACK format for Azure.
+ """
+
+ 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
+ Status: str
+ StatusExtended: str
+ ResourceId: str
+ CheckId: str
+ Muted: bool

prowler/lib/outputs/compliance/models.py

@@ -4,34 +4,6 @@
 
 
 # TODO: move this to outputs/<compliance>/models.py
-class Check_Output_MITRE_ATTACK(BaseModel):
- """
- Check_Output_MITRE_ATTACK generates a finding's output in CSV MITRE ATTACK format.
- """
-
- Provider: str
- Description: str
- AccountId: str
- Region: 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_AWSServices: 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_CSV_ENS_RD2022(BaseModel):
  """
  Check_Output_CSV_ENS_RD2022 generates a finding's output in CSV ENS RD2022 format.