chore(securityhub): add static credentials and role assumption support (#8539)

Co-authored-by: Adrián Jesús Peña Rodríguez <adrianjpr@gmail.com>
This commit is contained in:
Sergio Garcia
2025-08-22 11:58:35 +02:00
committed by GitHub
parent 2167683851
commit ea6d04ed3a
3 changed files with 1037 additions and 31 deletions
+1
View File
@@ -11,6 +11,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- Bedrock AgentCore privilege escalation combination for AWS provider [(#8526)](https://github.com/prowler-cloud/prowler/pull/8526)
- Add User Email and APP name/installations information in GitHub provider [(#8501)](https://github.com/prowler-cloud/prowler/pull/8501)
- Remove standalone iam:PassRole from privesc detection and add missing patterns [(#8530)](https://github.com/prowler-cloud/prowler/pull/8530)
- Support session/profile/role/static credentials in Security Hub integration [(#8539)](https://github.com/prowler-cloud/prowler/pull/8539)
- `eks_cluster_deletion_protection_enabled` check for AWS provider [(#8536)](https://github.com/prowler-cloud/prowler/pull/8536)
- ECS privilege escalation patterns (StartTask and RunTask) for AWS provider [(#8541)](https://github.com/prowler-cloud/prowler/pull/8541)
- Resource Explorer enumeration v2 API actions in `cloudtrail_threat_detection_enumeration` check [(#8557)](https://github.com/prowler-cloud/prowler/pull/8557)
@@ -1,18 +1,46 @@
import os
from dataclasses import dataclass
from typing import Optional
from boto3 import Session
from botocore.client import ClientError
from botocore.exceptions import NoCredentialsError, ProfileNotFound
from prowler.config.config import timestamp_utc
from prowler.lib.logger import logger
from prowler.lib.outputs.asff.asff import AWSSecurityFindingFormat
from prowler.providers.aws.aws_provider import AwsProvider
from prowler.providers.aws.config import (
AWS_STS_GLOBAL_ENDPOINT_REGION,
ROLE_SESSION_NAME,
)
from prowler.providers.aws.exceptions.exceptions import (
AWSAccessKeyIDInvalidError,
AWSArgumentTypeValidationError,
AWSAssumeRoleError,
AWSIAMRoleARNEmptyResourceError,
AWSIAMRoleARNInvalidAccountIDError,
AWSIAMRoleARNInvalidResourceTypeError,
AWSIAMRoleARNPartitionEmptyError,
AWSIAMRoleARNRegionNotEmtpyError,
AWSIAMRoleARNServiceNotIAMnorSTSError,
AWSNoCredentialsError,
AWSProfileNotFoundError,
AWSSecretAccessKeyInvalidError,
AWSSessionTokenExpiredError,
AWSSetUpSessionError,
)
from prowler.providers.aws.lib.arguments.arguments import (
validate_role_session_name,
validate_session_duration,
)
from prowler.providers.aws.lib.arn.arn import parse_iam_credentials_arn
from prowler.providers.aws.lib.security_hub.exceptions.exceptions import (
SecurityHubInvalidRegionError,
SecurityHubNoEnabledRegionsError,
)
from prowler.providers.aws.lib.session.aws_set_up_session import AwsSetUpSession
from prowler.providers.aws.models import AWSAssumeRoleInfo
from prowler.providers.common.models import Connection
SECURITY_HUB_INTEGRATION_NAME = "prowler/prowler"
@@ -26,10 +54,12 @@ class SecurityHubConnection(Connection):
Attributes:
enabled_regions (set): Set of regions where Security Hub is enabled.
disabled_regions (set): Set of regions where Security Hub is disabled.
partition (str): AWS partition (e.g., aws, aws-cn, aws-us-gov) where SecurityHub is deployed.
"""
enabled_regions: set = None
disabled_regions: set = None
partition: str = ""
class SecurityHub:
@@ -61,15 +91,15 @@ class SecurityHub:
def __init__(
self,
aws_account_id: str,
aws_partition: str,
aws_partition: str = None,
aws_session: Session = None,
findings: list[AWSSecurityFindingFormat] = [],
aws_security_hub_available_regions: list[str] = [],
send_only_fails: bool = False,
role_arn: str = None,
session_duration: int = None,
session_duration: int = 3600,
external_id: str = None,
role_session_name: str = None,
role_session_name: str = ROLE_SESSION_NAME,
mfa: bool = None,
profile: str = None,
aws_access_key_id: str = None,
@@ -116,8 +146,12 @@ class SecurityHub:
retries_max_attempts=retries_max_attempts,
regions=regions,
)
self._session = aws_setup_session._session
self._session = aws_setup_session._session.current_session
self._aws_account_id = aws_account_id
if not aws_partition:
aws_partition = AwsProvider.validate_credentials(
self._session, AWS_STS_GLOBAL_ENDPOINT_REGION
).arn.partition
self._aws_partition = aws_partition
self._enabled_regions = None
@@ -126,7 +160,7 @@ class SecurityHub:
if aws_security_hub_available_regions:
self._enabled_regions = self.verify_enabled_per_region(
aws_security_hub_available_regions,
aws_session,
self._session,
aws_account_id,
aws_partition,
)
@@ -363,34 +397,91 @@ class SecurityHub:
@staticmethod
def test_connection(
session: Session,
aws_account_id: str,
aws_partition: str,
aws_partition: str = None,
regions: set = None,
raise_on_exception: bool = True,
profile: str = None,
aws_region: str = AWS_STS_GLOBAL_ENDPOINT_REGION,
role_arn: str = None,
role_session_name: str = ROLE_SESSION_NAME,
session_duration: int = 3600,
external_id: str = None,
mfa_enabled: bool = False,
aws_access_key_id: str = None,
aws_secret_access_key: str = None,
aws_session_token: Optional[str] = None,
) -> SecurityHubConnection:
"""
Test the connection to AWS Security Hub by checking if Security Hub is enabled in the provided region
and if the Prowler integration is active.
Args:
session (Session): AWS session to use for authentication.
regions (set): Set of regions to check for Security Hub integration.
aws_account_id (str): AWS account ID to check for Prowler integration.
aws_partition (str): AWS partition (e.g., aws, aws-cn, aws-us-gov).
regions (set): Set of regions to check for Security Hub integration.
raise_on_exception (bool): Whether to raise an exception if an error occurs.
profile (str): AWS profile name to use for authentication.
aws_region (str): AWS region to use for the session.
role_arn (str): ARN of the IAM role to assume.
role_session_name (str): Name for the role session.
session_duration (int): Duration of the role session in seconds.
external_id (str): External ID to use when assuming the role.
mfa_enabled (bool): Whether MFA is enabled.
aws_access_key_id (str): AWS access key ID.
aws_secret_access_key (str): AWS secret access key.
aws_session_token (str): AWS session token.
Returns:
Connection: An object that contains the result of the test connection operation.
SecurityHubConnection: An object that contains the result of the test connection operation.
- is_connected (bool): Indicates whether the connection was successful.
- error (Exception): An exception object if an error occurs during the connection test.
enabled_regions (set): Set of regions where Security Hub is enabled.
disabled_regions (set): Set of regions where Security Hub is disabled.
- enabled_regions (set): Set of regions where Security Hub is enabled.
- disabled_regions (set): Set of regions where Security Hub is disabled.
"""
try:
disabled_regions = set()
enabled_regions = set()
# Set up AWS session
session = AwsProvider.setup_session(
mfa=mfa_enabled,
profile=profile,
aws_access_key_id=aws_access_key_id,
aws_secret_access_key=aws_secret_access_key,
aws_session_token=aws_session_token,
)
if not aws_partition:
aws_partition = AwsProvider.validate_credentials(
session, aws_region
).arn.partition
# Handle role assumption if role_arn is provided
if role_arn:
session_duration = validate_session_duration(session_duration)
role_session_name = validate_role_session_name(
role_session_name or ROLE_SESSION_NAME
)
role_arn = parse_iam_credentials_arn(role_arn)
assumed_role_information = AWSAssumeRoleInfo(
role_arn=role_arn,
session_duration=session_duration,
external_id=external_id,
mfa_enabled=mfa_enabled,
role_session_name=role_session_name,
)
assumed_role_credentials = AwsProvider.assume_role(
session,
assumed_role_information,
)
session = Session(
aws_access_key_id=assumed_role_credentials.aws_access_key_id,
aws_secret_access_key=assumed_role_credentials.aws_secret_access_key,
aws_session_token=assumed_role_credentials.aws_session_token,
region_name=aws_region,
profile_name=profile,
)
all_regions = AwsProvider.get_available_aws_service_regions(
service="securityhub", partition=aws_partition
)
@@ -427,6 +518,7 @@ class SecurityHub:
error=None,
enabled_regions=enabled_regions,
disabled_regions=disabled_regions,
partition=aws_partition,
)
if len(enabled_regions) == 0:
@@ -454,10 +546,206 @@ class SecurityHub:
error=None,
enabled_regions=enabled_regions,
disabled_regions=disabled_regions,
partition=aws_partition,
)
except AWSSetUpSessionError as setup_session_error:
logger.error(
f"{setup_session_error.__class__.__name__}[{setup_session_error.__traceback__.tb_lineno}]: {setup_session_error}"
)
if raise_on_exception:
raise setup_session_error
return SecurityHubConnection(
is_connected=False,
error=setup_session_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSArgumentTypeValidationError as validation_error:
logger.error(
f"{validation_error.__class__.__name__}[{validation_error.__traceback__.tb_lineno}]: {validation_error}"
)
if raise_on_exception:
raise validation_error
return SecurityHubConnection(
is_connected=False,
error=validation_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSIAMRoleARNRegionNotEmtpyError as arn_region_not_empty_error:
logger.error(
f"{arn_region_not_empty_error.__class__.__name__}[{arn_region_not_empty_error.__traceback__.tb_lineno}]: {arn_region_not_empty_error}"
)
if raise_on_exception:
raise arn_region_not_empty_error
return SecurityHubConnection(
is_connected=False,
error=arn_region_not_empty_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSIAMRoleARNPartitionEmptyError as arn_partition_empty_error:
logger.error(
f"{arn_partition_empty_error.__class__.__name__}[{arn_partition_empty_error.__traceback__.tb_lineno}]: {arn_partition_empty_error}"
)
if raise_on_exception:
raise arn_partition_empty_error
return SecurityHubConnection(
is_connected=False,
error=arn_partition_empty_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSIAMRoleARNServiceNotIAMnorSTSError as arn_service_not_iam_sts_error:
logger.error(
f"{arn_service_not_iam_sts_error.__class__.__name__}[{arn_service_not_iam_sts_error.__traceback__.tb_lineno}]: {arn_service_not_iam_sts_error}"
)
if raise_on_exception:
raise arn_service_not_iam_sts_error
return SecurityHubConnection(
is_connected=False,
error=arn_service_not_iam_sts_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSIAMRoleARNInvalidAccountIDError as arn_invalid_account_id_error:
logger.error(
f"{arn_invalid_account_id_error.__class__.__name__}[{arn_invalid_account_id_error.__traceback__.tb_lineno}]: {arn_invalid_account_id_error}"
)
if raise_on_exception:
raise arn_invalid_account_id_error
return SecurityHubConnection(
is_connected=False,
error=arn_invalid_account_id_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSIAMRoleARNInvalidResourceTypeError as arn_invalid_resource_type_error:
logger.error(
f"{arn_invalid_resource_type_error.__class__.__name__}[{arn_invalid_resource_type_error.__traceback__.tb_lineno}]: {arn_invalid_resource_type_error}"
)
if raise_on_exception:
raise arn_invalid_resource_type_error
return SecurityHubConnection(
is_connected=False,
error=arn_invalid_resource_type_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSIAMRoleARNEmptyResourceError as arn_empty_resource_error:
logger.error(
f"{arn_empty_resource_error.__class__.__name__}[{arn_empty_resource_error.__traceback__.tb_lineno}]: {arn_empty_resource_error}"
)
if raise_on_exception:
raise arn_empty_resource_error
return SecurityHubConnection(
is_connected=False,
error=arn_empty_resource_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSAssumeRoleError as assume_role_error:
logger.error(
f"{assume_role_error.__class__.__name__}[{assume_role_error.__traceback__.tb_lineno}]: {assume_role_error}"
)
if raise_on_exception:
raise assume_role_error
return SecurityHubConnection(
is_connected=False,
error=assume_role_error,
enabled_regions=set(),
disabled_regions=set(),
)
except ProfileNotFound as profile_not_found_error:
logger.error(
f"AWSProfileNotFoundError[{profile_not_found_error.__traceback__.tb_lineno}]: {profile_not_found_error}"
)
if raise_on_exception:
raise AWSProfileNotFoundError(
file=os.path.basename(__file__),
original_exception=profile_not_found_error,
) from profile_not_found_error
return SecurityHubConnection(
is_connected=False,
error=profile_not_found_error,
enabled_regions=set(),
disabled_regions=set(),
)
except NoCredentialsError as no_credentials_error:
logger.error(
f"AWSNoCredentialsError[{no_credentials_error.__traceback__.tb_lineno}]: {no_credentials_error}"
)
if raise_on_exception:
raise AWSNoCredentialsError(
file=os.path.basename(__file__),
original_exception=no_credentials_error,
) from no_credentials_error
return SecurityHubConnection(
is_connected=False,
error=no_credentials_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSAccessKeyIDInvalidError as access_key_id_invalid_error:
logger.error(
f"{access_key_id_invalid_error.__class__.__name__}[{access_key_id_invalid_error.__traceback__.tb_lineno}]: {access_key_id_invalid_error}"
)
if raise_on_exception:
raise access_key_id_invalid_error
return SecurityHubConnection(
is_connected=False,
error=access_key_id_invalid_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSSecretAccessKeyInvalidError as secret_access_key_invalid_error:
logger.error(
f"{secret_access_key_invalid_error.__class__.__name__}[{secret_access_key_invalid_error.__traceback__.tb_lineno}]: {secret_access_key_invalid_error}"
)
if raise_on_exception:
raise secret_access_key_invalid_error
return SecurityHubConnection(
is_connected=False,
error=secret_access_key_invalid_error,
enabled_regions=set(),
disabled_regions=set(),
)
except AWSSessionTokenExpiredError as session_token_expired:
logger.error(
f"{session_token_expired.__class__.__name__}[{session_token_expired.__traceback__.tb_lineno}]: {session_token_expired}"
)
if raise_on_exception:
raise session_token_expired
return SecurityHubConnection(
is_connected=False,
error=session_token_expired,
enabled_regions=set(),
disabled_regions=set(),
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
raise error
if raise_on_exception:
raise error
return SecurityHubConnection(
is_connected=False,
error=error,
enabled_regions=set(),
disabled_regions=set(),
)
@@ -412,14 +412,12 @@ class TestSecurityHub:
@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
def test_security_hub_test_connection_success(self):
session_mock = session.Session(region_name=AWS_REGION_EU_WEST_1)
# Test successful connection
connection = SecurityHub.test_connection(
session=session_mock,
regions={AWS_REGION_EU_WEST_1},
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
regions={AWS_REGION_EU_WEST_1},
raise_on_exception=False,
)
@@ -444,14 +442,11 @@ class TestSecurityHub:
error_response, operation_name
)
session_mock = session.Session(region_name=AWS_REGION_EU_WEST_1)
# Test connection failure due to invalid access
connection = SecurityHub.test_connection(
session=session_mock,
regions={AWS_REGION_EU_WEST_1},
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
regions={AWS_REGION_EU_WEST_1},
raise_on_exception=False,
)
@@ -468,14 +463,11 @@ class TestSecurityHub:
"ProductSubscriptions": []
}
session_mock = session.Session(region_name=AWS_REGION_EU_WEST_1)
# Test connection failure due to missing Prowler subscription
connection = SecurityHub.test_connection(
session=session_mock,
regions={AWS_REGION_EU_WEST_1},
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
regions={AWS_REGION_EU_WEST_1},
raise_on_exception=False,
)
@@ -489,14 +481,11 @@ class TestSecurityHub:
# Mock unexpected exception
mock_security_hub_client.side_effect = Exception("Unexpected error")
session_mock = session.Session(region_name=AWS_REGION_EU_WEST_1)
# Test connection failure due to an unexpected exception
connection = SecurityHub.test_connection(
session=session_mock,
regions={AWS_REGION_EU_WEST_1},
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
regions={AWS_REGION_EU_WEST_1},
raise_on_exception=False,
)
@@ -510,11 +499,8 @@ class TestSecurityHub:
# Mock unexpected exception
mock_security_hub_client.side_effect = Exception("Unexpected error")
session_mock = session.Session(region_name=AWS_REGION_EU_WEST_1)
# Test connection failure due to an unexpected exception
connection = SecurityHub.test_connection(
session=session_mock,
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
@@ -566,3 +552,734 @@ class TestSecurityHub:
str(e.value)
== "If no role ARN is provided, a profile, an AWS access key ID, or an AWS secret access key is required."
)
# Tests for new test_connection functionality - AWS Credential Management
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
@patch(
"prowler.providers.aws.aws_provider.AwsProvider.get_available_aws_service_regions"
)
@patch(
"prowler.providers.aws.lib.security_hub.security_hub.SecurityHub.verify_enabled_per_region"
)
def test_security_hub_test_connection_with_profile(
self, mock_verify_enabled, mock_get_regions, mock_setup_session
):
# Mock session setup
mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1)
mock_setup_session.return_value = mock_session
# Mock available regions
mock_get_regions.return_value = [AWS_REGION_EU_WEST_1]
# Mock enabled regions
mock_verify_enabled.return_value = {AWS_REGION_EU_WEST_1: mock_session}
# Test connection with profile
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
profile="test-profile",
raise_on_exception=False,
)
assert connection.is_connected is True
assert connection.error is None
mock_setup_session.assert_called_once_with(
mfa=False,
profile="test-profile",
aws_access_key_id=None,
aws_secret_access_key=None,
aws_session_token=None,
)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
@patch(
"prowler.providers.aws.aws_provider.AwsProvider.get_available_aws_service_regions"
)
@patch(
"prowler.providers.aws.lib.security_hub.security_hub.SecurityHub.verify_enabled_per_region"
)
def test_security_hub_test_connection_with_access_keys(
self, mock_verify_enabled, mock_get_regions, mock_setup_session
):
# Mock session setup
mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1)
mock_setup_session.return_value = mock_session
# Mock available regions
mock_get_regions.return_value = [AWS_REGION_EU_WEST_1]
# Mock enabled regions
mock_verify_enabled.return_value = {AWS_REGION_EU_WEST_1: mock_session}
# Test connection with access keys
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_access_key_id="test-key",
aws_secret_access_key="test-secret",
raise_on_exception=False,
)
assert connection.is_connected is True
assert connection.error is None
mock_setup_session.assert_called_once_with(
mfa=False,
profile=None,
aws_access_key_id="test-key",
aws_secret_access_key="test-secret",
aws_session_token=None,
)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
@patch(
"prowler.providers.aws.aws_provider.AwsProvider.get_available_aws_service_regions"
)
@patch(
"prowler.providers.aws.lib.security_hub.security_hub.SecurityHub.verify_enabled_per_region"
)
def test_security_hub_test_connection_with_session_token(
self, mock_verify_enabled, mock_get_regions, mock_setup_session
):
# Mock session setup
mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1)
mock_setup_session.return_value = mock_session
# Mock available regions
mock_get_regions.return_value = [AWS_REGION_EU_WEST_1]
# Mock enabled regions
mock_verify_enabled.return_value = {AWS_REGION_EU_WEST_1: mock_session}
# Test connection with session token
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_session_token="test-token",
raise_on_exception=False,
)
assert connection.is_connected is True
assert connection.error is None
mock_setup_session.assert_called_once_with(
mfa=False,
profile=None,
aws_access_key_id=None,
aws_secret_access_key=None,
aws_session_token="test-token",
)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
@patch(
"prowler.providers.aws.aws_provider.AwsProvider.get_available_aws_service_regions"
)
@patch(
"prowler.providers.aws.lib.security_hub.security_hub.SecurityHub.verify_enabled_per_region"
)
def test_security_hub_test_connection_with_mfa(
self, mock_verify_enabled, mock_get_regions, mock_setup_session
):
# Mock session setup
mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1)
mock_setup_session.return_value = mock_session
# Mock available regions
mock_get_regions.return_value = [AWS_REGION_EU_WEST_1]
# Mock enabled regions
mock_verify_enabled.return_value = {AWS_REGION_EU_WEST_1: mock_session}
# Test connection with MFA enabled
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
mfa_enabled=True,
raise_on_exception=False,
)
assert connection.is_connected is True
assert connection.error is None
mock_setup_session.assert_called_once_with(
mfa=True,
profile=None,
aws_access_key_id=None,
aws_secret_access_key=None,
aws_session_token=None,
)
# Tests for Role Assumption functionality
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
@patch("prowler.providers.aws.aws_provider.AwsProvider.assume_role")
@patch(
"prowler.providers.aws.aws_provider.AwsProvider.get_available_aws_service_regions"
)
@patch(
"prowler.providers.aws.lib.security_hub.security_hub.SecurityHub.verify_enabled_per_region"
)
def test_security_hub_test_connection_with_role_arn(
self,
mock_verify_enabled,
mock_get_regions,
mock_assume_role,
mock_setup_session,
):
# Mock initial session setup
mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1)
mock_setup_session.return_value = mock_session
# Mock assumed role credentials
from datetime import datetime, timezone
from prowler.providers.aws.models import AWSCredentials
mock_credentials = AWSCredentials(
aws_access_key_id="assumed-key",
aws_secret_access_key="assumed-secret",
aws_session_token="assumed-token",
expiration=datetime.now(timezone.utc),
)
mock_assume_role.return_value = mock_credentials
# Mock available regions
mock_get_regions.return_value = [AWS_REGION_EU_WEST_1]
# Mock enabled regions
mock_verify_enabled.return_value = {AWS_REGION_EU_WEST_1: mock_session}
# Test connection with role ARN
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
role_arn="arn:aws:iam::123456789012:role/test-role",
external_id="test-external-id",
session_duration=7200,
role_session_name="test-session",
raise_on_exception=False,
)
assert connection.is_connected is True
assert connection.error is None
mock_assume_role.assert_called_once()
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
@patch("prowler.providers.aws.aws_provider.AwsProvider.assume_role")
@patch(
"prowler.providers.aws.aws_provider.AwsProvider.get_available_aws_service_regions"
)
@patch(
"prowler.providers.aws.lib.security_hub.security_hub.SecurityHub.verify_enabled_per_region"
)
def test_security_hub_test_connection_with_role_arn_default_values(
self,
mock_verify_enabled,
mock_get_regions,
mock_assume_role,
mock_setup_session,
):
# Mock initial session setup
mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1)
mock_setup_session.return_value = mock_session
# Mock assumed role credentials
from datetime import datetime, timezone
from prowler.providers.aws.models import AWSCredentials
mock_credentials = AWSCredentials(
aws_access_key_id="assumed-key",
aws_secret_access_key="assumed-secret",
aws_session_token="assumed-token",
expiration=datetime.now(timezone.utc),
)
mock_assume_role.return_value = mock_credentials
# Mock available regions
mock_get_regions.return_value = [AWS_REGION_EU_WEST_1]
# Mock enabled regions
mock_verify_enabled.return_value = {AWS_REGION_EU_WEST_1: mock_session}
# Test connection with role ARN using default values
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
role_arn="arn:aws:iam::123456789012:role/test-role",
external_id="test-external-id",
raise_on_exception=False,
)
assert connection.is_connected is True
assert connection.error is None
mock_assume_role.assert_called_once()
# Tests for Error Handling - Session Setup Errors
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_setup_session_error(self, mock_setup_session):
from prowler.providers.aws.exceptions.exceptions import AWSSetUpSessionError
# Mock session setup error
mock_setup_session.side_effect = AWSSetUpSessionError(
file="test_file.py", original_exception=Exception("Session setup failed")
)
# Test connection failure due to session setup error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSSetUpSessionError)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_setup_session_error_raise(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import AWSSetUpSessionError
# Mock session setup error
mock_setup_session.side_effect = AWSSetUpSessionError(
file="test_file.py", original_exception=Exception("Session setup failed")
)
# Test that error is raised when raise_on_exception=True
with pytest.raises(AWSSetUpSessionError):
SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=True,
)
# Tests for Error Handling - Argument Validation Errors
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_argument_validation_error(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import (
AWSArgumentTypeValidationError,
)
# Mock session setup error
mock_setup_session.side_effect = AWSArgumentTypeValidationError(
file="test_file.py", original_exception=ValueError("Invalid argument")
)
# Test connection failure due to argument validation error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSArgumentTypeValidationError)
# Tests for Error Handling - Role ARN Validation Errors
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_role_arn_region_not_empty_error(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import (
AWSIAMRoleARNRegionNotEmtpyError,
)
# Mock session setup error
mock_setup_session.side_effect = AWSIAMRoleARNRegionNotEmtpyError(
file="test_file.py"
)
# Test connection failure due to role ARN region validation error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSIAMRoleARNRegionNotEmtpyError)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_role_arn_partition_empty_error(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import (
AWSIAMRoleARNPartitionEmptyError,
)
# Mock session setup error
mock_setup_session.side_effect = AWSIAMRoleARNPartitionEmptyError(
file="test_file.py"
)
# Test connection failure due to role ARN partition validation error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSIAMRoleARNPartitionEmptyError)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_role_arn_service_not_iam_sts_error(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import (
AWSIAMRoleARNServiceNotIAMnorSTSError,
)
# Mock session setup error
mock_setup_session.side_effect = AWSIAMRoleARNServiceNotIAMnorSTSError(
file="test_file.py"
)
# Test connection failure due to role ARN service validation error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSIAMRoleARNServiceNotIAMnorSTSError)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_role_arn_invalid_account_id_error(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import (
AWSIAMRoleARNInvalidAccountIDError,
)
# Mock session setup error
mock_setup_session.side_effect = AWSIAMRoleARNInvalidAccountIDError(
file="test_file.py"
)
# Test connection failure due to role ARN account ID validation error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSIAMRoleARNInvalidAccountIDError)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_role_arn_invalid_resource_type_error(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import (
AWSIAMRoleARNInvalidResourceTypeError,
)
# Mock session setup error
mock_setup_session.side_effect = AWSIAMRoleARNInvalidResourceTypeError(
file="test_file.py"
)
# Test connection failure due to role ARN resource type validation error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSIAMRoleARNInvalidResourceTypeError)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_role_arn_empty_resource_error(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import (
AWSIAMRoleARNEmptyResourceError,
)
# Mock session setup error
mock_setup_session.side_effect = AWSIAMRoleARNEmptyResourceError(
file="test_file.py"
)
# Test connection failure due to role ARN empty resource validation error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSIAMRoleARNEmptyResourceError)
# Tests for Error Handling - Role Assumption Errors
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_assume_role_error(self, mock_setup_session):
from prowler.providers.aws.exceptions.exceptions import AWSAssumeRoleError
# Mock session setup error
mock_setup_session.side_effect = AWSAssumeRoleError(
file="test_file.py", original_exception=Exception("Role assumption failed")
)
# Test connection failure due to role assumption error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSAssumeRoleError)
# Tests for Error Handling - Profile and Credential Errors
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_profile_not_found_error(
self, mock_setup_session
):
from botocore.exceptions import ProfileNotFound
# Mock session setup error
mock_setup_session.side_effect = ProfileNotFound(profile="test-profile")
# Test connection failure due to profile not found error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
profile="test-profile",
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, ProfileNotFound)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_profile_not_found_error_raise(
self, mock_setup_session
):
from botocore.exceptions import ProfileNotFound
from prowler.providers.aws.exceptions.exceptions import AWSProfileNotFoundError
# Mock session setup error
mock_setup_session.side_effect = ProfileNotFound(profile="test-profile")
# Test that error is raised when raise_on_exception=True
with pytest.raises(AWSProfileNotFoundError):
SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
profile="test-profile",
raise_on_exception=True,
)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_no_credentials_error(
self, mock_setup_session
):
from botocore.exceptions import NoCredentialsError
# Mock session setup error
mock_setup_session.side_effect = NoCredentialsError()
# Test connection failure due to no credentials error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, NoCredentialsError)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_no_credentials_error_raise(
self, mock_setup_session
):
from botocore.exceptions import NoCredentialsError
from prowler.providers.aws.exceptions.exceptions import AWSNoCredentialsError
# Mock session setup error
mock_setup_session.side_effect = NoCredentialsError()
# Test that error is raised when raise_on_exception=True
with pytest.raises(AWSNoCredentialsError):
SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=True,
)
# Tests for Error Handling - Access Key and Secret Key Errors
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_access_key_id_invalid_error(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import (
AWSAccessKeyIDInvalidError,
)
# Mock session setup error
mock_setup_session.side_effect = AWSAccessKeyIDInvalidError(
file="test_file.py", original_exception=ValueError("Invalid access key ID")
)
# Test connection failure due to invalid access key ID error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSAccessKeyIDInvalidError)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_secret_access_key_invalid_error(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import (
AWSSecretAccessKeyInvalidError,
)
# Mock session setup error
mock_setup_session.side_effect = AWSSecretAccessKeyInvalidError(
file="test_file.py",
original_exception=ValueError("Invalid secret access key"),
)
# Test connection failure due to invalid secret access key error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSSecretAccessKeyInvalidError)
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_session_token_expired_error(
self, mock_setup_session
):
from prowler.providers.aws.exceptions.exceptions import (
AWSSessionTokenExpiredError,
)
# Mock session setup error
mock_setup_session.side_effect = AWSSessionTokenExpiredError(
file="test_file.py", original_exception=ValueError("Session token expired")
)
# Test connection failure due to session token expired error
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, AWSSessionTokenExpiredError)
# Tests for Error Handling - Generic Exception
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_generic_exception(self, mock_setup_session):
# Mock session setup error
mock_setup_session.side_effect = Exception("Generic error")
# Test connection failure due to generic exception
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is False
assert isinstance(connection.error, Exception)
assert str(connection.error) == "Generic error"
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
def test_security_hub_test_connection_generic_exception_raise(
self, mock_setup_session
):
# Mock session setup error
mock_setup_session.side_effect = Exception("Generic error")
# Test that error is raised when raise_on_exception=True
with pytest.raises(Exception) as exc_info:
SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=True,
)
assert str(exc_info.value) == "Generic error"
# Tests for Edge Cases and Parameter Validation
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
@patch(
"prowler.providers.aws.aws_provider.AwsProvider.get_available_aws_service_regions"
)
@patch(
"prowler.providers.aws.lib.security_hub.security_hub.SecurityHub.verify_enabled_per_region"
)
def test_security_hub_test_connection_with_aws_region(
self, mock_verify_enabled, mock_get_regions, mock_setup_session
):
# Mock session setup
mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1)
mock_setup_session.return_value = mock_session
# Mock available regions
mock_get_regions.return_value = [AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
# Mock enabled regions
mock_verify_enabled.return_value = {AWS_REGION_EU_WEST_1: mock_session}
# Test connection with specific AWS region
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
aws_region=AWS_REGION_EU_WEST_1,
raise_on_exception=False,
)
assert connection.is_connected is True
assert connection.error is None
assert AWS_REGION_EU_WEST_1 in connection.enabled_regions
assert AWS_REGION_EU_WEST_2 in connection.disabled_regions
@patch("prowler.providers.aws.aws_provider.AwsProvider.setup_session")
@patch(
"prowler.providers.aws.aws_provider.AwsProvider.get_available_aws_service_regions"
)
@patch(
"prowler.providers.aws.lib.security_hub.security_hub.SecurityHub.verify_enabled_per_region"
)
def test_security_hub_test_connection_no_regions_specified(
self, mock_verify_enabled, mock_get_regions, mock_setup_session
):
# Mock session setup
mock_session = session.Session(region_name=AWS_REGION_EU_WEST_1)
mock_setup_session.return_value = mock_session
# Mock available regions
mock_get_regions.return_value = [AWS_REGION_EU_WEST_1, AWS_REGION_EU_WEST_2]
# Mock enabled regions
mock_verify_enabled.return_value = {AWS_REGION_EU_WEST_1: mock_session}
# Test connection without specifying regions
connection = SecurityHub.test_connection(
aws_account_id=AWS_ACCOUNT_NUMBER,
aws_partition=AWS_COMMERCIAL_PARTITION,
raise_on_exception=False,
)
assert connection.is_connected is True
assert connection.error is None
assert len(connection.enabled_regions) == 1
assert len(connection.disabled_regions) == 1