feat(compliance): add new Prowler Threat Score Compliance Framework (#7603)

Co-authored-by: MrCloudSec <hello@mistercloudsec.com>

Author: pedrooot

dashboard/compliance/prowler_threatscore_aws.py

@@ -0,0 +1,24 @@
+import warnings
+
+from dashboard.common_methods import get_section_containers_cis
+
+warnings.filterwarnings("ignore")
+
+
+def get_table(data):
+    aux = data[
+        [
+            "REQUIREMENTS_ID",
+            "REQUIREMENTS_DESCRIPTION",
+            "REQUIREMENTS_ATTRIBUTES_SECTION",
+            "CHECKID",
+            "STATUS",
+            "REGION",
+            "ACCOUNTID",
+            "RESOURCEID",
+        ]
+    ].copy()
+
+    return get_section_containers_cis(
+        aux, "REQUIREMENTS_ID", "REQUIREMENTS_ATTRIBUTES_SECTION"
+    )

dashboard/compliance/prowler_threatscore_azure.py

@@ -0,0 +1,24 @@
+import warnings
+
+from dashboard.common_methods import get_section_containers_cis
+
+warnings.filterwarnings("ignore")
+
+
+def get_table(data):
+    aux = data[
+        [
+            "REQUIREMENTS_ID",
+            "REQUIREMENTS_DESCRIPTION",
+            "REQUIREMENTS_ATTRIBUTES_SECTION",
+            "CHECKID",
+            "STATUS",
+            "REGION",
+            "ACCOUNTID",
+            "RESOURCEID",
+        ]
+    ].copy()
+
+    return get_section_containers_cis(
+        aux, "REQUIREMENTS_ID", "REQUIREMENTS_ATTRIBUTES_SECTION"
+    )

dashboard/compliance/prowler_threatscore_gcp.py

@@ -0,0 +1,24 @@
+import warnings
+
+from dashboard.common_methods import get_section_containers_cis
+
+warnings.filterwarnings("ignore")
+
+
+def get_table(data):
+    aux = data[
+        [
+            "REQUIREMENTS_ID",
+            "REQUIREMENTS_DESCRIPTION",
+            "REQUIREMENTS_ATTRIBUTES_SECTION",
+            "CHECKID",
+            "STATUS",
+            "REGION",
+            "ACCOUNTID",
+            "RESOURCEID",
+        ]
+    ].copy()
+
+    return get_section_containers_cis(
+        aux, "REQUIREMENTS_ID", "REQUIREMENTS_ATTRIBUTES_SECTION"
+    )

dashboard/pages/compliance.py

@@ -398,6 +398,10 @@ def load_csv_files(files):
                 f"dashboard.compliance.{current}"
             )
             data.drop_duplicates(keep="first", inplace=True)
+
+            if "threatscore" in analytics_input:
+                data = get_threatscore_mean_by_pillar(data)
+
             table = compliance_module.get_table(data)
         except ModuleNotFoundError:
             table = html.Div(
@@ -430,6 +434,9 @@ def load_csv_files(files):
         if "pci" in analytics_input:
             pie_2 = get_bar_graph(df, "REQUIREMENTS_ID")
             current_filter = "req_id"
+        elif "threatscore" in analytics_input:
+            pie_2 = get_table_prowler_threatscore(df)
+            current_filter = "threatscore"
         elif (
             "REQUIREMENTS_ATTRIBUTES_SECTION" in df.columns
             and not df["REQUIREMENTS_ATTRIBUTES_SECTION"].isnull().values.any()
@@ -488,6 +495,13 @@ def load_csv_files(files):
         pie_2, f"Top 5 failed {current_filter} by requirements"
     )
 
+    if "threatscore" in analytics_input:
+        security_level_graph = get_graph(
+            pie_2,
+            "Pillar Score by requirements (1 = Lowest Risk, 5 = Highest Risk)",
+            margin_top=0,
+        )
+
     return (
         table_output,
         overall_status_result_graph,
@@ -501,7 +515,7 @@ def load_csv_files(files):
     )
 
 
-def get_graph(pie, title):
+def get_graph(pie, title, margin_top=7):
     return [
         html.Span(
             title,
@@ -514,7 +528,7 @@ def get_graph(pie, title):
                 "display": "flex",
                 "justify-content": "center",
                 "align-items": "center",
-                "margin-top": "7%",
+                "margin-top": f"{margin_top}%",
             },
         ),
     ]
@@ -618,3 +632,87 @@ def get_table(current_compliance, table):
             className="relative flex flex-col bg-white shadow-provider rounded-xl px-4 py-3 flex-wrap w-full",
         ),
     ]
+
+
+def get_threatscore_mean_by_pillar(df):
+    modified_df = df[df["STATUS"] == "FAIL"]
+
+    modified_df["REQUIREMENTS_ATTRIBUTES_LEVELOFRISK"] = pd.to_numeric(
+        modified_df["REQUIREMENTS_ATTRIBUTES_LEVELOFRISK"], errors="coerce"
+    )
+
+    pillar_means = (
+        modified_df.groupby("REQUIREMENTS_ATTRIBUTES_SECTION")[
+            "REQUIREMENTS_ATTRIBUTES_LEVELOFRISK"
+        ]
+        .mean()
+        .round(2)
+    )
+
+    output = []
+    for pillar, mean in pillar_means.items():
+        output.append(f"{pillar} - [{mean}]")
+
+    for value in output:
+        if value.split(" - ")[0] in df["REQUIREMENTS_ATTRIBUTES_SECTION"].values:
+            df.loc[
+                df["REQUIREMENTS_ATTRIBUTES_SECTION"] == value.split(" - ")[0],
+                "REQUIREMENTS_ATTRIBUTES_SECTION",
+            ] = value
+    return df
+
+
+def get_table_prowler_threatscore(df):
+    df = df[df["STATUS"] == "FAIL"]
+
+    # Delete " - " from the column REQUIREMENTS_ATTRIBUTES_SECTION
+    df["REQUIREMENTS_ATTRIBUTES_SECTION"] = (
+        df["REQUIREMENTS_ATTRIBUTES_SECTION"].str.split(" - ").str[0]
+    )
+
+    df["REQUIREMENTS_ATTRIBUTES_LEVELOFRISK"] = pd.to_numeric(
+        df["REQUIREMENTS_ATTRIBUTES_LEVELOFRISK"], errors="coerce"
+    )
+
+    score_df = (
+        df.groupby("REQUIREMENTS_ATTRIBUTES_SECTION")[
+            "REQUIREMENTS_ATTRIBUTES_LEVELOFRISK"
+        ]
+        .mean()
+        .reset_index()
+        .rename(
+            columns={
+                "REQUIREMENTS_ATTRIBUTES_SECTION": "Pillar",
+                "REQUIREMENTS_ATTRIBUTES_LEVELOFRISK": "Score",
+            }
+        )
+    )
+
+    fig = px.bar(
+        score_df,
+        x="Pillar",
+        y="Score",
+        color="Score",
+        color_continuous_scale=[
+            "#45cc6e",
+            "#f4d44d",
+            "#e77676",
+        ],  # verde → amarillo → rojo
+        hover_data={"Score": True, "Pillar": True},
+        labels={"Score": "Average Risk Score", "Pillar": "Section"},
+        height=400,
+    )
+
+    fig.update_layout(
+        xaxis_title="Pillar",
+        yaxis_title="Level of Risk",
+        margin=dict(l=20, r=20, t=30, b=20),
+        plot_bgcolor="rgba(0,0,0,0)",
+        paper_bgcolor="rgba(0,0,0,0)",
+        coloraxis_colorbar=dict(title="Risk"),
+    )
+
+    return dcc.Graph(
+        figure=fig,
+        style={"height": "25rem", "width": "40rem"},
+    )

prowler/CHANGELOG.md

@@ -35,6 +35,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - Add new check `teams_meeting_recording_disabled` [(#7607)](https://github.com/prowler-cloud/prowler/pull/7607)
 - Add new check `teams_meeting_presenters_restricted` [(#7613)](https://github.com/prowler-cloud/prowler/pull/7613)
 - Add new check `teams_meeting_chat_anonymous_users_disabled` [(#7579)](https://github.com/prowler-cloud/prowler/pull/7579)
+- Add Prowler Threat Score Compliance Framework [(#7603)](https://github.com/prowler-cloud/prowler/pull/7603)
 
 ### Fixed
 

prowler/__main__.py

@@ -70,6 +70,15 @@
     AzureMitreAttack,
 )
 from prowler.lib.outputs.compliance.mitre_attack.mitre_attack_gcp import GCPMitreAttack
+from prowler.lib.outputs.compliance.prowler_threatscore.prowler_threatscore_aws import (
+    ProwlerThreatScoreAWS,
+)
+from prowler.lib.outputs.compliance.prowler_threatscore.prowler_threatscore_azure import (
+    ProwlerThreatScoreAzure,
+)
+from prowler.lib.outputs.compliance.prowler_threatscore.prowler_threatscore_gcp import (
+    ProwlerThreatScoreGCP,
+)
 from prowler.lib.outputs.csv.csv import CSV
 from prowler.lib.outputs.finding import Finding
 from prowler.lib.outputs.html.html import HTML
@@ -478,6 +487,18 @@ def prowler():
                 )
                 generated_outputs["compliance"].append(kisa_ismsp)
                 kisa_ismsp.batch_write_data_to_file()
+            elif compliance_name == "prowler_threatscore_aws":
+                filename = (
+                    f"{output_options.output_directory}/compliance/"
+                    f"{output_options.output_filename}_{compliance_name}.csv"
+                )
+                prowler_threatscore = ProwlerThreatScoreAWS(
+                    findings=finding_outputs,
+                    compliance=bulk_compliance_frameworks[compliance_name],
+                    file_path=filename,
+                )
+                generated_outputs["compliance"].append(prowler_threatscore)
+                prowler_threatscore.batch_write_data_to_file()
             else:
                 filename = (
                     f"{output_options.output_directory}/compliance/"
@@ -545,6 +566,18 @@ def prowler():
                 )
                 generated_outputs["compliance"].append(iso27001)
                 iso27001.batch_write_data_to_file()
+            elif compliance_name == "prowler_threatscore_azure":
+                filename = (
+                    f"{output_options.output_directory}/compliance/"
+                    f"{output_options.output_filename}_{compliance_name}.csv"
+                )
+                prowler_threatscore = ProwlerThreatScoreAzure(
+                    findings=finding_outputs,
+                    compliance=bulk_compliance_frameworks[compliance_name],
+                    file_path=filename,
+                )
+                generated_outputs["compliance"].append(prowler_threatscore)
+                prowler_threatscore.batch_write_data_to_file()
             else:
                 filename = (
                     f"{output_options.output_directory}/compliance/"
@@ -612,6 +645,18 @@ def prowler():
                 )
                 generated_outputs["compliance"].append(iso27001)
                 iso27001.batch_write_data_to_file()
+            elif compliance_name == "prowler_threatscore_gcp":
+                filename = (
+                    f"{output_options.output_directory}/compliance/"
+                    f"{output_options.output_filename}_{compliance_name}.csv"
+                )
+                prowler_threatscore = ProwlerThreatScoreGCP(
+                    findings=finding_outputs,
+                    compliance=bulk_compliance_frameworks[compliance_name],
+                    file_path=filename,
+                )
+                generated_outputs["compliance"].append(prowler_threatscore)
+                prowler_threatscore.batch_write_data_to_file()
             else:
                 filename = (
                     f"{output_options.output_directory}/compliance/"