chore(html): add HTML class (#4360)

This commit is contained in:
Sergio Garcia authored and GitHub committed 2024-07-04 13:28:09 -04:00
1 parent 673619c8a1
commit bf83f38c89
15 files changed
+877 -370

No files matched your search

+14 -9
View File
@@ -9,6 +9,7 @@ from colorama import Fore, Style
from prowler.config.config import (
csv_file_suffix,
get_available_compliance_frameworks,
html_file_suffix,
json_asff_file_suffix,
json_ocsf_file_suffix,
)
@@ -45,7 +46,7 @@ from prowler.lib.outputs.asff.asff import ASFF
from prowler.lib.outputs.compliance.compliance import display_compliance_table
from prowler.lib.outputs.csv.models import CSV
from prowler.lib.outputs.finding import Finding
from prowler.lib.outputs.html.html import add_html_footer, fill_html_overview_statistics
from prowler.lib.outputs.html.html import HTML
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
@@ -338,15 +339,19 @@ def prowler():
)
json_finding.batch_write_data_to_file()
if "html" in mode:
# TODO: generate HTML here
add_html_footer(
global_provider.output_options.output_filename,
global_provider.output_options.output_directory,
# Generate HTML Finding Object
filename = (
f"{global_provider.output_options.output_directory}/"
f"{global_provider.output_options.output_filename}{html_file_suffix}"
)
fill_html_overview_statistics(
stats,
global_provider.output_options.output_filename,
global_provider.output_options.output_directory,
html_finding = HTML(
findings=finding_outputs,
create_file_descriptor=True,
file_path=filename,
)
# Write HTML Finding Object to file
html_finding.batch_write_data_to_file(
provider=global_provider, stats=stats
)
# Send output to S3 if needed (-B / -D)
+5 -1
View File
@@ -127,7 +127,11 @@ class ASFF(Output):
None
"""
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
):
# Write JSON opening/header [
self._file_descriptor.write("[")
+5 -1
View File
@@ -30,7 +30,11 @@ class CSV(Output):
def batch_write_data_to_file(self) -> None:
"""Writes the findings to a file using the CSV format using the `Output._file_descriptor`."""
try:
if self._file_descriptor and not self._file_descriptor.closed:
if (
getattr(self, "_file_descriptor", None)
and not self._file_descriptor.closed
and self._data
):
csv_writer = DictWriter(
self._file_descriptor,
fieldnames=self._data[0].keys(),
+11 -17
View File
@@ -2,7 +2,7 @@ from csv import DictWriter
from io import TextIOWrapper
from typing import Any
from prowler.config.config import csv_file_suffix, html_file_suffix
from prowler.config.config import csv_file_suffix
from prowler.lib.logger import logger
from prowler.lib.outputs.compliance.mitre_attack.models import (
MitreAttackAWS,
@@ -20,7 +20,6 @@ from prowler.lib.outputs.compliance.models import (
Check_Output_CSV_KUBERNETES_CIS,
)
from prowler.lib.outputs.csv.csv import generate_csv_fields
from prowler.lib.outputs.html.html import add_html_header
from prowler.lib.outputs.output import Finding
from prowler.lib.utils.utils import file_exists, open_file
@@ -44,16 +43,13 @@ def initialize_file_descriptor(
filename,
"a",
)
if "html" in output_mode:
add_html_header(file_descriptor, provider)
else:
# Format is the class model of the CSV format to print the headers
csv_header = [x.upper() for x in generate_csv_fields(format)]
csv_writer = DictWriter(
file_descriptor, fieldnames=csv_header, delimiter=";"
)
if write_header:
csv_writer.writeheader()
# Format is the class model of the CSV format to print the headers
csv_header = [x.upper() for x in generate_csv_fields(format)]
csv_writer = DictWriter(
file_descriptor, fieldnames=csv_header, delimiter=";"
)
if write_header:
csv_writer.writeheader()
return file_descriptor
except Exception as error:
logger.error(
@@ -69,14 +65,12 @@ def fill_file_descriptors(output_modes, output_directory, output_filename, provi
# FIXME: Remove this once we always use the new CSV(Output)
if output_mode == "csv":
continue
elif output_mode == "json-ocsf":
continue
elif output_mode == "json-asff":
continue
elif output_mode == "html":
filename = f"{output_directory}/{output_filename}{html_file_suffix}"
file_descriptor = initialize_file_descriptor(
filename, output_mode, provider
)
file_descriptors.update({output_mode: file_descriptor})
continue
elif provider.type == "gcp":
filename = f"{output_directory}/compliance/{output_filename}_{output_mode}{csv_file_suffix}"
+313 -288
View File
@@ -1,27 +1,104 @@
import html
import importlib
import sys
from os import path
from io import TextIOWrapper
from prowler.config.config import (
enconding_format_utf_8,
html_file_suffix,
html_logo_url,
prowler_version,
square_logo_img,
timestamp,
)
from prowler.lib.logger import logger
from prowler.lib.outputs.output import Finding
from prowler.lib.outputs.output import Finding, Output
from prowler.lib.outputs.utils import parse_html_string, unroll_dict
from prowler.lib.utils.utils import open_file
from prowler.providers.common.provider import Provider
def add_html_header(file_descriptor, provider):
try:
file_descriptor.write(
f"""
<!DOCTYPE html>
class HTML(Output):
def transform(self, findings: list[Finding]) -> None:
"""Transforms the findings into the HTML format.
Args:
findings (list[Finding]): a list of Finding objects
"""
try:
for finding in findings:
row_class = "p-3 mb-2 bg-success-custom"
finding_status = finding.status.value
# Change the status of the finding if it's muted
if finding.muted:
finding_status = f"MUTED ({finding_status})"
row_class = "table-warning"
if finding.status == "MANUAL":
row_class = "table-info"
elif finding.status == "FAIL":
row_class = "table-danger"
self._data.append(
f"""
<tr class="{row_class}">
<td>{finding_status}</td>
<td>{finding.severity.value}</td>
<td>{finding.service_name}</td>
<td>{finding.region.lower()}</td>
<td>{finding.check_id.replace("_", "<wbr />_")}</td>
<td>{finding.check_title}</td>
<td>{finding.resource_uid.replace("<", "&lt;").replace(">", "&gt;").replace("_", "<wbr />_")}</td>
<td>{parse_html_string(finding.resource_tags)}</td>
<td>{finding.status_extended.replace("<", "&lt;").replace(">", "&gt;").replace("_", "<wbr />_")}</td>
<td><p class="show-read-more">{html.escape(finding.risk)}</p></td>
<td><p class="show-read-more">{html.escape(finding.remediation_recommendation_text)}</p> <a class="read-more" href="{finding.remediation_recommendation_url}"><i class="fas fa-external-link-alt"></i></a></td>
<td><p class="show-read-more">{parse_html_string(unroll_dict(finding.compliance))}</p></td>
</tr>
"""
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def batch_write_data_to_file(self, provider: Provider, stats: dict) -> None:
"""
Writes the findings to a file using the HTML format using the `Output._file_descriptor`.
Args:
provider (Provider): the provider object
output_filename (str): the name of the output file
output_directory (str): the directory where the output file will be saved
stats (dict): the statistics of the findings
"""
try:
if (
getattr(self, "_file_descriptor", None)
and not self._file_descriptor.closed
and self._data
):
HTML.write_header(self._file_descriptor, provider, stats)
for finding in self._data:
self._file_descriptor.write(finding)
HTML.write_footer(self._file_descriptor)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
@staticmethod
def write_header(
file_descriptor: TextIOWrapper, provider: Provider, stats: dict
) -> None:
"""
Writes the header of the HTML file.
Args:
file_descriptor (file): the file descriptor to write the header
provider (Provider): the provider object
stats (dict): the statistics of the findings
"""
try:
file_descriptor.write(
f"""
<!DOCTYPE html>
<html lang="en">
<head>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
@@ -81,7 +158,7 @@ def add_html_header(file_descriptor, provider):
</li>
</ul>
</div>
</div> {get_assessment_summary(provider)}
</div>{HTML.get_assessment_summary(provider)}
<div class="col-md-2">
<div class="card">
<div class="card-header">
@@ -89,16 +166,16 @@ def add_html_header(file_descriptor, provider):
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>Total Findings:</b> TOTAL_FINDINGS
<b>Total Findings:</b> {str(stats.get("findings_count", 0))}
</li>
<li class="list-group-item">
<b>Passed:</b> TOTAL_PASS
<b>Passed:</b> {str(stats.get("total_pass", 0))}
</li>
<li class="list-group-item">
<b>Failed:</b> TOTAL_FAIL
<b>Failed:</b> {str(stats.get("total_fail", 0))}
</li>
<li class="list-group-item">
<b>Total Resources:</b> TOTAL_RESOURCES
<b>Total Resources:</b> {str(stats.get("resources_count", 0))}
</li>
</ul>
</div>
@@ -124,107 +201,25 @@ def add_html_header(file_descriptor, provider):
<th scope="col">Compliance</th>
</tr>
</thead>
<tbody>
"""
)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
sys.exit(1)
def fill_html(file_descriptor, finding: Finding):
try:
row_class = "p-3 mb-2 bg-success-custom"
finding_status = finding.status.value
# Change the status of the finding if it's muted
if finding.muted:
finding_status = f"MUTED ({finding_status})"
row_class = "table-warning"
if finding.status == "MANUAL":
row_class = "table-info"
elif finding.status == "FAIL":
row_class = "table-danger"
file_descriptor.write(
f"""
<tr class="{row_class}">
<td>{finding_status}</td>
<td>{finding.severity.value}</td>
<td>{finding.service_name}</td>
<td>{finding.region.lower()}</td>
<td>{finding.check_id.replace("_", "<wbr />_")}</td>
<td>{finding.check_title}</td>
<td>{finding.resource_uid.replace("<", "&lt;").replace(">", "&gt;").replace("_", "<wbr />_")}</td>
<td>{parse_html_string(finding.resource_tags)}</td>
<td>{finding.status_extended.replace("<", "&lt;").replace(">", "&gt;").replace("_", "<wbr />_")}</td>
<td><p class="show-read-more">{html.escape(finding.risk)}</p></td>
<td><p class="show-read-more">{html.escape(finding.remediation_recommendation_text)}</p> <a class="read-more" href="{finding.remediation_recommendation_url}"><i class="fas fa-external-link-alt"></i></a></td>
<td><p class="show-read-more">{parse_html_string(unroll_dict(finding.compliance))}</p></td>
</tr>
"""
)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
sys.exit(1)
def fill_html_overview_statistics(stats, output_filename, output_directory):
try:
filename = f"{output_directory}/{output_filename}{html_file_suffix}"
# Read file
if path.isfile(filename):
with open(filename, "r", encoding=enconding_format_utf_8) as file:
filedata = file.read()
# Replace statistics
# TOTAL_FINDINGS
filedata = filedata.replace(
"TOTAL_FINDINGS", str(stats.get("findings_count", 0))
<tbody>"""
)
# TOTAL_RESOURCES
filedata = filedata.replace(
"TOTAL_RESOURCES", str(stats.get("resources_count", 0))
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
# TOTAL_PASS
filedata = filedata.replace("TOTAL_PASS", str(stats.get("total_pass", 0)))
# TOTAL_FAIL
filedata = filedata.replace("TOTAL_FAIL", str(stats.get("total_fail", 0)))
# Write file
with open(filename, "w", encoding=enconding_format_utf_8) as file:
file.write(filedata)
@staticmethod
def write_footer(file_descriptor: TextIOWrapper) -> None:
"""
Writes the footer of the HTML file.
except FileNotFoundError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
except UnicodeDecodeError as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
def add_html_footer(output_filename, output_directory):
try:
filename = f"{output_directory}/{output_filename}{html_file_suffix}"
# Close HTML file if exists
if path.isfile(filename):
file_descriptor = open_file(
filename,
"a",
)
Args:
file_descriptor (file): the file descriptor to write the footer
"""
try:
file_descriptor.write(
"""
</tbody>
</tbody>
</table>
</div>
</div>
@@ -306,231 +301,261 @@ def add_html_footer(output_filename, output_directory):
</html>
"""
)
file_descriptor.close()
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
sys.exit(1)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
@staticmethod
def get_aws_assessment_summary(provider: Provider) -> str:
"""
get_aws_assessment_summary gets the HTML assessment summary for the provider
def get_aws_html_assessment_summary(provider):
try:
if provider.__class__.__name__ == "AwsProvider":
Args:
provider (Provider): the provider object
Returns:
str: the HTML assessment summary
"""
try:
profile = (
provider._identity.profile
if provider._identity.profile is not None
provider.identity.profile
if provider.identity.profile is not None
else "default"
)
if isinstance(provider._identity.audited_regions, list):
audited_regions = " ".join(provider._identity.audited_regions)
elif not provider._identity.audited_regions:
if isinstance(provider.identity.audited_regions, list):
audited_regions = " ".join(provider.identity.audited_regions)
elif not provider.identity.audited_regions:
audited_regions = "All Regions"
else:
audited_regions = ", ".join(provider._identity.audited_regions)
audited_regions = ", ".join(provider.identity.audited_regions)
return f"""
<div class="col-md-2">
<div class="col-md-2">
<div class="card">
<div class="card-header">
AWS Assessment Summary
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>AWS Account:</b> {provider.identity.account}
</li>
<li class="list-group-item">
<b>AWS-CLI Profile:</b> {profile}
</li>
<li class="list-group-item">
<b>Audited Regions:</b> {audited_regions}
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
AWS Assessment Summary
AWS Credentials
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>AWS Account:</b> {provider._identity.account}
</li>
<li class="list-group-item">
<b>AWS-CLI Profile:</b> {profile}
</li>
<li class="list-group-item">
<b>Audited Regions:</b> {audited_regions}
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
AWS Credentials
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>User Id:</b> {provider._identity.user_id}
</li>
<li class="list-group-item">
<b>Caller Identity ARN:</b> {provider._identity.identity_arn}
</li>
</ul>
</div>
</div>
"""
<b>User Id:</b> {provider.identity.user_id}
</li>
<li class="list-group-item">
<b>Caller Identity ARN:</b> {provider.identity.identity_arn}
</li>
</ul>
</div>
</div>"""
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
return ""
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
sys.exit(1)
@staticmethod
def get_azure_assessment_summary(provider: Provider) -> str:
"""
get_azure_assessment_summary gets the HTML assessment summary for the provider
Args:
provider (Provider): the provider object
def get_azure_html_assessment_summary(provider):
try:
if provider.__class__.__name__ == "AzureProvider":
Returns:
str: the HTML assessment summary
"""
try:
printed_subscriptions = []
for key, value in provider._identity.subscriptions.items():
for key, value in provider.identity.subscriptions.items():
intermediate = f"{key} : {value}"
printed_subscriptions.append(intermediate)
# check if identity is str(coming from SP) or dict(coming from browser or)
if isinstance(provider._identity.identity_id, dict):
html_identity = provider._identity.identity_id.get(
if isinstance(provider.identity.identity_id, dict):
html_identity = provider.identity.identity_id.get(
"userPrincipalName", "Identity not found"
)
else:
html_identity = provider._identity.identity_id
html_identity = provider.identity.identity_id
return f"""
<div class="col-md-2">
<div class="col-md-2">
<div class="card">
<div class="card-header">
Azure Assessment Summary
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>Azure Tenant IDs:</b> {" ".join(provider.identity.tenant_ids)}
</li>
<li class="list-group-item">
<b>Azure Tenant Domain:</b> {provider.identity.tenant_domain}
</li>
<li class="list-group-item">
<b>Azure Subscriptions:</b> {" ".join(printed_subscriptions)}
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
Azure Assessment Summary
Azure Credentials
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>Azure Tenant IDs:</b> {" ".join(provider._identity.tenant_ids)}
</li>
<li class="list-group-item">
<b>Azure Tenant Domain:</b> {provider._identity.tenant_domain}
</li>
<li class="list-group-item">
<b>Azure Subscriptions:</b> {" ".join(printed_subscriptions)}
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
Azure Credentials
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>Azure Identity Type:</b> {provider._identity.identity_type}
</li>
<li class="list-group-item">
<b>Azure Identity ID:</b> {html_identity}
</li>
</ul>
</div>
</div>
"""
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
sys.exit(1)
<b>Azure Identity Type:</b> {provider.identity.identity_type}
</li>
<li class="list-group-item">
<b>Azure Identity ID:</b> {html_identity}
</li>
</ul>
</div>
</div>"""
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
return ""
@staticmethod
def get_gcp_assessment_summary(provider: Provider) -> str:
"""
get_gcp_assessment_summary gets the HTML assessment summary for the provider
def get_gcp_html_assessment_summary(provider):
try:
if provider.__class__.__name__ == "GcpProvider":
Args:
provider (Provider): the provider object
Returns:
str: the HTML assessment summary
"""
try:
try:
getattr(provider.credentials, "_service_account_email")
getattr(provider.session, "_service_account_email")
profile = (
provider.credentials._service_account_email
if provider.credentials._service_account_email is not None
provider.session._service_account_email
if provider.session._service_account_email is not None
else "default"
)
except AttributeError:
profile = "default"
return f"""
<div class="col-md-2">
<div class="card">
<div class="card-header">
GCP Assessment Summary
<div class="col-md-2">
<div class="card">
<div class="card-header">
GCP Assessment Summary
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>GCP Project IDs:</b> {", ".join(provider.project_ids)}
</li>
</ul>
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>GCP Project IDs:</b> {", ".join(provider.project_ids)}
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
GCP Credentials
<div class="col-md-4">
<div class="card">
<div class="card-header">
GCP Credentials
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>GCP Account:</b> {profile}
</li>
</ul>
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>GCP Account:</b> {profile}
</li>
</ul>
</div>
</div>
"""
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
sys.exit(1)
</div>"""
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"""
<div class="col-md-2">
<div class="card">
<div class="card-header">
Kubernetes Assessment Summary
<div class="col-md-2">
<div class="card">
<div class="card-header">
Kubernetes Assessment Summary
</div>
<ul class="list-group
list-group-flush">
<li class="list-group-item">
<b>Kubernetes Cluster:</b> {provider.identity.cluster}
</li>
</ul>
</div>
<ul class="list-group
list-group-flush">
<li class="list-group-item">
<b>Kubernetes Cluster:</b> {provider._identity.cluster}
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
Kubernetes Credentials
<div class="col-md-4">
<div class="card">
<div class="card-header">
Kubernetes Credentials
</div>
<ul class="list-group
list-group-flush">
<li class="list-group-item">
<b>Kubernetes Context:</b> {provider.identity.context}
</li>
</ul>
</div>
<ul class="list-group
list-group-flush">
<li class="list-group-item">
<b>Kubernetes Context:</b> {provider._identity.context}
</li>
</ul>
</div>
</div>
"""
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
)
sys.exit(1)
</div>"""
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 ""
+5 -1
View File
@@ -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(
+4 -3
View File
@@ -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):
-8
View File
@@ -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:
+516
View File
@@ -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 = """
<tr class="p-3 mb-2 bg-success-custom">
<td>PASS</td>
<td>high</td>
<td>test-service</td>
<td>eu-west-1</td>
<td>test-check-id</td>
<td>test-check-id</td>
<td></td>
<td></td>
<td></td>
<td><p class="show-read-more">test-risk</p></td>
<td><p class="show-read-more"></p> <a class="read-more" href=""><i class="fas fa-external-link-alt"></i></a></td>
<td><p class="show-read-more">
&#x2022;test-compliance: test-compliance
</p></td>
</tr>
"""
fail_html_finding = """
<tr class="table-danger">
<td>FAIL</td>
<td>high</td>
<td>test-service</td>
<td>eu-west-1</td>
<td>test-check-id</td>
<td>test-check-id</td>
<td></td>
<td></td>
<td></td>
<td><p class="show-read-more">test-risk</p></td>
<td><p class="show-read-more"></p> <a class="read-more" href=""><i class="fas fa-external-link-alt"></i></a></td>
<td><p class="show-read-more">
&#x2022;test-compliance: test-compliance
</p></td>
</tr>
"""
muted_html_finding = """
<tr class="table-warning">
<td>MUTED (PASS)</td>
<td>high</td>
<td>test-service</td>
<td>eu-west-1</td>
<td>test-check-id</td>
<td>test-check-id</td>
<td></td>
<td></td>
<td></td>
<td><p class="show-read-more">test-risk</p></td>
<td><p class="show-read-more"></p> <a class="read-more" href=""><i class="fas fa-external-link-alt"></i></a></td>
<td><p class="show-read-more">
&#x2022;test-compliance: test-compliance
</p></td>
</tr>
"""
manual_html_finding = """
<tr class="table-info">
<td>MANUAL</td>
<td>high</td>
<td>test-service</td>
<td>eu-west-1</td>
<td>test-check-id</td>
<td>test-check-id</td>
<td></td>
<td></td>
<td></td>
<td><p class="show-read-more">test-risk</p></td>
<td><p class="show-read-more"></p> <a class="read-more" href=""><i class="fas fa-external-link-alt"></i></a></td>
<td><p class="show-read-more">
&#x2022;test-compliance: test-compliance
</p></td>
</tr>
"""
aws_html_assessment_summary = """
<div class="col-md-2">
<div class="card">
<div class="card-header">
AWS Assessment Summary
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>AWS Account:</b> 123456789012
</li>
<li class="list-group-item">
<b>AWS-CLI Profile:</b> default
</li>
<li class="list-group-item">
<b>Audited Regions:</b> eu-west-1
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
AWS Credentials
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>User Id:</b> None
</li>
<li class="list-group-item">
<b>Caller Identity ARN:</b> None
</li>
</ul>
</div>
</div>"""
azure_html_assessment_summary = """
<div class="col-md-2">
<div class="card">
<div class="card-header">
Azure Assessment Summary
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>Azure Tenant IDs:</b> 00000000-0000-0000-0000-000000000000
</li>
<li class="list-group-item">
<b>Azure Tenant Domain:</b> Unknown tenant domain (missing AAD permissions)
</li>
<li class="list-group-item">
<b>Azure Subscriptions:</b> 4f647f43-15d2-4e3a-a7f0-8517cc4d977b : Subscription Name
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
Azure Credentials
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>Azure Identity Type:</b> Service Principal
</li>
<li class="list-group-item">
<b>Azure Identity ID:</b> 00000000-0000-0000-0000-000000000000
</li>
</ul>
</div>
</div>"""
gcp_html_assessment_summary = """
<div class="col-md-2">
<div class="card">
<div class="card-header">
GCP Assessment Summary
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>GCP Project IDs:</b> 123456789012
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
GCP Credentials
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>GCP Account:</b> test@test.com
</li>
</ul>
</div>
</div>"""
kubernetes_html_assessment_summary = """
<div class="col-md-2">
<div class="card">
<div class="card-header">
Kubernetes Assessment Summary
</div>
<ul class="list-group
list-group-flush">
<li class="list-group-item">
<b>Kubernetes Cluster:</b> None
</li>
</ul>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">
Kubernetes Credentials
</div>
<ul class="list-group
list-group-flush">
<li class="list-group-item">
<b>Kubernetes Context:</b> None
</li>
</ul>
</div>
</div>"""
aws_html_header = (
"""
<!DOCTYPE html>
<html lang="en">
<head>
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
<!-- Required meta tags -->
<meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no" />
<style>
.read-more {color: #00f;}
.bg-success-custom {background-color: #98dea7 !important;}
.bg-danger {background-color: #f28484 !important;}
</style>
<!-- Bootstrap CSS -->
<link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.5.0/css/bootstrap.min.css"
integrity="sha384-9aIt2nRpC12Uk9gS9baDl411NQApFmC26EwAOH8WgZl5MYYxFfc+NcPb1dKGj7Sk" crossorigin="anonymous" />
<!-- https://datatables.net/download/index with jQuery, DataTables, Buttons, SearchPanes, and Select //-->
<link rel="stylesheet" type="text/css"
href="https://cdn.datatables.net/v/dt/jqc-1.12.4/dt-1.10.25/b-1.7.1/sp-1.4.0/sl-1.3.3/datatables.min.css" />
<link rel="stylesheet" href="https://pro.fontawesome.com/releases/v5.10.0/css/all.css"
integrity="sha384-AYmEC3Yw5cVb3ZcuHtOA93w35dYTsvhLPVnYs9eStHfGJvOvKxVfELGroGkvsg+p" crossorigin="anonymous" />
<style>
.show-read-more .more-text {display: none;}
.dataTable {font-size: 14px;}
.container-fluid {font-size: 14px;}
.float-left { float: left !important; max-width: 100%; }
</style>
<title>Prowler - The Handy Cloud Security Tool</title>
</head>
<body>
<div class="container-fluid">
<div class="row mt-3">
<div class="col-md-4">
<a href="https://github.com/prowler-cloud/prowler/"><img class="float-left card-img-left mt-4 mr-4 ml-4"
src=https://prowler.com/wp-content/uploads/logo-html.png
alt="prowler-logo"
style="width: 15rem; height:auto;"/></a>
<div class="card">
<div class="card-header">
Report Information
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<div class="row">
<div class="col-md-auto">
<b>Version:</b> 4.2.4
</div>
</div>
</li>
<li class="list-group-item">
<b>Parameters used:</b> """
+ " ".join(sys.argv[1:])
+ """
</li>
<li class="list-group-item">
<b>Date:</b> """
+ timestamp.isoformat()
+ """
</li>
</ul>
</div>
</div>"""
+ aws_html_assessment_summary
+ """
<div class="col-md-2">
<div class="card">
<div class="card-header">
Assessment Overview
</div>
<ul class="list-group list-group-flush">
<li class="list-group-item">
<b>Total Findings:</b> 1
</li>
<li class="list-group-item">
<b>Passed:</b> 0
</li>
<li class="list-group-item">
<b>Failed:</b> 1
</li>
<li class="list-group-item">
<b>Total Resources:</b> 1
</li>
</ul>
</div>
</div>
</div>
</div>
<div class="row-mt-3">
<div class="col-md-12">
<table class="table compact stripe row-border ordering" id="findingsTable" data-order='[[ 5, "asc" ]]' data-page-length='100'>
<thead class="thead-light">
<tr>
<th scope="col">Status</th>
<th scope="col">Severity</th>
<th scope="col">Service Name</th>
<th scope="col">Region</th>
<th style="width:20%" scope="col">Check ID</th>
<th style="width:20%" scope="col">Check Title</th>
<th scope="col">Resource ID</th>
<th scope="col">Resource Tags</th>
<th scope="col">Status Extended</th>
<th scope="col">Risk</th>
<th scope="col">Recomendation</th>
<th scope="col">Compliance</th>
</tr>
</thead>
<tbody>"""
)
html_footer = """
</tbody>
</table>
</div>
</div>
<!-- Table search and paginator -->
<!-- Optional JavaScript -->
<!-- jQuery first, then Popper.js, then Bootstrap JS -->
<script src="https://code.jquery.com/jquery-3.5.1.min.js"
integrity="sha256-9/aliU8dGd2tb6OSsuzixeV4y/faTqgFtohetphbbj0=" crossorigin="anonymous"></script>
<script src="https://stackpath.bootstrapcdn.com/bootstrap/4.5.0/js/bootstrap.bundle.min.js"
integrity="sha384-1CmrxMRARb6aLqgBO7yyAxTOQE2AKb9GfXnEo760AUcUmFx3ibVJJAzGytlQcNXd"
crossorigin="anonymous"></script>
<!-- https://datatables.net/download/index with jQuery, DataTables, Buttons, SearchPanes, and Select //-->
<script type="text/javascript"
src="https://cdn.datatables.net/v/dt/jqc-1.12.4/dt-1.10.25/b-1.7.1/sp-1.4.0/sl-1.3.3/datatables.min.js"></script>
<script>
$(document).ready(function () {
// Initialise the table with 50 rows, and some search/filtering panes
$('#findingsTable').DataTable({
responsive: true,
// Show 25, 50, 100 and All records
lengthChange: true,
lengthMenu: [[25, 50, 100, -1], [25, 50, 100, "All"]],
searchPanes: {
cascadePanes: true,
viewTotal: true,
},
dom: 'Blfrtip',
language: {
// To enable a filter button instead of the filter row
searchPanes: {
clearMessage: 'Clear Filters',
collapse: { 0: 'Filters', _: 'Filters (%d)' },
initCollapsed: true
}
},
buttons: [
{
extend: 'searchPanes',
config: {
cascadePanes: true,
viewTotal: true,
orderable: false
}
}
],
columnDefs: [
{
searchPanes: {
show: true,
pagingType: 'numbers',
searching: true
},
// Show all filters
targets: [0, 1, 2, 3, 5, 7]
}
]
});
var maxLength = 30;
// ReadMore ReadLess
$(".show-read-more").each(function () {
var myStr = $(this).text();
if ($.trim(myStr).length > maxLength) {
var newStr = myStr.substring(0, maxLength);
var removedStr = myStr.substring(maxLength, $.trim(myStr).length);
$(this).empty().html(newStr);
$(this).append(' <a href="javascript:void(0);" class="read-more">read more...</a>');
$(this).append('<span class="more-text">' + removedStr + '</span>');
}
});
$(".read-more").click(function () {
$(this).siblings(".more-text").contents().unwrap();
$(this).remove();
});
});
</script>
</body>
</html>
"""
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
+1 -36
View File
@@ -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"]
@@ -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 ==
@@ -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",
@@ -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",
+2 -1
View File
@@ -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,
@@ -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 == {