WIP. flex routing basically works in mitm

bubble-server/src/main/resources/packer/roles/mitmproxy/files/bubble_api.py

@@ -25,7 +25,6 @@ HEADER_USER_AGENT = 'User-Agent'
 HEADER_CONTENT_SECURITY_POLICY = 'Content-Security-Policy'
 HEADER_REFERER = 'Referer'
 HEADER_FILTER_PASSTHRU = 'X-Bubble-Passthru'
-HEADER_FLEX_AUTH = 'X-Bubble-Flex-Auth'
 
 CTX_BUBBLE_MATCHERS = 'X-Bubble-Matchers'
 CTX_BUBBLE_ABORT = 'X-Bubble-Abort'

bubble-server/src/main/resources/packer/roles/mitmproxy/files/bubble_flex.py

@@ -1,16 +1,20 @@
 #
 # Copyright (c) 2020 Bubble, Inc.  All rights reserved. For personal (non-commercial) use, see license: https://getbubblenow.com/bubble-license/
 #
-import requests
+from mitmproxy import http
 from mitmproxy.net.http import headers as nheaders
+from bubble_api import bubble_get_flex_router
+from bubble_modify import bubble_filter_response
 
-from bubble_api import HEADER_FLEX_AUTH, bubble_get_flex_router
+import requests
 
 import logging
 from logging import INFO, DEBUG, WARNING, ERROR, CRITICAL
 
 bubble_log = logging.getLogger(__name__)
 
+FLEX_TIMEOUT = 20
+
 
 def prepend_remainder_to_stream(remainder, raw):
     first = True
@@ -37,61 +41,73 @@ def set_flex_response(client_addr, flex_host, flow):
 
     if bubble_log.isEnabledFor(INFO):
         bubble_log.info('set_flex_response: found router '+repr(router)+' for flex host: '+flex_host)
+    try:
+        process_flex(flex_host, flow, router)
+    except Exception as e:
+        if bubble_log.isEnabledFor(ERROR):
+            bubble_log.error('set_flex_response: error processing: '+repr(e))
+
+
+def process_flex(flex_host, flow, router):
+
+    # build the request URL
+    scheme = flow.request.scheme
+    url = scheme + '://' + flex_host + flow.request.path
+
+    # copy request headers
+    # see: https://stackoverflow.com/questions/16789840/python-requests-cant-send-multiple-headers-with-same-key
+    request_headers = {}
+    for name in flow.request.headers:
+        if bubble_log.isEnabledFor(DEBUG):
+            bubble_log.debug('process_flex: processing request header: '+repr(name))
+        if name in request_headers:
+            request_headers[name] = request_headers[name] + "," + flow.request.headers[name]
+        else:
+            request_headers[name] = flow.request.headers[name]
 
-    # build the request
-    url = flow.request.scheme + '://' + flex_host + flow.request.path
-    headers = flow.request.headers
-    headers[HEADER_FLEX_AUTH] = router['auth']
+    # setup proxies
     proxy_url = router['proxyUrl']
     proxies = {"http": proxy_url, "https": proxy_url}
 
     # send request to flex router
     if bubble_log.isEnabledFor(DEBUG):
-        bubble_log.debug('request: sending flex request for '+url+' to '+proxy_url+' with headers='+repr(headers))
-    response = requests.request(flow.request.method, url,
-                                headers=headers,
-                                timeout=(15, 15),
-                                stream=True,
-                                data=flow.request.stream,
-                                proxies=proxies)
-
-    # Parse the response, we have to do this raw to capture the full status line
-
-    # Status line
-    # 16K should be enough to capture status line and all headers
-    # see https://stackoverflow.com/questions/686217/maximum-on-http-header-values
-    next_bytes = response.raw.read(16384)
-    response_text = next_bytes.decode()
-    lines = response_text.splitlines()
-    status_line = lines[0]
-    status_line_parts = status_line.split()
-    flow.response.http_version = status_line_parts[0]
-    flow.response.status_code = int(status_line_parts[1])
-    flow.response.reason = status_line_parts[2]
-
-    # Headers
+        bubble_log.debug('process_flex: sending flex request for '+url+' to '+proxy_url+' with headers='+repr(request_headers))
+    try:
+        response = requests.request(flow.request.method, url,
+                                    headers=request_headers,
+                                    timeout=(15, 15),
+                                    stream=True,
+                                    data=flow.request.stream,  # use the original request body, if there is one
+                                    proxies=proxies)
+    except Exception as e:
+        if bubble_log.isEnabledFor(ERROR):
+            bubble_log.error('process_flex: error sending request to '+url+': '+repr(e))
+        return
+
+    # Status line -- http version is buried in response.raw.version
+    raw_version = response.raw.version
+    if raw_version == 10:
+        http_version = 'HTTP/1.0'
+    elif raw_version == 11:
+        http_version = 'HTTP/1.1'
+    else:
+        if bubble_log.isEnabledFor(DEBUG):
+            bubble_log.debug('process_flex: invalid HTTP version detected, response.raw.version=='+repr(raw_version))
+        return
+
+    # Headers -- copy from requests dict to Headers multimap
     response_headers = nheaders.Headers()
-    end_of_headers = False
-    lines = lines[1:]
-    bytes_consumed = len(status_line) + 1
-    while True:
-        for header_line in lines[:-1]:
-            if header_line == '':
-                bytes_consumed = bytes_consumed + 1
-                end_of_headers = True
-                break
-            header_parts = header_line.split(':', 1)
-            response_headers[header_parts[0].strip()] = header_parts[1].strip()
-            bytes_consumed = bytes_consumed + len(header_line) + 1
-        if end_of_headers:
-            break
-        next_bytes = response.raw.read(8192)  # wow, headers are big! continue reading in 8K chunks
-        next_text = lines[-1] + '\n' + next_bytes.decode()
-        lines = next_text.splitlines()
+    for name in response.headers:
+        response_headers[name] = response.headers[name]
 
-    flow.response.headers = response_headers
+    # Construct the response -- use the raw response stream
+    flow.response = http.HTTPResponse(http_version=http_version,
+                                      status_code=response.status_code,
+                                      reason=response.reason,
+                                      headers=response_headers)
 
-    # Body
-    # Determine the remainder left over from parsing headers, send that first
-    remainder = next_bytes[bytes_consumed:]
-    flow.response.stream = prepend_remainder_to_stream(remainder, response.raw)
+    # Apply filters
+    bubble_filter_response(flow, response.raw)
+
+    if bubble_log.isEnabledFor(INFO):
+        bubble_log.info('process_flex: successfully requested url '+url+' from flex router, proceeding ...')

bubble-server/src/main/resources/packer/roles/mitmproxy/files/bubble_modify.py

@@ -135,14 +135,22 @@ def filter_chunk(flow, chunk, req_id, user_agent, last, content_encoding=None, c
     return response.content
 
 
-def bubble_filter_chunks(flow, chunks, req_id, user_agent, content_encoding, content_type, csp):
+def bubble_filter_chunks(flow, chunks, flex_stream, req_id, user_agent, content_encoding, content_type, csp):
     """
     chunks is a generator that can be used to iterate over all chunks.
     """
+    if bubble_log.isEnabledFor(DEBUG):
+        bubble_log.debug('bubble_filter_chunks: starting...')
     first = True
     content_length = get_flow_ctx(flow, CTX_CONTENT_LENGTH)
+    if bubble_log.isEnabledFor(DEBUG):
+        bubble_log.debug('bubble_filter_chunks: found content_length='+str(content_length))
+    if flex_stream is not None:
+        chunks = flex_stream
     try:
         for chunk in chunks:
+            if bubble_log.isEnabledFor(DEBUG):
+                bubble_log.debug('bubble_filter_chunks: filtering chunk of size '+str(len(chunk)))
             if content_length:
                 bytes_sent = get_flow_ctx(flow, CTX_CONTENT_LENGTH_SENT)
                 chunk_len = len(chunk)
@@ -166,8 +174,11 @@ def bubble_filter_chunks(flow, chunks, req_id, user_agent, content_encoding, con
         yield None
 
 
-def bubble_modify(flow, req_id, user_agent, content_encoding, content_type, csp):
-    return lambda chunks: bubble_filter_chunks(flow, chunks, req_id, user_agent, content_encoding, content_type, csp)
+def bubble_modify(flow, flex_stream, req_id, user_agent, content_encoding, content_type, csp):
+    if bubble_log.isEnabledFor(DEBUG):
+        bubble_log.debug('bubble_modify: modifying req_id='+req_id)
+    return lambda chunks: bubble_filter_chunks(flow, chunks, flex_stream, req_id,
+                                               user_agent, content_encoding, content_type, csp)
 
 
 def send_bubble_response(response):
@@ -190,17 +201,16 @@ def abort_data(content_type):
     return EMPTY_DEFAULT
 
 
-def bubble_filter_response(flow):
-    return responseheaders(flow)
-
-
 def responseheaders(flow):
+    bubble_filter_response(flow, None)
 
+
+def bubble_filter_response(flow, flex_stream):
     path = flow.request.path
     if path and path.startswith(BUBBLE_URI_PREFIX):
         if path.startswith(HEALTH_CHECK_URI):
-            if bubble_log.isEnabledFor(DEBUG):
-                bubble_log.debug('responseheaders: special bubble health check request, responding with OK')
+            #if bubble_log.isEnabledFor(DEBUG):
+            #    bubble_log.debug('responseheaders: special bubble health check request, responding with OK')
             flow.response.headers = Headers()
             flow.response.headers[HEADER_HEALTH_CHECK] = 'OK'
             flow.response.headers[HEADER_CONTENT_LENGTH] = '3'
@@ -319,7 +329,9 @@ def responseheaders(flow):
                         content_length_value = flow.response.headers.pop(HEADER_CONTENT_LENGTH, None)
                         if bubble_log.isEnabledFor(DEBUG):
                             bubble_log.debug(prefix+'content_encoding='+repr(content_encoding) + ', content_type='+repr(content_type))
-                        flow.response.stream = bubble_modify(flow, req_id, user_agent, content_encoding, content_type, csp)
+
+                        flow.response.stream = bubble_modify(flow, flex_stream, req_id,
+                                                             user_agent, content_encoding, content_type, csp)
                         if content_length_value:
                             flow.response.headers['transfer-encoding'] = 'chunked'
                             # find server_conn to set fake_chunks on