Repository: getmoto/moto (python)

Issue:
moto does not enforce S3 object's tag limit
When calling https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3/client/put_object_tagging.html with tag set containing more than 10 elements moto does not respect AWS limits https://docs.aws.amazon.com/AmazonS3/latest/userguide/tagging-managing.html and allows to associate more than 10 tags with an object.

Versions:
moto==4.2.11
boto3==1.26.90
s3transfer==0.6.2

Pull request description:
S3: put_object_tagging() now validates the number of tags provided
Fixes #7104 

All tests changed in this PR are verified against AWS

Patch:
diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md
index 73285ec2e..f8349467c 100644
--- a/IMPLEMENTATION_COVERAGE.md
+++ b/IMPLEMENTATION_COVERAGE.md
@@ -6257,7 +6257,7 @@
 
 ## s3
 <details>
-<summary>65% implemented</summary>
+<summary>66% implemented</summary>
 
 - [X] abort_multipart_upload
 - [X] complete_multipart_upload
@@ -6351,7 +6351,7 @@
 - [X] put_object_legal_hold
 - [X] put_object_lock_configuration
 - [X] put_object_retention
-- [ ] put_object_tagging
+- [X] put_object_tagging
 - [X] put_public_access_block
 - [ ] restore_object
 - [X] select_object_content
diff --git a/docs/docs/services/s3.rst b/docs/docs/services/s3.rst
index 67639282f..88eef2c87 100644
--- a/docs/docs/services/s3.rst
+++ b/docs/docs/services/s3.rst
@@ -144,7 +144,7 @@ s3
 - [X] put_object_legal_hold
 - [X] put_object_lock_configuration
 - [X] put_object_retention
-- [ ] put_object_tagging
+- [X] put_object_tagging
 - [X] put_public_access_block
 - [ ] restore_object
 - [X] select_object_content
diff --git a/moto/awslambda/models.py b/moto/awslambda/models.py
index ffac0db24..3f4310744 100644
--- a/moto/awslambda/models.py
+++ b/moto/awslambda/models.py
@@ -620,7 +620,7 @@ class LambdaFunction(CloudFormationModel, DockerModel):
 
         self.description = spec.get("Description", "")
         self.memory_size = spec.get("MemorySize", 128)
-        self.package_type = spec.get("PackageType", None)
+        self.package_type = spec.get("PackageType", "Zip")
         self.publish = spec.get("Publish", False)  # this is ignored currently
         self.timeout = spec.get("Timeout", 3)
         self.layers: List[Dict[str, str]] = self._get_layers_data(
diff --git a/moto/s3/exceptions.py b/moto/s3/exceptions.py
index 737fa9d7a..b86d56995 100644
--- a/moto/s3/exceptions.py
+++ b/moto/s3/exceptions.py
@@ -65,6 +65,13 @@ class AccessForbidden(S3ClientError):
         super().__init__("AccessForbidden", msg)
 
 
+class BadRequest(S3ClientError):
+    code = 403
+
+    def __init__(self, msg: str):
+        super().__init__("BadRequest", msg)
+
+
 class BucketError(S3ClientError):
     def __init__(self, *args: Any, **kwargs: Any):
         kwargs.setdefault("template", "bucket_error")
diff --git a/moto/s3/models.py b/moto/s3/models.py
index 4665e96d0..a1ece84df 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -34,6 +34,7 @@ from moto.moto_api._internal import mock_random as random
 from moto.moto_api._internal.managed_state_model import ManagedState
 from moto.s3.exceptions import (
     AccessDeniedByLock,
+    BadRequest,
     BucketAlreadyExists,
     BucketNeedsToBeNew,
     CopyObjectMustChangeSomething,
@@ -2161,7 +2162,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
     def get_object_tagging(self, key: FakeKey) -> Dict[str, List[Dict[str, str]]]:
         return self.tagger.list_tags_for_resource(key.arn)
 
-    def set_key_tags(
+    def put_object_tagging(
         self,
         key: Optional[FakeKey],
         tags: Optional[Dict[str, str]],
@@ -2169,12 +2170,19 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
     ) -> FakeKey:
         if key is None:
             raise MissingKey(key=key_name)
-        boto_tags_dict = self.tagger.convert_dict_to_tags_input(tags)
-        errmsg = self.tagger.validate_tags(boto_tags_dict)
+        tags_input = self.tagger.convert_dict_to_tags_input(tags)
+        # Validation custom to S3
+        if tags:
+            if len(tags_input) > 10:
+                raise BadRequest("Object tags cannot be greater than 10")
+            if any([tagkey.startswith("aws") for tagkey in tags.keys()]):
+                raise InvalidTagError("Your TagKey cannot be prefixed with aws:")
+        # Validation shared across all services
+        errmsg = self.tagger.validate_tags(tags_input)
         if errmsg:
             raise InvalidTagError(errmsg)
         self.tagger.delete_all_tags_for_resource(key.arn)
-        self.tagger.tag_resource(key.arn, boto_tags_dict)
+        self.tagger.tag_resource(key.arn, tags_input)
         return key
 
     def get_bucket_tagging(self, bucket_name: str) -> Dict[str, List[Dict[str, str]]]:
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index fc589caec..186ee4d71 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -888,7 +888,7 @@ class S3Response(BaseResponse):
         elif "tagging" in querystring:
             tagging = self._bucket_tagging_from_body()
             self.backend.put_bucket_tagging(bucket_name, tagging)
-            return ""
+            return 204, {}, ""
         elif "website" in querystring:
             self.backend.set_bucket_website_configuration(bucket_name, self.body)
             return ""
@@ -1629,7 +1629,7 @@ class S3Response(BaseResponse):
                 bucket_name, key_name, version_id=version_id
             )
             tagging = self._tagging_from_xml(body)
-            self.backend.set_key_tags(key_to_tag, tagging, key_name)
+            self.backend.put_object_tagging(key_to_tag, tagging, key_name)
             return 200, response_headers, ""
 
         if "x-amz-copy-source" in request.headers:
@@ -1692,7 +1692,7 @@ class S3Response(BaseResponse):
             tdirective = request.headers.get("x-amz-tagging-directive")
             if tdirective == "REPLACE":
                 tagging = self._tagging_from_headers(request.headers)
-                self.backend.set_key_tags(new_key, tagging)
+                self.backend.put_object_tagging(new_key, tagging)
             if key_to_copy.version_id != "null":
                 response_headers[
                     "x-amz-copy-source-version-id"
@@ -1742,7 +1742,7 @@ class S3Response(BaseResponse):
         )
         if checksum_algorithm:
             new_key.checksum_algorithm = checksum_algorithm
-        self.backend.set_key_tags(new_key, tagging)
+        self.backend.put_object_tagging(new_key, tagging)
 
         response_headers.update(new_key.response_dict)
         # Remove content-length - the response body is empty for this request
@@ -2279,7 +2279,7 @@ class S3Response(BaseResponse):
                 )
                 key.checksum_value = checksum
 
-            self.backend.set_key_tags(key, multipart.tags)
+            self.backend.put_object_tagging(key, multipart.tags)
             self.backend.put_object_acl(
                 bucket_name=bucket_name,
                 key_name=key.name,


Test patch:
diff --git a/tests/test_awslambda/__init__.py b/tests/test_awslambda/__init__.py
index 08a1c1568..89d605282 100644
--- a/tests/test_awslambda/__init__.py
+++ b/tests/test_awslambda/__init__.py
@@ -1,1 +1,108 @@
-# This file is intentionally left blank.
+import json
+import os
+from functools import wraps
+from time import sleep
+from uuid import uuid4
+
+import boto3
+from botocore.exceptions import ClientError
+
+from moto import mock_iam, mock_lambda, mock_sts
+
+
+def lambda_aws_verified(func):
+    """
+    Function that is verified to work against AWS.
+    Can be run against AWS at any time by setting:
+      MOTO_TEST_ALLOW_AWS_REQUEST=true
+
+    If this environment variable is not set, the function runs in a `mock_lambda/mock_iam/mock_sts` context.
+
+    This decorator will:
+      - Create an IAM-role that can be used by AWSLambda functions table
+      - Run the test
+      - Delete the role
+    """
+
+    @wraps(func)
+    def pagination_wrapper():
+        role_name = "moto_test_role_" + str(uuid4())[0:6]
+
+        allow_aws_request = (
+            os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
+        )
+
+        if allow_aws_request:
+            return create_role_and_test(role_name)
+        else:
+            with mock_lambda(), mock_iam(), mock_sts():
+                return create_role_and_test(role_name)
+
+    def create_role_and_test(role_name):
+        from .test_lambda import _lambda_region
+
+        policy_doc = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Sid": "LambdaAssumeRole",
+                    "Effect": "Allow",
+                    "Principal": {"Service": "lambda.amazonaws.com"},
+                    "Action": "sts:AssumeRole",
+                }
+            ],
+        }
+        iam = boto3.client("iam", region_name=_lambda_region)
+        iam_role_arn = iam.create_role(
+            RoleName=role_name,
+            AssumeRolePolicyDocument=json.dumps(policy_doc),
+            Path="/",
+        )["Role"]["Arn"]
+
+        try:
+            # Role is not immediately available
+            check_iam_role_is_ready_to_use(iam_role_arn, role_name, _lambda_region)
+
+            resp = func(iam_role_arn)
+        finally:
+            iam.delete_role(RoleName=role_name)
+
+        return resp
+
+    def check_iam_role_is_ready_to_use(role_arn: str, role_name: str, region: str):
+        from .utilities import get_test_zip_file1
+
+        iam = boto3.client("iam", region_name=region)
+
+        # The waiter is normally used to wait until a resource is ready
+        wait_for_role = iam.get_waiter("role_exists")
+        wait_for_role.wait(RoleName=role_name)
+
+        # HOWEVER:
+        # Just because the role exists, doesn't mean it can be used by Lambda
+        # The only reliable way to check if our role is ready:
+        #   - try to create a function and see if it succeeds
+        #
+        # The alternive is to wait between 5 and 10 seconds, which is not a great solution either
+        _lambda = boto3.client("lambda", region)
+        for _ in range(15):
+            try:
+                fn_name = f"fn_for_{role_name}"
+                _lambda.create_function(
+                    FunctionName=fn_name,
+                    Runtime="python3.11",
+                    Role=role_arn,
+                    Handler="lambda_function.lambda_handler",
+                    Code={"ZipFile": get_test_zip_file1()},
+                )
+                # If it succeeded, our role is ready
+                _lambda.delete_function(FunctionName=fn_name)
+
+                return
+            except ClientError:
+                sleep(1)
+        raise Exception(
+            f"Couldn't create test Lambda FN using IAM role {role_name}, probably because it wasn't ready in time"
+        )
+
+    return pagination_wrapper
diff --git a/tests/test_awslambda/test_lambda.py b/tests/test_awslambda/test_lambda.py
index 0eb17edb3..46f3731cc 100644
--- a/tests/test_awslambda/test_lambda.py
+++ b/tests/test_awslambda/test_lambda.py
@@ -14,6 +14,7 @@ from moto import mock_ecr, mock_lambda, mock_s3, settings
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
 from tests.test_ecr.test_ecr_helpers import _create_image_manifest
 
+from . import lambda_aws_verified
 from .utilities import (
     _process_lambda,
     create_invalid_lambda,
@@ -40,10 +41,14 @@ def test_lambda_regions(region):
     assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
 
 
-@mock_lambda
-def test_list_functions():
+@pytest.mark.aws_verified
+@lambda_aws_verified
+def test_list_functions(iam_role_arn=None):
+    sts = boto3.client("sts", "eu-west-2")
+    account_id = sts.get_caller_identity()["Account"]
+
     conn = boto3.client("lambda", _lambda_region)
-    function_name = str(uuid4())[0:6]
+    function_name = "moto_test_" + str(uuid4())[0:6]
     initial_list = conn.list_functions()["Functions"]
     initial_names = [f["FunctionName"] for f in initial_list]
     assert function_name not in initial_names
@@ -51,10 +56,14 @@ def test_list_functions():
     conn.create_function(
         FunctionName=function_name,
         Runtime=PYTHON_VERSION,
-        Role=get_role_name(),
+        Role=iam_role_arn,
         Handler="lambda_function.lambda_handler",
         Code={"ZipFile": get_test_zip_file1()},
     )
+
+    wait_for_func = conn.get_waiter("function_active")
+    wait_for_func.wait(FunctionName=function_name, WaiterConfig={"Delay": 1})
+
     names = [f["FunctionName"] for f in conn.list_functions()["Functions"]]
     assert function_name in names
 
@@ -62,26 +71,31 @@ def test_list_functions():
     func_list = conn.list_functions()["Functions"]
     our_functions = [f for f in func_list if f["FunctionName"] == function_name]
     assert len(our_functions) == 1
+    assert our_functions[0]["PackageType"] == "Zip"
 
     # FunctionVersion=ALL means we should get a list of all versions
     full_list = conn.list_functions(FunctionVersion="ALL")["Functions"]
     our_functions = [f for f in full_list if f["FunctionName"] == function_name]
     assert len(our_functions) == 2
+    for func in our_functions:
+        assert func["PackageType"] == "Zip"
 
     v1 = [f for f in our_functions if f["Version"] == "1"][0]
     assert v1["Description"] == "v2"
     assert (
         v1["FunctionArn"]
-        == f"arn:aws:lambda:{_lambda_region}:{ACCOUNT_ID}:function:{function_name}:1"
+        == f"arn:aws:lambda:{_lambda_region}:{account_id}:function:{function_name}:1"
     )
 
     latest = [f for f in our_functions if f["Version"] == "$LATEST"][0]
     assert latest["Description"] == ""
     assert (
         latest["FunctionArn"]
-        == f"arn:aws:lambda:{_lambda_region}:{ACCOUNT_ID}:function:{function_name}:$LATEST"
+        == f"arn:aws:lambda:{_lambda_region}:{account_id}:function:{function_name}:$LATEST"
     )
 
+    conn.delete_function(FunctionName=function_name)
+
 
 @mock_lambda
 def test_create_based_on_s3_with_missing_bucket():
@@ -189,6 +203,7 @@ def test_create_function_from_zipfile():
         "Description": "test lambda function",
         "Timeout": 3,
         "MemorySize": 128,
+        "PackageType": "Zip",
         "CodeSha256": base64.b64encode(hashlib.sha256(zip_content).digest()).decode(
             "utf-8"
         ),
@@ -965,6 +980,7 @@ def test_list_create_list_get_delete_list():
             "FunctionName": function_name,
             "Handler": "lambda_function.lambda_handler",
             "MemorySize": 128,
+            "PackageType": "Zip",
             "Role": get_role_name(),
             "Runtime": PYTHON_VERSION,
             "Timeout": 3,
diff --git a/tests/test_s3/test_s3_tagging.py b/tests/test_s3/test_s3_tagging.py
index de5215270..bef48d7e5 100644
--- a/tests/test_s3/test_s3_tagging.py
+++ b/tests/test_s3/test_s3_tagging.py
@@ -6,6 +6,7 @@ from botocore.client import ClientError
 from moto import mock_s3, settings
 from moto.s3.responses import DEFAULT_REGION_NAME
 
+from . import s3_aws_verified
 from .test_s3 import add_proxy_details
 
 
@@ -21,12 +22,11 @@ def test_get_bucket_tagging_unknown_bucket():
     )
 
 
-@mock_s3
-def test_put_object_with_tagging():
+@pytest.mark.aws_verified
+@s3_aws_verified
+def test_put_object_with_tagging(bucket_name=None):
     s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
-    bucket_name = "mybucket"
     key = "key-with-tags"
-    s3_client.create_bucket(Bucket=bucket_name)
 
     # using system tags will fail
     with pytest.raises(ClientError) as err:
@@ -34,34 +34,34 @@ def test_put_object_with_tagging():
             Bucket=bucket_name, Key=key, Body="test", Tagging="aws:foo=bar"
         )
 
-    err_value = err.value
-    assert err_value.response["Error"]["Code"] == "InvalidTag"
+    err = err.value.response["Error"]
+    assert err["Code"] == "InvalidTag"
+    assert err["Message"] == "Your TagKey cannot be prefixed with aws:"
 
     s3_client.put_object(Bucket=bucket_name, Key=key, Body="test", Tagging="foo=bar")
 
-    assert {"Key": "foo", "Value": "bar"} in s3_client.get_object_tagging(
-        Bucket=bucket_name, Key=key
-    )["TagSet"]
+    tags = s3_client.get_object_tagging(Bucket=bucket_name, Key=key)
+    assert {"Key": "foo", "Value": "bar"} in tags["TagSet"]
 
     resp = s3_client.get_object(Bucket=bucket_name, Key=key)
     assert resp["TagCount"] == 1
 
     s3_client.delete_object_tagging(Bucket=bucket_name, Key=key)
 
-    assert s3_client.get_object_tagging(Bucket=bucket_name, Key=key)["TagSet"] == []
+    tags = s3_client.get_object_tagging(Bucket=bucket_name, Key=key)
+    assert tags["TagSet"] == []
 
 
-@mock_s3
-def test_put_bucket_tagging():
+@pytest.mark.aws_verified
+@s3_aws_verified
+def test_put_bucket_tagging(bucket_name=None):
     s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
-    bucket_name = "mybucket"
-    s3_client.create_bucket(Bucket=bucket_name)
 
     # With 1 tag:
     resp = s3_client.put_bucket_tagging(
         Bucket=bucket_name, Tagging={"TagSet": [{"Key": "TagOne", "Value": "ValueOne"}]}
     )
-    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 204
 
     # With multiple tags:
     resp = s3_client.put_bucket_tagging(
@@ -75,11 +75,7 @@ def test_put_bucket_tagging():
         },
     )
 
-    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
-
-    # No tags is also OK:
-    resp = s3_client.put_bucket_tagging(Bucket=bucket_name, Tagging={"TagSet": []})
-    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
+    assert resp["ResponseMetadata"]["HTTPStatusCode"] == 204
 
     # With duplicate tag keys:
     with pytest.raises(ClientError) as err:
@@ -92,11 +88,9 @@ def test_put_bucket_tagging():
                 ]
             },
         )
-    err_value = err.value
-    assert err_value.response["Error"]["Code"] == "InvalidTag"
-    assert err_value.response["Error"]["Message"] == (
-        "Cannot provide multiple Tags with the same key"
-    )
+    err = err.value.response["Error"]
+    assert err["Code"] == "InvalidTag"
+    assert err["Message"] == "Cannot provide multiple Tags with the same key"
 
     # Cannot put tags that are "system" tags - i.e. tags that start with "aws:"
     with pytest.raises(ClientError) as ce_exc:
@@ -104,11 +98,9 @@ def test_put_bucket_tagging():
             Bucket=bucket_name,
             Tagging={"TagSet": [{"Key": "aws:sometag", "Value": "nope"}]},
         )
-    err_value = ce_exc.value
-    assert err_value.response["Error"]["Code"] == "InvalidTag"
-    assert err_value.response["Error"]["Message"] == (
-        "System tags cannot be added/updated by requester"
-    )
+    err = ce_exc.value.response["Error"]
+    assert err["Code"] == "InvalidTag"
+    assert err["Message"] == "System tags cannot be added/updated by requester"
 
     # This is OK though:
     s3_client.put_bucket_tagging(
@@ -117,11 +109,11 @@ def test_put_bucket_tagging():
     )
 
 
-@mock_s3
-def test_get_bucket_tagging():
+@pytest.mark.aws_verified
+@s3_aws_verified
+def test_get_bucket_tagging(bucket_name=None):
     s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
-    bucket_name = "mybucket"
-    s3_client.create_bucket(Bucket=bucket_name)
+
     s3_client.put_bucket_tagging(
         Bucket=bucket_name,
         Tagging={
@@ -136,16 +128,18 @@ def test_get_bucket_tagging():
     resp = s3_client.get_bucket_tagging(Bucket=bucket_name)
     assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200
     assert len(resp["TagSet"]) == 2
+    assert {"Key": "TagOne", "Value": "ValueOne"} in resp["TagSet"]
+    assert {"Key": "TagTwo", "Value": "ValueTwo"} in resp["TagSet"]
 
     # With no tags:
     s3_client.put_bucket_tagging(Bucket=bucket_name, Tagging={"TagSet": []})
 
-    with pytest.raises(ClientError) as err:
+    with pytest.raises(ClientError) as exc:
         s3_client.get_bucket_tagging(Bucket=bucket_name)
 
-    err_value = err.value
-    assert err_value.response["Error"]["Code"] == "NoSuchTagSet"
-    assert err_value.response["Error"]["Message"] == "The TagSet does not exist"
+    err = exc.value.response["Error"]
+    assert err["Code"] == "NoSuchTagSet"
+    assert err["Message"] == "The TagSet does not exist"
 
 
 @mock_s3
@@ -175,12 +169,11 @@ def test_delete_bucket_tagging():
     assert err_value.response["Error"]["Message"] == "The TagSet does not exist"
 
 
-@mock_s3
-def test_put_object_tagging():
+@pytest.mark.aws_verified
+@s3_aws_verified
+def test_put_object_tagging(bucket_name=None):
     s3_client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
-    bucket_name = "mybucket"
     key = "key-with-tags"
-    s3_client.create_bucket(Bucket=bucket_name)
 
     with pytest.raises(ClientError) as err:
         s3_client.put_object_tagging(
@@ -194,13 +187,10 @@ def test_put_object_tagging():
             },
         )
 
-    err_value = err.value
-    assert err_value.response["Error"] == {
-        "Code": "NoSuchKey",
-        "Message": "The specified key does not exist.",
-        "Key": "key-with-tags",
-        "RequestID": "7a62c49f-347e-4fc4-9331-6e8eEXAMPLE",
-    }
+    err = err.value.response["Error"]
+    assert err["Code"] == "NoSuchKey"
+    assert err["Message"] == "The specified key does not exist."
+    assert err["Key"] == key
 
     s3_client.put_object(Bucket=bucket_name, Key=key, Body="test")
 
@@ -215,6 +205,19 @@ def test_put_object_tagging():
     err_value = err.value
     assert err_value.response["Error"]["Code"] == "InvalidTag"
 
+    # Can't put more than 10 tags at the same time
+    with pytest.raises(ClientError) as exc:
+        s3_client.put_object_tagging(
+            Bucket=bucket_name,
+            Key=key,
+            Tagging={
+                "TagSet": [{"Key": f"tag{i}", "Value": "too_many"} for i in range(11)]
+            },
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "BadRequest"
+    assert err["Message"] == "Object tags cannot be greater than 10"
+
     resp = s3_client.put_object_tagging(
         Bucket=bucket_name,
         Key=key,