diff --git a/moto/s3/responses.py b/moto/s3/responses.py index da42bac69..3b40c8701 100644 --- a/moto/s3/responses.py +++ b/moto/s3/responses.py @@ -1522,10 +1522,14 @@ class S3Response(BaseResponse): acl = self.backend.get_bucket(bucket_name).acl tagging = self._tagging_from_headers(request.headers) + if "versionId" in query: + version_id = query["versionId"][0] + else: + version_id = None + if "retention" in query: if not lock_enabled: raise LockNotEnabled - version_id = query.get("VersionId") retention = self._mode_until_from_body() self.backend.put_object_retention( bucket_name, key_name, version_id=version_id, retention=retention @@ -1535,7 +1539,6 @@ class S3Response(BaseResponse): if "legal-hold" in query: if not lock_enabled: raise LockNotEnabled - version_id = query.get("VersionId") legal_hold_status = self._legal_hold_status_from_xml(body) self.backend.put_object_legal_hold( bucket_name, key_name, version_id, legal_hold_status @@ -1547,10 +1550,6 @@ class S3Response(BaseResponse): return 200, response_headers, "" if "tagging" in query: - if "versionId" in query: - version_id = query["versionId"][0] - else: - version_id = None key_to_tag = self.backend.get_object( bucket_name, key_name, version_id=version_id ) diff --git a/tests/test_s3/test_s3_lock.py b/tests/test_s3/test_s3_lock.py index bb562fdfb..2a02fd512 100644 --- a/tests/test_s3/test_s3_lock.py +++ b/tests/test_s3/test_s3_lock.py @@ -1,9 +1,10 @@ import time import boto3 import datetime -import botocore +import pytest from moto import mock_s3 from botocore.config import Config +from botocore.client import ClientError from moto.s3.responses import DEFAULT_REGION_NAME import sure # noqa # pylint: disable=unused-import @@ -34,7 +35,7 @@ def test_locked_object(): try: s3.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id) deleted = True - except botocore.client.ClientError as e: + except ClientError as e: e.response["Error"]["Code"].should.equal("AccessDenied") deleted.should.equal(False) @@ -64,7 +65,7 @@ def test_fail_locked_object(): ObjectLockMode="COMPLIANCE", ObjectLockRetainUntilDate=until, ) - except botocore.client.ClientError as e: + except ClientError as e: e.response["Error"]["Code"].should.equal("InvalidRequest") failed = True @@ -99,7 +100,7 @@ def test_put_object_lock(): try: s3.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id) deleted = True - except botocore.client.ClientError as e: + except ClientError as e: e.response["Error"]["Code"].should.equal("AccessDenied") deleted.should.equal(False) @@ -135,7 +136,7 @@ def test_put_object_legal_hold(): try: s3.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id) deleted = True - except botocore.client.ClientError as e: + except ClientError as e: e.response["Error"]["Code"].should.equal("AccessDenied") deleted.should.equal(False) @@ -181,7 +182,7 @@ def test_put_default_lock(): try: s3.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id) deleted = True - except botocore.client.ClientError as e: + except ClientError as e: e.response["Error"]["Code"].should.equal("AccessDenied") deleted.should.equal(False) @@ -194,3 +195,111 @@ def test_put_default_lock(): response["ObjectLockConfiguration"]["Rule"]["DefaultRetention"][ "Days" ].should.equal(days) + + +@mock_s3 +def test_put_object_legal_hold_with_versions(): + s3 = boto3.client("s3", config=Config(region_name=DEFAULT_REGION_NAME)) + + bucket_name = "put-legal-bucket" + key_name = "file.txt" + + s3.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True) + + put_obj_1 = s3.put_object(Bucket=bucket_name, Body=b"test", Key=key_name) + version_id_1 = put_obj_1["VersionId"] + # lock the object with the version, locking the version 1 + s3.put_object_legal_hold( + Bucket=bucket_name, + Key=key_name, + VersionId=version_id_1, + LegalHold={"Status": "ON"}, + ) + + # put an object on the same key, effectively creating a version 2 of the object + put_obj_2 = s3.put_object(Bucket=bucket_name, Body=b"test", Key=key_name) + version_id_2 = put_obj_2["VersionId"] + # also lock the version 2 of the object + s3.put_object_legal_hold( + Bucket=bucket_name, + Key=key_name, + VersionId=version_id_2, + LegalHold={"Status": "ON"}, + ) + + # assert that the version 1 is locked + head_obj_1 = s3.head_object( + Bucket=bucket_name, Key=key_name, VersionId=version_id_1 + ) + assert head_obj_1["ObjectLockLegalHoldStatus"] == "ON" + + # remove the lock from the version 1 of the object + s3.put_object_legal_hold( + Bucket=bucket_name, + Key=key_name, + VersionId=version_id_1, + LegalHold={"Status": "OFF"}, + ) + + # assert that you can now delete the version 1 of the object + s3.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id_1) + + with pytest.raises(ClientError) as e: + s3.head_object(Bucket=bucket_name, Key=key_name, VersionId=version_id_1) + assert e.value.response["Error"]["Code"] == "404" + + # cleaning + s3.put_object_legal_hold( + Bucket=bucket_name, + Key=key_name, + VersionId=version_id_2, + LegalHold={"Status": "OFF"}, + ) + s3.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id_2) + s3.delete_bucket(Bucket=bucket_name) + + +@mock_s3 +def test_put_object_lock_with_versions(): + s3 = boto3.client("s3", config=Config(region_name=DEFAULT_REGION_NAME)) + + bucket_name = "put-lock-bucket-test" + key_name = "file.txt" + seconds_lock = 2 + + s3.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=True) + + put_obj_1 = s3.put_object(Bucket=bucket_name, Body=b"test", Key=key_name) + version_id_1 = put_obj_1["VersionId"] + put_obj_2 = s3.put_object(Bucket=bucket_name, Body=b"test", Key=key_name) + version_id_2 = put_obj_2["VersionId"] + + until = datetime.datetime.utcnow() + datetime.timedelta(seconds=seconds_lock) + + s3.put_object_retention( + Bucket=bucket_name, + Key=key_name, + VersionId=version_id_1, + Retention={"Mode": "COMPLIANCE", "RetainUntilDate": until}, + ) + + # assert that you can delete the locked version 1 of the object + deleted = False + try: + s3.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id_1) + deleted = True + except ClientError as e: + e.response["Error"]["Code"].should.equal("AccessDenied") + + deleted.should.equal(False) + + # assert that you can delete the version 2 of the object, not concerned by the lock + s3.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id_2) + with pytest.raises(ClientError) as e: + s3.head_object(Bucket=bucket_name, Key=key_name, VersionId=version_id_2) + assert e.value.response["Error"]["Code"] == "404" + + # cleaning + time.sleep(seconds_lock) + s3.delete_object(Bucket=bucket_name, Key=key_name, VersionId=version_id_1) + s3.delete_bucket(Bucket=bucket_name)