add ASGI support for embedded apps

This commit replaces our WSGI implementation with a new ASGI one,
which then uses `asgiref`'s compatibility mode to still support WSGI applications.
The ASGI implementation is a bit bare-bone, but good enough for our purposes.

The major changes are:

  - We now support ASGI apps.
  - Instead of taking connections out of mitmproxy's normal processing,
    we now just set flow.response and let things continue as usual.
    This allows users to see responses in mitmproxy, use the response hook
    to modify app responses, etc. Also important for us,
    this makes the new implementation work for shenanigans like sans-io.

CHANGELOG

@@ -9,6 +9,7 @@ Unreleased: mitmproxy next
     * Prevent transparent mode from connecting to itself in the basic cases (@prinzhorn)
     * Display HTTP trailers in mitmweb (@sanlengjingvv)
     * Revamp onboarding app (@mhils)
+    * Add ASGI support for embedded apps (@mhils)
 
     * --- TODO: add new PRs above this line ---
 

examples/addons/wsgi-flask-app.py

@@ -6,7 +6,7 @@
 a single simplest-possible page.
 """
 from flask import Flask
-from mitmproxy.addons import wsgiapp
+from mitmproxy.addons import asgiapp
 
 app = Flask("proxapp")
 
@@ -19,7 +19,7 @@ def hello_world() -> str:
 addons = [
     # Host app at the magic domain "example.com" on port 80. Requests to this
     # domain and port combination will now be routed to the WSGI app instance.
-    wsgiapp.WSGIApp(app, "example.com", 80)
+    asgiapp.WSGIApp(app, "example.com", 80)
     # SSL works too, but the magic domain needs to be resolvable from the mitmproxy machine due to mitmproxy's design.
     # mitmproxy will connect to said domain and use serve its certificate (unless --no-upstream-cert is set)
     # but won't send any data.

mitmproxy/addons/asgiapp.py

@@ -0,0 +1,129 @@
+import asyncio
+import urllib.parse
+
+import asgiref.compatibility
+import asgiref.wsgi
+
+from mitmproxy import ctx, http
+
+
+class ASGIApp:
+    """
+    An addon that hosts an ASGI/WSGI HTTP app within mitmproxy, at a specified hostname and port.
+
+    Some important caveats:
+        - This implementation will block and wait until the entire HTTP response is completed before sending out data.
+        - It currently only implements the HTTP protocol (Lifespan and WebSocket are unimplemented).
+    """
+
+    def __init__(self, asgi_app, host: str, port: int):
+        asgi_app = asgiref.compatibility.guarantee_single_callable(asgi_app)
+        self.asgi_app, self.host, self.port = asgi_app, host, port
+
+    @property
+    def name(self) -> str:
+        return f"asgiapp:{self.host}:{self.port}"
+
+    def request(self, flow: http.HTTPFlow) -> None:
+        assert flow.reply
+        if (flow.request.pretty_host, flow.request.port) == (self.host, self.port) and not flow.reply.has_message:
+            flow.reply.take()  # pause hook completion
+            asyncio.create_task(serve(self.asgi_app, flow))
+
+
+class WSGIApp(ASGIApp):
+    def __init__(self, wsgi_app, host: str, port: int):
+        asgi_app = asgiref.wsgi.WsgiToAsgi(wsgi_app)
+        super().__init__(asgi_app, host, port)
+
+
+HTTP_VERSION_MAP = {
+    "HTTP/1.0": "1.0",
+    "HTTP/1.1": "1.1",
+    "HTTP/2.0": "2",
+}
+
+
+def make_scope(flow: http.HTTPFlow) -> dict:
+    # %3F is a quoted question mark
+    quoted_path = urllib.parse.quote_from_bytes(flow.request.data.path).split("%3F", maxsplit=1)
+
+    # (Unicode string) – HTTP request target excluding any query string, with percent-encoded
+    # sequences and UTF-8 byte sequences decoded into characters.
+    path = quoted_path[0]
+
+    # (byte string) – URL portion after the ?, percent-encoded.
+    query_string: bytes
+    if len(quoted_path) > 1:
+        query_string = quoted_path[1].encode()
+    else:
+        query_string = b""
+
+    return {
+        "type": "http",
+        "asgi": {
+            "version": "3.0",
+            "spec_version": "2.1",
+        },
+        "http_version": HTTP_VERSION_MAP.get(flow.request.http_version, "1.1"),
+        "method": flow.request.method,
+        "scheme": flow.request.scheme,
+        "path": path,
+        "raw_path": flow.request.path,
+        "query_string": query_string,
+        "headers": list(list(x) for x in flow.request.headers.fields),
+        "client": flow.client_conn.address,
+        "extensions": {
+            "mitmproxy.master": ctx.master,
+        }
+    }
+
+
+async def serve(app, flow: http.HTTPFlow):
+    """
+    Serves app on flow.
+    """
+    assert flow.reply
+
+    scope = make_scope(flow)
+    done = asyncio.Event()
+    received_body = False
+
+    async def receive():
+        nonlocal received_body
+        if not received_body:
+            received_body = True
+            return {
+                "type": "http.request",
+                "body": flow.request.raw_content,
+            }
+        else:  # pragma: no cover
+            # We really don't expect this to be called a second time, but what to do?
+            # We just wait until the request is done before we continue here with sending a disconnect.
+            await done.wait()
+            return {
+                "type": "http.disconnect"
+            }
+
+    async def send(event):
+        if event["type"] == "http.response.start":
+            flow.response = http.HTTPResponse.make(event["status"], b"", event.get("headers", []))
+            flow.response.decode()
+        elif event["type"] == "http.response.body":
+            flow.response.content += event.get("body", b"")
+            if not event.get("more_body", False):
+                flow.reply.ack()
+        else:
+            raise AssertionError(f"Unexpected event: {event['type']}")
+
+    try:
+        await app(scope, receive, send)
+        if not flow.reply.has_message:
+            raise RuntimeError(f"no response sent.")
+    except Exception as e:
+        ctx.log.error(f"Error in asgi app: {e}")
+        flow.response = http.HTTPResponse.make(500, b"ASGI Error.")
+        flow.reply.ack(force=True)
+    finally:
+        flow.reply.commit()
+        done.set()

mitmproxy/addons/onboarding.py

@@ -1,16 +1,16 @@
-from mitmproxy.addons import wsgiapp
+from mitmproxy.addons import asgiapp
 from mitmproxy.addons.onboardingapp import app
 from mitmproxy import ctx
 
 APP_HOST = "mitm.it"
 APP_PORT = 80
 
 
-class Onboarding(wsgiapp.WSGIApp):
+class Onboarding(asgiapp.WSGIApp):
     name = "onboarding"
 
     def __init__(self):
-        super().__init__(app, None, None)
+        super().__init__(app, APP_HOST, APP_PORT)
 
     def load(self, loader):
         loader.add_option(

setup.py

@@ -61,6 +61,7 @@
     # https://packaging.python.org/en/latest/requirements/#install-requires
     # It is not considered best practice to use install_requires to pin dependencies to specific versions.
     install_requires=[
+        "asgiref>=3.2.10, <3.3",
         "blinker>=1.4, <1.5",
         "Brotli>=1.0,<1.1",
         "certifi>=2019.9.11",  # no semver here - this should always be on the last release!