Kanji
・클라우드 엔지니어 / 프리랜서 ・1993년생 ・에히메현 출신 / 도쿄도 시부야구 거주 ・AWS 경력 5년 프로필 상세
목차
<인용문> Prowler는 보안 모범 사례 평가, 감사, 사고 대응, 지속적인 모니터링, 강화 및 포렌식 준비를 수행하고 문제 해결도 수행하는 AWS, Azure, Google Cloud 및 Kubernetes용 오픈 소스 보안 도구입니다. Prowler 오픈 소스라고 하는 Prowler CLI(명령줄 인터페이스)와 Prowler SaaS라고 하는 그 위에 구축된 SaaS 서비스가 있습니다.
사용자 정의 검사는 공식 문서의 다음 페이지에 설명되어 있습니다.
기타 - Prowler 문서
아래는 번역된 발췌문이지만 공식 문서에는 사용자 정의 검사를 만드는 방법에 대한 자세한 내용이 제공되지 않습니다.Check 작성을 위한 세부 프로세스를 설명한다고 하는 개발자 가이드에는 실제로 Prowler 자체 개발 방법에 대한 정보가 포함되어 있으며 Custom Check에 대한 언급은 없습니다.
<인용문> 사용자 정의 검사 폴더에는 검사당 하나의 하위 폴더가 포함되어야 하며, 각 하위 폴더는 검사 이름을 따서 명명되어야 하며 다음을 포함해야 합니다.
빈 init .py: Python에게 이 검사 폴더를 패키지로 처리하도록 지시합니다. check_name.py: 검사 논리가 포함되어 있습니다. check_name.metadata.json: 수표의 메타데이터를 포함합니다. 확인 이름은 서비스 이름으로 시작하고 그 뒤에 밑줄이 와야 합니다(예: ec2_instance_public_ip).
수표 작성 방법에 대한 자세한 내용은 “개발자 가이드”를 참조하세요.
${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} 이라는 폴더를 생성합니다.
${parent_folder_name}
${AWS_service_name}_${check_name}
상위 폴더는 중첩된 구조에 배치될 수 있습니다.단, 인수로 지정된 폴더는 검사 폴더가 포함된 상위 폴더의 경로여야 합니다.
${AWS_service_name}_${check_name} 폴더를 여러 개 생성할 수 있습니다.
${AWS_service_name} 은 다음 중 하나여야 합니다.자세한 이유는 나중에 설명하겠습니다.(v3.12.1 기준)
${AWS_service_name}
${check_name} 은 소문자로 된 수표 이름을 지정합니다.
${check_name}
다음으로 파일 내용에 대해 설명하겠습니다.
파일 내용은 prowler/prowler/providers/aws/services at master · prowler-cloud/prowler 를 참조하여 생성됩니다.
__init__.py 는 빈 파일입니다.처리를 작성할 필요가 없습니다.
__init__.py
${AWS_service_name}_${check_name}.py 는 확인 논리를 설명합니다.
${AWS_service_name}_${check_name}.py
예를 들어 해당 계정에 대체 연락처가 등록되어 있는지 확인하기 위한 검사를 생성하려면 다음과 같이 설명합니다.
점검 보고서를 생성하는 패턴에는 여러 가지가 있지만, 다음 예에서는 점검 결과를 “FAIL” 또는 “PASS”로 반환하는 패턴을 설명합니다.
하이라이트 처리는 기본적으로 필요한 코드입니다. Check_Report_AWS 를 생성한 후 region , resource_arn , resource_id , status , status_extended 및 resource_tags 값을 설정합니다.
Check_Report_AWS
region
resource_arn
resource_id
status
status_extended
resource_tags
자세한 내용은 검사 - Prowler 문서 를 참조하세요.
보고서를 생성하려면 Prowler의 표준 클라이언트 account_client 를 사용하세요. account_client 는 prowler/providers/aws/services/${AWS_service_name}/${AWS_service_name}_client.py 및 prowler/providers/aws/services/${AWS_service_name}/${AWS_service_name}_service.py 의 Prowler GitHub 리포지토리에 정의되어 있습니다. ${AWS_service_name}_service.py 에서 AWSService 또는 BaseModel을 상속하는 클래스의 내용을 보면 account_client 와 함께 사용할 수 있는 변수를 확인할 수 있습니다.
account_client
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
여기에 정의되지 않은 변수를 사용하려면 자체 클라이언트를 생성해야 합니다.자세한 방법은 나중에 설명하겠습니다.
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
공급자
aws
CheckID
ServiceName
{ "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": "" }
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
${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
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)
Config(AWSService)
__describe_delivery_channels__
regional_client.describe_delivery_channels()
DeliveryChannel
self.delivery_channels
DeliveryChannel(BaseModel)
Optional
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
--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
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/__main__.py
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)
# Import custom checks from folder if checks_folder: parse_checks_from_folder(audit_info, checks_folder, provider)
# 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)
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)
prowler/lib/check/check.py
{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]
{prowler_dir[0]}/providers/{provider}/services/{check_service}/
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)
--only-logs
all_findings
execute
importlib.import_module
ModuleNotFoundError
logger.error
# 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