prowler/examples/sdk/aws_scan.ipynb

Setup and Configure Logger

This section configures the Python logging system to filter Prowler's output messages during security scans. We set the logging level to CRITICAL.

In [8]:

import logging

logging.basicConfig(level=logging.CRITICAL)

Initialize AWS Provider

In [ ]:

# Import the Prowler's provider you want to scan
from prowler.providers.aws.aws_provider import AwsProvider
import json

# Path to credentials file
credentials_path = "./secrets-sdk/credentials.json"

# Load credentials from JSON file
try:
    with open(credentials_path, "r") as f:
        aws_credentials = json.load(f)
    print("AWS credentials loaded successfully from file")
except (FileNotFoundError, json.JSONDecodeError):
    print("Invalid or missing JSON credentials file")
    aws_credentials = {
        "aws_access_key_id": "YOUR_ACCESS_KEY",
        "aws_secret_access_key": "YOUR_SECRET_KEY",
        "aws_session_token": "YOUR_SESSION_TOKEN"
    }

# Optional: Test the AWS provider credentials before instantiation to verify that credentials work
aws_connection = AwsProvider.test_connection(**aws_credentials)
print(f"AWS Test Connection:\n\t- Connected: {aws_connection.is_connected}\n\t- Error (if any): {aws_connection.error}\n")

# Initialize the AWS provider with static credentials
aws = AwsProvider(**aws_credentials)

# AWS Identity Information
print(f"AWS Identity Information:\n\t- Account Number: {aws.identity.account}\n\t- User ID: {aws.identity.user_id}\n")

# Alternative Providers (commented out)
# from prowler.providers.gcp.gcp_provider import GcpProvider
# from prowler.providers.azure.azure_provider import AzureProvider
# from prowler.providers.kubernetes.kubernetes_provider import KubernetesProvider
# from prowler.providers.m365.m365_provider import M365Provider

In [ ]:

import pprint
pprint.pp(aws.identity)

Mutelist

In [ ]:

# Mutelist
from prowler.providers.aws.lib.mutelist.mutelist import AWSMutelist


mutelist_content = {
    "Accounts": {
        "*": {
            "Checks": {
                "s3_account_level_public_access_blocks": {
                    "Tags": ["*"],
                    "Regions": ["*"],
                    "Resources": ["*"],
                }
            }
        }
    }
}
mutelist_object = AWSMutelist(
    mutelist_content=mutelist_content,
)
aws._mutelist = mutelist_object

List Available Security Checks

Explore different ways to list security checks by provider, service, severity, and category.

In [ ]:

# Import the CheckMetadata class to list security checks
from prowler.lib.check.models import CheckMetadata

# List all available checks
checks = CheckMetadata.list()
print(f"# Checks: {len(checks)}")

# List all AWS S3 checks
aws_s3_checks = CheckMetadata.list(provider="aws", service="s3")
print(f"AWS S3 Checks:\n\t- {'\n\t- '.join(aws_s3_checks)}")

# List all critical severity checks
critical_checks = CheckMetadata.list(provider="aws", severity="critical")
print(f"\n# Critical Checks: {len(critical_checks)}")

# List all checks in the internet-exposed category
internet_exposed_checks = CheckMetadata.list(provider="aws", category="internet-exposed")
print(f"\n# Internet-Exposed Category Checks: {len(internet_exposed_checks)}")

Execute Security Scans

Set up and execute security scans on AWS resources with different filtering options.

In [ ]:

# Import necessary libraries for scanning
from prowler.lib.scan.scan import Scan
# Auxiliary libraries
import itertools

# Set up the Scan class to scan all checks for the provider
scan = Scan(provider=aws)

# Parametrize the Scan to execute several checks, services, categories, compliances, etc.
scan_s3 = Scan(provider=aws, services=["s3"], severities=["critical", "high"])
# scan_critical = Scan(provider=aws, severities=["critical"])
# scan_internet_exposed = Scan(provider=aws, categories=["internet-exposed"])

# Start the scan with the `scan` method. This returns a generator with findings and progress.
print("\n##### Scanning AWS #####")
all_findings = []
total_findings = 0
for progress, findings in scan_s3.scan():
    all_findings.extend(findings)
    total_findings += len(findings)
    print(f"- Scan Progress: {progress}\n- # Findings: {total_findings}\n")

Process and Display Findings

Process the scan results and display detailed information about each finding.

In [ ]:

print("Finding's Details:")
for finding in all_findings:
    print(f"""
    - Check ID: {finding.metadata.CheckID}
    - Status: {str(finding.status)}
    - Status Extended: {finding.status_extended}
    - Resource ID: {finding.resource_uid}
    - Resource Metadata: {finding.resource_metadata}
    """)

# Retrieve all findings in one line
print("\n##### Getting all findings in one line #####")
scan_s3 = Scan(provider=aws, services=["s3"], severities=["critical"])
all_findings_one_line = list(itertools.chain.from_iterable(findings for _, findings in scan_s3.scan()))
print(f"Total findings collected in one line: {len(all_findings_one_line)}")

Check Metatada

CheckMetadata(
    Provider='aws'
    CheckID='s3_bucket_policy_public_write_access'
    CheckTitle='Check if S3 buckets have policies which allow WRITE access.'
    CheckType=['IAM']
    CheckAliases=[]
    ServiceName='s3'
    SubServiceName=''
    ResourceIdTemplate='arn:partition:s3:::bucket_name'
    Severity=<Severity.critical: 'critical'>
    ResourceType='AwsS3Bucket'
    Description='Check if S3 buckets have policies which allow WRITE access.'
    Risk='Non intended users can put objects in a given bucket.'
    RelatedUrl=''
    Remediation=
        Remediation(
            Code=Code(
                NativeIaC=''
                Terraform=''
                CLI=''
                Other='https://docs.prowler.com/checks/aws/s3-policies/s3_18-write-permissions-public#aws-console')
                Recommendation=Recommendation(Text='Ensure proper bucket policy is in place with the least privilege principle applied.'
                Url='https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_examples_s3_rw-bucket.html'
            )
        )
    Categories=['internet-exposed']
    DependsOn=[]
    RelatedTo=[]
    Notes=''
    # Compliance framework: A list of requirement IDs where the check is present.
    Compliance={
        "CIS-1.10": ["5.2.13"],
        "CIS-1.8": ["5.2.13"]
    }
)

Output Formats

In [ ]:

# Import necessary libraries for output
from prowler.lib.outputs.csv.csv import CSV
from prowler.lib.outputs.ocsf.ocsf import OCSF
from prowler.lib.outputs.asff.asff import ASFF  # Only for AWS
from prowler.lib.outputs.html.html import HTML
from prowler.lib.outputs.outputs import extract_findings_statistics
import datetime

# Get current date and time in YYYY-MM-DD_HH-MM-SS format for filenames
current_datetime = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")

# Write findings to CSV file
print("Writing findings to CSV file...")
csv_filename = f"./output/findings_{current_datetime}.csv"
csv = CSV(findings=all_findings, create_file_descriptor=True, file_path=csv_filename)
csv.batch_write_data_to_file()
print(f"Done! CSV File Path: {csv._file_descriptor.name}")

# Write findings to OCSF file
print("Writing findings to OCSF file...")
ocsf_filename = f"./output/findings_{current_datetime}.ocsf"
ocsf = OCSF(findings=all_findings, create_file_descriptor=True, file_path=ocsf_filename)
ocsf.batch_write_data_to_file()
print(f"Done! OCSF File Path: {ocsf._file_descriptor.name}")

# Write findings to ASFF file
print("Writing findings to ASFF file...")
asff_filename = f"./output/findings_{current_datetime}.asff"
asff = ASFF(findings=all_findings, create_file_descriptor=True, file_path=asff_filename)
asff.batch_write_data_to_file()
print(f"Done! ASFF File Path: {asff._file_descriptor.name}")

# Write findings to HTML file
print("Writing findings to HTML file...")
html_filename = f"./output/findings_{current_datetime}.html"
stats = extract_findings_statistics(all_findings)
html = HTML(findings=all_findings, create_file_descriptor=True, file_path=html_filename)
html.batch_write_data_to_file(provider=aws, stats=stats)
print(f"Done! HTML File Path: {html._file_descriptor.name}")

# IMPORTANT: The create_file_descriptor parameter will be removed in 5.4.0
# The file descriptor will be created by default when the Output class is instantiated

Integrate with AWS S3

Send findings to AWS S3.

In [ ]:

# Import the S3 class to send findings to AWS S3
from prowler.providers.aws.lib.s3.s3 import S3

print("\n##### Sending findings to S3 bucket #####")
generated_outputs = {"regular": [csv, ocsf, asff, html], "compliance": []}
s3_integration = S3(aws.session.current_session, bucket_name="sdk-core", output_directory="output")
s3_integration.send_to_bucket(generated_outputs)

# This upload the output files to the S3 bucket. In this case:
# sdk-core/output/csv/findings_2025-02-26_16-25-30.csv
# sdk-core/output/ocsf/findings_2025-02-26_16-25-30.ocsf
# sdk-core/output/asff/findings_2025-02-26_16-25-30.asff

Integrate with AWS Security Hub

Send findings to AWS Security Hub for centralized security monitoring.

In [ ]:

# Import the SecurityHub class to send findings to AWS Security Hub
from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub

# Print message indicating the start of the process
print("\n##### Sending findings to AWS Security Hub #####")

# Get available AWS regions for Security Hub.
# Each finding can only be sent to Security Hub within its own region.
# Additionally, it verifies that Prowler’s integration is active in
# Security Hub before sending
available_regions = aws.get_available_aws_service_regions(
    "securityhub",
    aws.identity.partition,
    aws.identity.audited_regions,
)

# Initialize the SecurityHub class with necessary parameters
security_hub = SecurityHub(
    aws_account_id=aws.identity.account,
    aws_partition=aws.identity.partition,
    aws_session=aws.session.current_session,
    findings=asff.data,
    send_only_fails=False,
    aws_security_hub_available_regions=available_regions,
)

# Send findings to AWS Security Hub
findings_sent_to_security_hub = security_hub.batch_send_to_security_hub()

# Print the number of findings sent to AWS Security Hub
print(f"{findings_sent_to_security_hub} findings sent to AWS Security Hub!")