fix: add default values for S3 class (#8417)

Co-authored-by: Pedro Martín <pedromarting3@gmail.com>
Co-authored-by: MrCloudSec <hello@mistercloudsec.com>
This commit is contained in:
Adrián Jesús Peña Rodríguez
2025-08-01 13:50:51 +02:00
committed by GitHub
parent 19a735aafe
commit 581afd38e6
7 changed files with 65 additions and 31 deletions
+3
View File
@@ -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)
+31 -18
View File
@@ -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
+2 -2
View File
@@ -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,
@@ -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."
+9 -2
View File
@@ -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")
+12 -1
View File
@@ -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."
@@ -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."
)