ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 04:21:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(ASFF): create class (#4368)

Browse Source 
Co-authored-by: pedrooot <pedromarting3@gmail.com>
This commit is contained in:
Pepe Fagoaga
2024-07-04 18:04:36 +02:00
committed by GitHub
parent 2345a7384b
commit 673619c8a1
18 changed files with 1255 additions and 1101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/__main__.py
+22 -9
View File
@@ -9,6 +9,7 @@ from colorama import Fore, Style
from prowler.config.config import (
csv_file_suffix,
get_available_compliance_frameworks,
json_asff_file_suffix,
json_ocsf_file_suffix,
)
from prowler.lib.banner import print_banner
@@ -40,6 +41,7 @@ from prowler.lib.check.custom_checks_metadata import (
)
from prowler.lib.cli.parser import ProwlerArgumentParser
from prowler.lib.logger import logger, set_logging_config
from prowler.lib.outputs.asff.asff import ASFF
from prowler.lib.outputs.compliance.compliance import display_compliance_table
from prowler.lib.outputs.csv.models import CSV
from prowler.lib.outputs.finding import Finding
@@ -51,7 +53,7 @@ from prowler.lib.outputs.summary_table import display_summary_table
from prowler.providers.aws.lib.s3.s3 import send_to_s3_bucket
from prowler.providers.aws.lib.security_hub.security_hub import (
batch_send_to_security_hub,
prepare_security_hub_findings,
filter_security_hub_findings_per_region,
resolve_security_hub_previous_findings,
verify_security_hub_integration_enabled_per_region,
)
@@ -295,20 +297,32 @@ def prowler():
if args.output_formats:
for mode in args.output_formats:
if "csv" in mode:
# Generate CSV Finding Object
filename = (
f"{global_provider.output_options.output_directory}/"
f"{global_provider.output_options.output_filename}{csv_file_suffix}"
)
csv_finding = CSV(
csv_output = CSV(
findings=finding_outputs,
create_file_descriptor=True,
file_path=filename,
)
# Write CSV Finding Object to file
csv_finding.batch_write_data_to_file()
csv_output.batch_write_data_to_file()
if "json-asff" in mode:
filename = (
f"{global_provider.output_options.output_directory}/"
f"{global_provider.output_options.output_filename}{json_asff_file_suffix}"
)
asff_output = ASFF(
findings=finding_outputs,
create_file_descriptor=True,
file_path=filename,
)
# Write ASFF Finding Object to file
asff_output.batch_write_data_to_file()
# Close json file if exists
# TODO: generate JSON here
@@ -376,13 +390,12 @@ def prowler():
aws_security_enabled_regions.append(region)
# Prepare the findings to be sent to Security Hub
security_hub_findings_per_region = prepare_security_hub_findings(
findings,
global_provider,
global_provider.output_options,
security_hub_findings_per_region = filter_security_hub_findings_per_region(
asff_output.data,
global_provider.output_options.send_sh_only_fails,
global_provider.output_options.status,
aws_security_enabled_regions,
)
# Send the findings to Security Hub
findings_sent_to_security_hub = batch_send_to_security_hub(
security_hub_findings_per_region, global_provider.session.current_session
prowler/lib/outputs/json_asff/__init__.py → prowler/lib/outputs/asff/__init__.py
View File
prowler/lib/outputs/asff/asff.py
+426
View File
@@ -0,0 +1,426 @@
from json import dump
from os import SEEK_SET
from typing import Optional
from pydantic import BaseModel, validator
from prowler.config.config import prowler_version, timestamp_utc
from prowler.lib.logger import logger
from prowler.lib.outputs.finding import Finding
from prowler.lib.outputs.output import Output
from prowler.lib.utils.utils import hash_sha512
class ASFF(Output):
"""
ASFF class represents a transformation of findings into AWS Security Finding Format (ASFF).
This class provides methods to transform a list of findings into the ASFF format required by AWS Security Hub. It includes operations such as generating unique identifiers, formatting timestamps, handling compliance frameworks, and ensuring the status values match the allowed values in ASFF.
Attributes:
- _data: A list to store the transformed findings.
- _file_descriptor: A file descriptor to write to file.
Methods:
- transform(findings: list[Finding]) -> None: Transforms a list of findings into ASFF format.
- batch_write_data_to_file() -> None: Writes the findings data to a file in JSON ASFF format.
- generate_status(status: str, muted: bool = False) -> str: Generates the ASFF status based on the provided status and muted flag.
- format_resource_tags(tags: str) -> dict: Transforms a string of tags into a dictionary format.
References:
- AWS Security Hub API Reference: https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html
- AWS Security Finding Format Syntax: https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-findings-format-syntax.html
"""
def transform(self, findings: list[Finding]) -> None:
"""
Transforms a list of findings into AWS Security Finding Format (ASFF).
This method iterates over the list of findings provided as input and transforms each finding into the ASFF format required by AWS Security Hub. It performs several operations for each finding, including generating unique identifiers, formatting timestamps, handling compliance frameworks, and ensuring the status values match the allowed values in ASFF.
Parameters:
- findings (list[Finding]): A list of Finding objects representing the findings to be transformed.
Returns:
- None
Notes:
- The method skips findings with a status of "MANUAL" as it is not valid in SecurityHub.
- It generates unique identifiers for each finding based on specific attributes.
- It formats timestamps in the required ASFF format.
- It handles compliance frameworks and associated standards for each finding.
- It ensures that the finding status matches the allowed values in ASFF.
References:
- AWS Security Hub API Reference: https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html
- AWS Security Finding Format Syntax: https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-findings-format-syntax.html
"""
try:
for finding in findings:
# MANUAL status is not valid in SecurityHub
# https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html
if finding.status == "MANUAL":
continue
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
resource_tags = ASFF.format_resource_tags(finding.resource_tags)
associated_standards, compliance_summary = ASFF.format_compliance(
finding.compliance
)
# Ensures finding_status matches allowed values in ASFF
finding_status = ASFF.generate_status(finding.status, finding.muted)
self._data.append(
AWSSecurityFindingFormat(
# The following line cannot be changed because it is the format we use to generate unique findings for AWS Security Hub
# If changed some findings could be lost because the unique identifier will be different
Id=f"prowler-{finding.check_id}-{finding.account_uid}-{finding.region}-{hash_sha512(finding.resource_uid)}",
ProductArn=f"arn:{finding.partition}:securityhub:{finding.region}::product/prowler/prowler",
ProductFields=ProductFields(
ProwlerResourceName=finding.resource_uid,
),
GeneratorId="prowler-" + finding.check_id,
AwsAccountId=finding.account_uid,
Types=finding.check_type.split(","),
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.severity.value),
Title=finding.check_title,
Description=finding.description,
Resources=[
Resource(
Id=finding.resource_uid,
Type=finding.resource_type,
Partition=finding.partition,
Region=finding.region,
Tags=resource_tags,
)
],
Compliance=Compliance(
Status=finding_status,
AssociatedStandards=associated_standards,
RelatedRequirements=compliance_summary,
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=finding.remediation_recommendation_text,
Url=finding.remediation_recommendation_url,
)
),
)
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def batch_write_data_to_file(self) -> None:
"""
Writes the findings data to a file in JSON ASFF format.
This method iterates over the findings data stored in the '_data' attribute and writes it to the file descriptor '_file_descriptor' in JSON format. It starts by writing the JSON opening/header '[', then iterates over each finding, dumping it to the file with an indent of 4 spaces. After writing all findings, it writes the closing ']' to complete the JSON array structure. Finally, it closes the file descriptor.
Returns:
None
"""
try:
if self._file_descriptor and not self._file_descriptor.closed:
# Write JSON opening/header [
self._file_descriptor.write("[")
# Write findings
for finding in self._data:
dump(
finding.dict(exclude_none=True),
self._file_descriptor,
indent=4,
)
self._file_descriptor.write(",")
# Write footer/closing ]
if self._file_descriptor.tell() > 0:
if self._file_descriptor.tell() != 1:
self._file_descriptor.seek(
self._file_descriptor.tell() - 1, SEEK_SET
)
self._file_descriptor.truncate()
self._file_descriptor.write("]")
# Close file descriptor
self._file_descriptor.close()
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
@staticmethod
def generate_status(status: str, muted: bool = False) -> str:
"""
Generates the ASFF status based on the provided status and muted flag.
Parameters:
- status (str): The status of the finding.
- muted (bool): Flag indicating if the finding is muted.
Returns:
- str: The ASFF status corresponding to the provided status and muted flag.
References:
- AWS Security Hub API Reference: https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html
"""
json_asff_status = ""
if muted:
# Per AWS Security Hub "MUTED" is not a valid status
# https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html
json_asff_status = "WARNING"
else:
if status == "PASS":
json_asff_status = "PASSED"
elif status == "FAIL":
json_asff_status = "FAILED"
else:
# MANUAL is set to NOT_AVAILABLE
json_asff_status = "NOT_AVAILABLE"
return json_asff_status
@staticmethod
def format_resource_tags(tags: str) -> dict:
"""
Transforms a string of tags into a dictionary format.
Parameters:
- tags (str): A string containing tags separated by ' | ' and key-value pairs separated by '='.
Returns:
- dict: A dictionary where keys are tag names and values are tag values.
Notes:
- If the input string is empty or None, it returns None.
- Each tag in the input string should be in the format 'key=value'.
- If the input string is not formatted correctly, it logs an error and returns None.
"""
try:
tags_dict = None
if tags:
tags = tags.split(" | ")
tags_dict = {}
for tag in tags:
value = tag.split("=")
tags_dict[value[0]] = value[1]
return tags_dict
except IndexError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return None
except AttributeError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return None
@staticmethod
def format_compliance(compliance: dict) -> tuple[list[dict], list[str]]:
"""
Transforms a dictionary of compliance data into a tuple of associated standards and compliance summaries.
Parameters:
- compliance (dict): A dictionary containing compliance data where keys are standards and values are lists of compliance details.
Returns:
- tuple[list[dict], list[str]]: A tuple containing a list of associated standards (each as a dictionary with 'StandardsId') and a list of compliance summaries.
Notes:
- The method limits the number of associated standards to 20.
- Each compliance summary is a concatenation of the standard key and its associated compliance details.
- If the concatenated summary exceeds 64 characters, it is truncated to 63 characters.
Example:
format_compliance({"standard1": ["detail1", "detail2"], "standard2": ["detail3"]}) -> ([{"StandardsId": "standard1"}, {"StandardsId": "standard2"}], ["standard1 detail1 detail2", "standard2 detail3"])
"""
compliance_summary = []
associated_standards = []
for key, value in compliance.items():
if (
len(associated_standards) < 20
): # AssociatedStandards should NOT have more than 20 items
associated_standards.append({"StandardsId": key})
item = f"{key} {' '.join(value)}"
if len(item) > 64:
item = item[0:63]
compliance_summary.append(item)
return associated_standards, compliance_summary
class ProductFields(BaseModel):
"""
Class representing the Product Fields of a finding in the AWS Security Finding Format.
Attributes:
- ProviderName (str): The name of the provider, default value is "Prowler".
- ProviderVersion (str): The version of the provider, fetched from the prowler_version in config.py.
- ProwlerResourceName (str): The name of the Prowler resource.
"""
ProviderName: str = "Prowler"
ProviderVersion: str = prowler_version
ProwlerResourceName: str
class Severity(BaseModel):
"""
Class representing the severity of a finding in the AWS Security Finding Format.
Attributes:
- Label (str): A string representing the severity label of the finding.
This class is used to define the severity level of a finding in the AWS Security Finding Format.
"""
Label: str
@validator("Label", pre=True, always=True)
def severity_uppercase(severity):
return severity.upper()
class Resource(BaseModel):
"""
Class representing a resource in the AWS Security Finding Format.
Attributes:
- Type (str): The type of the resource.
- Id (str): The unique identifier of the resource.
- Partition (str): The partition where the resource resides.
- Region (str): The region where the resource is located.
- Tags (Optional[dict]): Optional dictionary of tags associated with the resource.
This class defines the structure of a resource within the AWS Security Finding Format. It includes attributes to specify the type, unique identifier, partition, region, and optional tags of the resource.
"""
Type: str
Id: str
Partition: str
Region: str
Tags: Optional[dict]
class Compliance(BaseModel):
"""
Class representing the compliance details of a finding in the AWS Security Finding Format.
Attributes:
- Status (str): The compliance status of the finding.
- RelatedRequirements (list[str]): A list of related compliance requirements for the finding.
- AssociatedStandards (list[dict]): A list of associated standards with the finding, where each item is a dictionary containing the 'StandardsId'.
This class defines the structure of compliance information within the AWS Security Finding Format. It includes attributes to specify the compliance status, related requirements, and associated standards of a finding.
"""
Status: str
RelatedRequirements: list[str]
AssociatedStandards: list[dict]
@validator("Status", pre=True, always=True)
def status(status):
if status not in ["PASSED", "WARNING", "FAILED", "NOT_AVAILABLE"]:
raise ValueError("must contain a space")
return status
class Recommendation(BaseModel):
"""
Class representing a recommendation for remediation in the AWS Security Finding Format.
Attributes:
- Text (str): The text description of the recommendation.
- Url (str): The URL link for additional information related to the recommendation.
This class defines the structure of a recommendation within the AWS Security Finding Format. It includes attributes to specify the text description and URL link for further details regarding the recommendation.
"""
Text: str = ""
Url: str = ""
@validator("Text", pre=True, always=True)
def text_must_not_exceed_512_chars(text):
text_validated = text
if len(text) > 512:
text_validated = text[:509] + "..."
return text_validated
@validator("Url", pre=True, always=True)
def set_default_url_if_empty(url):
default_url = "https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html"
if url:
default_url = url
return default_url
class Remediation(BaseModel):
"""
Class representing a remediation action in the AWS Security Finding Format.
Attributes:
- Recommendation (Recommendation): An instance of the Recommendation class providing details for remediation.
This class defines the structure of a remediation action within the AWS Security Finding Format. It includes an attribute to specify the recommendation for remediation, which is an instance of the Recommendation class.
"""
Recommendation: Recommendation
class AWSSecurityFindingFormat(BaseModel):
"""
AWSSecurityFindingFormat generates a finding's output in JSON ASFF format: https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-findings-format-syntax.html
Attributes:
- SchemaVersion (str): The version of the ASFF schema being used, default value is "2018-10-08".
- Id (str): The unique identifier of the finding.
- ProductArn (str): The ARN of the product generating the finding.
- RecordState (str): The state of the finding record, default value is "ACTIVE".
- ProductFields (ProductFields): An instance of the ProductFields class representing the product fields of the finding.
- GeneratorId (str): The ID of the generator.
- AwsAccountId (str): The AWS account ID associated with the finding.
- Types (list[str]): A list of types associated with the finding, default value is None.
- FirstObservedAt (str): The timestamp when the finding was first observed.
- UpdatedAt (str): The timestamp when the finding was last updated.
- CreatedAt (str): The timestamp when the finding was created.
- Severity (Severity): An instance of the Severity class representing the severity of the finding.
- Title (str): The title of the finding.
- Description (str): The description of the finding, truncated to 1024 characters if longer.
- Resources (list[Resource]): A list of resources associated with the finding, default value is None.
- Compliance (Compliance): An instance of the Compliance class representing the compliance details of the finding.
- Remediation (Remediation): An instance of the Remediation class providing details for remediation.
This class defines the structure of a finding in the AWS Security Finding Format, including various attributes such as schema version, identifiers, timestamps, severity, title, description, resources, compliance details, and remediation information.
"""
SchemaVersion: str = "2018-10-08"
Id: str
ProductArn: str
RecordState: str = "ACTIVE"
ProductFields: ProductFields
GeneratorId: str
AwsAccountId: str
Types: list[str] = None
FirstObservedAt: str
UpdatedAt: str
CreatedAt: str
Severity: Severity
Title: str
Description: str
Resources: list[Resource] = None
Compliance: Compliance
Remediation: Remediation
@validator("Description", pre=True, always=True)
def description_must_not_exceed_1024_chars(description):
description_validated = description
if len(description) > 1024:
description_validated = description[:1021] + "..."
return description_validated
prowler/lib/outputs/csv/models.py
+2 -1
View File
@@ -30,7 +30,7 @@ class CSV(Output):
def batch_write_data_to_file(self) -> None:
"""Writes the findings to a file using the CSV format using the `Output._file_descriptor`."""
try:
if self._file_descriptor:
if self._file_descriptor and not self._file_descriptor.closed:
csv_writer = DictWriter(
self._file_descriptor,
fieldnames=self._data[0].keys(),
@@ -39,6 +39,7 @@ class CSV(Output):
csv_writer.writeheader()
for finding in self._data:
csv_writer.writerow(finding)
self._file_descriptor.close()
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
prowler/lib/outputs/file_descriptors.py
+52 -63
View File
@@ -2,11 +2,7 @@ from csv import DictWriter
from io import TextIOWrapper
from typing import Any
from prowler.config.config import (
csv_file_suffix,
html_file_suffix,
json_asff_file_suffix,
)
from prowler.config.config import csv_file_suffix, html_file_suffix
from prowler.lib.logger import logger
from prowler.lib.outputs.compliance.mitre_attack.models import (
MitreAttackAWS,
@@ -48,10 +44,7 @@ def initialize_file_descriptor(
filename,
"a",
)
if output_mode == "json-asff":
file_descriptor.write("[")
elif "html" in output_mode:
if "html" in output_mode:
add_html_header(file_descriptor, provider)
else:
# Format is the class model of the CSV format to print the headers
@@ -76,6 +69,8 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
# FIXME: Remove this once we always use the new CSV(Output)
if output_mode == "csv":
continue
elif output_mode == "json-asff":
continue
elif output_mode == "html":
filename = f"{output_directory}/{output_filename}{html_file_suffix}"
file_descriptor = initialize_file_descriptor(
@@ -157,68 +152,62 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
file_descriptors.update({output_mode: file_descriptor})
elif provider.type == "aws":
if output_mode == "json-asff":
filename = f"{output_directory}/{output_filename}{json_asff_file_suffix}"
# Compliance frameworks
filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
if output_mode == "ens_rd2022_aws":
file_descriptor = initialize_file_descriptor(
filename, output_mode
filename,
output_mode,
provider.type,
Check_Output_CSV_ENS_RD2022,
)
file_descriptors.update({output_mode: file_descriptor})
else: # Compliance frameworks
filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
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})
elif "cis_" in output_mode:
file_descriptor = initialize_file_descriptor(
filename,
output_mode,
provider.type,
Check_Output_CSV_AWS_CIS,
)
file_descriptors.update({output_mode: file_descriptor})
elif "cis_" in output_mode:
file_descriptor = initialize_file_descriptor(
filename,
output_mode,
provider.type,
Check_Output_CSV_AWS_CIS,
)
file_descriptors.update({output_mode: file_descriptor})
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})
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})
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 == "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})
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})
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})
except Exception as error:
logger.error(
prowler/lib/outputs/json_asff/json_asff.py
-137
View File
@@ -1,137 +0,0 @@
from prowler.config.config import timestamp_utc
from prowler.lib.logger import logger
from prowler.lib.outputs.compliance.compliance import get_check_compliance
from prowler.lib.outputs.json_asff.models import (
Check_Output_JSON_ASFF,
Compliance,
ProductFields,
Recommendation,
Remediation,
Resource,
Severity,
)
from prowler.lib.utils.utils import hash_sha512
def generate_json_asff_status(status: str, muted: bool = False) -> str:
json_asff_status = ""
if muted:
# Per AWS Security Hub "MUTED" is not a valid status
# https://docs.aws.amazon.com/securityhub/1.0/APIReference/API_Compliance.html
json_asff_status = "WARNING"
else:
if status == "PASS":
json_asff_status = "PASSED"
elif status == "FAIL":
json_asff_status = "FAILED"
else:
json_asff_status = "NOT_AVAILABLE"
return json_asff_status
def generate_json_asff_resource_tags(tags):
try:
resource_tags = {}
if tags and tags != [None]:
for tag in tags:
if "Key" in tag and "Value" in tag:
resource_tags[tag["Key"]] = tag["Value"]
else:
resource_tags.update(tag)
if len(resource_tags) == 0:
return None
else:
return None
return resource_tags
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def fill_json_asff(provider, finding):
"""
Fill the finding's output in JSON ASFF format.
Parameters:
- provider: The provider object containing information about the provider (e.g., AWS) and the output options object containing information about the desired output format.
- finding: The finding object containing information about the specific finding.
Returns:
- finding_output: The filled finding's output in JSON ASFF format.
"""
try:
# Check if there are no resources in the finding
if finding.resource_arn == "":
if finding.resource_id == "":
finding.resource_id = "NONE_PROVIDED"
finding.resource_arn = finding.resource_id
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
resource_tags = generate_json_asff_resource_tags(finding.resource_tags)
# Iterate for each compliance framework
compliance_summary = []
associated_standards = []
check_compliance = get_check_compliance(
finding, provider.type, provider.output_options
)
for key, value in check_compliance.items():
if (
len(associated_standards) < 20
): # AssociatedStandards should NOT have more than 20 items
associated_standards.append({"StandardsId": key})
item = f"{key} {' '.join(value)}"
if len(item) > 64:
item = item[0:63]
compliance_summary.append(item)
# Ensures finding_status matches allowed values in ASFF
finding_status = generate_json_asff_status(finding.status, finding.muted)
json_asff_output = Check_Output_JSON_ASFF(
# The following line cannot be changed because it is the format we use to generate unique findings for AWS Security Hub
# If changed some findings could be lost because the unique identifier will be different
# TODO: get this from the provider output
Id=f"prowler-{finding.check_metadata.CheckID}-{provider.identity.account}-{finding.region}-{hash_sha512(finding.resource_id)}",
ProductArn=f"arn:{provider.identity.partition}:securityhub:{finding.region}::product/prowler/prowler",
ProductFields=ProductFields(
ProwlerResourceName=finding.resource_arn,
),
GeneratorId="prowler-" + finding.check_metadata.CheckID,
AwsAccountId=provider.identity.account,
Types=finding.check_metadata.CheckType,
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.check_metadata.Severity.upper()),
Title=finding.check_metadata.CheckTitle,
Description=finding.status_extended,
Resources=[
Resource(
Id=finding.resource_arn,
Type=finding.check_metadata.ResourceType,
Partition=provider.identity.partition,
Region=finding.region,
Tags=resource_tags,
)
],
Compliance=Compliance(
Status=finding_status,
AssociatedStandards=associated_standards,
RelatedRequirements=compliance_summary,
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=finding.check_metadata.Remediation.Recommendation.Text,
Url=finding.check_metadata.Remediation.Recommendation.Url,
)
),
)
return json_asff_output
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
prowler/lib/outputs/json_asff/models.py
-83
View File
@@ -1,83 +0,0 @@
from typing import Optional
from pydantic import BaseModel, validator
from prowler.config.config import prowler_version
class ProductFields(BaseModel):
ProviderName: str = "Prowler"
ProviderVersion: str = prowler_version
ProwlerResourceName: str
class Severity(BaseModel):
Label: str
class Resource(BaseModel):
Type: str
Id: str
Partition: str
Region: str
Tags: Optional[dict]
class Compliance(BaseModel):
Status: str
RelatedRequirements: list[str]
AssociatedStandards: list[dict]
class Recommendation(BaseModel):
Text: str = ""
Url: str = ""
@validator("Text", pre=True, always=True)
def text_must_not_exceed_512_chars(text):
text_validated = text
if len(text) > 512:
text_validated = text[:509] + "..."
return text_validated
@validator("Url", pre=True, always=True)
def set_default_url_if_empty(url):
default_url = "https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html"
if url:
default_url = url
return default_url
class Remediation(BaseModel):
Recommendation: Recommendation
class Check_Output_JSON_ASFF(BaseModel):
"""
Check_Output_JSON_ASFF generates a finding's output in JSON ASFF format: https://docs.aws.amazon.com/securityhub/latest/userguide/securityhub-findings-format-syntax.html
"""
SchemaVersion: str = "2018-10-08"
Id: str
ProductArn: str
RecordState: str = "ACTIVE"
ProductFields: ProductFields
GeneratorId: str
AwsAccountId: str
Types: list[str] = None
FirstObservedAt: str
UpdatedAt: str
CreatedAt: str
Severity: Severity
Title: str
Description: str
Resources: list[Resource] = None
Compliance: Compliance
Remediation: Remediation
@validator("Description", pre=True, always=True)
def description_must_not_exceed_1024_chars(description):
description_validated = description
if len(description) > 1024:
description_validated = description[:1021] + "..."
return description_validated
prowler/lib/outputs/ocsf/ocsf.py
+3 -1
View File
@@ -171,7 +171,9 @@ class OCSF(Output):
if self._file_descriptor and not self._file_descriptor.closed:
self._file_descriptor.write("[")
for finding in self._data:
self._file_descriptor.write(finding.json(exclude_none=True))
self._file_descriptor.write(
finding.json(exclude_none=True, indent=4)
)
self._file_descriptor.write(",")
if self._file_descriptor.tell() > 0:
if self._file_descriptor.tell() != 1:
prowler/lib/outputs/output.py
+32 -1
View File
@@ -8,7 +8,23 @@ from prowler.lib.utils.utils import open_file
class Output(ABC):
_data: list[Finding]
"""
This class represents an abstract base class for defining different types of outputs for findings.
Attributes:
_data (list): A list to store transformed data from findings.
_file_descriptor (TextIOWrapper): A file descriptor to write data to a file.
Methods:
__init__: Initializes the Output class with findings, optionally creates a file descriptor.
data: Property to access the transformed data.
file_descriptor: Property to access the file descriptor.
transform: Abstract method to transform findings into a specific format.
batch_write_data_to_file: Abstract method to write data to a file in batches.
create_file_descriptor: Method to create a file descriptor for writing data to a file.
"""
_data: list
_file_descriptor: TextIOWrapper
def __init__(
@@ -39,6 +55,21 @@ class Output(ABC):
raise NotImplementedError
def create_file_descriptor(self, file_path) -> None:
"""
Creates a file descriptor for writing data to a file.
Parameters:
file_path (str): The path to the file where the data will be written.
Returns:
None
Raises:
Any exception that occurs during the file creation process will be caught and logged using the logger.
Note:
The file is opened in append mode ("a") to ensure data is written at the end of the file without overwriting existing content.
"""
try:
mode = "a"
self._file_descriptor = open_file(
prowler/lib/outputs/outputs.py
-16
View File
@@ -1,5 +1,3 @@
import json
from colorama import Fore, Style
from prowler.config.config import available_compliance_frameworks, orange_color
@@ -11,7 +9,6 @@ from prowler.lib.outputs.compliance.compliance import (
from prowler.lib.outputs.file_descriptors import fill_file_descriptors
from prowler.lib.outputs.finding import Finding
from prowler.lib.outputs.html.html import fill_html
from prowler.lib.outputs.json_asff.json_asff import fill_json_asff
def stdout_report(finding, color, verbose, status, fix):
@@ -95,19 +92,6 @@ def report(check_findings, provider):
input_compliance_frameworks,
)
# AWS specific outputs
if finding.check_metadata.Provider == "aws":
if "json-asff" in file_descriptors:
# Initialize this field using the class within fill_json_asff not here
json_asff_finding = fill_json_asff(provider, finding)
json.dump(
json_asff_finding.dict(exclude_none=True),
file_descriptors["json-asff"],
indent=4,
)
file_descriptors["json-asff"].write(",")
# Common Output Data
finding_output = Finding.generate_output(provider, finding)
prowler/providers/aws/lib/security_hub/security_hub.py
+84 -38
View File
@@ -3,52 +3,60 @@ from botocore.client import ClientError
from prowler.config.config import timestamp_utc
from prowler.lib.logger import logger
from prowler.lib.outputs.json_asff.json_asff import fill_json_asff
from prowler.lib.outputs.asff.asff import AWSSecurityFindingFormat
SECURITY_HUB_INTEGRATION_NAME = "prowler/prowler"
SECURITY_HUB_MAX_BATCH = 100
def prepare_security_hub_findings(
findings: list, provider, output_options, enabled_regions: list
def filter_security_hub_findings_per_region(
findings: list[AWSSecurityFindingFormat],
send_only_fails: bool,
status: list,
enabled_regions: list,
) -> dict:
security_hub_findings_per_region = {}
"""filter_security_hub_findings_per_region filters the findings by region and status. It returns a dictionary with the findings per region.
Args:
findings (list[AWSSecurityFindingFormat]): List of findings
send_only_fails (bool): Send only the findings that have failed
status (list): List of statuses to filter the findings
enabled_regions (list): List of enabled regions
Returns:
dict: Dictionary containing the findings per region
"""
security_hub_findings_per_region = {}
# Create a key per audited region
for region in enabled_regions:
security_hub_findings_per_region[region] = []
for finding in findings:
# We don't send the MANUAL findings to AWS Security Hub
if finding.status == "MANUAL":
continue
# We don't send findings to not enabled regions
if finding.region not in enabled_regions:
if finding.Resources[0].Region not in enabled_regions:
continue
if (
finding.status != "FAIL" or finding.muted
) and output_options.send_sh_only_fails:
finding.Compliance.Status != "FAILED"
or finding.Compliance.Status == "WARNING"
) and send_only_fails:
continue
if output_options.status:
if finding.status not in output_options.status:
# SecurityHub valid statuses are: PASSED, FAILED, WARNING
if status:
if finding.Compliance.Status == "PASSED" and "PASS" not in status:
continue
if finding.muted:
if finding.Compliance.Status == "FAILED" and "FAIL" not in status:
continue
# Check muted finding
if finding.Compliance.Status == "WARNING":
continue
# Get the finding region
region = finding.region
# We can do that since the finding always stores just one finding
region = finding.Resources[0].Region
# Format the finding in the JSON ASFF format
finding_json_asff = fill_json_asff(provider, finding)
# Include that finding within their region in the JSON format
security_hub_findings_per_region[region].append(
finding_json_asff.dict(exclude_none=True)
)
# Include that finding within their region
security_hub_findings_per_region[region].append(finding)
return security_hub_findings_per_region
@@ -59,6 +67,18 @@ def verify_security_hub_integration_enabled_per_region(
session: session.Session,
aws_account_number: str,
) -> bool:
"""
verify_security_hub_integration_enabled_per_region returns True if the Prowler integration is enabled for the given region. Otherwise returns false.
Args:
partition (str): AWS partition
region (str): AWS region
session (session.Session): AWS session
aws_account_number (str): AWS account number
Returns:
bool: True if the Prowler integration is enabled for the given region. Otherwise returns false.
"""
f"""verify_security_hub_integration_enabled returns True if the {SECURITY_HUB_INTEGRATION_NAME} is enabled for the given region. Otherwise returns false."""
prowler_integration_enabled = False
@@ -112,7 +132,14 @@ def batch_send_to_security_hub(
session: session.Session,
) -> int:
"""
send_to_security_hub sends findings to Security Hub and returns the number of findings that were successfully sent.
batch_send_to_security_hub sends findings to Security Hub and returns the number of findings that were successfully sent.
Args:
security_hub_findings_per_region (dict): Dictionary containing the findings per region
session (session.Session): AWS session
Returns:
int: Number of sent findings
"""
success_count = 0
@@ -120,11 +147,14 @@ def batch_send_to_security_hub(
# Iterate findings by region
for region, findings in security_hub_findings_per_region.items():
# Send findings to Security Hub
logger.info(f"Sending findings to Security Hub in the region {region}")
logger.info(
f"Sending {len(findings)} findings to Security Hub in the region {region}"
)
security_hub_client = session.client("securityhub", region_name=region)
success_count = __send_findings_to_security_hub__(
# Convert findings to dict
findings = [finding.dict(exclude_none=True) for finding in findings]
success_count += _send_findings_to_security_hub(
findings, region, security_hub_client
)
@@ -138,9 +168,16 @@ def batch_send_to_security_hub(
# Move previous Security Hub check findings to ARCHIVED (as prowler didn't re-detect them)
def resolve_security_hub_previous_findings(
security_hub_findings_per_region: dict, provider
) -> list:
) -> int:
"""
resolve_security_hub_previous_findings archives all the findings that does not appear in the current execution
Args:
security_hub_findings_per_region (dict): Dictionary containing the findings per region
provider: Provider object
Returns:
int: Number of archived findings
"""
logger.info("Checking previous findings in Security Hub to archive them.")
success_count = 0
@@ -150,7 +187,7 @@ def resolve_security_hub_previous_findings(
# Get current findings IDs
current_findings_ids = []
for finding in current_findings:
current_findings_ids.append(finding["Id"])
current_findings_ids.append(finding.Id)
# Get findings of that region
security_hub_client = provider.session.current_session.client(
"securityhub", region_name=region
@@ -165,7 +202,9 @@ def resolve_security_hub_previous_findings(
}
get_findings_paginator = security_hub_client.get_paginator("get_findings")
findings_to_archive = []
for page in get_findings_paginator.paginate(Filters=findings_filter):
for page in get_findings_paginator.paginate(
Filters=findings_filter, PaginationConfig={"PageSize": 100}
):
# Archive findings that have not appear in this execution
for finding in page["Findings"]:
if finding["Id"] not in current_findings_ids:
@@ -178,7 +217,7 @@ def resolve_security_hub_previous_findings(
logger.info(f"Archiving {len(findings_to_archive)} findings.")
# Send archive findings to SHub
success_count += __send_findings_to_security_hub__(
success_count += _send_findings_to_security_hub(
findings_to_archive, region, security_hub_client
)
except Exception as error:
@@ -188,17 +227,25 @@ def resolve_security_hub_previous_findings(
return success_count
def __send_findings_to_security_hub__(
def _send_findings_to_security_hub(
findings: list[dict], region: str, security_hub_client
):
"""Private function send_findings_to_security_hub chunks the findings in groups of 100 findings and send them to AWS Security Hub. It returns the number of sent findings."""
) -> int:
"""Private function send_findings_to_security_hub chunks the findings in groups of 100 findings and send them to AWS Security Hub. It returns the number of sent findings.
Args:
findings (list[dict]): List of findings to send to AWS Security Hub
region (str): AWS region to send the findings
security_hub_client: AWS Security Hub client
Returns:
int: Number of sent findings
"""
success_count = 0
try:
list_chunked = [
findings[i : i + SECURITY_HUB_MAX_BATCH]
for i in range(0, len(findings), SECURITY_HUB_MAX_BATCH)
]
for findings in list_chunked:
batch_import = security_hub_client.batch_import_findings(Findings=findings)
if batch_import["FailedCount"] > 0:
@@ -207,10 +254,9 @@ def __send_findings_to_security_hub__(
f"Failed to send findings to AWS Security Hub -- {failed_import['ErrorCode']} -- {failed_import['ErrorMessage']}"
)
success_count += batch_import["SuccessCount"]
return success_count
except Exception as error:
logger.error(
f"{error.__class__.__name__} -- [{error.__traceback__.tb_lineno}]:{error} in region {region}"
)
finally:
return success_count
tests/lib/outputs/asff/asff_test.py
+448
View File
@@ -0,0 +1,448 @@
from os import path
from prowler.config.config import prowler_version, timestamp_utc
from prowler.lib.outputs.asff.asff import (
ASFF,
AWSSecurityFindingFormat,
Compliance,
ProductFields,
Recommendation,
Remediation,
Resource,
Severity,
)
from prowler.lib.utils.utils import hash_sha512
from tests.lib.outputs.fixtures.fixtures import generate_finding_output
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_COMMERCIAL_PARTITION,
AWS_REGION_EU_WEST_1,
)
METADATA_FIXTURE_PATH = (
f"{path.dirname(path.realpath(__file__))}/../fixtures/metadata.json"
)
class TestASFF:
def test_asff(self):
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags="key1=value1",
)
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
associated_standards, compliance_summary = ASFF.format_compliance(
finding.compliance
)
expected = AWSSecurityFindingFormat(
Id=f"prowler-{finding.check_id}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}",
ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version,
ProwlerResourceName=finding.resource_uid,
),
GeneratorId="prowler-" + finding.check_id,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.check_type.split(","),
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.severity),
Title=finding.check_title,
Resources=[
Resource(
Id=finding.resource_uid,
Type=finding.resource_type,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags=ASFF.format_resource_tags(finding.resource_tags),
)
],
Compliance=Compliance(
Status=ASFF.generate_status(status),
RelatedRequirements=compliance_summary,
AssociatedStandards=associated_standards,
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=finding.remediation_recommendation_text,
Url=finding.remediation_recommendation_url,
)
),
Description=finding.description,
)
asff = ASFF(findings=[finding])
assert len(asff.data) == 1
asff_finding = asff.data[0]
assert asff_finding == expected
def test_asff_without_remediation_recommendation_url(self):
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags="key1=value1",
)
finding.remediation_recommendation_url = ""
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
associated_standards, compliance_summary = ASFF.format_compliance(
finding.compliance
)
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
expected = AWSSecurityFindingFormat(
Id=f"prowler-{finding.check_id}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}",
ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version,
ProwlerResourceName=finding.resource_uid,
),
GeneratorId="prowler-" + finding.check_id,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.check_type.split(","),
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.severity),
Title=finding.check_title,
Resources=[
Resource(
Id=finding.resource_uid,
Type=finding.resource_type,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags=ASFF.format_resource_tags(finding.resource_tags),
)
],
Compliance=Compliance(
Status=ASFF.generate_status(status),
RelatedRequirements=compliance_summary,
AssociatedStandards=associated_standards,
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=finding.remediation_recommendation_text,
Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html",
)
),
Description=finding.description,
)
asff = ASFF(findings=[finding])
assert len(asff.data) == 1
asff_finding = asff.data[0]
assert asff_finding == expected
def test_asff_with_long_description_and_remediation_recommendation_text(
self,
):
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags="key1=value1",
)
finding.remediation_recommendation_url = ""
finding.remediation_recommendation_text = "x" * 513
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
associated_standards, compliance_summary = ASFF.format_compliance(
finding.compliance
)
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
expected = AWSSecurityFindingFormat(
Id=f"prowler-{finding.check_id}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}",
ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version,
ProwlerResourceName=finding.resource_uid,
),
GeneratorId="prowler-" + finding.check_id,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.check_type.split(","),
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.severity),
Title=finding.check_title,
Resources=[
Resource(
Id=finding.resource_uid,
Type=finding.resource_type,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags=ASFF.format_resource_tags(finding.resource_tags),
)
],
Compliance=Compliance(
Status=ASFF.generate_status(status),
RelatedRequirements=compliance_summary,
AssociatedStandards=associated_standards,
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=f"{'x' * 509}...",
Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html",
)
),
Description=finding.description,
)
asff = ASFF(findings=[finding])
assert len(asff.data) == 1
asff_finding = asff.data[0]
assert asff_finding == expected
def test_fill_json_asff_with_long_associated_standards(self):
status = "PASS"
finding = generate_finding_output(
status=status,
status_extended="This is a test",
region=AWS_REGION_EU_WEST_1,
resource_details="Test resource details",
resource_name="test-resource",
resource_uid="test-arn",
resource_tags="key1=value1",
compliance={
"CISA": ["your-systems-3", "your-data-2"],
"SOC2": ["cc_2_1", "cc_7_2", "cc_a_1_2"],
"CIS-1.4": ["3.1"],
"CIS-1.5": ["3.1"],
"GDPR": ["article_25", "article_30"],
"AWS-Foundational-Security-Best-Practices": ["cloudtrail"],
"HIPAA": [
"164_308_a_1_ii_d",
"164_308_a_3_ii_a",
"164_308_a_6_ii",
"164_312_b",
"164_312_e_2_i",
],
"ISO27001": ["A.12.4"],
"GxP-21-CFR-Part-11": ["11.10-e", "11.10-k", "11.300-d"],
"AWS-Well-Architected-Framework-Security-Pillar": [
"SEC04-BP02",
"SEC04-BP03",
],
"GxP-EU-Annex-11": [
"1-risk-management",
"4.2-validation-documentation-change-control",
],
"NIST-800-171-Revision-2": [
"3_1_12",
"3_3_1",
"3_3_2",
"3_3_3",
"3_4_1",
"3_6_1",
"3_6_2",
"3_13_1",
"3_13_2",
"3_14_6",
"3_14_7",
],
"NIST-800-53-Revision-4": [
"ac_2_4",
"ac_2",
"au_2",
"au_3",
"au_12",
"cm_2",
],
"NIST-800-53-Revision-5": [
"ac_2_4",
"ac_3_1",
"ac_3_10",
"ac_4_26",
"ac_6_9",
"au_2_b",
"au_3_1",
"au_3_a",
"au_3_b",
"au_3_c",
"au_3_d",
"au_3_e",
"au_3_f",
"au_6_3",
"au_6_4",
"au_6_6",
"au_6_9",
"au_8_b",
"au_10",
"au_12_a",
"au_12_c",
"au_12_1",
"au_12_2",
"au_12_3",
"au_12_4",
"au_14_a",
"au_14_b",
"au_14_3",
"ca_7_b",
"cm_5_1_b",
"cm_6_a",
"cm_9_b",
"ia_3_3_b",
"ma_4_1_a",
"pm_14_a_1",
"pm_14_b",
"pm_31",
"sc_7_9_b",
"si_1_1_c",
"si_3_8_b",
"si_4_2",
"si_4_17",
"si_4_20",
"si_7_8",
"si_10_1_c",
],
"ENS-RD2022": [
"op.acc.6.r5.aws.iam.1",
"op.exp.5.aws.ct.1",
"op.exp.8.aws.ct.1",
"op.exp.8.aws.ct.6",
"op.exp.9.aws.ct.1",
"op.mon.1.aws.ct.1",
],
"NIST-CSF-1.1": [
"ae_1",
"ae_3",
"ae_4",
"cm_1",
"cm_3",
"cm_6",
"cm_7",
"am_3",
"ac_6",
"ds_5",
"ma_2",
"pt_1",
],
"RBI-Cyber-Security-Framework": ["annex_i_7_4"],
"FFIEC": [
"d2-ma-ma-b-1",
"d2-ma-ma-b-2",
"d3-dc-an-b-3",
"d3-dc-an-b-4",
"d3-dc-an-b-5",
"d3-dc-ev-b-1",
"d3-dc-ev-b-3",
"d3-pc-im-b-3",
"d3-pc-im-b-7",
"d5-dr-de-b-3",
],
"PCI-3.2.1": ["cloudtrail"],
"FedRamp-Moderate-Revision-4": [
"ac-2-4",
"ac-2-g",
"au-2-a-d",
"au-3",
"au-6-1-3",
"au-12-a-c",
"ca-7-a-b",
"si-4-16",
"si-4-2",
"si-4-4",
"si-4-5",
],
"FedRAMP-Low-Revision-4": ["ac-2", "au-2", "ca-7"],
},
)
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
associated_standards, compliance_summary = ASFF.format_compliance(
finding.compliance
)
expected = AWSSecurityFindingFormat(
Id=f"prowler-{finding.check_id}-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-{hash_sha512(finding.resource_uid)}",
ProductArn=f"arn:{AWS_COMMERCIAL_PARTITION}:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version,
ProwlerResourceName=finding.resource_uid,
),
GeneratorId="prowler-" + finding.check_id,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.check_type.split(","),
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.severity),
Title=finding.check_title,
Resources=[
Resource(
Id=finding.resource_uid,
Type=finding.resource_type,
Partition=AWS_COMMERCIAL_PARTITION,
Region=AWS_REGION_EU_WEST_1,
Tags=ASFF.format_resource_tags(finding.resource_tags),
)
],
Compliance=Compliance(
Status=ASFF.generate_status(status),
RelatedRequirements=compliance_summary,
AssociatedStandards=associated_standards,
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=finding.remediation_recommendation_text,
Url=finding.remediation_recommendation_url,
)
),
Description=finding.description,
)
asff = ASFF(findings=[finding])
assert len(asff.data) == 1
asff_finding = asff.data[0]
assert asff_finding == expected
def test_asff_generate_status(self):
assert ASFF.generate_status("PASS") == "PASSED"
assert ASFF.generate_status("FAIL") == "FAILED"
assert ASFF.generate_status("FAIL", True) == "WARNING"
assert ASFF.generate_status("SOMETHING ELSE") == "NOT_AVAILABLE"
def test_asff_format_resource_tags(self):
assert ASFF.format_resource_tags(None) is None
assert ASFF.format_resource_tags("") is None
assert ASFF.format_resource_tags([]) is None
assert ASFF.format_resource_tags([{}]) is None
assert ASFF.format_resource_tags("key1=value1") == {"key1": "value1"}
assert ASFF.format_resource_tags("key1=value1 | key2=value2") == {
"key1": "value1",
"key2": "value2",
}
tests/lib/outputs/csv/csv_test.py
+3 -2
View File
@@ -5,6 +5,7 @@ from typing import List
from unittest.mock import MagicMock
import pytest
from mock import patch
from prowler.lib.outputs.csv.csv import write_csv
from prowler.lib.outputs.csv.models import CSV
@@ -60,7 +61,6 @@ def generate_finding():
class TestCSV:
def test_output_transform(self, generate_finding):
findings = [generate_finding]
@@ -131,7 +131,8 @@ class TestCSV:
output = CSV(findings)
output._file_descriptor = mock_file
output.batch_write_data_to_file()
with patch.object(mock_file, "close", return_value=None):
output.batch_write_data_to_file()
mock_file.seek(0)
content = mock_file.read()
tests/lib/outputs/fixtures/fixtures.py
+18 -15
View File
@@ -2,20 +2,23 @@ from datetime import datetime
from prowler.config.config import prowler_version
from prowler.lib.outputs.finding import Finding
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1
# TODO: customize it per provider
def generate_finding_output(
status,
severity,
muted,
region,
status: str = "PASS",
status_extended: str = "",
severity: str = "high",
muted: bool = False,
region: str = AWS_REGION_EU_WEST_1,
resource_details: str = "",
resource_uid: str = "",
resource_name: str = "",
resource_tags: str = "",
compliance: dict = {"test-compliance": "test-compliance"},
timestamp: datetime = datetime.now(),
provider: str = "aws",
) -> Finding:
# TODO: Include metadata from a valid file
return Finding(
auth_method="profile: default",
timestamp=timestamp,
@@ -31,17 +34,17 @@ def generate_finding_output(
check_title="test-check-id",
check_type="test-type",
status=status,
status_extended="status extended",
status_extended=status_extended,
muted=muted,
service_name="test-service",
subservice_name="",
severity=severity,
resource_type="test-resource",
resource_uid="resource-id",
resource_name="resource_name",
resource_details="resource_details",
resource_tags="",
partition="aws",
resource_uid=resource_uid,
resource_name=resource_name,
resource_details=resource_details,
resource_tags=resource_tags,
partition=provider,
region=region,
description="check description",
risk="test-risk",
@@ -52,7 +55,7 @@ def generate_finding_output(
remediation_code_terraform="",
remediation_code_cli="",
remediation_code_other="",
compliance={"test-compliance": "test-compliance"},
compliance=compliance,
categories="test-category",
depends_on="test-dependency",
related_to="test-related-to",
tests/lib/outputs/json_asff/json_asff_test.py
-472
View File
@@ -1,472 +0,0 @@
from os import path
import mock
from prowler.config.config import prowler_version, timestamp_utc
from prowler.lib.check.models import Check_Report, load_check_metadata
from prowler.lib.outputs.json_asff.json_asff import (
fill_json_asff,
generate_json_asff_resource_tags,
generate_json_asff_status,
)
from prowler.lib.outputs.json_asff.models import (
Check_Output_JSON_ASFF,
Compliance,
ProductFields,
Recommendation,
Remediation,
Resource,
Severity,
)
from prowler.lib.utils.utils import hash_sha512
from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, set_mocked_aws_provider
METADATA_FIXTURE_PATH = (
f"{path.dirname(path.realpath(__file__))}/../fixtures/metadata.json"
)
class TestOutputJSONASFF:
def test_fill_json_asff(self):
aws_provider = set_mocked_aws_provider()
finding = Check_Report(load_check_metadata(METADATA_FIXTURE_PATH).json())
finding.resource_details = "Test resource details"
finding.resource_id = "test-resource"
finding.resource_arn = "test-arn"
finding.region = "eu-west-1"
finding.status = "PASS"
finding.status_extended = "This is a test"
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
expected = Check_Output_JSON_ASFF(
Id=f"prowler-{finding.check_metadata.CheckID}-123456789012-eu-west-1-{hash_sha512('test-resource')}",
ProductArn="arn:aws:securityhub:eu-west-1::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version, ProwlerResourceName="test-arn"
),
GeneratorId="prowler-" + finding.check_metadata.CheckID,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.check_metadata.CheckType,
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.check_metadata.Severity.upper()),
Title=finding.check_metadata.CheckTitle,
Description=finding.status_extended,
Resources=[
Resource(
Id="test-arn",
Type=finding.check_metadata.ResourceType,
Partition="aws",
Region="eu-west-1",
)
],
Compliance=Compliance(
Status="PASS" + "ED",
RelatedRequirements=[],
AssociatedStandards=[],
),
Remediation={
"Recommendation": finding.check_metadata.Remediation.Recommendation
},
)
assert fill_json_asff(aws_provider, finding) == expected
def test_fill_json_asff_without_remediation_recommendation_url(self):
aws_provider = set_mocked_aws_provider()
finding = Check_Report(load_check_metadata(METADATA_FIXTURE_PATH).json())
# Empty the Remediation.Recomendation.URL
finding.check_metadata.Remediation.Recommendation.Url = ""
finding.resource_details = "Test resource details"
finding.resource_id = "test-resource"
finding.resource_arn = "test-arn"
finding.region = "eu-west-1"
finding.status = "PASS"
finding.status_extended = "This is a test"
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
expected = Check_Output_JSON_ASFF(
Id=f"prowler-{finding.check_metadata.CheckID}-123456789012-eu-west-1-{hash_sha512('test-resource')}",
ProductArn="arn:aws:securityhub:eu-west-1::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version, ProwlerResourceName="test-arn"
),
GeneratorId="prowler-" + finding.check_metadata.CheckID,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.check_metadata.CheckType,
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.check_metadata.Severity.upper()),
Title=finding.check_metadata.CheckTitle,
Description=finding.status_extended,
Resources=[
Resource(
Id="test-arn",
Type=finding.check_metadata.ResourceType,
Partition="aws",
Region="eu-west-1",
)
],
Compliance=Compliance(
Status="PASS" + "ED",
RelatedRequirements=[],
AssociatedStandards=[],
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=finding.check_metadata.Remediation.Recommendation.Text,
Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html",
)
),
)
assert fill_json_asff(aws_provider, finding) == expected
def test_fill_json_asff_with_long_description_and_remediation_recommendation_text(
self,
):
aws_provider = set_mocked_aws_provider()
finding = Check_Report(load_check_metadata(METADATA_FIXTURE_PATH).json())
# Empty the Remediation.Recomendation.URL
finding.check_metadata.Remediation.Recommendation.Url = ""
finding.check_metadata.Remediation.Recommendation.Text = "x" * 513
finding.resource_details = "Test resource details"
finding.resource_id = "test-resource"
finding.resource_arn = "test-arn"
finding.region = "eu-west-1"
finding.status = "PASS"
finding.status_extended = "x" * 2000 # it has to be limited to 1000+...
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
expected = Check_Output_JSON_ASFF(
Id=f"prowler-{finding.check_metadata.CheckID}-123456789012-eu-west-1-{hash_sha512('test-resource')}",
ProductArn="arn:aws:securityhub:eu-west-1::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version, ProwlerResourceName="test-arn"
),
GeneratorId="prowler-" + finding.check_metadata.CheckID,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.check_metadata.CheckType,
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.check_metadata.Severity.upper()),
Title=finding.check_metadata.CheckTitle,
Description=f"{finding.status_extended[:1021]}...",
Resources=[
Resource(
Id="test-arn",
Type=finding.check_metadata.ResourceType,
Partition="aws",
Region="eu-west-1",
)
],
Compliance=Compliance(
Status="PASS" + "ED",
RelatedRequirements=[],
AssociatedStandards=[],
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=f"{'x' * 509}...",
Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html",
)
),
)
output_json_asff = fill_json_asff(aws_provider, finding)
assert isinstance(output_json_asff, Check_Output_JSON_ASFF)
assert output_json_asff.Id == expected.Id
assert output_json_asff.ProductArn == expected.ProductArn
assert output_json_asff.ProductFields == expected.ProductFields
assert output_json_asff.GeneratorId == expected.GeneratorId
assert output_json_asff.AwsAccountId == expected.AwsAccountId
assert output_json_asff.Types == expected.Types
assert output_json_asff.FirstObservedAt == expected.FirstObservedAt
assert output_json_asff.UpdatedAt == expected.UpdatedAt
assert output_json_asff.CreatedAt == expected.CreatedAt
assert output_json_asff.Severity == expected.Severity
assert output_json_asff.Title == expected.Title
assert output_json_asff.Description == expected.Description
assert output_json_asff.Resources == expected.Resources
assert output_json_asff.Compliance == expected.Compliance
assert isinstance(output_json_asff.Remediation, Remediation)
assert isinstance(output_json_asff.Remediation.Recommendation, Recommendation)
assert (
output_json_asff.Remediation.Recommendation.Text
== expected.Remediation.Recommendation.Text
)
assert (
output_json_asff.Remediation.Recommendation.Url
== expected.Remediation.Recommendation.Url
)
def test_fill_json_asff_with_long_associated_standards(self):
aws_provider = set_mocked_aws_provider()
with mock.patch(
"prowler.lib.outputs.json_asff.json_asff.get_check_compliance",
return_value={
"CISA": ["your-systems-3", "your-data-2"],
"SOC2": ["cc_2_1", "cc_7_2", "cc_a_1_2"],
"CIS-1.4": ["3.1"],
"CIS-1.5": ["3.1"],
"GDPR": ["article_25", "article_30"],
"AWS-Foundational-Security-Best-Practices": ["cloudtrail"],
"HIPAA": [
"164_308_a_1_ii_d",
"164_308_a_3_ii_a",
"164_308_a_6_ii",
"164_312_b",
"164_312_e_2_i",
],
"ISO27001": ["A.12.4"],
"GxP-21-CFR-Part-11": ["11.10-e", "11.10-k", "11.300-d"],
"AWS-Well-Architected-Framework-Security-Pillar": [
"SEC04-BP02",
"SEC04-BP03",
],
"GxP-EU-Annex-11": [
"1-risk-management",
"4.2-validation-documentation-change-control",
],
"NIST-800-171-Revision-2": [
"3_1_12",
"3_3_1",
"3_3_2",
"3_3_3",
"3_4_1",
"3_6_1",
"3_6_2",
"3_13_1",
"3_13_2",
"3_14_6",
"3_14_7",
],
"NIST-800-53-Revision-4": [
"ac_2_4",
"ac_2",
"au_2",
"au_3",
"au_12",
"cm_2",
],
"NIST-800-53-Revision-5": [
"ac_2_4",
"ac_3_1",
"ac_3_10",
"ac_4_26",
"ac_6_9",
"au_2_b",
"au_3_1",
"au_3_a",
"au_3_b",
"au_3_c",
"au_3_d",
"au_3_e",
"au_3_f",
"au_6_3",
"au_6_4",
"au_6_6",
"au_6_9",
"au_8_b",
"au_10",
"au_12_a",
"au_12_c",
"au_12_1",
"au_12_2",
"au_12_3",
"au_12_4",
"au_14_a",
"au_14_b",
"au_14_3",
"ca_7_b",
"cm_5_1_b",
"cm_6_a",
"cm_9_b",
"ia_3_3_b",
"ma_4_1_a",
"pm_14_a_1",
"pm_14_b",
"pm_31",
"sc_7_9_b",
"si_1_1_c",
"si_3_8_b",
"si_4_2",
"si_4_17",
"si_4_20",
"si_7_8",
"si_10_1_c",
],
"ENS-RD2022": [
"op.acc.6.r5.aws.iam.1",
"op.exp.5.aws.ct.1",
"op.exp.8.aws.ct.1",
"op.exp.8.aws.ct.6",
"op.exp.9.aws.ct.1",
"op.mon.1.aws.ct.1",
],
"NIST-CSF-1.1": [
"ae_1",
"ae_3",
"ae_4",
"cm_1",
"cm_3",
"cm_6",
"cm_7",
"am_3",
"ac_6",
"ds_5",
"ma_2",
"pt_1",
],
"RBI-Cyber-Security-Framework": ["annex_i_7_4"],
"FFIEC": [
"d2-ma-ma-b-1",
"d2-ma-ma-b-2",
"d3-dc-an-b-3",
"d3-dc-an-b-4",
"d3-dc-an-b-5",
"d3-dc-ev-b-1",
"d3-dc-ev-b-3",
"d3-pc-im-b-3",
"d3-pc-im-b-7",
"d5-dr-de-b-3",
],
"PCI-3.2.1": ["cloudtrail"],
"FedRamp-Moderate-Revision-4": [
"ac-2-4",
"ac-2-g",
"au-2-a-d",
"au-3",
"au-6-1-3",
"au-12-a-c",
"ca-7-a-b",
"si-4-16",
"si-4-2",
"si-4-4",
"si-4-5",
],
"FedRAMP-Low-Revision-4": ["ac-2", "au-2", "ca-7"],
},
):
finding = Check_Report(load_check_metadata(METADATA_FIXTURE_PATH).json())
# Empty the Remediation.Recomendation.URL
finding.check_metadata.Remediation.Recommendation.Url = ""
finding.resource_details = "Test resource details"
finding.resource_id = "test-resource"
finding.resource_arn = "test-arn"
finding.region = "eu-west-1"
finding.status = "PASS"
finding.status_extended = "This is a test"
timestamp = timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
expected = Check_Output_JSON_ASFF(
Id=f"prowler-{finding.check_metadata.CheckID}-123456789012-eu-west-1-{hash_sha512('test-resource')}",
ProductArn="arn:aws:securityhub:eu-west-1::product/prowler/prowler",
ProductFields=ProductFields(
ProviderVersion=prowler_version, ProwlerResourceName="test-arn"
),
GeneratorId="prowler-" + finding.check_metadata.CheckID,
AwsAccountId=AWS_ACCOUNT_NUMBER,
Types=finding.check_metadata.CheckType,
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
Severity=Severity(Label=finding.check_metadata.Severity.upper()),
Title=finding.check_metadata.CheckTitle,
Description=finding.status_extended,
Resources=[
Resource(
Id="test-arn",
Type=finding.check_metadata.ResourceType,
Partition="aws",
Region="eu-west-1",
)
],
Compliance=Compliance(
Status="PASS" + "ED",
RelatedRequirements=[
"CISA your-systems-3 your-data-2",
"SOC2 cc_2_1 cc_7_2 cc_a_1_2",
"CIS-1.4 3.1",
"CIS-1.5 3.1",
"GDPR article_25 article_30",
"AWS-Foundational-Security-Best-Practices cloudtrail",
"HIPAA 164_308_a_1_ii_d 164_308_a_3_ii_a 164_308_a_6_ii 164_312_",
"ISO27001 A.12.4",
"GxP-21-CFR-Part-11 11.10-e 11.10-k 11.300-d",
"AWS-Well-Architected-Framework-Security-Pillar SEC04-BP02 SEC04",
"GxP-EU-Annex-11 1-risk-management 4.2-validation-documentation-",
"NIST-800-171-Revision-2 3_1_12 3_3_1 3_3_2 3_3_3 3_4_1 3_6_1 3_",
"NIST-800-53-Revision-4 ac_2_4 ac_2 au_2 au_3 au_12 cm_2",
"NIST-800-53-Revision-5 ac_2_4 ac_3_1 ac_3_10 ac_4_26 ac_6_9 au_",
"ENS-RD2022 op.acc.6.r5.aws.iam.1 op.exp.5.aws.ct.1 op.exp.8.aws",
"NIST-CSF-1.1 ae_1 ae_3 ae_4 cm_1 cm_3 cm_6 cm_7 am_3 ac_6 ds_5 ",
"RBI-Cyber-Security-Framework annex_i_7_4",
"FFIEC d2-ma-ma-b-1 d2-ma-ma-b-2 d3-dc-an-b-3 d3-dc-an-b-4 d3-dc",
"PCI-3.2.1 cloudtrail",
"FedRamp-Moderate-Revision-4 ac-2-4 ac-2-g au-2-a-d au-3 au-6-1-",
],
AssociatedStandards=[
{"StandardsId": "CISA"},
{"StandardsId": "SOC2"},
{"StandardsId": "CIS-1.4"},
{"StandardsId": "CIS-1.5"},
{"StandardsId": "GDPR"},
{"StandardsId": "AWS-Foundational-Security-Best-Practices"},
{"StandardsId": "HIPAA"},
{"StandardsId": "ISO27001"},
{"StandardsId": "GxP-21-CFR-Part-11"},
{
"StandardsId": "AWS-Well-Architected-Framework-Security-Pillar"
},
{"StandardsId": "GxP-EU-Annex-11"},
{"StandardsId": "NIST-800-171-Revision-2"},
{"StandardsId": "NIST-800-53-Revision-4"},
{"StandardsId": "NIST-800-53-Revision-5"},
{"StandardsId": "ENS-RD2022"},
{"StandardsId": "NIST-CSF-1.1"},
{"StandardsId": "RBI-Cyber-Security-Framework"},
{"StandardsId": "FFIEC"},
{"StandardsId": "PCI-3.2.1"},
{"StandardsId": "FedRamp-Moderate-Revision-4"},
],
),
Remediation=Remediation(
Recommendation=Recommendation(
Text=finding.check_metadata.Remediation.Recommendation.Text,
Url="https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html",
)
),
)
assert fill_json_asff(aws_provider, finding) == expected
def test_generate_json_asff_status(self):
assert generate_json_asff_status("PASS") == "PASSED"
assert generate_json_asff_status("FAIL") == "FAILED"
assert generate_json_asff_status("FAIL", True) == "WARNING"
assert generate_json_asff_status("SOMETHING ELSE") == "NOT_AVAILABLE"
def test_generate_json_asff_resource_tags(self):
assert generate_json_asff_resource_tags(None) is None
assert generate_json_asff_resource_tags([]) is None
assert generate_json_asff_resource_tags([{}]) is None
assert generate_json_asff_resource_tags([{"key1": "value1"}]) == {
"key1": "value1"
}
assert generate_json_asff_resource_tags(
[{"Key": "key1", "Value": "value1"}]
) == {"key1": "value1"}
tests/lib/outputs/ocsf/ocsf_test.py
+24 -6
View File
@@ -100,7 +100,11 @@ expected_json_output = json.dumps(
class TestOCSF:
def test_transform(self):
findings = [generate_finding_output("FAIL", "low", False, AWS_REGION_EU_WEST_1)]
findings = [
generate_finding_output(
status="FAIL", severity="low", muted=False, region=AWS_REGION_EU_WEST_1
)
]
ocsf = OCSF(findings)
@@ -110,7 +114,17 @@ class TestOCSF:
def test_batch_write_data_to_file(self):
mock_file = StringIO()
findings = [
generate_finding_output("FAIL", "low", False, AWS_REGION_EU_WEST_1, now)
generate_finding_output(
status="FAIL",
severity="low",
muted=False,
region=AWS_REGION_EU_WEST_1,
timestamp=now,
resource_details="resource_details",
resource_name="resource_name",
resource_uid="resource-id",
status_extended="status extended",
)
]
output = OCSF(findings)
@@ -126,7 +140,7 @@ class TestOCSF:
def test_finding_output_cloud_pass_low_muted(self):
finding_output = generate_finding_output(
"PASS", "low", True, AWS_REGION_EU_WEST_1
status="PASS", severity="low", muted=True, region=AWS_REGION_EU_WEST_1
)
finding_ocsf = OCSF([finding_output])
@@ -232,7 +246,11 @@ class TestOCSF:
def test_finding_output_kubernetes(self):
finding_output = generate_finding_output(
"PASS", "low", True, AWS_REGION_EU_WEST_1, provider="kubernetes"
status="PASS",
severity="low",
muted=True,
region=AWS_REGION_EU_WEST_1,
provider="kubernetes",
)
finding_ocsf = OCSF([finding_output])
@@ -243,7 +261,7 @@ class TestOCSF:
def test_finding_output_cloud_fail_low_not_muted(self):
finding_output = generate_finding_output(
"FAIL", "low", False, AWS_REGION_EU_WEST_1
status="FAIL", severity="low", muted=False, region=AWS_REGION_EU_WEST_1
)
finding_ocsf = OCSF([finding_output])
@@ -257,7 +275,7 @@ class TestOCSF:
def test_finding_output_cloud_pass_low_not_muted(self):
finding_output = generate_finding_output(
"PASS", "low", False, AWS_REGION_EU_WEST_1
status="PASS", severity="low", muted=False, region=AWS_REGION_EU_WEST_1
)
finding_ocsf = OCSF([finding_output])
tests/lib/outputs/outputs_test.py
+1 -23
View File
@@ -5,11 +5,7 @@ from unittest import mock
import pytest
from colorama import Fore
from prowler.config.config import (
html_file_suffix,
json_asff_file_suffix,
output_file_timestamp,
)
from prowler.config.config import html_file_suffix, output_file_timestamp
from prowler.lib.check.compliance_models import (
CIS_Requirement_Attribute,
Compliance_Base_Model,
@@ -39,34 +35,16 @@ class TestOutputs:
output_directory = f"{os.path.dirname(os.path.realpath(__file__))}"
aws_provider = set_mocked_aws_provider()
test_output_modes = [
["json-asff"],
["html"],
["json-asff", "html"],
]
output_filename = f"prowler-output-{audited_account}-{output_file_timestamp}"
expected = [
{
"json-asff": open_file(
f"{output_directory}/{output_filename}{json_asff_file_suffix}",
"a",
)
},
{
"html": open_file(
f"{output_directory}/{output_filename}{html_file_suffix}",
"a",
)
},
{
"json-asff": open_file(
f"{output_directory}/{output_filename}{json_asff_file_suffix}",
"a",
),
"html": open_file(
f"{output_directory}/{output_filename}{html_file_suffix}",
"a",
),
},
]
for index, output_mode_list in enumerate(test_output_modes):
tests/providers/aws/lib/security_hub/security_hub_test.py
+140 -234
View File
@@ -1,69 +1,24 @@
from logging import ERROR, WARNING
from os import path
import botocore
from boto3 import session
from botocore.client import ClientError
from mock import MagicMock, patch
from mock import patch
from prowler.config.config import prowler_version, timestamp_utc
from prowler.lib.check.models import Check_Report, load_check_metadata
from prowler.lib.outputs.asff.asff import ASFF
from prowler.providers.aws.lib.security_hub.security_hub import (
batch_send_to_security_hub,
prepare_security_hub_findings,
filter_security_hub_findings_per_region,
verify_security_hub_integration_enabled_per_region,
)
from tests.lib.outputs.fixtures.fixtures import generate_finding_output
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_COMMERCIAL_PARTITION,
AWS_REGION_EU_WEST_1,
AWS_REGION_EU_WEST_2,
set_mocked_aws_provider,
)
def get_security_hub_finding(status: str):
return {
"SchemaVersion": "2018-10-08",
"Id": f"prowler-iam_user_accesskey_unused-{AWS_ACCOUNT_NUMBER}-{AWS_REGION_EU_WEST_1}-ee26b0dd4",
"ProductArn": f"arn:aws:securityhub:{AWS_REGION_EU_WEST_1}::product/prowler/prowler",
"RecordState": "ACTIVE",
"ProductFields": {
"ProviderName": "Prowler",
"ProviderVersion": prowler_version,
"ProwlerResourceName": "test",
},
"GeneratorId": "prowler-iam_user_accesskey_unused",
"AwsAccountId": f"{AWS_ACCOUNT_NUMBER}",
"Types": ["Software and Configuration Checks"],
"FirstObservedAt": timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ"),
"UpdatedAt": timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ"),
"CreatedAt": timestamp_utc.strftime("%Y-%m-%dT%H:%M:%SZ"),
"Severity": {"Label": "LOW"},
"Title": "Ensure Access Keys unused are disabled",
"Description": "test",
"Resources": [
{
"Type": "AwsIamAccessAnalyzer",
"Id": "test",
"Partition": "aws",
"Region": f"{AWS_REGION_EU_WEST_1}",
}
],
"Compliance": {
"Status": status,
"RelatedRequirements": [],
"AssociatedStandards": [],
},
"Remediation": {
"Recommendation": {
"Text": "Run sudo yum update and cross your fingers and toes.",
"Url": "https://myfp.com/recommendations/dangerous_things_and_how_to_fix_them.html",
}
},
}
# Mocking Security Hub Get Findings
make_api_call = botocore.client.BaseClient._make_api_call
@@ -92,41 +47,18 @@ def mock_make_api_call(self, operation_name, kwarg):
return make_api_call(self, operation_name, kwarg)
class Test_SecurityHub:
def generate_finding(self, status, region, muted=False):
finding = Check_Report(
load_check_metadata(
f"{path.dirname(path.realpath(__file__))}/fixtures/metadata.json"
).json()
)
finding.status = status
finding.status_extended = "test"
finding.resource_id = "test"
finding.resource_arn = "test"
finding.region = region
finding.muted = muted
def set_mocked_session(region=None):
# Create mock session
return session.Session(
region_name=region,
)
return finding
def set_mocked_output_options(
self, status: list[str] = [], send_sh_only_fails: bool = False
):
output_options = MagicMock
output_options.bulk_checks_metadata = {}
output_options.status = status
output_options.send_sh_only_fails = send_sh_only_fails
return output_options
def set_mocked_session(self, region):
# Create mock session
return session.Session(
region_name=region,
)
class TestSecurityHub:
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_verify_security_hub_integration_enabled_per_region(self):
session = self.set_mocked_session(AWS_REGION_EU_WEST_1)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
assert verify_security_hub_integration_enabled_per_region(
AWS_COMMERCIAL_PARTITION, AWS_REGION_EU_WEST_1, session, AWS_ACCOUNT_NUMBER
)
@@ -135,7 +67,7 @@ class Test_SecurityHub:
self, caplog
):
caplog.set_level(WARNING)
session = self.set_mocked_session(AWS_REGION_EU_WEST_1)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
with patch(
"prowler.providers.aws.lib.security_hub.security_hub.session.Session.client",
@@ -161,7 +93,7 @@ class Test_SecurityHub:
(
"root",
WARNING,
f"ClientError -- [70]: An error occurred ({error_code}) when calling the {operation_name} operation: {error_message}",
f"ClientError -- [90]: An error occurred ({error_code}) when calling the {operation_name} operation: {error_message}",
)
]
@@ -169,7 +101,7 @@ class Test_SecurityHub:
self, caplog
):
caplog.set_level(WARNING)
session = self.set_mocked_session(AWS_REGION_EU_WEST_1)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
with patch(
"prowler.providers.aws.lib.security_hub.security_hub.session.Session.client",
@@ -195,7 +127,7 @@ class Test_SecurityHub:
self, caplog
):
caplog.set_level(WARNING)
session = self.set_mocked_session(AWS_REGION_EU_WEST_1)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
with patch(
"prowler.providers.aws.lib.security_hub.security_hub.session.Session.client",
@@ -221,7 +153,7 @@ class Test_SecurityHub:
(
"root",
ERROR,
f"ClientError -- [70]: An error occurred ({error_code}) when calling the {operation_name} operation: {error_message}",
f"ClientError -- [90]: An error occurred ({error_code}) when calling the {operation_name} operation: {error_message}",
)
]
@@ -229,7 +161,7 @@ class Test_SecurityHub:
self, caplog
):
caplog.set_level(WARNING)
session = self.set_mocked_session(AWS_REGION_EU_WEST_1)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
with patch(
"prowler.providers.aws.lib.security_hub.security_hub.session.Session.client",
@@ -247,167 +179,141 @@ class Test_SecurityHub:
(
"root",
ERROR,
f"Exception -- [70]: {error_message}",
f"Exception -- [90]: {error_message}",
)
]
def test_prepare_security_hub_findings_enabled_region_all_statuses(self):
def test_filter_security_hub_findings_per_region_enabled_region_all_statuses(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options()
findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_1)]
aws_provider = set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
)
findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert prepare_security_hub_findings(
findings,
aws_provider,
output_options,
assert filter_security_hub_findings_per_region(
asff.data,
False,
[],
enabled_regions,
) == {
AWS_REGION_EU_WEST_1: [get_security_hub_finding("PASSED")],
}
) == {AWS_REGION_EU_WEST_1: [asff.data[0]]}
def test_prepare_security_hub_findings_all_statuses_MANUAL_finding(self):
def test_filter_security_hub_findings_per_region_all_statuses_MANUAL_finding(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options()
findings = [self.generate_finding("MANUAL", AWS_REGION_EU_WEST_1)]
aws_provider = set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
)
assert prepare_security_hub_findings(
findings,
aws_provider,
output_options,
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
def test_prepare_security_hub_findings_disabled_region(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options()
findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_2)]
aws_provider = set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
)
assert prepare_security_hub_findings(
findings,
aws_provider,
output_options,
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
def test_prepare_security_hub_findings_PASS_and_FAIL_statuses(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options(status=["FAIL"])
findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_1)]
aws_provider = set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
)
assert prepare_security_hub_findings(
findings,
aws_provider,
output_options,
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
def test_prepare_security_hub_findings_FAIL_and_FAIL_statuses(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options(status=["FAIL"])
findings = [self.generate_finding("FAIL", AWS_REGION_EU_WEST_1)]
aws_provider = set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
)
assert prepare_security_hub_findings(
findings,
aws_provider,
output_options,
enabled_regions,
) == {AWS_REGION_EU_WEST_1: [get_security_hub_finding("FAILED")]}
def test_prepare_security_hub_findings_send_sh_only_fails_PASS(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options(send_sh_only_fails=True)
findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_1)]
aws_provider = set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
)
assert prepare_security_hub_findings(
findings,
aws_provider,
output_options,
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
def test_prepare_security_hub_findings_send_sh_only_fails_FAIL(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options(send_sh_only_fails=True)
findings = [self.generate_finding("FAIL", AWS_REGION_EU_WEST_1)]
aws_provider = set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
)
assert prepare_security_hub_findings(
findings,
aws_provider,
output_options,
enabled_regions,
) == {AWS_REGION_EU_WEST_1: [get_security_hub_finding("FAILED")]}
def test_prepare_security_hub_findings_no_audited_regions(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options()
findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_1)]
aws_provider = set_mocked_aws_provider()
assert prepare_security_hub_findings(
findings,
aws_provider,
output_options,
enabled_regions,
) == {
AWS_REGION_EU_WEST_1: [get_security_hub_finding("PASSED")],
}
def test_prepare_security_hub_findings_muted_fail_with_send_sh_only_fails(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options(
send_sh_only_fails=True,
)
findings = [
self.generate_finding(
generate_finding_output(status="MANUAL", region=AWS_REGION_EU_WEST_1)
]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
def test_filter_security_hub_findings_per_region_disabled_region(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_2)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
def test_filter_security_hub_findings_per_region_PASS_and_FAIL_statuses(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
["FAIL"],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
def test_filter_security_hub_findings_per_region_FAIL_and_FAIL_statuses(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="FAIL", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
["FAIL"],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: [asff.data[0]]}
def test_filter_security_hub_findings_per_region_send_sh_only_fails_PASS(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
True,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
def test_filter_security_hub_findings_per_region_send_sh_only_fails_FAIL(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="FAIL", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
True,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: [asff.data[0]]}
def test_filter_security_hub_findings_per_region_no_audited_regions(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: [asff.data[0]]}
def test_filter_security_hub_findings_per_region_muted_fail_with_send_sh_only_fails(
self,
):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [
generate_finding_output(
status="FAIL", region=AWS_REGION_EU_WEST_1, muted=True
)
]
aws_provider = set_mocked_aws_provider()
asff = ASFF(findings=findings)
assert prepare_security_hub_findings(
findings,
aws_provider,
output_options,
assert filter_security_hub_findings_per_region(
asff.data,
True,
[],
enabled_regions,
) == {
AWS_REGION_EU_WEST_1: [],
}
def test_prepare_security_hub_findings_muted_fail_with_status_FAIL(self):
def test_filter_security_hub_findings_per_region_muted_fail_with_status_FAIL(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options(status=["FAIL"])
findings = [
self.generate_finding(
generate_finding_output(
status="FAIL", region=AWS_REGION_EU_WEST_1, muted=True
)
]
aws_provider = set_mocked_aws_provider()
asff = ASFF(findings=findings)
assert prepare_security_hub_findings(
findings,
aws_provider,
output_options,
assert filter_security_hub_findings_per_region(
asff.data,
False,
["FAIL"],
enabled_regions,
) == {
AWS_REGION_EU_WEST_1: [],
@@ -415,18 +321,18 @@ class Test_SecurityHub:
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_batch_send_to_security_hub_one_finding(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
output_options = self.set_mocked_output_options()
findings = [self.generate_finding("PASS", AWS_REGION_EU_WEST_1)]
aws_provider = set_mocked_aws_provider(
audited_regions=[AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
)
session = self.set_mocked_session(AWS_REGION_EU_WEST_1)
enabled_regions = [AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
findings = [
generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1),
generate_finding_output(status="FAIL", region=AWS_REGION_EU_WEST_2),
]
asff = ASFF(findings=findings)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
security_hub_findings = prepare_security_hub_findings(
findings,
aws_provider,
output_options,
security_hub_findings = filter_security_hub_findings_per_region(
asff.data,
False,
[],
enabled_regions,
)
@@ -435,5 +341,5 @@ class Test_SecurityHub:
security_hub_findings,
session,
)
== 1
== 2
)