ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 04:51:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(SecurityHub): create class to handle integration (#4397)

Browse Source 
Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Pepe Fagoaga
2024-07-09 17:47:47 +02:00
committed by GitHub
parent a0603b972e
commit 6336b1c0d9
4 changed files with 509 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/__main__.py
+12 -31
View File
@@ -52,12 +52,7 @@ from prowler.lib.outputs.outputs import extract_findings_statistics
from prowler.lib.outputs.slack.slack import Slack
from prowler.lib.outputs.summary_table import display_summary_table
from prowler.providers.aws.lib.s3.s3 import send_to_s3_bucket
from prowler.providers.aws.lib.security_hub.security_hub import (
batch_send_to_security_hub,
filter_security_hub_findings_per_region,
resolve_security_hub_previous_findings,
verify_security_hub_integration_enabled_per_region,
)
from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
from prowler.providers.common.provider import Provider
from prowler.providers.common.quick_inventory import run_provider_quick_inventory
@@ -360,35 +355,24 @@ def prowler():
print(
f"{Style.BRIGHT}\nSending findings to AWS Security Hub, please wait...{Style.RESET_ALL}"
)
# Verify where AWS Security Hub is enabled
aws_security_enabled_regions = []
security_hub_regions = (
global_provider.get_available_aws_service_regions("securityhub")
if not global_provider.identity.audited_regions
else global_provider.identity.audited_regions
)
for region in security_hub_regions:
# Save the regions where AWS Security Hub is enabled
if verify_security_hub_integration_enabled_per_region(
global_provider.identity.partition,
region,
global_provider.session.current_session,
global_provider.identity.account,
):
aws_security_enabled_regions.append(region)
# Prepare the findings to be sent to Security Hub
security_hub_findings_per_region = filter_security_hub_findings_per_region(
asff_output.data,
global_provider.output_options.send_sh_only_fails,
global_provider.output_options.status,
aws_security_enabled_regions,
security_hub = SecurityHub(
aws_account_id=global_provider.identity.account,
aws_partition=global_provider.identity.partition,
aws_session=global_provider.session.current_session,
findings=asff_output.data,
status=global_provider.output_options.status,
send_only_fails=global_provider.output_options.send_sh_only_fails,
aws_security_hub_available_regions=security_hub_regions,
)
# Send the findings to Security Hub
findings_sent_to_security_hub = batch_send_to_security_hub(
security_hub_findings_per_region, global_provider.session.current_session
)
findings_sent_to_security_hub = security_hub.batch_send_to_security_hub()
print(
f"{Style.BRIGHT}{Fore.GREEN}\n{findings_sent_to_security_hub} findings sent to AWS Security Hub!{Style.RESET_ALL}"
)
@@ -398,10 +382,7 @@ def prowler():
print(
f"{Style.BRIGHT}\nArchiving previous findings in AWS Security Hub, please wait...{Style.RESET_ALL}"
)
findings_archived_in_security_hub = resolve_security_hub_previous_findings(
security_hub_findings_per_region,
global_provider,
)
findings_archived_in_security_hub = security_hub.archive_previous_findings()
print(
f"{Style.BRIGHT}{Fore.GREEN}\n{findings_archived_in_security_hub} findings archived in AWS Security Hub!{Style.RESET_ALL}"
)
prowler/lib/outputs/asff/asff.py
+5 -1
View File
@@ -82,7 +82,11 @@ class ASFF(Output):
),
GeneratorId="prowler-" + finding.check_id,
AwsAccountId=finding.account_uid,
Types=finding.check_type.split(","),
Types=(
finding.check_type.split(",")
if finding.check_type
else ["Software and Configuration Checks"]
),
FirstObservedAt=timestamp,
UpdatedAt=timestamp,
CreatedAt=timestamp,
prowler/providers/aws/lib/security_hub/security_hub.py
+259 -225
View File
@@ -1,4 +1,4 @@
from boto3 import session
from boto3 import Session
from botocore.client import ClientError
from prowler.config.config import timestamp_utc
@@ -9,254 +9,288 @@ SECURITY_HUB_INTEGRATION_NAME = "prowler/prowler"
SECURITY_HUB_MAX_BATCH = 100
def filter_security_hub_findings_per_region(
findings: list[AWSSecurityFindingFormat],
send_only_fails: bool,
status: list,
enabled_regions: list,
) -> dict:
"""filter_security_hub_findings_per_region filters the findings by region and status. It returns a dictionary with the findings per region.
Args:
findings (list[AWSSecurityFindingFormat]): List of findings
send_only_fails (bool): Send only the findings that have failed
status (list): List of statuses to filter the findings
enabled_regions (list): List of enabled regions
Returns:
dict: Dictionary containing the findings per region
class SecurityHub:
"""
security_hub_findings_per_region = {}
# Create a key per audited region
for region in enabled_regions:
security_hub_findings_per_region[region] = []
for finding in findings:
# We don't send findings to not enabled regions
if finding.Resources[0].Region not in enabled_regions:
continue
Class representing a SecurityHub object for managing findings and interactions with AWS Security Hub.
if (
finding.Compliance.Status != "FAILED"
or finding.Compliance.Status == "WARNING"
) and send_only_fails:
continue
Attributes:
_session (Session): AWS session object for authentication and communication with AWS services.
_aws_account_id (str): AWS account ID associated with the SecurityHub instance.
_aws_partition (str): AWS partition (e.g., aws, aws-cn, aws-us-gov) where SecurityHub is deployed.
_findings_per_region (dict): Dictionary containing findings per region.
_enabled_regions (dict): Dictionary containing enabled regions with SecurityHub clients.
# SecurityHub valid statuses are: PASSED, FAILED, WARNING
if status:
if finding.Compliance.Status == "PASSED" and "PASS" not in status:
continue
if finding.Compliance.Status == "FAILED" and "FAIL" not in status:
continue
# Check muted finding
if finding.Compliance.Status == "WARNING":
Methods:
__init__: Initializes the SecurityHub object with necessary attributes.
filter: Filters findings based on region and status, returning a dictionary with findings per region.
verify_enabled_per_region: Verifies and stores enabled regions with SecurityHub clients.
batch_send_to_security_hub: Sends findings to Security Hub and returns the count of successfully sent findings.
archive_previous_findings: Archives findings that are not present in the current execution.
_send_findings_to_security_hub: Sends findings to AWS Security Hub in batches and returns the count of successfully sent findings.
"""
_session: Session
_aws_account_id: str
_aws_partition: str
_findings_per_region: dict[str, list[AWSSecurityFindingFormat]]
_enabled_regions: dict[str, Session]
def __init__(
self,
aws_session: Session,
aws_account_id: str,
aws_partition: str,
findings: list[AWSSecurityFindingFormat] = [],
status: list[str] = [],
aws_security_hub_available_regions: list[str] = [],
send_only_fails: bool = False,
) -> "SecurityHub":
self._session = aws_session
self._aws_account_id = aws_account_id
self._aws_partition = aws_partition
self._enabled_regions = None
self._findings_per_region = None
if aws_security_hub_available_regions:
self._enabled_regions = self.verify_enabled_per_region(
aws_security_hub_available_regions
)
if findings and self._enabled_regions:
self._findings_per_region = self.filter(findings, send_only_fails, status)
def filter(
self,
findings: list[AWSSecurityFindingFormat],
send_only_fails: bool,
status: list[str],
) -> dict:
"""
Filters the given list of findings based on the provided criteria and returns a dictionary containing findings per region.
Args:
findings (list[AWSSecurityFindingFormat]): List of findings to filter.
send_only_fails (bool): Flag indicating whether to send only findings with status 'FAILED'.
status (list[str]): List of valid statuses to filter the findings.
Returns:
dict: A dictionary containing findings per region after applying the filtering criteria.
"""
findings_per_region = {}
# Create a key per audited region
for region in self._enabled_regions.keys():
findings_per_region[region] = []
for finding in findings:
# We don't send findings to not enabled regions
if finding.Resources[0].Region not in findings_per_region:
continue
# Get the finding region
# We can do that since the finding always stores just one finding
region = finding.Resources[0].Region
if (
finding.Compliance.Status != "FAILED"
or finding.Compliance.Status == "WARNING"
) and send_only_fails:
continue
# Include that finding within their region
security_hub_findings_per_region[region].append(finding)
# SecurityHub valid statuses are: PASSED, FAILED, WARNING
if status:
if finding.Compliance.Status == "PASSED" and "PASS" not in status:
continue
if finding.Compliance.Status == "FAILED" and "FAIL" not in status:
continue
# Check muted finding
if finding.Compliance.Status == "WARNING":
continue
return security_hub_findings_per_region
# Get the finding region
# We can do that since the finding always stores just one finding
region = finding.Resources[0].Region
# Include that finding within their region
findings_per_region[region].append(finding)
def verify_security_hub_integration_enabled_per_region(
partition: str,
region: str,
session: session.Session,
aws_account_number: str,
) -> bool:
"""
verify_security_hub_integration_enabled_per_region returns True if the Prowler integration is enabled for the given region. Otherwise returns false.
return findings_per_region
Args:
partition (str): AWS partition
region (str): AWS region
session (session.Session): AWS session
aws_account_number (str): AWS account number
def verify_enabled_per_region(
self,
aws_security_hub_available_regions: list[str],
) -> dict[str, Session]:
"""
Filters the given list of regions where AWS Security Hub is enabled and returns a dictionary containing the region and their boto3 client if the region and the Prowler integration is enabled.
Returns:
bool: True if the Prowler integration is enabled for the given region. Otherwise returns false.
"""
f"""verify_security_hub_integration_enabled returns True if the {SECURITY_HUB_INTEGRATION_NAME} is enabled for the given region. Otherwise returns false."""
prowler_integration_enabled = False
Args:
aws_security_hub_available_regions (list[str]): List of AWS regions to check for Security Hub integration.
try:
logger.info(
f"Checking if the {SECURITY_HUB_INTEGRATION_NAME} is enabled in the {region} region."
)
# Check if security hub is enabled in current region
security_hub_client = session.client("securityhub", region_name=region)
security_hub_client.describe_hub()
Returns:
dict: A dictionary containing enabled regions with SecurityHub clients.
"""
enabled_regions = {}
for region in aws_security_hub_available_regions:
try:
logger.info(
f"Checking if the {SECURITY_HUB_INTEGRATION_NAME} is enabled in the {region} region."
)
# Check if security hub is enabled in current region
security_hub_client = self._session.client(
"securityhub", region_name=region
)
security_hub_client.describe_hub()
# Check if Prowler integration is enabled in Security Hub
security_hub_prowler_integration_arn = f"arn:{partition}:securityhub:{region}:{aws_account_number}:product-subscription/{SECURITY_HUB_INTEGRATION_NAME}"
if security_hub_prowler_integration_arn not in str(
security_hub_client.list_enabled_products_for_import()
):
logger.warning(
f"Security Hub is enabled in {region} but Prowler integration does not accept findings. More info: https://docs.prowler.cloud/en/latest/tutorials/aws/securityhub/"
)
else:
prowler_integration_enabled = True
# Check if Prowler integration is enabled in Security Hub
security_hub_prowler_integration_arn = f"arn:{self._aws_partition}:securityhub:{region}:{self._aws_account_id}:product-subscription/{SECURITY_HUB_INTEGRATION_NAME}"
if security_hub_prowler_integration_arn not in str(
security_hub_client.list_enabled_products_for_import()
):
logger.warning(
f"Security Hub is enabled in {region} but Prowler integration does not accept findings. More info: https://docs.prowler.cloud/en/latest/tutorials/aws/securityhub/"
)
else:
enabled_regions[region] = self._session.client(
"securityhub", region_name=region
)
# Handle all the permissions / configuration errors
except ClientError as client_error:
# Check if Account is subscribed to Security Hub
error_code = client_error.response["Error"]["Code"]
error_message = client_error.response["Error"]["Message"]
if (
error_code == "InvalidAccessException"
and f"Account {aws_account_number} is not subscribed to AWS Security Hub"
in error_message
):
logger.warning(
f"{client_error.__class__.__name__} -- [{client_error.__traceback__.tb_lineno}]: {client_error}"
)
else:
logger.error(
f"{client_error.__class__.__name__} -- [{client_error.__traceback__.tb_lineno}]: {client_error}"
)
except Exception as error:
logger.error(
f"{error.__class__.__name__} -- [{error.__traceback__.tb_lineno}]: {error}"
)
# Handle all the permissions / configuration errors
except ClientError as client_error:
# Check if Account is subscribed to Security Hub
error_code = client_error.response["Error"]["Code"]
error_message = client_error.response["Error"]["Message"]
if (
error_code == "InvalidAccessException"
and f"Account {self._aws_account_id} is not subscribed to AWS Security Hub"
in error_message
):
logger.warning(
f"{client_error.__class__.__name__} -- [{client_error.__traceback__.tb_lineno}]: {client_error}"
)
else:
logger.error(
f"{client_error.__class__.__name__} -- [{client_error.__traceback__.tb_lineno}]: {client_error}"
)
except Exception as error:
logger.error(
f"{error.__class__.__name__} -- [{error.__traceback__.tb_lineno}]: {error}"
)
return enabled_regions
finally:
return prowler_integration_enabled
def batch_send_to_security_hub(
self,
) -> int:
"""
Sends the findings to AWS Security Hub in batches for each region and returns the count of successfully sent findings.
def batch_send_to_security_hub(
security_hub_findings_per_region: dict,
session: session.Session,
) -> int:
"""
batch_send_to_security_hub sends findings to Security Hub and returns the number of findings that were successfully sent.
Args:
security_hub_findings_per_region (dict): Dictionary containing the findings per region
session (session.Session): AWS session
Returns:
int: Number of sent findings
"""
success_count = 0
try:
# Iterate findings by region
for region, findings in security_hub_findings_per_region.items():
# Send findings to Security Hub
logger.info(
f"Sending {len(findings)} findings to Security Hub in the region {region}"
)
security_hub_client = session.client("securityhub", region_name=region)
# Convert findings to dict
findings = [finding.dict(exclude_none=True) for finding in findings]
success_count += _send_findings_to_security_hub(
findings, region, security_hub_client
)
except Exception as error:
logger.error(
f"{error.__class__.__name__} -- [{error.__traceback__.tb_lineno}]:{error} in region {region}"
)
return success_count
# Move previous Security Hub check findings to ARCHIVED (as prowler didn't re-detect them)
def resolve_security_hub_previous_findings(
security_hub_findings_per_region: dict, provider
) -> int:
"""
resolve_security_hub_previous_findings archives all the findings that does not appear in the current execution
Args:
security_hub_findings_per_region (dict): Dictionary containing the findings per region
provider: Provider object
Returns:
int: Number of archived findings
"""
logger.info("Checking previous findings in Security Hub to archive them.")
success_count = 0
for region in security_hub_findings_per_region.keys():
Returns:
int: Number of successfully sent findings to AWS Security Hub.
"""
success_count = 0
try:
current_findings = security_hub_findings_per_region[region]
# Get current findings IDs
current_findings_ids = []
for finding in current_findings:
current_findings_ids.append(finding.Id)
# Get findings of that region
security_hub_client = provider.session.current_session.client(
"securityhub", region_name=region
)
findings_filter = {
"ProductName": [{"Value": "Prowler", "Comparison": "EQUALS"}],
"RecordState": [{"Value": "ACTIVE", "Comparison": "EQUALS"}],
"AwsAccountId": [
{"Value": provider.identity.account, "Comparison": "EQUALS"}
],
"Region": [{"Value": region, "Comparison": "EQUALS"}],
}
get_findings_paginator = security_hub_client.get_paginator("get_findings")
findings_to_archive = []
for page in get_findings_paginator.paginate(
Filters=findings_filter, PaginationConfig={"PageSize": 100}
):
# Archive findings that have not appear in this execution
for finding in page["Findings"]:
if finding["Id"] not in current_findings_ids:
finding["RecordState"] = "ARCHIVED"
finding["UpdatedAt"] = timestamp_utc.strftime(
"%Y-%m-%dT%H:%M:%SZ"
)
# Iterate findings by region
for region, findings in self._findings_per_region.items():
# Send findings to Security Hub
logger.info(
f"Sending {len(findings)} findings to Security Hub in the region {region}"
)
findings_to_archive.append(finding)
logger.info(f"Archiving {len(findings_to_archive)} findings.")
# Convert findings to dict
findings = [finding.dict(exclude_none=True) for finding in findings]
success_count += self._send_findings_in_batches(
findings,
region,
)
# Send archive findings to SHub
success_count += _send_findings_to_security_hub(
findings_to_archive, region, security_hub_client
)
except Exception as error:
logger.error(
f"{error.__class__.__name__} -- [{error.__traceback__.tb_lineno}]:{error} in region {region}"
)
return success_count
return success_count
def archive_previous_findings(self) -> int:
"""
Checks previous findings in Security Hub to archive them.
def _send_findings_to_security_hub(
findings: list[dict], region: str, security_hub_client
) -> int:
"""Private function send_findings_to_security_hub chunks the findings in groups of 100 findings and send them to AWS Security Hub. It returns the number of sent findings.
Args:
findings (list[dict]): List of findings to send to AWS Security Hub
region (str): AWS region to send the findings
security_hub_client: AWS Security Hub client
Returns:
int: Number of sent findings
"""
success_count = 0
try:
list_chunked = [
findings[i : i + SECURITY_HUB_MAX_BATCH]
for i in range(0, len(findings), SECURITY_HUB_MAX_BATCH)
]
for findings in list_chunked:
batch_import = security_hub_client.batch_import_findings(Findings=findings)
if batch_import["FailedCount"] > 0:
failed_import = batch_import["FailedFindings"][0]
logger.error(
f"Failed to send findings to AWS Security Hub -- {failed_import['ErrorCode']} -- {failed_import['ErrorMessage']}"
Returns:
int: Number of successfully archived findings.
"""
logger.info("Checking previous findings in Security Hub to archive them.")
success_count = 0
for region in self._findings_per_region.keys():
try:
current_findings = self._findings_per_region[region]
# Get current findings IDs
current_findings_ids = []
for finding in current_findings:
current_findings_ids.append(finding.Id)
# Get findings of that region
findings_filter = {
"ProductName": [{"Value": "Prowler", "Comparison": "EQUALS"}],
"RecordState": [{"Value": "ACTIVE", "Comparison": "EQUALS"}],
"AwsAccountId": [
{"Value": self._aws_account_id, "Comparison": "EQUALS"}
],
"Region": [{"Value": region, "Comparison": "EQUALS"}],
}
get_findings_paginator = self._enabled_regions[region].get_paginator(
"get_findings"
)
findings_to_archive = []
for page in get_findings_paginator.paginate(
Filters=findings_filter, PaginationConfig={"PageSize": 100}
):
# Archive findings that have not appear in this execution
for finding in page["Findings"]:
if finding["Id"] not in current_findings_ids:
finding["RecordState"] = "ARCHIVED"
finding["UpdatedAt"] = timestamp_utc.strftime(
"%Y-%m-%dT%H:%M:%SZ"
)
findings_to_archive.append(finding)
logger.info(f"Archiving {len(findings_to_archive)} findings.")
# Send archive findings to SHub
success_count += self._send_findings_in_batches(
findings_to_archive,
region,
)
except Exception as error:
logger.error(
f"{error.__class__.__name__} -- [{error.__traceback__.tb_lineno}]:{error} in region {region}"
)
success_count += batch_import["SuccessCount"]
return success_count
except Exception as error:
logger.error(
f"{error.__class__.__name__} -- [{error.__traceback__.tb_lineno}]:{error} in region {region}"
)
return success_count
def _send_findings_in_batches(
self, findings: list[AWSSecurityFindingFormat], region: str
) -> int:
"""
Sends the given findings to AWS Security Hub in batches for a specific region and returns the count of successfully sent findings.
Args:
findings (list[AWSSecurityFindingFormat]): List of findings to send to AWS Security Hub.
region (str): The AWS region where the findings will be sent.
Returns:
int: Number of successfully sent findings to AWS Security Hub.
"""
success_count = 0
try:
list_chunked = [
findings[i : i + SECURITY_HUB_MAX_BATCH]
for i in range(0, len(findings), SECURITY_HUB_MAX_BATCH)
]
for findings in list_chunked:
batch_import = self._enabled_regions[region].batch_import_findings(
Findings=findings
)
if batch_import["FailedCount"] > 0:
failed_import = batch_import["FailedFindings"][0]
logger.error(
f"Failed to send findings to AWS Security Hub -- {failed_import['ErrorCode']} -- {failed_import['ErrorMessage']}"
)
success_count += batch_import["SuccessCount"]
return success_count
except Exception as error:
logger.error(
f"{error.__class__.__name__} -- [{error.__traceback__.tb_lineno}]:{error} in region {region}"
)
return success_count
tests/providers/aws/lib/security_hub/security_hub_test.py
+233 -148
View File
@@ -1,4 +1,5 @@
from logging import ERROR, WARNING
import re
from logging import WARNING
import botocore
from boto3 import session
@@ -6,11 +7,7 @@ from botocore.client import ClientError
from mock import patch
from prowler.lib.outputs.asff.asff import ASFF
from prowler.providers.aws.lib.security_hub.security_hub import (
batch_send_to_security_hub,
filter_security_hub_findings_per_region,
verify_security_hub_integration_enabled_per_region,
)
from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
from tests.lib.outputs.fixtures.fixtures import generate_finding_output
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
@@ -41,36 +38,36 @@ def mock_make_api_call(self, operation_name, kwarg):
return {
"ProductSubscriptions": [
f"arn:aws:securityhub:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:product-subscription/prowler/prowler",
f"arn:aws:securityhub:{AWS_REGION_EU_WEST_2}:{AWS_ACCOUNT_NUMBER}:product-subscription/prowler/prowler",
]
}
return make_api_call(self, operation_name, kwarg)
def set_mocked_session(region=None):
# Create mock session
return session.Session(
region_name=region,
)
class TestSecurityHub:
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_verify_security_hub_integration_enabled_per_region(self):
session = set_mocked_session(AWS_REGION_EU_WEST_1)
assert verify_security_hub_integration_enabled_per_region(
AWS_COMMERCIAL_PARTITION, AWS_REGION_EU_WEST_1, session, AWS_ACCOUNT_NUMBER
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
)
assert security_hub._enabled_regions
assert len(security_hub._enabled_regions) == 1
assert AWS_REGION_EU_WEST_1 in security_hub._enabled_regions
def test_verify_security_hub_integration_enabled_per_region_security_hub_disabled(
self, caplog
):
caplog.set_level(WARNING)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
with patch(
"prowler.providers.aws.lib.security_hub.security_hub.session.Session.client",
"prowler.providers.aws.lib.security_hub.security_hub.Session.client",
) as mock_security_hub:
error_message = f"Account {AWS_ACCOUNT_NUMBER} is not subscribed to AWS Security Hub in region {AWS_REGION_EU_WEST_1}"
error_code = "InvalidAccessException"
@@ -83,38 +80,50 @@ class TestSecurityHub:
operation_name = "DescribeHub"
mock_security_hub.side_effect = ClientError(error_response, operation_name)
assert not verify_security_hub_integration_enabled_per_region(
AWS_COMMERCIAL_PARTITION,
AWS_REGION_EU_WEST_1,
session,
AWS_ACCOUNT_NUMBER,
)
assert caplog.record_tuples == [
(
"root",
WARNING,
f"ClientError -- [90]: An error occurred ({error_code}) when calling the {operation_name} operation: {error_message}",
log_pattern = re.compile(
r"ClientError -- \[\d+\]: An error occurred \({error_code}\) when calling the {operation_name} operation: {error_message}".format(
error_code=re.escape(error_code),
operation_name=re.escape(operation_name),
error_message=re.escape(error_message),
)
]
)
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
)
assert security_hub._enabled_regions == {}
assert any(
log_pattern.match(record.message) for record in caplog.records
), "Expected log message not found"
def test_verify_security_hub_integration_enabled_per_region_prowler_not_subscribed(
self, caplog
):
caplog.set_level(WARNING)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
with patch(
"prowler.providers.aws.lib.security_hub.security_hub.session.Session.client",
"prowler.providers.aws.lib.security_hub.security_hub.Session.client",
) as mock_security_hub:
mock_security_hub.describe_hub.return_value = None
mock_security_hub.list_enabled_products_for_import.return_value = []
assert not verify_security_hub_integration_enabled_per_region(
AWS_COMMERCIAL_PARTITION,
AWS_REGION_EU_WEST_1,
session,
AWS_ACCOUNT_NUMBER,
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
)
assert security_hub._enabled_regions == {}
assert caplog.record_tuples == [
(
"root",
@@ -127,10 +136,9 @@ class TestSecurityHub:
self, caplog
):
caplog.set_level(WARNING)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
with patch(
"prowler.providers.aws.lib.security_hub.security_hub.session.Session.client",
"prowler.providers.aws.lib.security_hub.security_hub.Session.client",
) as mock_security_hub:
error_message = f"Another exception in region {AWS_REGION_EU_WEST_1}"
error_code = "AnotherException"
@@ -143,148 +151,215 @@ class TestSecurityHub:
operation_name = "DescribeHub"
mock_security_hub.side_effect = ClientError(error_response, operation_name)
assert not verify_security_hub_integration_enabled_per_region(
AWS_COMMERCIAL_PARTITION,
AWS_REGION_EU_WEST_1,
session,
AWS_ACCOUNT_NUMBER,
)
assert caplog.record_tuples == [
(
"root",
ERROR,
f"ClientError -- [90]: An error occurred ({error_code}) when calling the {operation_name} operation: {error_message}",
log_pattern = re.compile(
r"ClientError -- \[\d+\]: An error occurred \({error_code}\) when calling the {operation_name} operation: {error_message}".format(
error_code=re.escape(error_code),
operation_name=re.escape(operation_name),
error_message=re.escape(error_message),
)
]
)
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
)
assert security_hub._enabled_regions == {}
assert any(
log_pattern.match(record.message) for record in caplog.records
), "Expected log message not found"
def test_verify_security_hub_integration_enabled_per_region_another_Exception(
self, caplog
):
caplog.set_level(WARNING)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
with patch(
"prowler.providers.aws.lib.security_hub.security_hub.session.Session.client",
"prowler.providers.aws.lib.security_hub.security_hub.Session.client",
) as mock_security_hub:
error_message = f"Another exception in region {AWS_REGION_EU_WEST_1}"
mock_security_hub.side_effect = Exception(error_message)
assert not verify_security_hub_integration_enabled_per_region(
AWS_COMMERCIAL_PARTITION,
AWS_REGION_EU_WEST_1,
session,
AWS_ACCOUNT_NUMBER,
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
)
assert caplog.record_tuples == [
(
"root",
ERROR,
f"Exception -- [90]: {error_message}",
)
]
log_pattern = re.compile(
r"Exception -- \[\d+\]: {error_message}".format(
error_message=re.escape(error_message),
)
)
assert security_hub._enabled_regions == {}
assert any(
log_pattern.match(record.message) for record in caplog.records
), "Expected log message not found"
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_filter_security_hub_findings_per_region_enabled_region_all_statuses(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: [asff.data[0]]}
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
findings=asff.data,
)
assert security_hub._findings_per_region == {
AWS_REGION_EU_WEST_1: [asff.data[0]]
}
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_filter_security_hub_findings_per_region_all_statuses_MANUAL_finding(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [
generate_finding_output(status="MANUAL", region=AWS_REGION_EU_WEST_1)
]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
findings=asff.data,
)
assert security_hub._findings_per_region is None
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_filter_security_hub_findings_per_region_disabled_region(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_2)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
findings=asff.data,
)
assert security_hub._findings_per_region == {AWS_REGION_EU_WEST_1: []}
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_filter_security_hub_findings_per_region_PASS_and_FAIL_statuses(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
["FAIL"],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
findings=asff.data,
status=["FAIL"],
)
assert security_hub._findings_per_region == {AWS_REGION_EU_WEST_1: []}
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_filter_security_hub_findings_per_region_FAIL_and_FAIL_statuses(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="FAIL", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
["FAIL"],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: [asff.data[0]]}
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
findings=asff.data,
status=["FAIL"],
)
assert security_hub._findings_per_region == {
AWS_REGION_EU_WEST_1: [asff.data[0]]
}
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_filter_security_hub_findings_per_region_send_sh_only_fails_PASS(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
True,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: []}
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
findings=asff.data,
status=[],
send_only_fails=True,
)
assert security_hub._findings_per_region == {AWS_REGION_EU_WEST_1: []}
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_filter_security_hub_findings_per_region_send_sh_only_fails_FAIL(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="FAIL", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
True,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: [asff.data[0]]}
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
findings=asff.data,
status=[],
send_only_fails=True,
)
assert security_hub._findings_per_region == {
AWS_REGION_EU_WEST_1: [asff.data[0]]
}
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_filter_security_hub_findings_per_region_no_audited_regions(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [generate_finding_output(status="PASS", region=AWS_REGION_EU_WEST_1)]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
[],
enabled_regions,
) == {AWS_REGION_EU_WEST_1: [asff.data[0]]}
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[],
findings=asff.data,
status=[],
send_only_fails=True,
)
assert security_hub._findings_per_region is None
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_filter_security_hub_findings_per_region_muted_fail_with_send_sh_only_fails(
self,
):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [
generate_finding_output(
status="FAIL", region=AWS_REGION_EU_WEST_1, muted=True
@@ -292,17 +367,24 @@ class TestSecurityHub:
]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
True,
[],
enabled_regions,
) == {
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
findings=asff.data,
status=[],
send_only_fails=True,
)
assert security_hub._findings_per_region == {
AWS_REGION_EU_WEST_1: [],
}
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_filter_security_hub_findings_per_region_muted_fail_with_status_FAIL(self):
enabled_regions = [AWS_REGION_EU_WEST_1]
findings = [
generate_finding_output(
status="FAIL", region=AWS_REGION_EU_WEST_1, muted=True
@@ -310,12 +392,19 @@ class TestSecurityHub:
]
asff = ASFF(findings=findings)
assert filter_security_hub_findings_per_region(
asff.data,
False,
["FAIL"],
enabled_regions,
) == {
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
findings=asff.data,
status=["FAIL"],
send_only_fails=True,
)
assert security_hub._findings_per_region == {
AWS_REGION_EU_WEST_1: [],
}
@@ -327,19 +416,15 @@ class TestSecurityHub:
generate_finding_output(status="FAIL", region=AWS_REGION_EU_WEST_2),
]
asff = ASFF(findings=findings)
session = set_mocked_session(AWS_REGION_EU_WEST_1)
security_hub_findings = filter_security_hub_findings_per_region(
asff.data,
False,
[],
enabled_regions,
security_hub = SecurityHub(
aws_session=session.Session(
region_name=AWS_REGION_EU_WEST_1,
),
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_security_hub_available_regions=enabled_regions,
findings=asff.data,
)
assert (
batch_send_to_security_hub(
security_hub_findings,
session,
)
== 2
)
assert security_hub.batch_send_to_security_hub() == 2