【提示/解決方法】如何使用Prowler建立自定義檢查

📅 發布日: 2025/02/11, 更新日: 2025/02/11 ⏱️ 預計閱讀時間約12分鐘 (4,723字)

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

# 【提示/解決方法】如何使用Prowler建立自定義檢查
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 中建立自定義檢查時有所幫助。