Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

S3 - missing features (#4793)

Browse Source
This commit is contained in:
Bert Blommers 2022-01-25 18:25:39 -01:00 committed by GitHub
parent 38ad5193d4
commit 35d3c72039
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 84 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
moto/s3/models.py
View File

@ -115,6 +115,7 @@ class FakeKey(BaseModel):
lock_mode=None,
lock_legal_status=None,
lock_until=None,
s3_backend=None,
):
self.name = name
self.last_modified = datetime.datetime.utcnow()
@ -147,6 +148,8 @@ class FakeKey(BaseModel):
# Default metadata values
self._metadata["Content-Type"] = "binary/octet-stream"
self.s3_backend = s3_backend
@property
def version_id(self):
return self._version_id
@ -265,6 +268,9 @@ class FakeKey(BaseModel):
res["x-amz-object-lock-retain-until-date"] = self.lock_until
if self.lock_mode:
res["x-amz-object-lock-mode"] = self.lock_mode
tags = s3_backend.tagger.get_tag_dict_for_resource(self.arn)
if tags:
res["x-amz-tagging-count"] = len(tags.keys())
return res
@ -1620,6 +1626,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
lock_mode=lock_mode,
lock_legal_status=lock_legal_status,
lock_until=lock_until,
s3_backend=s3_backend,
)
keys = [
@ -1650,8 +1657,16 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
key.lock_mode = retention[0]
key.lock_until = retention[1]
def get_object(self, bucket_name, key_name, version_id=None, part_number=None):
key_name = clean_key_name(key_name)
def get_object(
self,
bucket_name,
key_name,
version_id=None,
part_number=None,
key_is_clean=False,
):
if not key_is_clean:
key_name = clean_key_name(key_name)
bucket = self.get_bucket(bucket_name)
key = None
@ -1991,37 +2006,35 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
def copy_object(
self,
src_bucket_name,
src_key_name,
src_key,
dest_bucket_name,
dest_key_name,
storage=None,
acl=None,
src_version_id=None,
encryption=None,
kms_key_id=None,
bucket_key_enabled=False,
):
key = self.get_object(src_bucket_name, src_key_name, version_id=src_version_id)
new_key = self.put_object(
bucket_name=dest_bucket_name,
key_name=dest_key_name,
value=key.value,
storage=storage or key.storage_class,
multipart=key.multipart,
encryption=encryption or key.encryption,
kms_key_id=kms_key_id or key.kms_key_id,
bucket_key_enabled=key.bucket_key_enabled,
lock_mode=key.lock_mode,
lock_legal_status=key.lock_legal_status,
lock_until=key.lock_until,
value=src_key.value,
storage=storage or src_key.storage_class,
multipart=src_key.multipart,
encryption=encryption or src_key.encryption,
kms_key_id=kms_key_id or src_key.kms_key_id,
bucket_key_enabled=bucket_key_enabled or src_key.bucket_key_enabled,
lock_mode=src_key.lock_mode,
lock_legal_status=src_key.lock_legal_status,
lock_until=src_key.lock_until,
)
self.tagger.copy_tags(key.arn, new_key.arn)
new_key.set_metadata(key.metadata)
self.tagger.copy_tags(src_key.arn, new_key.arn)
new_key.set_metadata(src_key.metadata)
if acl is not None:
new_key.set_acl(acl)
if key.storage_class in "GLACIER":
if src_key.storage_class in "GLACIER":
# Object copied from Glacier object should not have expiry
new_key.set_expiry(None)

19
moto/s3/responses.py
View File

@ -1365,6 +1365,7 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
kms_key_id = request.headers.get(
"x-amz-server-side-encryption-aws-kms-key-id", None
)
bucket_key_enabled = request.headers.get(
"x-amz-server-side-encryption-bucket-key-enabled", None
)
@ -1437,13 +1438,15 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
if isinstance(copy_source, bytes):
copy_source = copy_source.decode("utf-8")
copy_source_parsed = urlparse(copy_source)
src_bucket, src_key = copy_source_parsed.path.lstrip("/").split("/", 1)
src_bucket, src_key = (
unquote(copy_source_parsed.path).lstrip("/").split("/", 1)
)
src_version_id = parse_qs(copy_source_parsed.query).get(
"versionId", [None]
)[0]
key = self.backend.get_object(
src_bucket, src_key, version_id=src_version_id
src_bucket, src_key, version_id=src_version_id, key_is_clean=True
)
if key is not None:
@ -1455,16 +1458,22 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
):
raise ObjectNotInActiveTierError(key)
bucket_key_enabled = (
request.headers.get(
"x-amz-server-side-encryption-bucket-key-enabled", ""
).lower()
== "true"
)
self.backend.copy_object(
src_bucket,
src_key,
key,
bucket_name,
key_name,
storage=storage_class,
acl=acl,
src_version_id=src_version_id,
kms_key_id=kms_key_id,
encryption=encryption,
bucket_key_enabled=bucket_key_enabled,
)
else:
raise MissingKey(key=src_key)

1
tests/terraform-tests.success.txt
View File

@ -121,6 +121,7 @@ TestAccAWSENI_basic
TestAccAWSENI_IPv6
TestAccAWSENI_disappears
TestAccAWSS3BucketObject_
TestAccAWSS3ObjectCopy
TestAccAWSIAMPolicy_
TestAccAWSIAMGroup_
TestAccAWSIAMRolePolicy

38
tests/test_s3/test_s3.py
View File

@ -592,6 +592,41 @@ def test_copy_key_with_version_boto3():
resp["Body"].read().should.equal(b"some value")
@mock_s3
def test_copy_object_with_bucketkeyenabled_returns_the_value():
s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
bucket_name = "test-copy-object-with-bucketkeyenabled"
s3.create_bucket(Bucket=bucket_name)
key = s3.Object(bucket_name, "the-key")
key.put(Body=b"some value")
key2 = s3.Object(bucket_name, "new-key")
key2.copy_from(
CopySource=f"{bucket_name}/the-key",
BucketKeyEnabled=True,
ServerSideEncryption="aws:kms",
)
resp = client.get_object(Bucket=bucket_name, Key="the-key")
src_headers = resp["ResponseMetadata"]["HTTPHeaders"]
src_headers.shouldnt.have.key("x-amz-server-side-encryption")
src_headers.shouldnt.have.key("x-amz-server-side-encryption-aws-kms-key-id")
src_headers.shouldnt.have.key("x-amz-server-side-encryption-bucket-key-enabled")
resp = client.get_object(Bucket=bucket_name, Key="new-key")
target_headers = resp["ResponseMetadata"]["HTTPHeaders"]
target_headers.should.have.key("x-amz-server-side-encryption")
# AWS will also return the KMS default key id - not yet implemented
# target_headers.should.have.key("x-amz-server-side-encryption-aws-kms-key-id")
# This field is only returned if encryption is set to 'aws:kms'
target_headers.should.have.key("x-amz-server-side-encryption-bucket-key-enabled")
str(
target_headers["x-amz-server-side-encryption-bucket-key-enabled"]
).lower().should.equal("true")
@mock_s3
def test_set_metadata_boto3():
s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
@ -2719,6 +2754,9 @@ def test_boto3_put_object_with_tagging():
{"Key": "foo", "Value": "bar"}
)
resp = s3.get_object(Bucket=bucket_name, Key=key)
resp.should.have.key("TagCount").equals(1)
s3.delete_object_tagging(Bucket=bucket_name, Key=key)
s3.get_object_tagging(Bucket=bucket_name, Key=key)["TagSet"].should.equal([])