diff --git a/prowler/__main__.py b/prowler/__main__.py index ac4e35a182..bc5c832a4d 100644 --- a/prowler/__main__.py +++ b/prowler/__main__.py @@ -6,7 +6,11 @@ from os import environ from colorama import Fore, Style -from prowler.config.config import csv_file_suffix, get_available_compliance_frameworks +from prowler.config.config import ( + csv_file_suffix, + get_available_compliance_frameworks, + json_ocsf_file_suffix, +) from prowler.lib.banner import print_banner from prowler.lib.check.check import ( bulk_load_checks_metadata, @@ -40,7 +44,7 @@ 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 from prowler.lib.outputs.html.html import add_html_footer, fill_html_overview_statistics -from prowler.lib.outputs.json.json import close_json +from prowler.lib.outputs.ocsf.ocsf import OCSF from prowler.lib.outputs.outputs import extract_findings_statistics from prowler.lib.outputs.slack.slack import Slack from prowler.lib.outputs.summary_table import display_summary_table @@ -308,13 +312,17 @@ def prowler(): # Close json file if exists # TODO: generate JSON here - if "json" in mode: - close_json( - global_provider.output_options.output_filename, - global_provider.output_options.output_directory, - mode, + if "json-ocsf" in mode: + filename = ( + f"{global_provider.output_options.output_directory}/" + f"{global_provider.output_options.output_filename}{json_ocsf_file_suffix}" ) - + json_finding = OCSF( + findings=finding_outputs, + create_file_descriptor=True, + file_path=filename, + ) + json_finding.batch_write_data_to_file() if "html" in mode: # TODO: generate HTML here add_html_footer( diff --git a/prowler/lib/outputs/file_descriptors.py b/prowler/lib/outputs/file_descriptors.py index cf90dd76a3..ce86662c66 100644 --- a/prowler/lib/outputs/file_descriptors.py +++ b/prowler/lib/outputs/file_descriptors.py @@ -6,7 +6,6 @@ from prowler.config.config import ( csv_file_suffix, html_file_suffix, json_asff_file_suffix, - json_ocsf_file_suffix, ) from prowler.lib.logger import logger from prowler.lib.outputs.compliance.mitre_attack.models import ( @@ -50,7 +49,7 @@ def initialize_file_descriptor( "a", ) - if output_mode in ("json-asff", "json-ocsf"): + if output_mode == "json-asff": file_descriptor.write("[") elif "html" in output_mode: add_html_header(file_descriptor, provider) @@ -84,13 +83,6 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi ) file_descriptors.update({output_mode: file_descriptor}) - elif output_mode == "json-ocsf": - filename = ( - f"{output_directory}/{output_filename}{json_ocsf_file_suffix}" - ) - file_descriptor = initialize_file_descriptor(filename, output_mode) - file_descriptors.update({output_mode: file_descriptor}) - elif provider.type == "gcp": filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}" if "cis_" in output_mode: diff --git a/prowler/lib/outputs/json/json.py b/prowler/lib/outputs/json/json.py deleted file mode 100644 index 1d1184f298..0000000000 --- a/prowler/lib/outputs/json/json.py +++ /dev/null @@ -1,39 +0,0 @@ -import os -import sys - -from prowler.config.config import ( - json_asff_file_suffix, - json_file_suffix, - json_ocsf_file_suffix, -) -from prowler.lib.logger import logger -from prowler.lib.utils.utils import open_file - - -def close_json(output_filename, output_directory, mode): - """close_json closes the output JSON file replacing the last comma with ]""" - try: - suffix = json_file_suffix - if mode == "json-asff": - suffix = json_asff_file_suffix - elif mode == "json-ocsf": - suffix = json_ocsf_file_suffix - filename = f"{output_directory}/{output_filename}{suffix}" - # Close JSON file if exists - if os.path.isfile(filename): - file_descriptor = open_file( - filename, - "a", - ) - # Replace last comma for square bracket if not empty - if file_descriptor.tell() > 0: - if file_descriptor.tell() != 1: - file_descriptor.seek(file_descriptor.tell() - 1, os.SEEK_SET) - file_descriptor.truncate() - file_descriptor.write("]") - file_descriptor.close() - except Exception as error: - logger.critical( - f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}" - ) - sys.exit(1) diff --git a/prowler/lib/outputs/json_ocsf/__init__.py b/prowler/lib/outputs/json_ocsf/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/prowler/lib/outputs/json_ocsf/json_ocsf.py b/prowler/lib/outputs/json_ocsf/json_ocsf.py deleted file mode 100644 index 9a47791ff5..0000000000 --- a/prowler/lib/outputs/json_ocsf/json_ocsf.py +++ /dev/null @@ -1,155 +0,0 @@ -from py_ocsf_models.events.base_event import SeverityID, StatusID -from py_ocsf_models.events.findings.detection_finding import DetectionFinding -from py_ocsf_models.events.findings.detection_finding import ( - TypeID as DetectionFindingTypeID, -) -from py_ocsf_models.events.findings.finding import ActivityID, FindingInformation -from py_ocsf_models.objects.account import Account, TypeID -from py_ocsf_models.objects.cloud import Cloud -from py_ocsf_models.objects.container import Container -from py_ocsf_models.objects.group import Group -from py_ocsf_models.objects.metadata import Metadata -from py_ocsf_models.objects.organization import Organization -from py_ocsf_models.objects.product import Product -from py_ocsf_models.objects.remediation import Remediation -from py_ocsf_models.objects.resource_details import ResourceDetails - -from prowler.lib.logger import logger -from prowler.lib.outputs.finding import Finding - - -def get_account_type_id_by_provider(provider: str) -> TypeID: - type_id = TypeID.Other - if provider == "aws": - type_id = TypeID.AWS_Account - elif provider == "azure": - type_id = TypeID.Azure_AD_Account - elif provider == "gcp": - type_id = TypeID.GCP_Account - return type_id - - -def get_finding_status_id(status: str, muted: bool) -> StatusID: - status_id = StatusID.Other - if status == "FAIL": - status_id = StatusID.New - if muted: - status_id = StatusID.Suppressed - return status_id - - -def fill_json_ocsf(finding_output: Finding) -> DetectionFinding: - try: - finding_activity = ActivityID.Create - cloud_account_type = get_account_type_id_by_provider(finding_output.provider) - finding_severity = getattr( - SeverityID, finding_output.severity.capitalize(), SeverityID.Unknown - ) - finding_status = get_finding_status_id( - finding_output.status, finding_output.muted - ) - - detection_finding = DetectionFinding( - activity_id=finding_activity.value, - activity_name=finding_activity.name, - finding_info=FindingInformation( - created_time=finding_output.timestamp, - desc=finding_output.description, - title=finding_output.check_title, - uid=finding_output.finding_uid, - product_uid="prowler", - ), - event_time=finding_output.timestamp, - remediation=Remediation( - desc=finding_output.remediation_recommendation_text, - references=list( - filter( - None, - [ - finding_output.remediation_code_nativeiac, - finding_output.remediation_code_terraform, - finding_output.remediation_code_cli, - finding_output.remediation_code_other, - finding_output.remediation_recommendation_url, - ], - ) - ), - ), - severity_id=finding_severity.value, - severity=finding_severity.name, - status_id=finding_status.value, - status=finding_status.name, - status_code=finding_output.status, - status_detail=finding_output.status_extended, - risk_details=finding_output.risk, - resources=[ - ResourceDetails( - # TODO: Check labels for other providers - labels=( - finding_output.resource_tags.split(",") - if finding_output.resource_tags - else [] - ), - name=finding_output.resource_name, - uid=finding_output.resource_uid, - group=Group(name=finding_output.service_name), - type=finding_output.resource_type, - # TODO: this should be included only if using the Cloud profile - cloud_partition=finding_output.partition, - region=finding_output.region, - data={"details": finding_output.resource_details}, - ) - ], - metadata=Metadata( - event_code=finding_output.check_id, - product=Product( - name="Prowler", - vendor_name="Prowler", - version=finding_output.prowler_version, - ), - ), - type_uid=DetectionFindingTypeID.Create, - type_name=DetectionFindingTypeID.Create.name, - unmapped={ - "check_type": finding_output.check_type, - "related_url": finding_output.related_url, - "categories": finding_output.categories, - "depends_on": finding_output.depends_on, - "related_to": finding_output.related_to, - "notes": finding_output.notes, - "compliance": finding_output.compliance, - }, - ) - - if finding_output.provider == "kubernetes": - detection_finding.container = ( - Container( - name=finding_output.resource_name, - uid=finding_output.resource_uid, - ), - ) - # TODO: Get the PID of the namespace (we only have the name of the namespace) - # detection_finding.namespace_pid=, - else: - detection_finding.cloud = Cloud( - account=Account( - name=finding_output.account_name, - type_id=cloud_account_type.value, - type=cloud_account_type.name, - uid=finding_output.account_uid, - labels=finding_output.account_tags, - ), - org=Organization( - uid=finding_output.account_organization_uid, - name=finding_output.account_organization_name, - ), - provider=finding_output.provider, - region=finding_output.region, - ) - - return detection_finding - - except Exception as error: - logger.error( - f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" - ) diff --git a/prowler/lib/outputs/json/__init__.py b/prowler/lib/outputs/ocsf/__init__.py similarity index 100% rename from prowler/lib/outputs/json/__init__.py rename to prowler/lib/outputs/ocsf/__init__.py diff --git a/prowler/lib/outputs/ocsf/ocsf.py b/prowler/lib/outputs/ocsf/ocsf.py new file mode 100644 index 0000000000..d753fc17ed --- /dev/null +++ b/prowler/lib/outputs/ocsf/ocsf.py @@ -0,0 +1,227 @@ +import os +import sys +from typing import List + +from py_ocsf_models.events.base_event import SeverityID, StatusID +from py_ocsf_models.events.findings.detection_finding import DetectionFinding +from py_ocsf_models.events.findings.detection_finding import ( + TypeID as DetectionFindingTypeID, +) +from py_ocsf_models.events.findings.finding import ActivityID, FindingInformation +from py_ocsf_models.objects.account import Account, TypeID +from py_ocsf_models.objects.cloud import Cloud +from py_ocsf_models.objects.container import Container +from py_ocsf_models.objects.group import Group +from py_ocsf_models.objects.metadata import Metadata +from py_ocsf_models.objects.organization import Organization +from py_ocsf_models.objects.product import Product +from py_ocsf_models.objects.remediation import Remediation +from py_ocsf_models.objects.resource_details import ResourceDetails + +from prowler.lib.logger import logger +from prowler.lib.outputs.finding import Finding +from prowler.lib.outputs.output import Output + + +class OCSF(Output): + """ + OCSF class that transforms the findings into the OCSF format. + + This class provides methods to transform the findings into the OCSF format and write them to a file. + + Attributes: + - _data: A list to store the transformed findings. + - _file_descriptor: A file descriptor to write the findings to a file. + + Methods: + - transform(findings: List[Finding]) -> None: Transforms the findings into the OCSF format. + - batch_write_data_to_file() -> None: Writes the findings to a file using the OCSF format using the `Output._file_descriptor`. + - get_account_type_id_by_provider(provider: str) -> TypeID: Returns the TypeID based on the provider. + - get_finding_status_id(status: str, muted: bool) -> StatusID: Returns the StatusID based on the status and muted values. + + References: + - OCSF: https://docs.aws.amazon.com/security-lake/latest/userguide/open-cybersecurity-schema-framework.html + - PY-OCSF-Model: https://github.com/prowler-cloud/py-ocsf-models + """ + + def transform(self, findings: List[Finding]) -> None: + """Transforms the findings into the OCSF format. + + Args: + findings (List[Finding]): a list of Finding objects + """ + try: + for finding in findings: + finding_activity = ActivityID.Create + cloud_account_type = self.get_account_type_id_by_provider( + finding.provider + ) + finding_severity = getattr( + SeverityID, finding.severity.capitalize(), SeverityID.Unknown + ) + finding_status = self.get_finding_status_id( + finding.status, finding.muted + ) + + detection_finding = DetectionFinding( + activity_id=finding_activity.value, + activity_name=finding_activity.name, + finding_info=FindingInformation( + created_time=finding.timestamp, + desc=finding.description, + title=finding.check_title, + uid=finding.finding_uid, + product_uid="prowler", + ), + event_time=finding.timestamp, + remediation=Remediation( + desc=finding.remediation_recommendation_text, + references=list( + filter( + None, + [ + finding.remediation_code_nativeiac, + finding.remediation_code_terraform, + finding.remediation_code_cli, + finding.remediation_code_other, + finding.remediation_recommendation_url, + ], + ) + ), + ), + severity_id=finding_severity.value, + severity=finding_severity.name, + status_id=finding_status.value, + status=finding_status.name, + status_code=finding.status, + status_detail=finding.status_extended, + risk_details=finding.risk, + resources=[ + ResourceDetails( + # TODO: Check labels for other providers + labels=( + finding.resource_tags.split(",") + if finding.resource_tags + else [] + ), + name=finding.resource_name, + uid=finding.resource_uid, + group=Group(name=finding.service_name), + type=finding.resource_type, + # TODO: this should be included only if using the Cloud profile + cloud_partition=finding.partition, + region=finding.region, + data={"details": finding.resource_details}, + ) + ], + metadata=Metadata( + event_code=finding.check_id, + product=Product( + name="Prowler", + vendor_name="Prowler", + version=finding.prowler_version, + ), + ), + type_uid=DetectionFindingTypeID.Create, + type_name=DetectionFindingTypeID.Create.name, + unmapped={ + "check_type": finding.check_type, + "related_url": finding.related_url, + "categories": finding.categories, + "depends_on": finding.depends_on, + "related_to": finding.related_to, + "notes": finding.notes, + "compliance": finding.compliance, + }, + ) + + if finding.provider == "kubernetes": + detection_finding.container = Container( + name=finding.resource_name, + uid=finding.resource_uid, + ) + # TODO: Get the PID of the namespace (we only have the name of the namespace) + # detection_finding.namespace_pid=, + else: + detection_finding.cloud = Cloud( + account=Account( + name=finding.account_name, + type_id=cloud_account_type.value, + type=cloud_account_type.name, + uid=finding.account_uid, + labels=finding.account_tags, + ), + org=Organization( + uid=finding.account_organization_uid, + name=finding.account_organization_name, + ), + provider=finding.provider, + region=finding.region, + ) + + self._data.append(detection_finding) + 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 to a file using the OCSF format using the `Output._file_descriptor`.""" + try: + 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(",") + if self._file_descriptor.tell() > 0: + if self._file_descriptor.tell() != 1: + self._file_descriptor.seek( + self._file_descriptor.tell() - 1, os.SEEK_SET + ) + self._file_descriptor.truncate() + self._file_descriptor.write("]") + self._file_descriptor.close() + except Exception as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + sys.exit(1) + + @staticmethod + def get_account_type_id_by_provider(provider: str) -> TypeID: + """ + Returns the TypeID based on the provider. + + Args: + provider (str): The provider name + + Returns: + TypeID: The TypeID based on the provider + """ + type_id = TypeID.Other + if provider == "aws": + type_id = TypeID.AWS_Account + elif provider == "azure": + type_id = TypeID.Azure_AD_Account + elif provider == "gcp": + type_id = TypeID.GCP_Account + return type_id + + @staticmethod + def get_finding_status_id(status: str, muted: bool) -> StatusID: + """ + Returns the StatusID based on the status and muted values. + + Args: + status (str): The status value + muted (bool): The muted value + + Returns: + StatusID: The StatusID based on the status and muted values + """ + status_id = StatusID.Other + if status == "FAIL": + status_id = StatusID.New + if muted: + status_id = StatusID.Suppressed + return status_id diff --git a/prowler/lib/outputs/output.py b/prowler/lib/outputs/output.py index 284c3e138d..43159c5061 100644 --- a/prowler/lib/outputs/output.py +++ b/prowler/lib/outputs/output.py @@ -8,8 +8,8 @@ from prowler.lib.utils.utils import open_file class Output(ABC): - _data: list[Finding] = [] - _file_descriptor: TextIOWrapper = None + _data: list[Finding] + _file_descriptor: TextIOWrapper def __init__( self, @@ -17,6 +17,7 @@ class Output(ABC): create_file_descriptor: bool = False, file_path: str = None, ) -> None: + self._data = [] self.transform(findings) if create_file_descriptor: self.create_file_descriptor(file_path) diff --git a/prowler/lib/outputs/outputs.py b/prowler/lib/outputs/outputs.py index 7318e4bd93..09e9c1bc46 100644 --- a/prowler/lib/outputs/outputs.py +++ b/prowler/lib/outputs/outputs.py @@ -12,7 +12,6 @@ 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 -from prowler.lib.outputs.json_ocsf.json_ocsf import fill_json_ocsf def stdout_report(finding, color, verbose, status, fix): @@ -112,15 +111,6 @@ def report(check_findings, provider): # Common Output Data finding_output = Finding.generate_output(provider, finding) - # JSON - if "json-ocsf" in file_descriptors: - detection_finding = fill_json_ocsf(finding_output) - - file_descriptors["json-ocsf"].write( - detection_finding.json(exclude_none=True, indent=4) - ) - file_descriptors["json-ocsf"].write(",") - if "html" in file_descriptors: fill_html(file_descriptors["html"], finding_output) diff --git a/tests/lib/outputs/fixtures/fixtures.py b/tests/lib/outputs/fixtures/fixtures.py index 7999fec250..35414b3019 100644 --- a/tests/lib/outputs/fixtures/fixtures.py +++ b/tests/lib/outputs/fixtures/fixtures.py @@ -6,12 +6,19 @@ from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER # TODO: customize it per provider -def generate_finding_output(status, severity, muted, region) -> Finding: +def generate_finding_output( + status, + severity, + muted, + region, + timestamp: datetime = datetime.now(), + provider: str = "aws", +) -> Finding: # TODO: Include metadata from a valid file return Finding( auth_method="profile: default", - timestamp=datetime.now(), + timestamp=timestamp, account_uid=AWS_ACCOUNT_NUMBER, account_name=AWS_ACCOUNT_NUMBER, account_email="", @@ -19,7 +26,7 @@ def generate_finding_output(status, severity, muted, region) -> Finding: account_organization_name="test-organization", account_tags=["test-tag:test-value"], finding_uid="test-unique-finding", - provider="aws", + provider=provider, check_id="test-check-id", check_title="test-check-id", check_type="test-type", diff --git a/tests/lib/outputs/json_ocsf/json_ocsf_test.py b/tests/lib/outputs/json_ocsf/json_ocsf_test.py deleted file mode 100644 index 623158d5bf..0000000000 --- a/tests/lib/outputs/json_ocsf/json_ocsf_test.py +++ /dev/null @@ -1,242 +0,0 @@ -# from datetime import datetime -from os import path - -from py_ocsf_models.events.base_event import SeverityID, StatusID -from py_ocsf_models.events.findings.detection_finding import ( - TypeID as DetectionFindingTypeID, -) -from py_ocsf_models.events.findings.finding import ActivityID, FindingInformation -from py_ocsf_models.objects.account import Account, TypeID -from py_ocsf_models.objects.cloud import Cloud -from py_ocsf_models.objects.group import Group -from py_ocsf_models.objects.metadata import Metadata -from py_ocsf_models.objects.organization import Organization -from py_ocsf_models.objects.product import Product - -# from py_ocsf_models.events.findings.detection_finding import DetectionFinding -from py_ocsf_models.objects.remediation import Remediation -from py_ocsf_models.objects.resource_details import ResourceDetails - -from prowler.config.config import prowler_version -from prowler.lib.outputs.json_ocsf.json_ocsf import ( - fill_json_ocsf, - get_account_type_id_by_provider, - get_finding_status_id, -) -from tests.lib.outputs.fixtures.fixtures import generate_finding_output -from tests.providers.aws.utils import AWS_REGION_EU_WEST_1 - -METADATA_FIXTURE_PATH = ( - f"{path.dirname(path.realpath(__file__))}/../fixtures/metadata.json" -) - - -class TestOutputJSONOCSF: - # test_fill_json_ocsf_iso_format_timestamp - def test_finding_output_cloud_pass_low_muted(self): - finding_output = generate_finding_output( - "PASS", "low", True, AWS_REGION_EU_WEST_1 - ) - - finding_json_ocsf = fill_json_ocsf(finding_output) - - # Activity - assert finding_json_ocsf.activity_id == ActivityID.Create.value - assert finding_json_ocsf.activity_name == ActivityID.Create.name - - # Finding Information - finding_information = finding_json_ocsf.finding_info - - assert isinstance(finding_information, FindingInformation) - assert finding_information.created_time == finding_output.timestamp - assert finding_information.desc == finding_output.description - assert finding_information.title == finding_output.check_title - assert finding_information.uid == finding_output.finding_uid - assert finding_information.product_uid == "prowler" - - # Event time - assert finding_json_ocsf.event_time == finding_output.timestamp - - # Remediation - remediation = finding_json_ocsf.remediation - assert isinstance(remediation, Remediation) - assert remediation.desc == finding_output.remediation_recommendation_text - assert remediation.references == [] - - # Severity - assert finding_json_ocsf.severity_id == SeverityID.Low - assert finding_json_ocsf.severity == SeverityID.Low.name - - # Status - assert finding_json_ocsf.status_id == StatusID.Suppressed.value - assert finding_json_ocsf.status == StatusID.Suppressed.name - assert finding_json_ocsf.status_code == finding_output.status - assert finding_json_ocsf.status_detail == finding_output.status_extended - - # Risk - assert finding_json_ocsf.risk_details == finding_output.risk - - # Unmapped Data - assert finding_json_ocsf.unmapped == { - "check_type": finding_output.check_type, - "related_url": finding_output.related_url, - "categories": finding_output.categories, - "depends_on": finding_output.depends_on, - "related_to": finding_output.related_to, - "notes": finding_output.notes, - "compliance": finding_output.compliance, - } - - # ResourceDetails - resource_details = finding_json_ocsf.resources - - assert len(resource_details) == 1 - assert isinstance(resource_details, list) - assert isinstance(resource_details[0], ResourceDetails) - assert resource_details[0].labels == [] - assert resource_details[0].name == finding_output.resource_name - assert resource_details[0].uid == finding_output.resource_uid - assert resource_details[0].type == finding_output.resource_type - assert resource_details[0].cloud_partition == finding_output.partition - assert resource_details[0].region == finding_output.region - assert resource_details[0].data == {"details": finding_output.resource_details} - - resource_details_group = resource_details[0].group - assert isinstance(resource_details_group, Group) - assert resource_details_group.name == finding_output.service_name - - # Metadata - metadata = finding_json_ocsf.metadata - assert isinstance(metadata, Metadata) - assert metadata.event_code == finding_output.check_id - - metadata_product = metadata.product - assert isinstance(metadata_product, Product) - assert metadata_product.name == "Prowler" - assert metadata_product.vendor_name == "Prowler" - assert metadata_product.version == prowler_version - - # Type - assert finding_json_ocsf.type_uid == DetectionFindingTypeID.Create - assert finding_json_ocsf.type_name == DetectionFindingTypeID.Create.name - - # Cloud - cloud = finding_json_ocsf.cloud - assert isinstance(cloud, Cloud) - assert cloud.provider == "aws" - assert cloud.region == finding_output.region - - cloud_account = cloud.account - assert isinstance(cloud_account, Account) - assert cloud_account.name == finding_output.account_name - assert cloud_account.type_id == TypeID.AWS_Account - assert cloud_account.type == TypeID.AWS_Account.name - assert cloud_account.uid == finding_output.account_uid - assert cloud_account.labels == finding_output.account_tags - - cloud_organization = cloud.org - assert isinstance(cloud_organization, Organization) - assert cloud_organization.uid == finding_output.account_organization_uid - assert cloud_organization.name == finding_output.account_organization_name - - def test_finding_output_cloud_fail_low_not_muted(self): - finding_output = generate_finding_output( - "FAIL", "low", False, AWS_REGION_EU_WEST_1 - ) - - finding_json_ocsf = fill_json_ocsf(finding_output) - - # Status - assert finding_json_ocsf.status_id == StatusID.New.value - assert finding_json_ocsf.status == StatusID.New.name - assert finding_json_ocsf.status_code == finding_output.status - assert finding_json_ocsf.status_detail == finding_output.status_extended - - def test_finding_output_cloud_pass_low_not_muted(self): - finding_output = generate_finding_output( - "PASS", "low", False, AWS_REGION_EU_WEST_1 - ) - - finding_json_ocsf = fill_json_ocsf(finding_output) - - # Status - assert finding_json_ocsf.status_id == StatusID.Other.value - assert finding_json_ocsf.status == StatusID.Other.name - assert finding_json_ocsf.status_code == finding_output.status - assert finding_json_ocsf.status_detail == finding_output.status_extended - - # Returns TypeID.AWS_Account when provider is 'aws' - def test_returns_aws_account_when_provider_is_aws(self): - provider = "aws" - result = get_account_type_id_by_provider(provider) - - assert result == TypeID.AWS_Account - - # Returns TypeID.Azure_AD_Account when provider is 'azure' - def test_returns_azure_ad_account_when_provider_is_azure(self): - provider = "azure" - result = get_account_type_id_by_provider(provider) - - assert result == TypeID.Azure_AD_Account - - # Returns TypeID.GCP_Account when provider is 'gcp' - def test_returns_gcp_account_when_provider_is_gcp(self): - provider = "gcp" - result = get_account_type_id_by_provider(provider) - - assert result == TypeID.GCP_Account - - # Returns TypeID.Other when provider is None - def test_returns_other_when_provider_is_none(self): - provider = None - result = get_account_type_id_by_provider(provider) - - assert result == TypeID.Other - - # Returns StatusID.New when status is "FAIL" and muted is False - def test_new_when_status_fail_and_not_muted(self): - status = "FAIL" - muted = False - result = get_finding_status_id(status, muted) - - assert result == StatusID.New - - # Returns StatusID.Suppressed when status is "FAIL" and muted is True - def test_suppressed_when_status_fail_and_muted(self): - status = "FAIL" - muted = True - result = get_finding_status_id(status, muted) - - assert result == StatusID.Suppressed - - # Returns StatusID.Other when status is None and muted is False - def test_other_when_status_whatever_and_not_muted(self): - status = None - muted = False - result = get_finding_status_id(status, muted) - - assert result == StatusID.Other - - # Returns StatusID.Suppresed when status is None and muted is True - def test_other_when_status_whatever_and_muted(self): - status = None - muted = True - result = get_finding_status_id(status, muted) - - assert result == StatusID.Suppressed - - # Returns StatusID.Suppressed when muted is True and status is not "FAIL" - def test_suppressed_when_status_pass_and_muted(self): - status = "PASS" - muted = True - result = get_finding_status_id(status, muted) - - assert result == StatusID.Suppressed - - # Returns StatusID.Other when muted is False and status is not "FAIL" - def test_other_when_status_pass_and_not_muted(self): - status = "PASS" - muted = False - result = get_finding_status_id(status, muted) - - assert result == StatusID.Other diff --git a/tests/lib/outputs/ocsf/ocsf_test.py b/tests/lib/outputs/ocsf/ocsf_test.py new file mode 100644 index 0000000000..3263b449e2 --- /dev/null +++ b/tests/lib/outputs/ocsf/ocsf_test.py @@ -0,0 +1,326 @@ +import json +from datetime import datetime +from io import StringIO + +from mock import patch +from py_ocsf_models.events.base_event import SeverityID, StatusID +from py_ocsf_models.events.findings.detection_finding import DetectionFinding +from py_ocsf_models.events.findings.detection_finding import ( + TypeID as DetectionFindingTypeID, +) +from py_ocsf_models.events.findings.finding import ActivityID, FindingInformation +from py_ocsf_models.objects.account import Account, TypeID +from py_ocsf_models.objects.cloud import Cloud +from py_ocsf_models.objects.group import Group +from py_ocsf_models.objects.metadata import Metadata +from py_ocsf_models.objects.organization import Organization +from py_ocsf_models.objects.product import Product +from py_ocsf_models.objects.remediation import Remediation +from py_ocsf_models.objects.resource_details import ResourceDetails + +from prowler.config.config import prowler_version +from prowler.lib.outputs.ocsf.ocsf import OCSF +from tests.lib.outputs.fixtures.fixtures import generate_finding_output +from tests.providers.aws.utils import AWS_REGION_EU_WEST_1 + +now = datetime.now() +expected_json_output = json.dumps( + [ + { + "metadata": { + "event_code": "test-check-id", + "product": { + "name": "Prowler", + "vendor_name": "Prowler", + "version": "4.2.4", + }, + "version": "1.2.0", + }, + "severity_id": 2, + "severity": "Low", + "status": "New", + "status_code": "FAIL", + "status_detail": "status extended", + "status_id": 1, + "unmapped": { + "check_type": "test-type", + "related_url": "test-url", + "categories": "test-category", + "depends_on": "test-dependency", + "related_to": "test-related-to", + "notes": "test-notes", + "compliance": {"test-compliance": "test-compliance"}, + }, + "activity_name": "Create", + "activity_id": 1, + "finding_info": { + "created_time": now.isoformat(), + "desc": "check description", + "product_uid": "prowler", + "title": "test-check-id", + "uid": "test-unique-finding", + }, + "resources": [ + { + "cloud_partition": "aws", + "region": "eu-west-1", + "data": {"details": "resource_details"}, + "group": {"name": "test-service"}, + "labels": [], + "name": "resource_name", + "type": "test-resource", + "uid": "resource-id", + } + ], + "category_name": "Findings", + "category_uid": 2, + "class_name": "DetectionFinding", + "class_uid": 2004, + "cloud": { + "account": { + "name": "123456789012", + "type": "AWS_Account", + "type_id": 10, + "uid": "123456789012", + "labels": ["test-tag:test-value"], + }, + "org": {"name": "test-organization", "uid": "test-organization-id"}, + "provider": "aws", + "region": "eu-west-1", + }, + "event_time": now.isoformat(), + "remediation": {"desc": "", "references": []}, + "risk_details": "test-risk", + "type_uid": 200401, + "type_name": "Create", + } + ] +) + + +class TestOCSF: + def test_transform(self): + findings = [generate_finding_output("FAIL", "low", False, AWS_REGION_EU_WEST_1)] + + ocsf = OCSF(findings) + + output_data = ocsf.data[0] + assert isinstance(output_data, DetectionFinding) + + def test_batch_write_data_to_file(self): + mock_file = StringIO() + findings = [ + generate_finding_output("FAIL", "low", False, AWS_REGION_EU_WEST_1, now) + ] + + output = OCSF(findings) + 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() + + assert json.loads(content) == json.loads(expected_json_output) + + def test_finding_output_cloud_pass_low_muted(self): + finding_output = generate_finding_output( + "PASS", "low", True, AWS_REGION_EU_WEST_1 + ) + + finding_ocsf = OCSF([finding_output]) + finding_ocsf = finding_ocsf.data[0] + # Activity + assert finding_ocsf.activity_id == ActivityID.Create.value + assert finding_ocsf.activity_name == ActivityID.Create.name + + # Finding Information + finding_information = finding_ocsf.finding_info + + assert isinstance(finding_information, FindingInformation) + assert finding_information.created_time == finding_output.timestamp + assert finding_information.desc == finding_output.description + assert finding_information.title == finding_output.check_title + assert finding_information.uid == finding_output.finding_uid + assert finding_information.product_uid == "prowler" + + # Event time + assert finding_ocsf.event_time == finding_output.timestamp + + # Remediation + remediation = finding_ocsf.remediation + assert isinstance(remediation, Remediation) + assert remediation.desc == finding_output.remediation_recommendation_text + assert remediation.references == [] + + # Severity + assert finding_ocsf.severity_id == SeverityID.Low + assert finding_ocsf.severity == SeverityID.Low.name + + # Status + assert finding_ocsf.status_id == StatusID.Suppressed.value + assert finding_ocsf.status == StatusID.Suppressed.name + assert finding_ocsf.status_code == finding_output.status + assert finding_ocsf.status_detail == finding_output.status_extended + + # Risk + assert finding_ocsf.risk_details == finding_output.risk + + # Unmapped Data + assert finding_ocsf.unmapped == { + "check_type": finding_output.check_type, + "related_url": finding_output.related_url, + "categories": finding_output.categories, + "depends_on": finding_output.depends_on, + "related_to": finding_output.related_to, + "notes": finding_output.notes, + "compliance": finding_output.compliance, + } + + # ResourceDetails + resource_details = finding_ocsf.resources + + assert len(resource_details) == 1 + assert isinstance(resource_details, list) + assert isinstance(resource_details[0], ResourceDetails) + assert resource_details[0].labels == [] + assert resource_details[0].name == finding_output.resource_name + assert resource_details[0].uid == finding_output.resource_uid + assert resource_details[0].type == finding_output.resource_type + assert resource_details[0].cloud_partition == finding_output.partition + assert resource_details[0].region == finding_output.region + assert resource_details[0].data == {"details": finding_output.resource_details} + + resource_details_group = resource_details[0].group + assert isinstance(resource_details_group, Group) + assert resource_details_group.name == finding_output.service_name + + # Metadata + metadata = finding_ocsf.metadata + assert isinstance(metadata, Metadata) + assert metadata.event_code == finding_output.check_id + + metadata_product = metadata.product + assert isinstance(metadata_product, Product) + assert metadata_product.name == "Prowler" + assert metadata_product.vendor_name == "Prowler" + assert metadata_product.version == prowler_version + + # Type + assert finding_ocsf.type_uid == DetectionFindingTypeID.Create + assert finding_ocsf.type_name == DetectionFindingTypeID.Create.name + + # Cloud + cloud = finding_ocsf.cloud + assert isinstance(cloud, Cloud) + assert cloud.provider == "aws" + assert cloud.region == finding_output.region + + cloud_account = cloud.account + assert isinstance(cloud_account, Account) + assert cloud_account.name == finding_output.account_name + assert cloud_account.type_id == TypeID.AWS_Account + assert cloud_account.type == TypeID.AWS_Account.name + assert cloud_account.uid == finding_output.account_uid + assert cloud_account.labels == finding_output.account_tags + + cloud_organization = cloud.org + assert isinstance(cloud_organization, Organization) + assert cloud_organization.uid == finding_output.account_organization_uid + assert cloud_organization.name == finding_output.account_organization_name + + def test_finding_output_kubernetes(self): + finding_output = generate_finding_output( + "PASS", "low", True, AWS_REGION_EU_WEST_1, provider="kubernetes" + ) + + finding_ocsf = OCSF([finding_output]) + finding_ocsf = finding_ocsf.data[0] + + assert finding_ocsf.container.name == finding_output.resource_name + assert finding_ocsf.container.uid == finding_output.resource_uid + + def test_finding_output_cloud_fail_low_not_muted(self): + finding_output = generate_finding_output( + "FAIL", "low", False, AWS_REGION_EU_WEST_1 + ) + + finding_ocsf = OCSF([finding_output]) + finding_ocsf = finding_ocsf.data[0] + + # Status + assert finding_ocsf.status_id == StatusID.New.value + assert finding_ocsf.status == StatusID.New.name + assert finding_ocsf.status_code == finding_output.status + assert finding_ocsf.status_detail == finding_output.status_extended + + def test_finding_output_cloud_pass_low_not_muted(self): + finding_output = generate_finding_output( + "PASS", "low", False, AWS_REGION_EU_WEST_1 + ) + + finding_ocsf = OCSF([finding_output]) + finding_ocsf = finding_ocsf.data[0] + + # Status + assert finding_ocsf.status_id == StatusID.Other.value + assert finding_ocsf.status == StatusID.Other.name + assert finding_ocsf.status_code == finding_output.status + assert finding_ocsf.status_detail == finding_output.status_extended + + # Returns TypeID.AWS_Account when provider is 'aws' + def test_returns_aws_account_when_provider_is_aws(self): + provider = "aws" + assert OCSF.get_account_type_id_by_provider(provider) == TypeID.AWS_Account + + # Returns TypeID.Azure_AD_Account when provider is 'azure' + def test_returns_azure_ad_account_when_provider_is_azure(self): + provider = "azure" + assert OCSF.get_account_type_id_by_provider(provider) == TypeID.Azure_AD_Account + + # Returns TypeID.GCP_Account when provider is 'gcp' + def test_returns_gcp_account_when_provider_is_gcp(self): + provider = "gcp" + assert OCSF.get_account_type_id_by_provider(provider) == TypeID.GCP_Account + + # Returns TypeID.Other when provider is None + def test_returns_other_when_provider_is_none(self): + provider = "None" + assert OCSF.get_account_type_id_by_provider(provider) == TypeID.Other + + # Returns StatusID.New when status is "FAIL" and muted is False + def test_new_when_status_fail_and_not_muted(self): + status = "FAIL" + muted = False + assert OCSF.get_finding_status_id(status, muted) == StatusID.New + + # Returns StatusID.Suppressed when status is "FAIL" and muted is True + def test_suppressed_when_status_fail_and_muted(self): + status = "FAIL" + muted = True + assert OCSF.get_finding_status_id(status, muted) == StatusID.Suppressed + + # Returns StatusID.Other when status is PASS and muted is False + def test_other_when_status_whatever_and_not_muted(self): + status = "PASS" + muted = False + assert OCSF.get_finding_status_id(status, muted) == StatusID.Other + + # Returns StatusID.Suppresed when status is PASS and muted is True + def test_other_when_status_whatever_and_muted(self): + status = "PASS" + muted = True + assert OCSF.get_finding_status_id(status, muted) == StatusID.Suppressed + + # Returns StatusID.Suppressed when muted is True and status is not "FAIL" + def test_suppressed_when_status_pass_and_muted(self): + status = "PASS" + muted = True + assert OCSF.get_finding_status_id(status, muted) == StatusID.Suppressed + + # Returns StatusID.Other when muted is False and status is not "FAIL" + def test_other_when_status_pass_and_not_muted(self): + status = "PASS" + muted = False + assert OCSF.get_finding_status_id(status, muted) == StatusID.Other diff --git a/tests/lib/outputs/outputs_test.py b/tests/lib/outputs/outputs_test.py index a22f884b75..b72f4e13b8 100644 --- a/tests/lib/outputs/outputs_test.py +++ b/tests/lib/outputs/outputs_test.py @@ -8,7 +8,6 @@ from colorama import Fore from prowler.config.config import ( html_file_suffix, json_asff_file_suffix, - json_ocsf_file_suffix, output_file_timestamp, ) from prowler.lib.check.compliance_models import ( @@ -41,9 +40,8 @@ class TestOutputs: aws_provider = set_mocked_aws_provider() test_output_modes = [ ["json-asff"], - ["json-ocsf"], ["html"], - ["json-asff", "json-ocsf", "html"], + ["json-asff", "html"], ] output_filename = f"prowler-output-{audited_account}-{output_file_timestamp}" expected = [ @@ -53,12 +51,6 @@ class TestOutputs: "a", ) }, - { - "json-ocsf": open_file( - f"{output_directory}/{output_filename}{json_ocsf_file_suffix}", - "a", - ) - }, { "html": open_file( f"{output_directory}/{output_filename}{html_file_suffix}", @@ -70,10 +62,6 @@ class TestOutputs: f"{output_directory}/{output_filename}{json_asff_file_suffix}", "a", ), - "json-ocsf": open_file( - f"{output_directory}/{output_filename}{json_ocsf_file_suffix}", - "a", - ), "html": open_file( f"{output_directory}/{output_filename}{html_file_suffix}", "a",