diff --git a/prowler/__main__.py b/prowler/__main__.py index 966b340e2e..f9c321d579 100644 --- a/prowler/__main__.py +++ b/prowler/__main__.py @@ -298,59 +298,42 @@ 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}" - ) + filename = ( + f"{global_provider.output_options.output_directory}/" + f"{global_provider.output_options.output_filename}" + ) + if mode == "csv": csv_output = CSV( findings=finding_outputs, create_file_descriptor=True, - file_path=filename, + file_path=f"{filename}{csv_file_suffix}", ) # Write CSV Finding Object 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}" - ) + if mode == "json-asff": asff_output = ASFF( findings=finding_outputs, create_file_descriptor=True, - file_path=filename, + file_path=f"{filename}{json_asff_file_suffix}", ) # Write ASFF Finding Object to file asff_output.batch_write_data_to_file() - # Close json file if exists - # TODO: generate JSON here - 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( + if mode == "json-ocsf": + json_output = OCSF( findings=finding_outputs, create_file_descriptor=True, - file_path=filename, + file_path=f"{filename}{json_ocsf_file_suffix}", ) - json_finding.batch_write_data_to_file() - if "html" in mode: - # Generate HTML Finding Object - filename = ( - f"{global_provider.output_options.output_directory}/" - f"{global_provider.output_options.output_filename}{html_file_suffix}" - ) - html_finding = HTML( + json_output.batch_write_data_to_file() + if mode == "html": + html_output = HTML( findings=finding_outputs, create_file_descriptor=True, - file_path=filename, + file_path=f"{filename}{html_file_suffix}", ) - # Write HTML Finding Object to file - html_finding.batch_write_data_to_file( + html_output.batch_write_data_to_file( provider=global_provider, stats=stats ) diff --git a/prowler/lib/outputs/html/html.py b/prowler/lib/outputs/html/html.py index 759c44b4fb..e8dfd5f6af 100644 --- a/prowler/lib/outputs/html/html.py +++ b/prowler/lib/outputs/html/html.py @@ -78,6 +78,8 @@ class HTML(Output): for finding in self._data: self._file_descriptor.write(finding) HTML.write_footer(self._file_descriptor) + # Close file descriptor + self._file_descriptor.close() except Exception as error: logger.error( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" diff --git a/prowler/lib/outputs/ocsf/ocsf.py b/prowler/lib/outputs/ocsf/ocsf.py index 7a677f353a..14bd869f87 100644 --- a/prowler/lib/outputs/ocsf/ocsf.py +++ b/prowler/lib/outputs/ocsf/ocsf.py @@ -1,5 +1,4 @@ import os -import sys from typing import List from py_ocsf_models.events.base_event import SeverityID, StatusID @@ -25,22 +24,22 @@ from prowler.lib.outputs.output import Output class OCSF(Output): """ - OCSF class that transforms the findings into the OCSF format. + OCSF class that transforms the findings into the OCSF Detection Finding format. - This class provides methods to transform the findings into the OCSF format and write them to a file. + This class provides methods to transform the findings into the OCSF Detection Finding 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`. + - transform(findings: List[Finding]) -> None: Transforms the findings into the OCSF Detection Finding format. + - batch_write_data_to_file() -> None: Writes the findings to a file using the OCSF Detection Finding 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 + - OCSF: https://schema.ocsf.io/1.2.0/classes/detection_finding - PY-OCSF-Model: https://github.com/prowler-cloud/py-ocsf-models """ @@ -191,7 +190,6 @@ class OCSF(Output): 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: diff --git a/tests/lib/outputs/asff/asff_test.py b/tests/lib/outputs/asff/asff_test.py index 9a7e1ab0c5..b5f7abb4e9 100644 --- a/tests/lib/outputs/asff/asff_test.py +++ b/tests/lib/outputs/asff/asff_test.py @@ -1,5 +1,11 @@ +from datetime import datetime +from io import StringIO +from json import loads from os import path +from freezegun import freeze_time +from mock import patch + from prowler.config.config import prowler_version, timestamp_utc from prowler.lib.outputs.asff.asff import ( ASFF, @@ -430,6 +436,82 @@ class TestASFF: assert asff_finding == expected + @freeze_time(datetime.now()) + def test_asff_write_to_file(self): + mock_file = StringIO() + + 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") + + expected_asff = [ + { + "SchemaVersion": "2018-10-08", + "Id": "prowler-test-check-id-123456789012-eu-west-1-1aa220687", + "ProductArn": "arn:aws:securityhub:eu-west-1::product/prowler/prowler", + "RecordState": "ACTIVE", + "ProductFields": { + "ProviderName": "Prowler", + "ProviderVersion": "4.2.4", + "ProwlerResourceName": "test-arn", + }, + "GeneratorId": "prowler-test-check-id", + "AwsAccountId": "123456789012", + "Types": ["test-type"], + "FirstObservedAt": timestamp, + "UpdatedAt": timestamp, + "CreatedAt": timestamp, + "Severity": {"Label": "HIGH"}, + "Title": "test-check-id", + "Description": "check description", + "Resources": [ + { + "Type": "test-resource", + "Id": "test-arn", + "Partition": "aws", + "Region": "eu-west-1", + "Tags": {"key1": "value1"}, + } + ], + "Compliance": { + "Status": "PASSED", + "RelatedRequirements": [ + "test-compliance t e s t - c o m p l i a n c e" + ], + "AssociatedStandards": [{"StandardsId": "test-compliance"}], + }, + "Remediation": { + "Recommendation": { + "Text": "", + "Url": "https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html", + } + }, + } + ] + + asff = ASFF(findings=[finding]) + asff._file_descriptor = mock_file + + with patch.object(mock_file, "close", return_value=None): + asff.batch_write_data_to_file() + + mock_file.seek(0) + content = mock_file.read() + assert loads(content) == expected_asff + + def test_batch_write_data_to_file_without_findings(self): + assert not hasattr(ASFF([]), "_file_descriptor") + def test_asff_generate_status(self): assert ASFF.generate_status("PASS") == "PASSED" assert ASFF.generate_status("FAIL") == "FAILED" diff --git a/tests/lib/outputs/csv/csv_test.py b/tests/lib/outputs/csv/csv_test.py index 9c8fe53e50..8d21d9b1c2 100644 --- a/tests/lib/outputs/csv/csv_test.py +++ b/tests/lib/outputs/csv/csv_test.py @@ -5,67 +5,45 @@ from typing import List from unittest.mock import MagicMock import pytest +from freezegun import freeze_time from mock import patch +from prowler.config.config import prowler_version from prowler.lib.outputs.csv.csv import write_csv from prowler.lib.outputs.csv.models import CSV -from prowler.lib.outputs.finding import Finding, Severity, Status +from prowler.lib.outputs.finding import Finding from prowler.lib.outputs.output import Output - - -@pytest.fixture -def generate_finding(): - return Finding( - auth_method="OAuth", - timestamp=datetime.now(), - account_uid="12345", - account_name="Example Account", - account_email="example@example.com", - account_organization_uid="org-123", - account_organization_name="Example Org", - account_tags=["tag1", "tag2"], - finding_uid="finding-123", - provider="aws", - check_id="check-123", - check_title="Example Check", - check_type="Security", - status=Status("FAIL"), - status_extended="Extended status", - muted=False, - service_name="Example Service", - subservice_name="Example Subservice", - severity=Severity("critical"), - resource_type="Instance", - resource_uid="resource-123", - resource_name="Example Resource", - resource_details="Detailed information about the resource", - resource_tags="tag1,tag2", - partition="aws", - region="us-west-1", - description="Description of the finding", - risk="High", - related_url="http://example.com", - remediation_recommendation_text="Recommendation text", - remediation_recommendation_url="http://example.com/remediation", - remediation_code_nativeiac="native-iac-code", - remediation_code_terraform="terraform-code", - remediation_code_cli="cli-code", - remediation_code_other="other-code", - compliance={"compliance_key": "compliance_value"}, - categories="category1,category2", - depends_on="dependency", - related_to="related finding", - notes="Notes about the finding", - prowler_version="1.0", - ) +from tests.lib.outputs.fixtures.fixtures import generate_finding_output +from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, AWS_REGION_EU_WEST_1 class TestCSV: - def test_output_transform(self, generate_finding): - findings = [generate_finding] - - # Clear the data from CSV class - CSV._data = [] + def test_output_transform(self): + findings = [ + generate_finding_output( + status="PASS", + status_extended="status-extended", + resource_uid="resource-123", + resource_name="Example Resource", + resource_details="Detailed information about the resource", + resource_tags="tag1,tag2", + partition="aws", + description="Description of the finding", + risk="High", + related_url="http://example.com", + remediation_recommendation_text="Recommendation text", + remediation_recommendation_url="http://example.com/remediation", + remediation_code_nativeiac="native-iac-code", + remediation_code_terraform="terraform-code", + remediation_code_other="other-code", + remediation_code_cli="cli-code", + compliance={"compliance_key": "compliance_value"}, + categories="category1,category2", + depends_on="dependency", + related_to="related finding", + notes="Notes about the finding", + ) + ] output = CSV(findings) output_data = output.data[0] @@ -77,25 +55,25 @@ class TestCSV: assert isinstance(output_data["MUTED"], bool) assert isinstance(output_data["COMPLIANCE"], str) - assert output_data["AUTH_METHOD"] == "OAuth" - assert output_data["ACCOUNT_UID"] == "12345" - assert output_data["ACCOUNT_NAME"] == "Example Account" - assert output_data["ACCOUNT_EMAIL"] == "example@example.com" - assert output_data["ACCOUNT_ORGANIZATION_UID"] == "org-123" - assert output_data["ACCOUNT_ORGANIZATION_NAME"] == "Example Org" - assert output_data["ACCOUNT_TAGS"] == "tag1 | tag2" - assert output_data["FINDING_UID"] == "finding-123" + assert output_data["AUTH_METHOD"] == "profile: default" + assert output_data["ACCOUNT_UID"] == AWS_ACCOUNT_NUMBER + assert output_data["ACCOUNT_NAME"] == AWS_ACCOUNT_NUMBER + assert output_data["ACCOUNT_EMAIL"] == "" + assert output_data["ACCOUNT_ORGANIZATION_UID"] == "test-organization-id" + assert output_data["ACCOUNT_ORGANIZATION_NAME"] == "test-organization" + assert output_data["ACCOUNT_TAGS"] == "test-tag:test-value" + assert output_data["FINDING_UID"] == "test-unique-finding" assert output_data["PROVIDER"] == "aws" - assert output_data["CHECK_ID"] == "check-123" - assert output_data["CHECK_TITLE"] == "Example Check" - assert output_data["CHECK_TYPE"] == "Security" - assert output_data["STATUS"] == "FAIL" - assert output_data["STATUS_EXTENDED"] == "Extended status" + assert output_data["CHECK_ID"] == "test-check-id" + assert output_data["CHECK_TITLE"] == "test-check-id" + assert output_data["CHECK_TYPE"] == "test-type" + assert output_data["STATUS"] == "PASS" + assert output_data["STATUS_EXTENDED"] == "status-extended" assert output_data["MUTED"] is False - assert output_data["SERVICE_NAME"] == "Example Service" - assert output_data["SUBSERVICE_NAME"] == "Example Subservice" - assert output_data["SEVERITY"] == "critical" - assert output_data["RESOURCE_TYPE"] == "Instance" + assert output_data["SERVICE_NAME"] == "test-service" + assert output_data["SUBSERVICE_NAME"] == "" + assert output_data["SEVERITY"] == "high" + assert output_data["RESOURCE_TYPE"] == "test-resource" assert output_data["RESOURCE_UID"] == "resource-123" assert output_data["RESOURCE_NAME"] == "Example Resource" assert ( @@ -103,7 +81,7 @@ class TestCSV: ) assert output_data["RESOURCE_TAGS"] == "tag1,tag2" assert output_data["PARTITION"] == "aws" - assert output_data["REGION"] == "us-west-1" + assert output_data["REGION"] == AWS_REGION_EU_WEST_1 assert output_data["DESCRIPTION"] == "Description of the finding" assert output_data["RISK"] == "High" assert output_data["RELATED_URL"] == "http://example.com" @@ -121,13 +99,13 @@ class TestCSV: assert output_data["DEPENDS_ON"] == "dependency" assert output_data["RELATED_TO"] == "related finding" assert output_data["NOTES"] == "Notes about the finding" - assert output_data["PROWLER_VERSION"] == "1.0" + assert output_data["PROWLER_VERSION"] == prowler_version - def test_csv_write_to_file(self, generate_finding): + @freeze_time(datetime.now()) + def test_csv_write_to_file(self): mock_file = StringIO() - findings = [generate_finding] - # Clear the data from CSV class - CSV._data = [] + findings = [generate_finding_output()] + output = CSV(findings) output._file_descriptor = mock_file @@ -135,15 +113,13 @@ class TestCSV: output.batch_write_data_to_file() mock_file.seek(0) + expected_csv = f"AUTH_METHOD;TIMESTAMP;ACCOUNT_UID;ACCOUNT_NAME;ACCOUNT_EMAIL;ACCOUNT_ORGANIZATION_UID;ACCOUNT_ORGANIZATION_NAME;ACCOUNT_TAGS;FINDING_UID;PROVIDER;CHECK_ID;CHECK_TITLE;CHECK_TYPE;STATUS;STATUS_EXTENDED;MUTED;SERVICE_NAME;SUBSERVICE_NAME;SEVERITY;RESOURCE_TYPE;RESOURCE_UID;RESOURCE_NAME;RESOURCE_DETAILS;RESOURCE_TAGS;PARTITION;REGION;DESCRIPTION;RISK;RELATED_URL;REMEDIATION_RECOMMENDATION_TEXT;REMEDIATION_RECOMMENDATION_URL;REMEDIATION_CODE_NATIVEIAC;REMEDIATION_CODE_TERRAFORM;REMEDIATION_CODE_CLI;REMEDIATION_CODE_OTHER;COMPLIANCE;CATEGORIES;DEPENDS_ON;RELATED_TO;NOTES;PROWLER_VERSION\r\nprofile: default;{datetime.now()};123456789012;123456789012;;test-organization-id;test-organization;test-tag:test-value;test-unique-finding;aws;test-check-id;test-check-id;test-type;PASS;;False;test-service;;high;test-resource;;;;;aws;eu-west-1;check description;test-risk;test-url;;;;;;;test-compliance: test-compliance;test-category;test-dependency;test-related-to;test-notes;4.2.4\r\n" content = mock_file.read() - content = content.split("PROWLER_VERSION")[1] - content = content.removeprefix("\r\n") - content = content.removesuffix("\r\n") - string = "" - for value in output.data[0].values(): - string += f"{value};" - string = string.removesuffix(";") - assert string in content + + assert content == expected_csv + + def test_batch_write_data_to_file_without_findings(self): + assert not hasattr(CSV([]), "_file_descriptor") @pytest.fixture def mock_output_class(self): diff --git a/tests/lib/outputs/fixtures/fixtures.py b/tests/lib/outputs/fixtures/fixtures.py index 62ca5b6acf..909995f52c 100644 --- a/tests/lib/outputs/fixtures/fixtures.py +++ b/tests/lib/outputs/fixtures/fixtures.py @@ -16,12 +16,26 @@ def generate_finding_output( resource_name: str = "", resource_tags: str = "", compliance: dict = {"test-compliance": "test-compliance"}, - timestamp: datetime = datetime.now(), + timestamp: datetime = None, provider: str = "aws", + partition: str = "aws", + description: str = "check description", + risk: str = "test-risk", + related_url: str = "test-url", + remediation_recommendation_text: str = "", + remediation_recommendation_url: str = "", + remediation_code_nativeiac: str = "", + remediation_code_terraform: str = "", + remediation_code_cli: str = "", + remediation_code_other: str = "", + categories: str = "test-category", + depends_on: str = "test-dependency", + related_to: str = "test-related-to", + notes: str = "test-notes", ) -> Finding: return Finding( auth_method="profile: default", - timestamp=timestamp, + timestamp=timestamp if timestamp else datetime.now(), account_uid=AWS_ACCOUNT_NUMBER, account_name=AWS_ACCOUNT_NUMBER, account_email="", @@ -44,21 +58,21 @@ def generate_finding_output( resource_name=resource_name, resource_details=resource_details, resource_tags=resource_tags, - partition=provider, + partition=partition, region=region, - description="check description", - risk="test-risk", - related_url="test-url", - remediation_recommendation_text="", - remediation_recommendation_url="", - remediation_code_nativeiac="", - remediation_code_terraform="", - remediation_code_cli="", - remediation_code_other="", + description=description, + risk=risk, + related_url=related_url, + remediation_recommendation_text=remediation_recommendation_text, + remediation_recommendation_url=remediation_recommendation_url, + remediation_code_nativeiac=remediation_code_nativeiac, + remediation_code_terraform=remediation_code_terraform, + remediation_code_cli=remediation_code_cli, + remediation_code_other=remediation_code_other, compliance=compliance, - categories="test-category", - depends_on="test-dependency", - related_to="test-related-to", - notes="test-notes", + categories=categories, + depends_on=depends_on, + related_to=related_to, + notes=notes, prowler_version=prowler_version, ) diff --git a/tests/lib/outputs/html/html_test.py b/tests/lib/outputs/html/html_test.py index 6c906b08d0..77f900c211 100644 --- a/tests/lib/outputs/html/html_test.py +++ b/tests/lib/outputs/html/html_test.py @@ -1,6 +1,8 @@ import sys from io import StringIO +from mock import patch + from prowler.config.config import timestamp from prowler.lib.outputs.html.html import HTML from tests.lib.outputs.fixtures.fixtures import generate_finding_output @@ -214,8 +216,18 @@ kubernetes_html_assessment_summary = """ """ -aws_html_header = ( + +def __get_aws_html_header__(args: list) -> str: """ + Generate the HTML header for AWS + + Args: + args (list): List of arguments passed to the script + + Returns: + str: HTML header for AWS + """ + aws_html_header = f""" @@ -223,11 +235,11 @@ aws_html_header = ( Prowler - The Handy Cloud Security Tool @@ -269,20 +281,14 @@ aws_html_header = (
  • - Parameters used: """ - + " ".join(sys.argv[1:]) - + """ + Parameters used: {" ".join(args)}
  • - Date: """ - + timestamp.isoformat() - + """ + Date: {timestamp.isoformat()}
    • - """ - + aws_html_assessment_summary - + """ + {aws_html_assessment_summary}
    @@ -326,7 +332,8 @@ aws_html_header = ( """ -) + return aws_html_header + html_footer = """ @@ -448,11 +455,18 @@ class TestHTML: output._file_descriptor = mock_file provider = set_mocked_aws_provider(audited_regions=[AWS_REGION_EU_WEST_1]) - output.batch_write_data_to_file(provider, html_stats) + with patch.object(mock_file, "close", return_value=None): + output.batch_write_data_to_file(provider, html_stats) mock_file.seek(0) content = mock_file.read() - assert content == aws_html_header + pass_html_finding + html_footer + args = sys.argv[1:] + assert ( + content == __get_aws_html_header__(args) + pass_html_finding + html_footer + ) + + def test_batch_write_data_to_file_without_findings(self): + assert not hasattr(HTML([]), "_file_descriptor") def test_write_header(self): mock_file = StringIO() @@ -465,7 +479,8 @@ class TestHTML: mock_file.seek(0) content = mock_file.read() - assert content == aws_html_header + args = sys.argv[1:] + assert content == __get_aws_html_header__(args) def test_write_footer(self): mock_file = StringIO() diff --git a/tests/lib/outputs/ocsf/ocsf_test.py b/tests/lib/outputs/ocsf/ocsf_test.py index 9a6dfaa1b8..9903f8e643 100644 --- a/tests/lib/outputs/ocsf/ocsf_test.py +++ b/tests/lib/outputs/ocsf/ocsf_test.py @@ -2,6 +2,7 @@ import json from datetime import datetime from io import StringIO +from freezegun import freeze_time from mock import patch from py_ocsf_models.events.base_event import SeverityID, StatusID from py_ocsf_models.events.findings.detection_finding import DetectionFinding @@ -23,82 +24,9 @@ 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: + # TODO: improve this test checking the fields def test_transform(self): findings = [ generate_finding_output( @@ -155,6 +83,7 @@ class TestOCSF: "compliance": findings[0].compliance, } + @freeze_time(datetime.now()) def test_batch_write_data_to_file(self): mock_file = StringIO() findings = [ @@ -163,7 +92,7 @@ class TestOCSF: severity="low", muted=False, region=AWS_REGION_EU_WEST_1, - timestamp=now, + timestamp=datetime.now(), resource_details="resource_details", resource_name="resource_name", resource_uid="resource-id", @@ -171,6 +100,80 @@ class TestOCSF: ) ] + expected_json_output = [ + { + "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": datetime.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": datetime.now().isoformat(), + "remediation": {"desc": "", "references": []}, + "risk_details": "test-risk", + "type_uid": 200401, + "type_name": "Create", + } + ] + output = OCSF(findings) output._file_descriptor = mock_file @@ -180,7 +183,10 @@ class TestOCSF: mock_file.seek(0) content = mock_file.read() - assert json.loads(content) == json.loads(expected_json_output) + assert json.loads(content) == expected_json_output + + def test_batch_write_data_to_file_without_findings(self): + assert not hasattr(OCSF([]), "_file_descriptor") def test_finding_output_cloud_pass_low_muted(self): finding_output = generate_finding_output(