From 79450d6977c53b087e0497ea02aad73c63ff7093 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20Mart=C3=ADn?= Date: Wed, 3 Sep 2025 17:52:09 +0200 Subject: [PATCH] fix(securityhub): resolve TypeError from Python3.9 (#8619) Co-authored-by: Hugo Pereira Brito <101209179+HugoPBrito@users.noreply.github.com> --- prowler/CHANGELOG.md | 1 + .../aws/lib/security_hub/security_hub.py | 4 +- .../aws/lib/security_hub/security_hub_test.py | 164 ++++++++++++++++++ 3 files changed, 167 insertions(+), 2 deletions(-) diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 2e30a158e5..4c8b6996e1 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -20,6 +20,7 @@ All notable changes to the **Prowler SDK** are documented in this file. ## [v5.11.1] (Prowler UNRELEASED) ### Fixed +- TypeError from Python 3.9 in Security Hub module by updating type annotations [(#8619)](https://github.com/prowler-cloud/prowler/pull/8619) --- diff --git a/prowler/providers/aws/lib/security_hub/security_hub.py b/prowler/providers/aws/lib/security_hub/security_hub.py index b2809b25a4..5dcd0b199c 100644 --- a/prowler/providers/aws/lib/security_hub/security_hub.py +++ b/prowler/providers/aws/lib/security_hub/security_hub.py @@ -1,7 +1,7 @@ import os from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass -from typing import Optional +from typing import Optional, Union from boto3 import Session from botocore.client import ClientError @@ -219,7 +219,7 @@ class SecurityHub: session: Session, aws_account_id: str, aws_partition: str, - ) -> tuple[str, Session | None]: + ) -> tuple[str, Union[Session, None]]: """ Check if Security Hub is enabled in a specific region and if Prowler integration is active. diff --git a/tests/providers/aws/lib/security_hub/security_hub_test.py b/tests/providers/aws/lib/security_hub/security_hub_test.py index 17dec4505c..0ee74a4c30 100644 --- a/tests/providers/aws/lib/security_hub/security_hub_test.py +++ b/tests/providers/aws/lib/security_hub/security_hub_test.py @@ -1283,3 +1283,167 @@ class TestSecurityHub: assert connection.error is None assert len(connection.enabled_regions) == 1 assert len(connection.disabled_regions) == 1 + + # Tests for _check_region_security_hub static method + @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) + def test_check_region_security_hub_success(self): + # Test successful security hub check + mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1) + + region, client = SecurityHub._check_region_security_hub( + region=AWS_REGION_EU_WEST_1, + session=mock_session, + aws_account_id=AWS_ACCOUNT_NUMBER, + aws_partition=AWS_COMMERCIAL_PARTITION, + ) + + assert region == AWS_REGION_EU_WEST_1 + assert client is not None + assert hasattr(client, "_make_api_call") + + def test_check_region_security_hub_invalid_access_exception(self, caplog): + caplog.set_level(WARNING) + + with patch("boto3.Session.client") as mock_client: + error_message = ( + f"Account {AWS_ACCOUNT_NUMBER} is not subscribed to AWS Security Hub" + ) + error_code = "InvalidAccessException" + error_response = { + "Error": { + "Code": error_code, + "Message": error_message, + } + } + operation_name = "DescribeHub" + mock_client.side_effect = ClientError(error_response, operation_name) + + mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1) + + region, client = SecurityHub._check_region_security_hub( + region=AWS_REGION_EU_WEST_1, + session=mock_session, + aws_account_id=AWS_ACCOUNT_NUMBER, + aws_partition=AWS_COMMERCIAL_PARTITION, + ) + + assert region == AWS_REGION_EU_WEST_1 + assert client is None + + # Check that warning was logged for InvalidAccessException + 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), + ) + ) + assert any( + log_pattern.match(record.message) for record in caplog.records + ), "Expected log message not found" + + def test_check_region_security_hub_prowler_integration_not_enabled(self, caplog): + from logging import INFO + + caplog.set_level(INFO) + + with patch("boto3.Session.client") as mock_client: + mock_security_hub_client = mock_client.return_value + mock_security_hub_client.describe_hub.return_value = {} + mock_security_hub_client.list_enabled_products_for_import.return_value = { + "ProductSubscriptions": [] + } + + mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1) + + region, client = SecurityHub._check_region_security_hub( + region=AWS_REGION_EU_WEST_1, + session=mock_session, + aws_account_id=AWS_ACCOUNT_NUMBER, + aws_partition=AWS_COMMERCIAL_PARTITION, + ) + + assert region == AWS_REGION_EU_WEST_1 + assert client is None + + # Check that warning was logged for missing Prowler integration + assert caplog.record_tuples == [ + ( + "root", + INFO, + f"Checking if the prowler/prowler is enabled in the {AWS_REGION_EU_WEST_1} region.", + ), + ( + "root", + WARNING, + f"Security Hub is enabled in {AWS_REGION_EU_WEST_1} but Prowler integration does not accept findings. More info: https://docs.prowler.cloud/en/latest/tutorials/aws/securityhub/", + ), + ] + + def test_check_region_security_hub_other_client_error(self, caplog): + caplog.set_level(WARNING) + + with patch("boto3.Session.client") as mock_client: + error_message = "Some other error" + error_code = "SomeOtherException" + error_response = { + "Error": { + "Code": error_code, + "Message": error_message, + } + } + operation_name = "DescribeHub" + mock_client.side_effect = ClientError(error_response, operation_name) + + mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1) + + region, client = SecurityHub._check_region_security_hub( + region=AWS_REGION_EU_WEST_1, + session=mock_session, + aws_account_id=AWS_ACCOUNT_NUMBER, + aws_partition=AWS_COMMERCIAL_PARTITION, + ) + + assert region == AWS_REGION_EU_WEST_1 + assert client is None + + # Check that error was logged for other ClientError + 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), + ) + ) + assert any( + log_pattern.match(record.message) for record in caplog.records + ), "Expected log message not found" + + def test_check_region_security_hub_generic_exception(self, caplog): + caplog.set_level(WARNING) + + with patch("boto3.Session.client") as mock_client: + error_message = "Generic exception occurred" + mock_client.side_effect = Exception(error_message) + + mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1) + + region, client = SecurityHub._check_region_security_hub( + region=AWS_REGION_EU_WEST_1, + session=mock_session, + aws_account_id=AWS_ACCOUNT_NUMBER, + aws_partition=AWS_COMMERCIAL_PARTITION, + ) + + assert region == AWS_REGION_EU_WEST_1 + assert client is None + + # Check that error was logged for generic exception + log_pattern = re.compile( + r"Exception -- \[\d+\]: {error_message}".format( + error_message=re.escape(error_message), + ) + ) + assert any( + log_pattern.match(record.message) for record in caplog.records + ), "Expected log message not found"