diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 379e9946c7..18e401937a 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -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) diff --git a/prowler/providers/aws/lib/security_hub/security_hub.py b/prowler/providers/aws/lib/security_hub/security_hub.py index 00a1d1c044..4c894a026f 100644 --- a/prowler/providers/aws/lib/security_hub/security_hub.py +++ b/prowler/providers/aws/lib/security_hub/security_hub.py @@ -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(), + ) 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 3cb8119316..17dec4505c 100644 --- a/tests/providers/aws/lib/security_hub/security_hub_test.py +++ b/tests/providers/aws/lib/security_hub/security_hub_test.py @@ -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