Merge branch 'master' into fix/resources-filters-and-drawer-polish

This commit is contained in:
Alejandro Bailo
2026-04-23 16:05:52 +02:00
committed by GitHub
23 changed files with 2412 additions and 106 deletions
+5 -1
View File
@@ -4,6 +4,10 @@ All notable changes to the **Prowler API** are documented in this file.
## [1.26.0] (Prowler UNRELEASED)
### 🚀 Added
- CIS Benchmark PDF report generation for scans, exposing the latest CIS version per provider via `GET /scans/{id}/cis/{name}/` and picking the variant dynamically via `_pick_latest_cis_variant` (no hard-coded provider → version mapping) [(#10650)](https://github.com/prowler-cloud/prowler/pull/10650)
### 🔄 Changed
- Allows tenant owners to expel users from their organizations [(#10787)](https://github.com/prowler-cloud/prowler/pull/10787)
@@ -736,4 +740,4 @@ All notable changes to the **Prowler API** are documented in this file.
- Findings metadata endpoint received a performance improvement [(#6863)](https://github.com/prowler-cloud/prowler/pull/6863)
- Increased the allowed length of the provider UID for Kubernetes providers [(#6869)](https://github.com/prowler-cloud/prowler/pull/6869)
---
---
+45
View File
@@ -4113,6 +4113,51 @@ class TestScanViewSet:
assert cd.startswith('attachment; filename="')
assert cd.endswith(f'filename="{fname.name}"')
def test_cis_no_output(self, authenticated_client, scans_fixture):
"""CIS PDF endpoint must 404 when the scan has no output_location."""
scan = scans_fixture[0]
scan.state = StateChoices.COMPLETED
scan.output_location = ""
scan.save()
url = reverse("scan-cis", kwargs={"pk": scan.id})
resp = authenticated_client.get(url)
assert resp.status_code == status.HTTP_404_NOT_FOUND
assert (
resp.json()["errors"]["detail"]
== "The scan has no reports, or the CIS report generation task has not started yet."
)
def test_cis_local_file(self, authenticated_client, scans_fixture, monkeypatch):
"""CIS PDF endpoint must serve the latest generated PDF."""
scan = scans_fixture[0]
scan.state = StateChoices.COMPLETED
with tempfile.TemporaryDirectory() as tmp:
tmp_path = Path(tmp)
base = tmp_path / "reports"
cis_dir = base / "cis"
cis_dir.mkdir(parents=True, exist_ok=True)
fname = cis_dir / "prowler-output-aws-20260101000000_cis_report.pdf"
fname.write_bytes(b"%PDF-1.4 fake pdf")
scan.output_location = str(base / "scan.zip")
scan.save()
monkeypatch.setattr(
glob,
"glob",
lambda p: [str(fname)] if p.endswith("*_cis_report.pdf") else [],
)
url = reverse("scan-cis", kwargs={"pk": scan.id})
resp = authenticated_client.get(url)
assert resp.status_code == status.HTTP_200_OK
assert resp["Content-Type"] == "application/pdf"
cd = resp["Content-Disposition"]
assert cd.startswith('attachment; filename="')
assert cd.endswith(f'filename="{fname.name}"')
@patch("api.v1.views.Task.objects.get")
@patch("api.v1.views.TaskSerializer")
def test__get_task_status_returns_none_if_task_not_executing(
+63
View File
@@ -1926,6 +1926,27 @@ class ProviderViewSet(DisablePaginationMixin, BaseRLSViewSet):
),
},
),
cis=extend_schema(
tags=["Scan"],
summary="Retrieve CIS Benchmark compliance report",
description="Download the CIS Benchmark compliance report as a PDF file. "
"When a provider ships multiple CIS versions, the report is generated "
"for the highest available version.",
request=None,
responses={
200: OpenApiResponse(
description="PDF file containing the CIS compliance report"
),
202: OpenApiResponse(description="The task is in progress"),
401: OpenApiResponse(
description="API key missing or user not Authenticated"
),
403: OpenApiResponse(description="There is a problem with credentials"),
404: OpenApiResponse(
description="The scan has no CIS reports, or the CIS report generation task has not started yet"
),
},
),
)
@method_decorator(CACHE_DECORATOR, name="list")
@method_decorator(CACHE_DECORATOR, name="retrieve")
@@ -1994,6 +2015,9 @@ class ScanViewSet(BaseRLSViewSet):
elif self.action == "csa":
if hasattr(self, "response_serializer_class"):
return self.response_serializer_class
elif self.action == "cis":
if hasattr(self, "response_serializer_class"):
return self.response_serializer_class
return super().get_serializer_class()
def partial_update(self, request, *args, **kwargs):
@@ -2236,6 +2260,45 @@ class ScanViewSet(BaseRLSViewSet):
content, filename = loader
return self._serve_file(content, filename, "text/csv")
@action(
detail=True,
methods=["get"],
url_name="cis",
)
def cis(self, request, pk=None):
scan = self.get_object()
running_resp = self._get_task_status(scan)
if running_resp:
return running_resp
if not scan.output_location:
return Response(
{
"detail": "The scan has no reports, or the CIS 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),
"cis",
"*_cis_report.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, "cis", "*_cis_report.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")
@action(
detail=True,
methods=["get"],
Binary file not shown.

After

Width:  |  Height:  |  Size: 131 KiB

+223 -10
View File
@@ -1,3 +1,6 @@
import gc
import re
from collections.abc import Iterable
from pathlib import Path
from shutil import rmtree
@@ -6,6 +9,7 @@ from config.django.base import DJANGO_TMP_OUTPUT_DIRECTORY
from tasks.jobs.export import _generate_compliance_output_directory, _upload_to_s3
from tasks.jobs.reports import (
FRAMEWORK_REGISTRY,
CISReportGenerator,
CSAReportGenerator,
ENSReportGenerator,
NIS2ReportGenerator,
@@ -17,10 +21,53 @@ from tasks.jobs.threatscore_utils import _aggregate_requirement_statistics_from_
from api.db_router import READ_REPLICA_ALIAS
from api.db_utils import rls_transaction
from api.models import Provider, ScanSummary, ThreatScoreSnapshot
from prowler.lib.check.compliance_models import Compliance
from prowler.lib.outputs.finding import Finding as FindingOutput
logger = get_task_logger(__name__)
# Matches CIS compliance_ids like "cis_1.4_aws", "cis_5.0_azure",
# "cis_1.10_kubernetes", "cis_3.0.1_aws". Requires at least one dotted
# component so malformed inputs like "cis_._aws" or "cis_5._aws" are rejected
# at the regex stage, rather than by a later ValueError fallback.
_CIS_VARIANT_RE = re.compile(r"^cis_(?P<version>\d+(?:\.\d+)+)_(?P<provider>.+)$")
def _pick_latest_cis_variant(compliance_ids: Iterable[str]) -> str | None:
"""Return the CIS compliance_id with the highest semantic version.
CIS ships many variants per provider (e.g. cis_1.4_aws, ..., cis_6.0_aws).
A lexicographic sort is incorrect for version strings like ``1.10`` vs
``1.2``; this helper parses the version into a tuple of ints so ``1.10``
is correctly ordered after ``1.2``. Malformed names are skipped so a
broken JSON cannot crash the whole CIS pipeline.
Args:
compliance_ids: Iterable of CIS compliance identifiers. Expected to
belong to a single provider (callers should pass the already
filtered keys from ``Compliance.get_bulk(provider_type)``).
Returns:
The compliance_id with the highest parsed version, or ``None`` if no
well-formed CIS identifier was found.
"""
best_key: tuple[int, ...] | None = None
best_name: str | None = None
for name in compliance_ids:
match = _CIS_VARIANT_RE.match(name)
if not match:
continue
try:
key = tuple(int(part) for part in match.group("version").split("."))
except ValueError:
# Defensive: the regex already guarantees numeric chunks, but we
# keep the guard so a future regex change cannot crash callers.
continue
if best_key is None or key > best_key:
best_key = key
best_name = name
return best_name
def generate_threatscore_report(
tenant_id: str,
@@ -191,6 +238,53 @@ def generate_csa_report(
)
def generate_cis_report(
tenant_id: str,
scan_id: str,
compliance_id: str,
output_path: str,
provider_id: str,
only_failed: bool = True,
include_manual: bool = False,
provider_obj: Provider | None = None,
requirement_statistics: dict[str, dict[str, int]] | None = None,
findings_cache: dict[str, list[FindingOutput]] | None = None,
) -> None:
"""
Generate a PDF compliance report for a specific CIS Benchmark variant.
Unlike single-version frameworks (ENS, NIS2, CSA), CIS has multiple
variants per provider (e.g., cis_1.4_aws, cis_5.0_aws, cis_6.0_aws). This
wrapper is called once per variant, receiving the specific compliance_id.
Args:
tenant_id: The tenant ID for Row-Level Security context.
scan_id: ID of the scan executed by Prowler.
compliance_id: ID of the specific CIS variant (e.g., "cis_5.0_aws").
output_path: Output PDF file path.
provider_id: Provider ID for the scan.
only_failed: If True, only include failed requirements in detailed section.
include_manual: If True, include manual requirements in detailed section.
provider_obj: Pre-fetched Provider object to avoid duplicate queries.
requirement_statistics: Pre-aggregated requirement statistics.
findings_cache: Cache of already loaded findings to avoid duplicate queries.
"""
generator = CISReportGenerator(FRAMEWORK_REGISTRY["cis"])
generator.generate(
tenant_id=tenant_id,
scan_id=scan_id,
compliance_id=compliance_id,
output_path=output_path,
provider_id=provider_id,
provider_obj=provider_obj,
requirement_statistics=requirement_statistics,
findings_cache=findings_cache,
only_failed=only_failed,
include_manual=include_manual,
)
def generate_compliance_reports(
tenant_id: str,
scan_id: str,
@@ -199,6 +293,7 @@ def generate_compliance_reports(
generate_ens: bool = True,
generate_nis2: bool = True,
generate_csa: bool = True,
generate_cis: bool = True,
only_failed_threatscore: bool = True,
min_risk_level_threatscore: int = 4,
include_manual_ens: bool = True,
@@ -206,6 +301,8 @@ def generate_compliance_reports(
only_failed_nis2: bool = True,
only_failed_csa: bool = True,
include_manual_csa: bool = False,
only_failed_cis: bool = True,
include_manual_cis: bool = False,
) -> dict[str, dict[str, bool | str]]:
"""
Generate multiple compliance reports with shared database queries.
@@ -215,6 +312,13 @@ def generate_compliance_reports(
- Aggregating requirement statistics once (shared across all reports)
- Reusing compliance framework data when possible
For CIS a single PDF is produced per run: the one matching the highest
available CIS version for the scan's provider (picked dynamically from
``Compliance.get_bulk`` via :func:`_pick_latest_cis_variant`). The
returned ``results["cis"]`` entry has the same flat shape as the other
single-version frameworks — the picked variant is an internal detail,
not surfaced in the result.
Args:
tenant_id: The tenant ID for Row-Level Security context.
scan_id: The ID of the scan to generate reports for.
@@ -223,6 +327,8 @@ def generate_compliance_reports(
generate_ens: Whether to generate ENS report.
generate_nis2: Whether to generate NIS2 report.
generate_csa: Whether to generate CSA CCM report.
generate_cis: Whether to generate a CIS Benchmark report for the
latest CIS version available for the provider.
only_failed_threatscore: For ThreatScore, only include failed requirements.
min_risk_level_threatscore: Minimum risk level for ThreatScore critical requirements.
include_manual_ens: For ENS, include manual requirements.
@@ -230,22 +336,26 @@ def generate_compliance_reports(
only_failed_nis2: For NIS2, only include failed requirements.
only_failed_csa: For CSA CCM, only include failed requirements.
include_manual_csa: For CSA CCM, include manual requirements.
only_failed_cis: For CIS, only include failed requirements in detailed section.
include_manual_cis: For CIS, include manual requirements in detailed section.
Returns:
Dictionary with results for each report type.
Dictionary with results for each report type. Every value has the
same flat shape: ``{"upload": bool, "path": str, "error"?: str}``.
"""
logger.info(
"Generating compliance reports for scan %s with provider %s"
" (ThreatScore: %s, ENS: %s, NIS2: %s, CSA: %s)",
" (ThreatScore: %s, ENS: %s, NIS2: %s, CSA: %s, CIS: %s)",
scan_id,
provider_id,
generate_threatscore,
generate_ens,
generate_nis2,
generate_csa,
generate_cis,
)
results = {}
results: dict = {}
# Validate that the scan has findings and get provider info
with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
@@ -259,6 +369,8 @@ def generate_compliance_reports(
results["nis2"] = {"upload": False, "path": ""}
if generate_csa:
results["csa"] = {"upload": False, "path": ""}
if generate_cis:
results["cis"] = {"upload": False, "path": ""}
return results
provider_obj = Provider.objects.get(id=provider_id)
@@ -299,11 +411,18 @@ def generate_compliance_reports(
results["csa"] = {"upload": False, "path": ""}
generate_csa = False
# For CIS we do NOT pre-check the provider against a hard-coded whitelist
# (that list drifts the moment a new CIS JSON ships). Instead, we let
# `_pick_latest_cis_variant` over `Compliance.get_bulk(provider_type)`
# return None for providers that lack CIS, and treat that as "nothing to
# do" below.
if (
not generate_threatscore
and not generate_ens
and not generate_nis2
and not generate_csa
and not generate_cis
):
return results
@@ -350,6 +469,13 @@ def generate_compliance_reports(
scan_id,
compliance_framework="csa",
)
cis_path = _generate_compliance_output_directory(
DJANGO_TMP_OUTPUT_DIRECTORY,
provider_uid,
tenant_id,
scan_id,
compliance_framework="cis",
)
out_dir = str(Path(threatscore_path).parent.parent)
except Exception as e:
logger.error("Error generating output directory: %s", e)
@@ -362,6 +488,8 @@ def generate_compliance_reports(
results["nis2"] = error_dict.copy()
if generate_csa:
results["csa"] = error_dict.copy()
if generate_cis:
results["cis"] = error_dict.copy()
return results
# Generate ThreatScore report
@@ -569,12 +697,92 @@ def generate_compliance_reports(
logger.error("Error generating CSA CCM report: %s", e)
results["csa"] = {"upload": False, "path": "", "error": str(e)}
# Clean up temporary files if all reports were uploaded successfully
all_uploaded = all(
result.get("upload", False)
for result in results.values()
if result.get("upload") is not None
)
# Generate CIS Benchmark report for the latest available version only.
# CIS ships multiple versions per provider (e.g. cis_1.4_aws, cis_5.0_aws,
# cis_6.0_aws); we dynamically pick the highest semantic version at run
# time rather than hard-coding a per-provider mapping. `Compliance.get_bulk`
# is the single source of truth for which providers have CIS.
if generate_cis:
latest_cis: str | None = None
try:
frameworks_bulk = Compliance.get_bulk(provider_type)
latest_cis = _pick_latest_cis_variant(
name for name in frameworks_bulk.keys() if name.startswith("cis_")
)
except Exception as e:
logger.error("Error discovering CIS variants for %s: %s", provider_type, e)
results["cis"] = {"upload": False, "path": "", "error": str(e)}
if "cis" not in results:
if latest_cis is None:
logger.info("No CIS variants available for provider %s", provider_type)
results["cis"] = {"upload": False, "path": ""}
else:
logger.info(
"Selected latest CIS variant for provider %s: %s",
provider_type,
latest_cis,
)
pdf_path_cis = f"{cis_path}_cis_report.pdf"
try:
generate_cis_report(
tenant_id=tenant_id,
scan_id=scan_id,
compliance_id=latest_cis,
output_path=pdf_path_cis,
provider_id=provider_id,
only_failed=only_failed_cis,
include_manual=include_manual_cis,
provider_obj=provider_obj,
requirement_statistics=requirement_statistics,
findings_cache=findings_cache,
)
upload_uri_cis = _upload_to_s3(
tenant_id,
scan_id,
pdf_path_cis,
f"cis/{Path(pdf_path_cis).name}",
)
if upload_uri_cis:
results["cis"] = {
"upload": True,
"path": upload_uri_cis,
}
logger.info(
"CIS report %s uploaded to %s",
latest_cis,
upload_uri_cis,
)
else:
results["cis"] = {"upload": False, "path": out_dir}
logger.warning(
"CIS report %s saved locally at %s",
latest_cis,
out_dir,
)
except Exception as e:
logger.error("Error generating CIS report %s: %s", latest_cis, e)
results["cis"] = {
"upload": False,
"path": "",
"error": str(e),
}
finally:
# Free ReportLab/matplotlib memory before moving on.
gc.collect()
# Clean up temporary files only if every requested report has been
# successfully uploaded. All result entries now share the same flat
# shape, so the check is a single comprehension.
upload_flags = [
bool(entry.get("upload", False))
for entry in results.values()
if isinstance(entry, dict) and entry.get("upload") is not None
]
all_uploaded = bool(upload_flags) and all(upload_flags)
if all_uploaded:
try:
@@ -595,6 +803,7 @@ def generate_compliance_reports_job(
generate_ens: bool = True,
generate_nis2: bool = True,
generate_csa: bool = True,
generate_cis: bool = True,
) -> dict[str, dict[str, bool | str]]:
"""
Celery task wrapper for generate_compliance_reports.
@@ -607,9 +816,12 @@ def generate_compliance_reports_job(
generate_ens: Whether to generate ENS report.
generate_nis2: Whether to generate NIS2 report.
generate_csa: Whether to generate CSA CCM report.
generate_cis: Whether to generate the CIS Benchmark report for the
latest CIS version available for the provider.
Returns:
Dictionary with results for each report type.
Dictionary with results for each report type. Every entry shares the
same flat ``{"upload", "path", "error"?}`` shape.
"""
return generate_compliance_reports(
tenant_id=tenant_id,
@@ -619,4 +831,5 @@ def generate_compliance_reports_job(
generate_ens=generate_ens,
generate_nis2=generate_nis2,
generate_csa=generate_csa,
generate_cis=generate_cis,
)
@@ -17,6 +17,9 @@ from .charts import (
get_chart_color_for_percentage,
)
# Framework-specific generators
from .cis import CISReportGenerator
# Reusable components
# Reusable components: Color helpers, Badge components, Risk component,
# Table components, Section components
@@ -31,10 +34,12 @@ from .components import (
create_section_header,
create_status_badge,
create_summary_table,
escape_html,
get_color_for_compliance,
get_color_for_risk_level,
get_color_for_weight,
get_status_color,
truncate_text,
)
# Framework configuration: Main configuration, Color constants, ENS colors,
@@ -90,8 +95,6 @@ from .config import (
FrameworkConfig,
get_framework_config,
)
# Framework-specific generators
from .csa import CSAReportGenerator
from .ens import ENSReportGenerator
from .nis2 import NIS2ReportGenerator
@@ -109,6 +112,7 @@ __all__ = [
"ENSReportGenerator",
"NIS2ReportGenerator",
"CSAReportGenerator",
"CISReportGenerator",
# Configuration
"FrameworkConfig",
"FRAMEWORK_REGISTRY",
@@ -182,6 +186,9 @@ __all__ = [
# Section components
"create_section_header",
"create_summary_table",
# Text helpers
"truncate_text",
"escape_html",
# Chart functions
"get_chart_color_for_percentage",
"create_vertical_bar_chart",
+755
View File
@@ -0,0 +1,755 @@
import os
import re
from collections import defaultdict
from typing import Any
from reportlab.lib.units import inch
from reportlab.platypus import Image, PageBreak, Paragraph, Spacer, Table, TableStyle
from api.models import StatusChoices
from .base import (
BaseComplianceReportGenerator,
ComplianceData,
RequirementData,
get_requirement_metadata,
)
from .charts import (
create_horizontal_bar_chart,
create_pie_chart,
create_stacked_bar_chart,
get_chart_color_for_percentage,
)
from .components import ColumnConfig, create_data_table, escape_html, truncate_text
from .config import (
CHART_COLOR_GREEN_1,
CHART_COLOR_RED,
CHART_COLOR_YELLOW,
COLOR_BG_BLUE,
COLOR_BLUE,
COLOR_BORDER_GRAY,
COLOR_DARK_GRAY,
COLOR_GRAY,
COLOR_GRID_GRAY,
COLOR_HIGH_RISK,
COLOR_LIGHT_BLUE,
COLOR_SAFE,
COLOR_WHITE,
)
# Ordered buckets used both in the executive summary tables and the charts
# section. Exposed as module constants so the two call sites never drift.
_PROFILE_BUCKET_ORDER: tuple[str, ...] = ("L1", "L2", "Other")
_ASSESSMENT_BUCKET_ORDER: tuple[str, ...] = ("Automated", "Manual")
# Anchored matchers for profile normalization — substring checks on "L1"/"L2"
# would happily match unrelated tokens like "CL2 Worker" or "HL2" coming from
# future CIS profile enum values.
_LEVEL_2_RE = re.compile(r"(?:\bLevel\s*2\b|\bL2\b|Level_2)")
_LEVEL_1_RE = re.compile(r"(?:\bLevel\s*1\b|\bL1\b|Level_1)")
def _normalize_profile(profile: Any) -> str:
"""Bucket a CIS Profile enum/string into one of: ``L1``, ``L2``, ``Other``.
The ``CIS_Requirement_Attribute_Profile`` enum has values like
``"Level 1"``, ``"Level 2"``, ``"E3 Level 1"``, ``"E5 Level 2"``. We
collapse them into three buckets to keep charts and badges readable
across CIS variants, using anchored regex matches so that future enum
values cannot accidentally promote e.g. ``"CL2 Worker"`` into ``L2``.
Args:
profile: The profile value (enum member, string, or ``None``).
Returns:
One of ``"L1"``, ``"L2"``, ``"Other"``.
"""
if profile is None:
return "Other"
value = getattr(profile, "value", None) or str(profile)
if _LEVEL_2_RE.search(value):
return "L2"
if _LEVEL_1_RE.search(value):
return "L1"
return "Other"
def _profile_badge_text(bucket: str) -> str:
"""Map a normalized profile bucket (L1/L2/Other) to a short badge label."""
return {"L1": "Level 1", "L2": "Level 2"}.get(bucket, "Other")
# =============================================================================
# CIS Report Generator
# =============================================================================
class CISReportGenerator(BaseComplianceReportGenerator):
"""
PDF report generator for CIS (Center for Internet Security) Benchmarks.
CIS differs from single-version frameworks (ENS, NIS2, CSA) in that:
- Each provider has multiple CIS versions (e.g. AWS: 1.4, 1.5, ..., 6.0).
- Section names differ across versions and providers and MUST be derived
at runtime from the loaded compliance data.
- Requirements carry Profile (Level 1/Level 2) and AssessmentStatus
(Automated/Manual) attributes that drive the executive summary and
charts.
This generator produces:
- Cover page with Prowler logo and dynamic CIS version/provider metadata
- Executive summary with overall compliance score, counts, and breakdowns
by Profile and AssessmentStatus
- Charts: overall status pie, pass rate by section (horizontal bar),
Level 1 vs Level 2 pass/fail distribution (stacked bar)
- Requirements index grouped by dynamic section
- Detailed findings for FAIL requirements with CIS-specific audit /
remediation / rationale details
"""
# Per-run memoization cache for ``_compute_statistics``. ``generate()``
# is the public entry point and is called once per PDF, so scoping the
# cache to the last seen ComplianceData instance is enough to avoid the
# double computation between executive summary and charts section.
_stats_cache_key: int | None = None
_stats_cache_value: dict | None = None
# Body section ordering — ensure every top-level section starts on its
# own clean page. The base class only puts a PageBreak AFTER Charts and
# Requirements Index, so Executive Summary and Charts end up sharing a
# page. This override prepends a PageBreak so Compliance Analysis always
# begins on a fresh page.
def _build_body_sections(self, data: ComplianceData) -> list:
return [PageBreak(), *super()._build_body_sections(data)]
# -------------------------------------------------------------------------
# Cover page override — shows dynamic CIS version + provider in the title
# -------------------------------------------------------------------------
def create_cover_page(self, data: ComplianceData) -> list:
"""Create the CIS report cover page with Prowler + CIS logos side by side."""
elements = []
# Create logos side by side (same pattern as NIS2 / ENS)
prowler_logo_path = os.path.join(
os.path.dirname(__file__), "../../assets/img/prowler_logo.png"
)
cis_logo_path = os.path.join(
os.path.dirname(__file__), "../../assets/img/cis_logo.png"
)
if os.path.exists(cis_logo_path):
prowler_logo = Image(prowler_logo_path, width=3.5 * inch, height=0.7 * inch)
cis_logo = Image(cis_logo_path, width=2.3 * inch, height=1.1 * inch)
logos_table = Table(
[[prowler_logo, cis_logo]], colWidths=[4 * inch, 2.5 * inch]
)
logos_table.setStyle(
TableStyle(
[
("ALIGN", (0, 0), (0, 0), "LEFT"),
("ALIGN", (1, 0), (1, 0), "RIGHT"),
("VALIGN", (0, 0), (0, 0), "MIDDLE"),
("VALIGN", (1, 0), (1, 0), "MIDDLE"),
]
)
)
elements.append(logos_table)
elif os.path.exists(prowler_logo_path):
# Fallback: only the Prowler logo if the CIS asset is missing
elements.append(Image(prowler_logo_path, width=5 * inch, height=1 * inch))
elements.append(Spacer(1, 0.5 * inch))
# Dynamic title: "CIS Benchmark v5.0 — AWS Compliance Report"
provider_label = ""
if data.provider_obj:
provider_label = f"{data.provider_obj.provider.upper()}"
title_text = (
f"CIS Benchmark v{data.version}{provider_label}<br/>Compliance Report"
)
elements.append(Paragraph(title_text, self.styles["title"]))
elements.append(Spacer(1, 0.5 * inch))
# Metadata table via base class helper
info_rows = self._build_info_rows(data, language=self.config.language)
metadata_data = []
for label, value in info_rows:
if label in ("Name:", "Description:") and value:
metadata_data.append(
[label, Paragraph(str(value), self.styles["normal_center"])]
)
else:
metadata_data.append([label, value])
metadata_table = Table(metadata_data, colWidths=[2 * inch, 4 * inch])
metadata_table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (0, -1), COLOR_BLUE),
("TEXTCOLOR", (0, 0), (0, -1), COLOR_WHITE),
("FONTNAME", (0, 0), (0, -1), "FiraCode"),
("BACKGROUND", (1, 0), (1, -1), COLOR_BG_BLUE),
("TEXTCOLOR", (1, 0), (1, -1), COLOR_GRAY),
("FONTNAME", (1, 0), (1, -1), "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, COLOR_BORDER_GRAY),
("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(metadata_table)
return elements
# -------------------------------------------------------------------------
# Executive Summary
# -------------------------------------------------------------------------
def create_executive_summary(self, data: ComplianceData) -> list:
"""Create the CIS executive summary section."""
elements = []
elements.append(Paragraph("Executive Summary", self.styles["h1"]))
elements.append(Spacer(1, 0.1 * inch))
stats = self._compute_statistics(data)
# --- Summary metrics table ---
summary_data = [
["Metric", "Value"],
["Total Requirements", str(stats["total"])],
["Passed", str(stats["passed"])],
["Failed", str(stats["failed"])],
["Manual", str(stats["manual"])],
["Overall Compliance", f"{stats['overall_compliance']:.1f}%"],
]
summary_table = Table(summary_data, colWidths=[3 * inch, 2 * inch])
summary_table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, 0), COLOR_BLUE),
("TEXTCOLOR", (0, 0), (-1, 0), COLOR_WHITE),
("BACKGROUND", (0, 2), (0, 2), COLOR_SAFE),
("TEXTCOLOR", (0, 2), (0, 2), COLOR_WHITE),
("BACKGROUND", (0, 3), (0, 3), COLOR_HIGH_RISK),
("TEXTCOLOR", (0, 3), (0, 3), COLOR_WHITE),
("BACKGROUND", (0, 4), (0, 4), COLOR_DARK_GRAY),
("TEXTCOLOR", (0, 4), (0, 4), COLOR_WHITE),
("FONTNAME", (0, 0), (-1, 0), "PlusJakartaSans"),
("FONTSIZE", (0, 0), (-1, 0), 12),
("FONTSIZE", (0, 1), (-1, -1), 10),
("ALIGN", (0, 0), (-1, -1), "CENTER"),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("GRID", (0, 0), (-1, -1), 0.5, COLOR_BORDER_GRAY),
("BOTTOMPADDING", (0, 0), (-1, 0), 10),
(
"ROWBACKGROUNDS",
(1, 1),
(1, -1),
[COLOR_WHITE, COLOR_BG_BLUE],
),
]
)
)
elements.append(summary_table)
elements.append(Spacer(1, 0.25 * inch))
# --- Profile breakdown table ---
elements.append(Paragraph("Breakdown by Profile", self.styles["h2"]))
elements.append(Spacer(1, 0.1 * inch))
profile_counts = stats["profile_counts"]
profile_table_data = [["Profile", "Passed", "Failed", "Manual", "Total"]]
for bucket in _PROFILE_BUCKET_ORDER:
counts = profile_counts.get(bucket, {"passed": 0, "failed": 0, "manual": 0})
total = counts["passed"] + counts["failed"] + counts["manual"]
if total == 0:
continue
profile_table_data.append(
[
_profile_badge_text(bucket),
str(counts["passed"]),
str(counts["failed"]),
str(counts["manual"]),
str(total),
]
)
profile_table = Table(
profile_table_data,
colWidths=[1.5 * inch, 1 * inch, 1 * inch, 1 * inch, 1 * inch],
)
profile_table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, 0), COLOR_BLUE),
("TEXTCOLOR", (0, 0), (-1, 0), COLOR_WHITE),
("FONTNAME", (0, 0), (-1, 0), "FiraCode"),
("FONTSIZE", (0, 0), (-1, 0), 10),
("ALIGN", (0, 0), (-1, -1), "CENTER"),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("FONTSIZE", (0, 1), (-1, -1), 9),
("GRID", (0, 0), (-1, -1), 0.5, COLOR_GRID_GRAY),
(
"ROWBACKGROUNDS",
(0, 1),
(-1, -1),
[COLOR_WHITE, COLOR_BG_BLUE],
),
]
)
)
elements.append(profile_table)
elements.append(Spacer(1, 0.25 * inch))
# --- Assessment status breakdown ---
elements.append(Paragraph("Breakdown by Assessment Status", self.styles["h2"]))
elements.append(Spacer(1, 0.1 * inch))
assessment_counts = stats["assessment_counts"]
assessment_table_data = [["Assessment", "Passed", "Failed", "Manual", "Total"]]
for bucket in _ASSESSMENT_BUCKET_ORDER:
counts = assessment_counts.get(
bucket, {"passed": 0, "failed": 0, "manual": 0}
)
total = counts["passed"] + counts["failed"] + counts["manual"]
if total == 0:
continue
assessment_table_data.append(
[
bucket,
str(counts["passed"]),
str(counts["failed"]),
str(counts["manual"]),
str(total),
]
)
assessment_table = Table(
assessment_table_data,
colWidths=[1.5 * inch, 1 * inch, 1 * inch, 1 * inch, 1 * inch],
)
assessment_table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, 0), COLOR_LIGHT_BLUE),
("TEXTCOLOR", (0, 0), (-1, 0), COLOR_WHITE),
("FONTNAME", (0, 0), (-1, 0), "FiraCode"),
("FONTSIZE", (0, 0), (-1, 0), 10),
("ALIGN", (0, 0), (-1, -1), "CENTER"),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("FONTSIZE", (0, 1), (-1, -1), 9),
("GRID", (0, 0), (-1, -1), 0.5, COLOR_GRID_GRAY),
(
"ROWBACKGROUNDS",
(0, 1),
(-1, -1),
[COLOR_WHITE, COLOR_BG_BLUE],
),
]
)
)
elements.append(assessment_table)
elements.append(Spacer(1, 0.25 * inch))
# --- Top 5 failing sections ---
top_failing = stats["top_failing_sections"]
if top_failing:
elements.append(
Paragraph("Top Sections with Lowest Compliance", self.styles["h2"])
)
elements.append(Spacer(1, 0.1 * inch))
top_table_data = [["Section", "Passed", "Failed", "Compliance"]]
for section_label, section_stats in top_failing:
passed = section_stats["passed"]
failed = section_stats["failed"]
total = passed + failed
pct = (passed / total * 100) if total > 0 else 100
top_table_data.append(
[
truncate_text(section_label, 55),
str(passed),
str(failed),
f"{pct:.1f}%",
]
)
top_table = Table(
top_table_data,
colWidths=[3.5 * inch, 0.9 * inch, 0.9 * inch, 1.2 * inch],
)
top_table.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, 0), COLOR_HIGH_RISK),
("TEXTCOLOR", (0, 0), (-1, 0), COLOR_WHITE),
("FONTNAME", (0, 0), (-1, 0), "FiraCode"),
("FONTSIZE", (0, 0), (-1, 0), 10),
("ALIGN", (0, 0), (-1, -1), "CENTER"),
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
("FONTSIZE", (0, 1), (-1, -1), 9),
("GRID", (0, 0), (-1, -1), 0.5, COLOR_GRID_GRAY),
(
"ROWBACKGROUNDS",
(0, 1),
(-1, -1),
[COLOR_WHITE, COLOR_BG_BLUE],
),
]
)
)
elements.append(top_table)
return elements
# -------------------------------------------------------------------------
# Charts section
# -------------------------------------------------------------------------
def create_charts_section(self, data: ComplianceData) -> list:
"""Create the CIS charts section."""
elements = []
elements.append(Paragraph("Compliance Analysis", self.styles["h1"]))
elements.append(Spacer(1, 0.1 * inch))
# --- Pie chart: overall Pass / Fail / Manual ---
stats = self._compute_statistics(data)
pie_labels = []
pie_values = []
pie_colors = []
if stats["passed"] > 0:
pie_labels.append(f"Pass ({stats['passed']})")
pie_values.append(stats["passed"])
pie_colors.append(CHART_COLOR_GREEN_1)
if stats["failed"] > 0:
pie_labels.append(f"Fail ({stats['failed']})")
pie_values.append(stats["failed"])
pie_colors.append(CHART_COLOR_RED)
if stats["manual"] > 0:
pie_labels.append(f"Manual ({stats['manual']})")
pie_values.append(stats["manual"])
pie_colors.append(CHART_COLOR_YELLOW)
if pie_values:
elements.append(Paragraph("Overall Status Distribution", self.styles["h2"]))
elements.append(Spacer(1, 0.1 * inch))
pie_buffer = create_pie_chart(
labels=pie_labels,
values=pie_values,
colors=pie_colors,
)
pie_buffer.seek(0)
elements.append(Image(pie_buffer, width=4.5 * inch, height=4.5 * inch))
elements.append(Spacer(1, 0.2 * inch))
# --- Horizontal bar: pass rate by section ---
section_stats = stats["section_stats"]
if section_stats:
elements.append(PageBreak())
elements.append(Paragraph("Compliance by Section", self.styles["h1"]))
elements.append(Spacer(1, 0.1 * inch))
elements.append(
Paragraph(
"The following chart shows compliance percentage for each CIS "
"section based on automated checks:",
self.styles["normal_center"],
)
)
elements.append(Spacer(1, 0.1 * inch))
# Sort sections by pass rate descending for readability
sorted_sections = sorted(
section_stats.items(),
key=lambda item: (
(item[1]["passed"] / (item[1]["passed"] + item[1]["failed"]) * 100)
if (item[1]["passed"] + item[1]["failed"]) > 0
else 100
),
reverse=True,
)
bar_labels = []
bar_values = []
for section_label, section_data in sorted_sections:
total = section_data["passed"] + section_data["failed"]
if total == 0:
continue
pct = (section_data["passed"] / total) * 100
bar_labels.append(truncate_text(section_label, 60))
bar_values.append(pct)
if bar_values:
bar_buffer = create_horizontal_bar_chart(
labels=bar_labels,
values=bar_values,
xlabel="Compliance (%)",
color_func=get_chart_color_for_percentage,
label_fontsize=9,
)
bar_buffer.seek(0)
elements.append(Image(bar_buffer, width=6.5 * inch, height=5 * inch))
# --- Stacked bar: Level 1 vs Level 2 pass/fail ---
profile_counts = stats["profile_counts"]
has_profile_data = any(
(counts["passed"] + counts["failed"]) > 0
for counts in profile_counts.values()
)
if has_profile_data:
elements.append(PageBreak())
elements.append(Paragraph("Profile Breakdown", self.styles["h1"]))
elements.append(Spacer(1, 0.1 * inch))
elements.append(
Paragraph(
"Distribution of Pass / Fail / Manual across CIS profile levels.",
self.styles["normal_center"],
)
)
elements.append(Spacer(1, 0.1 * inch))
profile_labels = []
pass_series = []
fail_series = []
manual_series = []
for bucket in _PROFILE_BUCKET_ORDER:
counts = profile_counts.get(bucket)
if not counts:
continue
total = counts["passed"] + counts["failed"] + counts["manual"]
if total == 0:
continue
profile_labels.append(_profile_badge_text(bucket))
pass_series.append(counts["passed"])
fail_series.append(counts["failed"])
manual_series.append(counts["manual"])
if profile_labels:
stacked_buffer = create_stacked_bar_chart(
labels=profile_labels,
data_series={
"Pass": pass_series,
"Fail": fail_series,
"Manual": manual_series,
},
xlabel="Profile",
ylabel="Requirements",
)
stacked_buffer.seek(0)
elements.append(Image(stacked_buffer, width=6 * inch, height=4 * inch))
return elements
# -------------------------------------------------------------------------
# Requirements Index
# -------------------------------------------------------------------------
def create_requirements_index(self, data: ComplianceData) -> list:
"""Create the CIS requirements index grouped by dynamic section."""
elements = []
elements.append(Paragraph("Requirements Index", self.styles["h1"]))
elements.append(Spacer(1, 0.1 * inch))
sections = self._derive_sections(data)
by_section: dict[str, list[dict]] = defaultdict(list)
for req in data.requirements:
meta = get_requirement_metadata(req.id, data.attributes_by_requirement_id)
section = "Other"
profile_bucket = "Other"
assessment = ""
if meta:
section = getattr(meta, "Section", "Other") or "Other"
profile_bucket = _normalize_profile(getattr(meta, "Profile", None))
assessment_enum = getattr(meta, "AssessmentStatus", None)
assessment = getattr(assessment_enum, "value", None) or str(
assessment_enum or ""
)
by_section[section].append(
{
"id": req.id,
"description": truncate_text(req.description, 80),
"profile": _profile_badge_text(profile_bucket),
"assessment": assessment or "-",
"status": (req.status or "").upper(),
}
)
columns = [
ColumnConfig("ID", 0.9 * inch, "id", align="LEFT"),
ColumnConfig("Description", 3.0 * inch, "description", align="LEFT"),
ColumnConfig("Profile", 0.9 * inch, "profile"),
ColumnConfig("Assessment", 1 * inch, "assessment"),
ColumnConfig("Status", 0.9 * inch, "status"),
]
for section in sections:
rows = by_section.get(section, [])
if not rows:
continue
elements.append(Paragraph(truncate_text(section, 90), self.styles["h2"]))
elements.append(Spacer(1, 0.05 * inch))
table = create_data_table(
data=rows,
columns=columns,
header_color=self.config.primary_color,
normal_style=self.styles["normal_center"],
)
elements.append(table)
elements.append(Spacer(1, 0.15 * inch))
return elements
# -------------------------------------------------------------------------
# Detailed findings hook — inject CIS-specific rationale / audit content
# -------------------------------------------------------------------------
def _render_requirement_detail_extras(
self, req: RequirementData, data: ComplianceData
) -> list:
"""Render CIS rationale, impact, audit, remediation and references."""
extras = []
meta = get_requirement_metadata(req.id, data.attributes_by_requirement_id)
if meta is None:
return extras
field_map = [
("Rationale", "RationaleStatement"),
("Impact", "ImpactStatement"),
("Audit Procedure", "AuditProcedure"),
("Remediation", "RemediationProcedure"),
("References", "References"),
]
for label, attr_name in field_map:
value = getattr(meta, attr_name, None)
if not value:
continue
text = str(value).strip()
if not text:
continue
extras.append(Paragraph(f"<b>{label}:</b>", self.styles["h3"]))
extras.append(Paragraph(escape_html(text), self.styles["normal"]))
extras.append(Spacer(1, 0.08 * inch))
return extras
# -------------------------------------------------------------------------
# Private helpers
# -------------------------------------------------------------------------
def _derive_sections(self, data: ComplianceData) -> list[str]:
"""Extract ordered unique Section names from loaded compliance data."""
seen: dict[str, bool] = {}
for req in data.requirements:
meta = get_requirement_metadata(req.id, data.attributes_by_requirement_id)
if meta is None:
continue
section = getattr(meta, "Section", None) or "Other"
if section not in seen:
seen[section] = True
return list(seen.keys())
def _compute_statistics(self, data: ComplianceData) -> dict:
"""Aggregate all statistics needed for summary and charts.
Memoized per-``ComplianceData`` instance via ``_stats_cache_*``: the
executive summary and the charts section both need the same numbers,
so they would otherwise re-iterate the requirements twice. We key on
``id(data)`` because ``ComplianceData`` is a dataclass and its
instances are not hashable.
Returns a dict with:
- total, passed, failed, manual: int
- overall_compliance: float (percentage)
- profile_counts: {"L1": {"passed", "failed", "manual"}, ...}
- assessment_counts: {"Automated": {...}, "Manual": {...}}
- section_stats: {section_name: {"passed", "failed", "manual"}, ...}
- top_failing_sections: list[(section_name, stats)] (up to 5)
"""
cache_key = id(data)
if self._stats_cache_key == cache_key and self._stats_cache_value is not None:
return self._stats_cache_value
stats = self._compute_statistics_uncached(data)
self._stats_cache_key = cache_key
self._stats_cache_value = stats
return stats
def _compute_statistics_uncached(self, data: ComplianceData) -> dict:
"""Actual aggregation kernel; call ``_compute_statistics`` instead."""
total = len(data.requirements)
passed = sum(1 for r in data.requirements if r.status == StatusChoices.PASS)
failed = sum(1 for r in data.requirements if r.status == StatusChoices.FAIL)
manual = sum(1 for r in data.requirements if r.status == StatusChoices.MANUAL)
evaluated = passed + failed
overall_compliance = (passed / evaluated * 100) if evaluated > 0 else 100.0
profile_counts: dict[str, dict[str, int]] = {
"L1": {"passed": 0, "failed": 0, "manual": 0},
"L2": {"passed": 0, "failed": 0, "manual": 0},
"Other": {"passed": 0, "failed": 0, "manual": 0},
}
assessment_counts: dict[str, dict[str, int]] = {
"Automated": {"passed": 0, "failed": 0, "manual": 0},
"Manual": {"passed": 0, "failed": 0, "manual": 0},
}
section_stats: dict[str, dict[str, int]] = defaultdict(
lambda: {"passed": 0, "failed": 0, "manual": 0}
)
for req in data.requirements:
meta = get_requirement_metadata(req.id, data.attributes_by_requirement_id)
if meta is None:
continue
profile_bucket = _normalize_profile(getattr(meta, "Profile", None))
assessment_enum = getattr(meta, "AssessmentStatus", None)
assessment_value = getattr(assessment_enum, "value", None) or str(
assessment_enum or ""
)
assessment_bucket = (
"Automated" if assessment_value == "Automated" else "Manual"
)
section = getattr(meta, "Section", None) or "Other"
status_key = {
StatusChoices.PASS: "passed",
StatusChoices.FAIL: "failed",
StatusChoices.MANUAL: "manual",
}.get(req.status)
if status_key is None:
continue
profile_counts[profile_bucket][status_key] += 1
assessment_counts[assessment_bucket][status_key] += 1
section_stats[section][status_key] += 1
# Top 5 sections with lowest pass rate (only sections with evaluated reqs)
def _section_rate(item):
_, stats_ = item
evaluated_ = stats_["passed"] + stats_["failed"]
if evaluated_ == 0:
return 101 # sort evaluated=0 to the bottom
return stats_["passed"] / evaluated_ * 100
top_failing_sections = sorted(
(
item
for item in section_stats.items()
if (item[1]["passed"] + item[1]["failed"]) > 0
),
key=_section_rate,
)[:5]
return {
"total": total,
"passed": passed,
"failed": failed,
"manual": manual,
"overall_compliance": overall_compliance,
"profile_counts": profile_counts,
"assessment_counts": assessment_counts,
"section_stats": dict(section_stats),
"top_failing_sections": top_failing_sections,
}
@@ -26,6 +26,52 @@ from .config import (
)
def truncate_text(text: str, max_len: int) -> str:
"""Truncate ``text`` to ``max_len`` characters, appending an ellipsis if cut.
Used by report generators that need to squeeze long descriptions, section
titles or finding titles into a fixed-width table cell.
Args:
text: Source string. ``None`` and non-string values are treated as empty.
max_len: Maximum output length including the ellipsis. Values < 4 are
clamped so the result never grows beyond ``max_len``.
Returns:
The original string if short enough, otherwise ``text[: max_len - 3] + "..."``.
When ``max_len < 4`` a plain substring of length ``max_len`` is returned
so callers never get a string longer than they asked for.
"""
if not text:
return ""
text = str(text)
if len(text) <= max_len:
return text
if max_len < 4:
return text[:max_len]
return text[: max_len - 3] + "..."
def escape_html(text: str) -> str:
"""Escape the minimal HTML entities required for safe ReportLab Paragraph rendering.
ReportLab's ``Paragraph`` parses a small HTML subset, so raw ``<``, ``>``
and ``&`` in user-provided content (rationale, remediation, etc.) would
break layout or be interpreted as tags. This helper mirrors
``html.escape`` but avoids pulling in the stdlib dependency and keeps the
output deterministic.
Args:
text: Untrusted source string.
Returns:
A string safe to embed inside a ReportLab Paragraph.
"""
return (
str(text or "").replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
)
def get_color_for_risk_level(risk_level: int) -> colors.Color:
"""
Get color based on risk level.
@@ -313,6 +313,32 @@ FRAMEWORK_REGISTRY: dict[str, FrameworkConfig] = {
has_niveles=False,
has_weight=False,
),
"cis": FrameworkConfig(
name="cis",
display_name="CIS Benchmark",
logo_filename=None,
primary_color=COLOR_BLUE,
secondary_color=COLOR_LIGHT_BLUE,
bg_color=COLOR_BG_BLUE,
attribute_fields=[
"Section",
"SubSection",
"Profile",
"AssessmentStatus",
"Description",
"RationaleStatement",
"ImpactStatement",
"RemediationProcedure",
"AuditProcedure",
"References",
],
sections=None, # Derived dynamically per CIS variant (section names differ across versions/providers)
language="en",
has_risk_levels=False,
has_dimensions=False,
has_niveles=False,
has_weight=False,
),
}
@@ -336,5 +362,7 @@ def get_framework_config(compliance_id: str) -> FrameworkConfig | None:
return FRAMEWORK_REGISTRY["nis2"]
if "csa" in compliance_lower or "ccm" in compliance_lower:
return FRAMEWORK_REGISTRY["csa_ccm"]
if compliance_lower.startswith("cis_") or "cis" in compliance_lower:
return FRAMEWORK_REGISTRY["cis"]
return None
+6 -1
View File
@@ -1000,13 +1000,17 @@ def jira_integration_task(
@handle_provider_deletion
def generate_compliance_reports_task(tenant_id: str, scan_id: str, provider_id: str):
"""
Optimized task to generate ThreatScore, ENS, NIS2, and CSA CCM reports with shared queries.
Optimized task to generate ThreatScore, ENS, NIS2, CSA CCM and CIS reports with shared queries.
This task is more efficient than running separate report tasks because it reuses database queries:
- Provider object fetched once (instead of multiple times)
- Requirement statistics aggregated once (instead of multiple times)
- Can reduce database load by up to 50-70%
CIS emits a single PDF per run: the one matching the highest CIS version
available for the scan's provider, picked dynamically from
``Compliance.get_bulk`` (no hard-coded provider → version mapping).
Args:
tenant_id (str): The tenant identifier.
scan_id (str): The scan identifier.
@@ -1023,6 +1027,7 @@ def generate_compliance_reports_task(tenant_id: str, scan_id: str, provider_id:
generate_ens=True,
generate_nis2=True,
generate_csa=True,
generate_cis=True,
)
+265 -1
View File
@@ -4,7 +4,11 @@ from unittest.mock import Mock, patch
import matplotlib
import pytest
from reportlab.lib import colors
from tasks.jobs.report import generate_compliance_reports, generate_threatscore_report
from tasks.jobs.report import (
_pick_latest_cis_variant,
generate_compliance_reports,
generate_threatscore_report,
)
from tasks.jobs.reports import (
CHART_COLOR_GREEN_1,
CHART_COLOR_GREEN_2,
@@ -422,6 +426,266 @@ class TestGenerateComplianceReportsOptimized:
mock_ens.assert_not_called()
mock_nis2.assert_not_called()
@patch("tasks.jobs.report._upload_to_s3")
@patch("tasks.jobs.report.generate_cis_report")
def test_no_findings_returns_flat_cis_entry(
self,
mock_cis,
mock_upload,
tenants_fixture,
scans_fixture,
providers_fixture,
):
"""Scan with no findings and ``generate_cis=True`` must yield a flat
``{"upload": False, "path": ""}`` entry, consistent with the other
frameworks (no nested dict, no sentinel keys)."""
tenant = tenants_fixture[0]
scan = scans_fixture[0]
provider = providers_fixture[0]
result = generate_compliance_reports(
tenant_id=str(tenant.id),
scan_id=str(scan.id),
provider_id=str(provider.id),
generate_threatscore=False,
generate_ens=False,
generate_nis2=False,
generate_csa=False,
generate_cis=True,
)
assert result["cis"] == {"upload": False, "path": ""}
mock_cis.assert_not_called()
@pytest.mark.django_db
class TestGenerateComplianceReportsCIS:
"""Test suite covering the CIS branch of generate_compliance_reports."""
def _force_scan_has_findings(self, monkeypatch):
"""Bypass the ScanSummary.exists() early-return guard."""
class _FakeManager:
def filter(self, **kwargs):
class _Q:
def exists(self):
return True
return _Q()
monkeypatch.setattr("tasks.jobs.report.ScanSummary.objects", _FakeManager())
@patch("tasks.jobs.report._aggregate_requirement_statistics_from_database")
@patch("tasks.jobs.report._upload_to_s3")
@patch("tasks.jobs.report.generate_cis_report")
@patch("tasks.jobs.report.Compliance.get_bulk")
def test_cis_picks_latest_version(
self,
mock_get_bulk,
mock_cis,
mock_upload,
mock_stats,
monkeypatch,
tenants_fixture,
scans_fixture,
providers_fixture,
):
"""CIS branch should generate a single PDF for the highest version.
The returned ``results["cis"]`` must have the same flat shape as the
other single-version frameworks (``{"upload", "path"}``) — the picked
variant is an internal detail and is not exposed in the result.
"""
tenant = tenants_fixture[0]
scan = scans_fixture[0]
provider = providers_fixture[0]
self._force_scan_has_findings(monkeypatch)
mock_stats.return_value = {}
# Multiple CIS variants + a non-CIS framework that must be ignored.
# Includes 1.10 to verify the selection is not lexicographic.
mock_get_bulk.return_value = {
"cis_1.4_aws": Mock(),
"cis_1.10_aws": Mock(),
"cis_2.0_aws": Mock(),
"cis_5.0_aws": Mock(),
"ens_rd2022_aws": Mock(),
}
mock_upload.return_value = "s3://bucket/path"
result = generate_compliance_reports(
tenant_id=str(tenant.id),
scan_id=str(scan.id),
provider_id=str(provider.id),
generate_threatscore=False,
generate_ens=False,
generate_nis2=False,
generate_csa=False,
generate_cis=True,
)
# Exactly one call for the latest version, never for older variants
# or non-CIS frameworks.
assert mock_cis.call_count == 1
assert mock_cis.call_args.kwargs["compliance_id"] == "cis_5.0_aws"
assert result["cis"]["upload"] is True
assert result["cis"]["path"] == "s3://bucket/path"
assert "compliance_id" not in result["cis"]
@patch("tasks.jobs.report._aggregate_requirement_statistics_from_database")
@patch("tasks.jobs.report._upload_to_s3")
@patch("tasks.jobs.report.generate_cis_report")
@patch("tasks.jobs.report.Compliance.get_bulk")
def test_cis_latest_variant_failure_captured_in_results(
self,
mock_get_bulk,
mock_cis,
mock_upload,
mock_stats,
monkeypatch,
tenants_fixture,
scans_fixture,
providers_fixture,
):
"""A failure in the latest CIS variant must be surfaced in the flat results entry."""
tenant = tenants_fixture[0]
scan = scans_fixture[0]
provider = providers_fixture[0]
self._force_scan_has_findings(monkeypatch)
mock_stats.return_value = {}
mock_get_bulk.return_value = {
"cis_1.4_aws": Mock(),
"cis_5.0_aws": Mock(),
}
mock_cis.side_effect = RuntimeError("boom")
result = generate_compliance_reports(
tenant_id=str(tenant.id),
scan_id=str(scan.id),
provider_id=str(provider.id),
generate_threatscore=False,
generate_ens=False,
generate_nis2=False,
generate_csa=False,
generate_cis=True,
)
# Only the latest variant is attempted; its failure lands in a flat
# entry keyed under "cis" with the same shape as sibling frameworks.
assert mock_cis.call_count == 1
assert result["cis"]["upload"] is False
assert result["cis"]["error"] == "boom"
assert "compliance_id" not in result["cis"]
@patch("tasks.jobs.report._aggregate_requirement_statistics_from_database")
@patch("tasks.jobs.report._upload_to_s3")
@patch("tasks.jobs.report.generate_cis_report")
@patch("tasks.jobs.report.Compliance.get_bulk")
def test_cis_provider_without_cis_skipped_cleanly(
self,
mock_get_bulk,
mock_cis,
mock_upload,
mock_stats,
monkeypatch,
tenants_fixture,
scans_fixture,
providers_fixture,
):
"""When ``Compliance.get_bulk`` returns no CIS entry the CIS branch
must skip cleanly and record a flat ``{"upload": False, "path": ""}``
entry — no hard-coded provider whitelist is consulted."""
tenant = tenants_fixture[0]
scan = scans_fixture[0]
provider = providers_fixture[0]
self._force_scan_has_findings(monkeypatch)
mock_stats.return_value = {}
# No ``cis_*`` keys in the bulk → no variant picked.
mock_get_bulk.return_value = {"ens_rd2022_aws": Mock()}
result = generate_compliance_reports(
tenant_id=str(tenant.id),
scan_id=str(scan.id),
provider_id=str(provider.id),
generate_threatscore=False,
generate_ens=False,
generate_nis2=False,
generate_csa=False,
generate_cis=True,
)
assert result["cis"] == {"upload": False, "path": ""}
mock_cis.assert_not_called()
class TestPickLatestCisVariant:
"""Unit tests for `_pick_latest_cis_variant` helper."""
def test_empty_returns_none(self):
assert _pick_latest_cis_variant([]) is None
def test_single_variant(self):
assert _pick_latest_cis_variant(["cis_5.0_aws"]) == "cis_5.0_aws"
def test_numeric_not_lexicographic(self):
"""1.10 must beat 1.2 (lex sort would pick 1.2)."""
variants = ["cis_1.2_kubernetes", "cis_1.10_kubernetes"]
assert _pick_latest_cis_variant(variants) == "cis_1.10_kubernetes"
def test_major_version_wins(self):
variants = ["cis_1.4_aws", "cis_2.0_aws", "cis_5.0_aws", "cis_6.0_aws"]
assert _pick_latest_cis_variant(variants) == "cis_6.0_aws"
def test_minor_version_breaks_tie(self):
variants = ["cis_3.0_aws", "cis_3.1_aws", "cis_2.9_aws"]
assert _pick_latest_cis_variant(variants) == "cis_3.1_aws"
def test_three_part_version(self):
"""Versions like 3.0.1 must win over 3.0."""
variants = ["cis_3.0_aws", "cis_3.0.1_aws"]
assert _pick_latest_cis_variant(variants) == "cis_3.0.1_aws"
def test_malformed_names_ignored(self):
variants = ["notcis_1.0_aws", "cis_abc_aws", "cis_5.0_aws"]
assert _pick_latest_cis_variant(variants) == "cis_5.0_aws"
def test_only_malformed_returns_none(self):
variants = ["notcis_1.0_aws", "cis_abc_aws"]
assert _pick_latest_cis_variant(variants) is None
def test_multidigit_provider_name(self):
"""Provider name with underscores (e.g. googleworkspace) must parse."""
variants = ["cis_1.3_googleworkspace"]
assert _pick_latest_cis_variant(variants) == "cis_1.3_googleworkspace"
def test_accepts_iterator(self):
"""The helper must accept any iterable, not just lists."""
def _gen():
yield "cis_1.4_aws"
yield "cis_5.0_aws"
assert _pick_latest_cis_variant(_gen()) == "cis_5.0_aws"
def test_rejects_single_integer_version(self):
"""The regex requires at least one dotted component. ``cis_5_aws``
without a minor version is malformed per the backend contract."""
assert _pick_latest_cis_variant(["cis_5_aws"]) is None
def test_rejects_trailing_dot(self):
"""Inputs like ``cis_5._aws`` must be rejected at the regex stage
instead of silently normalising to ``(5, 0)``."""
assert _pick_latest_cis_variant(["cis_5._aws", "cis_1.0_aws"]) == "cis_1.0_aws"
def test_rejects_lone_dot_version(self):
"""``cis_._aws`` has no numeric component and must be skipped."""
assert _pick_latest_cis_variant(["cis_._aws", "cis_1.0_aws"]) == "cis_1.0_aws"
class TestOptimizationImprovements:
"""Test suite for optimization-related functionality."""
@@ -0,0 +1,532 @@
from unittest.mock import Mock, patch
import pytest
from reportlab.platypus import Image, LongTable, Paragraph, Table
from tasks.jobs.reports import FRAMEWORK_REGISTRY, ComplianceData, RequirementData
from tasks.jobs.reports.cis import (
CISReportGenerator,
_normalize_profile,
_profile_badge_text,
)
from api.models import StatusChoices
# =============================================================================
# Fixtures
# =============================================================================
@pytest.fixture
def cis_generator():
"""Create a CISReportGenerator instance for testing."""
config = FRAMEWORK_REGISTRY["cis"]
return CISReportGenerator(config)
def _make_attr(
section: str,
profile_value: str = "Level 1",
assessment_value: str = "Automated",
sub_section: str = "",
**extras,
) -> Mock:
"""Build a mock CIS_Requirement_Attribute with duck-typed fields."""
attr = Mock()
attr.Section = section
attr.SubSection = sub_section
# CIS enums have `.value`. Use a simple Mock that exposes `.value`.
attr.Profile = Mock(value=profile_value)
attr.AssessmentStatus = Mock(value=assessment_value)
attr.Description = extras.get("description", "desc")
attr.RationaleStatement = extras.get("rationale", "the rationale")
attr.ImpactStatement = extras.get("impact", "the impact")
attr.RemediationProcedure = extras.get("remediation", "the remediation")
attr.AuditProcedure = extras.get("audit", "the audit")
attr.AdditionalInformation = ""
attr.DefaultValue = ""
attr.References = extras.get("references", "https://example.com")
return attr
@pytest.fixture
def basic_cis_compliance_data():
"""Create basic ComplianceData for CIS testing (no requirements)."""
return ComplianceData(
tenant_id="tenant-123",
scan_id="scan-456",
provider_id="provider-789",
compliance_id="cis_5.0_aws",
framework="CIS",
name="CIS Amazon Web Services Foundations Benchmark v5.0.0",
version="5.0",
description="Center for Internet Security AWS Foundations Benchmark",
)
@pytest.fixture
def populated_cis_compliance_data(basic_cis_compliance_data):
"""CIS data with mixed requirements across 2 sections, Profile L1/L2, Pass/Fail/Manual."""
data = basic_cis_compliance_data
data.requirements = [
RequirementData(
id="1.1",
description="Maintain current contact details",
status=StatusChoices.PASS,
passed_findings=5,
failed_findings=0,
total_findings=5,
checks=["aws_check_1"],
),
RequirementData(
id="1.2",
description="Ensure root account has no access keys",
status=StatusChoices.FAIL,
passed_findings=0,
failed_findings=3,
total_findings=3,
checks=["aws_check_2"],
),
RequirementData(
id="1.3",
description="Ensure MFA is enabled for all IAM users",
status=StatusChoices.MANUAL,
checks=[],
),
RequirementData(
id="2.1",
description="Ensure S3 Buckets are logging",
status=StatusChoices.PASS,
passed_findings=2,
failed_findings=0,
total_findings=2,
checks=["aws_check_3"],
),
RequirementData(
id="2.2",
description="Ensure encryption at rest is enabled",
status=StatusChoices.FAIL,
passed_findings=0,
failed_findings=4,
total_findings=4,
checks=["aws_check_4"],
),
]
data.attributes_by_requirement_id = {
"1.1": {
"attributes": {
"req_attributes": [
_make_attr(
"1 Identity and Access Management",
profile_value="Level 1",
assessment_value="Automated",
)
],
"checks": ["aws_check_1"],
}
},
"1.2": {
"attributes": {
"req_attributes": [
_make_attr(
"1 Identity and Access Management",
profile_value="Level 1",
assessment_value="Automated",
)
],
"checks": ["aws_check_2"],
}
},
"1.3": {
"attributes": {
"req_attributes": [
_make_attr(
"1 Identity and Access Management",
profile_value="Level 2",
assessment_value="Manual",
)
],
"checks": [],
}
},
"2.1": {
"attributes": {
"req_attributes": [
_make_attr(
"2 Storage",
profile_value="Level 2",
assessment_value="Automated",
)
],
"checks": ["aws_check_3"],
}
},
"2.2": {
"attributes": {
"req_attributes": [
_make_attr(
"2 Storage",
profile_value="Level 1",
assessment_value="Automated",
)
],
"checks": ["aws_check_4"],
}
},
}
return data
# =============================================================================
# Helper function tests
# =============================================================================
class TestNormalizeProfile:
"""Test suite for _normalize_profile helper."""
def test_level_1_string(self):
assert _normalize_profile(Mock(value="Level 1")) == "L1"
def test_level_2_string(self):
assert _normalize_profile(Mock(value="Level 2")) == "L2"
def test_e3_level_1(self):
assert _normalize_profile(Mock(value="E3 Level 1")) == "L1"
def test_e5_level_2(self):
assert _normalize_profile(Mock(value="E5 Level 2")) == "L2"
def test_none_returns_other(self):
assert _normalize_profile(None) == "Other"
def test_substring_trap_rejected(self):
"""Unrelated tokens containing the literal ``L2`` must NOT map to L2."""
# A future enum value like "CL2 Kubernetes Worker" would be silently
# misclassified by a naive substring check.
assert _normalize_profile(Mock(value="CL2 Worker")) == "Other"
assert _normalize_profile(Mock(value="HL2 Legacy")) == "Other"
def test_raw_string_level_1(self):
# Mock without .value falls back to str(profile); use a real string
class NoValue:
def __str__(self):
return "Level 1"
assert _normalize_profile(NoValue()) == "L1"
def test_unknown_profile_returns_other(self):
assert _normalize_profile(Mock(value="Custom Profile")) == "Other"
class TestProfileBadgeText:
def test_l1_label(self):
assert _profile_badge_text("L1") == "Level 1"
def test_l2_label(self):
assert _profile_badge_text("L2") == "Level 2"
def test_other_label(self):
assert _profile_badge_text("Other") == "Other"
# =============================================================================
# Generator initialization
# =============================================================================
class TestCISGeneratorInitialization:
def test_generator_created(self, cis_generator):
assert cis_generator is not None
assert cis_generator.config.name == "cis"
def test_generator_language(self, cis_generator):
assert cis_generator.config.language == "en"
def test_generator_sections_dynamic(self, cis_generator):
# CIS sections differ per variant so config.sections MUST be None
assert cis_generator.config.sections is None
def test_attribute_fields_contain_cis_specific(self, cis_generator):
for field in ("Profile", "AssessmentStatus", "RationaleStatement"):
assert field in cis_generator.config.attribute_fields
# =============================================================================
# _derive_sections
# =============================================================================
class TestDeriveSections:
def test_preserves_first_seen_order(
self, cis_generator, populated_cis_compliance_data
):
sections = cis_generator._derive_sections(populated_cis_compliance_data)
assert sections == [
"1 Identity and Access Management",
"2 Storage",
]
def test_deduplicates_sections(self, cis_generator, basic_cis_compliance_data):
basic_cis_compliance_data.requirements = [
RequirementData(id="1.1", description="a", status=StatusChoices.PASS),
RequirementData(id="1.2", description="b", status=StatusChoices.PASS),
]
attr = _make_attr("1 IAM")
basic_cis_compliance_data.attributes_by_requirement_id = {
"1.1": {"attributes": {"req_attributes": [attr], "checks": []}},
"1.2": {"attributes": {"req_attributes": [attr], "checks": []}},
}
assert cis_generator._derive_sections(basic_cis_compliance_data) == ["1 IAM"]
def test_empty_data_returns_empty(self, cis_generator, basic_cis_compliance_data):
basic_cis_compliance_data.requirements = []
basic_cis_compliance_data.attributes_by_requirement_id = {}
assert cis_generator._derive_sections(basic_cis_compliance_data) == []
# =============================================================================
# _compute_statistics
# =============================================================================
class TestComputeStatistics:
def test_totals(self, cis_generator, populated_cis_compliance_data):
stats = cis_generator._compute_statistics(populated_cis_compliance_data)
assert stats["total"] == 5
assert stats["passed"] == 2
assert stats["failed"] == 2
assert stats["manual"] == 1
def test_overall_compliance_excludes_manual(
self, cis_generator, populated_cis_compliance_data
):
stats = cis_generator._compute_statistics(populated_cis_compliance_data)
# 2 passed / 4 evaluated (pass + fail) = 50%
assert stats["overall_compliance"] == pytest.approx(50.0)
def test_overall_compliance_all_manual(
self, cis_generator, basic_cis_compliance_data
):
basic_cis_compliance_data.requirements = [
RequirementData(id="x", description="d", status=StatusChoices.MANUAL),
]
attr = _make_attr("1 IAM", profile_value="Level 1", assessment_value="Manual")
basic_cis_compliance_data.attributes_by_requirement_id = {
"x": {"attributes": {"req_attributes": [attr], "checks": []}},
}
stats = cis_generator._compute_statistics(basic_cis_compliance_data)
# No evaluated → defaults to 100%
assert stats["overall_compliance"] == 100.0
def test_profile_counts(self, cis_generator, populated_cis_compliance_data):
stats = cis_generator._compute_statistics(populated_cis_compliance_data)
profile = stats["profile_counts"]
# From fixture:
# L1: 1.1 (PASS, Auto), 1.2 (FAIL, Auto), 2.2 (FAIL, Auto) → pass=1, fail=2, manual=0
# L2: 1.3 (MANUAL, Manual), 2.1 (PASS, Auto) → pass=1, fail=0, manual=1
assert profile["L1"] == {"passed": 1, "failed": 2, "manual": 0}
assert profile["L2"] == {"passed": 1, "failed": 0, "manual": 1}
def test_assessment_counts(self, cis_generator, populated_cis_compliance_data):
stats = cis_generator._compute_statistics(populated_cis_compliance_data)
assessment = stats["assessment_counts"]
# Automated: 1.1 PASS, 1.2 FAIL, 2.1 PASS, 2.2 FAIL → pass=2, fail=2, manual=0
# Manual: 1.3 MANUAL → pass=0, fail=0, manual=1
assert assessment["Automated"] == {"passed": 2, "failed": 2, "manual": 0}
assert assessment["Manual"] == {"passed": 0, "failed": 0, "manual": 1}
def test_top_failing_sections_includes_all_evaluated(
self, cis_generator, populated_cis_compliance_data
):
stats = cis_generator._compute_statistics(populated_cis_compliance_data)
top = stats["top_failing_sections"]
# Both sections have 1 PASS + 1 FAIL evaluated → tied at 50%. The
# sort is stable, so both must appear and both must be capped at
# 5 entries.
assert len(top) == 2
section_names = {name for name, _ in top}
assert section_names == {
"1 Identity and Access Management",
"2 Storage",
}
def test_compute_statistics_is_memoized(
self, cis_generator, populated_cis_compliance_data
):
"""Calling ``_compute_statistics`` twice with the same data must
reuse the cached value and not re-run the uncached kernel."""
with patch.object(
CISReportGenerator,
"_compute_statistics_uncached",
wraps=cis_generator._compute_statistics_uncached,
) as spy:
cis_generator._compute_statistics(populated_cis_compliance_data)
cis_generator._compute_statistics(populated_cis_compliance_data)
assert spy.call_count == 1
# =============================================================================
# Executive summary
# =============================================================================
class TestCISExecutiveSummary:
def test_title_present(self, cis_generator, populated_cis_compliance_data):
elements = cis_generator.create_executive_summary(populated_cis_compliance_data)
paragraphs = [e for e in elements if isinstance(e, Paragraph)]
text = " ".join(str(p.text) for p in paragraphs)
assert "Executive Summary" in text
def test_tables_rendered(self, cis_generator, populated_cis_compliance_data):
elements = cis_generator.create_executive_summary(populated_cis_compliance_data)
tables = [e for e in elements if isinstance(e, Table)]
# Exact count: Summary, Profile, Assessment, Top Failing Sections = 4.
assert len(tables) == 4
def test_no_requirements(self, cis_generator, basic_cis_compliance_data):
basic_cis_compliance_data.requirements = []
basic_cis_compliance_data.attributes_by_requirement_id = {}
elements = cis_generator.create_executive_summary(basic_cis_compliance_data)
# With no requirements: Summary table always renders, and both Profile
# and Assessment breakdown tables render with a 0-filled default row,
# but Top Failing Sections is suppressed → exactly 3 tables.
tables = [e for e in elements if isinstance(e, Table)]
assert len(tables) == 3
# =============================================================================
# Charts section
# =============================================================================
class TestCISChartsSection:
def test_charts_rendered(self, cis_generator, populated_cis_compliance_data):
elements = cis_generator.create_charts_section(populated_cis_compliance_data)
# At least 1 image for the pie + 1 for section bar + 1 for stacked
images = [e for e in elements if isinstance(e, Image)]
assert len(images) >= 1
def test_charts_no_data_no_crash(self, cis_generator, basic_cis_compliance_data):
basic_cis_compliance_data.requirements = []
basic_cis_compliance_data.attributes_by_requirement_id = {}
elements = cis_generator.create_charts_section(basic_cis_compliance_data)
# Must not raise; may or may not have any Image
assert isinstance(elements, list)
# =============================================================================
# Requirements index
# =============================================================================
class TestCISRequirementsIndex:
def test_title_present(self, cis_generator, populated_cis_compliance_data):
elements = cis_generator.create_requirements_index(
populated_cis_compliance_data
)
paragraphs = [e for e in elements if isinstance(e, Paragraph)]
text = " ".join(str(p.text) for p in paragraphs)
assert "Requirements Index" in text
def test_groups_by_section(self, cis_generator, populated_cis_compliance_data):
elements = cis_generator.create_requirements_index(
populated_cis_compliance_data
)
paragraphs = [e for e in elements if isinstance(e, Paragraph)]
text = " ".join(str(p.text) for p in paragraphs)
assert "1 Identity and Access Management" in text
assert "2 Storage" in text
def test_renders_tables_per_section(
self, cis_generator, populated_cis_compliance_data
):
elements = cis_generator.create_requirements_index(
populated_cis_compliance_data
)
# One table per section with requirements. ``create_data_table``
# returns a LongTable when the row count exceeds its threshold and a
# plain Table otherwise — both are valid.
tables = [e for e in elements if isinstance(e, (Table, LongTable))]
assert len(tables) == 2
# =============================================================================
# Detailed findings extras hook
# =============================================================================
class TestRenderRequirementDetailExtras:
def test_inserts_all_fields(self, cis_generator, populated_cis_compliance_data):
req = populated_cis_compliance_data.requirements[1] # 1.2 FAIL
extras = cis_generator._render_requirement_detail_extras(
req, populated_cis_compliance_data
)
text = " ".join(str(p.text) for p in extras if isinstance(p, Paragraph))
assert "Rationale" in text
assert "Impact" in text
assert "Audit Procedure" in text
assert "Remediation" in text
assert "References" in text
def test_missing_metadata_returns_empty(
self, cis_generator, basic_cis_compliance_data
):
basic_cis_compliance_data.attributes_by_requirement_id = {}
req = RequirementData(id="99", description="unknown", status=StatusChoices.FAIL)
extras = cis_generator._render_requirement_detail_extras(
req, basic_cis_compliance_data
)
assert extras == []
def test_escapes_html_chars(self, cis_generator, basic_cis_compliance_data):
attr = _make_attr(
"1 IAM",
rationale="<script>alert('x')</script>",
)
basic_cis_compliance_data.attributes_by_requirement_id = {
"1.1": {"attributes": {"req_attributes": [attr], "checks": []}}
}
req = RequirementData(id="1.1", description="d", status=StatusChoices.FAIL)
extras = cis_generator._render_requirement_detail_extras(
req, basic_cis_compliance_data
)
text = " ".join(str(p.text) for p in extras if isinstance(p, Paragraph))
assert "<script>" not in text
assert "&lt;script&gt;" in text
# =============================================================================
# Cover page
# =============================================================================
class TestCISCoverPage:
@patch("tasks.jobs.reports.cis.Image")
def test_cover_page_has_logo(
self, mock_image, cis_generator, basic_cis_compliance_data
):
elements = cis_generator.create_cover_page(basic_cis_compliance_data)
assert len(elements) > 0
assert mock_image.call_count >= 1
def test_cover_page_title_includes_version(
self, cis_generator, basic_cis_compliance_data
):
elements = cis_generator.create_cover_page(basic_cis_compliance_data)
paragraphs = [e for e in elements if isinstance(e, Paragraph)]
content = " ".join(str(p.text) for p in paragraphs)
assert "CIS Benchmark" in content
assert "5.0" in content
def test_cover_page_title_includes_provider_when_set(
self, cis_generator, basic_cis_compliance_data
):
provider = Mock()
provider.provider = "aws"
provider.uid = "123456789012"
provider.alias = "test-account"
basic_cis_compliance_data.provider_obj = provider
elements = cis_generator.create_cover_page(basic_cis_compliance_data)
paragraphs = [e for e in elements if isinstance(e, Paragraph)]
content = " ".join(str(p.text) for p in paragraphs)
assert "AWS" in content
+4 -1
View File
@@ -4,11 +4,14 @@ All notable changes to the **Prowler UI** are documented in this file.
## [1.25.0] (Prowler UNRELEASED)
### 🚀 Added
- Download PDF button for CIS Benchmark compliance cards, surfaced only on the latest CIS variant per provider to match the backend's latest-only PDF generation [(#10650)](https://github.com/prowler-cloud/prowler/pull/10650)
### 🔄 Changed
- Redesign compliance page, client-side search for compliance frameworks, compact scan selector trigger, enhanced compliance cards [(#10767)](https://github.com/prowler-cloud/prowler/pull/10767)
- Allows tenant owners to expel users from their organizations [(#10787)](https://github.com/prowler-cloud/prowler/pull/10787)
- Shared filter dropdowns now support local option search and auto-scroll to the first visible match across table and provider filters
- Resources now use batch-applied filters, render metadata JSON with syntax highlighting, simplify metadata/tags tabs, and keep finding-row Learn more links interactive inside the drawer [(#10861)](https://github.com/prowler-cloud/prowler/pull/10861)
- Shared filter dropdowns now support local option search and auto-scroll to the first visible match across table and provider filters [(#10859)](https://github.com/prowler-cloud/prowler/pull/10859)
- Backward-compatibility middleware redirect from `/sign-up?invitation_token=…` to `/invitation/accept?invitation_token=…`; new invitation emails use `/invitation/accept` directly [(#10797)](https://github.com/prowler-cloud/prowler/pull/10797)
+62 -63
View File
@@ -318,55 +318,87 @@ export const getExportsZip = async (scanId: string) => {
}
};
export const getComplianceCsv = async (
scanId: string,
complianceId: string,
) => {
const headers = await getAuthHeaders({ contentType: false });
/**
* Discriminated union returned by {@link _fetchScanBinary}.
*
* Exported so `ui/lib/helper.ts::downloadFile` can type-narrow on the
* `success` / `pending` / `error` tags without resorting to `any`.
*/
export type ScanBinaryResult =
| { success: true; data: string; filename: string }
| { pending: true; state: string | undefined; taskId: string | undefined }
| { error: string };
const url = new URL(
`${apiBaseUrl}/scans/${scanId}/compliance/${complianceId}`,
);
/**
* Shared binary-report fetcher used by CSV and PDF report downloads.
*
* All report endpoints (`/scans/{id}/compliance/{name}`,
* `/scans/{id}/{reportType}`) speak the same protocol: Bearer auth, 202
* ACCEPTED while the generation task is still running, 2xx with a binary
* body when the artifact is ready, JSON error body otherwise. This helper
* encapsulates all of that so the public wrappers only have to build the
* URL and pick a filename.
*
* @param urlPath Path segment under `{apiBaseUrl}/scans/{scanId}/`.
* @param filename Download filename to surface to the user.
* @param errorLabel Friendly label used when the backend error body is empty.
* @returns A ``{ success, data, filename }`` object on 2xx, a
* ``{ pending, state, taskId }`` object on 202, or
* ``{ error }`` on any failure.
*/
const _fetchScanBinary = async (
scanId: string,
urlPath: string,
filename: string,
errorLabel: string,
): Promise<ScanBinaryResult> => {
const headers = await getAuthHeaders({ contentType: false });
const url = new URL(`${apiBaseUrl}/scans/${scanId}/${urlPath}`);
try {
const response = await fetch(url.toString(), { headers });
if (response.status === 202) {
const json = await response.json();
const taskId = json?.data?.id;
const state = json?.data?.attributes?.state;
return {
pending: true,
state,
taskId,
state: json?.data?.attributes?.state,
taskId: json?.data?.id,
};
}
if (!response.ok) {
const errorData = await response.json();
const errorData = await response.json().catch(() => ({}));
throw new Error(
errorData?.errors?.detail ||
"Unable to retrieve compliance report. Contact support if the issue continues.",
`Unable to retrieve ${errorLabel}. Contact support if the issue continues.`,
);
}
const arrayBuffer = await response.arrayBuffer();
const base64 = Buffer.from(arrayBuffer).toString("base64");
return {
success: true,
data: base64,
filename: `scan-${scanId}-compliance-${complianceId}.csv`,
};
return { success: true, data: base64, filename };
} catch (error) {
return {
error: getErrorMessage(error),
};
return { error: getErrorMessage(error) };
}
};
export const getComplianceCsv = async (scanId: string, complianceId: string) =>
_fetchScanBinary(
scanId,
`compliance/${complianceId}`,
`scan-${scanId}-compliance-${complianceId}.csv`,
"compliance report",
);
/**
* Generic function to get a compliance PDF report (ThreatScore, ENS, etc.)
* Get a compliance PDF report for any supported framework.
*
* For frameworks with multiple variants per provider (currently CIS) the
* backend generates a single PDF for the highest available version, so
* callers only need to pass the generic report type.
*
* @param scanId - The scan ID
* @param reportType - Type of report (from COMPLIANCE_REPORT_TYPES)
* @returns Promise with the PDF data or error
@@ -375,44 +407,11 @@ export const getCompliancePdfReport = async (
scanId: string,
reportType: ComplianceReportType,
) => {
const headers = await getAuthHeaders({ contentType: false });
const url = new URL(`${apiBaseUrl}/scans/${scanId}/${reportType}`);
try {
const response = await fetch(url.toString(), { headers });
if (response.status === 202) {
const json = await response.json();
const taskId = json?.data?.id;
const state = json?.data?.attributes?.state;
return {
pending: true,
state,
taskId,
};
}
if (!response.ok) {
const errorData = await response.json();
const reportName = COMPLIANCE_REPORT_DISPLAY_NAMES[reportType];
throw new Error(
errorData?.errors?.detail ||
`Unable to retrieve ${reportName} PDF report. Contact support if the issue continues.`,
);
}
const arrayBuffer = await response.arrayBuffer();
const base64 = Buffer.from(arrayBuffer).toString("base64");
return {
success: true,
data: base64,
filename: `scan-${scanId}-${reportType}.pdf`,
};
} catch (error) {
return {
error: getErrorMessage(error),
};
}
const reportName = COMPLIANCE_REPORT_DISPLAY_NAMES[reportType];
return _fetchScanBinary(
scanId,
reportType,
`scan-${scanId}-${reportType}.pdf`,
`${reportName} PDF report`,
);
};
@@ -5,6 +5,7 @@ import {
getComplianceAttributes,
getComplianceOverviewMetadataInfo,
getComplianceRequirements,
getCompliancesOverview,
} from "@/actions/compliances";
import { getThreatScore } from "@/actions/overview";
import { getScan } from "@/actions/scans";
@@ -25,7 +26,10 @@ import {
import { getComplianceIcon } from "@/components/icons/compliance/IconCompliance";
import { ContentLayout } from "@/components/ui";
import { getComplianceMapper } from "@/lib/compliance/compliance-mapper";
import { getReportTypeForFramework } from "@/lib/compliance/compliance-report-types";
import {
getReportTypeForCompliance,
pickLatestCisPerProvider,
} from "@/lib/compliance/compliance-report-types";
import { cn } from "@/lib/utils";
import {
AttributesData,
@@ -113,6 +117,27 @@ export default async function ComplianceDetail({
}
}
// Only CIS variants need the "is this the latest version per provider?"
// check to gate the PDF download button. Every other framework either
// always has a PDF (ENS/NIS2/CSA/ThreatScore) or none at all, so we skip
// the extra compliance-overview roundtrip for non-CIS detail pages.
const needsCisLatestCheck =
typeof complianceId === "string" && complianceId.startsWith("cis_");
let latestCisIds: Set<string> = new Set<string>();
if (needsCisLatestCheck && selectedScanId) {
const scanCompliancesData = await getCompliancesOverview({
scanId: selectedScanId,
});
const scanComplianceIds: string[] = Array.isArray(scanCompliancesData?.data)
? scanCompliancesData.data
.map((c: { id?: string }) => c?.id)
.filter(
(id: string | undefined): id is string => typeof id === "string",
)
: [];
latestCisIds = pickLatestCisPerProvider(scanComplianceIds);
}
const uniqueRegions = metadataInfoData?.data?.attributes?.regions || [];
// Detect if this is a ThreatScore compliance view
@@ -158,8 +183,10 @@ export default async function ComplianceDetail({
<ComplianceDownloadContainer
scanId={selectedScanId}
complianceId={complianceId}
reportType={getReportTypeForFramework(
reportType={getReportTypeForCompliance(
attributesData?.data?.[0]?.attributes?.framework,
complianceId,
latestCisIds.has(complianceId),
)}
/>
</div>
+12
View File
@@ -17,6 +17,7 @@ import { ComplianceOverviewGrid } from "@/components/compliance/compliance-overv
import { Alert, AlertDescription } from "@/components/shadcn/alert";
import { Card, CardContent } from "@/components/shadcn/card/card";
import { ContentLayout } from "@/components/ui";
import { pickLatestCisPerProvider } from "@/lib/compliance/compliance-report-types";
import {
ExpandedScanData,
ScanEntity,
@@ -232,12 +233,23 @@ const SSRComplianceGrid = async ({
);
}
// Compute the set of latest CIS variants per provider once, so each card
// can gate its PDF button without re-parsing on every render. The backend
// only generates a CIS PDF for the latest version per provider, so any
// other CIS card must not expose the PDF download button.
const latestCisIds = pickLatestCisPerProvider(
compliancesData.data.map(
(compliance: ComplianceOverviewData) => compliance.id,
),
);
return (
<ComplianceOverviewPanel>
<ComplianceOverviewGrid
frameworks={frameworks}
scanId={scanId}
selectedScan={selectedScan}
latestCisIds={latestCisIds}
/>
</ComplianceOverviewPanel>
);
+14 -2
View File
@@ -10,7 +10,7 @@ import {
TooltipContent,
TooltipTrigger,
} from "@/components/shadcn/tooltip";
import { getReportTypeForFramework } from "@/lib/compliance/compliance-report-types";
import { getReportTypeForCompliance } from "@/lib/compliance/compliance-report-types";
import {
getScoreIndicatorClass,
type ScoreColorVariant,
@@ -31,6 +31,13 @@ interface ComplianceCardProps {
complianceId: string;
id: string;
selectedScan?: ScanEntity;
/**
* True when this compliance_id is the highest CIS version for its provider.
* Only the latest CIS variant per provider has a PDF generated by the
* backend, so older variants must not expose the PDF download button.
* Ignored for non-CIS frameworks.
*/
isLatestCisForProvider?: boolean;
}
export const ComplianceCard: React.FC<ComplianceCardProps> = ({
@@ -41,6 +48,7 @@ export const ComplianceCard: React.FC<ComplianceCardProps> = ({
scanId,
complianceId,
id,
isLatestCisForProvider = false,
}) => {
const searchParams = useSearchParams();
const router = useRouter();
@@ -112,7 +120,11 @@ export const ComplianceCard: React.FC<ComplianceCardProps> = ({
presentation="dropdown"
scanId={scanId}
complianceId={complianceId}
reportType={getReportTypeForFramework(title)}
reportType={getReportTypeForCompliance(
title,
complianceId,
isLatestCisForProvider,
)}
disabled={hasRegionFilter}
/>
</div>
@@ -6,15 +6,16 @@ import { render, screen } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import { beforeEach, describe, expect, it, vi } from "vitest";
const { downloadComplianceCsvMock, downloadComplianceReportPdfMock } =
vi.hoisted(() => ({
const { downloadComplianceCsvMock, downloadCompliancePdfMock } = vi.hoisted(
() => ({
downloadComplianceCsvMock: vi.fn(),
downloadComplianceReportPdfMock: vi.fn(),
}));
downloadCompliancePdfMock: vi.fn(),
}),
);
vi.mock("@/lib/helper", () => ({
downloadComplianceCsv: downloadComplianceCsvMock,
downloadComplianceReportPdf: downloadComplianceReportPdfMock,
downloadCompliancePdf: downloadCompliancePdfMock,
}));
vi.mock("@/components/ui", () => ({
@@ -124,7 +125,7 @@ describe("ComplianceDownloadContainer", () => {
"compliance-1",
{},
);
expect(downloadComplianceReportPdfMock).toHaveBeenCalledWith(
expect(downloadCompliancePdfMock).toHaveBeenCalledWith(
"scan-1",
"threatscore",
{},
@@ -15,10 +15,7 @@ import {
} from "@/components/shadcn/tooltip";
import { toast } from "@/components/ui";
import type { ComplianceReportType } from "@/lib/compliance/compliance-report-types";
import {
downloadComplianceCsv,
downloadComplianceReportPdf,
} from "@/lib/helper";
import { downloadComplianceCsv, downloadCompliancePdf } from "@/lib/helper";
import { cn } from "@/lib/utils";
interface ComplianceDownloadContainerProps {
@@ -61,7 +58,7 @@ export const ComplianceDownloadContainer = ({
if (!reportType || isDownloadingPdf) return;
setIsDownloadingPdf(true);
try {
await downloadComplianceReportPdf(scanId, reportType, toast);
await downloadCompliancePdf(scanId, reportType, toast);
} finally {
setIsDownloadingPdf(false);
}
@@ -11,12 +11,19 @@ interface ComplianceOverviewGridProps {
frameworks: ComplianceOverviewData[];
scanId: string;
selectedScan?: ScanEntity;
/**
* Subset of compliance_ids that represent the latest CIS variant per
* provider. Only those cards expose the PDF download button, matching
* the backend's latest-only CIS PDF generation.
*/
latestCisIds?: ReadonlySet<string>;
}
export const ComplianceOverviewGrid = ({
frameworks,
scanId,
selectedScan,
latestCisIds,
}: ComplianceOverviewGridProps) => {
const [searchTerm, setSearchTerm] = useState("");
@@ -61,6 +68,7 @@ export const ComplianceOverviewGrid = ({
complianceId={id}
id={id}
selectedScan={selectedScan}
isLatestCisForProvider={latestCisIds?.has(id) ?? false}
/>
);
})}
@@ -0,0 +1,156 @@
import { describe, expect, it } from "vitest";
import {
COMPLIANCE_REPORT_TYPES,
getReportTypeForCompliance,
getReportTypeForFramework,
pickLatestCisPerProvider,
} from "./compliance-report-types";
describe("getReportTypeForFramework", () => {
it("returns the framework-mapped type for single-version frameworks", () => {
expect(getReportTypeForFramework("ENS")).toBe(COMPLIANCE_REPORT_TYPES.ENS);
expect(getReportTypeForFramework("NIS2")).toBe(
COMPLIANCE_REPORT_TYPES.NIS2,
);
expect(getReportTypeForFramework("CSA-CCM")).toBe(
COMPLIANCE_REPORT_TYPES.CSA_CCM,
);
expect(getReportTypeForFramework("ProwlerThreatScore")).toBe(
COMPLIANCE_REPORT_TYPES.THREATSCORE,
);
});
it("returns undefined for CIS — callers must go through getReportTypeForCompliance", () => {
expect(getReportTypeForFramework("CIS")).toBeUndefined();
});
it("returns undefined for unknown frameworks", () => {
expect(getReportTypeForFramework("SomethingElse")).toBeUndefined();
});
it("returns undefined when framework is missing", () => {
expect(getReportTypeForFramework(undefined)).toBeUndefined();
});
});
describe("pickLatestCisPerProvider", () => {
it("returns an empty set for an empty input", () => {
const latest = pickLatestCisPerProvider([]);
expect(Array.from(latest)).toEqual([]);
});
it("returns a single variant when only one is provided", () => {
const latest = pickLatestCisPerProvider(["cis_5.0_aws"]);
expect(Array.from(latest)).toEqual(["cis_5.0_aws"]);
});
it("selects numerically, not lexicographically (1.10 beats 1.2)", () => {
const latest = pickLatestCisPerProvider([
"cis_1.2_kubernetes",
"cis_1.10_kubernetes",
]);
expect(Array.from(latest)).toEqual(["cis_1.10_kubernetes"]);
});
it("picks the highest major version across many variants", () => {
const latest = pickLatestCisPerProvider([
"cis_1.4_aws",
"cis_2.0_aws",
"cis_5.0_aws",
"cis_6.0_aws",
]);
expect(Array.from(latest)).toEqual(["cis_6.0_aws"]);
});
it("breaks ties on the minor version", () => {
const latest = pickLatestCisPerProvider([
"cis_3.0_aws",
"cis_3.1_aws",
"cis_2.9_aws",
]);
expect(Array.from(latest)).toEqual(["cis_3.1_aws"]);
});
it("considers three-part versions higher than two-part prefixes", () => {
const latest = pickLatestCisPerProvider(["cis_3.0_aws", "cis_3.0.1_aws"]);
expect(Array.from(latest)).toEqual(["cis_3.0.1_aws"]);
});
it("picks one latest per provider when multiple providers are mixed", () => {
const latest = pickLatestCisPerProvider([
"cis_1.4_aws",
"cis_5.0_aws",
"cis_2.0_azure",
"cis_5.0_azure",
"cis_1.12_kubernetes",
"cis_1.8_kubernetes",
]);
expect(new Set(latest)).toEqual(
new Set(["cis_5.0_aws", "cis_5.0_azure", "cis_1.12_kubernetes"]),
);
});
it("ignores non-CIS compliance ids mixed into the input", () => {
const latest = pickLatestCisPerProvider([
"ens_rd2022_aws",
"nis2_aws",
"csa_ccm_4.0_aws",
"prowler_threatscore_aws",
"cis_5.0_aws",
]);
expect(Array.from(latest)).toEqual(["cis_5.0_aws"]);
});
it("skips malformed names silently", () => {
const latest = pickLatestCisPerProvider([
"cis_abc_aws",
"cis_._aws",
"cis_5._aws",
"cis_5_aws",
"notcis_1.0_aws",
"cis_5.0_aws",
]);
expect(Array.from(latest)).toEqual(["cis_5.0_aws"]);
});
it("returns an empty set when every input is malformed", () => {
const latest = pickLatestCisPerProvider(["cis_abc_aws", "notcis_1.0_aws"]);
expect(latest.size).toBe(0);
});
});
describe("getReportTypeForCompliance", () => {
it("returns the framework-mapped type for single-version frameworks", () => {
expect(getReportTypeForCompliance("ENS", "ens_rd2022_aws", false)).toBe(
COMPLIANCE_REPORT_TYPES.ENS,
);
expect(getReportTypeForCompliance("NIS2", "nis2_aws", false)).toBe(
COMPLIANCE_REPORT_TYPES.NIS2,
);
});
it("returns CIS only when isLatestCisForProvider is true", () => {
expect(getReportTypeForCompliance("CIS", "cis_5.0_aws", true)).toBe(
COMPLIANCE_REPORT_TYPES.CIS,
);
});
it("hides CIS when isLatestCisForProvider is false (fail-closed default)", () => {
// An older CIS variant must NOT surface a PDF button because the
// backend only generates the PDF for the latest version.
expect(
getReportTypeForCompliance("CIS", "cis_1.4_aws", false),
).toBeUndefined();
});
it("defaults isLatestCisForProvider to false when omitted", () => {
expect(getReportTypeForCompliance("CIS", "cis_5.0_aws")).toBeUndefined();
});
it("returns undefined for unknown frameworks", () => {
expect(
getReportTypeForCompliance("SomethingElse", "unknown_1.0_foo", false),
).toBeUndefined();
});
});
+118 -6
View File
@@ -14,8 +14,8 @@ export const COMPLIANCE_REPORT_TYPES = {
ENS: "ens",
NIS2: "nis2",
CSA_CCM: "csa",
CIS: "cis",
// Future report types can be added here:
// CIS: "cis",
// NIST: "nist",
} as const;
@@ -36,6 +36,7 @@ export const COMPLIANCE_REPORT_DISPLAY_NAMES: Record<
[COMPLIANCE_REPORT_TYPES.ENS]: "ENS RD2022",
[COMPLIANCE_REPORT_TYPES.NIS2]: "NIS2",
[COMPLIANCE_REPORT_TYPES.CSA_CCM]: "CSA CCM",
[COMPLIANCE_REPORT_TYPES.CIS]: "CIS Benchmark",
// Add display names for future report types here
};
@@ -50,12 +51,21 @@ export const COMPLIANCE_REPORT_BUTTON_LABELS: Record<
[COMPLIANCE_REPORT_TYPES.ENS]: "PDF ENS Report",
[COMPLIANCE_REPORT_TYPES.NIS2]: "PDF NIS2 Report",
[COMPLIANCE_REPORT_TYPES.CSA_CCM]: "PDF CSA CCM Report",
[COMPLIANCE_REPORT_TYPES.CIS]: "PDF CIS Benchmark Report",
// Add button labels for future report types here
};
/**
* Maps compliance framework names (from API) to their report types
* This mapping determines which frameworks support PDF reporting
* Maps compliance framework names (from API) to their report types.
* This mapping determines which frameworks support PDF reporting.
*
* NOTE: CIS is intentionally NOT listed here. CIS has multiple variants per
* provider (e.g. cis_1.4_aws, cis_5.0_aws, cis_6.0_aws) and the backend only
* generates the PDF for the latest version per provider. The UI must gate
* the PDF button to the "latest" card only, otherwise every CIS version would
* expose a button pointing to the same latest-only artifact. The gating lives
* in `getReportTypeForCompliance` below, which callers should use instead of
* `getReportTypeForFramework` whenever CIS may be in play.
*/
const FRAMEWORK_TO_REPORT_TYPE: Record<string, ComplianceReportType> = {
ProwlerThreatScore: COMPLIANCE_REPORT_TYPES.THREATSCORE,
@@ -63,13 +73,16 @@ const FRAMEWORK_TO_REPORT_TYPE: Record<string, ComplianceReportType> = {
NIS2: COMPLIANCE_REPORT_TYPES.NIS2,
"CSA-CCM": COMPLIANCE_REPORT_TYPES.CSA_CCM,
// Add new framework mappings here as PDF support is added:
// "CIS-1.5": COMPLIANCE_REPORT_TYPES.CIS,
// "NIST-800-53": COMPLIANCE_REPORT_TYPES.NIST,
};
/**
* Helper function to get report type from framework name
* Returns undefined if framework doesn't support PDF reporting
* Helper function to get report type from framework name.
* Returns undefined if framework doesn't support PDF reporting.
*
* For CIS this will always return undefined — use
* `getReportTypeForCompliance` instead, which additionally checks whether
* the specific compliance_id is the latest variant for its provider.
*/
export const getReportTypeForFramework = (
framework: string | undefined,
@@ -77,3 +90,102 @@ export const getReportTypeForFramework = (
if (!framework) return undefined;
return FRAMEWORK_TO_REPORT_TYPE[framework];
};
/**
* Matches CIS compliance_ids like `cis_5.0_aws`, `cis_1.10_kubernetes`,
* `cis_3.0.1_aws`. Must stay in lock-step with the backend regex in
* `api/src/backend/tasks/jobs/report.py::_CIS_VARIANT_RE`.
*
* The version chunk requires at least one dotted component so malformed
* inputs like ``cis_5._aws``, ``cis_._aws`` or ``cis_5_aws`` are rejected
* at the regex stage instead of reaching the comparator with phantom zeros.
*
* Uses positional groups (not named) so the regex works under ES5 target.
* Group 1 = version string, group 2 = provider.
*/
const CIS_VARIANT_RE = /^cis_(\d+(?:\.\d+)+)_(.+)$/;
/**
* From an arbitrary set of compliance_ids (as returned by the compliance
* overview endpoint), return the subset of CIS variants that correspond to
* the highest semantic version per provider.
*
* Why: the backend only generates the CIS PDF for the latest version per
* provider (see `_pick_latest_cis_variant` in report.py). This helper
* mirrors that selection in the UI so only the "latest" CIS card exposes
* the PDF download button; older variants keep the CSV option.
*
* A lexicographic sort would be wrong (`cis_1.10_kubernetes` must beat
* `cis_1.2_kubernetes`), so the version chunk is parsed into a numeric
* tuple. Malformed names are silently skipped.
*
* Non-CIS compliance_ids are ignored — the returned Set only contains
* compliance_ids that satisfy `complianceId.startsWith("cis_")`.
*/
export const pickLatestCisPerProvider = (
complianceIds: readonly string[],
): Set<string> => {
const bestByProvider: Record<string, { key: number[]; id: string }> = {};
const compareVersions = (a: number[], b: number[]): number => {
const length = Math.max(a.length, b.length);
for (let i = 0; i < length; i++) {
// Use nullish coalescing — `||` would also collapse a legitimate 0.0
// into 0 (harmless here, but semantically wrong for a version chunk
// comparator). Missing trailing components are treated as `.0`.
const diff = (a[i] ?? 0) - (b[i] ?? 0);
if (diff !== 0) return diff;
}
return 0;
};
complianceIds.forEach((id) => {
const match = id.match(CIS_VARIANT_RE);
if (!match) return;
const version = match[1];
const provider = match[2];
const parts = version.split(".").map((chunk: string) => Number(chunk));
if (parts.some((n: number) => !Number.isFinite(n))) return;
const current = bestByProvider[provider];
if (!current || compareVersions(parts, current.key) > 0) {
bestByProvider[provider] = { key: parts, id };
}
});
const latest = new Set<string>();
Object.keys(bestByProvider).forEach((provider) => {
latest.add(bestByProvider[provider].id);
});
return latest;
};
/**
* Resolve the report type for a compliance card.
*
* Single-version frameworks (ENS/NIS2/CSA/ThreatScore) resolve via the
* framework-name mapping. CIS is gated by `isLatestCisForProvider`: only
* the card representing the highest CIS version per provider gets a PDF
* button, because the backend only generates a single PDF per provider
* (the latest variant). Callers must compute the "latest" set ahead of
* time via `pickLatestCisPerProvider` — this defaults to `false` so
* forgetting to pass it fails closed (no phantom PDF button on older CIS
* versions).
*/
export const getReportTypeForCompliance = (
framework: string | undefined,
complianceId: string | undefined,
isLatestCisForProvider: boolean = false,
): ComplianceReportType | undefined => {
const fromFramework = getReportTypeForFramework(framework);
if (fromFramework) return fromFramework;
if (
isLatestCisForProvider &&
typeof complianceId === "string" &&
complianceId.startsWith("cis_")
) {
return COMPLIANCE_REPORT_TYPES.CIS;
}
return undefined;
};
+23 -6
View File
@@ -2,6 +2,7 @@ import {
getComplianceCsv,
getCompliancePdfReport,
getExportsZip,
type ScanBinaryResult,
} from "@/actions/scans";
import { getTask } from "@/actions/task";
import { auth } from "@/auth.config";
@@ -149,12 +150,12 @@ export const downloadScanZip = async (
* Generic function to download a file from base64 data
*/
const downloadFile = async (
result: any,
result: ScanBinaryResult,
outputType: string,
successMessage: string,
toast: ReturnType<typeof useToast>["toast"],
): Promise<void> => {
if (result?.pending) {
if ("pending" in result && result.pending) {
toast({
title: "The report is still being generated",
description: "Please try again in a few minutes.",
@@ -162,7 +163,7 @@ const downloadFile = async (
return;
}
if (result?.success && result.data) {
if ("success" in result && result.success) {
try {
const binaryString = window.atob(result.data);
const bytes = new Uint8Array(binaryString.length);
@@ -194,7 +195,7 @@ const downloadFile = async (
return;
}
if (result?.error) {
if ("error" in result) {
toast({
variant: "destructive",
title: "Download Failed",
@@ -230,12 +231,18 @@ export const downloadComplianceCsv = async (
};
/**
* Generic function to download a compliance PDF report (ThreatScore, ENS, etc.)
* Download a compliance PDF report.
*
* The call hits ``/scans/{id}/{reportType}`` for every supported framework.
* For CIS — which ships multiple versions per provider — the backend only
* generates the PDF for the highest available version, so the call site
* does not need to resolve a specific variant.
*
* @param scanId - The scan ID
* @param reportType - Type of report (from COMPLIANCE_REPORT_TYPES)
* @param toast - Toast notification function
*/
export const downloadComplianceReportPdf = async (
export const downloadCompliancePdf = async (
scanId: string,
reportType: ComplianceReportType,
toast: ReturnType<typeof useToast>["toast"],
@@ -254,6 +261,16 @@ export const downloadComplianceReportPdf = async (
);
};
/**
* @deprecated Use {@link downloadCompliancePdf} instead. Kept as a thin
* wrapper for callers not yet migrated.
*/
export const downloadComplianceReportPdf = async (
scanId: string,
reportType: ComplianceReportType,
toast: ReturnType<typeof useToast>["toast"],
): Promise<void> => downloadCompliancePdf(scanId, reportType, toast);
export const isGoogleOAuthEnabled =
!!process.env.SOCIAL_GOOGLE_OAUTH_CLIENT_ID &&
!!process.env.SOCIAL_GOOGLE_OAUTH_CLIENT_SECRET;