diff --git a/api/src/backend/api/v1/serializers.py b/api/src/backend/api/v1/serializers.py
index 2c9b044d51..ff64eb6890 100644
--- a/api/src/backend/api/v1/serializers.py
+++ b/api/src/backend/api/v1/serializers.py
@@ -1776,6 +1776,15 @@ class ComplianceOverviewMetadataSerializer(serializers.Serializer):
resource_name = "compliance-overviews-metadata"
+class ComplianceOverviewThreatscoreReportSerializer(serializers.Serializer):
+ id = serializers.CharField(source="scan")
+ name = serializers.CharField()
+
+ class Meta:
+ resource_name = "scan-threatscore-reports"
+ fields = ["id", "name"]
+
+
# Overviews
diff --git a/api/src/backend/api/v1/views.py b/api/src/backend/api/v1/views.py
index fa732edba8..81d566de13 100644
--- a/api/src/backend/api/v1/views.py
+++ b/api/src/backend/api/v1/views.py
@@ -137,6 +137,7 @@ from api.v1.serializers import (
ComplianceOverviewDetailSerializer,
ComplianceOverviewMetadataSerializer,
ComplianceOverviewSerializer,
+ ComplianceOverviewThreatscoreReportSerializer,
FindingDynamicFilterSerializer,
FindingMetadataSerializer,
FindingSerializer,
@@ -3117,6 +3118,8 @@ class ComplianceOverviewViewSet(BaseRLSViewSet, TaskManagementMixin):
return ComplianceOverviewAttributesSerializer
elif self.action == "requirements":
return ComplianceOverviewDetailSerializer
+ elif self.action == "threatscore":
+ return ComplianceOverviewThreatscoreReportSerializer
return super().get_serializer_class()
@extend_schema(exclude=True)
@@ -3424,6 +3427,75 @@ class ComplianceOverviewViewSet(BaseRLSViewSet, TaskManagementMixin):
serializer = self.get_serializer(attribute_data, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)
+ @action(
+ detail=True,
+ methods=["get"],
+ url_path="threatscore",
+ url_name="threatscore",
+ )
+ def threatscore(self, request):
+ scan_id = request.query_params.get("filter[scan_id]")
+ if not scan_id:
+ raise ValidationError(
+ [
+ {
+ "detail": "This query parameter is required.",
+ "status": 400,
+ "source": {"pointer": "filter[scan_id]"},
+ "code": "required",
+ }
+ ]
+ )
+ try:
+ if task := self.get_task_response_if_running(
+ task_name="scan-threatscore-report",
+ task_kwargs={"tenant_id": self.request.tenant_id, "scan_id": scan_id},
+ raise_on_not_found=False,
+ ):
+ return task
+ except TaskFailedException:
+ return Response(
+ {"detail": "Task failed to generate threatscore report."},
+ status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+ )
+ try:
+ scan = Scan.objects.get(id=scan_id)
+ except Scan.DoesNotExist:
+ return Response(
+ {"detail": "Scan not found."},
+ status=status.HTTP_404_NOT_FOUND,
+ )
+
+ if not scan.output_location:
+ return Response(
+ {
+ "detail": "The scan has no reports, or the report generation task has not started yet."
+ },
+ status=status.HTTP_404_NOT_FOUND,
+ )
+
+ if scan.output_location.startswith("s3://"):
+ bucket = env.str("DJANGO_OUTPUT_S3_AWS_OUTPUT_BUCKET", "")
+ key_prefix = scan.output_location.removeprefix(f"s3://{bucket}/")
+ prefix = os.path.join(
+ os.path.dirname(key_prefix),
+ "compliance",
+ f"{scan.provider.provider}_threatscore.pdf",
+ )
+ loader = self._load_file(prefix, s3=True, bucket=bucket, list_objects=True)
+ else:
+ base = os.path.dirname(scan.output_location)
+ pattern = os.path.join(
+ base, "compliance", f"*_{scan.provider.provider}_threatscore.pdf"
+ )
+ loader = self._load_file(pattern, s3=False)
+
+ if isinstance(loader, Response):
+ return loader
+
+ content, filename = loader
+ return self._serve_file(content, filename, "application/pdf")
+
@extend_schema(tags=["Overview"])
@extend_schema_view(
@@ -3937,4 +4009,4 @@ class ProcessorViewSet(BaseRLSViewSet):
return ProcessorCreateSerializer
elif self.action == "partial_update":
return ProcessorUpdateSerializer
- return super().get_serializer_class()
\ No newline at end of file
+ return super().get_serializer_class()
diff --git a/api/src/backend/tasks/jobs/report.py b/api/src/backend/tasks/jobs/report.py
new file mode 100644
index 0000000000..1af4abb4b7
--- /dev/null
+++ b/api/src/backend/tasks/jobs/report.py
@@ -0,0 +1,912 @@
+import io
+import os
+
+import matplotlib.pyplot as plt
+from celery.utils.log import get_task_logger
+from config.django.base import DJANGO_TMP_OUTPUT_DIRECTORY
+from reportlab.lib import colors
+from reportlab.lib.enums import TA_CENTER
+from reportlab.lib.pagesizes import letter
+from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
+from reportlab.lib.units import inch
+from reportlab.pdfbase import pdfmetrics
+from reportlab.pdfbase.ttfonts import TTFont
+from reportlab.pdfgen import canvas
+from reportlab.platypus import (
+ Image,
+ PageBreak,
+ Paragraph,
+ SimpleDocTemplate,
+ Spacer,
+ Table,
+ TableStyle,
+)
+from tasks.jobs.export import _generate_output_directory, _upload_to_s3
+
+from api.models import Finding, Provider, Task
+from api.utils import initialize_prowler_provider
+from prowler.lib.check.compliance_models import Compliance
+from prowler.lib.outputs.finding import Finding as FindingOutput
+
+pdfmetrics.registerFont(
+ TTFont(
+ "PlusJakartaSans",
+ os.path.join(
+ os.path.dirname(__file__), "assets/fonts/PlusJakartaSans-Regular.ttf"
+ ),
+ )
+)
+
+pdfmetrics.registerFont(
+ TTFont(
+ "FiraCode",
+ os.path.join(os.path.dirname(__file__), "assets/fonts/FiraCode-Regular.ttf"),
+ )
+)
+
+
+def generate_threatscore_report(
+ scan_id: str,
+ compliance_id: str,
+ output_path: str,
+ provider_id: str,
+ only_failed: bool = True,
+ min_risk_level: int = 4,
+):
+ """
+ Generate a PDF compliance report based on Prowler ORM objects.
+
+ Parameters:
+ - scan_id: ID of the scan executed by Prowler.
+ - compliance_id: ID of the compliance framework (e.g., "nis2_azure").
+ - output_path: Output PDF file path (e.g., "threatscore_report.pdf").
+ - provider_id: Provider ID for the scan.
+ - only_failed: If True, only requirements with status "FAIL" will be included in the list of requirements.
+ - min_risk_level: Minimum risk level for critical failed requirements.
+ """
+ styles = getSampleStyleSheet()
+
+ prowler_dark_green = colors.Color(0.1, 0.5, 0.2)
+
+ title_style = ParagraphStyle(
+ "CustomTitle",
+ parent=styles["Title"],
+ fontSize=24,
+ textColor=prowler_dark_green,
+ spaceAfter=20,
+ fontName="PlusJakartaSans",
+ alignment=TA_CENTER,
+ )
+
+ h1 = ParagraphStyle(
+ "CustomH1",
+ parent=styles["Heading1"],
+ fontSize=18,
+ textColor=colors.Color(0.2, 0.4, 0.6),
+ spaceBefore=20,
+ spaceAfter=12,
+ fontName="PlusJakartaSans",
+ leftIndent=0,
+ borderWidth=2,
+ borderColor=colors.Color(0.2, 0.4, 0.6),
+ borderPadding=8,
+ backColor=colors.Color(0.95, 0.97, 1.0),
+ )
+
+ h2 = ParagraphStyle(
+ "CustomH2",
+ parent=styles["Heading2"],
+ fontSize=14,
+ textColor=colors.Color(0.3, 0.5, 0.7),
+ spaceBefore=15,
+ spaceAfter=8,
+ fontName="PlusJakartaSans",
+ leftIndent=10,
+ borderWidth=1,
+ borderColor=colors.Color(0.7, 0.8, 0.9),
+ borderPadding=5,
+ backColor=colors.Color(0.98, 0.99, 1.0),
+ )
+
+ h3 = ParagraphStyle(
+ "CustomH3",
+ parent=styles["Heading3"],
+ fontSize=12,
+ textColor=colors.Color(0.4, 0.6, 0.8),
+ spaceBefore=10,
+ spaceAfter=6,
+ fontName="PlusJakartaSans",
+ leftIndent=20,
+ )
+
+ normal = ParagraphStyle(
+ "CustomNormal",
+ parent=styles["Normal"],
+ fontSize=10,
+ textColor=colors.Color(0.2, 0.2, 0.2),
+ spaceBefore=4,
+ spaceAfter=4,
+ leftIndent=30,
+ fontName="PlusJakartaSans",
+ )
+
+ normal_center = ParagraphStyle(
+ "CustomNormalCenter",
+ parent=styles["Normal"],
+ fontSize=10,
+ textColor=colors.Color(0.2, 0.2, 0.2),
+ fontName="PlusJakartaSans",
+ )
+
+ provider_obj = Provider.objects.get(id=provider_id)
+ prowler_provider = initialize_prowler_provider(provider_obj)
+ provider_type = provider_obj.provider
+
+ frameworks_bulk = Compliance.get_bulk(provider_type)
+ compliance_obj = frameworks_bulk[compliance_id]
+ compliance_name = compliance_obj.name
+ compliance_version = getattr(compliance_obj, "Version", "N/A")
+ compliance_description = getattr(compliance_obj, "Description", "")
+
+ findings_qs = Finding.all_objects.filter(scan_id=scan_id).order_by("uid")
+ findings = [
+ FindingOutput.transform_api_finding(f, prowler_provider) for f in findings_qs
+ ]
+
+ attrs_map = {}
+ resp_reqs = []
+ resp_attrs = []
+ for req_id, req in compliance_obj.requirements.items():
+ attrs_map[req_id] = {
+ "id": req_id,
+ "attributes": {
+ "metadata": [req.metadata],
+ "check_ids": req.check_ids,
+ },
+ "description": getattr(req, "description", ""),
+ }
+ resp_attrs.append(
+ {
+ "id": req_id,
+ "attributes": {
+ "framework_description": compliance_description,
+ "attributes": {"metadata": [req.metadata]},
+ },
+ }
+ )
+ status = "UNKNOWN"
+ description = getattr(req, "description", "")
+ for f in findings:
+ if f["check_id"] in req.check_ids:
+ status = f["status"]
+ description = f.get("description", description)
+ break
+ resp_reqs.append(
+ {
+ "id": req_id,
+ "attributes": {
+ "framework": compliance_name,
+ "version": compliance_version,
+ "status": status,
+ "description": description,
+ },
+ }
+ )
+
+ def create_risk_component(risk_level, weight, score=0):
+ """Create a visual risk component similar to the UI design"""
+ if risk_level >= 4:
+ risk_color = colors.Color(0.8, 0.2, 0.2)
+ elif risk_level >= 3:
+ risk_color = colors.Color(0.9, 0.6, 0.2)
+ elif risk_level >= 2:
+ risk_color = colors.Color(0.9, 0.9, 0.2)
+ else:
+ risk_color = colors.Color(0.2, 0.8, 0.2)
+
+ if weight <= 50:
+ weight_color = colors.Color(0.2, 0.8, 0.2)
+ elif weight <= 100:
+ weight_color = colors.Color(0.9, 0.9, 0.2)
+ else:
+ weight_color = colors.Color(0.8, 0.2, 0.2)
+
+ score_color = colors.Color(0.4, 0.4, 0.4)
+
+ data = [
+ [
+ "Risk Level:",
+ str(risk_level),
+ "Weight:",
+ str(weight),
+ "Score:",
+ str(score),
+ ]
+ ]
+
+ table = Table(
+ data,
+ colWidths=[
+ 0.8 * inch,
+ 0.4 * inch,
+ 0.6 * inch,
+ 0.4 * inch,
+ 0.5 * inch,
+ 0.4 * inch,
+ ],
+ )
+
+ table.setStyle(
+ TableStyle(
+ [
+ ("BACKGROUND", (0, 0), (0, 0), colors.Color(0.9, 0.9, 0.9)),
+ ("BACKGROUND", (1, 0), (1, 0), risk_color),
+ ("TEXTCOLOR", (1, 0), (1, 0), colors.white),
+ ("FONTNAME", (1, 0), (1, 0), "FiraCode"),
+ ("BACKGROUND", (2, 0), (2, 0), colors.Color(0.9, 0.9, 0.9)),
+ ("BACKGROUND", (3, 0), (3, 0), weight_color),
+ ("TEXTCOLOR", (3, 0), (3, 0), colors.white),
+ ("FONTNAME", (3, 0), (3, 0), "FiraCode"),
+ ("BACKGROUND", (4, 0), (4, 0), colors.Color(0.9, 0.9, 0.9)),
+ ("BACKGROUND", (5, 0), (5, 0), score_color),
+ ("TEXTCOLOR", (5, 0), (5, 0), colors.white),
+ ("FONTNAME", (5, 0), (5, 0), "FiraCode"),
+ ("ALIGN", (0, 0), (-1, -1), "CENTER"),
+ ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
+ ("FONTSIZE", (0, 0), (-1, -1), 10),
+ ("GRID", (0, 0), (-1, -1), 0.5, colors.black),
+ ("LEFTPADDING", (0, 0), (-1, -1), 6),
+ ("RIGHTPADDING", (0, 0), (-1, -1), 6),
+ ("TOPPADDING", (0, 0), (-1, -1), 8),
+ ("BOTTOMPADDING", (0, 0), (-1, -1), 8),
+ ]
+ )
+ )
+
+ return table
+
+ def create_status_component(status):
+ """Create a visual status component with colors"""
+ if status.upper() == "PASS":
+ status_color = colors.Color(0.2, 0.8, 0.2)
+ elif status.upper() == "FAIL":
+ status_color = colors.Color(0.8, 0.2, 0.2)
+ else:
+ status_color = colors.Color(0.4, 0.4, 0.4)
+
+ data = [["State:", status.upper()]]
+
+ elements.append(Spacer(1, 0.1 * inch))
+
+ table = Table(data, colWidths=[0.6 * inch, 0.8 * inch])
+
+ table.setStyle(
+ TableStyle(
+ [
+ ("BACKGROUND", (0, 0), (0, 0), colors.Color(0.9, 0.9, 0.9)),
+ ("FONTNAME", (0, 0), (0, 0), "PlusJakartaSans"),
+ ("BACKGROUND", (1, 0), (1, 0), status_color),
+ ("TEXTCOLOR", (1, 0), (1, 0), colors.white),
+ ("FONTNAME", (1, 0), (1, 0), "FiraCode"),
+ ("ALIGN", (0, 0), (-1, -1), "CENTER"),
+ ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
+ ("FONTSIZE", (0, 0), (-1, -1), 12),
+ ("GRID", (0, 0), (-1, -1), 0.5, colors.black),
+ ("LEFTPADDING", (0, 0), (-1, -1), 8),
+ ("RIGHTPADDING", (0, 0), (-1, -1), 8),
+ ("TOPPADDING", (0, 0), (-1, -1), 10),
+ ("BOTTOMPADDING", (0, 0), (-1, -1), 10),
+ ]
+ )
+ )
+
+ return table
+
+ def create_section_score_chart(resp_reqs, attrs_map):
+ """Create a bar chart showing compliance score by section"""
+ sections_data = {}
+
+ for req in resp_reqs:
+ req_id = req["id"]
+ attr = attrs_map.get(req_id, {})
+ status = req["attributes"]["status"]
+
+ metadata = attr.get("attributes", {}).get("metadata", [])
+ if metadata:
+ m = metadata[0]
+ section = m.get("Section", "Unknown")
+ risk_level = m.get("LevelOfRisk", 0)
+ weight = m.get("Weight", 0)
+
+ if section not in sections_data:
+ sections_data[section] = {"total_score": 0, "max_possible_score": 0}
+
+ max_score = risk_level * weight
+ sections_data[section]["max_possible_score"] += max_score
+
+ if status == "PASS":
+ sections_data[section]["total_score"] += max_score
+
+ section_names = []
+ compliance_percentages = []
+
+ for section, data in sections_data.items():
+ if data["max_possible_score"] > 0:
+ compliance_percentage = (
+ data["total_score"] / data["max_possible_score"]
+ ) * 100
+ else:
+ compliance_percentage = 0
+
+ section_names.append(section)
+ compliance_percentages.append(compliance_percentage)
+
+ sorted_data = sorted(
+ zip(section_names, compliance_percentages), key=lambda x: x[1], reverse=True
+ )
+ section_names, compliance_percentages = (
+ zip(*sorted_data) if sorted_data else ([], [])
+ )
+
+ fig, ax = plt.subplots(figsize=(12, 8))
+
+ colors_list = []
+ for percentage in compliance_percentages:
+ if percentage >= 80:
+ color = "#4CAF50"
+ elif percentage >= 60:
+ color = "#8BC34A"
+ elif percentage >= 40:
+ color = "#FFEB3B"
+ elif percentage >= 20:
+ color = "#FF9800"
+ else:
+ color = "#F44336"
+ colors_list.append(color)
+
+ bars = ax.bar(section_names, compliance_percentages, color=colors_list)
+
+ ax.set_ylabel("Compliance Score (%)", fontsize=12)
+ ax.set_xlabel("Section", fontsize=12)
+ ax.set_ylim(0, 100)
+
+ for bar, percentage in zip(bars, compliance_percentages):
+ height = bar.get_height()
+ ax.text(
+ bar.get_x() + bar.get_width() / 2.0,
+ height + 1,
+ f"{percentage:.1f}%",
+ ha="center",
+ va="bottom",
+ fontweight="bold",
+ )
+
+ plt.xticks(rotation=45, ha="right")
+
+ ax.grid(True, alpha=0.3, axis="y")
+
+ plt.tight_layout()
+
+ buffer = io.BytesIO()
+ plt.savefig(buffer, format="png", dpi=300, bbox_inches="tight")
+ buffer.seek(0)
+ plt.close()
+
+ return buffer
+
+ def get_finding_info(check_id: str):
+ return [f for f in findings if f["check_id"] == check_id]
+
+ doc = SimpleDocTemplate(
+ output_path,
+ pagesize=letter,
+ title=f"Prowler ThreatScore Report - {compliance_name}",
+ author="Prowler",
+ subject=f"Compliance Report for {compliance_name}",
+ creator="Prowler Compliance Generator",
+ keywords=f"compliance,{compliance_name},security,framework,prowler",
+ )
+ elements = []
+
+ try:
+ logo = Image(
+ "util/compliance_report/assets/img/prowler_logo.png",
+ width=5 * inch,
+ height=0.8 * inch,
+ )
+ elements.append(logo)
+ except Exception:
+ pass
+ elements.append(Spacer(1, 0.5 * inch))
+ elements.append(Paragraph("Prowler ThreatScore Report", title_style))
+ elements.append(Spacer(1, 0.5 * inch))
+
+ info_data = [
+ ["Compliance Framework:", compliance_name],
+ ["Compliance ID:", compliance_id],
+ ["Version:", compliance_version],
+ ["Scan ID:", scan_id],
+ ["Description:", Paragraph(compliance_description, normal_center)],
+ ]
+ info_table = Table(info_data, colWidths=[2 * inch, 4 * inch])
+ info_table.setStyle(
+ TableStyle(
+ [
+ ("BACKGROUND", (0, 0), (0, 4), colors.Color(0.2, 0.4, 0.6)),
+ ("TEXTCOLOR", (0, 0), (0, 4), colors.white),
+ ("FONTNAME", (0, 0), (0, 4), "FiraCode"),
+ ("BACKGROUND", (1, 0), (1, 4), colors.Color(0.95, 0.97, 1.0)),
+ ("TEXTCOLOR", (1, 0), (1, 4), colors.Color(0.2, 0.2, 0.2)),
+ ("FONTNAME", (1, 0), (1, 4), "PlusJakartaSans"),
+ ("ALIGN", (0, 0), (-1, -1), "LEFT"),
+ ("VALIGN", (0, 0), (-1, -1), "TOP"),
+ ("FONTSIZE", (0, 0), (-1, -1), 11),
+ ("GRID", (0, 0), (-1, -1), 1, colors.Color(0.7, 0.8, 0.9)),
+ ("LEFTPADDING", (0, 0), (-1, -1), 10),
+ ("RIGHTPADDING", (0, 0), (-1, -1), 10),
+ ("TOPPADDING", (0, 0), (-1, -1), 8),
+ ("BOTTOMPADDING", (0, 0), (-1, -1), 8),
+ ]
+ )
+ )
+
+ elements.append(info_table)
+
+ elements.append(PageBreak())
+
+ elements.append(Paragraph("Compliance Score by Sections", h1))
+ elements.append(Spacer(1, 0.2 * inch))
+
+ chart_buffer = create_section_score_chart(resp_reqs, attrs_map)
+ chart_image = Image(chart_buffer, width=7 * inch, height=5.5 * inch)
+ elements.append(chart_image)
+
+ total_score = 0
+ max_possible_score = 0
+
+ for req in resp_reqs:
+ req_id = req["id"]
+ attr = attrs_map.get(req_id, {})
+ status = req["attributes"]["status"]
+
+ metadata = attr.get("attributes", {}).get("metadata", [])
+ if metadata:
+ m = metadata[0]
+ risk_level = m.get("LevelOfRisk", 0)
+ weight = m.get("Weight", 0)
+ max_score = risk_level * weight
+ max_possible_score += max_score
+
+ if status == "PASS":
+ total_score += max_score
+
+ overall_compliance = (
+ (total_score / max_possible_score * 100) if max_possible_score > 0 else 0
+ )
+
+ elements.append(Spacer(1, 0.3 * inch))
+
+ summary_data = [
+ ["Total Score:", f"{total_score:,}"],
+ ["Max Possible Score:", f"{max_possible_score:,}"],
+ ["Overall Compliance:", f"{overall_compliance:.2f}%"],
+ ]
+
+ if overall_compliance >= 80:
+ compliance_color = colors.Color(0.2, 0.8, 0.2)
+ elif overall_compliance >= 60:
+ compliance_color = colors.Color(0.8, 0.8, 0.2)
+ else:
+ compliance_color = colors.Color(0.8, 0.2, 0.2)
+
+ summary_table = Table(summary_data, colWidths=[2.5 * inch, 2 * inch])
+ summary_table.setStyle(
+ TableStyle(
+ [
+ ("BACKGROUND", (0, 0), (0, 1), colors.Color(0.3, 0.5, 0.7)),
+ ("TEXTCOLOR", (0, 0), (0, 1), colors.white),
+ ("FONTNAME", (0, 0), (0, 1), "FiraCode"),
+ ("BACKGROUND", (0, 2), (0, 2), colors.Color(0.1, 0.3, 0.5)),
+ ("TEXTCOLOR", (0, 2), (0, 2), colors.white),
+ ("FONTNAME", (0, 2), (0, 2), "FiraCode"),
+ ("FONTSIZE", (0, 2), (0, 2), 12),
+ ("BACKGROUND", (1, 0), (1, 1), colors.Color(0.95, 0.97, 1.0)),
+ ("TEXTCOLOR", (1, 0), (1, 1), colors.Color(0.2, 0.2, 0.2)),
+ ("FONTNAME", (1, 0), (1, 1), "PlusJakartaSans"),
+ ("BACKGROUND", (1, 2), (1, 2), compliance_color),
+ ("TEXTCOLOR", (1, 2), (1, 2), colors.white),
+ ("FONTNAME", (1, 2), (1, 2), "FiraCode"),
+ ("FONTSIZE", (1, 2), (1, 2), 14),
+ ("ALIGN", (0, 0), (-1, -1), "CENTER"),
+ ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
+ ("FONTSIZE", (0, 0), (1, 1), 11),
+ ("GRID", (0, 0), (-1, -1), 1.5, colors.Color(0.5, 0.6, 0.7)),
+ ("LEFTPADDING", (0, 0), (-1, -1), 12),
+ ("RIGHTPADDING", (0, 0), (-1, -1), 12),
+ ("TOPPADDING", (0, 0), (-1, -1), 10),
+ ("BOTTOMPADDING", (0, 0), (-1, -1), 10),
+ ]
+ )
+ )
+
+ elements.append(summary_table)
+
+ elements.append(PageBreak())
+
+ elements.append(Paragraph("Requirements Index", h1))
+
+ sections = {}
+ for req in resp_attrs:
+ meta = req["attributes"]["attributes"]["metadata"][0]
+ section = meta["Section"]
+ subsection = meta["SubSection"]
+ req_id = req["id"]
+ title = meta["Title"]
+
+ if section not in sections:
+ sections[section] = {}
+ if subsection not in sections[section]:
+ sections[section][subsection] = []
+
+ sections[section][subsection].append({"id": req_id, "title": title})
+
+ section_num = 1
+ for section_name, subsections in sections.items():
+ elements.append(Paragraph(f"{section_num}. {section_name}", h2))
+
+ subsection_num = 1
+ for subsection_name, requirements in subsections.items():
+ elements.append(Paragraph(f"{subsection_name}", h3))
+
+ req_num = 1
+ for req in requirements:
+ elements.append(Paragraph(f"{req['id']} - {req['title']}", normal))
+ req_num += 1
+
+ subsection_num += 1
+
+ section_num += 1
+ elements.append(Spacer(1, 0.1 * inch))
+
+ elements.append(PageBreak())
+
+ elements.append(Paragraph("Top Requirements by Level of Risk", h1))
+ elements.append(Spacer(1, 0.1 * inch))
+ elements.append(
+ Paragraph(f"Critical Failed Requirements (Risk Level ≥ {min_risk_level})", h2)
+ )
+ elements.append(Spacer(1, 0.2 * inch))
+
+ critical_reqs = []
+ for req in resp_reqs:
+ req_id = req["id"]
+ attr = attrs_map.get(req_id, {})
+ status = req["attributes"]["status"]
+
+ if status == "FAIL":
+ metadata = attr.get("attributes", {}).get("metadata", [])
+ if metadata:
+ m = metadata[0]
+ risk_level = m.get("LevelOfRisk", 0)
+ weight = m.get("Weight", 0)
+
+ if risk_level >= min_risk_level:
+ critical_reqs.append(
+ {
+ "req": req,
+ "attr": attr,
+ "risk_level": risk_level,
+ "weight": weight,
+ "metadata": m,
+ }
+ )
+
+ critical_reqs.sort(key=lambda x: (x["risk_level"], x["weight"]), reverse=True)
+
+ if not critical_reqs:
+ elements.append(
+ Paragraph("✅ No critical failed requirements found. Great job!", normal)
+ )
+ else:
+ elements.append(
+ Paragraph(
+ f"Found {len(critical_reqs)} critical failed requirements that require immediate attention:",
+ normal,
+ )
+ )
+ elements.append(Spacer(1, 0.5 * inch))
+
+ table_data = [["Risk", "Weight", "Requirement ID", "Title", "Section"]]
+
+ for idx, critical_req in enumerate(critical_reqs):
+ req_id = critical_req["req"]["id"]
+ risk_level = critical_req["risk_level"]
+ weight = critical_req["weight"]
+ title = critical_req["metadata"].get("Title", "N/A")
+ section = critical_req["metadata"].get("Section", "N/A")
+
+ if len(title) > 50:
+ title = title[:47] + "..."
+
+ table_data.append([str(risk_level), str(weight), req_id, title, section])
+
+ critical_table = Table(
+ table_data,
+ colWidths=[0.6 * inch, 0.8 * inch, 1.2 * inch, 3 * inch, 1.4 * inch],
+ )
+
+ critical_table.setStyle(
+ TableStyle(
+ [
+ ("BACKGROUND", (0, 0), (-1, 0), colors.Color(0.8, 0.2, 0.2)),
+ ("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
+ ("FONTNAME", (0, 0), (-1, 0), "FiraCode"),
+ ("FONTSIZE", (0, 0), (-1, 0), 10),
+ ("BACKGROUND", (0, 1), (0, -1), colors.Color(0.8, 0.2, 0.2)),
+ ("TEXTCOLOR", (0, 1), (0, -1), colors.white),
+ ("FONTNAME", (0, 1), (0, -1), "FiraCode"),
+ ("ALIGN", (0, 1), (0, -1), "CENTER"),
+ ("FONTSIZE", (0, 1), (0, -1), 12),
+ ("ALIGN", (1, 1), (1, -1), "CENTER"),
+ ("FONTNAME", (1, 1), (1, -1), "FiraCode"),
+ ("FONTNAME", (2, 1), (2, -1), "FiraCode"),
+ ("FONTSIZE", (2, 1), (2, -1), 9),
+ ("FONTNAME", (3, 1), (-1, -1), "PlusJakartaSans"),
+ ("FONTSIZE", (3, 1), (-1, -1), 8),
+ ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
+ ("GRID", (0, 0), (-1, -1), 1, colors.Color(0.7, 0.7, 0.7)),
+ ("LEFTPADDING", (0, 0), (-1, -1), 6),
+ ("RIGHTPADDING", (0, 0), (-1, -1), 6),
+ ("TOPPADDING", (0, 0), (-1, -1), 8),
+ ("BOTTOMPADDING", (0, 0), (-1, -1), 8),
+ ("BACKGROUND", (1, 1), (-1, -1), colors.Color(0.98, 0.98, 0.98)),
+ ]
+ )
+ )
+
+ for idx, critical_req in enumerate(critical_reqs):
+ row_idx = idx + 1
+ weight = critical_req["weight"]
+
+ if weight >= 150:
+ weight_color = colors.Color(0.8, 0.2, 0.2)
+ elif weight >= 100:
+ weight_color = colors.Color(0.9, 0.6, 0.2)
+ else:
+ weight_color = colors.Color(0.9, 0.9, 0.2)
+
+ critical_table.setStyle(
+ TableStyle(
+ [
+ ("BACKGROUND", (1, row_idx), (1, row_idx), weight_color),
+ ("TEXTCOLOR", (1, row_idx), (1, row_idx), colors.white),
+ ]
+ )
+ )
+
+ elements.append(critical_table)
+ elements.append(Spacer(1, 0.2 * inch))
+
+ warning_text = """
+ IMMEDIATE ACTION REQUIRED:
+ These requirements have the highest risk levels and have failed compliance checks.
+ Please prioritize addressing these issues to improve your security posture.
+ """
+
+ warning_style = ParagraphStyle(
+ "Warning",
+ parent=styles["Normal"],
+ fontSize=11,
+ textColor=colors.Color(0.8, 0.2, 0.2),
+ spaceBefore=10,
+ spaceAfter=10,
+ leftIndent=20,
+ rightIndent=20,
+ fontName="PlusJakartaSans",
+ backColor=colors.Color(1.0, 0.95, 0.95),
+ borderWidth=2,
+ borderColor=colors.Color(0.8, 0.2, 0.2),
+ borderPadding=10,
+ )
+
+ elements.append(Paragraph(warning_text, warning_style))
+
+ elements.append(PageBreak())
+
+ def get_weight(req):
+ req_id = req["id"]
+ attr = attrs_map.get(req_id, {})
+ metadata = attr.get("attributes", {}).get("metadata", [])
+ if metadata:
+ return metadata[0].get("Weight", 0)
+ return 0
+
+ sorted_reqs = sorted(resp_reqs, key=get_weight, reverse=True)
+
+ if only_failed:
+ sorted_reqs = [
+ req for req in sorted_reqs if req["attributes"]["status"] == "FAIL"
+ ]
+
+ for req in sorted_reqs:
+ req_id = req["id"]
+ attr = attrs_map.get(req_id, {})
+ desc = req["attributes"]["description"]
+ status = req["attributes"]["status"]
+
+ elements.append(Paragraph(f"{req_id}: {attr.get('description', desc)}", h1))
+
+ status_component = create_status_component(status)
+ elements.append(status_component)
+ elements.append(Spacer(1, 0.1 * inch))
+
+ metadata = attr.get("attributes", {}).get("metadata", [])
+ if metadata:
+ m = metadata[0]
+ elements.append(Paragraph("Title: ", h3))
+ elements.append(Paragraph(f"{m.get('Title')}", normal))
+ elements.append(Paragraph("Section: ", h3))
+ elements.append(Paragraph(f"{m.get('Section')}", normal))
+ elements.append(Paragraph("SubSection: ", h3))
+ elements.append(Paragraph(f"{m.get('SubSection')}", normal))
+ elements.append(Paragraph("Description: ", h3))
+ elements.append(Paragraph(f"{m.get('AttributeDescription')}", normal))
+ elements.append(Paragraph("Additional Information: ", h3))
+ elements.append(Paragraph(f"{m.get('AdditionalInformation')}", normal))
+ elements.append(Spacer(1, 0.1 * inch))
+
+ risk_level = m.get("LevelOfRisk", 0)
+ weight = m.get("Weight", 0)
+
+ if status == "PASS":
+ score = risk_level * weight
+ else:
+ score = 0
+
+ risk_component = create_risk_component(risk_level, weight, score)
+ elements.append(risk_component)
+ elements.append(Spacer(1, 0.1 * inch))
+
+ checks = attr.get("attributes", {}).get("check_ids", [])
+ for cid in checks:
+ elements.append(Paragraph(f"Check: {cid}", h2))
+ elements.append(Spacer(1, 0.1 * inch))
+ finds = get_finding_info(cid)
+ if not finds:
+ elements.append(
+ Paragraph("- No information for this finding currently", normal)
+ )
+ else:
+ findings_table_data = [
+ [
+ "Finding",
+ "Resource name",
+ "Severity",
+ "Status",
+ "Region",
+ ]
+ ]
+ for f in finds:
+ attr = f["attributes"]
+ check_meta = attr.get("check_metadata", {})
+ title = check_meta.get("checktitle", attr.get("check_id", ""))
+ resource_name = attr.get("resource_id")
+ if not resource_name:
+ rel_resources = (
+ f.get("relationships", {})
+ .get("resources", {})
+ .get("data", [])
+ )
+ if rel_resources and isinstance(rel_resources, list):
+ resource_name = rel_resources[0].get("id", "")
+ else:
+ resource_name = ""
+ severity = attr.get("severity", "").capitalize()
+ status = attr.get("status", "").upper()
+ region = attr.get("region", "global")
+
+ findings_table_data.append(
+ [
+ Paragraph(title, normal_center),
+ Paragraph(resource_name, normal_center),
+ Paragraph(severity, normal_center),
+ Paragraph(status, normal_center),
+ Paragraph(region, normal_center),
+ ]
+ )
+ findings_table = Table(
+ findings_table_data,
+ colWidths=[
+ 2.5 * inch,
+ 3 * inch,
+ 0.9 * inch,
+ 0.9 * inch,
+ 0.9 * inch,
+ ],
+ )
+ findings_table.setStyle(
+ TableStyle(
+ [
+ (
+ "BACKGROUND",
+ (0, 0),
+ (-1, 0),
+ colors.Color(0.2, 0.4, 0.6),
+ ),
+ ("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
+ ("FONTNAME", (0, 0), (-1, 0), "FiraCode"),
+ ("ALIGN", (0, 0), (0, 0), "CENTER"),
+ ("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
+ ("FONTSIZE", (0, 0), (-1, -1), 9),
+ (
+ "GRID",
+ (0, 0),
+ (-1, -1),
+ 0.1,
+ colors.Color(0.7, 0.8, 0.9),
+ ),
+ ("LEFTPADDING", (0, 0), (0, 0), 0),
+ ("RIGHTPADDING", (0, 0), (0, 0), 0),
+ ("TOPPADDING", (0, 0), (-1, -1), 4),
+ ("BOTTOMPADDING", (0, 0), (-1, -1), 4),
+ ]
+ )
+ )
+ elements.append(findings_table)
+ elements.append(Spacer(1, 0.1 * inch))
+
+ elements.append(PageBreak())
+
+ def add_footer(canvas: canvas.Canvas, doc):
+ width, height = doc.pagesize
+ page_num_text = f"Page {doc.page}"
+ canvas.setFont("PlusJakartaSans", 9)
+ canvas.setFillColorRGB(0.4, 0.4, 0.4)
+ canvas.drawString(30, 20, page_num_text)
+ powered_text = "Powered by Prowler"
+ text_width = canvas.stringWidth(powered_text, "PlusJakartaSans", 9)
+ canvas.drawString(width - text_width - 30, 20, powered_text)
+
+ doc.build(elements, onFirstPage=add_footer, onLaterPages=add_footer)
+
+
+logger = get_task_logger(__name__)
+
+
+def generate_threatscore_report_job(tenant_id: str, scan_id: str, provider_id: str):
+ provider_obj = Provider.objects.get(id=provider_id)
+ provider_uid = provider_obj.uid
+ provider_type = provider_obj.provider
+
+ # This compliance is hardcoded because is the only one that is available for the threatscore report
+ compliance_id = f"prowler_threatscore_{provider_type}"
+
+ out_dir, _ = _generate_output_directory(
+ DJANGO_TMP_OUTPUT_DIRECTORY, provider_uid, tenant_id, scan_id
+ )
+ pdf_path = f"{out_dir}_threatscore_report.pdf"
+
+ generate_threatscore_report(
+ scan_id=scan_id,
+ compliance_id=compliance_id,
+ output_path=pdf_path,
+ provider_id=provider_id,
+ only_failed=True,
+ min_risk_level=4,
+ )
+
+ s3_uri = None
+ if os.path.exists(pdf_path):
+ try:
+ s3_uri = _upload_to_s3(tenant_id, pdf_path, scan_id)
+ except Exception as e:
+ logger.error(f"Error uploading PDF to S3: {e}")
+
+ try:
+ task = Task.objects.get(scan_id=scan_id)
+ result_path = s3_uri if s3_uri else pdf_path
+ task.result = result_path
+ task.save(update_fields=["result"])
+ except Exception as e:
+ logger.error(f"Error saving PDF location to Task: {e}")
+ return {"pdf_location": s3_uri if s3_uri else pdf_path}
diff --git a/api/src/backend/tasks/tasks.py b/api/src/backend/tasks/tasks.py
index 544103a1e3..0e2c2e3790 100644
--- a/api/src/backend/tasks/tasks.py
+++ b/api/src/backend/tasks/tasks.py
@@ -17,6 +17,7 @@ from tasks.jobs.export import (
_generate_output_directory,
_upload_to_s3,
)
+from tasks.jobs.report import generate_threatscore_report_job
from tasks.jobs.scan import (
aggregate_findings,
create_compliance_requirements,
@@ -55,6 +56,11 @@ def _perform_scan_complete_tasks(tenant_id: str, scan_id: str, provider_id: str)
scan_id=scan_id, provider_id=provider_id, tenant_id=tenant_id
),
).apply_async()
+ chain(
+ generate_threatscore_report_task.si(
+ tenant_id=tenant_id, scan_id=scan_id, provider_id=provider_id
+ ),
+ ).apply_async()
@shared_task(base=RLSTask, name="provider-connection-check")
@@ -419,4 +425,24 @@ def check_lighthouse_connection_task(lighthouse_config_id: str, tenant_id: str =
- 'error' (str or None): The error message if the connection failed, otherwise `None`.
- 'available_models' (list): List of available models if connection is successful.
"""
- return check_lighthouse_connection(lighthouse_config_id=lighthouse_config_id)
\ No newline at end of file
+ return check_lighthouse_connection(lighthouse_config_id=lighthouse_config_id)
+
+
+@shared_task(
+ base=RLSTask,
+ name="scan-threatscore-report",
+ queue="scan-reports",
+)
+@set_tenant(keep_tenant=True)
+def generate_threatscore_report_task(tenant_id: str, scan_id: str, provider_id: str):
+ """
+ Task to generate a threatscore report for a given scan.
+
+ Args:
+ tenant_id (str): The tenant identifier.
+ scan_id (str): The scan identifier.
+ provider_id (str): The provider identifier.
+ """
+ return generate_threatscore_report_job(
+ tenant_id=tenant_id, scan_id=scan_id, provider_id=provider_id
+ )