WIP. passthru + flex routing works. code cleanup is next

bubble-server/src/main/java/bubble/service/device/FlexRouterInfo.java

@@ -29,8 +29,8 @@ public class FlexRouterInfo {
     }
 
     @JsonIgnore public String getVpnIp () { return router.getIp(); }
-    @JsonIgnore public int getPort () { return router.getPort(); }
 
+    public int getPort () { return router.getPort(); }
     public String getProxyUrl () { return router.proxyBaseUri(); }
 
     public boolean hasGeoLocation () { return hasDeviceStatus() && deviceStatus.getLocation() != null && deviceStatus.getLocation().hasLatLon(); }

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

@@ -139,7 +139,7 @@ def bubble_get_flex_router(client_addr):
                 bubble_log.debug('bubble_get_flex_routes: no router found for '+client_addr)
         else:
             if bubble_log.isEnabledFor(ERROR):
-                bubble_log.error('bubble_get_flex_routes: API call failed with status: '+response.status_code)
+                bubble_log.error('bubble_get_flex_routes: API call failed with HTTP status: '+str(response.status_code))
         return None
 
     except Exception as e:
@@ -284,6 +284,14 @@ def is_flex_domain(client_addr, fqdn):
     return False
 
 
+def original_flex_ip(client_addr, fqdns):
+    for fqdn in fqdns:
+        ip = REDIS.get("flexOriginal~"+client_addr+"~"+fqdn)
+        if ip is not None:
+            return ip.decode()
+    return None
+
+
 def health_check_response(flow):
     #if bubble_log.isEnabledFor(DEBUG):
     #    bubble_log.debug('health_check_response: special bubble health check request, responding with OK')

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

@@ -33,8 +33,9 @@ from logging import INFO, DEBUG, WARNING, ERROR, CRITICAL
 
 import traceback
 from bubble_api import bubble_conn_check, bubble_activity_log, REDIS, redis_set, \
-    is_bubble_request, is_sage_request, is_not_from_vpn
+    is_bubble_request, is_sage_request, is_not_from_vpn, is_flex_domain, bubble_get_flex_router, original_flex_ip
 from bubble_config import bubble_host, bubble_host_alias, bubble_sage_host, bubble_sage_ip4, bubble_sage_ip6, cert_validation_host
+from bubble_flex_passthru import BubbleFlexPassthruLayer
 
 bubble_log = logging.getLogger(__name__)
 
@@ -182,36 +183,26 @@ def check_bubble_connection(client_addr, server_addr, fqdns, security_level):
         if security_level['level'] == SEC_MAX:
             if bubble_log.isEnabledFor(DEBUG):
                 bubble_log.debug('check_bubble_connection: bubble API returned ' + str(check_response) +' for FQDN/addr ' + str(fqdns) +'/' + str(server_addr) + ', security_level=maximum, returning Block')
-            return {'fqdns': fqdns, 'addr': server_addr, 'passthru': False, 'flex': False, 'block': True, 'reason': 'bubble_error'}
+            return {'fqdns': fqdns, 'addr': server_addr, 'passthru': False, 'block': True, 'reason': 'bubble_error'}
         else:
             if bubble_log.isEnabledFor(DEBUG):
                 bubble_log.debug('check_bubble_connection: bubble API returned ' + str(check_response) +' for FQDN/addr ' + str(fqdns) +'/' + str(server_addr) + ', returning True')
-            return {'fqdns': fqdns, 'addr': server_addr, 'passthru': True, 'flex': False, 'reason': 'bubble_error'}
+            return {'fqdns': fqdns, 'addr': server_addr, 'passthru': True, 'reason': 'bubble_error'}
 
     elif check_response == 'passthru':
         if bubble_log.isEnabledFor(DEBUG):
             bubble_log.debug('check_bubble_connection: bubble API returned ' + str(check_response) +' for FQDN/addr ' + str(fqdns) +'/' + str(server_addr) + ', returning True')
-        return {'fqdns': fqdns, 'addr': server_addr, 'passthru': True, 'flex': False, 'reason': 'bubble_passthru'}
+        return {'fqdns': fqdns, 'addr': server_addr, 'passthru': True, 'reason': 'bubble_passthru'}
 
     elif check_response == 'block':
         if bubble_log.isEnabledFor(DEBUG):
             bubble_log.debug('check_bubble_connection: bubble API returned ' + str(check_response) +' for FQDN/addr ' + str(fqdns) +'/' + str(server_addr) + ', returning Block')
-        return {'fqdns': fqdns, 'addr': server_addr, 'passthru': False, 'flex': False, 'block': True, 'reason': 'bubble_block'}
-
-    elif check_response == 'passthru_flex':
-        if bubble_log.isEnabledFor(DEBUG):
-            bubble_log.debug('check_bubble_connection: bubble API returned ' + str(check_response) +' for FQDN/addr ' + str(fqdns) +'/' + str(server_addr) + ', returning True')
-        return {'fqdns': fqdns, 'addr': server_addr, 'passthru': True, 'flex': True, 'reason': 'bubble_passthru_flex'}
-
-    elif check_response == 'noop_flex':
-        if bubble_log.isEnabledFor(DEBUG):
-            bubble_log.debug('check_bubble_connection: bubble API returned ' + str(check_response) +' for FQDN/addr ' + str(fqdns) +'/' + str(server_addr) + ', returning True')
-        return {'fqdns': fqdns, 'addr': server_addr, 'passthru': False, 'flex': True, 'reason': 'bubble_no_passthru_flex'}
+        return {'fqdns': fqdns, 'addr': server_addr, 'passthru': False, 'block': True, 'reason': 'bubble_block'}
 
     else:
         if bubble_log.isEnabledFor(DEBUG):
             bubble_log.debug('check_bubble_connection: bubble API returned ' + str(check_response) +' for FQDN/addr ' + str(fqdns) +'/' + str(server_addr) + ', returning False')
-        return {'fqdns': fqdns, 'addr': server_addr, 'passthru': False, 'flex': False, 'reason': 'bubble_no_passthru'}
+        return {'fqdns': fqdns, 'addr': server_addr, 'passthru': False, 'reason': 'bubble_no_passthru'}
 
 
 def check_connection(client_addr, server_addr, fqdns, security_level):
@@ -240,28 +231,50 @@ def check_connection(client_addr, server_addr, fqdns, security_level):
     return check_response
 
 
-def next_layer(next_layer):
-    if isinstance(next_layer, TlsLayer) and next_layer._client_tls:
-        client_hello = net_tls.ClientHello.from_file(next_layer.client_conn.rfile)
-        client_addr = next_layer.client_conn.address[0]
-        server_addr = next_layer.server_conn.address[0]
+def check_passthru_flex(client_addr, server_addr, fqdns):
+    if fqdns:
+        for fqdn in fqdns:
+            if is_flex_domain(client_addr, fqdn):
+                return True
+    else:
+        return is_flex_domain(client_addr, server_addr)
+
+
+def passthru_flex_port(client_addr, fqdns):
+    router = bubble_get_flex_router(client_addr)
+    if router is None or 'auth' not in router:
+        if bubble_log.isEnabledFor(INFO):
+            bubble_log.info('apply_passthru_flex: no flex router for fqdn(s): '+repr(fqdns))
+    elif 'port' in router:
+        return router['port']
+    else:
+        if bubble_log.isEnabledFor(WARNING):
+            bubble_log.warning('apply_passthru_flex: flex router found but has no port ('+repr(router)+') for fqdn(s): '+repr(fqdns))
+    return None
+
+
+def next_layer(layer):
+    if isinstance(layer, TlsLayer) and layer._client_tls:
+        client_hello = net_tls.ClientHello.from_file(layer.client_conn.rfile)
+        client_addr = layer.client_conn.address[0]
+        server_addr = layer.server_conn.address[0]
         if bubble_log.isEnabledFor(DEBUG):
             bubble_log.debug('next_layer: STARTING: client='+ client_addr+' server='+server_addr)
         if client_hello.sni:
             fqdn = client_hello.sni.decode()
             if bubble_log.isEnabledFor(DEBUG):
                 bubble_log.debug('next_layer: using fqdn in SNI: '+ fqdn)
-            fqdns = [ fqdn ]
+            fqdns = [fqdn]
         else:
             fqdns = fqdns_for_addr(server_addr)
             if bubble_log.isEnabledFor(DEBUG):
                 bubble_log.debug('next_layer: NO fqdn in sni, using fqdns from DNS: '+ str(fqdns))
-        next_layer.fqdns = fqdns
+        layer.fqdns = fqdns
         no_fqdns = fqdns is None or len(fqdns) == 0
         security_level = get_device_security_level(client_addr, fqdns)
-        next_layer.security_level = security_level
-        next_layer.do_block = False
-        called_check_api = False
+        layer.security_level = security_level
+        layer.do_block = False
+        check_for_flex = False
         if is_bubble_request(server_addr, fqdns):
             if bubble_log.isEnabledFor(INFO):
                 bubble_log.info('next_layer: enabling passthru for LOCAL bubble='+server_addr+' (bubble_host ('+bubble_host+') in fqdns or bubble_host_alias ('+bubble_host_alias+') in fqdns) regardless of security_level='+repr(security_level)+' for client='+client_addr+', fqdns='+repr(fqdns))
@@ -277,7 +290,7 @@ def next_layer(next_layer):
             if bubble_log.isEnabledFor(WARNING):
                 bubble_log.warning('next_layer: enabling block for non-VPN client='+client_addr+', fqdns='+str(fqdns))
             bubble_activity_log(client_addr, server_addr, 'conn_block_non_vpn', fqdns)
-            next_layer.__class__ = TlsBlock
+            layer.__class__ = TlsBlock
             return
 
         elif security_level['level'] == SEC_OFF:
@@ -304,34 +317,43 @@ def next_layer(next_layer):
             if bubble_log.isEnabledFor(INFO):
                 bubble_log.info('next_layer: calling check_connection for server='+server_addr+', fqdns='+str(fqdns)+', client='+client_addr+' with security_level='+repr(security_level))
             check = check_connection(client_addr, server_addr, fqdns, security_level)
-            called_check_api = True
 
-        if check is None or ('passthru' in check and check['passthru'] and ('flex' not in check or not check['flex'])):
+        if check is None or ('passthru' in check and check['passthru']):
             if bubble_log.isEnabledFor(INFO):
                 bubble_log.info('next_layer: enabling passthru for server=' + server_addr+', fqdns='+str(fqdns))
             bubble_activity_log(client_addr, server_addr, 'tls_passthru', fqdns)
-            next_layer_replacement = RawTCPLayer(next_layer.ctx, ignore=True)
-            next_layer.reply.send(next_layer_replacement)
+            flex_port = None
+            if check_passthru_flex(client_addr, server_addr, fqdns):
+                if bubble_log.isEnabledFor(DEBUG):
+                    bubble_log.debug('next_layer: applying flex passthru for server=' + server_addr+', fqdns='+str(fqdns))
+                flex_port = passthru_flex_port(client_addr, fqdns)
+                if flex_port:
+                    layer_replacement = BubbleFlexPassthruLayer(layer.ctx, ('127.0.0.1', flex_port), fqdns[0], 443)
+                    layer.reply.send(layer_replacement)
+            if flex_port is None:
+                layer_replacement = RawTCPLayer(layer.ctx, ignore=True)
+                layer.reply.send(layer_replacement)
 
         elif 'block' in check and check['block']:
             if bubble_log.isEnabledFor(INFO):
                 bubble_log.info('next_layer: enabling block for server=' + server_addr+', fqdns='+str(fqdns))
             bubble_activity_log(client_addr, server_addr, 'conn_block', fqdns)
             if show_block_stats(client_addr, fqdns) and security_level['level'] != SEC_BASIC:
-                next_layer.do_block = True
-                next_layer.__class__ = TlsFeedback
+                layer.do_block = True
+                layer.__class__ = TlsFeedback
             else:
-                next_layer.__class__ = TlsBlock
+                layer.__class__ = TlsBlock
 
         elif security_level['level'] == SEC_BASIC:
             if bubble_log.isEnabledFor(INFO):
                 bubble_log.info('next_layer: check='+repr(check)+' but security_level='+repr(security_level)+', enabling passthru for server=' + server_addr+', fqdns='+str(fqdns))
             bubble_activity_log(client_addr, server_addr, 'tls_passthru', fqdns)
-            next_layer_replacement = RawTCPLayer(next_layer.ctx, ignore=True)
-            next_layer.reply.send(next_layer_replacement)
+            # todo
+            layer_replacement = RawTCPLayer(layer.ctx, ignore=True)
+            layer.reply.send(layer_replacement)
 
         else:
             if bubble_log.isEnabledFor(INFO):
                 bubble_log.info('next_layer: disabling passthru (with TlsFeedback) for client_addr='+client_addr+', server_addr='+server_addr+', fqdns='+str(fqdns))
             bubble_activity_log(client_addr, server_addr, 'tls_intercept', fqdns)
-            next_layer.__class__ = TlsFeedback
+            layer.__class__ = TlsFeedback

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

@@ -0,0 +1,123 @@
+#
+# Copyright (c) 2020 Bubble, Inc.  All rights reserved. For personal (non-commercial) use, see license: https://getbubblenow.com/bubble-license/
+#
+# Parts of this are borrowed from rawtcp.py in the mitmproxy project. The mitmproxy license is reprinted here:
+#
+# Copyright (c) 2013, Aldo Cortesi. All rights reserved.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+import socket
+
+from OpenSSL import SSL
+
+import mitmproxy.net.tcp
+from mitmproxy.exceptions import MitmproxyException
+from mitmproxy import tcp
+from mitmproxy import flow
+from mitmproxy import exceptions
+from mitmproxy.proxy.protocol import base
+from mitmproxy.connections import ServerConnection
+from mitmproxy.http import make_connect_request
+from mitmproxy.net.http.http1 import assemble_request, read_response
+
+import traceback
+import logging
+from logging import INFO, DEBUG, WARNING, ERROR, CRITICAL
+
+bubble_log = logging.getLogger(__name__)
+
+
+class BubbleFlexPassthruException(MitmproxyException):
+    pass
+
+
+class BubbleFlexPassthruLayer(base.Layer):
+    chunk_size = 4096
+    proxy_addr = None
+    host = None
+    port = None
+
+    def __init__(self, ctx, proxy_addr, host, port):
+        bubble_log.info('__init__ called with ctx='+repr(ctx)+' and ctx.server_conn=('+repr(ctx.server_conn)+') and proxy_addr='+repr(proxy_addr))
+        self.ignore = True
+        self.proxy_addr = proxy_addr
+        self.server_conn = ServerConnection(proxy_addr)
+        self.host = host
+        self.port = port
+        ctx.server_conn = self.server_conn
+        super().__init__(ctx)
+        bubble_log.info('__init__ finished, self.server_conn='+repr(self.server_conn))
+
+    def __call__(self):
+        bubble_log.info('__call__ starting, self.server_conn='+repr(self.server_conn))
+        self.connect()
+        client = self.client_conn.connection
+        server = self.server_conn.connection
+
+        buf = memoryview(bytearray(self.chunk_size))
+
+        connect_req = make_connect_request((self.host, self.port))
+        server.send(assemble_request(connect_req))
+        resp = server.recv(1024).decode()
+        if not resp.startswith('HTTP/1.1 200 OK'):
+            raise BubbleFlexPassthruException('CONNECT request error: '+resp)
+
+        conns = [client, server]
+
+        # https://github.com/openssl/openssl/issues/6234
+        for conn in conns:
+            if isinstance(conn, SSL.Connection) and hasattr(SSL._lib, "SSL_clear_mode"):
+                SSL._lib.SSL_clear_mode(conn._ssl, SSL._lib.SSL_MODE_AUTO_RETRY)
+
+        try:
+            while not self.channel.should_exit.is_set():
+                r = mitmproxy.net.tcp.ssl_read_select(conns, 10)
+                for conn in r:
+                    dst = server if conn == client else client
+                    try:
+                        size = conn.recv_into(buf, self.chunk_size)
+                    except (SSL.WantReadError, SSL.WantWriteError):
+                        continue
+                    if not size:
+                        conns.remove(conn)
+                        # Shutdown connection to the other peer
+                        if isinstance(conn, SSL.Connection):
+                            # We can't half-close a connection, so we just close everything here.
+                            # Sockets will be cleaned up on a higher level.
+                            return
+                        else:
+                            dst.shutdown(socket.SHUT_WR)
+
+                        if len(conns) == 0:
+                            return
+                        continue
+
+                    tcp_message = tcp.TCPMessage(dst == server, buf[:size].tobytes())
+                    dst.sendall(tcp_message.content)
+
+        except (socket.error, exceptions.TcpException, SSL.Error) as e:
+            just_the_string = traceback.format_exc()
+            bubble_log.info('~~~~~~~~~~~~~~~~~~~~~~~~~~~~ __call__ exception: '+repr(e)+' from '+just_the_string)
+            if not self.ignore:
+                f.error = flow.Error("TCP connection closed unexpectedly: {}".format(repr(e)))
+                self.channel.tell("tcp_error", f)
+        finally:
+            if not self.ignore:
+                self.channel.tell("tcp_end", f)

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

@@ -252,19 +252,19 @@ class Rerouter:
         return host
 
     def request(self, flow):
-        flex_host = self.bubble_handle_request(flow)
+        host = self.bubble_handle_request(flow)
         path = flow.request.path
         if is_bubble_special_path(path):
             #if bubble_log.isEnabledFor(DEBUG):
             #    bubble_log.debug('request: is_bubble_special_path('+path+') returned true, sending special bubble response')
             special_bubble_response(flow)
 
-        elif flex_host is not None:
+        elif host is not None:
             client_addr = flow.client_conn.address[0]
-            if is_flex_domain(client_addr, flex_host):
+            if is_flex_domain(client_addr, host):
                 if bubble_log.isEnabledFor(DEBUG):
-                    bubble_log.debug('request: is_flex_domain('+flex_host+') returned true, sending flex response')
-                set_flex_response(client_addr, flex_host, flow)
+                    bubble_log.debug('request: is_flex_domain('+host+') returned true, sending flex response')
+                set_flex_response(client_addr, host, flow)
 
 
 addons = [Rerouter()]

bubble-server/src/main/resources/packer/roles/mitmproxy/files/run_mitm.sh

@@ -23,7 +23,7 @@ BUBBLE_PORT=${PORT} mitmdump \
   --set block_private=false \
   --set termlog_verbosity=warn \
   --set flow_detail=0 \
-  --set stream_large_bodies=5m \
+  --set stream_large_bodies=1 \
   --set keep_host_header \
   -s ./bubble_debug.py \
   -s ./bubble_conn_check.py \