-
- GCP Assessment Summary
+
-
+
-
+
-
+ GCP Assessment Summary
+
+ -
+
- + GCP Project IDs: {", ".join(provider.project_ids)} + +
-
-
- - GCP Project IDs: {", ".join(provider.project_ids)} - -
-
-
"""
+ except Exception as error:
+ logger.error(
+ f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
+ )
+ return ""
+ @staticmethod
+ def get_kubernetes_assessment_summary(provider: Provider) -> str:
+ """
+ get_kubernetes_assessment_summary gets the HTML assessment summary for the provider
-def get_kubernetes_html_assessment_summary(provider):
- try:
- if provider.__class__.__name__ == "KubernetesProvider":
+ Args:
+ provider (Provider): the provider object
+
+ Returns:
+ str: the HTML assessment summary
+ """
+ try:
return f"""
-
- GCP Credentials
+
- """
- except Exception as error:
- logger.critical(
- f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
- )
- sys.exit(1)
+
+
-
+
-
+ GCP Credentials
+
+ -
+
- + GCP Account: {profile} + +
-
-
- - GCP Account: {profile} - -
-
-
- Kubernetes Assessment Summary
+
-
+
-
+
-
+ Kubernetes Assessment Summary
+
+ -
+
- + Kubernetes Cluster: {provider.identity.cluster} + +
-
-
- - Kubernetes Cluster: {provider._identity.cluster} - -
-
+ PASS
+ high
+ test-service
+ eu-west-1
+ test-check-id
+ test-check-id
+
+
+ """
+fail_html_finding = """
+
+ FAIL
+ high
+ test-service
+ eu-west-1
+ test-check-id
+ test-check-id
+
+
+ """
+muted_html_finding = """
+
+ MUTED (PASS)
+ high
+ test-service
+ eu-west-1
+ test-check-id
+ test-check-id
+
+
+ """
+manual_html_finding = """
+
+ MANUAL
+ high
+ test-service
+ eu-west-1
+ test-check-id
+ test-check-id
+
+
+ """
+aws_html_assessment_summary = """
+ Prowler - The Handy Cloud Security Tool
+
+
+
-
"""
+ except Exception as error:
+ logger.error(
+ f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
+ )
+ return ""
+ @staticmethod
+ def get_assessment_summary(provider: Provider) -> str:
+ """
+ get_assessment_summary gets the HTML assessment summary for the provider
-def get_assessment_summary(provider):
- """
- get_assessment_summary gets the HTML assessment summary for the provider
- """
- try:
- # This is based in the Provider_provider class
- # It is not pretty but useful
- # AWS_provider --> aws
- # GCP_provider --> gcp
- # Azure_provider --> azure
+ Args:
+ provider (Provider): the provider object
- # Dynamically get the Provider quick inventory handler
- provider_html_assessment_summary_function = (
- f"get_{provider.type}_html_assessment_summary"
- )
- return getattr(
- importlib.import_module(__name__), provider_html_assessment_summary_function
- )(provider)
- except Exception as error:
- logger.critical(
- f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
- )
- sys.exit(1)
+ Returns:
+ str: the HTML assessment summary
+ """
+ try:
+ # This is based in the Provider_provider class
+ # It is not pretty but useful
+ # AWS_provider --> aws
+ # GCP_provider --> gcp
+ # Azure_provider --> azure
+ # Kubernetes_provider --> kubernetes
+
+ # Dynamically get the Provider quick inventory handler
+ provider_html_assessment_summary_function = (
+ f"get_{provider.type}_assessment_summary"
+ )
+ return getattr(HTML, provider_html_assessment_summary_function)(provider)
+ except Exception as error:
+ logger.error(
+ f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
+ )
+ return ""
diff --git a/prowler/lib/outputs/ocsf/ocsf.py b/prowler/lib/outputs/ocsf/ocsf.py
index 5fc5663aea..7a677f353a 100644
--- a/prowler/lib/outputs/ocsf/ocsf.py
+++ b/prowler/lib/outputs/ocsf/ocsf.py
@@ -168,7 +168,11 @@ class OCSF(Output):
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:
+ if (
+ getattr(self, "_file_descriptor", None)
+ and not self._file_descriptor.closed
+ and self._data
+ ):
self._file_descriptor.write("[")
for finding in self._data:
self._file_descriptor.write(
diff --git a/prowler/lib/outputs/output.py b/prowler/lib/outputs/output.py
index f9dfde3752..9abc267deb 100644
--- a/prowler/lib/outputs/output.py
+++ b/prowler/lib/outputs/output.py
@@ -34,9 +34,10 @@ class Output(ABC):
file_path: str = None,
) -> None:
self._data = []
- self.transform(findings)
- if create_file_descriptor:
- self.create_file_descriptor(file_path)
+ if findings:
+ self.transform(findings)
+ if create_file_descriptor:
+ self.create_file_descriptor(file_path)
@property
def data(self):
diff --git a/prowler/lib/outputs/outputs.py b/prowler/lib/outputs/outputs.py
index 252e6d730b..88c903ac42 100644
--- a/prowler/lib/outputs/outputs.py
+++ b/prowler/lib/outputs/outputs.py
@@ -7,8 +7,6 @@ from prowler.lib.outputs.compliance.compliance import (
fill_compliance,
)
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
def stdout_report(finding, color, verbose, status, fix):
@@ -92,12 +90,6 @@ def report(check_findings, provider):
input_compliance_frameworks,
)
- # Common Output Data
- finding_output = Finding.generate_output(provider, finding)
-
- if "html" in file_descriptors:
- fill_html(file_descriptors["html"], finding_output)
-
else: # No service resources in the whole account
color = set_report_color("MANUAL")
if output_options.verbose:
diff --git a/tests/lib/outputs/html/html_test.py b/tests/lib/outputs/html/html_test.py
new file mode 100644
index 0000000000..6c906b08d0
--- /dev/null
+++ b/tests/lib/outputs/html/html_test.py
@@ -0,0 +1,516 @@
+import sys
+from io import StringIO
+
+from prowler.config.config import timestamp
+from prowler.lib.outputs.html.html import HTML
+from tests.lib.outputs.fixtures.fixtures import generate_finding_output
+from tests.providers.aws.utils import AWS_REGION_EU_WEST_1, set_mocked_aws_provider
+from tests.providers.azure.azure_fixtures import set_mocked_azure_provider
+from tests.providers.gcp.gcp_fixtures import GCP_PROJECT_ID, set_mocked_gcp_provider
+from tests.providers.kubernetes.kubernetes_fixtures import (
+ set_mocked_kubernetes_provider,
+)
+
+html_stats = {
+ "total_pass": 0,
+ "total_fail": 1,
+ "resources_count": 1,
+ "findings_count": 1,
+}
+pass_html_finding = """
+
- Kubernetes Credentials
+
- """
- except Exception as error:
- logger.critical(
- f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
- )
- sys.exit(1)
+
+
-
+
-
+ Kubernetes Credentials
+
+ -
+
- + Kubernetes Context: {provider.identity.context} + +
-
-
- - Kubernetes Context: {provider._identity.context} - -
test-risk
+•test-compliance: test-compliance +
test-risk
+•test-compliance: test-compliance +
test-risk
+•test-compliance: test-compliance +
test-risk
+•test-compliance: test-compliance +
+
+
+
+
+ AWS Assessment Summary
+
+ -
+
- + AWS Account: 123456789012 + +
- + AWS-CLI Profile: default + +
- + Audited Regions: eu-west-1 + +
+
"""
+
+azure_html_assessment_summary = """
+
+
+
+
+ AWS Credentials
+
+ -
+
- + User Id: None + +
- + Caller Identity ARN: None + +
+
+
+
+
+ Azure Assessment Summary
+
+ -
+
- + Azure Tenant IDs: 00000000-0000-0000-0000-000000000000 + +
- + Azure Tenant Domain: Unknown tenant domain (missing AAD permissions) + +
- + Azure Subscriptions: 4f647f43-15d2-4e3a-a7f0-8517cc4d977b : Subscription Name + +
+
"""
+
+gcp_html_assessment_summary = """
+
+
+
+ Azure Credentials
+
+ -
+
- + Azure Identity Type: Service Principal + +
- + Azure Identity ID: 00000000-0000-0000-0000-000000000000 + +
+
+
+
+
+ GCP Assessment Summary
+
+ -
+
- + GCP Project IDs: 123456789012 + +
+
"""
+
+kubernetes_html_assessment_summary = """
+
+
+
+ GCP Credentials
+
+ -
+
- + GCP Account: test@test.com + +
+
+
+
+
+ Kubernetes Assessment Summary
+
+ -
+
- + Kubernetes Cluster: None + +
+
"""
+
+aws_html_header = (
+ """
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ Kubernetes Credentials
+
+ -
+
- + Kubernetes Context: None + +
+
+
+
+
+ 
+
"""
+ + aws_html_assessment_summary
+ + """
+
+
+
+
+ Report Information
+
+ -
+
-
+ +++ Version: 4.2.4 ++
+
- + Parameters used: """ + + " ".join(sys.argv[1:]) + + """ + +
- + Date: """ + + timestamp.isoformat() + + """ + +
+
+
+
+
+ Assessment Overview
+
+ -
+
- + Total Findings: 1 + +
- + Passed: 0 + +
- + Failed: 1 + +
- + Total Resources: 1 + +
+
+
+
+
+
+
+
+
+
+
+
+
+"""
+
+
+class TestHTML:
+ def test_transform_fail_finding(self):
+ findings = [generate_finding_output(status="FAIL")]
+ html = HTML(findings)
+ output_data = html.data[0]
+ assert isinstance(output_data, str)
+ assert output_data == fail_html_finding
+
+ def test_transform_pass_finding(self):
+ findings = [generate_finding_output()]
+ html = HTML(findings)
+ output_data = html.data[0]
+ assert isinstance(output_data, str)
+ assert output_data == pass_html_finding
+
+ def test_transform_muted_finding(self):
+ findings = [generate_finding_output(muted=True)]
+ html = HTML(findings)
+ output_data = html.data[0]
+ assert isinstance(output_data, str)
+ assert output_data == muted_html_finding
+
+ def test_transform_manual_finding(self):
+ findings = [generate_finding_output(status="MANUAL")]
+ html = HTML(findings)
+ output_data = html.data[0]
+ assert isinstance(output_data, str)
+ assert output_data == manual_html_finding
+
+ def test_batch_write_data_to_file(self):
+ mock_file = StringIO()
+ findings = [generate_finding_output()]
+ output = HTML(findings)
+ 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)
+
+ mock_file.seek(0)
+ content = mock_file.read()
+ assert content == aws_html_header + pass_html_finding + html_footer
+
+ def test_write_header(self):
+ mock_file = StringIO()
+ findings = [generate_finding_output()]
+ output = HTML(findings)
+ output._file_descriptor = mock_file
+ provider = set_mocked_aws_provider(audited_regions=[AWS_REGION_EU_WEST_1])
+
+ output.write_header(mock_file, provider, html_stats)
+
+ mock_file.seek(0)
+ content = mock_file.read()
+ assert content == aws_html_header
+
+ def test_write_footer(self):
+ mock_file = StringIO()
+ findings = [generate_finding_output()]
+ output = HTML(findings)
+ output._file_descriptor = mock_file
+
+ output.write_footer(mock_file)
+
+ mock_file.seek(0)
+ content = mock_file.read()
+ assert content == html_footer
+
+ def test_aws_get_assessment_summary(self):
+ findings = [generate_finding_output()]
+ output = HTML(findings)
+ provider = set_mocked_aws_provider(audited_regions=[AWS_REGION_EU_WEST_1])
+
+ summary = output.get_assessment_summary(provider)
+
+ assert summary == aws_html_assessment_summary
+
+ def test_azure_get_assessment_summary(self):
+ findings = [generate_finding_output()]
+ output = HTML(findings)
+ provider = set_mocked_azure_provider()
+
+ summary = output.get_assessment_summary(provider)
+
+ assert summary == summary
+
+ def test_gcp_get_assessment_summary(self):
+ findings = [generate_finding_output()]
+ output = HTML(findings)
+ provider = set_mocked_gcp_provider(project_ids=[GCP_PROJECT_ID])
+
+ summary = output.get_assessment_summary(provider)
+
+ assert summary == gcp_html_assessment_summary
+
+ def test_kubernetes_get_assessment_summary(self):
+ findings = [generate_finding_output()]
+ output = HTML(findings)
+ provider = set_mocked_kubernetes_provider()
+
+ summary = output.get_assessment_summary(provider)
+
+ assert summary == kubernetes_html_assessment_summary
diff --git a/tests/lib/outputs/outputs_test.py b/tests/lib/outputs/outputs_test.py
index 58a9ec510b..b4c12ea62e 100644
--- a/tests/lib/outputs/outputs_test.py
+++ b/tests/lib/outputs/outputs_test.py
@@ -1,11 +1,9 @@
-import os
-from os import path, remove
+from os import path
from unittest import mock
import pytest
from colorama import Fore
-from prowler.config.config import html_file_suffix, output_file_timestamp
from prowler.lib.check.compliance_models import (
CIS_Requirement_Attribute,
Compliance_Base_Model,
@@ -14,7 +12,6 @@ from prowler.lib.check.compliance_models import (
from prowler.lib.check.models import Check_Report, load_check_metadata
from prowler.lib.outputs.compliance.compliance import get_check_compliance
from prowler.lib.outputs.csv.csv import generate_csv_fields
-from prowler.lib.outputs.file_descriptors import fill_file_descriptors
from prowler.lib.outputs.finding import Finding
from prowler.lib.outputs.outputs import extract_findings_statistics, set_report_color
from prowler.lib.outputs.utils import (
@@ -25,41 +22,9 @@ from prowler.lib.outputs.utils import (
unroll_list,
unroll_tags,
)
-from prowler.lib.utils.utils import open_file
-from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, set_mocked_aws_provider
class TestOutputs:
- def test_fill_file_descriptors_aws(self):
- audited_account = AWS_ACCOUNT_NUMBER
- output_directory = f"{os.path.dirname(os.path.realpath(__file__))}"
- aws_provider = set_mocked_aws_provider()
- test_output_modes = [
- ["html"],
- ]
- output_filename = f"prowler-output-{audited_account}-{output_file_timestamp}"
- expected = [
- {
- "html": open_file(
- f"{output_directory}/{output_filename}{html_file_suffix}",
- "a",
- )
- },
- ]
-
- for index, output_mode_list in enumerate(test_output_modes):
- test_output_file_descriptors = fill_file_descriptors(
- output_mode_list,
- output_directory,
- output_filename,
- aws_provider,
- )
- for output_mode in output_mode_list:
- assert (
- test_output_file_descriptors[output_mode].name
- == expected[index][output_mode].name
- )
- remove(expected[index][output_mode].name)
def test_set_report_color(self):
test_status = ["PASS", "FAIL", "MANUAL"]
diff --git a/tests/providers/aws/services/autoscaling/autoscaling_service_test.py b/tests/providers/aws/services/autoscaling/autoscaling_service_test.py
index fe85a0725d..dbc4f05381 100644
--- a/tests/providers/aws/services/autoscaling/autoscaling_service_test.py
+++ b/tests/providers/aws/services/autoscaling/autoscaling_service_test.py
@@ -94,7 +94,7 @@ class Test_AutoScaling_Service:
KeyName="the_keys",
SecurityGroups=["default", "default2"],
)
- asg = autoscaling_client.create_auto_scaling_group(
+ _ = autoscaling_client.create_auto_scaling_group(
AutoScalingGroupName="my-autoscaling-group",
LaunchConfigurationName="test",
MinSize=0,
@@ -112,7 +112,6 @@ class Test_AutoScaling_Service:
# AutoScaling client for this test class
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
autoscaling = AutoScaling(aws_provider)
- print("asg", asg)
assert len(autoscaling.groups) == 1
# create_auto_scaling_group doesn't return the ARN, can't check it
# assert autoscaling.groups[0].arn ==
diff --git a/tests/providers/aws/services/iam/iam_inline_policy_allows_privilege_escalation/iam_inline_policy_allows_privilege_escalation_test.py b/tests/providers/aws/services/iam/iam_inline_policy_allows_privilege_escalation/iam_inline_policy_allows_privilege_escalation_test.py
index 4e5162334a..00e4c57b1e 100644
--- a/tests/providers/aws/services/iam/iam_inline_policy_allows_privilege_escalation/iam_inline_policy_allows_privilege_escalation_test.py
+++ b/tests/providers/aws/services/iam/iam_inline_policy_allows_privilege_escalation/iam_inline_policy_allows_privilege_escalation_test.py
@@ -516,7 +516,6 @@ class Test_iam_inline_policy_allows_privilege_escalation:
policy_name = "privileged_policy"
for values in privilege_escalation_policies_combination.values():
- print(list(values))
# We create a new statement in each loop with the combinations required to allow the privilege escalation
policy_document = {
"Version": "2012-10-17",
diff --git a/tests/providers/aws/services/iam/iam_policy_allows_privilege_escalation/iam_policy_allows_privilege_escalation_test.py b/tests/providers/aws/services/iam/iam_policy_allows_privilege_escalation/iam_policy_allows_privilege_escalation_test.py
index ea004609c0..41fe22aaca 100644
--- a/tests/providers/aws/services/iam/iam_policy_allows_privilege_escalation/iam_policy_allows_privilege_escalation_test.py
+++ b/tests/providers/aws/services/iam/iam_policy_allows_privilege_escalation/iam_policy_allows_privilege_escalation_test.py
@@ -454,7 +454,6 @@ class Test_iam_policy_allows_privilege_escalation:
iam_client = client("iam", region_name=AWS_REGION_US_EAST_1)
policy_name = "privileged_policy"
for values in privilege_escalation_policies_combination.values():
- print(list(values))
# We create a new statement in each loop with the combinations required to allow the privilege escalation
policy_document = {
"Version": "2012-10-17",
diff --git a/tests/providers/gcp/gcp_fixtures.py b/tests/providers/gcp/gcp_fixtures.py
index 4ffb13e918..4968fc77b3 100644
--- a/tests/providers/gcp/gcp_fixtures.py
+++ b/tests/providers/gcp/gcp_fixtures.py
@@ -13,7 +13,8 @@ def set_mocked_gcp_provider(
) -> GcpProvider:
provider = MagicMock()
provider.type = "gcp"
- provider.session = None
+ provider.session = MagicMock()
+ provider.session._service_account_email = "test@test.com"
provider.project_ids = project_ids
provider.identity = GCPIdentityInfo(
profile=profile,
diff --git a/tests/providers/kubernetes/kubernetes_provider_test.py b/tests/providers/kubernetes/kubernetes_provider_test.py
index 5579a9bc50..ba714ee51a 100644
--- a/tests/providers/kubernetes/kubernetes_provider_test.py
+++ b/tests/providers/kubernetes/kubernetes_provider_test.py
@@ -55,7 +55,6 @@ class TestKubernetesProvider:
# Instantiate the KubernetesProvider with mocked arguments
kubernetes_provider = KubernetesProvider(arguments)
- print(kubernetes_provider.__dict__)
assert isinstance(kubernetes_provider.session, KubernetesSession)
assert kubernetes_provider.session.api_client is not None
assert kubernetes_provider.session.context == {
+
+
+
+
+ | Status | +Severity | +Service Name | +Region | +Check ID | +Check Title | +Resource ID | +Resource Tags | +Status Extended | +Risk | +Recomendation | +Compliance | +
|---|