remove use of http2_server in replicator aws proxy

Author: thrau

aws-replicator/aws_replicator/client/auth_proxy.py

@@ -12,27 +12,23 @@
 import requests
 from botocore.awsrequest import AWSPreparedRequest
 from botocore.model import OperationModel
-from localstack import config
 from localstack import config as localstack_config
-from localstack.aws.api import HttpRequest
 from localstack.aws.protocol.parser import create_parser
 from localstack.aws.spec import load_service
 from localstack.config import external_service_url
 from localstack.constants import AWS_REGION_US_EAST_1, DOCKER_IMAGE_NAME_PRO
-from localstack.http import Request
-from localstack.utils.aws.aws_responses import requests_response
 from localstack.utils.bootstrap import setup_logging
 from localstack.utils.collections import select_attributes
 from localstack.utils.container_utils.container_client import PortMappings
 from localstack.utils.docker_utils import DOCKER_CLIENT, reserve_available_container_port
 from localstack.utils.files import new_tmp_file, save_file
 from localstack.utils.functions import run_safe
 from localstack.utils.net import get_docker_host_from_container, get_free_tcp_port
-from localstack.utils.server.http2_server import run_server
 from localstack.utils.serving import Server
 from localstack.utils.strings import short_uid, to_bytes, to_str, truncate
 from localstack_ext.bootstrap.licensingv2 import ENV_LOCALSTACK_API_KEY, ENV_LOCALSTACK_AUTH_TOKEN
-from requests import Response
+from werkzeug import Request, Response
+from werkzeug import serving as werkzeug_serving
 
 from aws_replicator import config as repl_config
 from aws_replicator.client.utils import truncate_content
@@ -41,7 +37,7 @@
 
 LOG = logging.getLogger(__name__)
 LOG.setLevel(logging.INFO)
-if config.DEBUG:
+if localstack_config.DEBUG:
     LOG.setLevel(logging.DEBUG)
 
 # TODO make configurable
@@ -62,17 +58,31 @@ def __init__(self, config: ProxyConfig, port: int = None):
         self.config = config
         port = port or get_free_tcp_port()
         super().__init__(port=port)
+        self._server = None
 
     def do_run(self):
         self.register_in_instance()
         bind_host = self.config.get("bind_host") or DEFAULT_BIND_HOST
-        proxy = run_server(port=self.port, bind_addresses=[bind_host], handler=self.proxy_request)
-        proxy.join()
 
-    def proxy_request(self, request: Request, data: bytes) -> Response:
+        # werkzeug uses under the hood a stdlib ``http.server.HTTPServer``, which should be enough to serve
+        # a simple AWS proxy. if HTTP2 and websockets are ever needed, then we should move to a more
+        # sophisticated runtime (like hypercorn or twisted)
+        self._server = werkzeug_serving.make_server(bind_host, self.port, self._wsgi_app)
+        self._server.serve_forever()
+
+    def do_shutdown(self):
+        if self._server:
+            self._server.shutdown()
+
+    @Request.application
+    def _wsgi_app(self, request: Request) -> Response:
+        """A wsgi-compatible interface for serving the proxy server."""
+        return self.proxy_request(request)
+
+    def proxy_request(self, request: Request) -> Response:
         parsed = self._extract_region_and_service(request.headers)
         if not parsed:
-            return requests_response("", status_code=400)
+            return Response(status=400)
         region_name, service_name = parsed
         query_string = to_str(request.query_string or "")
 
@@ -85,13 +95,6 @@ def proxy_request(self, request: Request, data: bytes) -> Response:
             query_string,
         )
 
-        request = HttpRequest(
-            body=data,
-            method=request.method,
-            headers=request.headers,
-            path=request.path,
-            query_string=query_string,
-        )
         session = boto3.Session()
         client = session.client(service_name, region_name=region_name)
 
@@ -119,24 +122,23 @@ def proxy_request(self, request: Request, data: bytes) -> Response:
             # send request to upstream AWS
             result = client._endpoint.make_request(operation_model, request_dict)
 
-            # create response object - TODO: to be replaced with localstack.http.Response over time
-            response = requests_response(
+            response = Response(
                 result[0].content,
-                status_code=result[0].status_code,
+                status=result[0].status_code,
                 headers=dict(result[0].headers),
             )
 
             LOG.debug(
                 "Received response for service %s from AWS: %s - %s",
                 service_name,
                 response.status_code,
-                truncate_content(response.content, max_length=500),
+                truncate_content(response.data, max_length=500),
             )
             return response
         except Exception as e:
             if LOG.isEnabledFor(logging.DEBUG):
                 LOG.exception("Error when making request to AWS service %s: %s", service_name, e)
-            return requests_response("", status_code=400)
+            return Response(status=400)
 
     def register_in_instance(self):
         port = getattr(self, "port", None)
@@ -156,7 +158,7 @@ def register_in_instance(self):
             raise
 
     def _parse_aws_request(
-        self, request: HttpRequest, service_name: str, region_name: str, client
+        self, request: Request, service_name: str, region_name: str, client
     ) -> Tuple[OperationModel, AWSPreparedRequest, Dict]:
         parser = create_parser(load_service(service_name))
         operation_model, parsed_request = parser.parse(request)
@@ -245,7 +247,7 @@ def _adjust_request_dict(self, service_name: str, request_dict: Dict):
                 req_json["QueueOwnerAWSAccountId"] = account_id
                 request_dict["body"] = to_bytes(json.dumps(req_json))
 
-    def _fix_headers(self, request: HttpRequest, service_name: str):
+    def _fix_headers(self, request: Request, service_name: str):
         if service_name == "s3":
             # fix the Host header, to avoid bucket addressing issues
             host = request.headers.get("Host") or ""

aws-replicator/tests/test_proxy_requests.py

@@ -8,7 +8,6 @@
 from localstack.aws.connect import connect_to
 from localstack.constants import TEST_AWS_ACCOUNT_ID
 from localstack.utils.aws.arns import sqs_queue_arn, sqs_queue_url_for_arn
-from localstack.utils.net import wait_for_port_open
 from localstack.utils.sync import retry
 
 from aws_replicator.client.auth_proxy import start_aws_auth_proxy
@@ -24,7 +23,7 @@ def start_aws_proxy():
 
     def _start(config: dict = None):
         proxy = start_aws_auth_proxy(config)
-        wait_for_port_open(proxy.port)
+        proxy.wait_is_up()
         proxies.append(proxy)
         return proxy