mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 12:31:54 +00:00
StylusFrost
1240 lines
42 KiB
Python
1240 lines
42 KiB
Python
import json
|
|
import os
|
|
import tempfile
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from pydantic.v1 import ValidationError
|
|
|
|
from prowler.lib.check.compliance_models import (
|
|
AttributeMetadata,
|
|
ChartConfig,
|
|
Compliance,
|
|
ComplianceFramework,
|
|
CriticalRequirementsFilter,
|
|
EnumValueDisplay,
|
|
I18nLabels,
|
|
OutputFormats,
|
|
OutputsConfig,
|
|
PDFConfig,
|
|
ReportFilter,
|
|
ScoringConfig,
|
|
ScoringFormula,
|
|
SplitByConfig,
|
|
TableConfig,
|
|
TableLabels,
|
|
UniversalComplianceRequirement,
|
|
adapt_legacy_to_universal,
|
|
get_bulk_compliance_frameworks_universal,
|
|
load_compliance_framework_universal,
|
|
)
|
|
from tests.lib.outputs.compliance.fixtures import (
|
|
CIS_1_4_AWS,
|
|
ENS_RD2022_AWS,
|
|
KISA_ISMSP_AWS,
|
|
MITRE_ATTACK_AWS,
|
|
NIST_800_53_REVISION_4_AWS,
|
|
PROWLER_THREATSCORE_AWS,
|
|
)
|
|
|
|
|
|
class TestOutputFormats:
|
|
def test_defaults(self):
|
|
of = OutputFormats()
|
|
assert of.csv is True
|
|
assert of.ocsf is True
|
|
|
|
def test_explicit_false(self):
|
|
of = OutputFormats(csv=False, ocsf=False)
|
|
assert of.csv is False
|
|
assert of.ocsf is False
|
|
|
|
|
|
class TestAttributeMetadata:
|
|
def test_basic(self):
|
|
meta = AttributeMetadata(key="Section", type="str")
|
|
assert meta.key == "Section"
|
|
assert meta.type == "str"
|
|
assert meta.output_formats.csv is True
|
|
assert meta.required is False
|
|
|
|
def test_with_enum(self):
|
|
meta = AttributeMetadata(
|
|
key="Profile",
|
|
type="str",
|
|
enum=["Level 1", "Level 2"],
|
|
)
|
|
assert meta.enum == ["Level 1", "Level 2"]
|
|
|
|
def test_int_type(self):
|
|
meta = AttributeMetadata(key="LevelOfRisk", type="int", required=True)
|
|
assert meta.type == "int"
|
|
assert meta.required is True
|
|
|
|
def test_enum_display_field(self):
|
|
meta = AttributeMetadata(
|
|
key="Dimensiones",
|
|
type="str",
|
|
enum=["confidencialidad", "integridad", "trazabilidad"],
|
|
enum_display={
|
|
"confidencialidad": {
|
|
"label": "Confidencialidad",
|
|
"abbreviation": "C",
|
|
"color": "#FF6347",
|
|
},
|
|
"integridad": {
|
|
"label": "Integridad",
|
|
"abbreviation": "I",
|
|
"color": "#4286F4",
|
|
},
|
|
"trazabilidad": {
|
|
"label": "Trazabilidad",
|
|
"abbreviation": "T",
|
|
"color": "#32CD32",
|
|
},
|
|
},
|
|
)
|
|
assert meta.enum_display is not None
|
|
assert meta.enum_display["confidencialidad"]["abbreviation"] == "C"
|
|
assert meta.enum_display["integridad"]["color"] == "#4286F4"
|
|
|
|
def test_enum_order_field(self):
|
|
meta = AttributeMetadata(
|
|
key="Nivel",
|
|
type="str",
|
|
enum=["opcional", "bajo", "medio", "alto"],
|
|
enum_order=["alto", "medio", "bajo", "opcional"],
|
|
)
|
|
assert meta.enum_order == ["alto", "medio", "bajo", "opcional"]
|
|
|
|
def test_chart_label_field(self):
|
|
meta = AttributeMetadata(
|
|
key="Section",
|
|
type="str",
|
|
chart_label="Security Domain",
|
|
)
|
|
assert meta.chart_label == "Security Domain"
|
|
|
|
def test_output_formats_default_true(self):
|
|
meta = AttributeMetadata(key="Section")
|
|
assert meta.output_formats.csv is True
|
|
assert meta.output_formats.ocsf is True
|
|
|
|
def test_output_formats_explicit_false(self):
|
|
meta = AttributeMetadata(
|
|
key="InternalNote",
|
|
output_formats=OutputFormats(csv=False, ocsf=False),
|
|
)
|
|
assert meta.output_formats.csv is False
|
|
assert meta.output_formats.ocsf is False
|
|
|
|
def test_new_fields_default_none(self):
|
|
meta = AttributeMetadata(key="Section")
|
|
assert meta.enum_display is None
|
|
assert meta.enum_order is None
|
|
assert meta.chart_label is None
|
|
|
|
|
|
class TestEnumValueDisplay:
|
|
def test_basic(self):
|
|
evd = EnumValueDisplay(label="Test")
|
|
assert evd.label == "Test"
|
|
assert evd.abbreviation is None
|
|
assert evd.color is None
|
|
assert evd.icon is None
|
|
|
|
def test_dimension_style(self):
|
|
evd = EnumValueDisplay(
|
|
label="Trazabilidad",
|
|
abbreviation="T",
|
|
color="#4286F4",
|
|
)
|
|
assert evd.label == "Trazabilidad"
|
|
assert evd.abbreviation == "T"
|
|
assert evd.color == "#4286F4"
|
|
|
|
def test_tipo_style(self):
|
|
evd = EnumValueDisplay(
|
|
label="Requisito",
|
|
icon="⚠️",
|
|
)
|
|
assert evd.icon == "⚠️"
|
|
assert evd.abbreviation is None
|
|
|
|
|
|
class TestChartConfig:
|
|
def test_horizontal_bar(self):
|
|
chart = ChartConfig(
|
|
id="section_compliance",
|
|
type="horizontal_bar",
|
|
group_by="Section",
|
|
title="Compliance Score by Domain",
|
|
y_label="Domain",
|
|
x_label="Compliance %",
|
|
)
|
|
assert chart.type == "horizontal_bar"
|
|
assert chart.group_by == "Section"
|
|
assert chart.value_source == "compliance_percent"
|
|
assert chart.color_mode == "by_value"
|
|
|
|
def test_vertical_bar(self):
|
|
chart = ChartConfig(
|
|
id="risk_distribution",
|
|
type="vertical_bar",
|
|
group_by="LevelOfRisk",
|
|
color_mode="fixed",
|
|
fixed_color="#336699",
|
|
)
|
|
assert chart.type == "vertical_bar"
|
|
assert chart.fixed_color == "#336699"
|
|
|
|
def test_radar(self):
|
|
chart = ChartConfig(
|
|
id="dimension_radar",
|
|
type="radar",
|
|
group_by="Dimensiones",
|
|
)
|
|
assert chart.type == "radar"
|
|
|
|
def test_defaults(self):
|
|
chart = ChartConfig(id="test", type="vertical_bar", group_by="Section")
|
|
assert chart.title is None
|
|
assert chart.x_label is None
|
|
assert chart.y_label is None
|
|
assert chart.value_source == "compliance_percent"
|
|
assert chart.color_mode == "by_value"
|
|
assert chart.fixed_color is None
|
|
|
|
|
|
class TestScoringFormula:
|
|
def test_threatscore_style(self):
|
|
formula = ScoringFormula(
|
|
risk_field="LevelOfRisk",
|
|
weight_field="Weight",
|
|
risk_boost_factor=0.25,
|
|
)
|
|
assert formula.risk_field == "LevelOfRisk"
|
|
assert formula.weight_field == "Weight"
|
|
assert formula.risk_boost_factor == 0.25
|
|
|
|
def test_custom_boost_factor(self):
|
|
formula = ScoringFormula(
|
|
risk_field="Risk",
|
|
weight_field="Impact",
|
|
risk_boost_factor=0.5,
|
|
)
|
|
assert formula.risk_boost_factor == 0.5
|
|
|
|
def test_default_boost_factor(self):
|
|
formula = ScoringFormula(risk_field="LevelOfRisk", weight_field="Weight")
|
|
assert formula.risk_boost_factor == 0.25
|
|
|
|
|
|
class TestCriticalRequirementsFilter:
|
|
def test_int_based(self):
|
|
crf = CriticalRequirementsFilter(
|
|
filter_field="LevelOfRisk",
|
|
min_value=4,
|
|
title="Critical Failed Requirements",
|
|
)
|
|
assert crf.filter_field == "LevelOfRisk"
|
|
assert crf.min_value == 4
|
|
assert crf.filter_value is None
|
|
assert crf.status_filter == "FAIL"
|
|
assert crf.title == "Critical Failed Requirements"
|
|
|
|
def test_string_based(self):
|
|
crf = CriticalRequirementsFilter(
|
|
filter_field="Nivel",
|
|
filter_value="alto",
|
|
)
|
|
assert crf.filter_value == "alto"
|
|
assert crf.min_value is None
|
|
|
|
def test_defaults(self):
|
|
crf = CriticalRequirementsFilter(filter_field="LevelOfRisk")
|
|
assert crf.status_filter == "FAIL"
|
|
assert crf.title is None
|
|
assert crf.min_value is None
|
|
assert crf.filter_value is None
|
|
|
|
|
|
class TestReportFilter:
|
|
def test_defaults(self):
|
|
rf = ReportFilter()
|
|
assert rf.only_failed is True
|
|
assert rf.include_manual is False
|
|
|
|
def test_custom(self):
|
|
rf = ReportFilter(only_failed=False, include_manual=True)
|
|
assert rf.only_failed is False
|
|
assert rf.include_manual is True
|
|
|
|
|
|
class TestI18nLabels:
|
|
def test_english_defaults(self):
|
|
labels = I18nLabels()
|
|
assert labels.page_label == "Page"
|
|
assert labels.powered_by == "Powered by Prowler"
|
|
assert labels.framework_label == "Framework:"
|
|
assert labels.provider_label == "Provider:"
|
|
assert labels.report_title is None
|
|
|
|
def test_spanish_override(self):
|
|
labels = I18nLabels(
|
|
report_title="Informe de Cumplimiento ENS",
|
|
page_label="Página",
|
|
powered_by="Generado por Prowler",
|
|
framework_label="Marco:",
|
|
version_label="Versión:",
|
|
provider_label="Proveedor:",
|
|
description_label="Descripción:",
|
|
compliance_score_label="Puntuación de Cumplimiento por Secciones",
|
|
requirements_index_label="Índice de Requisitos",
|
|
detailed_findings_label="Hallazgos Detallados",
|
|
)
|
|
assert labels.page_label == "Página"
|
|
assert labels.provider_label == "Proveedor:"
|
|
assert labels.report_title == "Informe de Cumplimiento ENS"
|
|
|
|
|
|
class TestSplitByConfig:
|
|
def test_cis_style(self):
|
|
config = SplitByConfig(field="Profile", values=["Level 1", "Level 2"])
|
|
assert config.field == "Profile"
|
|
assert len(config.values) == 2
|
|
|
|
def test_ens_style(self):
|
|
config = SplitByConfig(
|
|
field="Nivel",
|
|
values=["alto", "medio", "bajo", "opcional"],
|
|
)
|
|
assert len(config.values) == 4
|
|
|
|
|
|
class TestScoringConfig:
|
|
def test_threatscore_style(self):
|
|
config = ScoringConfig(risk_field="LevelOfRisk", weight_field="Weight")
|
|
assert config.risk_field == "LevelOfRisk"
|
|
assert config.weight_field == "Weight"
|
|
|
|
|
|
class TestTableLabels:
|
|
def test_defaults(self):
|
|
labels = TableLabels()
|
|
assert labels.pass_label == "PASS"
|
|
assert labels.fail_label == "FAIL"
|
|
assert labels.provider_header == "Provider"
|
|
|
|
def test_ens_spanish(self):
|
|
labels = TableLabels(
|
|
pass_label="CUMPLE",
|
|
fail_label="NO CUMPLE",
|
|
provider_header="Proveedor",
|
|
)
|
|
assert labels.pass_label == "CUMPLE"
|
|
|
|
|
|
class TestTableConfig:
|
|
def test_grouped_mode(self):
|
|
tc = TableConfig(group_by="Section")
|
|
assert tc.group_by == "Section"
|
|
assert tc.split_by is None
|
|
assert tc.scoring is None
|
|
|
|
def test_split_mode(self):
|
|
tc = TableConfig(
|
|
group_by="Section",
|
|
split_by=SplitByConfig(field="Profile", values=["Level 1", "Level 2"]),
|
|
)
|
|
assert tc.split_by is not None
|
|
assert tc.split_by.field == "Profile"
|
|
|
|
def test_scored_mode(self):
|
|
tc = TableConfig(
|
|
group_by="Section",
|
|
scoring=ScoringConfig(risk_field="LevelOfRisk", weight_field="Weight"),
|
|
)
|
|
assert tc.scoring is not None
|
|
|
|
|
|
class TestPDFConfig:
|
|
def test_defaults(self):
|
|
pdf = PDFConfig()
|
|
assert pdf.language == "en"
|
|
assert pdf.logo_filename is None
|
|
assert pdf.primary_color is None
|
|
assert pdf.sections is None
|
|
assert pdf.section_short_names is None
|
|
assert pdf.group_by_field is None
|
|
assert pdf.sub_group_by_field is None
|
|
assert pdf.section_titles is None
|
|
assert pdf.charts is None
|
|
assert pdf.scoring is None
|
|
assert pdf.critical_filter is None
|
|
assert pdf.filter is None
|
|
assert pdf.labels is None
|
|
|
|
def test_csa_ccm_style(self):
|
|
pdf = PDFConfig(
|
|
primary_color="#336699",
|
|
secondary_color="#4D80B3",
|
|
bg_color="#F2F8FF",
|
|
group_by_field="Section",
|
|
sections=["Audit & Assurance", "Identity & Access Management"],
|
|
section_short_names={"Identity & Access Management": "IAM"},
|
|
charts=[
|
|
ChartConfig(
|
|
id="section_compliance",
|
|
type="horizontal_bar",
|
|
group_by="Section",
|
|
title="Compliance Score by Domain",
|
|
).dict()
|
|
],
|
|
filter=ReportFilter(only_failed=True, include_manual=False),
|
|
)
|
|
assert pdf.primary_color == "#336699"
|
|
assert len(pdf.sections) == 2
|
|
assert pdf.section_short_names["Identity & Access Management"] == "IAM"
|
|
assert pdf.group_by_field == "Section"
|
|
assert pdf.charts is not None
|
|
assert len(pdf.charts) == 1
|
|
assert pdf.filter.only_failed is True
|
|
|
|
def test_ens_style(self):
|
|
pdf = PDFConfig(
|
|
language="es",
|
|
logo_filename="ens_logo.png",
|
|
primary_color="#CC3333",
|
|
group_by_field="Marco",
|
|
sub_group_by_field="Categoria",
|
|
labels=I18nLabels(
|
|
page_label="Página",
|
|
provider_label="Proveedor:",
|
|
),
|
|
)
|
|
assert pdf.language == "es"
|
|
assert pdf.logo_filename == "ens_logo.png"
|
|
assert pdf.group_by_field == "Marco"
|
|
assert pdf.sub_group_by_field == "Categoria"
|
|
assert pdf.labels.page_label == "Página"
|
|
|
|
def test_threatscore_style(self):
|
|
pdf = PDFConfig(
|
|
primary_color="#336699",
|
|
sections=["1. IAM", "2. Attack Surface"],
|
|
scoring=ScoringFormula(
|
|
risk_field="LevelOfRisk",
|
|
weight_field="Weight",
|
|
risk_boost_factor=0.25,
|
|
),
|
|
critical_filter=CriticalRequirementsFilter(
|
|
filter_field="LevelOfRisk",
|
|
min_value=4,
|
|
title="Critical Failed Requirements",
|
|
),
|
|
)
|
|
assert pdf.scoring is not None
|
|
assert pdf.scoring.risk_field == "LevelOfRisk"
|
|
assert pdf.critical_filter.min_value == 4
|
|
|
|
def test_section_titles(self):
|
|
pdf = PDFConfig(
|
|
section_titles={
|
|
"1": "1. Policy on Security",
|
|
"2": "2. Risk Management",
|
|
},
|
|
)
|
|
assert pdf.section_titles["1"] == "1. Policy on Security"
|
|
|
|
def test_in_framework(self):
|
|
fw = ComplianceFramework(
|
|
framework="Test",
|
|
name="Test Framework",
|
|
description="Test",
|
|
requirements=[],
|
|
outputs=OutputsConfig(
|
|
pdf_config=PDFConfig(
|
|
primary_color="#336699",
|
|
sections=["Section A"],
|
|
charts=[
|
|
ChartConfig(
|
|
id="test_chart",
|
|
type="vertical_bar",
|
|
group_by="Section",
|
|
).dict()
|
|
],
|
|
),
|
|
),
|
|
)
|
|
assert fw.outputs is not None
|
|
assert fw.outputs.pdf_config is not None
|
|
assert fw.outputs.pdf_config.primary_color == "#336699"
|
|
assert fw.outputs.pdf_config.sections == ["Section A"]
|
|
assert fw.outputs.pdf_config.charts is not None
|
|
assert len(fw.outputs.pdf_config.charts) == 1
|
|
assert fw.outputs.pdf_config.charts[0]["id"] == "test_chart"
|
|
assert fw.outputs.pdf_config.charts[0]["type"] == "vertical_bar"
|
|
|
|
def test_framework_without_pdf_config(self):
|
|
fw = ComplianceFramework(
|
|
framework="Test",
|
|
name="Test Framework",
|
|
description="Test",
|
|
requirements=[],
|
|
)
|
|
assert fw.outputs is None
|
|
|
|
|
|
class TestUniversalComplianceRequirement:
|
|
def test_flat_dict_attributes(self):
|
|
req = UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="Test requirement",
|
|
attributes={"Section": "IAM", "Profile": "Level 1"},
|
|
checks={"aws": ["check_a", "check_b"]},
|
|
)
|
|
assert req.attributes["Section"] == "IAM"
|
|
assert len(req.checks["aws"]) == 2
|
|
|
|
def test_mitre_optional_fields(self):
|
|
req = UniversalComplianceRequirement(
|
|
id="T1190",
|
|
description="Exploit Public-Facing Application",
|
|
attributes={},
|
|
checks={"aws": ["drs_job_exist"]},
|
|
tactics=["Initial Access"],
|
|
sub_techniques=[],
|
|
platforms=["IaaS", "Linux"],
|
|
technique_url="https://attack.mitre.org/techniques/T1190/",
|
|
)
|
|
assert req.tactics == ["Initial Access"]
|
|
assert req.technique_url == "https://attack.mitre.org/techniques/T1190/"
|
|
|
|
def test_dict_checks_multi_provider(self):
|
|
req = UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="Multi-provider",
|
|
attributes={},
|
|
checks={"aws": ["check_a"], "azure": ["check_b"]},
|
|
)
|
|
assert isinstance(req.checks, dict)
|
|
assert "aws" in req.checks
|
|
|
|
def test_empty_checks(self):
|
|
req = UniversalComplianceRequirement(
|
|
id="manual-1",
|
|
description="Manual requirement",
|
|
attributes={"Section": "Governance"},
|
|
checks={},
|
|
)
|
|
assert req.checks == {}
|
|
|
|
def test_checks_default_is_empty_dict(self):
|
|
req = UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="No checks provided",
|
|
)
|
|
assert req.checks == {}
|
|
|
|
|
|
class TestComplianceFramework:
|
|
def test_basic_framework(self):
|
|
fw = ComplianceFramework(
|
|
framework="TestFW",
|
|
name="Test Framework",
|
|
provider="AWS",
|
|
version="1.0",
|
|
description="A test framework",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="Test",
|
|
attributes={"Section": "IAM"},
|
|
checks={"aws": ["check_a"]},
|
|
)
|
|
],
|
|
attributes_metadata=[
|
|
AttributeMetadata(key="Section", type="str"),
|
|
],
|
|
outputs=OutputsConfig(table_config=TableConfig(group_by="Section")),
|
|
)
|
|
assert fw.framework == "TestFW"
|
|
assert fw.outputs.table_config.group_by == "Section"
|
|
assert len(fw.attributes_metadata) == 1
|
|
assert len(fw.requirements) == 1
|
|
|
|
def test_optional_provider(self):
|
|
fw = ComplianceFramework(
|
|
framework="MultiCloud",
|
|
name="Multi-cloud framework",
|
|
description="A multi-provider framework",
|
|
requirements=[],
|
|
)
|
|
assert fw.provider is None
|
|
|
|
def test_get_providers_from_dict_checks(self):
|
|
fw = ComplianceFramework(
|
|
framework="MultiCloud",
|
|
name="Multi-cloud",
|
|
description="test",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="test",
|
|
attributes={},
|
|
checks={
|
|
"aws": ["check_a"],
|
|
"azure": ["check_b"],
|
|
"gcp": ["check_c"],
|
|
},
|
|
),
|
|
UniversalComplianceRequirement(
|
|
id="1.2",
|
|
description="test2",
|
|
attributes={},
|
|
checks={"aws": ["check_d"]},
|
|
),
|
|
],
|
|
)
|
|
providers = fw.get_providers()
|
|
assert providers == ["aws", "azure", "gcp"]
|
|
|
|
def test_get_providers_fallback_to_explicit(self):
|
|
fw = ComplianceFramework(
|
|
framework="SingleCloud",
|
|
name="Single-cloud",
|
|
provider="AWS",
|
|
description="test",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="test",
|
|
attributes={},
|
|
checks={},
|
|
),
|
|
],
|
|
)
|
|
providers = fw.get_providers()
|
|
assert providers == ["aws"]
|
|
|
|
def test_supports_provider_dict_checks(self):
|
|
fw = ComplianceFramework(
|
|
framework="MultiCloud",
|
|
name="Multi-cloud",
|
|
description="test",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="test",
|
|
attributes={},
|
|
checks={"aws": ["check_a"], "azure": ["check_b"]},
|
|
),
|
|
],
|
|
)
|
|
assert fw.supports_provider("aws") is True
|
|
assert fw.supports_provider("azure") is True
|
|
assert fw.supports_provider("gcp") is False
|
|
|
|
def test_supports_provider_explicit_only(self):
|
|
"""Framework with explicit provider but no per-requirement checks still supports the provider."""
|
|
fw = ComplianceFramework(
|
|
framework="SingleCloud",
|
|
name="Single-cloud",
|
|
provider="AWS",
|
|
description="test",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="Manual requirement",
|
|
attributes={},
|
|
checks={},
|
|
),
|
|
],
|
|
)
|
|
assert fw.supports_provider("aws") is True
|
|
assert fw.supports_provider("azure") is False
|
|
|
|
def test_no_provider_field_with_dict_checks(self):
|
|
"""Multi-provider JSON has no Provider field — providers derived from checks."""
|
|
fw = ComplianceFramework(
|
|
framework="CSA_CCM",
|
|
name="CSA CCM 4.0",
|
|
description="Cloud Controls Matrix",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="A&A-01",
|
|
description="Audit & Assurance",
|
|
attributes={"Domain": "A&A"},
|
|
checks={
|
|
"aws": ["check_a"],
|
|
"azure": ["check_b"],
|
|
"gcp": ["check_c"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
assert fw.provider is None
|
|
assert fw.get_providers() == ["aws", "azure", "gcp"]
|
|
assert fw.supports_provider("aws")
|
|
assert fw.supports_provider("azure")
|
|
assert fw.supports_provider("gcp")
|
|
assert not fw.supports_provider("kubernetes")
|
|
|
|
def test_icon_field(self):
|
|
fw = ComplianceFramework(
|
|
framework="CSA_CCM",
|
|
name="CSA CCM 4.0",
|
|
description="Cloud Controls Matrix",
|
|
icon="csa",
|
|
requirements=[],
|
|
)
|
|
assert fw.icon == "csa"
|
|
|
|
def test_icon_defaults_to_none(self):
|
|
fw = ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[],
|
|
)
|
|
assert fw.icon is None
|
|
|
|
|
|
class TestAdaptLegacyToUniversal:
|
|
def test_adapt_cis(self):
|
|
fw = adapt_legacy_to_universal(CIS_1_4_AWS)
|
|
assert fw.framework == "CIS"
|
|
assert fw.provider == "AWS"
|
|
assert len(fw.requirements) == 2
|
|
# First requirement should have flat attributes
|
|
req = fw.requirements[0]
|
|
assert "Section" in req.attributes
|
|
assert req.attributes["Section"] == "2. Storage"
|
|
assert req.tactics is None
|
|
# Checks must be wrapped in dict keyed by provider
|
|
assert isinstance(req.checks, dict)
|
|
assert "aws" in req.checks
|
|
|
|
def test_adapt_ens(self):
|
|
fw = adapt_legacy_to_universal(ENS_RD2022_AWS)
|
|
assert fw.framework == "ENS"
|
|
req = fw.requirements[0]
|
|
assert "Marco" in req.attributes
|
|
assert req.attributes["Marco"] == "operacional"
|
|
|
|
def test_adapt_mitre(self):
|
|
fw = adapt_legacy_to_universal(MITRE_ATTACK_AWS)
|
|
assert fw.framework == "MITRE-ATTACK"
|
|
req = fw.requirements[0]
|
|
assert req.tactics == ["Initial Access"]
|
|
assert req.technique_url == "https://attack.mitre.org/techniques/T1190/"
|
|
assert "_raw_attributes" in req.attributes
|
|
assert isinstance(req.checks, dict)
|
|
assert "aws" in req.checks
|
|
|
|
def test_adapt_threatscore(self):
|
|
fw = adapt_legacy_to_universal(PROWLER_THREATSCORE_AWS)
|
|
req = fw.requirements[0]
|
|
assert req.attributes["LevelOfRisk"] == 5
|
|
assert req.attributes["Weight"] == 1000
|
|
|
|
def test_adapt_generic(self):
|
|
fw = adapt_legacy_to_universal(NIST_800_53_REVISION_4_AWS)
|
|
req = fw.requirements[0]
|
|
assert "Section" in req.attributes
|
|
|
|
def test_adapt_kisa(self):
|
|
fw = adapt_legacy_to_universal(KISA_ISMSP_AWS)
|
|
req = fw.requirements[0]
|
|
assert "Domain" in req.attributes
|
|
|
|
def test_inferred_metadata_cis(self):
|
|
fw = adapt_legacy_to_universal(CIS_1_4_AWS)
|
|
assert fw.attributes_metadata is not None
|
|
keys = [m.key for m in fw.attributes_metadata]
|
|
assert "Section" in keys
|
|
assert "Profile" in keys
|
|
|
|
def test_inferred_metadata_mitre_is_none(self):
|
|
fw = adapt_legacy_to_universal(MITRE_ATTACK_AWS)
|
|
assert fw.attributes_metadata is None
|
|
|
|
def test_table_config_is_none(self):
|
|
fw = adapt_legacy_to_universal(CIS_1_4_AWS)
|
|
assert fw.outputs is None
|
|
|
|
|
|
class TestLoadComplianceFrameworkUniversal:
|
|
def test_load_universal_format(self, tmp_path):
|
|
data = {
|
|
"framework": "TestFW",
|
|
"name": "Test",
|
|
"provider": "AWS",
|
|
"version": "1.0",
|
|
"description": "desc",
|
|
"icon": "prowlerthreatscore",
|
|
"attributes_metadata": [{"key": "Section", "type": "str"}],
|
|
"outputs": {"table_config": {"group_by": "Section"}},
|
|
"requirements": [
|
|
{
|
|
"id": "1.1",
|
|
"description": "test",
|
|
"attributes": {"Section": "IAM"},
|
|
"checks": {"aws": ["check_a"]},
|
|
}
|
|
],
|
|
}
|
|
path = tmp_path / "test.json"
|
|
path.write_text(json.dumps(data))
|
|
fw = load_compliance_framework_universal(str(path))
|
|
assert fw is not None
|
|
assert fw.framework == "TestFW"
|
|
assert fw.icon == "prowlerthreatscore"
|
|
assert fw.outputs.table_config.group_by == "Section"
|
|
|
|
def test_load_universal_multi_provider(self, tmp_path):
|
|
data = {
|
|
"framework": "CSA_CCM",
|
|
"name": "CSA CCM 4.0",
|
|
"version": "4.0",
|
|
"description": "Cloud Controls Matrix",
|
|
"attributes_metadata": [{"key": "Domain", "type": "str"}],
|
|
"outputs": {"table_config": {"group_by": "Domain"}},
|
|
"requirements": [
|
|
{
|
|
"id": "A&A-01",
|
|
"description": "Audit",
|
|
"attributes": {"Domain": "Audit"},
|
|
"checks": {
|
|
"aws": ["check_a"],
|
|
"azure": ["check_b"],
|
|
"gcp": ["check_c"],
|
|
},
|
|
}
|
|
],
|
|
}
|
|
path = tmp_path / "csa_ccm_4.0.json"
|
|
path.write_text(json.dumps(data))
|
|
fw = load_compliance_framework_universal(str(path))
|
|
assert fw is not None
|
|
assert fw.provider is None
|
|
assert fw.get_providers() == ["aws", "azure", "gcp"]
|
|
assert fw.supports_provider("aws")
|
|
assert not fw.supports_provider("kubernetes")
|
|
|
|
def test_load_legacy_format(self, tmp_path):
|
|
data = {
|
|
"Framework": "SOC2",
|
|
"Name": "SOC2",
|
|
"Provider": "AWS",
|
|
"Version": "",
|
|
"Description": "desc",
|
|
"Requirements": [
|
|
{
|
|
"Id": "1.1",
|
|
"Description": "test",
|
|
"Attributes": [{"Section": "Access Control"}],
|
|
"Checks": ["check_a"],
|
|
}
|
|
],
|
|
}
|
|
path = tmp_path / "legacy.json"
|
|
path.write_text(json.dumps(data))
|
|
fw = load_compliance_framework_universal(str(path))
|
|
assert fw is not None
|
|
assert fw.framework == "SOC2"
|
|
assert fw.outputs is None
|
|
assert fw.requirements[0].attributes["Section"] == "Access Control"
|
|
assert fw.requirements[0].checks == {"aws": ["check_a"]}
|
|
|
|
|
|
class TestSmokeLoadAllJSONs:
|
|
"""Parametrized smoke test: every existing compliance JSON must load as ComplianceFramework."""
|
|
|
|
@staticmethod
|
|
def _find_all_compliance_jsons():
|
|
base = os.path.join(
|
|
os.path.dirname(__file__),
|
|
"..",
|
|
"..",
|
|
"..",
|
|
"prowler",
|
|
"compliance",
|
|
)
|
|
base = os.path.normpath(base)
|
|
jsons = []
|
|
if os.path.isdir(base):
|
|
# Top-level JSONs (multi-provider)
|
|
for filename in os.listdir(base):
|
|
if filename.endswith(".json"):
|
|
jsons.append(os.path.join(base, filename))
|
|
# Provider sub-directory JSONs
|
|
for provider_dir in os.listdir(base):
|
|
provider_path = os.path.join(base, provider_dir)
|
|
if os.path.isdir(provider_path):
|
|
for filename in os.listdir(provider_path):
|
|
if filename.endswith(".json"):
|
|
jsons.append(os.path.join(provider_path, filename))
|
|
return jsons
|
|
|
|
@pytest.mark.parametrize(
|
|
"json_path",
|
|
_find_all_compliance_jsons.__func__(),
|
|
ids=lambda p: os.path.basename(p),
|
|
)
|
|
def test_loads_as_universal(self, json_path):
|
|
fw = load_compliance_framework_universal(json_path)
|
|
assert fw is not None, f"Failed to load {json_path}"
|
|
assert fw.framework
|
|
assert fw.name
|
|
assert len(fw.requirements) >= 0
|
|
|
|
|
|
class TestBackwardCompat:
|
|
"""Ensure Compliance.get_bulk still returns Compliance objects."""
|
|
|
|
def test_get_bulk_still_works(self):
|
|
# This test just validates the legacy path still returns Compliance objects
|
|
# We test with a constructed Compliance object
|
|
legacy = CIS_1_4_AWS
|
|
assert isinstance(legacy, Compliance)
|
|
assert legacy.Framework == "CIS"
|
|
|
|
|
|
class TestAttributesMetadataValidation:
|
|
"""Validate that Requirement attributes match their attributes_metadata schema."""
|
|
|
|
def _metadata(self, required=False, enum=None, type_str="str"):
|
|
return [
|
|
AttributeMetadata(key="Section", type="str", required=True),
|
|
AttributeMetadata(key="Level", type=type_str, required=required, enum=enum),
|
|
]
|
|
|
|
def test_valid_attributes_pass(self):
|
|
fw = ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={"Section": "IAM", "Level": "high"},
|
|
checks={},
|
|
),
|
|
],
|
|
attributes_metadata=self._metadata(),
|
|
)
|
|
assert len(fw.requirements) == 1
|
|
|
|
def test_missing_required_key_raises(self):
|
|
with pytest.raises(
|
|
ValidationError, match="missing required attribute 'Section'"
|
|
):
|
|
ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={"Level": "high"},
|
|
checks={},
|
|
),
|
|
],
|
|
attributes_metadata=self._metadata(),
|
|
)
|
|
|
|
def test_invalid_enum_value_raises(self):
|
|
with pytest.raises(ValidationError, match="not in"):
|
|
ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={"Section": "IAM", "Level": "invalid"},
|
|
checks={},
|
|
),
|
|
],
|
|
attributes_metadata=self._metadata(enum=["high", "low"]),
|
|
)
|
|
|
|
def test_valid_enum_value_passes(self):
|
|
fw = ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={"Section": "IAM", "Level": "high"},
|
|
checks={},
|
|
),
|
|
],
|
|
attributes_metadata=self._metadata(enum=["high", "low"]),
|
|
)
|
|
assert len(fw.requirements) == 1
|
|
|
|
def test_wrong_type_int_raises(self):
|
|
with pytest.raises(ValidationError, match="expected type int"):
|
|
ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={"Section": "IAM", "Level": "not_a_number"},
|
|
checks={},
|
|
),
|
|
],
|
|
attributes_metadata=self._metadata(type_str="int"),
|
|
)
|
|
|
|
def test_correct_type_int_passes(self):
|
|
fw = ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={"Section": "IAM", "Level": 5},
|
|
checks={},
|
|
),
|
|
],
|
|
attributes_metadata=self._metadata(type_str="int"),
|
|
)
|
|
assert fw.requirements[0].attributes["Level"] == 5
|
|
|
|
def test_none_optional_value_skips_validation(self):
|
|
"""None values for non-required keys should not trigger type/enum errors."""
|
|
fw = ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={"Section": "IAM", "Level": None},
|
|
checks={},
|
|
),
|
|
],
|
|
attributes_metadata=self._metadata(enum=["high", "low"]),
|
|
)
|
|
assert len(fw.requirements) == 1
|
|
|
|
def test_no_metadata_skips_validation(self):
|
|
"""Frameworks without attributes_metadata should not be validated."""
|
|
fw = ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={"anything": "goes"},
|
|
checks={},
|
|
),
|
|
],
|
|
)
|
|
assert len(fw.requirements) == 1
|
|
|
|
def test_unknown_attribute_key_raises(self):
|
|
"""Typos like 'Sectoin' must be rejected by the schema validator."""
|
|
with pytest.raises(ValidationError, match="unknown attribute 'Sectoin'"):
|
|
ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={"Sectoin": "IAM", "Level": "high"},
|
|
checks={},
|
|
),
|
|
],
|
|
attributes_metadata=self._metadata(enum=["high", "low"]),
|
|
)
|
|
|
|
def test_multiple_unknown_keys_all_reported(self):
|
|
"""Every unknown key must appear in the validation error (deterministic order)."""
|
|
with pytest.raises(
|
|
ValidationError,
|
|
match=r"unknown attribute 'Bogus1'[\s\S]*unknown attribute 'Bogus2'",
|
|
):
|
|
ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={
|
|
"Section": "IAM",
|
|
"Level": "high",
|
|
"Bogus1": "x",
|
|
"Bogus2": "y",
|
|
},
|
|
checks={},
|
|
),
|
|
],
|
|
attributes_metadata=self._metadata(enum=["high", "low"]),
|
|
)
|
|
|
|
def test_multiple_errors_reported(self):
|
|
"""All validation errors should be collected and reported together."""
|
|
with pytest.raises(
|
|
ValidationError, match="missing required attribute 'Section'"
|
|
):
|
|
ComplianceFramework(
|
|
framework="Test",
|
|
name="Test",
|
|
description="d",
|
|
requirements=[
|
|
UniversalComplianceRequirement(
|
|
id="1.1",
|
|
description="d",
|
|
attributes={"Level": "bad"},
|
|
checks={},
|
|
),
|
|
UniversalComplianceRequirement(
|
|
id="1.2",
|
|
description="d",
|
|
attributes={"Level": "also_bad"},
|
|
checks={},
|
|
),
|
|
],
|
|
attributes_metadata=self._metadata(enum=["high", "low"]),
|
|
)
|
|
|
|
|
|
class TestGetBulkUniversalEntryPoints:
|
|
"""Entry-point discovery for universal (multi-provider) compliance frameworks."""
|
|
|
|
@staticmethod
|
|
def _write_universal_json(directory, filename, framework, display_name):
|
|
data = {
|
|
"framework": framework,
|
|
"name": display_name,
|
|
"version": "1.0",
|
|
"description": "External multi-provider framework",
|
|
"requirements": [
|
|
{
|
|
"id": "1",
|
|
"name": "Requirement 1",
|
|
"description": "desc",
|
|
"checks": {"fakeexternal": ["check_a"]},
|
|
}
|
|
],
|
|
}
|
|
with open(os.path.join(directory, filename), "w") as f:
|
|
json.dump(data, f)
|
|
|
|
@staticmethod
|
|
def _entry_point(path):
|
|
module = MagicMock()
|
|
module.__path__ = [path]
|
|
ep = MagicMock()
|
|
ep.name = "fakeexternal"
|
|
ep.group = "prowler.compliance.universal"
|
|
ep.load.return_value = module
|
|
return ep
|
|
|
|
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
|
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
|
def test_includes_external_universal_framework(self, mock_list_modules, mock_ep):
|
|
mock_list_modules.return_value = []
|
|
with tempfile.TemporaryDirectory() as ep_dir:
|
|
self._write_universal_json(
|
|
ep_dir, "customuniversal_1.0.json", "CustomUniversal", "Custom"
|
|
)
|
|
mock_ep.return_value = [self._entry_point(ep_dir)]
|
|
|
|
bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
|
|
|
|
mock_ep.assert_called_with(group="prowler.compliance.universal")
|
|
assert "customuniversal_1.0" in bulk
|
|
assert bulk["customuniversal_1.0"].framework == "CustomUniversal"
|
|
|
|
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
|
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
|
def test_builtin_wins_over_external_on_name_collision(
|
|
self, mock_list_modules, mock_ep
|
|
):
|
|
with (
|
|
tempfile.TemporaryDirectory() as root,
|
|
tempfile.TemporaryDirectory() as ep_dir,
|
|
):
|
|
builtin_sub = os.path.join(root, "builtinprov")
|
|
os.makedirs(builtin_sub)
|
|
self._write_universal_json(
|
|
builtin_sub, "shared_1.0.json", "SharedFramework", "Built-in"
|
|
)
|
|
builtin_module = MagicMock()
|
|
builtin_module.module_finder.path = root
|
|
builtin_module.name = "prowler.compliance.builtinprov"
|
|
mock_list_modules.return_value = [builtin_module]
|
|
|
|
self._write_universal_json(
|
|
ep_dir, "shared_1.0.json", "SharedFramework", "External"
|
|
)
|
|
mock_ep.return_value = [self._entry_point(ep_dir)]
|
|
|
|
bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
|
|
|
|
assert "shared_1.0" in bulk
|
|
assert bulk["shared_1.0"].name == "Built-in"
|
|
|
|
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
|
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
|
def test_loads_all_frameworks_in_a_single_entry_point_path(
|
|
self, mock_list_modules, mock_ep
|
|
):
|
|
"""All JSONs in one entry-point directory are added, not collapsed to one."""
|
|
mock_list_modules.return_value = []
|
|
with tempfile.TemporaryDirectory() as ep_dir:
|
|
self._write_universal_json(ep_dir, "fw_a_1.0.json", "FwA", "Framework A")
|
|
self._write_universal_json(ep_dir, "fw_b_1.0.json", "FwB", "Framework B")
|
|
mock_ep.return_value = [self._entry_point(ep_dir)]
|
|
|
|
bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
|
|
|
|
assert "fw_a_1.0" in bulk
|
|
assert "fw_b_1.0" in bulk
|
|
|
|
@patch("prowler.lib.check.compliance_models.importlib.metadata.entry_points")
|
|
@patch("prowler.lib.check.compliance_models.list_compliance_modules")
|
|
def test_merges_frameworks_from_multiple_packages_same_provider(
|
|
self, mock_list_modules, mock_ep
|
|
):
|
|
"""Two packages under the same provider name are both discovered."""
|
|
mock_list_modules.return_value = []
|
|
with (
|
|
tempfile.TemporaryDirectory() as dir_a,
|
|
tempfile.TemporaryDirectory() as dir_b,
|
|
):
|
|
self._write_universal_json(dir_a, "pkg_a_1.0.json", "PkgA", "Package A")
|
|
self._write_universal_json(dir_b, "pkg_b_1.0.json", "PkgB", "Package B")
|
|
mock_ep.return_value = [
|
|
self._entry_point(dir_a),
|
|
self._entry_point(dir_b),
|
|
]
|
|
|
|
bulk = get_bulk_compliance_frameworks_universal("fakeexternal")
|
|
|
|
assert "pkg_a_1.0" in bulk
|
|
assert "pkg_b_1.0" in bulk
|