Files
prowler/tests/lib/check/compliance_check_test.py
2025-09-22 14:53:27 +02:00

557 lines
21 KiB
Python

from unittest import mock
from prowler.lib.check.compliance import update_checks_metadata_with_compliance
from prowler.lib.check.compliance_models import (
CIS_Requirement_Attribute,
CIS_Requirement_Attribute_AssessmentStatus,
CIS_Requirement_Attribute_Profile,
Compliance,
Compliance_Requirement,
)
from prowler.lib.check.models import CheckMetadata
custom_compliance_metadata = {
"framework1_aws": Compliance(
Framework="Framework1",
Name="Framework 1",
Provider="aws",
Version="1.0",
Description="Framework 1 Description",
Requirements=[
Compliance_Requirement(
Id="1.1.1",
Description="description",
Attributes=[
CIS_Requirement_Attribute(
Section="1. Identity",
Profile=CIS_Requirement_Attribute_Profile("Level 1"),
AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus(
"Manual"
),
Description="Description",
RationaleStatement="Rationale",
ImpactStatement="Impact",
RemediationProcedure="Remediation",
AuditProcedure="Audit",
AdditionalInformation="Additional",
References="References",
)
],
Checks=[
"accessanalyzer_enabled",
"iam_user_mfa_enabled_console_access",
],
),
# Manual requirement
Compliance_Requirement(
Id="1.1.2",
Description="description",
Attributes=[
CIS_Requirement_Attribute(
Section="1. Identity",
Profile=CIS_Requirement_Attribute_Profile("Level 1"),
AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus(
"Manual"
),
Description="Description",
RationaleStatement="Rationale",
ImpactStatement="Impact",
RemediationProcedure="Remediation",
AuditProcedure="Audit",
AdditionalInformation="Additional",
References="References",
)
],
Checks=[],
),
],
),
"framework1_azure": Compliance(
Framework="Framework1",
Name="Framework 1",
Provider="azure",
Version="1.0",
Description="Framework 2 Description",
Requirements=[
Compliance_Requirement(
Id="1.1.1",
Description="description",
Attributes=[
CIS_Requirement_Attribute(
Section="1. Identity",
Profile=CIS_Requirement_Attribute_Profile("Level 1"),
AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus(
"Manual"
),
Description="Description",
RationaleStatement="Rationale",
ImpactStatement="Impact",
RemediationProcedure="Remediation",
AuditProcedure="Audit",
AdditionalInformation="Additional",
References="References",
)
],
Checks=[],
)
],
),
"framework1_gcp": Compliance(
Framework="Framework1",
Name="Framework 1",
Provider="gcp",
Version="1.0",
Description="Framework 2 Description",
Requirements=[
Compliance_Requirement(
Id="1.1.1",
Description="description",
Attributes=[
CIS_Requirement_Attribute(
Section="1. Identity",
Profile=CIS_Requirement_Attribute_Profile("Level 1"),
AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus(
"Manual"
),
Description="Description",
RationaleStatement="Rationale",
ImpactStatement="Impact",
RemediationProcedure="Remediation",
AuditProcedure="Audit",
AdditionalInformation="Additional",
References="References",
)
],
Checks=[],
)
],
),
"framework1_k8s": Compliance(
Framework="Framework1",
Name="Framework 1",
Provider="Kubernetes",
Version="1.0",
Description="Framework 2 Description",
Requirements=[
Compliance_Requirement(
Id="1.1.1",
Description="description",
Attributes=[
CIS_Requirement_Attribute(
Section="1. Identity",
Profile=CIS_Requirement_Attribute_Profile("Level 1"),
AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus(
"Manual"
),
Description="Description",
RationaleStatement="Rationale",
ImpactStatement="Impact",
RemediationProcedure="Remediation",
AuditProcedure="Audit",
AdditionalInformation="Additional",
References="References",
)
],
Checks=[],
)
],
),
"framework1_m365": Compliance(
Framework="Framework1",
Name="Framework 1",
Provider="m365",
Version="1.0",
Description="Framework 2 Description",
Requirements=[
Compliance_Requirement(
Id="1.1.1",
Description="description",
Attributes=[
CIS_Requirement_Attribute(
Section="1. Identity",
Profile=CIS_Requirement_Attribute_Profile("E3 Level 1"),
AssessmentStatus=CIS_Requirement_Attribute_AssessmentStatus(
"Manual"
),
Description="Description",
RationaleStatement="Rationale",
ImpactStatement="Impact",
RemediationProcedure="Remediation",
AuditProcedure="Audit",
AdditionalInformation="Additional",
References="References",
)
],
Checks=[],
)
],
),
}
class TestCompliance:
def get_custom_check_metadata(self):
return {
"accessanalyzer_enabled": CheckMetadata(
Provider="aws",
CheckID="accessanalyzer_enabled",
CheckTitle="Check 1",
CheckType=["type1"],
ServiceName="accessanalyzer",
SubServiceName="subservice1",
ResourceIdTemplate="template1",
Severity="high",
ResourceType="resource1",
Description="Description 1",
Risk="risk1",
RelatedUrl="url1",
Remediation={
"Code": {
"CLI": "cli1",
"NativeIaC": "native1",
"Other": "other1",
"Terraform": "terraform1",
},
"Recommendation": {"Text": "text1", "Url": "url1"},
},
Categories=["categoryone"],
DependsOn=["dependency1"],
RelatedTo=["related1"],
Notes="notes1",
Compliance=[],
),
"iam_user_mfa_enabled_console_access": CheckMetadata(
Provider="aws",
CheckID="iam_user_mfa_enabled_console_access",
CheckTitle="Check 2",
CheckType=["type2"],
ServiceName="iam",
SubServiceName="subservice2",
ResourceIdTemplate="template2",
Severity="medium",
ResourceType="resource2",
Description="Description 2",
Risk="risk2",
RelatedUrl="url2",
Remediation={
"Code": {
"CLI": "cli2",
"NativeIaC": "native2",
"Other": "other2",
"Terraform": "terraform2",
},
"Recommendation": {"Text": "text2", "Url": "url2"},
},
Categories=["categorytwo"],
DependsOn=["dependency2"],
RelatedTo=["related2"],
Notes="notes2",
Compliance=[],
),
}
def test_update_checks_metadata(self):
bulk_compliance_frameworks = custom_compliance_metadata
bulk_checks_metadata = self.get_custom_check_metadata()
updated_metadata = update_checks_metadata_with_compliance(
bulk_compliance_frameworks, bulk_checks_metadata
)
assert "accessanalyzer_enabled" in updated_metadata
assert "iam_user_mfa_enabled_console_access" in updated_metadata
accessanalyzer_enabled_compliance = updated_metadata[
"accessanalyzer_enabled"
].Compliance[0]
assert len(updated_metadata["accessanalyzer_enabled"].Compliance) == 1
assert accessanalyzer_enabled_compliance.Framework == "Framework1"
assert accessanalyzer_enabled_compliance.Provider == "aws"
assert accessanalyzer_enabled_compliance.Version == "1.0"
assert (
accessanalyzer_enabled_compliance.Description == "Framework 1 Description"
)
assert len(accessanalyzer_enabled_compliance.Requirements) == 1
accessanalyzer_enabled_requirement = (
accessanalyzer_enabled_compliance.Requirements[0]
)
assert accessanalyzer_enabled_requirement.Id == "1.1.1"
assert accessanalyzer_enabled_requirement.Description == "description"
assert len(accessanalyzer_enabled_requirement.Attributes) == 1
accessanalyzer_enabled_attribute = (
accessanalyzer_enabled_requirement.Attributes[0]
)
assert accessanalyzer_enabled_attribute.Section == "1. Identity"
assert accessanalyzer_enabled_attribute.Profile == "Level 1"
assert accessanalyzer_enabled_attribute.AssessmentStatus == "Manual"
assert accessanalyzer_enabled_attribute.Description == "Description"
assert accessanalyzer_enabled_attribute.RationaleStatement == "Rationale"
assert accessanalyzer_enabled_attribute.ImpactStatement == "Impact"
assert accessanalyzer_enabled_attribute.RemediationProcedure == "Remediation"
assert accessanalyzer_enabled_attribute.AuditProcedure == "Audit"
assert accessanalyzer_enabled_attribute.AdditionalInformation == "Additional"
assert accessanalyzer_enabled_attribute.References == "References"
def test_list_no_provider(self):
bulk_compliance_frameworks = custom_compliance_metadata
list_compliance = Compliance.list(bulk_compliance_frameworks)
assert len(list_compliance) == 5
assert list_compliance[0] == "framework1_aws"
assert list_compliance[1] == "framework1_azure"
assert list_compliance[2] == "framework1_gcp"
assert list_compliance[3] == "framework1_k8s"
assert list_compliance[4] == "framework1_m365"
def test_list_with_provider_aws(self):
bulk_compliance_frameworks = custom_compliance_metadata
list_compliance = Compliance.list(bulk_compliance_frameworks, provider="aws")
assert len(list_compliance) == 1
assert list_compliance[0] == "framework1_aws"
def test_list_with_provider_azure(self):
bulk_compliance_frameworks = custom_compliance_metadata
list_compliance = Compliance.list(bulk_compliance_frameworks, provider="azure")
assert len(list_compliance) == 1
assert list_compliance[0] == "framework1_azure"
def test_list_with_provider_gcp(self):
bulk_compliance_frameworks = custom_compliance_metadata
list_compliance = Compliance.list(bulk_compliance_frameworks, provider="gcp")
assert len(list_compliance) == 1
assert list_compliance[0] == "framework1_gcp"
def test_list_with_provider_k8s(self):
bulk_compliance_frameworks = custom_compliance_metadata
list_compliance = Compliance.list(bulk_compliance_frameworks, provider="k8s")
assert len(list_compliance) == 1
assert list_compliance[0] == "framework1_k8s"
def test_list_with_provider_m365(self):
bulk_compliance_frameworks = custom_compliance_metadata
list_compliance = Compliance.list(bulk_compliance_frameworks, provider="m365")
assert len(list_compliance) == 1
assert list_compliance[0] == "framework1_m365"
def test_get_compliance_frameworks(self):
bulk_compliance_frameworks = custom_compliance_metadata
compliance_framework = Compliance.get(
bulk_compliance_frameworks, compliance_framework_name="framework1_aws"
)
assert compliance_framework.Framework == "Framework1"
assert compliance_framework.Provider == "aws"
assert compliance_framework.Version == "1.0"
assert compliance_framework.Description == "Framework 1 Description"
assert len(compliance_framework.Requirements) == 2
compliance_framework = Compliance.get(
bulk_compliance_frameworks, compliance_framework_name="framework1_azure"
)
assert compliance_framework.Framework == "Framework1"
assert compliance_framework.Provider == "azure"
assert compliance_framework.Version == "1.0"
assert compliance_framework.Description == "Framework 2 Description"
assert len(compliance_framework.Requirements) == 1
compliance_framework = Compliance.get(
bulk_compliance_frameworks, compliance_framework_name="framework1_gcp"
)
assert compliance_framework.Framework == "Framework1"
assert compliance_framework.Provider == "gcp"
assert compliance_framework.Version == "1.0"
assert compliance_framework.Description == "Framework 2 Description"
assert len(compliance_framework.Requirements) == 1
compliance_framework = Compliance.get(
bulk_compliance_frameworks, compliance_framework_name="framework1_k8s"
)
assert compliance_framework.Framework == "Framework1"
assert compliance_framework.Provider == "Kubernetes"
assert compliance_framework.Version == "1.0"
assert compliance_framework.Description == "Framework 2 Description"
assert len(compliance_framework.Requirements) == 1
compliance_framework = Compliance.get(
bulk_compliance_frameworks, compliance_framework_name="framework1_m365"
)
assert compliance_framework.Framework == "Framework1"
assert compliance_framework.Provider == "m365"
assert compliance_framework.Version == "1.0"
assert compliance_framework.Description == "Framework 2 Description"
assert len(compliance_framework.Requirements) == 1
assert compliance_framework.Requirements[0].Id == "1.1.1"
assert compliance_framework.Requirements[0].Description == "description"
assert len(compliance_framework.Requirements[0].Attributes) == 1
assert (
compliance_framework.Requirements[0].Attributes[0].Section == "1. Identity"
)
assert (
compliance_framework.Requirements[0].Attributes[0].Profile == "E3 Level 1"
)
assert (
compliance_framework.Requirements[0].Attributes[0].AssessmentStatus
== "Manual"
)
assert (
compliance_framework.Requirements[0].Attributes[0].Description
== "Description"
)
assert (
compliance_framework.Requirements[0].Attributes[0].RationaleStatement
== "Rationale"
)
assert (
compliance_framework.Requirements[0].Attributes[0].ImpactStatement
== "Impact"
)
assert (
compliance_framework.Requirements[0].Attributes[0].RemediationProcedure
== "Remediation"
)
assert (
compliance_framework.Requirements[0].Attributes[0].AuditProcedure == "Audit"
)
assert (
compliance_framework.Requirements[0].Attributes[0].AdditionalInformation
== "Additional"
)
assert (
compliance_framework.Requirements[0].Attributes[0].References
== "References"
)
def test_get_non_existent_framework(self):
bulk_compliance_frameworks = custom_compliance_metadata
compliance_framework = Compliance.get(
bulk_compliance_frameworks, compliance_framework_name="non_existent"
)
assert compliance_framework is None
def test_list_compliance_requirements_no_compliance(self):
bulk_compliance_frameworks = custom_compliance_metadata
list_requirements = Compliance.list_requirements(bulk_compliance_frameworks)
assert len(list_requirements) == 0
def test_list_compliance_requirements_with_compliance(self):
bulk_compliance_frameworks = custom_compliance_metadata
list_requirements = Compliance.list_requirements(
bulk_compliance_frameworks, compliance_framework="framework1_aws"
)
assert len(list_requirements) == 2
assert list_requirements[0] == "1.1.1"
assert list_requirements[1] == "1.1.2"
list_requirements = Compliance.list_requirements(
bulk_compliance_frameworks, compliance_framework="framework1_azure"
)
assert len(list_requirements) == 1
assert list_requirements[0] == "1.1.1"
def test_get_compliance_requirement(self):
bulk_compliance_frameworks = custom_compliance_metadata
compliance_requirement = Compliance.get_requirement(
bulk_compliance_frameworks,
compliance_framework="framework1_aws",
requirement_id="1.1.1",
)
assert compliance_requirement.Id == "1.1.1"
assert compliance_requirement.Description == "description"
assert len(compliance_requirement.Attributes) == 1
compliance_requirement = Compliance.get_requirement(
bulk_compliance_frameworks,
compliance_framework="framework1_aws",
requirement_id="1.1.2",
)
assert compliance_requirement.Id == "1.1.2"
assert compliance_requirement.Description == "description"
assert len(compliance_requirement.Attributes) == 1
compliance_requirement = Compliance.get_requirement(
bulk_compliance_frameworks,
compliance_framework="framework1_azure",
requirement_id="1.1.1",
)
assert compliance_requirement.Id == "1.1.1"
assert compliance_requirement.Description == "description"
assert len(compliance_requirement.Attributes) == 1
def test_get_compliance_requirement_not_found(self):
bulk_compliance_frameworks = custom_compliance_metadata
compliance_requirement = Compliance.get_requirement(
bulk_compliance_frameworks,
compliance_framework="framework1_aws",
requirement_id="1.1.3",
)
assert compliance_requirement is None
@mock.patch("prowler.lib.check.compliance_models.load_compliance_framework")
@mock.patch("os.stat")
@mock.patch("os.path.isfile")
@mock.patch("os.listdir")
@mock.patch("prowler.lib.check.compliance_models.list_compliance_modules")
def test_get_bulk(
self,
mock_list_modules,
mock_listdir,
mock_isfile,
mock_stat,
mock_load_compliance,
):
object = mock.Mock()
object.path = "/path/to/compliance"
object.name = "framework1_aws"
mock_list_modules.return_value = [object]
mock_listdir.return_value = ["framework1_aws.json"]
mock_isfile.return_value = True
mock_stat.return_value.st_size = 100
mock_load_compliance.return_value = mock.Mock(
Framework="Framework1", Provider="aws"
)
from prowler.lib.check.compliance_models import Compliance
result = Compliance.get_bulk(provider="aws")
assert len(result) == 1
assert "framework1_aws" in result.keys()
mock_list_modules.assert_called_once()