diff --git a/moto/s3/models.py b/moto/s3/models.py index 31df9c21d..5c71675fb 100644 --- a/moto/s3/models.py +++ b/moto/s3/models.py @@ -1298,7 +1298,7 @@ class FakeBucket(CloudFormationModel): "The target bucket for logging does not exist." ) - target_prefix = self.logging.get("TargetPrefix", None) + target_prefix = logging_config.get("TargetPrefix", None) has_policy_permissions = self._log_permissions_enabled_policy( target_bucket=target_bucket, target_prefix=target_prefix ) diff --git a/tests/test_s3/__init__.py b/tests/test_s3/__init__.py index 2a177eb21..e89b1490c 100644 --- a/tests/test_s3/__init__.py +++ b/tests/test_s3/__init__.py @@ -51,22 +51,24 @@ def s3_aws_verified(func): finally: ### CLEANUP ### - versions = client.list_object_versions(Bucket=bucket_name).get( - "Versions", [] - ) - for key in versions: - client.delete_object( - Bucket=bucket_name, Key=key["Key"], VersionId=key.get("VersionId") - ) - delete_markers = client.list_object_versions(Bucket=bucket_name).get( - "DeleteMarkers", [] - ) - for key in delete_markers: - client.delete_object( - Bucket=bucket_name, Key=key["Key"], VersionId=key.get("VersionId") - ) + empty_bucket(client, bucket_name) client.delete_bucket(Bucket=bucket_name) return resp return pagination_wrapper + + +def empty_bucket(client, bucket_name): + versions = client.list_object_versions(Bucket=bucket_name).get("Versions", []) + for key in versions: + client.delete_object( + Bucket=bucket_name, Key=key["Key"], VersionId=key.get("VersionId") + ) + delete_markers = client.list_object_versions(Bucket=bucket_name).get( + "DeleteMarkers", [] + ) + for key in delete_markers: + client.delete_object( + Bucket=bucket_name, Key=key["Key"], VersionId=key.get("VersionId") + ) diff --git a/tests/test_s3/test_s3_logging.py b/tests/test_s3/test_s3_logging.py index 152b6fdec..a891b4a14 100644 --- a/tests/test_s3/test_s3_logging.py +++ b/tests/test_s3/test_s3_logging.py @@ -1,6 +1,7 @@ import json from unittest import SkipTest from unittest.mock import patch +from uuid import uuid4 import boto3 import pytest @@ -11,6 +12,7 @@ from moto.core import DEFAULT_ACCOUNT_ID from moto.s3 import s3_backends from moto.s3.models import FakeBucket from moto.s3.responses import DEFAULT_REGION_NAME +from tests.test_s3 import empty_bucket, s3_aws_verified @mock_aws @@ -587,3 +589,76 @@ def test_bucket_policy_resource(): assert FakeBucket._log_permissions_enabled_policy( target_bucket=log_bucket_obj, target_prefix="prefix" ) + + +@s3_aws_verified +@pytest.mark.aws_verified +def test_put_logging_w_bucket_policy_no_prefix(bucket_name=None): + s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + + log_bucket_name = f"{uuid4()}" + s3_client.create_bucket(Bucket=log_bucket_name) + bucket_policy = { + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "S3ServerAccessLogsPolicy", + "Effect": "Allow", + "Principal": {"Service": "logging.s3.amazonaws.com"}, + "Action": ["s3:PutObject"], + "Resource": f"arn:aws:s3:::{log_bucket_name}/*", + } + ], + } + s3_client.put_bucket_policy( + Bucket=log_bucket_name, Policy=json.dumps(bucket_policy) + ) + s3_client.put_bucket_logging( + Bucket=bucket_name, + BucketLoggingStatus={ + "LoggingEnabled": {"TargetBucket": log_bucket_name, "TargetPrefix": ""} + }, + ) + result = s3_client.get_bucket_logging(Bucket=bucket_name) + assert result["LoggingEnabled"]["TargetBucket"] == log_bucket_name + assert result["LoggingEnabled"]["TargetPrefix"] == "" + + empty_bucket(s3_client, log_bucket_name) + s3_client.delete_bucket(Bucket=log_bucket_name) + + +@s3_aws_verified +@pytest.mark.aws_verified +def test_put_logging_w_bucket_policy_w_prefix(bucket_name=None): + s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + + log_bucket_name = f"{uuid4()}" + s3_client.create_bucket(Bucket=log_bucket_name) + prefix = "some-prefix" + bucket_policy = { + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "S3ServerAccessLogsPolicy", + "Effect": "Allow", + "Principal": {"Service": "logging.s3.amazonaws.com"}, + "Action": ["s3:PutObject"], + "Resource": f"arn:aws:s3:::{log_bucket_name}/{prefix}*", + } + ], + } + s3_client.put_bucket_policy( + Bucket=log_bucket_name, Policy=json.dumps(bucket_policy) + ) + s3_client.put_bucket_logging( + Bucket=bucket_name, + BucketLoggingStatus={ + "LoggingEnabled": {"TargetBucket": log_bucket_name, "TargetPrefix": prefix} + }, + ) + result = s3_client.get_bucket_logging(Bucket=bucket_name) + assert result["LoggingEnabled"]["TargetBucket"] == log_bucket_name + assert result["LoggingEnabled"]["TargetPrefix"] == prefix + + empty_bucket(s3_client, log_bucket_name) + s3_client.delete_bucket(Bucket=log_bucket_name)