mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 20:42:02 +00:00
chore(mitre): add MITRE ATT&CK output class (#4425)
This commit is contained in:
Sergio Garcia
@@ -48,6 +48,11 @@ from prowler.lib.outputs.compliance.cis.cis_azure import AzureCIS
|
||||
from prowler.lib.outputs.compliance.cis.cis_gcp import GCPCIS
|
||||
from prowler.lib.outputs.compliance.cis.cis_kubernetes import KubernetesCIS
|
||||
from prowler.lib.outputs.compliance.compliance import display_compliance_table
|
||||
from prowler.lib.outputs.compliance.mitre_attack.mitre_attack_aws import AWSMitreAttack
|
||||
from prowler.lib.outputs.compliance.mitre_attack.mitre_attack_azure import (
|
||||
AzureMitreAttack,
|
||||
)
|
||||
from prowler.lib.outputs.compliance.mitre_attack.mitre_attack_gcp import GCPMitreAttack
|
||||
from prowler.lib.outputs.csv.models import CSV
|
||||
from prowler.lib.outputs.finding import Finding
|
||||
from prowler.lib.outputs.html.html import HTML
|
||||
@@ -373,6 +378,19 @@ def prowler():
|
||||
file_path=filename,
|
||||
)
|
||||
cis_finding.batch_write_data_to_file()
|
||||
elif compliance_name == "mitre_attack_aws":
|
||||
# Generate MITRE ATT&CK Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
mitre_attack_finding = AWSMitreAttack(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
mitre_attack_finding.batch_write_data_to_file()
|
||||
|
||||
elif provider == "azure":
|
||||
for compliance_name in input_compliance_frameworks:
|
||||
@@ -389,6 +407,19 @@ def prowler():
|
||||
file_path=filename,
|
||||
)
|
||||
cis_finding.batch_write_data_to_file()
|
||||
elif compliance_name == "mitre_attack_azure":
|
||||
# Generate MITRE ATT&CK Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
mitre_attack_finding = AzureMitreAttack(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
mitre_attack_finding.batch_write_data_to_file()
|
||||
|
||||
elif provider == "gcp":
|
||||
for compliance_name in input_compliance_frameworks:
|
||||
@@ -405,6 +436,19 @@ def prowler():
|
||||
file_path=filename,
|
||||
)
|
||||
cis_finding.batch_write_data_to_file()
|
||||
elif compliance_name == "mitre_attack_gcp":
|
||||
# Generate MITRE ATT&CK Finding Object
|
||||
filename = (
|
||||
f"{global_provider.output_options.output_directory}/compliance/"
|
||||
f"{global_provider.output_options.output_filename}_{compliance_name}.csv"
|
||||
)
|
||||
mitre_attack_finding = GCPMitreAttack(
|
||||
findings=finding_outputs,
|
||||
compliance=bulk_compliance_frameworks[compliance_name],
|
||||
create_file_descriptor=True,
|
||||
file_path=filename,
|
||||
)
|
||||
mitre_attack_finding.batch_write_data_to_file()
|
||||
|
||||
elif provider == "kubernetes":
|
||||
for compliance_name in input_compliance_frameworks:
|
||||
|
||||
@@ -19,7 +19,6 @@ from prowler.lib.outputs.compliance.iso27001_2013_aws import (
|
||||
)
|
||||
from prowler.lib.outputs.compliance.mitre_attack.mitre_attack import (
|
||||
get_mitre_attack_table,
|
||||
write_compliance_row_mitre_attack,
|
||||
)
|
||||
|
||||
|
||||
@@ -102,8 +101,11 @@ def fill_compliance(
|
||||
file_descriptors, finding, compliance, output_options, provider
|
||||
)
|
||||
|
||||
# FIXME: Remove this once we merge all the compliance frameworks
|
||||
elif compliance.Framework == "CIS":
|
||||
continue
|
||||
elif compliance.Framework == "MITRE-ATTACK" and compliance.Version == "":
|
||||
continue
|
||||
|
||||
elif (
|
||||
"AWS-Well-Architected-Framework" in compliance.Framework
|
||||
@@ -122,11 +124,6 @@ def fill_compliance(
|
||||
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:
|
||||
write_compliance_row_generic(
|
||||
file_descriptors, finding, compliance, output_options, provider
|
||||
|
||||
@@ -1,101 +1,7 @@
|
||||
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.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(file_descriptors, finding, compliance, provider):
|
||||
try:
|
||||
compliance_output = compliance.Framework
|
||||
if compliance.Version != "":
|
||||
compliance_output += "_" + compliance.Version
|
||||
if compliance.Provider != "":
|
||||
compliance_output += "_" + compliance.Provider
|
||||
|
||||
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:
|
||||
|
||||
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
|
||||
)
|
||||
elif compliance.Provider == "GCP":
|
||||
attributes_services = ", ".join(
|
||||
attribute.GCPService 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
|
||||
)
|
||||
elif compliance.Provider == "GCP":
|
||||
common_data["ProjectId"] = unroll_list(provider.projects)
|
||||
|
||||
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}"
|
||||
)
|
||||
from prowler.config.config import orange_color
|
||||
|
||||
|
||||
def get_mitre_attack_table(
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
from csv import DictWriter
|
||||
from venv import logger
|
||||
|
||||
from prowler.lib.check.compliance_models import ComplianceBaseModel
|
||||
from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput
|
||||
from prowler.lib.outputs.compliance.mitre_attack.models import MitreAttackAWS
|
||||
from prowler.lib.outputs.finding import Finding
|
||||
from prowler.lib.outputs.utils import unroll_list
|
||||
|
||||
|
||||
class AWSMitreAttack(ComplianceOutput):
|
||||
"""
|
||||
This class represents the AWS MITRE ATT&CK compliance output.
|
||||
|
||||
Attributes:
|
||||
- _data (list): A list to store transformed data from findings.
|
||||
- _file_descriptor (TextIOWrapper): A file descriptor to write data to a file.
|
||||
|
||||
Methods:
|
||||
- transform: Transforms findings into AWS MITRE ATT&CK compliance format.
|
||||
- batch_write_data_to_file: Writes the findings data to a CSV file in AWS MITRE ATT&CK compliance format.
|
||||
"""
|
||||
|
||||
def transform(
|
||||
self,
|
||||
findings: list[Finding],
|
||||
compliance: ComplianceBaseModel,
|
||||
compliance_name: str,
|
||||
) -> None:
|
||||
"""
|
||||
Transforms a list of findings into AWS MITRE ATT&CK compliance format.
|
||||
|
||||
Parameters:
|
||||
- findings (list): A list of findings.
|
||||
- compliance (ComplianceBaseModel): A compliance model.
|
||||
- compliance_name (str): The name of the compliance model.
|
||||
|
||||
Returns:
|
||||
- None
|
||||
"""
|
||||
for finding in findings:
|
||||
# Get the compliance requirements for the finding
|
||||
finding_requirements = finding.compliance.get(compliance_name, [])
|
||||
for requirement in compliance.Requirements:
|
||||
if requirement.Id in finding_requirements:
|
||||
compliance_row = MitreAttackAWS(
|
||||
Provider=finding.provider,
|
||||
Description=compliance.Description,
|
||||
AccountId=finding.account_uid,
|
||||
Region=finding.region,
|
||||
AssessmentDate=str(finding.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=", ".join(
|
||||
attribute.AWSService for attribute in requirement.Attributes
|
||||
),
|
||||
Requirements_Attributes_Categories=", ".join(
|
||||
attribute.Category for attribute in requirement.Attributes
|
||||
),
|
||||
Requirements_Attributes_Values=", ".join(
|
||||
attribute.Value for attribute in requirement.Attributes
|
||||
),
|
||||
Requirements_Attributes_Comments=", ".join(
|
||||
attribute.Comment for attribute in requirement.Attributes
|
||||
),
|
||||
Status=finding.status,
|
||||
StatusExtended=finding.status_extended,
|
||||
ResourceId=finding.resource_uid,
|
||||
ResourceName=finding.resource_name,
|
||||
CheckId=finding.check_id,
|
||||
Muted=finding.muted,
|
||||
)
|
||||
self._data.append(compliance_row)
|
||||
|
||||
def batch_write_data_to_file(self) -> None:
|
||||
"""
|
||||
Writes the findings data to a CSV file in AWS MITRE ATT&CK compliance format.
|
||||
|
||||
Returns:
|
||||
- None
|
||||
"""
|
||||
try:
|
||||
if (
|
||||
getattr(self, "_file_descriptor", None)
|
||||
and not self._file_descriptor.closed
|
||||
and self._data
|
||||
):
|
||||
csv_writer = DictWriter(
|
||||
self._file_descriptor,
|
||||
fieldnames=[field.upper() for field in self._data[0].dict().keys()],
|
||||
delimiter=";",
|
||||
)
|
||||
csv_writer.writeheader()
|
||||
for finding in self._data:
|
||||
csv_writer.writerow(
|
||||
{k.upper(): v for k, v in finding.dict().items()}
|
||||
)
|
||||
self._file_descriptor.close()
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
@@ -0,0 +1,110 @@
|
||||
from csv import DictWriter
|
||||
from venv import logger
|
||||
|
||||
from prowler.lib.check.compliance_models import ComplianceBaseModel
|
||||
from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput
|
||||
from prowler.lib.outputs.compliance.mitre_attack.models import MitreAttackAzure
|
||||
from prowler.lib.outputs.finding import Finding
|
||||
from prowler.lib.outputs.utils import unroll_list
|
||||
|
||||
|
||||
class AzureMitreAttack(ComplianceOutput):
|
||||
"""
|
||||
This class represents the Azure MITRE ATT&CK compliance output.
|
||||
|
||||
Attributes:
|
||||
- _data (list): A list to store transformed data from findings.
|
||||
- _file_descriptor (TextIOWrapper): A file descriptor to write data to a file.
|
||||
|
||||
Methods:
|
||||
- transform: Transforms findings into Azure MITRE ATT&CK compliance format.
|
||||
- batch_write_data_to_file: Writes the findings data to a CSV file in Azure MITRE ATT&CK compliance format.
|
||||
"""
|
||||
|
||||
def transform(
|
||||
self,
|
||||
findings: list[Finding],
|
||||
compliance: ComplianceBaseModel,
|
||||
compliance_name: str,
|
||||
) -> None:
|
||||
"""
|
||||
Transforms a list of findings into Azure MITRE ATT&CK compliance format.
|
||||
|
||||
Parameters:
|
||||
- findings (list): A list of findings.
|
||||
- compliance (ComplianceBaseModel): A compliance model.
|
||||
- compliance_name (str): The name of the compliance model.
|
||||
|
||||
Returns:
|
||||
- None
|
||||
"""
|
||||
for finding in findings:
|
||||
# Get the compliance requirements for the finding
|
||||
finding_requirements = finding.compliance.get(compliance_name, [])
|
||||
for requirement in compliance.Requirements:
|
||||
if requirement.Id in finding_requirements:
|
||||
compliance_row = MitreAttackAzure(
|
||||
Provider=finding.provider,
|
||||
Description=compliance.Description,
|
||||
SubscriptionId=finding.account_uid,
|
||||
Location=finding.region,
|
||||
AssessmentDate=str(finding.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=", ".join(
|
||||
attribute.AzureService
|
||||
for attribute in requirement.Attributes
|
||||
),
|
||||
Requirements_Attributes_Categories=", ".join(
|
||||
attribute.Category for attribute in requirement.Attributes
|
||||
),
|
||||
Requirements_Attributes_Values=", ".join(
|
||||
attribute.Value for attribute in requirement.Attributes
|
||||
),
|
||||
Requirements_Attributes_Comments=", ".join(
|
||||
attribute.Comment for attribute in requirement.Attributes
|
||||
),
|
||||
Status=finding.status,
|
||||
StatusExtended=finding.status_extended,
|
||||
ResourceId=finding.resource_uid,
|
||||
ResourceName=finding.resource_name,
|
||||
CheckId=finding.check_id,
|
||||
Muted=finding.muted,
|
||||
)
|
||||
self._data.append(compliance_row)
|
||||
|
||||
def batch_write_data_to_file(self) -> None:
|
||||
"""
|
||||
Writes the findings data to a CSV file in Azure MITRE ATT&CK compliance format.
|
||||
|
||||
Returns:
|
||||
- None
|
||||
"""
|
||||
try:
|
||||
if (
|
||||
getattr(self, "_file_descriptor", None)
|
||||
and not self._file_descriptor.closed
|
||||
and self._data
|
||||
):
|
||||
csv_writer = DictWriter(
|
||||
self._file_descriptor,
|
||||
fieldnames=[field.upper() for field in self._data[0].dict().keys()],
|
||||
delimiter=";",
|
||||
)
|
||||
csv_writer.writeheader()
|
||||
for finding in self._data:
|
||||
csv_writer.writerow(
|
||||
{k.upper(): v for k, v in finding.dict().items()}
|
||||
)
|
||||
self._file_descriptor.close()
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
@@ -0,0 +1,109 @@
|
||||
from csv import DictWriter
|
||||
from venv import logger
|
||||
|
||||
from prowler.lib.check.compliance_models import ComplianceBaseModel
|
||||
from prowler.lib.outputs.compliance.compliance_output import ComplianceOutput
|
||||
from prowler.lib.outputs.compliance.mitre_attack.models import MitreAttackGCP
|
||||
from prowler.lib.outputs.finding import Finding
|
||||
from prowler.lib.outputs.utils import unroll_list
|
||||
|
||||
|
||||
class GCPMitreAttack(ComplianceOutput):
|
||||
"""
|
||||
This class represents the GCP MITRE ATT&CK compliance output.
|
||||
|
||||
Attributes:
|
||||
- _data (list): A list to store transformed data from findings.
|
||||
- _file_descriptor (TextIOWrapper): A file descriptor to write data to a file.
|
||||
|
||||
Methods:
|
||||
- transform: Transforms findings into GCP MITRE ATT&CK compliance format.
|
||||
- batch_write_data_to_file: Writes the findings data to a CSV file in GCP MITRE ATT&CK compliance format.
|
||||
"""
|
||||
|
||||
def transform(
|
||||
self,
|
||||
findings: list[Finding],
|
||||
compliance: ComplianceBaseModel,
|
||||
compliance_name: str,
|
||||
) -> None:
|
||||
"""
|
||||
Transforms a list of findings into GCP MITRE ATT&CK compliance format.
|
||||
|
||||
Parameters:
|
||||
- findings (list): A list of findings.
|
||||
- compliance (ComplianceBaseModel): A compliance model.
|
||||
- compliance_name (str): The name of the compliance model.
|
||||
|
||||
Returns:
|
||||
- None
|
||||
"""
|
||||
for finding in findings:
|
||||
# Get the compliance requirements for the finding
|
||||
finding_requirements = finding.compliance.get(compliance_name, [])
|
||||
for requirement in compliance.Requirements:
|
||||
if requirement.Id in finding_requirements:
|
||||
compliance_row = MitreAttackGCP(
|
||||
Provider=finding.provider,
|
||||
Description=compliance.Description,
|
||||
ProjectId=finding.account_uid,
|
||||
Location=finding.region,
|
||||
AssessmentDate=str(finding.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=", ".join(
|
||||
attribute.GCPService for attribute in requirement.Attributes
|
||||
),
|
||||
Requirements_Attributes_Categories=", ".join(
|
||||
attribute.Category for attribute in requirement.Attributes
|
||||
),
|
||||
Requirements_Attributes_Values=", ".join(
|
||||
attribute.Value for attribute in requirement.Attributes
|
||||
),
|
||||
Requirements_Attributes_Comments=", ".join(
|
||||
attribute.Comment for attribute in requirement.Attributes
|
||||
),
|
||||
Status=finding.status,
|
||||
StatusExtended=finding.status_extended,
|
||||
ResourceId=finding.resource_uid,
|
||||
ResourceName=finding.resource_name,
|
||||
CheckId=finding.check_id,
|
||||
Muted=finding.muted,
|
||||
)
|
||||
self._data.append(compliance_row)
|
||||
|
||||
def batch_write_data_to_file(self) -> None:
|
||||
"""
|
||||
Writes the findings data to a CSV file in GCP MITRE ATT&CK compliance format.
|
||||
|
||||
Returns:
|
||||
- None
|
||||
"""
|
||||
try:
|
||||
if (
|
||||
getattr(self, "_file_descriptor", None)
|
||||
and not self._file_descriptor.closed
|
||||
and self._data
|
||||
):
|
||||
csv_writer = DictWriter(
|
||||
self._file_descriptor,
|
||||
fieldnames=[field.upper() for field in self._data[0].dict().keys()],
|
||||
delimiter=";",
|
||||
)
|
||||
csv_writer.writeheader()
|
||||
for finding in self._data:
|
||||
csv_writer.writerow(
|
||||
{k.upper(): v for k, v in finding.dict().items()}
|
||||
)
|
||||
self._file_descriptor.close()
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
@@ -27,6 +27,7 @@ class MitreAttackAWS(BaseModel):
|
||||
ResourceId: str
|
||||
CheckId: str
|
||||
Muted: bool
|
||||
ResourceName: str
|
||||
|
||||
|
||||
class MitreAttackAzure(BaseModel):
|
||||
@@ -54,6 +55,8 @@ class MitreAttackAzure(BaseModel):
|
||||
ResourceId: str
|
||||
CheckId: str
|
||||
Muted: bool
|
||||
ResourceName: str
|
||||
Location: str
|
||||
|
||||
|
||||
class MitreAttackGCP(BaseModel):
|
||||
@@ -81,3 +84,9 @@ class MitreAttackGCP(BaseModel):
|
||||
ResourceId: str
|
||||
CheckId: str
|
||||
Muted: bool
|
||||
ResourceName: str
|
||||
Location: str
|
||||
|
||||
|
||||
# TODO: Create a parent class for the common fields of MITRE ATT&CK and have the specific classes from each provider to inherit from it.
|
||||
# It is not done yet because it is needed to respect the current order of the fields in the output file.
|
||||
|
||||
@@ -4,11 +4,6 @@ from typing import Any
|
||||
|
||||
from prowler.config.config import csv_file_suffix
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.lib.outputs.compliance.mitre_attack.models import (
|
||||
MitreAttackAWS,
|
||||
MitreAttackAzure,
|
||||
MitreAttackGCP,
|
||||
)
|
||||
from prowler.lib.outputs.compliance.models import (
|
||||
Check_Output_CSV_AWS_ISO27001_2013,
|
||||
Check_Output_CSV_AWS_Well_Architected,
|
||||
@@ -22,12 +17,9 @@ from prowler.lib.utils.utils import file_exists, open_file
|
||||
|
||||
def initialize_file_descriptor(
|
||||
filename: str,
|
||||
output_mode: str,
|
||||
provider: Any = None,
|
||||
format: Any = Finding,
|
||||
write_header: bool = True,
|
||||
) -> TextIOWrapper:
|
||||
"""Open/Create the output file. If needed include headers or the required format, by default will use the Finding"""
|
||||
"""Open/Create the output file. If needed include headers or the required format, by default will use the FindingOutput"""
|
||||
try:
|
||||
if file_exists(filename):
|
||||
file_descriptor = open_file(
|
||||
@@ -44,8 +36,7 @@ def initialize_file_descriptor(
|
||||
csv_writer = DictWriter(
|
||||
file_descriptor, fieldnames=csv_header, delimiter=";"
|
||||
)
|
||||
if write_header:
|
||||
csv_writer.writeheader()
|
||||
csv_writer.writeheader()
|
||||
return file_descriptor
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
@@ -70,54 +61,32 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
|
||||
# FIXME: Remove this once we merge all the compliance frameworks
|
||||
if "cis_" in output_mode:
|
||||
continue
|
||||
elif "mitre_attack_" in output_mode:
|
||||
continue
|
||||
|
||||
elif provider.type == "gcp":
|
||||
filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
|
||||
if output_mode == "mitre_attack_gcp":
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
output_mode,
|
||||
provider.type,
|
||||
MitreAttackGCP,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
else:
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
output_mode,
|
||||
provider.type,
|
||||
Check_Output_CSV_Generic_Compliance,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
Check_Output_CSV_Generic_Compliance,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
|
||||
elif provider.type == "kubernetes":
|
||||
filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
output_mode,
|
||||
provider.type,
|
||||
Check_Output_CSV_Generic_Compliance,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
|
||||
elif provider.type == "azure":
|
||||
filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
|
||||
if output_mode == "mitre_attack_azure":
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
output_mode,
|
||||
provider.type,
|
||||
MitreAttackAzure,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
else:
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
output_mode,
|
||||
provider.type,
|
||||
Check_Output_CSV_Generic_Compliance,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
Check_Output_CSV_Generic_Compliance,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
|
||||
elif provider.type == "aws":
|
||||
# Compliance frameworks
|
||||
@@ -125,8 +94,6 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
|
||||
if output_mode == "ens_rd2022_aws":
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
output_mode,
|
||||
provider.type,
|
||||
Check_Output_CSV_ENS_RD2022,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
@@ -134,8 +101,6 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
|
||||
elif "aws_well_architected_framework" in output_mode:
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
output_mode,
|
||||
provider.type,
|
||||
Check_Output_CSV_AWS_Well_Architected,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
@@ -143,27 +108,14 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
|
||||
elif output_mode == "iso27001_2013_aws":
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
output_mode,
|
||||
provider.type,
|
||||
Check_Output_CSV_AWS_ISO27001_2013,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
|
||||
elif output_mode == "mitre_attack_aws":
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
output_mode,
|
||||
provider.type,
|
||||
MitreAttackAWS,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
|
||||
else:
|
||||
# Generic Compliance framework
|
||||
file_descriptor = initialize_file_descriptor(
|
||||
filename,
|
||||
output_mode,
|
||||
provider.type,
|
||||
Check_Output_CSV_Generic_Compliance,
|
||||
)
|
||||
file_descriptors.update({output_mode: file_descriptor})
|
||||
|
||||
-1
@@ -78,7 +78,6 @@ class TestAWSCIS:
|
||||
def test_batch_write_data_to_file(self):
|
||||
mock_file = StringIO()
|
||||
findings = [generate_finding_output(compliance={"CIS-1.4": "2.1.3"})]
|
||||
# Clear the data from CSV class
|
||||
output = AWSCIS(findings, CIS_1_4_AWS)
|
||||
output._file_descriptor = mock_file
|
||||
|
||||
@@ -2,6 +2,10 @@ from prowler.lib.check.compliance_models import (
|
||||
CIS_Requirement_Attribute,
|
||||
Compliance_Requirement,
|
||||
ComplianceBaseModel,
|
||||
Mitre_Requirement,
|
||||
Mitre_Requirement_Attribute_AWS,
|
||||
Mitre_Requirement_Attribute_Azure,
|
||||
Mitre_Requirement_Attribute_GCP,
|
||||
)
|
||||
|
||||
CIS_1_4_AWS_NAME = "cis_1.4_aws"
|
||||
@@ -147,6 +151,138 @@ CIS_1_5_AWS = ComplianceBaseModel(
|
||||
],
|
||||
)
|
||||
|
||||
MITRE_ATTACK_AWS_NAME = "mitre_attack_aws"
|
||||
MITRE_ATTACK_AWS = ComplianceBaseModel(
|
||||
Framework="MITRE-ATTACK",
|
||||
Provider="AWS",
|
||||
Version="",
|
||||
Description="MITRE ATT&CK® is a globally-accessible knowledge base of adversary tactics and techniques based on real-world observations. The ATT&CK knowledge base is used as a foundation for the development of specific threat models and methodologies in the private sector, in government, and in the cybersecurity product and service community.",
|
||||
Requirements=[
|
||||
Mitre_Requirement(
|
||||
Name="Exploit Public-Facing Application",
|
||||
Id="T1190",
|
||||
Tactics=["Initial Access"],
|
||||
SubTechniques=[],
|
||||
Description="Adversaries may attempt to exploit a weakness in an Internet-facing host or system to initially access a network. The weakness in the system can be a software bug, a temporary glitch, or a misconfiguration.",
|
||||
Platforms=["Containers", "IaaS", "Linux", "Network", "Windows", "macOS"],
|
||||
TechniqueURL="https://attack.mitre.org/techniques/T1190/",
|
||||
Attributes=[
|
||||
Mitre_Requirement_Attribute_AWS(
|
||||
AWSService="AWS CloudEndure Disaster Recovery",
|
||||
Category="Respond",
|
||||
Value="Significant",
|
||||
Comment="AWS CloudEndure Disaster Recovery enables the replication and recovery of servers into AWS Cloud. In the event that a public-facing application or server is compromised, AWS CloudEndure can be used to provision an instance of the server from a previous point in time within minutes. As a result, this mapping is given a score of Significant.",
|
||||
)
|
||||
],
|
||||
Checks=[
|
||||
"drs_job_exist",
|
||||
"config_recorder_all_regions_enabled",
|
||||
"rds_instance_minor_version_upgrade_enabled",
|
||||
"rds_instance_backup_enabled",
|
||||
"securityhub_enabled",
|
||||
"elbv2_waf_acl_attached",
|
||||
"guardduty_is_enabled",
|
||||
"inspector2_is_enabled",
|
||||
"inspector2_active_findings_exist",
|
||||
"awslambda_function_not_publicly_accessible",
|
||||
"ec2_instance_public_ip",
|
||||
],
|
||||
)
|
||||
],
|
||||
)
|
||||
MITRE_ATTACK_AZURE_NAME = "mitre_attack_azure"
|
||||
MITRE_ATTACK_AZURE = ComplianceBaseModel(
|
||||
Framework="MITRE-ATTACK",
|
||||
Provider="Azure",
|
||||
Version="",
|
||||
Description="MITRE ATT&CK® is a globally-accessible knowledge base of adversary tactics and techniques based on real-world observations. The ATT&CK knowledge base is used as a foundation for the development of specific threat models and methodologies in the private sector, in government, and in the cybersecurity product and service community.",
|
||||
Requirements=[
|
||||
Mitre_Requirement(
|
||||
Name="Exploit Public-Facing Application",
|
||||
Id="T1190",
|
||||
Tactics=["Initial Access"],
|
||||
SubTechniques=[],
|
||||
Description="Adversaries may attempt to exploit a weakness in an Internet-facing host or system to initially access a network. The weakness in the system can be a software bug, a temporary glitch, or a misconfiguration.",
|
||||
Platforms=["Containers", "IaaS", "Linux", "Network", "Windows", "macOS"],
|
||||
TechniqueURL="https://attack.mitre.org/techniques/T1190/",
|
||||
Attributes=[
|
||||
Mitre_Requirement_Attribute_Azure(
|
||||
AzureService="Azure SQL Database",
|
||||
Category="Detect",
|
||||
Value="Minimal",
|
||||
Comment="This control may alert on usage of faulty SQL statements. This generates an alert for a possible SQL injection by an application. Alerts may not be generated on usage of valid SQL statements by attackers for malicious purposes.",
|
||||
)
|
||||
],
|
||||
Checks=[
|
||||
"aks_clusters_created_with_private_nodes",
|
||||
"aks_clusters_public_access_disabled",
|
||||
"app_ensure_java_version_is_latest",
|
||||
"app_ensure_php_version_is_latest",
|
||||
"app_ensure_python_version_is_latest",
|
||||
"defender_assessments_vm_endpoint_protection_installed",
|
||||
"defender_assessments_vm_endpoint_protection_installed",
|
||||
"defender_auto_provisioning_log_analytics_agent_vms_on",
|
||||
"defender_auto_provisioning_vulnerabilty_assessments_machines_on",
|
||||
"defender_container_images_resolved_vulnerabilities",
|
||||
"defender_container_images_scan_enabled",
|
||||
"defender_ensure_defender_for_app_services_is_on",
|
||||
"defender_ensure_defender_for_arm_is_on",
|
||||
"defender_ensure_defender_for_azure_sql_databases_is_on",
|
||||
"defender_ensure_defender_for_containers_is_on",
|
||||
"defender_ensure_defender_for_cosmosdb_is_on",
|
||||
"defender_ensure_defender_for_databases_is_on",
|
||||
"defender_ensure_defender_for_dns_is_on",
|
||||
"defender_ensure_defender_for_keyvault_is_on",
|
||||
"defender_ensure_defender_for_os_relational_databases_is_on",
|
||||
"defender_ensure_defender_for_server_is_on",
|
||||
"defender_ensure_defender_for_sql_servers_is_on",
|
||||
"defender_ensure_defender_for_storage_is_on",
|
||||
"defender_ensure_iot_hub_defender_is_on",
|
||||
"defender_ensure_mcas_is_enabled",
|
||||
"defender_ensure_notify_alerts_severity_is_high",
|
||||
"defender_ensure_notify_emails_to_owners",
|
||||
"defender_ensure_system_updates_are_applied",
|
||||
"defender_ensure_wdatp_is_enabled",
|
||||
],
|
||||
)
|
||||
],
|
||||
)
|
||||
MITRE_ATTACK_GCP_NAME = "mitre_attack_gcp"
|
||||
MITRE_ATTACK_GCP = ComplianceBaseModel(
|
||||
Framework="MITRE-ATTACK",
|
||||
Provider="GCP",
|
||||
Version="",
|
||||
Description="MITRE ATT&CK® is a globally-accessible knowledge base of adversary tactics and techniques based on real-world observations. The ATT&CK knowledge base is used as a foundation for the development of specific threat models and methodologies in the private sector, in government, and in the cybersecurity product and service community.",
|
||||
Requirements=[
|
||||
Mitre_Requirement(
|
||||
Name="Exploit Public-Facing Application",
|
||||
Id="T1190",
|
||||
Tactics=["Initial Access"],
|
||||
SubTechniques=[],
|
||||
Description="Adversaries may attempt to exploit a weakness in an Internet-facing host or system to initially access a network. The weakness in the system can be a software bug, a temporary glitch, or a misconfiguration.",
|
||||
Platforms=["Containers", "IaaS", "Linux", "Network", "Windows", "macOS"],
|
||||
TechniqueURL="https://attack.mitre.org/techniques/T1190/",
|
||||
Attributes=[
|
||||
Mitre_Requirement_Attribute_GCP(
|
||||
GCPService="Artifact Registry",
|
||||
Category="Protect",
|
||||
Value="Partial",
|
||||
Comment="Once this control is deployed, it can detect known vulnerabilities in various Linux OS packages. This information can be used to patch, isolate, or remove vulnerable software and machines. This control does not directly protect against exploitation and is not effective against zero day attacks, vulnerabilities with no available patch, and other end-of-life packages.",
|
||||
)
|
||||
],
|
||||
Checks=[
|
||||
"cloudsql_instance_public_access",
|
||||
"cloudsql_instance_public_ip",
|
||||
"cloudstorage_bucket_public_access",
|
||||
"compute_firewall_rdp_access_from_the_internet_allowed",
|
||||
"compute_firewall_ssh_access_from_the_internet_allowed",
|
||||
"compute_instance_public_ip",
|
||||
"compute_public_address_shodan",
|
||||
"kms_key_not_publicly_accessible",
|
||||
],
|
||||
)
|
||||
],
|
||||
)
|
||||
NOT_PRESENT_COMPLIANCE_NAME = "not_present_compliance_name"
|
||||
NOT_PRESENT_COMPLIANCE = ComplianceBaseModel(
|
||||
Framework="NOT_EXISTENT",
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
from datetime import datetime
|
||||
from io import StringIO
|
||||
|
||||
from freezegun import freeze_time
|
||||
from mock import patch
|
||||
|
||||
from prowler.lib.outputs.compliance.mitre_attack.mitre_attack_aws import AWSMitreAttack
|
||||
from prowler.lib.outputs.compliance.mitre_attack.models import MitreAttackAWS
|
||||
from prowler.lib.outputs.utils import unroll_list
|
||||
from tests.lib.outputs.compliance.fixtures import MITRE_ATTACK_AWS
|
||||
from tests.lib.outputs.fixtures.fixtures import generate_finding_output
|
||||
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
|
||||
|
||||
|
||||
class TestAWSCIS:
|
||||
def test_output_transform(self):
|
||||
findings = [generate_finding_output(compliance={"MITRE-ATTACK": "T1190"})]
|
||||
|
||||
output = AWSMitreAttack(findings, MITRE_ATTACK_AWS)
|
||||
output_data = output.data[0]
|
||||
assert isinstance(output_data, MitreAttackAWS)
|
||||
assert output_data.Provider == "aws"
|
||||
assert output_data.Description == MITRE_ATTACK_AWS.Description
|
||||
assert output_data.AccountId == AWS_ACCOUNT_NUMBER
|
||||
assert output_data.Region == AWS_REGION_EU_WEST_1
|
||||
assert output_data.Requirements_Id == MITRE_ATTACK_AWS.Requirements[0].Id
|
||||
assert output_data.Requirements_Name == MITRE_ATTACK_AWS.Requirements[0].Name
|
||||
assert (
|
||||
output_data.Requirements_Description
|
||||
== MITRE_ATTACK_AWS.Requirements[0].Description
|
||||
)
|
||||
assert output_data.Requirements_Tactics == unroll_list(
|
||||
MITRE_ATTACK_AWS.Requirements[0].Tactics
|
||||
)
|
||||
assert output_data.Requirements_SubTechniques == unroll_list(
|
||||
MITRE_ATTACK_AWS.Requirements[0].SubTechniques
|
||||
)
|
||||
assert output_data.Requirements_Platforms == unroll_list(
|
||||
MITRE_ATTACK_AWS.Requirements[0].Platforms
|
||||
)
|
||||
assert (
|
||||
output_data.Requirements_TechniqueURL
|
||||
== MITRE_ATTACK_AWS.Requirements[0].TechniqueURL
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Services == ", ".join(
|
||||
attribute.AWSService
|
||||
for attribute in MITRE_ATTACK_AWS.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Categories == ", ".join(
|
||||
attribute.Category
|
||||
for attribute in MITRE_ATTACK_AWS.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Values == ", ".join(
|
||||
attribute.Value for attribute in MITRE_ATTACK_AWS.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Comments == ", ".join(
|
||||
attribute.Comment
|
||||
for attribute in MITRE_ATTACK_AWS.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Status == "PASS"
|
||||
assert output_data.StatusExtended == ""
|
||||
assert output_data.ResourceId == ""
|
||||
assert output_data.ResourceName == ""
|
||||
assert output_data.CheckId == "test-check-id"
|
||||
assert not output_data.Muted
|
||||
|
||||
@freeze_time(datetime.now())
|
||||
def test_batch_write_data_to_file(self):
|
||||
mock_file = StringIO()
|
||||
findings = [generate_finding_output(compliance={"MITRE-ATTACK": "T1190"})]
|
||||
output = AWSMitreAttack(findings, MITRE_ATTACK_AWS)
|
||||
output._file_descriptor = mock_file
|
||||
|
||||
with patch.object(mock_file, "close", return_value=None):
|
||||
output.batch_write_data_to_file()
|
||||
|
||||
mock_file.seek(0)
|
||||
content = mock_file.read()
|
||||
print(repr(content) + "\n")
|
||||
expected_csv = f"PROVIDER;DESCRIPTION;ACCOUNTID;REGION;ASSESSMENTDATE;REQUIREMENTS_ID;REQUIREMENTS_NAME;REQUIREMENTS_DESCRIPTION;REQUIREMENTS_TACTICS;REQUIREMENTS_SUBTECHNIQUES;REQUIREMENTS_PLATFORMS;REQUIREMENTS_TECHNIQUEURL;REQUIREMENTS_ATTRIBUTES_SERVICES;REQUIREMENTS_ATTRIBUTES_CATEGORIES;REQUIREMENTS_ATTRIBUTES_VALUES;REQUIREMENTS_ATTRIBUTES_COMMENTS;STATUS;STATUSEXTENDED;RESOURCEID;CHECKID;MUTED;RESOURCENAME\r\naws;MITRE ATT&CK® is a globally-accessible knowledge base of adversary tactics and techniques based on real-world observations. The ATT&CK knowledge base is used as a foundation for the development of specific threat models and methodologies in the private sector, in government, and in the cybersecurity product and service community.;123456789012;eu-west-1;{datetime.now()};T1190;Exploit Public-Facing Application;Adversaries may attempt to exploit a weakness in an Internet-facing host or system to initially access a network. The weakness in the system can be a software bug, a temporary glitch, or a misconfiguration.;Initial Access;;Containers | IaaS | Linux | Network | Windows | macOS;https://attack.mitre.org/techniques/T1190/;AWS CloudEndure Disaster Recovery;Respond;Significant;AWS CloudEndure Disaster Recovery enables the replication and recovery of servers into AWS Cloud. In the event that a public-facing application or server is compromised, AWS CloudEndure can be used to provision an instance of the server from a previous point in time within minutes. As a result, this mapping is given a score of Significant.;PASS;;;test-check-id;False;\r\n"
|
||||
assert content == expected_csv
|
||||
@@ -0,0 +1,102 @@
|
||||
from datetime import datetime
|
||||
from io import StringIO
|
||||
|
||||
from freezegun import freeze_time
|
||||
from mock import patch
|
||||
|
||||
from prowler.lib.outputs.compliance.mitre_attack.mitre_attack_azure import (
|
||||
AzureMitreAttack,
|
||||
)
|
||||
from prowler.lib.outputs.compliance.mitre_attack.models import MitreAttackAzure
|
||||
from prowler.lib.outputs.utils import unroll_list
|
||||
from tests.lib.outputs.compliance.fixtures import MITRE_ATTACK_AZURE
|
||||
from tests.lib.outputs.fixtures.fixtures import generate_finding_output
|
||||
from tests.providers.azure.azure_fixtures import (
|
||||
AZURE_SUBSCRIPTION_ID,
|
||||
AZURE_SUBSCRIPTION_NAME,
|
||||
)
|
||||
|
||||
|
||||
class TestAzureCIS:
|
||||
def test_output_transform(self):
|
||||
findings = [
|
||||
generate_finding_output(
|
||||
provider="azure",
|
||||
compliance={"MITRE-ATTACK": "T1190"},
|
||||
account_name=AZURE_SUBSCRIPTION_NAME,
|
||||
account_uid=AZURE_SUBSCRIPTION_ID,
|
||||
region="",
|
||||
)
|
||||
]
|
||||
|
||||
output = AzureMitreAttack(findings, MITRE_ATTACK_AZURE)
|
||||
output_data = output.data[0]
|
||||
assert isinstance(output_data, MitreAttackAzure)
|
||||
assert output_data.Provider == "azure"
|
||||
assert output_data.Description == MITRE_ATTACK_AZURE.Description
|
||||
assert output_data.SubscriptionId == AZURE_SUBSCRIPTION_ID
|
||||
assert output_data.Location == ""
|
||||
assert output_data.Requirements_Id == MITRE_ATTACK_AZURE.Requirements[0].Id
|
||||
assert output_data.Requirements_Name == MITRE_ATTACK_AZURE.Requirements[0].Name
|
||||
assert (
|
||||
output_data.Requirements_Description
|
||||
== MITRE_ATTACK_AZURE.Requirements[0].Description
|
||||
)
|
||||
assert output_data.Requirements_Tactics == unroll_list(
|
||||
MITRE_ATTACK_AZURE.Requirements[0].Tactics
|
||||
)
|
||||
assert output_data.Requirements_SubTechniques == unroll_list(
|
||||
MITRE_ATTACK_AZURE.Requirements[0].SubTechniques
|
||||
)
|
||||
assert output_data.Requirements_Platforms == unroll_list(
|
||||
MITRE_ATTACK_AZURE.Requirements[0].Platforms
|
||||
)
|
||||
assert (
|
||||
output_data.Requirements_TechniqueURL
|
||||
== MITRE_ATTACK_AZURE.Requirements[0].TechniqueURL
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Services == ", ".join(
|
||||
attribute.AzureService
|
||||
for attribute in MITRE_ATTACK_AZURE.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Categories == ", ".join(
|
||||
attribute.Category
|
||||
for attribute in MITRE_ATTACK_AZURE.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Values == ", ".join(
|
||||
attribute.Value
|
||||
for attribute in MITRE_ATTACK_AZURE.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Comments == ", ".join(
|
||||
attribute.Comment
|
||||
for attribute in MITRE_ATTACK_AZURE.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Status == "PASS"
|
||||
assert output_data.StatusExtended == ""
|
||||
assert output_data.ResourceId == ""
|
||||
assert output_data.ResourceName == ""
|
||||
assert output_data.CheckId == "test-check-id"
|
||||
assert not output_data.Muted
|
||||
|
||||
@freeze_time(datetime.now())
|
||||
def test_batch_write_data_to_file(self):
|
||||
mock_file = StringIO()
|
||||
findings = [
|
||||
generate_finding_output(
|
||||
provider="azure",
|
||||
compliance={"MITRE-ATTACK": "T1190"},
|
||||
account_name=AZURE_SUBSCRIPTION_NAME,
|
||||
account_uid=AZURE_SUBSCRIPTION_ID,
|
||||
region="",
|
||||
)
|
||||
]
|
||||
output = AzureMitreAttack(findings, MITRE_ATTACK_AZURE)
|
||||
output._file_descriptor = mock_file
|
||||
|
||||
with patch.object(mock_file, "close", return_value=None):
|
||||
output.batch_write_data_to_file()
|
||||
|
||||
mock_file.seek(0)
|
||||
content = mock_file.read()
|
||||
expected_csv = f"PROVIDER;DESCRIPTION;SUBSCRIPTIONID;ASSESSMENTDATE;REQUIREMENTS_ID;REQUIREMENTS_NAME;REQUIREMENTS_DESCRIPTION;REQUIREMENTS_TACTICS;REQUIREMENTS_SUBTECHNIQUES;REQUIREMENTS_PLATFORMS;REQUIREMENTS_TECHNIQUEURL;REQUIREMENTS_ATTRIBUTES_SERVICES;REQUIREMENTS_ATTRIBUTES_CATEGORIES;REQUIREMENTS_ATTRIBUTES_VALUES;REQUIREMENTS_ATTRIBUTES_COMMENTS;STATUS;STATUSEXTENDED;RESOURCEID;CHECKID;MUTED;RESOURCENAME;LOCATION\r\nazure;MITRE ATT&CK® is a globally-accessible knowledge base of adversary tactics and techniques based on real-world observations. The ATT&CK knowledge base is used as a foundation for the development of specific threat models and methodologies in the private sector, in government, and in the cybersecurity product and service community.;{AZURE_SUBSCRIPTION_ID};{datetime.now()};T1190;Exploit Public-Facing Application;Adversaries may attempt to exploit a weakness in an Internet-facing host or system to initially access a network. The weakness in the system can be a software bug, a temporary glitch, or a misconfiguration.;Initial Access;;Containers | IaaS | Linux | Network | Windows | macOS;https://attack.mitre.org/techniques/T1190/;Azure SQL Database;Detect;Minimal;This control may alert on usage of faulty SQL statements. This generates an alert for a possible SQL injection by an application. Alerts may not be generated on usage of valid SQL statements by attackers for malicious purposes.;PASS;;;test-check-id;False;;\r\n"
|
||||
assert content == expected_csv
|
||||
@@ -0,0 +1,96 @@
|
||||
from datetime import datetime
|
||||
from io import StringIO
|
||||
|
||||
from freezegun import freeze_time
|
||||
from mock import patch
|
||||
|
||||
from prowler.lib.outputs.compliance.mitre_attack.mitre_attack_gcp import GCPMitreAttack
|
||||
from prowler.lib.outputs.compliance.mitre_attack.models import MitreAttackGCP
|
||||
from prowler.lib.outputs.utils import unroll_list
|
||||
from tests.lib.outputs.compliance.fixtures import MITRE_ATTACK_GCP
|
||||
from tests.lib.outputs.fixtures.fixtures import generate_finding_output
|
||||
from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID
|
||||
|
||||
|
||||
class TestGCPCIS:
|
||||
def test_output_transform(self):
|
||||
findings = [
|
||||
generate_finding_output(
|
||||
provider="gcp",
|
||||
compliance={"MITRE-ATTACK": "T1190"},
|
||||
account_name=GCP_PROJECT_ID,
|
||||
account_uid=GCP_PROJECT_ID,
|
||||
region="",
|
||||
)
|
||||
]
|
||||
|
||||
output = GCPMitreAttack(findings, MITRE_ATTACK_GCP)
|
||||
output_data = output.data[0]
|
||||
assert isinstance(output_data, MitreAttackGCP)
|
||||
assert output_data.Provider == "gcp"
|
||||
assert output_data.Description == MITRE_ATTACK_GCP.Description
|
||||
assert output_data.ProjectId == GCP_PROJECT_ID
|
||||
assert output_data.Location == ""
|
||||
assert output_data.Requirements_Id == MITRE_ATTACK_GCP.Requirements[0].Id
|
||||
assert output_data.Requirements_Name == MITRE_ATTACK_GCP.Requirements[0].Name
|
||||
assert (
|
||||
output_data.Requirements_Description
|
||||
== MITRE_ATTACK_GCP.Requirements[0].Description
|
||||
)
|
||||
assert output_data.Requirements_Tactics == unroll_list(
|
||||
MITRE_ATTACK_GCP.Requirements[0].Tactics
|
||||
)
|
||||
assert output_data.Requirements_SubTechniques == unroll_list(
|
||||
MITRE_ATTACK_GCP.Requirements[0].SubTechniques
|
||||
)
|
||||
assert output_data.Requirements_Platforms == unroll_list(
|
||||
MITRE_ATTACK_GCP.Requirements[0].Platforms
|
||||
)
|
||||
assert (
|
||||
output_data.Requirements_TechniqueURL
|
||||
== MITRE_ATTACK_GCP.Requirements[0].TechniqueURL
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Services == ", ".join(
|
||||
attribute.GCPService
|
||||
for attribute in MITRE_ATTACK_GCP.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Categories == ", ".join(
|
||||
attribute.Category
|
||||
for attribute in MITRE_ATTACK_GCP.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Values == ", ".join(
|
||||
attribute.Value for attribute in MITRE_ATTACK_GCP.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Requirements_Attributes_Comments == ", ".join(
|
||||
attribute.Comment
|
||||
for attribute in MITRE_ATTACK_GCP.Requirements[0].Attributes
|
||||
)
|
||||
assert output_data.Status == "PASS"
|
||||
assert output_data.StatusExtended == ""
|
||||
assert output_data.ResourceId == ""
|
||||
assert output_data.ResourceName == ""
|
||||
assert output_data.CheckId == "test-check-id"
|
||||
assert not output_data.Muted
|
||||
|
||||
@freeze_time(datetime.now())
|
||||
def test_batch_write_data_to_file(self):
|
||||
mock_file = StringIO()
|
||||
findings = [
|
||||
generate_finding_output(
|
||||
provider="gcp",
|
||||
compliance={"MITRE-ATTACK": "T1190"},
|
||||
account_name=GCP_PROJECT_ID,
|
||||
account_uid=GCP_PROJECT_ID,
|
||||
region="",
|
||||
)
|
||||
]
|
||||
output = GCPMitreAttack(findings, MITRE_ATTACK_GCP)
|
||||
output._file_descriptor = mock_file
|
||||
|
||||
with patch.object(mock_file, "close", return_value=None):
|
||||
output.batch_write_data_to_file()
|
||||
|
||||
mock_file.seek(0)
|
||||
content = mock_file.read()
|
||||
expected_csv = f"PROVIDER;DESCRIPTION;PROJECTID;ASSESSMENTDATE;REQUIREMENTS_ID;REQUIREMENTS_NAME;REQUIREMENTS_DESCRIPTION;REQUIREMENTS_TACTICS;REQUIREMENTS_SUBTECHNIQUES;REQUIREMENTS_PLATFORMS;REQUIREMENTS_TECHNIQUEURL;REQUIREMENTS_ATTRIBUTES_SERVICES;REQUIREMENTS_ATTRIBUTES_CATEGORIES;REQUIREMENTS_ATTRIBUTES_VALUES;REQUIREMENTS_ATTRIBUTES_COMMENTS;STATUS;STATUSEXTENDED;RESOURCEID;CHECKID;MUTED;RESOURCENAME;LOCATION\r\ngcp;MITRE ATT&CK® is a globally-accessible knowledge base of adversary tactics and techniques based on real-world observations. The ATT&CK knowledge base is used as a foundation for the development of specific threat models and methodologies in the private sector, in government, and in the cybersecurity product and service community.;{GCP_PROJECT_ID};{datetime.now()};T1190;Exploit Public-Facing Application;Adversaries may attempt to exploit a weakness in an Internet-facing host or system to initially access a network. The weakness in the system can be a software bug, a temporary glitch, or a misconfiguration.;Initial Access;;Containers | IaaS | Linux | Network | Windows | macOS;https://attack.mitre.org/techniques/T1190/;Artifact Registry;Protect;Partial;Once this control is deployed, it can detect known vulnerabilities in various Linux OS packages. This information can be used to patch, isolate, or remove vulnerable software and machines. This control does not directly protect against exploitation and is not effective against zero day attacks, vulnerabilities with no available patch, and other end-of-life packages.;PASS;;;test-check-id;False;;\r\n"
|
||||
assert content == expected_csv
|
||||
Reference in New Issue
Block a user