mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 04:51:51 +00:00
fix(check): custom check folder validation (#9335)
This commit is contained in:
committed by
GitHub
parent
bb43e924ee
commit
1250f582a5
@@ -14,6 +14,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
|
||||
|
||||
### Fixed
|
||||
- `sharepoint_external_sharing_managed` check to handle external sharing disabled at organization level [(#9298)](https://github.com/prowler-cloud/prowler/pull/9298)
|
||||
- Custom check folder metadata validation [(#9335)](https://github.com/prowler-cloud/prowler/pull/9335)
|
||||
- Support multiple Exchange mailbox policies in M365 `exchange_mailbox_policy_additional_storage_restricted` check [(#9241)](https://github.com/prowler-cloud/prowler/pull/9241)
|
||||
|
||||
---
|
||||
|
||||
@@ -24,6 +24,7 @@ from prowler.lib.check.check import (
|
||||
list_checks_json,
|
||||
list_fixers,
|
||||
list_services,
|
||||
load_custom_checks_metadata,
|
||||
parse_checks_from_file,
|
||||
parse_checks_from_folder,
|
||||
print_categories,
|
||||
@@ -185,6 +186,11 @@ def prowler():
|
||||
logger.debug("Loading checks metadata from .metadata.json files")
|
||||
bulk_checks_metadata = CheckMetadata.get_bulk(provider)
|
||||
|
||||
# Load custom checks metadata before validation
|
||||
if checks_folder:
|
||||
custom_folder_metadata = load_custom_checks_metadata(checks_folder)
|
||||
bulk_checks_metadata.update(custom_folder_metadata)
|
||||
|
||||
if args.list_categories:
|
||||
print_categories(list_categories(bulk_checks_metadata))
|
||||
sys.exit()
|
||||
|
||||
@@ -14,7 +14,7 @@ from colorama import Fore, Style
|
||||
import prowler
|
||||
from prowler.config.config import orange_color
|
||||
from prowler.lib.check.custom_checks_metadata import update_check_metadata
|
||||
from prowler.lib.check.models import Check
|
||||
from prowler.lib.check.models import Check, load_check_metadata
|
||||
from prowler.lib.check.utils import recover_checks_from_provider
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.lib.outputs.outputs import report
|
||||
@@ -110,6 +110,48 @@ def parse_checks_from_folder(provider, input_folder: str) -> set:
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def load_custom_checks_metadata(input_folder: str) -> dict:
|
||||
"""
|
||||
Load check metadata from a custom checks folder without copying the checks.
|
||||
This is used to validate check names before the provider is initialized.
|
||||
|
||||
Args:
|
||||
input_folder (str): Path to the folder containing custom checks.
|
||||
|
||||
Returns:
|
||||
dict: A dictionary with CheckID as key and CheckMetadata as value.
|
||||
"""
|
||||
custom_checks_metadata = {}
|
||||
|
||||
try:
|
||||
if not os.path.isdir(input_folder):
|
||||
return custom_checks_metadata
|
||||
|
||||
with os.scandir(input_folder) as checks:
|
||||
for check in checks:
|
||||
if check.is_dir():
|
||||
check_name = check.name
|
||||
metadata_file = os.path.join(
|
||||
input_folder, check_name, f"{check_name}.metadata.json"
|
||||
)
|
||||
if os.path.isfile(metadata_file):
|
||||
try:
|
||||
check_metadata = load_check_metadata(metadata_file)
|
||||
custom_checks_metadata[check_metadata.CheckID] = (
|
||||
check_metadata
|
||||
)
|
||||
except Exception as error:
|
||||
logger.warning(
|
||||
f"Could not load metadata from {metadata_file}: {error}"
|
||||
)
|
||||
return custom_checks_metadata
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
|
||||
)
|
||||
return custom_checks_metadata
|
||||
|
||||
|
||||
# Load checks from custom folder
|
||||
def remove_custom_checks_module(input_folder: str, provider: str):
|
||||
# Check if input folder is a S3 URI
|
||||
|
||||
@@ -18,6 +18,7 @@ from prowler.lib.check.check import (
|
||||
list_categories,
|
||||
list_checks_json,
|
||||
list_services,
|
||||
load_custom_checks_metadata,
|
||||
parse_checks_from_file,
|
||||
parse_checks_from_folder,
|
||||
remove_custom_checks_module,
|
||||
@@ -483,6 +484,49 @@ class TestCheck:
|
||||
)
|
||||
remove_custom_checks_module(check_folder, provider)
|
||||
|
||||
def test_load_custom_checks_metadata(self, tmp_path):
|
||||
"""Test loading check metadata from a custom checks folder."""
|
||||
check_name = "custom_test_check"
|
||||
check_folder = tmp_path / check_name
|
||||
check_folder.mkdir()
|
||||
|
||||
metadata = {
|
||||
"Provider": "aws",
|
||||
"CheckID": check_name,
|
||||
"CheckTitle": "Test Custom Check",
|
||||
"CheckType": [],
|
||||
"ServiceName": "custom",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "arn:aws:custom:::resource",
|
||||
"Severity": "low",
|
||||
"ResourceType": "AwsCustomResource",
|
||||
"Description": "A test custom check",
|
||||
"Risk": "Test risk",
|
||||
"RelatedUrl": "https://example.com",
|
||||
"Remediation": {
|
||||
"Code": {"CLI": "", "NativeIaC": "", "Other": "", "Terraform": ""},
|
||||
"Recommendation": {"Text": "", "Url": ""},
|
||||
},
|
||||
"Categories": [],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "",
|
||||
}
|
||||
metadata_file = check_folder / f"{check_name}.metadata.json"
|
||||
metadata_file.write_text(json.dumps(metadata))
|
||||
|
||||
result = load_custom_checks_metadata(str(tmp_path))
|
||||
|
||||
assert check_name in result
|
||||
assert result[check_name].CheckID == check_name
|
||||
assert result[check_name].Provider == "aws"
|
||||
assert result[check_name].Severity == "low"
|
||||
|
||||
def test_load_custom_checks_metadata_nonexistent_path(self):
|
||||
"""Test that nonexistent paths return empty dict."""
|
||||
result = load_custom_checks_metadata("/nonexistent/path/to/checks")
|
||||
assert result == {}
|
||||
|
||||
def test_exclude_checks_to_run(self):
|
||||
test_cases = [
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user