【提示/解决方法】如何使用Prowler创建自定义检查

📅 发布日: 2025/02/11, 更新日: 2025/02/11 ⏱️ 预计阅读时间约12分钟 (4,714字)

Prowler是用于检查AWS、Azure和GCP帐户安全设置的工具。
除了默认提供的检查之外，您还可以创建自定义检查。
由于官方文档中没有详细说明创建自定义检查的方法，因此我们分析了源代码以确认创建自定义检查的方法。本文介绍如何使用Prowler创建自定义检查。

什么是徘徊者？
官方文档对自定义检查有何说明？
（先总结）自定义检查的文件结构和内容
1. 使用 Prowler 的标准客户端
2. 使用自定义客户端
自定义检查开发时的调试方法
阅读 Prowler 的源代码（从 v1.12.1 开始）
总结

什么是徘徊者？

如 GitHub - prowler-cloud/prowler 中所述：

<块引用>
Prowler 是一款适用于 AWS、Azure、Google Cloud 和 Kubernetes 的开源安全工具，可执行安全最佳实践评估、审计、事件响应、持续监控、强化和取证准备，并执行修复。
有一个名为 Prowler Open Source 的 Prowler CLI（命令行界面）和一个构建在其之上的名为 Prowler SaaS 的 SaaS 服务。

来源：GitHub - prowler-cloud/prowler

Prowler 可以基于 CIS、NIST 和 PCI-DSS 等安全框架执行检查。
正如文章提示：调查支持 CIS AWS Foundations Benchmark v3.0.0 的安全检查工具（Security Hub、CFn Guard、Prowler）中所总结的，Prowler 可以检查安全范围内更广泛的规则与 Security Hub 相比的框架。

官方文档对自定义检查有何说明？

官方文档的下页描述了自定义检查：
其他 - Prowler 文档
下面是翻译后的摘录，但官方文档没有提供有关如何创建自定义检查的详细信息。《开发者指南》据说描述了编写检查的详细过程，实际上包含了如何开发 Prowler 本身的信息，并没有提及自定义检查。

<块引用>
自定义检查文件夹必须包含每张检查一个子文件夹，每个子文件夹必须以检查命名并包含以下内容：

空的 init .py：这告诉 Python 将此检查文件夹视为一个包。
check_name.py：包含检查的逻辑。
check_name.metadata.json：包含检查的元数据。
检查名称必须以服务名称开头，后跟下划线（例如 ec2_instance_public_ip）。

有关如何编写支票的更多详细信息，请参阅“开发人员指南”。

来源：其他 - Prowler 文档

（先总结）自定义检查的文件结构和内容

创建自定义检查时，文件结构有两种模式：“使用 Prowler 的标准客户端”和“使用自定义客户端”。

使用 Prowler 的标准客户端

使用 Prowler 的标准客户端时，您可以使用以下文件结构创建自定义检查。

${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} 指定适当的文件夹名称。直接在父文件夹下创建一个名为“ ${AWS_service_name}_$ {check_name}”的文件夹。
父文件夹可以放置在任何嵌套结构中。但是，指定为参数的文件夹必须是包含检查文件夹的父文件夹的路径。
可以创建多个“ ${AWS_service_name}_$ {check_name}”文件夹。
${AWS_service_name} 必须是以下之一。详细原因稍后解释。（从 v3.12.1 开始）

<详情>

AWS 服务名称列表

访问分析器
帐户
acm
API网关
-apigatewayv2
应用程序流
雅典娜
自动缩放
-awslambda
备份
云信息
云前
云迹
云观察
代码神器
代码构建
知道
配置
目录服务
dlm
文档数据库
博士
dynamodb
एक2
电子病历
ECS
EFS
例如
弹性疼痛
埃尔布
elbv2
电子病历
调频管理系统
冰川
全球加速器
胶水
警卫职责
我是
检查员2
公里数
干得好
海王星
网络防火墙
开放搜索
组织
RD
红移
资源浏览器2
53号公路
c3
贤者
秘密管理器
安全中心
盾
社交网络
sqs
ssm
ssmincidents
值得信赖的顾问
专有网络
瓦夫
WAFV2
架构良好
工作空间

${check_name} 以小写蛇形形式指定检查的名称。
接下来，我将解释文件的内容。
文件的内容是参考 prowler/prowler/providers/aws/services at master · prowler-cloud/prowler 创建的。
__init__.py 是一个空文件。不需要写任何处理。
${AWS_service_name}_${check_name}.py 描述检查的逻辑。
例如，要创建检查以查看该帐户是否注册了替代联系信息，描述如下。
生成检查报告的模式有多种，但以下示例解释了将检查结果返回为“FAIL”或“PASS”的模式。
突出显示的处理基本上是必要的代码。生成“Check_Report_AWS”后，设置“region”、“resource_arn”、“resource_id”、“status”、“status_extended”和“resource_tags”的值。
有关详细信息，请参阅检查 - Prowler 文档。
要生成报告，请使用 Prowler 的标准客户端“account_client”。 account_client 在 Prowler GitHub 存储库的 prowler/providers/aws/services/${AWS_service_name}/${AWS_service_name}_client.py 和 prowler/providers/aws/services/${AWS_service_name}/${AWS_service_name}_service.py 中定义。通过查看 ${AWS_service_name}_service.py 中继承AWSService或BaseModel的类的内容，可以检查可以与 account_client 一起使用的变量。
如果您想使用此处未定义的变量，您需要创建自己的客户端。详细方法稍后解释。

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 描述检查的元数据。
对于检查该帐户是否注册了备用联系信息，描述如下。
元数据键必须包含在 JSON 中，但某些键值可以为空。在下面的示例中，仅指定可能需要的项目。
必须满足以下值格式。
Provider 是 aws
CheckID 是 ${AWS_service_name}_${check_name} （小写蛇形字母）
ServiceName 是 ${AWS_service_name} （如 Prowler 中定义）
“严重性”是“低”、“中”、“高”、“严重”之一

{
  "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": ""
}

使用自定义客户端

使用自定义客户端时，您可以使用以下文件结构创建自定义检查。文件命名与使用 Prowler 标准客户端时相同。
突出显示的 cst_*.py 文件名可以是任意的，但建议加载时使用易于理解的名称。

${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

接下来，我将解释文件的内容。
__init__.py 和 ${AWS_service_name}_${check_name}.metadata.json 与使用 Prowler 标准客户端时相同。
在 ${AWS_service_name}_${check_name}.py 中，您需要更改客户端的加载方式。在这里，我将解释创建一个检查以查看 Config 传递通道是否已启用的示例。
关键点是下面的第3-7行。由于放置在 ${AWS_service_name}_${check_name}/ 目录下的文件将无法按原样加载，因此请使用 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

cst_config_client.py 文件的内容如下。在这里，我将解释为Config定义自定义客户端的示例。
内容很简单，实例化了 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)

cst_config_service.py 文件的内容如下。这里我以定义Config交付记录器为例进行说明。
处理内容是根据Prowler标准客户端的内容开发的。
Config(AWSService) 类定义客户端初始化时要执行的处理。
__describe_delivery_channels__ 函数执行 regional_client.describe_delivery_channels() 来获取Config传递通道。
它生成 DeliveryChannel 类的实例并将其添加到 self.delivery_channels 列表中。
self.delivery_channels 列表包含有关配置传递通道的信息。检查结果是根据该信息生成的。
对于检查安全相关资源的处理，需要定义AWS资源不存在时的处理。相反，对于检查多个AWS资源（例如EC2实例和S3存储桶）的处理，需要执行List或Describe API来获取有关所有资源的信息。
DeliveryChannel(BaseModel) 类根据 CloudFormation 和 API 文档定义了 DeliveryChannel 的结构。作为信息不强制的参数被指定为“可选”。即使AWS资源不存在，在输出结果时，除了名称和区域之外的参数也可能被指定为“None”，因此指定“Optional”。
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

自定义检查开发时的调试方法

在自定义检查开发期间调试的最快方法是从 Prowler 执行它。
通过使用如下所示的“–checks”选项仅指定正在开发的检查，可以缩短调试所需的时间。
如果指定 --log-level ，您可以检查记录器输出的日志。

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

当在DEBUG级别输出日志时，建议将日志输出到文件，因为在终端中查看比较困难。

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

阅读 Prowler 的源代码（从 v1.12.1 开始）

入口点 - prowler()

Prowler 的入口点是 prowler/__main__.py 中的 prowler() 。
在以下代码中读取运行 Prowler 的参数。如果使用“-x”或“–checks-folder”指定了自定义检查的文件夹，则该值将设置为“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

如果指定了参数，则调用“parse_checks_from_folder(audit_info,checks_folder,provider)”函数。该函数读取自定义检查的文件夹，并将加载的文件夹中的检查文件复制到Prowler中的特定路径。由于该函数的行为是决定自定义检查的文件结构和内容的关键点，因此稍后会详细解释。

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

复制文件后，会根据运行时指定的参数生成“checks_to_execute”列表，缩小要执行的检查范围。

    # 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)

根据生成的“checks_to_execute”列表，调用“execute_checks(checks_to_execute,provider,audit_info,audit_output_options,custom_checks_metadata)”函数。该函数执行“checks_to_execute”列表中包含的检查，并将结果存储在“findings”列表中。这个功能也是你在创建自己的客户端时可能会绊倒的点，所以稍后会解释。
后续处理根据‘findings’列表输出检查结果，此处省略详细内容。

    # 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"
        )

最后，如果指定了用于自定义检查的文件夹，则会调用“remove_custom_checks_module(checks_folder,provider)”函数。此功能删除用于自定义检查的文件夹。

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

解析文件夹中的检查 - parse_checks_from_folder(audit_info,checks_folder,provider)

-“parse_checks_from_folder(audit_info,checks_folder,provider)”函数在“prowler/lib/check/check.py”中定义。
- 在第 131-133 行中，它确定 checks_folder 是否是 S3 URI。如果是 S3 URI，则检查指定 URI 前缀下的对象并下载它们。
- 第146-151行，遍历 checks_folder 下的文件夹并识别检查文件夹。如果识别到检查文件夹，则会将 checks_folder 下的检查文件夹复制到 Prowler 的路径 {prowler_dir[0]}/providers/{provider}/services/{check_service}/{check.name} 。例如，如果结构为“custom-checks-folder/account_alternative_contact_information_is_registered/”，它将被复制到“{prowler_dir[0]}/providers/aws/services/account/account_alternative_contact_information_is_registered/”。
- 如 check_service = check.name.split("_")[0] 中，snake_case 名称的第一个单词成为 AWS 服务名称。因此，如前所述， 您必须使用 Prowler 中已定义的 AWS 服务名称。
- 每个AWS服务客户端都在 {prowler_dir[0]}/providers/{provider}/services/{check_service}/ 下定义，但由于仅复制check文件夹，即使您将自定义客户端放置在父文件夹下，文件也不会被复制。
- 如果您想创建自定义客户端，则需要将其包含在检查文件夹中。_

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)

执行检查 -execute_checks(checks_to_execute,provider,audit_info,audit_output_options,custom_checks_metadata)

execute_checks(checks_to_execute,provider,audit_info,audit_output_options,custom_checks_metadata) 函数在 prowler/lib/check/check.py 中定义。
也有指定 --only-logs 选项时的处理，但下面描述默认的执行过程。
第510-519行，执行“checks_to_execute”列表中包含的检查，并将结果存储在“all_findings”列表中。
在“execute”函数中使用“importlib.import_module”加载检查文件时，如果发生“ModuleNotFoundError”，则会输出“logger.error”错误。_当无法找到使用“–checks”选项指定的检查名称以及无法在检查的 Python 脚本中正确加载模块时，都会发生此“ModuleNotFoundError”。_要确定原因，您需要在自定义 Python 脚本中输出调试日志并进行调查。

        # 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

总结

本文总结了在 Prowler 中创建自定义检查所需的信息。
虽然 Prowler 提供了许多标准检查，但创建自定义检查允许您运行适合您组织的安全策略的检查。
如果您不需要像Config规则那样定期/连续运行检查，或者如果您想降低与Config规则相关的成本，则可以使用Prowler以较低的成本运行检查。
我希望这篇文章对在 Prowler 中创建自定义检查时有所帮助。