Host prefix bug (#2911)

---------

Co-authored-by: Nate Prewitt <[email protected]>

botocore/serialize.py

@@ -46,6 +46,7 @@
 
 from botocore import validate
 from botocore.compat import formatdate
+from botocore.exceptions import ParamValidationError
 from botocore.utils import (
     has_header,
     is_json_value_header,
@@ -58,6 +59,7 @@
 ISO8601 = '%Y-%m-%dT%H:%M:%SZ'
 # Same as ISO8601, but with microsecond precision.
 ISO8601_MICRO = '%Y-%m-%dT%H:%M:%S.%fZ'
+HOST_PREFIX_RE = re.compile(r"^[A-Za-z0-9\.\-]+$")
 
 
 def create_serializer(protocol_name, include_validation=True):
@@ -183,8 +185,21 @@ def _expand_host_prefix(self, parameters, operation_model):
             for member, shape in input_members.items()
             if shape.serialization.get('hostLabel')
         ]
-        format_kwargs = {name: parameters[name] for name in host_labels}
-
+        format_kwargs = {}
+        bad_labels = []
+        for name in host_labels:
+            param = parameters[name]
+            if not HOST_PREFIX_RE.match(param):
+                bad_labels.append(name)
+            format_kwargs[name] = param
+        if bad_labels:
+            raise ParamValidationError(
+                report=(
+                    f"Invalid value for parameter(s): {', '.join(bad_labels)}. "
+                    "Must contain only alphanumeric characters, hyphen, "
+                    "or period."
+                )
+            )
         return host_prefix_expression.format(**format_kwargs)
 
 

tests/functional/test_s3.py

@@ -1377,6 +1377,17 @@ def test_endpoint_redirection_fails_with_accelerate_endpoint(self):
                 RequestToken="SecretToken",
             )
 
+    def test_invalid_request_route_raises(self):
+        self.client, self.http_stubber = self.create_stubbed_s3_client(
+            region_name="us-west-2"
+        )
+        self.http_stubber.add_response()
+        with self.assertRaises(ParamValidationError):
+            self.client.write_get_object_response(
+                RequestRoute="my-route/",
+                RequestToken="SecretToken",
+            )
+
 
 class TestS3SigV4(BaseS3OperationTest):
     def setUp(self):