TUN-3615: added support to proxy tcp streams

added ingress.DefaultStreamHandler and a basic test for tcp stream proxy
moved websocket.Stream to ingress
cloudflared no longer picks tcpstream host from header

ingress/ingress.go

@@ -24,6 +24,11 @@ var (
 	ErrURLIncompatibleWithIngress = errors.New("You can't set the --url flag (or $TUNNEL_URL) when using multiple-origin ingress rules")
 )
 
+const (
+	ServiceBastion = "bastion"
+	ServiceTeamnet = "teamnet-proxy"
+)
+
 // FindMatchingRule returns the index of the Ingress Rule which matches the given
 // hostname and path. This function assumes the last rule matches everything,
 // which is the case if the rules were instantiated via the ingress#Validate method
@@ -90,7 +95,7 @@ func parseSingleOriginService(c *cli.Context, allowURLFromArgs bool) (originServ
 		return new(helloWorld), nil
 	}
 	if c.IsSet(config.BastionFlag) {
-		return newBridgeService(), nil
+		return newBridgeService(nil), nil
 	}
 	if c.IsSet("url") {
 		originURL, err := config.ValidateUrl(c, allowURLFromArgs)
@@ -159,12 +164,14 @@ func validate(ingress []config.UnvalidatedIngressRule, defaults OriginRequestCon
 			service = &srv
 		} else if r.Service == "hello_world" || r.Service == "hello-world" || r.Service == "helloworld" {
 			service = new(helloWorld)
-		} else if r.Service == "bastion" || cfg.BastionMode {
+		} else if r.Service == ServiceBastion || cfg.BastionMode {
 			// Bastion mode will always start a Websocket proxy server, which will
 			// overwrite the localService.URL field when `start` is called. So,
 			// leave the URL field empty for now.
 			cfg.BastionMode = true
-			service = newBridgeService()
+			service = newBridgeService(nil)
+		} else if r.Service == ServiceTeamnet {
+			service = newBridgeService(DefaultStreamHandler)
 		} else {
 			// Validate URL services
 			u, err := url.Parse(r.Service)

ingress/ingress_test.go

@@ -315,7 +315,7 @@ ingress:
 			want: []Rule{
 				{
 					Hostname: "bastion.foo.com",
-					Service:  newBridgeService(),
+					Service:  newBridgeService(nil),
 					Config:   setConfig(originRequestFromYAML(config.OriginRequestConfig{}), config.OriginRequestConfig{BastionMode: &tr}),
 				},
 				{
@@ -335,7 +335,7 @@ ingress:
 			want: []Rule{
 				{
 					Hostname: "bastion.foo.com",
-					Service:  newBridgeService(),
+					Service:  newBridgeService(nil),
 					Config:   setConfig(originRequestFromYAML(config.OriginRequestConfig{}), config.OriginRequestConfig{BastionMode: &tr}),
 				},
 				{

ingress/origin_connection.go

@@ -17,10 +17,36 @@ type OriginConnection interface {
 	Close()
 }
 
+type streamHandlerFunc func(originConn io.ReadWriter, remoteConn net.Conn)
+
+// Stream copies copy data to & from provided io.ReadWriters.
+func Stream(conn, backendConn io.ReadWriter) {
+	proxyDone := make(chan struct{}, 2)
+
+	go func() {
+		io.Copy(conn, backendConn)
+		proxyDone <- struct{}{}
+	}()
+
+	go func() {
+		io.Copy(backendConn, conn)
+		proxyDone <- struct{}{}
+	}()
+
+	// If one side is done, we are done.
+	<-proxyDone
+}
+
+// DefaultStreamHandler is an implementation of streamHandlerFunc that
+// performs a two way io.Copy between originConn and remoteConn.
+func DefaultStreamHandler(originConn io.ReadWriter, remoteConn net.Conn) {
+	Stream(originConn, remoteConn)
+}
+
 // tcpConnection is an OriginConnection that directly streams to raw TCP.
 type tcpConnection struct {
 	conn          net.Conn
-	streamHandler func(tunnelConn io.ReadWriter, originConn net.Conn)
+	streamHandler streamHandlerFunc
 }
 
 func (tc *tcpConnection) Stream(tunnelConn io.ReadWriter) {
@@ -39,7 +65,7 @@ type wsConnection struct {
 }
 
 func (wsc *wsConnection) Stream(tunnelConn io.ReadWriter) {
-	websocket.Stream(tunnelConn, wsc.wsConn.UnderlyingConn())
+	Stream(tunnelConn, wsc.wsConn.UnderlyingConn())
 }
 
 func (wsc *wsConnection) Close() {

ingress/origin_proxy.go

@@ -2,13 +2,14 @@ package ingress
 
 import (
 	"fmt"
-	"io"
 	"net"
 	"net/http"
 	"net/url"
 	"strings"
 
+	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/h2mux"
+	"github.com/pkg/errors"
 )
 
 // HTTPOriginProxy can be implemented by origin services that want to proxy http requests.
@@ -63,7 +64,21 @@ func (o *bridgeService) EstablishConnection(r *http.Request) (OriginConnection,
 	return o.client.connect(r, dest)
 }
 
+// getRequestHost returns the host of the http.Request.
+func getRequestHost(r *http.Request) (string, error) {
+	if r.Host != "" {
+		return r.Host, nil
+	}
+	if r.URL != nil {
+		return r.URL.Host, nil
+	}
+	return "", errors.New("host not found")
+}
+
 func (o *bridgeService) destination(r *http.Request) (string, error) {
+	if connection.IsTCPStream(r) {
+		return getRequestHost(r)
+	}
 	jumpDestination := r.Header.Get(h2mux.CFJumpDestinationHeader)
 	if jumpDestination == "" {
 		return "", fmt.Errorf("Did not receive final destination from client. The --destination flag is likely not set on the client side")
@@ -85,7 +100,7 @@ func (o *singleTCPService) EstablishConnection(r *http.Request) (OriginConnectio
 }
 
 type tcpClient struct {
-	streamHandler func(originConn io.ReadWriter, remoteConn net.Conn)
+	streamHandler streamHandlerFunc
 }
 
 func (c *tcpClient) connect(r *http.Request, addr string) (OriginConnection, error) {

ingress/origin_proxy_test.go

@@ -91,7 +91,7 @@ func TestBridgeServiceDestination(t *testing.T) {
 			wantErr: true,
 		},
 	}
-	s := newBridgeService()
+	s := newBridgeService(nil)
 	for _, test := range tests {
 		r := &http.Request{
 			Header: test.header,

ingress/origin_service.go

@@ -81,9 +81,12 @@ type bridgeService struct {
 	client *tcpClient
 }
 
-func newBridgeService() *bridgeService {
+// if streamHandler is nil, a default one is set.
+func newBridgeService(streamHandler streamHandlerFunc) *bridgeService {
 	return &bridgeService{
-		client: &tcpClient{},
+		client: &tcpClient{
+			streamHandler: streamHandler,
+		},
 	}
 }
 
@@ -92,10 +95,15 @@ func (o *bridgeService) String() string {
 }
 
 func (o *bridgeService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
+	// streamHandler is already set by the constructor.
+	if o.client.streamHandler != nil {
+		return nil
+	}
+
 	if cfg.ProxyType == socksProxy {
 		o.client.streamHandler = socks.StreamHandler
 	} else {
-		o.client.streamHandler = websocket.DefaultStreamHandler
+		o.client.streamHandler = DefaultStreamHandler
 	}
 	return nil
 }
@@ -136,7 +144,7 @@ func (o *singleTCPService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdo
 	if cfg.ProxyType == socksProxy {
 		o.client.streamHandler = socks.StreamHandler
 	} else {
-		o.client.streamHandler = websocket.DefaultStreamHandler
+		o.client.streamHandler = DefaultStreamHandler
 	}
 	return nil
 }