mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 12:31:54 +00:00
chore(ocsf): add OCSF class for outputs (#4355)
This commit is contained in:
Pedro Martín
+16
-8
@@ -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(
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
@@ -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}"
|
||||
)
|
||||
@@ -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
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user