fix S3 PutObjectRetention and PutObjectLegalHold not using versionId (#6295)

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
             )

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)