feat(aws): add static credentials for S3 and SH (#7322)

2026-07-23 12:31:54 +00:00 · 2025-05-06 17:55:53 +02:00
parent d823b2b9de
commit fe5a78e4d4
8 changed files with 398 additions and 8 deletions
@@ -49,6 +49,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 - Add new check for MailTips full enabled for Exchange in M365 [(#7637)](https://github.com/prowler-cloud/prowler/pull/7637)
 - Add new check for Comprehensive Attachments Filter Applied for Defender in M365 [(#7661)](https://github.com/prowler-cloud/prowler/pull/7661)
 - Modified check `exchange_mailbox_properties_auditing_enabled` to make it configurable [(#7662)](https://github.com/prowler-cloud/prowler/pull/7662)
+- Add support for static credentials for sending findings to Amazon S3 and AWS Security Hub [(#7322)](https://github.com/prowler-cloud/prowler/pull/7322)

 ### Fixed

@@ -193,6 +193,7 @@ class AwsProvider(Provider):
        ######## AWS Session
        logger.info("Generating original session ...")

+        # TODO: Use AwsSetUpSession ?????
        # Configure the initial AWS Session using the local credentials: profile or environment variables
        aws_session = self.setup_session(
            mfa=mfa,
@@ -1,8 +1,8 @@
 import tempfile
 from os import path
 from tempfile import NamedTemporaryFile
+from typing import Optional

-from boto3 import Session
 from botocore import exceptions

 from prowler.lib.logger import logger
@@ -14,6 +14,8 @@ from prowler.providers.aws.lib.s3.exceptions.exceptions import (
    S3InvalidBucketNameError,
    S3TestConnectionError,
 )
+from prowler.providers.aws.lib.session.aws_set_up_session import AwsSetUpSession
+from prowler.providers.aws.models import AWSIdentityInfo, AWSSession
 from prowler.providers.common.models import Connection


@@ -33,22 +35,68 @@ class S3:
    - send_to_bucket: Sends the provided outputs to the S3 bucket.
    """

-    _session: Session
+    _session: AWSSession
+    _identity: AWSIdentityInfo
    _bucket_name: str
    _output_directory: str

    def __init__(
-        self, session: Session, bucket_name: str, output_directory: str
+        self,
+        bucket_name: str,
+        output_directory: str,
+        session: AWSSession = None,
+        role_arn: str = None,
+        session_duration: int = None,
+        external_id: str = None,
+        role_session_name: str = None,
+        mfa: bool = None,
+        profile: str = None,
+        aws_access_key_id: str = None,
+        aws_secret_access_key: str = None,
+        aws_session_token: Optional[str] = None,
+        retries_max_attempts: int = 3,
+        regions: set = set(),
    ) -> None:
        """
        Initializes a new instance of the `S3` class.

-        Parameters:
-        - session: An instance of the `Session` class representing the AWS session.
+        Args:
+        - session: An instance of the `AWSSession` class representing the AWS session.
        - bucket_name: A string representing the name of the S3 bucket.
        - output_directory: A string representing the output directory path.
+        - role_arn: The ARN of the IAM role to assume.
+        - session_duration: The duration of the session in seconds, between 900 and 43200.
+        - external_id: The external ID to use when assuming the IAM role.
+        - role_session_name: The name of the session when assuming the IAM role.
+        - mfa: A boolean indicating whether MFA is enabled.
+        - profile: The name of the AWS CLI profile to use.
+        - aws_access_key_id: The AWS access key ID.
+        - aws_secret_access_key: The AWS secret access key.
+        - aws_session_token: The AWS session token, optional.
+        - retries_max_attempts: The maximum number of retries for the AWS client.
+        - regions: A set of regions to audit.
+
+        Returns:
+        - None
        """
-        self._session = session.client(__class__.__name__.lower())
+        if session:
+            self._session = session.client(__class__.__name__.lower())
+        else:
+            aws_setup_session = AwsSetUpSession(
+                role_arn=role_arn,
+                session_duration=session_duration,
+                external_id=external_id,
+                role_session_name=role_session_name,
+                mfa=mfa,
+                profile=profile,
+                aws_access_key_id=aws_access_key_id,
+                aws_secret_access_key=aws_secret_access_key,
+                aws_session_token=aws_session_token,
+                retries_max_attempts=retries_max_attempts,
+                regions=regions,
+            )
+            self._session = aws_setup_session._session
+
        self._bucket_name = bucket_name
        self._output_directory = output_directory

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from typing import Optional

 from boto3 import Session
 from botocore.client import ClientError
@@ -11,6 +12,7 @@ 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.common.models import Connection

 SECURITY_HUB_INTEGRATION_NAME = "prowler/prowler"
@@ -58,14 +60,63 @@ class SecurityHub:

    def __init__(
        self,
-        aws_session: Session,
        aws_account_id: str,
        aws_partition: str,
+        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,
+        external_id: str = None,
+        role_session_name: str = None,
+        mfa: bool = None,
+        profile: str = None,
+        aws_access_key_id: str = None,
+        aws_secret_access_key: str = None,
+        aws_session_token: Optional[str] = None,
+        retries_max_attempts: int = 3,
+        regions: set = set(),
    ) -> "SecurityHub":
-        self._session = aws_session
+        """
+        Initializes the SecurityHub object with the necessary attributes.
+
+        Args:
+        - aws_session (Session): AWS session object for authentication and communication with AWS services.
+        - aws_account_id (str): AWS account ID associated with the SecurityHub instance.
+        - aws_partition (str): AWS partition (e.g., aws, aws-cn, aws-us-gov) where SecurityHub is deployed.
+        - findings (list[AWSSecurityFindingFormat]): List of findings to filter and send to Security Hub.
+        - aws_security_hub_available_regions (list[str]): List of regions where Security Hub is available.
+        - send_only_fails (bool): Flag indicating whether to send only findings with status 'FAILED'.
+        - role_arn: The ARN of the IAM role to assume.
+        - session_duration: The duration of the session in seconds, between 900 and 43200.
+        - external_id: The external ID to use when assuming the IAM role.
+        - role_session_name: The name of the session when assuming the IAM role.
+        - mfa: A boolean indicating whether MFA is enabled.
+        - profile: The name of the AWS CLI profile to use.
+        - aws_access_key_id: The AWS access key ID.
+        - aws_secret_access_key: The AWS secret access key.
+        - aws_session_token: The AWS session token, optional.
+        - retries_max_attempts: The maximum number of retries for the AWS client.
+        - regions: A set of regions to audit.
+        """
+        if aws_session:
+            self._session = aws_session
+        else:
+            aws_setup_session = AwsSetUpSession(
+                role_arn=role_arn,
+                session_duration=session_duration,
+                external_id=external_id,
+                role_session_name=role_session_name,
+                mfa=mfa,
+                profile=profile,
+                aws_access_key_id=aws_access_key_id,
+                aws_secret_access_key=aws_secret_access_key,
+                aws_session_token=aws_session_token,
+                retries_max_attempts=retries_max_attempts,
+                regions=regions,
+            )
+            self._session = aws_setup_session._session
        self._aws_account_id = aws_account_id
        self._aws_partition = aws_partition

@@ -0,0 +1,201 @@
+from typing import Optional
+
+from prowler.lib.logger import logger
+from prowler.providers.aws.aws_provider import (
+    AwsProvider,
+    get_aws_region_for_sts,
+    parse_iam_credentials_arn,
+)
+from prowler.providers.aws.models import (
+    AWSAssumeRoleConfiguration,
+    AWSAssumeRoleInfo,
+    AWSIdentityInfo,
+    AWSSession,
+)
+
+
+class AwsSetUpSession:
+    """
+    A class to set up the AWS session.
+
+    Attributes:
+    - _session: An instance of the AWSSession class.
+
+    Methods:
+    - __init__: The constructor for the AwsSetUpSession class.
+    """
+
+    _session: AWSSession
+    _identity: AWSIdentityInfo
+
+    def __init__(
+        self,
+        role_arn: str = None,
+        session_duration: int = None,
+        external_id: str = None,
+        role_session_name: str = None,
+        mfa: bool = None,
+        profile: str = None,
+        aws_access_key_id: str = None,
+        aws_secret_access_key: str = None,
+        aws_session_token: Optional[str] = None,
+        retries_max_attempts: int = 3,
+        regions: set = set(),
+    ) -> None:
+        """
+        The constructor for the AwsSetUpSession class.
+
+        Parameters:
+        - role_arn: The ARN of the IAM role to assume.
+        - session_duration: The duration of the session in seconds, between 900 and 43200.
+        - external_id: The external ID to use when assuming the IAM role.
+        - role_session_name: The name of the session when assuming the IAM role.
+        - mfa: A boolean indicating whether MFA is enabled.
+        - profile: The name of the AWS CLI profile to use.
+        - aws_access_key_id: The AWS access key ID.
+        - aws_secret_access_key: The AWS secret access key.
+        - aws_session_token: The AWS session token, optional.
+        - retries_max_attempts: The maximum number of retries for the AWS client.
+        - regions: A set of regions to audit.
+
+        Returns:
+
+        An instance of the AwsSetUpSession class.
+        """
+
+        validate_arguments(
+            role_arn=role_arn,
+            session_duration=session_duration,
+            external_id=external_id,
+            role_session_name=role_session_name,
+            profile=profile,
+            aws_access_key_id=aws_access_key_id,
+            aws_secret_access_key=aws_secret_access_key,
+        )
+        # Setup the AWS session
+        aws_session = AwsProvider.setup_session(
+            mfa=mfa,
+            profile=profile,
+            aws_access_key_id=aws_access_key_id,
+            aws_secret_access_key=aws_secret_access_key,
+            aws_session_token=aws_session_token,
+        )
+        session_config = AwsProvider.set_session_config(retries_max_attempts)
+        self._session = AWSSession(
+            current_session=aws_session,
+            session_config=session_config,
+            original_session=aws_session,
+        )
+
+        ######## Validate AWS credentials
+        # After the session is created, validate it
+        logger.info("Validating credentials ...")
+        sts_region = get_aws_region_for_sts(
+            self._session.current_session.region_name, regions
+        )
+
+        # Validate the credentials
+        caller_identity = AwsProvider.validate_credentials(
+            session=self._session.current_session,
+            aws_region=sts_region,
+        )
+
+        logger.info("Credentials validated")
+        ########
+
+        ######## AWS Provider Identity
+        # Get profile region
+        profile_region = AwsProvider.get_profile_region(self._session.current_session)
+
+        # Set identity
+        self._identity = AwsProvider.set_identity(
+            caller_identity=caller_identity,
+            profile=profile,
+            regions=regions,
+            profile_region=profile_region,
+        )
+        ########
+
+        ######## AWS Session with Assume Role (if needed)
+        if role_arn:
+            # Validate the input role
+            valid_role_arn = parse_iam_credentials_arn(role_arn)
+            # Set assume IAM Role information
+            assumed_role_information = AWSAssumeRoleInfo(
+                role_arn=valid_role_arn,
+                session_duration=session_duration,
+                external_id=external_id,
+                mfa_enabled=mfa,
+                role_session_name=role_session_name,
+                sts_region=sts_region,
+            )
+            # Assume the IAM Role
+            logger.info(f"Assuming role: {assumed_role_information.role_arn.arn}")
+            assumed_role_credentials = self.assume_role(
+                self._session.current_session,
+                assumed_role_information,
+            )
+            logger.info(f"IAM Role assumed: {assumed_role_information.role_arn.arn}")
+
+            assumed_role_configuration = AWSAssumeRoleConfiguration(
+                info=assumed_role_information, credentials=assumed_role_credentials
+            )
+            # Store the assumed role configuration since it'll be needed to refresh the credentials
+            self._assumed_role_configuration = assumed_role_configuration
+
+            # Store a new current session using the assumed IAM Role
+            self._session.current_session = self.setup_assumed_session(
+                assumed_role_configuration.credentials
+            )
+            logger.info("Audit session is the new session created assuming an IAM Role")
+
+            # Modify identity for the IAM Role assumed since this will be the identity to audit with
+            logger.info("Setting new identity for the AWS IAM Role assumed")
+            self._identity.account = assumed_role_configuration.info.role_arn.account_id
+            self._identity.partition = (
+                assumed_role_configuration.info.role_arn.partition
+            )
+            self._identity.account_arn = f"arn:{assumed_role_configuration.info.role_arn.partition}:iam::{assumed_role_configuration.info.role_arn.account_id}:root"
+        ########
+
+
+def validate_arguments(
+    role_arn: str = None,
+    session_duration: int = None,
+    external_id: str = None,
+    role_session_name: str = None,
+    profile: str = None,
+    aws_access_key_id: str = None,
+    aws_secret_access_key: str = None,
+) -> None:
+    """
+    Validate the arguments provided to the S3 class."
+
+    Parameters:
+    - role_arn: The ARN of the IAM role to assume.
+    - session_duration: The duration of the session in seconds, between 900 and 43200.
+    - external_id: The external ID to use when assuming the IAM role.
+    - role_session_name: The name of the session when assuming the IAM role.
+    - mfa: A boolean indicating whether MFA is enabled.
+    - profile: The name of the AWS CLI profile to use.
+    - aws_access_key_id: The AWS access key ID.
+    - aws_secret_access_key: The AWS secret access key.
+    - aws_session_token: The AWS session token, optional.
+    - retries_max_attempts: The maximum number of retries for the AWS client.
+    - regions: A set of regions to audit.
+    """
+
+    if role_arn:
+        if not session_duration or not external_id or not role_session_name:
+            raise ValueError(
+                "If a role ARN is provided, a session duration, an external ID, and a role session name are required."
+            )
+    else:
+        if session_duration or external_id or role_session_name:
+            raise ValueError(
+                "If a session duration, an external ID, or a role session name is provided, a role ARN is required."
+            )
+        if not profile and not aws_access_key_id and not aws_secret_access_key:
+            raise ValueError(
+                "If no role ARN is provided, a profile, an AWS access key ID, or an AWS secret access key is required."
+            )
@@ -344,3 +344,46 @@ class TestS3:
            assert s3 is not None
            assert s3.is_connected is False
            assert s3.error is not None
+
+    @mock_aws
+    def test_init_without_session(self):
+        with pytest.raises(ValueError) as e:
+            S3(
+                session=None,
+                bucket_name=S3_BUCKET_NAME,
+                output_directory=CURRENT_DIRECTORY,
+            )
+
+        assert (
+            str(e.value)
+            == "If no role ARN is provided, a profile, an AWS access key ID, or an AWS secret access key is required."
+        )
+
+    @mock_aws
+    def test_init_without_session_but_role_arn(self):
+        with pytest.raises(ValueError) as e:
+            S3(
+                session=None,
+                bucket_name=S3_BUCKET_NAME,
+                output_directory=CURRENT_DIRECTORY,
+                role_arn="arn:aws:iam::123456789012:role/role_name",
+            )
+
+        assert (
+            str(e.value)
+            == "If a role ARN is provided, a session duration, an external ID, and a role session name are required."
+        )
+
+    @mock_aws
+    def test_init_without_session_and_role_arn_but_session_duration(self):
+        with pytest.raises(ValueError) as e:
+            S3(
+                session=None,
+                bucket_name=S3_BUCKET_NAME,
+                output_directory=CURRENT_DIRECTORY,
+                session_duration=3600,
+            )
+        assert (
+            str(e.value)
+            == "If a session duration, an external ID, or a role session name is provided, a role ARN is required."
+        )
@@ -2,6 +2,7 @@ import re
 from logging import WARNING

 import botocore
+import pytest
 from boto3 import session
 from botocore.client import ClientError
 from mock import patch
@@ -521,3 +522,47 @@ class TestSecurityHub:

        assert connection.is_connected is False
        assert isinstance(connection.error, SecurityHubNoEnabledRegionsError)
+
+    def test_init_without_session(self):
+        with pytest.raises(ValueError) as e:
+            SecurityHub(
+                aws_session=None,
+                aws_account_id=AWS_ACCOUNT_NUMBER,
+                aws_partition=AWS_COMMERCIAL_PARTITION,
+                aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
+            )
+
+        assert (
+            str(e.value)
+            == "If no role ARN is provided, a profile, an AWS access key ID, or an AWS secret access key is required."
+        )
+
+    def test_init_without_session_but_role_arn(self):
+        with pytest.raises(ValueError) as e:
+            SecurityHub(
+                aws_session=None,
+                aws_account_id=AWS_ACCOUNT_NUMBER,
+                aws_partition=AWS_COMMERCIAL_PARTITION,
+                aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
+                role_arn="arn:aws:iam::123456789012:role/my-role",
+            )
+
+        assert (
+            str(e.value)
+            == "If a role ARN is provided, a session duration, an external ID, and a role session name are required."
+        )
+
+    def test_init_without_session_and_role_arn_but_session_duration(self):
+        with pytest.raises(ValueError) as e:
+            SecurityHub(
+                aws_session=None,
+                aws_account_id=AWS_ACCOUNT_NUMBER,
+                aws_partition=AWS_COMMERCIAL_PARTITION,
+                aws_security_hub_available_regions=[AWS_REGION_EU_WEST_1],
+                session_duration=3600,
+            )
+
+        assert (
+            str(e.value)
+            == "If a session duration, an external ID, or a role session name is provided, a role ARN is required."
+        )