ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 04:21:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(s3): Use HeadBucket instead of GetBucketLocation (#8456)

Browse Source
This commit is contained in:
Pepe Fagoaga
2025-08-06 15:35:52 +02:00
committed by GitHub
parent 0ee0fc082a
commit 260fada3eb
6 changed files with 86 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
permissions/templates/cloudformation/prowler-scan-role.yml
+1 -1
View File
@@ -156,7 +156,7 @@ Resources:
"s3:ResourceAccount": !Sub ${S3IntegrationBucketAccountId}
- Effect: Allow
Action:
- "s3:GetBucketLocation"
- "s3:ListBucket"
Resource:
- !Sub "arn:${AWS::Partition}:s3:::${S3IntegrationBucketName}"
Condition:
permissions/templates/terraform/s3-integration/main.tf
+1 -1
View File
@@ -38,7 +38,7 @@ resource "aws_iam_role_policy" "prowler_s3_integration" {
{
Effect = "Allow"
Action = [
"s3:GetBucketLocation"
"s3:ListBucket"
]
Resource = [
"arn:${data.aws_partition.current.partition}:s3:::${var.s3_integration_bucket_name}"
prowler/CHANGELOG.md
+1
View File
@@ -28,6 +28,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
- 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)
- GitHub App authentication fails to generate output files and HTML header sections [(#8423)](https://github.com/prowler-cloud/prowler/pull/8423)
- S3 `test_connection` uses AWS S3 API `HeadBucket` instead of `GetBucketLocation` [(#8456)](https://github.com/prowler-cloud/prowler/pull/8456)
---
prowler/providers/aws/lib/s3/exceptions/exceptions.py
+11
View File
@@ -26,6 +26,10 @@ class S3BaseException(ProwlerException):
"message": "The specified location constraint is not valid.",
"remediation": "Check the location constraint.",
},
(6005, "S3InvalidBucketRegionError"): {
"message": "The specified bucket region is invalid.",
"remediation": "Check the bucket region.",
},
}
def __init__(self, code, file=None, original_exception=None, message=None):
@@ -75,3 +79,10 @@ class S3IllegalLocationConstraintError(S3BaseException):
super().__init__(
6004, file=file, original_exception=original_exception, message=message
)
class S3InvalidBucketRegionError(S3BaseException):
def __init__(self, file=None, original_exception=None, message=None):
super().__init__(
6005, file=file, original_exception=original_exception, message=message
)
prowler/providers/aws/lib/s3/s3.py
+21 -16
View File
@@ -39,6 +39,7 @@ from prowler.providers.aws.lib.s3.exceptions.exceptions import (
S3ClientError,
S3IllegalLocationConstraintError,
S3InvalidBucketNameError,
S3InvalidBucketRegionError,
S3TestConnectionError,
)
from prowler.providers.aws.lib.session.aws_set_up_session import (
@@ -312,28 +313,25 @@ class S3:
region_name=aws_region,
profile_name=profile,
)
s3_client = session.client(__class__.__name__.lower())
if "s3://" in bucket_name:
bucket_name = bucket_name.removeprefix("s3://")
# Check for the bucket location
bucket_location = s3_client.get_bucket_location(Bucket=bucket_name)
if bucket_location["LocationConstraint"] == "EU":
bucket_location["LocationConstraint"] = "eu-west-1"
if (
bucket_location["LocationConstraint"] == ""
or bucket_location["LocationConstraint"] is None
):
bucket_location["LocationConstraint"] = "us-east-1"
# Check bucket location, requires s3:ListBucket permission
# https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadBucket.html
bucket_region = s3_client.head_bucket(Bucket=bucket_name).get(
"BucketRegion"
)
if bucket_region is None:
exception = S3InvalidBucketRegionError()
if raise_on_exception:
raise exception
return Connection(error=exception)
# If the bucket location is not the same as the session region, change the session region
if (
session.region_name != bucket_location["LocationConstraint"]
and bucket_location["LocationConstraint"] is not None
):
if session.region_name != bucket_region:
s3_client = session.client(
__class__.__name__.lower(),
region_name=bucket_location["LocationConstraint"],
region_name=bucket_region,
)
# Set a Temp file to upload
with tempfile.TemporaryFile() as temp_file:
@@ -466,12 +464,19 @@ class S3:
if raise_on_exception:
raise session_token_expired
return Connection(error=session_token_expired)
except S3InvalidBucketRegionError as invalid_bucket_region_error:
logger.error(
f"{invalid_bucket_region_error.__class__.__name__}[{invalid_bucket_region_error.__traceback__.tb_lineno}]: {invalid_bucket_region_error}"
)
if raise_on_exception:
raise invalid_bucket_region_error
return Connection(error=invalid_bucket_region_error)
except ClientError as client_error:
if raise_on_exception:
if (
"specified bucket does not exist"
in client_error.response["Error"]["Message"]
or "Not Found" in client_error.response["Error"]["Message"]
):
raise S3InvalidBucketNameError(original_exception=client_error)
elif (
tests/providers/aws/lib/s3/s3_test.py
+51 -1
View File
@@ -1,7 +1,9 @@
from os import path, remove
from pathlib import Path
from unittest import mock
import boto3
import botocore
import pytest
from moto import mock_aws
@@ -9,7 +11,10 @@ from prowler.lib.outputs.compliance.iso27001.iso27001_aws import AWSISO27001
from prowler.lib.outputs.csv.csv import CSV
from prowler.lib.outputs.html.html import HTML
from prowler.lib.outputs.ocsf.ocsf import OCSF
from prowler.providers.aws.lib.s3.exceptions.exceptions import S3InvalidBucketNameError
from prowler.providers.aws.lib.s3.exceptions.exceptions import (
S3InvalidBucketNameError,
S3InvalidBucketRegionError,
)
from prowler.providers.aws.lib.s3.s3 import S3
from prowler.providers.common.models import Connection
from tests.lib.outputs.compliance.fixtures import ISO27001_2013_AWS
@@ -18,6 +23,7 @@ from tests.providers.aws.utils import AWS_REGION_US_EAST_1
CURRENT_DIRECTORY = str(Path(path.dirname(path.realpath(__file__))))
S3_BUCKET_NAME = "test_bucket"
S3_BUCKET_ARN = f"arn:aws:s3:::{S3_BUCKET_NAME}"
OUTPUT_MODE_CSV = "csv"
OUTPUT_MODE_JSON_OCSF = "json-ocsf"
OUTPUT_MODE_JSON_ASFF = "json-asff"
@@ -46,6 +52,7 @@ FINDING = generate_finding_output(
related_to=["related"],
notes="Notes about the finding",
)
make_api_call = botocore.client.BaseClient._make_api_call
class TestS3:
@@ -373,6 +380,49 @@ class TestS3:
aws_secret_access_key=access_key["SecretAccessKey"],
)
def mock_make_api_head_bucket(self, operation_name, kwarg):
if operation_name == "HeadBucket":
if kwarg["Bucket"] == "bucket_without_region":
return {"BucketArn": S3_BUCKET_ARN}
else:
return {
"BucketArn": S3_BUCKET_ARN,
"BucketRegion": AWS_REGION_US_EAST_1,
}
@mock_aws
def test_test_connection_S3_bucket_invalid_region_raise_on_exception(self):
with mock.patch(
"botocore.client.BaseClient._make_api_call",
new=self.mock_make_api_head_bucket,
):
with pytest.raises(S3InvalidBucketRegionError):
S3.test_connection(
aws_region=AWS_REGION_US_EAST_1,
# Bucket without region to force exception
bucket_name="bucket_without_region",
# aws_access_key_id=access_key["AccessKeyId"],
# aws_secret_access_key=access_key["SecretAccessKey"],
raise_on_exception=True,
)
def test_test_connection_S3_bucket_invalid_region_no_raise_on_exception(self):
with mock.patch(
"botocore.client.BaseClient._make_api_call",
new=self.mock_make_api_head_bucket,
):
connection = S3.test_connection(
aws_region=AWS_REGION_US_EAST_1,
bucket_name="bucket_without_region",
raise_on_exception=False,
)
assert connection.is_connected is False
assert (
str(connection.error)
== "S3InvalidBucketRegionError[6005]: The specified bucket region is invalid."
)
@mock_aws
def test_init_without_session(self):
with pytest.raises(ValueError) as e: