ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 04:51:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: lib/check tests

Browse Source
This commit is contained in:
HugoPBrito
2025-08-27 11:34:17 +02:00
parent 2fa8b63e58
commit 3d19a496e5
6 changed files with 1160 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/lib/check/models.py
+14 -8
View File
@@ -27,20 +27,26 @@ def _validate_aws_check_type_in_config(check_type: str) -> bool:
bool: True if the CheckType path exists in the config hierarchy
"""
try:
# Get config directly from global provider like custom checks do
from prowler.providers.common.provider import Provider
import json
import os
if not hasattr(Provider, "_global_provider") or not Provider._global_provider:
if not check_type:
return False
# Access config directly like: service_client.audit_config.get("key", default)
hierarchy = Provider._global_provider.audit_config.get("aws", {}).get(
"valid_check_types", {}
# Get the path to the AWS CheckTypes configuration
current_dir = os.path.dirname(os.path.abspath(__file__))
check_types_file = os.path.join(
current_dir, "..", "..", "providers", "aws", "config", "check_types.json"
)
check_types_file = os.path.normpath(check_types_file)
if not check_type or not hierarchy:
# Load the CheckTypes hierarchy from JSON file
if not os.path.exists(check_types_file):
return False
with open(check_types_file, "r") as f:
hierarchy = json.load(f)
# Split the path by '/' to get each level
path_parts = check_type.split("/")
@@ -53,7 +59,7 @@ def _validate_aws_check_type_in_config(check_type: str) -> bool:
return True
except (KeyError, AttributeError):
except (KeyError, AttributeError, FileNotFoundError, json.JSONDecodeError):
return False
tests/lib/check/check_loader_test.py
+38 -34
View File
@@ -1,4 +1,4 @@
from mock import patch
from unittest.mock import patch
from prowler.lib.check.checks_loader import (
load_checks_to_execute,
@@ -30,7 +30,9 @@ class TestCheckLoader:
Provider="aws",
CheckID=S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME,
CheckTitle="Check S3 Bucket Level Public Access Block.",
CheckType=["Data Protection"],
CheckType=[
"Software and Configuration Checks/AWS Security Best Practices/Network Reachability"
],
CheckAliases=[S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME_CUSTOM_ALIAS],
ServiceName=S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME_SERVICE,
SubServiceName="",
@@ -64,7 +66,9 @@ class TestCheckLoader:
Provider="aws",
CheckID=IAM_USER_NO_MFA_NAME,
CheckTitle="Check IAM User No MFA.",
CheckType=["Data Protection"],
CheckType=[
"Software and Configuration Checks/Industry and Regulatory Standards/CIS AWS Foundations Benchmark"
],
CheckAliases=[IAM_USER_NO_MFA_NAME_CUSTOM_ALIAS],
ServiceName=IAM_USER_NO_MFA_NAME_SERVICE,
SubServiceName="",
@@ -98,7 +102,7 @@ class TestCheckLoader:
Provider="aws",
CheckID=CLOUDTRAIL_THREAT_DETECTION_ENUMERATION_NAME,
CheckTitle="Ensure there are no potential enumeration threats in CloudTrail",
CheckType=[],
CheckType=["TTPs/Discovery"],
ServiceName="cloudtrail",
SubServiceName="",
ResourceIdTemplate="arn:partition:service:region:account-id:resource-id",
@@ -122,55 +126,55 @@ class TestCheckLoader:
)
def test_load_checks_to_execute(self):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
provider=self.provider,
)
def test_load_checks_to_execute_with_check_list(self):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
check_list = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME]
assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
check_list=check_list,
provider=self.provider,
)
def test_load_checks_to_execute_with_severities(self):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
severities = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_SEVERITY]
assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
severities=severities,
provider=self.provider,
)
def test_load_checks_to_execute_with_severities_and_services(self):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
service_list = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME_SERVICE]
severities = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_SEVERITY]
assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
service_list=service_list,
severities=severities,
provider=self.provider,
)
def test_load_checks_to_execute_with_severities_and_services_multiple(self):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata(),
IAM_USER_NO_MFA_NAME: self.get_custom_check_iam_metadata(),
}
@@ -181,7 +185,7 @@ class TestCheckLoader:
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME,
IAM_USER_NO_MFA_NAME,
} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
service_list=service_list,
severities=severities,
provider=self.provider,
@@ -190,14 +194,14 @@ class TestCheckLoader:
def test_load_checks_to_execute_with_severities_and_services_not_within_severity(
self,
):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
service_list = ["ec2"]
severities = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_SEVERITY]
assert set() == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
service_list=service_list,
severities=severities,
provider=self.provider,
@@ -206,7 +210,7 @@ class TestCheckLoader:
def test_load_checks_to_execute_with_checks_file(
self,
):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
checks_file = "path/to/test_file"
@@ -215,7 +219,7 @@ class TestCheckLoader:
return_value={S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME},
):
assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
checks_file=checks_file,
provider=self.provider,
)
@@ -223,13 +227,13 @@ class TestCheckLoader:
def test_load_checks_to_execute_with_service_list(
self,
):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
service_list = [S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME_SERVICE]
assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
service_list=service_list,
provider=self.provider,
)
@@ -237,7 +241,7 @@ class TestCheckLoader:
def test_load_checks_to_execute_with_compliance_frameworks(
self,
):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
bulk_compliance_frameworks = {
@@ -259,7 +263,7 @@ class TestCheckLoader:
compliance_frameworks = ["soc2_aws"]
assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
bulk_compliance_frameworks=bulk_compliance_frameworks,
compliance_frameworks=compliance_frameworks,
provider=self.provider,
@@ -268,24 +272,24 @@ class TestCheckLoader:
def test_load_checks_to_execute_with_categories(
self,
):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
categories = {"internet-exposed"}
assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
categories=categories,
provider=self.provider,
)
def test_load_checks_to_execute_no_bulk_checks_metadata(self):
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
with patch(
"prowler.lib.check.checks_loader.CheckMetadata.get_bulk",
return_value=bulk_checks_metatada,
return_value=bulk_checks_metadata,
):
assert {S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME} == load_checks_to_execute(
provider=self.provider,
@@ -311,13 +315,13 @@ class TestCheckLoader:
compliance_frameworks = ["soc2_aws"]
bulk_checks_metatada = {
bulk_checks_metadata = {
S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME: self.get_custom_check_s3_metadata()
}
with (
patch(
"prowler.lib.check.checks_loader.CheckMetadata.get_bulk",
return_value=bulk_checks_metatada,
return_value=bulk_checks_metadata,
),
patch(
"prowler.lib.check.checks_loader.Compliance.get_bulk",
@@ -344,38 +348,38 @@ class TestCheckLoader:
)
def test_threat_detection_category(self):
bulk_checks_metatada = {
bulk_checks_metadata = {
CLOUDTRAIL_THREAT_DETECTION_ENUMERATION_NAME: self.get_threat_detection_check_metadata()
}
categories = {"threat-detection"}
assert {CLOUDTRAIL_THREAT_DETECTION_ENUMERATION_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
categories=categories,
provider=self.provider,
)
def test_discard_threat_detection_checks(self):
bulk_checks_metatada = {
bulk_checks_metadata = {
CLOUDTRAIL_THREAT_DETECTION_ENUMERATION_NAME: self.get_threat_detection_check_metadata()
}
categories = {}
assert set() == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
categories=categories,
provider=self.provider,
)
def test_threat_detection_single_check(self):
bulk_checks_metatada = {
bulk_checks_metadata = {
CLOUDTRAIL_THREAT_DETECTION_ENUMERATION_NAME: self.get_threat_detection_check_metadata()
}
categories = {}
check_list = [CLOUDTRAIL_THREAT_DETECTION_ENUMERATION_NAME]
assert {CLOUDTRAIL_THREAT_DETECTION_ENUMERATION_NAME} == load_checks_to_execute(
bulk_checks_metadata=bulk_checks_metatada,
bulk_checks_metadata=bulk_checks_metadata,
check_list=check_list,
categories=categories,
provider=self.provider,
tests/lib/check/compliance_check_test.py
+2 -2
View File
@@ -192,7 +192,7 @@ class TestCompliance:
Provider="aws",
CheckID="accessanalyzer_enabled",
CheckTitle="Check 1",
CheckType=["type1"],
CheckType=["TTPs/Initial Access"],
ServiceName="accessanalyzer",
SubServiceName="subservice1",
ResourceIdTemplate="template1",
@@ -220,7 +220,7 @@ class TestCompliance:
Provider="aws",
CheckID="iam_user_mfa_enabled_console_access",
CheckTitle="Check 2",
CheckType=["type2"],
CheckType=["TTPs/Credential Access"],
ServiceName="iam",
SubServiceName="subservice2",
ResourceIdTemplate="template2",
tests/lib/check/custom_checks_metadata_test.py
+1 -1
View File
@@ -36,7 +36,7 @@ class TestCustomChecksMetadata:
Provider="aws",
CheckID=S3_BUCKET_LEVEL_PUBLIC_ACCESS_BLOCK_NAME,
CheckTitle="Check S3 Bucket Level Public Access Block.",
CheckType=["Data Protection"],
CheckType=["Sensitive Data Identifications/PII"],
CheckAliases=[],
ServiceName="s3",
SubServiceName="",
tests/lib/check/fixtures/bulk_checks_metadata.py
+15 -5
View File
@@ -5,7 +5,9 @@ test_bulk_checks_metadata = {
Provider="aws",
CheckID="vpc_peering_routing_tables_with_least_privilege",
CheckTitle="Ensure routing tables for VPC peering are least access.",
CheckType=["Infrastructure Security"],
CheckType=[
"Software and Configuration Checks/AWS Security Best Practices/Network Reachability"
],
ServiceName="vpc",
SubServiceName="route_table",
ResourceIdTemplate="arn:partition:service:region:account-id:resource-id",
@@ -36,7 +38,9 @@ test_bulk_checks_metadata = {
Provider="aws",
CheckID="vpc_subnet_different_az",
CheckTitle="Ensure all vpc has subnets in more than one availability zone",
CheckType=["Infrastructure Security"],
CheckType=[
"Software and Configuration Checks/AWS Security Best Practices/Network Reachability"
],
ServiceName="vpc",
SubServiceName="subnet",
ResourceIdTemplate="arn:partition:service:region:account-id:resource-id",
@@ -64,7 +68,9 @@ test_bulk_checks_metadata = {
Provider="aws",
CheckID="vpc_subnet_separate_private_public",
CheckTitle="Ensure all vpc has public and private subnets defined",
CheckType=["Infrastructure Security"],
CheckType=[
"Software and Configuration Checks/AWS Security Best Practices/Network Reachability"
],
ServiceName="vpc",
SubServiceName="subnet",
ResourceIdTemplate="arn:partition:service:region:account-id:resource-id",
@@ -91,7 +97,9 @@ test_bulk_checks_metadata = {
Provider="aws",
CheckID="workspaces_volume_encryption_enabled",
CheckTitle="Ensure that your Amazon WorkSpaces storage volumes are encrypted in order to meet security and compliance requirements",
CheckType=[],
CheckType=[
"Software and Configuration Checks/AWS Security Best Practices/Runtime Behavior Analysis"
],
ServiceName="workspaces",
SubServiceName="",
ResourceIdTemplate="arn:aws:workspaces:region:account-id:workspace",
@@ -122,7 +130,9 @@ test_bulk_checks_metadata = {
Provider="aws",
CheckID="workspaces_vpc_2private_1public_subnets_nat",
CheckTitle="Ensure that the Workspaces VPC are deployed following the best practices using 1 public subnet and 2 private subnets with a NAT Gateway attached",
CheckType=[],
CheckType=[
"Software and Configuration Checks/AWS Security Best Practices/Runtime Behavior Analysis"
],
ServiceName="workspaces",
SubServiceName="",
ResourceIdTemplate="arn:aws:workspaces:region:account-id:workspace",
tests/lib/check/models_test.py
+1090 -4
View File
File diff suppressed because it is too large Load Diff