fix sqs query-api endpoint strategy routing

localstack/services/sqs/provider.py

@@ -302,8 +302,8 @@ def url(self, context: RequestContext) -> str:
             scheme = context.request.scheme
             host_url = f"{scheme}://{region}queue.{constants.LOCALHOST_HOSTNAME}:{config.EDGE_PORT}"
         elif config.SQS_ENDPOINT_STRATEGY == "path":
-            # localhost:4566/queue/us-east-1/00000000000/my-queue (us-east-1)
-            host_url = f"{context.request.host}/queue/{self.region}"
+            # https?://localhost:4566/queue/us-east-1/00000000000/my-queue (us-east-1)
+            host_url = f"{context.request.host_url}/queue/{self.region}"
         else:
             if config.SQS_PORT_EXTERNAL:
                 host_url = external_service_url("sqs")

localstack/services/sqs/query_api.py

@@ -20,6 +20,7 @@
 from localstack.http.dispatcher import Handler
 from localstack.services.sqs.provider import MissingParameter
 from localstack.utils.aws import aws_stack
+from localstack.utils.aws.request_context import extract_region_from_headers
 
 LOG = logging.getLogger(__name__)
 
@@ -38,14 +39,19 @@ def path_strategy_handler(request: Request, region, account_id: str, queue_name:
 
 @route(
     '/<regex("[0-9]{12}"):account_id>/<regex("[a-zA-Z0-9_-]+(.fifo)?"):queue_name>',
-    host="sqs.<regex('([a-z0-9-]+)?'):region>.localstack.cloud<regex('(:[0-9]{2,5})?'):port>",
+    host='<regex("([a-z0-9-]+\\.)?"):region>queue.localhost.localstack.cloud<regex("(:[0-9]{2,5})?"):port>',
     methods=["POST", "GET"],
 )
 def domain_strategy_handler(
     request: Request, account_id: str, queue_name: str, region: str = None, port: int = None
 ):
     """Uses the endpoint host to extract the region. See:
     https://docs.aws.amazon.com/general/latest/gr/sqs-service.html"""
+    if not region:
+        region = config.DEFAULT_REGION
+    else:
+        region = region.rstrip(".")
+
     return handle_request(request, region)
 
 
@@ -55,8 +61,18 @@ def domain_strategy_handler(
 )
 def legacy_handler(request: Request, account_id: str, queue_name: str) -> Response:
     # previously, Queue URLs were created as http://localhost:4566/000000000000/my-queue-name. Because the region is
-    # ambiguous in this request, we fall back to the default region and hope for the best.
-    return handle_request(request, config.DEFAULT_REGION)
+    # ambiguous in this request, we fall back to the region that the request is coming from (this is not how AWS
+    # behaves though).
+    if "X-Amz-Credential" in request.args:
+        region = request.args["X-Amz-Credential"].split("/")[2]
+    else:
+        region = extract_region_from_headers(request.headers)
+
+    LOG.debug(
+        "Region of queue URL %s is ambiguous, got region %s from request", request.url, region
+    )
+
+    return handle_request(request, region)
 
 
 def register(router: Router[Handler]):

tests/integration/test_sqs.py

@@ -2327,6 +2327,58 @@ def test_get_queue_url_work_for_different_queue(self, sqs_create_queue, sqs_http
         assert queue1_url not in response.text
         assert response.status_code == 200
 
+    @pytest.mark.aws_validated
+    @pytest.mark.parametrize("strategy", ["domain", "path", "off"])
+    def test_endpoint_strategy_with_multi_region(
+        self,
+        strategy,
+        sqs_http_client,
+        create_boto_client,
+        aws_http_client_factory,
+        monkeypatch,
+        cleanups,
+    ):
+        monkeypatch.setattr(config, "SQS_ENDPOINT_STRATEGY", strategy)
+
+        queue_name = f"test-queue-{short_uid()}"
+
+        sqs_region1 = create_boto_client("sqs", "us-east-1")
+        sqs_region2 = create_boto_client("sqs", "eu-west-1")
+
+        queue_region1 = sqs_region1.create_queue(QueueName=queue_name)["QueueUrl"]
+        cleanups.append(lambda: sqs_region1.delete_queue(QueueUrl=queue_region1))
+        queue_region2 = sqs_region2.create_queue(QueueName=queue_name)["QueueUrl"]
+        cleanups.append(lambda: sqs_region2.delete_queue(QueueUrl=queue_region2))
+
+        if strategy == "off":
+            assert queue_region1 == queue_region2
+        else:
+            assert queue_region1 != queue_region2
+            assert "eu-west-1" in queue_region2
+            # us-east-1 is the default region, so it's not necessarily part of the queue URL
+
+        client_region1 = aws_http_client_factory("sqs", "us-east-1")
+        client_region2 = aws_http_client_factory("sqs", "eu-west-1")
+
+        response = client_region1.get(
+            queue_region1, params={"Action": "SendMessage", "MessageBody": "foobar"}
+        )
+        assert response.ok
+
+        # shouldn't return anything
+        response = client_region2.get(
+            queue_region2, params={"Action": "ReceiveMessage", "VisibilityTimeout": "0"}
+        )
+        assert response.ok
+        assert "foobar" not in response.text
+
+        # should return the message
+        response = client_region1.get(
+            queue_region1, params={"Action": "ReceiveMessage", "VisibilityTimeout": "0"}
+        )
+        assert response.ok
+        assert "foobar" in response.text
+
     @pytest.mark.aws_validated
     def test_overwrite_queue_url_in_params(self, sqs_create_queue, sqs_http_client):
         # here, queue1 url simply serves as AWS endpoint but we pass queue2 url in the request arg