tessl/pypi-prowler
Open source cloud security assessment tool for AWS, Azure, GCP, and Kubernetes with hundreds of compliance checks.
—
Pending
check-management.mddocs/
Check Management and Execution
Core functionality for loading, filtering, and executing security checks across cloud providers. This module provides the foundation for Prowler's security assessment capabilities, supporting custom checks, service filtering, compliance framework mapping, and parallel execution.
Capabilities
Check Execution
Primary function for executing security checks against cloud resources and collecting findings.
def execute_checks(
checks: list,
provider: Provider,
) -> list[Finding]:
"""
Execute security checks and return findings.
Coordinates the execution of loaded security checks against the configured
provider, collecting results and generating Finding objects for each check.
Handles parallel execution, exception management, and finding generation.
Parameters:
- checks: list - List of check modules/functions to execute
- provider: Provider - Provider instance (AWS, Azure, GCP, etc.)
Returns:
list[Finding]: List of Finding objects representing check results with status,
metadata, resource information, and compliance mappings
Raises:
Exception: On critical check execution errors that cannot be handled
"""Check Loading and Filtering
Load checks based on provider capabilities and filtering criteria.
def load_checks_to_execute(
provider: Provider,
check_list: list = None,
service_list: list = None,
severities: list = None,
compliance_frameworks: list = None,
categories: set = None,
checks_file: str = None,
checks_folder: str = None,
) -> list:
"""
Load checks based on provider and filtering criteria.
Discovers available checks for the provider and applies various filters
to determine which checks should be executed. This function is imported
from prowler.lib.check.checks_loader module.
Parameters:
- provider: Provider - Provider instance to load checks for
- check_list: list = None - Optional list of specific check IDs to run
- service_list: list = None - Optional list of services to include
- severities: list = None - Optional list of severity levels to filter by
- compliance_frameworks: list = None - Optional list of compliance frameworks
- categories: set = None - Optional set of check categories to include
- checks_file: str = None - Path to file containing check list
- checks_folder: str = None - Path to folder containing custom checks
Returns:
list: List of check modules ready for execution
Raises:
Exception: On check loading or filtering errors
"""Service and Check Exclusion
Functions for excluding specific checks and services from execution.
def exclude_checks_to_run(checks_to_execute: set, excluded_checks: list) -> set:
"""
Exclude specific checks from execution set.
Parameters:
- checks_to_execute: set - Set of checks to execute
- excluded_checks: list - List of check IDs to exclude
Returns:
set: Filtered set of checks with exclusions removed
"""
def exclude_services_to_run(
checks_to_execute: set,
excluded_services: list,
bulk_checks_metadata: dict,
) -> set:
"""
Exclude checks from specific services.
Parameters:
- checks_to_execute: set - Set of checks to execute
- excluded_services: list - List of service names to exclude
- bulk_checks_metadata: dict - Bulk check metadata for service mapping
Returns:
set: Filtered set of checks with service exclusions applied
"""
def parse_checks_from_file(input_file: str, provider: str) -> set:
"""
Parse checks from input file.
Parameters:
- input_file: str - Path to file containing check list
- provider: str - Provider name for validation
Returns:
set: Set of valid check names from the file
"""
def parse_checks_from_folder(provider, input_folder: str) -> set:
"""
Parse custom checks from folder.
Parameters:
- provider: Provider instance
- input_folder: str - Path to folder containing custom checks
Returns:
set: Set of custom check names discovered in the folder
"""
def list_services(provider: str) -> set:
"""
List available services for a provider.
Parameters:
- provider: str - Provider name
Returns:
set: Set of available service names
"""
def list_categories(bulk_checks_metadata: dict) -> set:
"""
List available check categories.
Parameters:
- bulk_checks_metadata: dict - Bulk check metadata
Returns:
set: Set of available category names
"""
def print_checks(
check_list: list,
bulk_checks_metadata: dict,
output_format: str = "table",
) -> None:
"""
Print checks in specified format.
Parameters:
- check_list: list - List of checks to print
- bulk_checks_metadata: dict - Check metadata for details
- output_format: str - Output format ("table", "json", etc.)
Returns:
None (prints to stdout)
"""
def print_compliance_frameworks(
bulk_compliance_frameworks: dict,
provider: str = None,
) -> None:
"""
Print available compliance frameworks.
Parameters:
- bulk_compliance_frameworks: dict - Bulk compliance framework data
- provider: str = None - Optional provider filter
Returns:
None (prints to stdout)
"""
def run_fixer(check_findings: list) -> int:
"""
Run automatic remediation for findings.
Parameters:
- check_findings: list - List of findings to attempt remediation
Returns:
int: Number of successfully remediated findings
"""
Filtered list of checks with exclusions applied
"""
def exclude_services_to_run(checks: list, excluded_services: list) -> list:
"""
Exclude entire services from check execution.
Parameters:
- checks: List of available checks
- excluded_services: List of service names to exclude
Returns:
Filtered list of checks with service exclusions applied
"""Check Discovery and Listing
Functions for discovering and listing available checks, services, and categories.
def list_checks_json(provider: str, check_list: list = None) -> dict:
"""
Export available checks as JSON for programmatic access.
Parameters:
- provider: Provider name (aws, azure, gcp, etc.)
- check_list: Optional list to filter specific checks
Returns:
Dictionary containing check metadata organized by service
"""
def list_services(provider: str) -> list[str]:
"""
List available services for a provider.
Parameters:
- provider: Provider name
Returns:
List of available service names
"""
def list_categories(provider: str) -> list[str]:
"""
List available check categories for a provider.
Parameters:
- provider: Provider name
Returns:
List of available category names
"""
def list_fixers(provider: str) -> list[str]:
"""
List available automatic fixers for a provider.
Parameters:
- provider: Provider name
Returns:
List of fixer names for checks that support remediation
"""Check Information Display
Functions for displaying check information to users.
def print_checks(
provider: str,
check_list: list = None,
service_list: list = None
):
"""
Print available checks with metadata.
Parameters:
- provider: Provider name
- check_list: Optional list to filter specific checks
- service_list: Optional list to filter by services
Returns:
None (prints to stdout)
"""
def print_services(provider: str):
"""
Print available services for a provider.
Parameters:
- provider: Provider name
Returns:
None (prints to stdout)
"""
def print_categories(provider: str):
"""
Print available check categories.
Parameters:
- provider: Provider name
Returns:
None (prints to stdout)
"""
def print_fixers(provider: str):
"""
Print available fixers with their associated checks.
Parameters:
- provider: Provider name
Returns:
None (prints to stdout)
"""Custom Check Management
Functions for loading and managing custom security checks.
def parse_checks_from_folder(custom_checks_folder: str) -> list:
"""
Load custom checks from a specified folder.
Scans the folder for Python modules containing security checks
and loads them for execution alongside built-in checks.
Parameters:
- custom_checks_folder: Path to folder containing custom check modules
Returns:
List of loaded custom check modules
Raises:
ProwlerException: On folder access or module loading errors
"""
def remove_custom_checks_module():
"""
Remove custom checks module from system to prevent conflicts.
Cleans up custom check modules after execution to ensure
they don't interfere with subsequent runs.
Returns:
None
"""Compliance Framework Integration
Functions for integrating compliance frameworks with check execution.
def print_compliance_frameworks(provider: str = None):
"""
Print available compliance frameworks.
Parameters:
- provider: Optional provider name to filter frameworks
Returns:
None (prints to stdout)
"""
def print_compliance_requirements(
provider: str,
compliance_framework: str
):
"""
Print requirements for a specific compliance framework.
Parameters:
- provider: Provider name
- compliance_framework: Framework name (e.g., 'cis_1.5_aws')
Returns:
None (prints to stdout)
"""Automatic Remediation
Execute automatic fixes for failed security checks where available.
def run_fixer(
findings: list[Finding],
provider: Provider,
fixer_list: list = None
) -> dict:
"""
Run automatic remediation for failed checks.
Attempts to automatically fix security issues identified by checks
where fixers are available and appropriate.
Parameters:
- findings: List of Finding objects with failed checks
- provider: Provider instance for executing fixes
- fixer_list: Optional list of specific fixers to run
Returns:
Dictionary containing fix results and any errors encountered
Raises:
ProwlerException: On fixer execution errors
"""Usage Examples
Basic Check Execution
from prowler.lib.check.check import execute_checks
from prowler.lib.check.checks_loader import load_checks_to_execute
from prowler.providers.aws.aws_provider import AWSProvider
# Initialize provider
provider = AWSProvider()
# Load all checks for provider
checks = load_checks_to_execute(provider)
# Execute checks and collect findings
findings = execute_checks(checks, provider)
# Process results
for finding in findings:
print(f"Check: {finding.metadata.CheckID}")
print(f"Status: {finding.status}")
print(f"Resource: {finding.resource_uid}")Filtered Check Execution
from prowler.lib.check.checks_loader import load_checks_to_execute
from prowler.providers.azure.azure_provider import AzureProvider
provider = AzureProvider()
# Load specific checks only
specific_checks = load_checks_to_execute(
provider,
check_list=['storage_account_public_access_disabled', 'vm_disk_encryption_enabled']
)
# Load checks for specific services
service_checks = load_checks_to_execute(
provider,
service_list=['storage', 'compute']
)
# Load checks with exclusions
filtered_checks = load_checks_to_execute(
provider,
excluded_services=['network'],
excluded_checks=['vm_old_image_version']
)Compliance Framework Filtering
from prowler.lib.check.checks_loader import load_checks_to_execute
from prowler.providers.gcp.gcp_provider import GCPProvider
provider = GCPProvider()
# Load checks for CIS compliance
cis_checks = load_checks_to_execute(
provider,
compliance_frameworks=['cis_1.3_gcp']
)
# Load checks for multiple frameworks
multi_compliance_checks = load_checks_to_execute(
provider,
compliance_frameworks=['cis_1.3_gcp', 'nist_csf_1.1_gcp']
)Custom Check Loading
from prowler.lib.check.check import parse_checks_from_folder, execute_checks
from prowler.providers.aws.aws_provider import AWSProvider
provider = AWSProvider()
# Load custom checks from folder
custom_checks = parse_checks_from_folder('/path/to/custom/checks')
# Execute custom checks
findings = execute_checks(custom_checks, provider)
# Clean up custom modules
remove_custom_checks_module()Information Discovery
from prowler.lib.check.check import (
list_services,
list_categories,
list_checks_json,
print_checks
)
# Get available services for AWS
aws_services = list_services('aws')
print(f"AWS Services: {aws_services}")
# Get check categories
categories = list_categories('azure')
print(f"Azure Categories: {categories}")
# Export check metadata as JSON
check_metadata = list_checks_json('gcp', check_list=['compute_instance_public_ip'])
# Print formatted check information
print_checks('aws', service_list=['iam', 'ec2'])Install with Tessl CLI
npx tessl i tessl/pypi-prowler