diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 47c133e254..b723e3bd28 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -25,6 +25,7 @@ All notable changes to the **Prowler SDK** are documented in this file. - Add missing audit evidence for controls 1.1.4 and 2.5.5 for ISMS-P compliance. [(#8386)](https://github.com/prowler-cloud/prowler/pull/8386) - Use the correct @staticmethod decorator for `set_identity` and `set_session_config` methods in AwsProvider [(#8056)](https://github.com/prowler-cloud/prowler/pull/8056) - Use the correct default value for `role_session_name` and `session_duration` in AwsSetUpSession [(#8056)](https://github.com/prowler-cloud/prowler/pull/8056) +- Use the correct default value for `role_session_name` and `session_duration` in S3 [(#8417)](https://github.com/prowler-cloud/prowler/pull/8417) --- @@ -38,6 +39,8 @@ All notable changes to the **Prowler SDK** are documented in this file. - Use default tenant domain instead of first domain in list for Azure and M365 providers [(#8402)](https://github.com/prowler-cloud/prowler/pull/8402) - Avoid multiple module error calls in M365 provider [(#8353)](https://github.com/prowler-cloud/prowler/pull/8353) - Tweaks from Prowler ThreatScore in order to handle the correct reqs [(#8401)](https://github.com/prowler-cloud/prowler/pull/8401) +- Make `setup_assumed_session` static for the AWS provider [(#8419)](https://github.com/prowler-cloud/prowler/pull/8419) + --- ## [v5.9.2] (Prowler v5.9.2) diff --git a/prowler/providers/aws/aws_provider.py b/prowler/providers/aws/aws_provider.py index 14dfa6ec4e..9893f249bc 100644 --- a/prowler/providers/aws/aws_provider.py +++ b/prowler/providers/aws/aws_provider.py @@ -254,7 +254,7 @@ class AwsProvider(Provider): ) # Assume the IAM Role logger.info(f"Assuming role: {assumed_role_information.role_arn.arn}") - assumed_role_credentials = self.assume_role( + assumed_role_credentials = AwsProvider.assume_role( self._session.current_session, assumed_role_information, ) @@ -267,8 +267,10 @@ class AwsProvider(Provider): 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 + self._session.current_session = AwsProvider.setup_assumed_session( + self._identity, + assumed_role_configuration, + self._session, ) logger.info("Audit session is the new session created assuming an IAM Role") @@ -316,8 +318,10 @@ class AwsProvider(Provider): credentials=organizations_assumed_role_credentials, ) # Get a new session using the AWS Organizations IAM Role assumed - aws_organizations_session = self.setup_assumed_session( - organizations_assumed_role_configuration.credentials + aws_organizations_session = AwsProvider.setup_assumed_session( + self._identity, + organizations_assumed_role_configuration, + self._session, ) logger.info( "Generated new session for to get the AWS Organizations metadata" @@ -576,9 +580,11 @@ class AwsProvider(Provider): file=pathlib.Path(__file__).name, ) + @staticmethod def setup_assumed_session( - self, - assumed_role_credentials: AWSCredentials, + identity: AWSIdentityInfo, + assumed_role_configuration: AWSAssumeRoleConfiguration, + session: AWSSession, ) -> Session: """ Sets up an assumed session using the provided assumed role credentials. @@ -588,7 +594,9 @@ class AwsProvider(Provider): refreshing of the assumed role credentials. Args: + identity (AWSIdentityInfo): The identity information. assumed_role_credentials (AWSCredentials): The assumed role credentials. + session (AWSSession): The AWS provider session. Returns: Session: The assumed session. @@ -607,20 +615,22 @@ class AwsProvider(Provider): # that needs to be a method without arguments that retrieves a new set of fresh credentials # assuming the role again. assumed_refreshable_credentials = RefreshableCredentials( - access_key=assumed_role_credentials.aws_access_key_id, - secret_key=assumed_role_credentials.aws_secret_access_key, - token=assumed_role_credentials.aws_session_token, - expiry_time=assumed_role_credentials.expiration, - refresh_using=self.refresh_credentials, + access_key=assumed_role_configuration.credentials.aws_access_key_id, + secret_key=assumed_role_configuration.credentials.aws_secret_access_key, + token=assumed_role_configuration.credentials.aws_session_token, + expiry_time=assumed_role_configuration.credentials.expiration, + refresh_using=lambda: AwsProvider.refresh_credentials( + assumed_role_configuration, session + ), method="sts-assume-role", ) # Here we need the botocore session since it needs to use refreshable credentials assumed_session = BotocoreSession() assumed_session._credentials = assumed_refreshable_credentials - assumed_session.set_config_variable("region", self._identity.profile_region) + assumed_session.set_config_variable("region", identity.profile_region) return Session( - profile_name=self._identity.profile, + profile_name=identity.profile, botocore_session=assumed_session, ) except Exception as error: @@ -630,7 +640,10 @@ class AwsProvider(Provider): raise error # TODO: maybe this can be improved with botocore.credentials.DeferredRefreshableCredentials https://stackoverflow.com/a/75576540 - def refresh_credentials(self) -> dict: + @staticmethod + def refresh_credentials( + assumed_role_configuration: AWSAssumeRoleConfiguration, session: AWSSession + ) -> dict: """ Refresh credentials method using AWS STS Assume Role. @@ -640,7 +653,7 @@ class AwsProvider(Provider): logger.info("Refreshing assumed credentials...") # Since this method does not accept arguments, we need to get the original_session and the assumed role credentials - current_credentials = self._assumed_role_configuration.credentials + current_credentials = assumed_role_configuration.credentials refreshed_credentials = { "access_key": current_credentials.aws_access_key_id, "secret_key": current_credentials.aws_secret_access_key, @@ -655,8 +668,8 @@ class AwsProvider(Provider): if datetime.fromisoformat(refreshed_credentials["expiry_time"]) <= datetime.now( get_localzone() ): - assume_role_response = self.assume_role( - self._session.original_session, self._assumed_role_configuration.info + assume_role_response = AwsProvider.assume_role( + session.original_session, assumed_role_configuration.info ) refreshed_credentials = dict( # Keys of the dict has to be the same as those that are being searched in the parent class diff --git a/prowler/providers/aws/lib/s3/s3.py b/prowler/providers/aws/lib/s3/s3.py index f90a6844b7..f195a98663 100644 --- a/prowler/providers/aws/lib/s3/s3.py +++ b/prowler/providers/aws/lib/s3/s3.py @@ -76,9 +76,9 @@ class S3: output_directory: str, session: AWSSession = None, 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, diff --git a/prowler/providers/aws/lib/session/aws_set_up_session.py b/prowler/providers/aws/lib/session/aws_set_up_session.py index d62822c57c..9246c0a9eb 100644 --- a/prowler/providers/aws/lib/session/aws_set_up_session.py +++ b/prowler/providers/aws/lib/session/aws_set_up_session.py @@ -132,7 +132,7 @@ class AwsSetUpSession: ) # Assume the IAM Role logger.info(f"Assuming role: {assumed_role_information.role_arn.arn}") - assumed_role_credentials = self.assume_role( + assumed_role_credentials = AwsProvider.assume_role( self._session.current_session, assumed_role_information, ) @@ -145,8 +145,10 @@ class AwsSetUpSession: 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 + self._session.current_session = AwsProvider.setup_assumed_session( + self._identity, + self._assumed_role_configuration, + self._session, ) logger.info("Audit session is the new session created assuming an IAM Role") @@ -192,10 +194,8 @@ def validate_arguments( "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 external_id: + raise ValueError("If an external ID 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." diff --git a/tests/providers/aws/aws_provider_test.py b/tests/providers/aws/aws_provider_test.py index 5153729551..c14218f987 100644 --- a/tests/providers/aws/aws_provider_test.py +++ b/tests/providers/aws/aws_provider_test.py @@ -2008,7 +2008,12 @@ aws: ).isoformat(), } - assert aws_provider.refresh_credentials() == refreshed_credentials + assert ( + AwsProvider.refresh_credentials( + aws_provider._assumed_role_configuration, aws_provider._session + ) + == refreshed_credentials + ) @mock_aws def test_refresh_credentials_after_expiration(self): @@ -2025,7 +2030,9 @@ aws: current_credentials = aws_provider._assumed_role_configuration.credentials # Refresh credentials - refreshed_credentials = aws_provider.refresh_credentials() + refreshed_credentials = AwsProvider.refresh_credentials( + aws_provider._assumed_role_configuration, aws_provider._session + ) # Assert that the refreshed credentials are different access_key = refreshed_credentials.get("access_key") diff --git a/tests/providers/aws/lib/s3/s3_test.py b/tests/providers/aws/lib/s3/s3_test.py index d6797bb1b9..c32b81460d 100644 --- a/tests/providers/aws/lib/s3/s3_test.py +++ b/tests/providers/aws/lib/s3/s3_test.py @@ -413,5 +413,16 @@ class TestS3: ) assert ( str(e.value) - == "If a session duration, an external ID, or a role session name is provided, a role ARN is required." + == "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_and_role_arn_but_profile(self): + with pytest.raises(ValueError) as e: + S3( + session=None, + bucket_name=S3_BUCKET_NAME, + output_directory=CURRENT_DIRECTORY, + external_id="1234567890", + ) + assert str(e.value) == "If an external ID is provided, a role ARN is required." 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 1095667090..3cb8119316 100644 --- a/tests/providers/aws/lib/security_hub/security_hub_test.py +++ b/tests/providers/aws/lib/security_hub/security_hub_test.py @@ -564,5 +564,5 @@ class TestSecurityHub: assert ( str(e.value) - == "If a session duration, an external ID, or a role session name is provided, a role ARN is required." + == "If no role ARN is provided, a profile, an AWS access key ID, or an AWS secret access key is required." )