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

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
This commit is contained in:
Nacho Rivera
2024-04-30 17:34:10 +02:00
committed by GitHub
parent 484cf6f49d
commit 722554ad3f
11 changed files with 2690 additions and 100 deletions
@@ -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"
)
+7
View File
@@ -267,6 +267,7 @@ def display_data(
# 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 display_data(
):
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(
File diff suppressed because it is too large Load Diff
+14 -2
View File
@@ -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]
+5 -10
View File
@@ -17,9 +17,9 @@ from prowler.lib.outputs.compliance.generic import (
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:
@@ -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(
@@ -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
-28
View File
@@ -4,34 +4,6 @@ from pydantic import BaseModel
# 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.
+12 -2
View File
@@ -9,6 +9,10 @@ from prowler.config.config import (
)
from prowler.lib.logger import logger
from prowler.lib.outputs.common_models import FindingOutput
from prowler.lib.outputs.compliance.mitre_attack.models import (
MitreAttackAWS,
MitreAttackAzure,
)
from prowler.lib.outputs.compliance.models import (
Check_Output_CSV_AWS_CIS,
Check_Output_CSV_AWS_ISO27001_2013,
@@ -18,7 +22,6 @@ from prowler.lib.outputs.compliance.models import (
Check_Output_CSV_GCP_CIS,
Check_Output_CSV_Generic_Compliance,
Check_Output_CSV_KUBERNETES_CIS,
Check_Output_MITRE_ATTACK,
)
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.utils.utils import file_exists, open_file
@@ -117,6 +120,13 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
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,
MitreAttackAzure,
)
file_descriptors.update({output_mode: file_descriptor})
else:
file_descriptor = initialize_file_descriptor(
filename,
@@ -170,7 +180,7 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
file_descriptor = initialize_file_descriptor(
filename,
output_mode,
Check_Output_MITRE_ATTACK,
MitreAttackAWS,
)
file_descriptors.update({output_mode: file_descriptor})
-1
View File
@@ -88,7 +88,6 @@ def report(check_findings, provider):
available_compliance_frameworks
)
)
fill_compliance(
output_options,
finding,