【Consejos/Solución】Cómo crear cheques personalizados con Prowler

📅 Creado: 2025/02/11, Actualizado: 2025/02/11 ⏱️ Tiempo estimado de lectura ~9 mins (1,733 palabras)

Prowler is a tool for checking the security settings of AWS, Azure, and GCP accounts.
In addition to the checks provided by default, you can also create custom checks.
Since the method for creating custom checks was not detailed in the official documentation, we analyzed the source code to confirm the method for creating custom checks.This article describes how to create custom checks with Prowler.

Índice

What is Prowler?
What does the official documentation say about custom checks?
(Summary first) File structure and contents of custom checks
1. Using Prowler’s standard client
2. Using a custom client
Debugging method during custom check development
Reading Prowler’s Source Code (as of v1.12.1)
Resumen

What is Prowler?

As described on GitHub - prowler-cloud/prowler :

Prowler is an open-source security tool for AWS, Azure, Google Cloud, and Kubernetes that performs security best practices assessments, audits, incident response, continuous monitoring, hardening, and forensics readiness, and also performs remediation.
There is a Prowler CLI (Command Line Interface) called Prowler Open Source and a SaaS service built on top of it called Prowler SaaS.

Source: GitHub - prowler-cloud/prowler

Prowler can perform checks based on security frameworks such as CIS, NIST, and PCI-DSS.
As summarized in the article Tips: Investigating security check tools that support CIS AWS Foundations Benchmark v3.0.0 (Security Hub, CFn Guard, Prowler) , Prowler can check a wider range of rules within security frameworks compared to Security Hub.

What does the official documentation say about custom checks?

Custom checks are described on the following page of the official documentation:
Miscellaneous - Prowler Documentation
Below is a translated excerpt, but the official documentation does not provide much detail on how to create custom checks. The Developer Guide, which is said to describe the detailed process for writing checks, actually contains information on how to develop Prowler itself and does not mention custom checks.

The custom checks folder must contain one subfolder per check, each subfolder must be named after the check and contain the following:

An empty init .py: This tells Python to treat this check folder as a package.
check_name.py: Contains the logic of the check.
check_name.metadata.json: Contains the metadata of the check.
The check name must start with the service name followed by an underscore (e.g., ec2_instance_public_ip).

For more details on how to write checks, refer to the “Developer Guide”.

Source: Miscellaneous - Prowler Documentation

(Summary first) File structure and contents of custom checks

When creating custom checks, there are two patterns for the file structure: “Using Prowler’s standard client” and “Using a custom client”.

Using Prowler’s standard client

When using Prowler’s standard client, you can create custom checks with the following file structure.

${parent_folder_name}/
└── ${AWS_service_name}_${check_name}/
    ├── __init__.py
    ├── ${AWS_service_name}_${check_name}.py
    └── ${AWS_service_name}_${check_name}.metadata.json

# Example file structure: Check if alternative contact information is registered for the account
custom-checks-folder/
∟ account_alternative_contact_information_is_registered/
　 ∟ __init__.py
　 ∟ account_alternative_contact_information_is_registered.py
　 ∟ account_alternative_contact_information_is_registered.metadata.json

${parent_folder_name} specifies an appropriate folder name. Create a folder named ${AWS_service_name}_${check_name} directly under the parent folder.
The parent folder can be placed in any nested structure. However, the folder specified as an argument must be the path of the parent folder containing the check folder.
Multiple ${AWS_service_name}_${check_name} folders can be created.
${AWS_service_name} must be one of the following. The detailed reason is explained later. (As of v3.12.1)

List of AWS service names

accessanalyzer
account
acm
apigateway
apigatewayv2
appstream
athena
autoscaling
awslambda
backup
cloudformation
cloudfront
cloudtrail
cloudwatch
codeartifact
codebuild
cognito
config
directoryservice
dlm
documentdb
drs
dynamodb
ec2
ecr
ecs
efs
eks
elasticache
elb
elbv2
emr
fms
glacier
globalaccelerator
glue
guardduty
iam
inspector2
kms
macie
neptune
networkfirewall
opensearch
organizations
rds
redshift
resourceexplorer2
route53
s3
sagemaker
secretsmanager
securityhub
shield
sns
sqs
ssm
ssmincidents
trustedadvisor
vpc
waf
wafv2
wellarchitected
workspaces

${check_name} specifies the name of the check in lower snake case.
Next, I will explain the content of the files.
The content of the files is created with reference to prowler/prowler/providers/aws/services at master · prowler-cloud/prowler .
__init__.py is an empty file. There is no need to write any processing.
${AWS_service_name}_${check_name}.py describes the logic of the check.
For example, to create a check to see if alternative contact information is registered for the account, it is described as follows.
There are several patterns for generating check reports, but the following example explains the pattern that returns the result of the check as “FAIL” or “PASS”.
The highlighted processing is basically necessary code. After generating Check_Report_AWS , set the values of region , resource_arn , resource_id , status , status_extended , and resource_tags .
For details, refer to Checks - Prowler Documentation .
To generate the report, use Prowler’s standard client account_client . account_client is defined in the Prowler GitHub repository in prowler/providers/aws/services/${AWS_service_name}/${AWS_service_name}_client.py and prowler/providers/aws/services/${AWS_service_name}/${AWS_service_name}_service.py . By looking at the content of the class that inherits AWSService or BaseModel in ${AWS_service_name}_service.py , you can check the variables that can be used with account_client .
If you want to use variables that are not defined here, you need to create your own client. The detailed method is explained later.

import json
from prowler.lib.logger import logger
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.account.account_client import account_client

class account_alternative_contact_information_is_registered(Check):
    def execute(self):
        desired_alternative_contact_count = account_client.audit_config.get("desired_alternative_contact_count", 1)
        findings = []
        report = Check_Report_AWS(self.metadata())

        # For each Prowler check we MUST fill the following
        # Check_Report_AWS fields:
        # - region
        # - resource_id
        # - resource_arn
        # - status
        # - status_extended
        # - resource_tags (optional)

        logger.debug(f"account_client.contact_names: {account_client.contact_names}")
        logger.debug(f"account_client.contact_phone_numbers: {account_client.contact_phone_numbers}")
        logger.debug(f"account_client.contact_emails: {account_client.contact_emails}")

        contact_emails = account_client.contact_emails
        contact_emails.discard(None)

        report = Check_Report_AWS(self.metadata())
        report.region = account_client.region
        report.resource_arn = account_client.audited_account_arn
        report.resource_id = account_client.audited_account

        if len(contact_emails) != desired_alternative_contact_count:
            report.status = "FAIL"
        else:
            report.status = "PASS"

        report.status_extended = json.dumps(
            {
                "desired_alternative_contact_count": desired_alternative_contact_count,
                "contact_emails_count": len(contact_emails),
                "contact_emails": list(contact_emails),
            },
        )

        findings.append(report)

        return findings

${AWS_service_name}_${check_name}.metadata.json describes the metadata of the check.
For the check to see if alternative contact information is registered for the account, it is described as follows.
The metadata keys must be included in the JSON, but some key values can be empty. In the example below, only the items that are likely to be required are specified.
The following value formats must be met.
Provider is aws
CheckID is ${AWS_service_name}_${check_name} (lower snake case)
ServiceName is ${AWS_service_name} (as defined in Prowler)
Severity is one of low , medium , high , critical

{
  "Provider": "aws",
  "CheckID": "account_alternative_contact_information_is_registered",
  "CheckTitle": "Account alternative contact information is registered",
  "CheckType": [
    "IAM"
  ],
  "ServiceName": "account",
  "SubServiceName": "",
  "ResourceIdTemplate": "arn:partition:access-recorder:region:account-id:recorder/resource-id",
  "Severity": "medium",
  "ResourceType": "Other",
  "Description": "",
  "Risk": "",
  "RelatedUrl": "",
  "Remediation": {
    "Code": {
      "CLI": "",
      "NativeIaC": "",
      "Other": "",
      "Terraform": ""
    },
    "Recommendation": {
      "Text": "",
      "Url": ""
    }
  },
  "Categories": [
    "custom-checks"
  ],
  "DependsOn": [],
  "RelatedTo": [],
  "Notes": ""
}

Using a custom client

When using a custom client, you can create custom checks with the following file structure. The file naming is the same as when using Prowler’s standard client.
The highlighted cst_*.py file names can be arbitrary, but it is recommended to use names that are easy to understand when loading.

${parent_folder_name}/
∟ ${AWS_service_name}_${check_name}
　 ∟ __init__.py
　 ∟ ${AWS_service_name}_${check_name}.py
　 ∟ ${AWS_service_name}_${check_name}.metadata.json
　 ∟ cst_${AWS_service_name}_client.py
　 ∟ cst_${AWS_service_name}_service.py

# Example file structure: Check if the Config delivery channel is enabled
custom-checks-folder/
∟ config_delivery_channel_enabled/
　 ∟ __init__.py
　 ∟ config_delivery_channel_enabled.py
　 ∟ config_delivery_channel_enabled.metadata.json
　 ∟ cst_config_client.py
　 ∟ cst_config_service.py

Next, I will explain the content of the files.
__init__.py and ${AWS_service_name}_${check_name}.metadata.json are the same as when using Prowler’s standard client.
In ${AWS_service_name}_${check_name}.py , you need to change the way the client is loaded. Here, I will explain the example of creating a check to see if the Config delivery channel is enabled.
The key point is lines 3-7 below. Since files placed under the ${AWS_service_name}_${check_name}/ directory will fail to load as they are, add the directory where the file exists with sys.path.append .

from prowler.lib.logger import logger
from prowler.lib.check.models import Check, Check_Report_AWS
import os
import sys

sys.path.append(os.path.join(os.path.dirname(__file__)))
from cst_config_client import config_client  # noqa: E402

class config_delivery_channel_enabled(Check):
    def execute(self):
        findings = []
        report = Check_Report_AWS(self.metadata())

// Omitted

The content of the cst_config_client.py file is as follows. Here, I will explain the example of defining a custom client for Config.
The content is simple, and it instantiates the client defined in cst_config_service.py .

from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
from cst_config_service import Config

config_client = Config(current_audit_info)

The content of the cst_config_service.py file is as follows. Here, I will explain the example of defining the Config delivery recorder.
The processing content is developed based on the content of Prowler’s standard client.
The Config(AWSService) class defines the processing to be executed when the client is initialized.
The __describe_delivery_channels__ function executes regional_client.describe_delivery_channels() to obtain the Config delivery channel.
It generates an instance of the DeliveryChannel class and adds it to the self.delivery_channels list.
The self.delivery_channels list contains information about the Config delivery channel. The check results are generated based on this information.
For processing that checks security-related resources, it is necessary to define the processing for when AWS resources do not exist. Conversely, for processing that checks multiple AWS resources such as EC2 instances and S3 buckets, it is necessary to execute List or Describe APIs to obtain information about all resources.
The DeliveryChannel(BaseModel) class defines the structure of the DeliveryChannel based on CloudFormation and API documentation. Parameters that are not mandatory as information are specified as Optional . When outputting results even if AWS resources do not exist, parameters other than name and region may be specified as None , so Optional is specified.
AWS::Config::DeliveryChannel - AWS CloudFormation
DescribeDeliveryChannels - AWS Config

from typing import Optional

from pydantic import BaseModel

from prowler.lib.logger import logger
from prowler.lib.scan_filters.scan_filters import is_resource_filtered
from prowler.providers.aws.lib.service.service import AWSService

class Config(AWSService):
    def __init__(self, audit_info):
        super().__init__(__class__.__name__, audit_info)
        self.delivery_channels = []
        self.__threading_call__(self.__describe_delivery_channels__)

    def __describe_delivery_channels__(self, regional_client):
        logger.info("Config - Listing Delivery Channels...")
        try:
            delivery_channel_count = 0
            delivery_channels = regional_client.describe_delivery_channels()[
                "DeliveryChannels"
            ]
            for delivery_channel in delivery_channels:
                if not self.audit_resources or (
                    is_resource_filtered(delivery_channel["name"], self.audit_resources)
                ):
                    self.delivery_channels.append(
                        DeliveryChannel(
                            name=delivery_channel["name"],
                            s3_bucket_name=delivery_channel["s3BucketName"],
                            s3_key_prefix=delivery_channel["s3KeyPrefix"],
                            sns_topic_arn=delivery_channel["snsTopicARN"],
                            region=regional_client.region,
                        )
                    )
                    delivery_channel_count += 1

            # No delivery channels in region
            if delivery_channel_count == 0:
                self.delivery_channels.append(
                    DeliveryChannel(
                        name=self.audited_account,
                        s3_bucket_name=None,
                        s3_key_prefix=None,
                        sns_topic_arn=None,
                        region=regional_client.region,
                    )
                )
        except Exception as error:
            logger.error(
                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
            )

class DeliveryChannel(BaseModel):
    name: str
    s3_bucket_name: Optional[str]
    s3_key_prefix: Optional[str]
    sns_topic_arn: Optional[str]
    region: str

Debugging method during custom check development

The quickest way to debug during custom check development is to execute it from Prowler.
By specifying only the check under development with the --checks option as shown below, you can shorten the time required for debugging.
If you specify --log-level , you can check the logs output by the logger.

prowler aws \
--checks-folder ./custom-checks-folder/ \
--checks \
account_alternative_contact_information_is_registered \
--log-level ERROR \
-M csv \
--output-directory ./output

When outputting logs at the DEBUG level, it is recommended to output the logs to a file as it is difficult to check in the terminal.

prowler aws \
--checks-folder ./custom-checks-folder/ \
--checks \
account_alternative_contact_information_is_registered \
--log-level DEBUG \
-M csv \
--output-directory ./output > debug.log 2>&1

Reading Prowler’s Source Code (as of v1.12.1)

Entry Point - prowler()

The entry point for Prowler is prowler() in prowler/__main__.py .
The arguments for running Prowler are read in the following code. If a folder for custom checks is specified with -x or --checks-folder , the value is set to checks_folder .

def prowler():
    # Parse Arguments
    parser = ProwlerArgumentParser()
    args = parser.parse()

    # Save Arguments
    provider = args.provider
    checks = args.checks
    excluded_checks = args.excluded_checks
    excluded_services = args.excluded_services
    services = args.services
    categories = args.categories
    checks_file = args.checks_file
    checks_folder = args.checks_folder
    severities = args.severity
    compliance_framework = args.compliance
    custom_checks_metadata_file = args.custom_checks_metadata_file

If the argument is specified, the parse_checks_from_folder(audit_info, checks_folder, provider) function is called. This function reads the folder for custom checks and copies the check files in the loaded folder to a specific path in Prowler. Since the behavior of this function is a key point that determines the file structure and contents of custom checks, it will be explained in detail later.

    # Import custom checks from folder
    if checks_folder:
        parse_checks_from_folder(audit_info, checks_folder, provider)

After the files are copied, the checks_to_execute list is generated based on the arguments specified at runtime, narrowing down the checks to be executed.

    # Exclude checks if -e/--excluded-checks
    if excluded_checks:
        checks_to_execute = exclude_checks_to_run(checks_to_execute, excluded_checks)

    # Exclude services if --excluded-services
    if excluded_services:
        checks_to_execute = exclude_services_to_run(
            checks_to_execute, excluded_services, provider
        )

    # Once the audit_info is set and we have the eventual checks based on the resource identifier,
    # it is time to check what Prowler's checks are going to be executed
    if audit_info.audit_resources:
        checks_from_resources = set_provider_execution_parameters(provider, audit_info)
        checks_to_execute = checks_to_execute.intersection(checks_from_resources)

    # Sort final check list
    checks_to_execute = sorted(checks_to_execute)

Based on the generated checks_to_execute list, the execute_checks(checks_to_execute,provider,audit_info,audit_output_options,custom_checks_metadata) function is called. This function executes the checks included in the checks_to_execute list and stores the results in the findings list. This function is also a point where you may stumble when creating your own client, so it will be explained later.
The subsequent processing outputs the check results based on the findings list, but the details are omitted here.

    # Execute checks
    findings = []
    if len(checks_to_execute):
        findings = execute_checks(
            checks_to_execute,
            provider,
            audit_info,
            audit_output_options,
            custom_checks_metadata,
        )
    else:
        logger.error(
            "There are no checks to execute. Please, check your input arguments"
        )

Finally, if a folder for custom checks was specified, the remove_custom_checks_module(checks_folder, provider) function is called. This function deletes the folder for custom checks.

    # If custom checks were passed, remove the modules
    if checks_folder:
        remove_custom_checks_module(checks_folder, provider)

Parsing Checks from Folder - parse_checks_from_folder(audit_info, checks_folder, provider)

The parse_checks_from_folder(audit_info, checks_folder, provider) function is defined in prowler/lib/check/check.py .
In lines 131-133, it determines whether checks_folder is an S3 URI. If it is an S3 URI, it checks the objects under the specified URI prefix and downloads them.
In lines 146-151, it traverses the folders under checks_folder and identifies the check folders. If a check folder is identified, it copies the check folder under checks_folder to Prowler’s path {prowler_dir[0]}/providers/{provider}/services/{check_service}/{check.name} . For example, if the structure is custom-checks-folder/account_alternative_contact_information_is_registered/ , it will be copied to {prowler_dir[0]}/providers/aws/services/account/account_alternative_contact_information_is_registered/ .
As in check_service = check.name.split("_")[0] , the first word of the snake_case name becomes the AWS service name. For this reason, as mentioned earlier, you must use AWS service names already defined in Prowler.
Each AWS service client is defined under {prowler_dir[0]}/providers/{provider}/services/{check_service}/ , but since only the check folder is copied, even if you place a custom client under the parent folder, the file will not be copied.
If you want to create a custom client, you need to include it in the check folder.

def parse_checks_from_folder(audit_info, input_folder: str, provider: str) -> int:
    try:
        imported_checks = 0
        # Check if input folder is a S3 URI
        if provider == "aws" and re.search(
            "^s3://([^/]+)/(.*?([^/]+))/$", input_folder
        ):
            bucket = input_folder.split("/")[2]
            key = ("/").join(input_folder.split("/")[3:])
            s3_resource = audit_info.audit_session.resource("s3")
            bucket = s3_resource.Bucket(bucket)
            for obj in bucket.objects.filter(Prefix=key):
                if not os.path.exists(os.path.dirname(obj.key)):
                    os.makedirs(os.path.dirname(obj.key))
                if obj.key[-1] == "/":
                    continue
                bucket.download_file(obj.key, obj.key)
            input_folder = key
        # Import custom checks by moving the checks folders to the corresponding services
        with os.scandir(input_folder) as checks:
            for check in checks:
                if check.is_dir():
                    check_module = input_folder + "/" + check.name
                    # Copy checks to specific provider/service folder
                    check_service = check.name.split("_")[0]
                    prowler_dir = prowler.__path__
                    prowler_module = f"{prowler_dir[0]}/providers/{provider}/services/{check_service}/{check.name}"
                    if os.path.exists(prowler_module):
                        shutil.rmtree(prowler_module)
                    shutil.copytree(check_module, prowler_module)
                    imported_checks += 1
        return imported_checks
    except Exception as error:
        logger.critical(
            f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
        )
        sys.exit(1)

Executing Checks - execute_checks(checks_to_execute,provider,audit_info,audit_output_options,custom_checks_metadata)

The execute_checks(checks_to_execute,provider,audit_info,audit_output_options,custom_checks_metadata) function is defined in prowler/lib/check/check.py .
There is also processing for when the --only-logs option is specified, but the following describes the default execution process.
In lines 510-519, the checks included in the checks_to_execute list are executed, and the results are stored in the all_findings list.
When loading the check file using importlib.import_module inside the execute function, if a ModuleNotFoundError occurs, an error is output with logger.error . This ModuleNotFoundError can occur both when the check name specified with the --checks option cannot be found and when a module cannot be loaded properly within the Python script for the check. To determine which is the cause, you need to output debug logs in your custom Python script and investigate.

        # Default execution
        checks_num = len(checks_to_execute)
        plural_string = "checks"
        singular_string = "check"

        check_noun = plural_string if checks_num > 1 else singular_string
        print(
            f"{Style.BRIGHT}Executing {checks_num} {check_noun}, please wait...{Style.RESET_ALL}\n"
        )
        with alive_bar(
            total=len(checks_to_execute),
            ctrl_c=False,
            bar="blocks",
            spinner="classic",
            stats=False,
            enrich_print=False,
        ) as bar:
            for check_name in checks_to_execute:
                # Recover service from check name
                service = check_name.split("_")[0]
                bar.title = (
                    f"-> Scanning {orange_color}{service}{Style.RESET_ALL} service"
                )
                try:
                    check_findings = execute(
                        service,
                        check_name,
                        provider,
                        audit_output_options,
                        audit_info,
                        services_executed,
                        checks_executed,
                        custom_checks_metadata,
                    )
                    all_findings.extend(check_findings)

                # If check does not exists in the provider or is from another provider
                except ModuleNotFoundError:
                    logger.error(
                        f"Check '{check_name}' was not found for the {provider.upper()} provider"
                    )
                except Exception as error:
                    logger.error(
                        f"{check_name} - {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
                    )
                bar()
            bar.title = f"-> {Fore.GREEN}Scan completed!{Style.RESET_ALL}"
    return all_findings

Resumen

This article summarizes the information necessary for creating custom checks in Prowler.
While Prowler provides many standard checks, creating custom checks allows you to run checks tailored to your organization’s security policies.
If you do not need to run checks periodically/continuously like Config rules, or if you want to reduce the cost associated with Config rules, you can use Prowler to run checks at a lower cost.
I hope this article will be helpful when creating custom checks in Prowler.