mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 04:51:51 +00:00
fix(securityhub): resolve TypeError from Python3.9 (#8619)
Co-authored-by: Hugo Pereira Brito <101209179+HugoPBrito@users.noreply.github.com>
(cherry picked from commit 79450d6977)
# Conflicts:
# prowler/CHANGELOG.md
This commit is contained in:
@@ -1,6 +1,31 @@
|
||||
# Prowler SDK Changelog
|
||||
|
||||
All notable changes to the **Prowler SDK** are documented in this file.
|
||||
<<<<<<< HEAD
|
||||
=======
|
||||
## [v5.12.0] (Prowler UNRELEASED)
|
||||
|
||||
### Added
|
||||
- Get Jira Project's metadata [(#8630)](https://github.com/prowler-cloud/prowler/pull/8630)
|
||||
- Add more fields for the Jira ticket and handle custom fields errors [(#8601)](https://github.com/prowler-cloud/prowler/pull/8601)
|
||||
- Get Jira projects from test_connection [(#8634)](https://github.com/prowler-cloud/prowler/pull/8634)
|
||||
- `AdditionalUrls` field in CheckMetadata [(#8590)](https://github.com/prowler-cloud/prowler/pull/8590)
|
||||
- Support color for MANUAL finidngs in Jira tickets [(#8642)](https://github.com/prowler-cloud/prowler/pull/8642)
|
||||
|
||||
### Changed
|
||||
|
||||
### Fixed
|
||||
- Renamed `AdditionalUrls` to `AdditionalURLs` field in CheckMetadata [(#8639)](https://github.com/prowler-cloud/prowler/pull/8639)
|
||||
|
||||
---
|
||||
|
||||
## [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)
|
||||
|
||||
---
|
||||
>>>>>>> 79450d697 (fix(securityhub): resolve TypeError from Python3.9 (#8619))
|
||||
|
||||
## [v5.11.0] (Prowler v5.11.0)
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user