Proxy protocol on outbound tcp, tcp+sni and tcp with tls connection (#…

…598)

* Proxy protocol on outbound tcp, tcp+sni and tcp with tls connections; integration tests

* Replace header fmt call with concat

docs/content/cfg/_index.md

@@ -30,6 +30,7 @@ Option                                     | Description
 `deny=ip:10.0.0.0/8,ip:fe80::1234`         | Deny requests that source from the `10.0.0.0/8` CIDR mask or `fe80::1234`.  All other requests will be allowed.
 `strip=/path`                              | Forward `/path/to/file` as `/to/file`
 `proto=tcp`                                | Upstream service is TCP, `dst` must be `:port`
+`pxyproto=true`                            | Enables PROXY protocol on outbount TCP connection
 `proto=https`                              | Upstream service is HTTPS
 `tlsskipverify=true`                       | Disable TLS cert validation for HTTPS upstream
 `host=name`                                | Set the `Host` header to `name`. If `name == 'dst'` then the `Host` header will be set to the registered upstream host name

docs/content/quickstart/_index.md

@@ -46,6 +46,7 @@ and you need to add a separate `urlprefix-` tag for every `host/path` prefix the
 
 	# TCP examples
 	urlprefix-:3306 proto=tcp                          # route external port 3306
+	urlprefix-:3306 proto=tcp pxyproto=true            # enables PROXY protocol on outbount TCP connection
 
 	# GRPC/S examples
 	urlprefix-/my.service/Method proto=grpc                      # method specific route

proxy/tcp/proxy_proto.go

@@ -0,0 +1,24 @@
+package tcp
+
+import (
+	"net"
+)
+
+// WriteProxyHeader extracts remote and local IP address and port
+// combinations from incoming connection and writes the PROXY proto
+// header to the outgoing connection
+func WriteProxyHeader(out, in net.Conn) error {
+	clientAddr, clientPort, _ := net.SplitHostPort(in.RemoteAddr().String())
+	serverAddr, serverPort, _ := net.SplitHostPort(in.LocalAddr().String())
+
+	var proto string
+	if net.ParseIP(clientAddr).To4() != nil {
+		proto = "TCP4"
+	} else {
+		proto = "TCP6"
+	}
+
+	header := "PROXY " + proto + " " + clientAddr + " " + serverAddr + " " + clientPort + " " + serverPort + "\r\n"
+	_, err := out.Write([]byte(header))
+	return err
+}

proxy/tcp/sni_proxy.go

@@ -113,6 +113,18 @@ func (p *SNIProxy) ServeTCP(in net.Conn) error {
 	}
 	defer out.Close()
 
+	// enable PROXY protocol support on outbound connection
+	if t.ProxyProto {
+		err := WriteProxyHeader(out, in)
+		if err != nil {
+			log.Print("[WARN] tcp+sni: write proxy protocol header failed. ", err)
+			if p.ConnFail != nil {
+				p.ConnFail.Inc(1)
+			}
+			return err
+		}
+	}
+
 	// write the data already read from the connection
 	n, err := out.Write(data)
 	if err != nil {

proxy/tcp/tcp_proxy.go

@@ -62,6 +62,18 @@ func (p *Proxy) ServeTCP(in net.Conn) error {
 	}
 	defer out.Close()
 
+	// enable PROXY protocol support on outbound connection
+	if t.ProxyProto {
+		err := WriteProxyHeader(out, in)
+		if err != nil {
+			log.Print("[WARN] tcp: write proxy protocol header failed. ", err)
+			if p.ConnFail != nil {
+				p.ConnFail.Inc(1)
+			}
+			return err
+		}
+	}
+
 	errc := make(chan error, 2)
 	cp := func(dst io.Writer, src io.Reader, c metrics.Counter) {
 		errc <- copyBuffer(dst, src, c)

proxy/tcp/tcptest/dialer.go

@@ -18,13 +18,24 @@ func NewRetryDialer() *RetryDialer {
 // the timeout has been reached. The default timeout is one
 // second and the default sleep interval is 100ms.
 type RetryDialer struct {
-	Dialer  net.Dialer
-	Timeout time.Duration
-	Sleep   time.Duration
+	Dialer     net.Dialer
+	Timeout    time.Duration
+	Sleep      time.Duration
+	ProxyProto bool
 }
 
 func (d *RetryDialer) Dial(network, addr string) (c net.Conn, err error) {
-	dial := func() (net.Conn, error) { return d.Dialer.Dial(network, addr) }
+	dial := func() (net.Conn, error) {
+		conn, err := d.Dialer.Dial(network, addr)
+		if err != nil {
+			return nil, err
+		}
+		if d.ProxyProto {
+			pxy := "PROXY TCP4 1.2.3.4 5.6.7.8 12345 54321\r\n"
+			conn.Write([]byte(pxy))
+		}
+		return conn, err
+	}
 	return retry(dial, d.Timeout, d.Sleep)
 }
 
@@ -33,14 +44,25 @@ func NewTLSRetryDialer(cfg *tls.Config) *TLSRetryDialer {
 }
 
 type TLSRetryDialer struct {
-	TLS     *tls.Config
-	Dialer  net.Dialer
-	Timeout time.Duration
-	Sleep   time.Duration
+	TLS        *tls.Config
+	Dialer     net.Dialer
+	Timeout    time.Duration
+	Sleep      time.Duration
+	ProxyProto bool
 }
 
 func (d *TLSRetryDialer) Dial(network, addr string) (c net.Conn, err error) {
-	dial := func() (net.Conn, error) { return tls.Dial(network, addr, d.TLS) }
+	dial := func() (net.Conn, error) {
+		conn, err := net.Dial(network, addr)
+		if err != nil {
+			return nil, err
+		}
+		if d.ProxyProto {
+			pxy := "PROXY TCP4 1.2.3.4 5.6.7.8 12345 54321\r\n"
+			conn.Write([]byte(pxy))
+		}
+		return tls.Client(conn, d.TLS), nil
+	}
 	return retry(dial, d.Timeout, d.Sleep)
 }
 

proxy/tcp/tcptest/server.go

@@ -4,7 +4,9 @@ import (
 	"crypto/tls"
 	"fmt"
 	"net"
+	"time"
 
+	proxyproto "github.com/armon/go-proxyproto"
 	"github.com/fabiolb/fabio/proxy/internal"
 	"github.com/fabiolb/fabio/proxy/tcp"
 )
@@ -108,3 +110,25 @@ func newLocalListener() net.Listener {
 	}
 	return l
 }
+
+func NewServerWithProxyProto(h tcp.Handler) *Server {
+	srv := NewUnstartedServerWithProxyProto(h)
+	srv.Start()
+	return srv
+}
+
+func NewTLSServerWithProxyProto(h tcp.Handler) *Server {
+	srv := NewUnstartedServerWithProxyProto(h)
+	srv.StartTLS()
+	return srv
+}
+
+func NewUnstartedServerWithProxyProto(h tcp.Handler) *Server {
+	return &Server{
+		Listener: &proxyproto.Listener{
+			Listener:           newLocalListener(),
+			ProxyHeaderTimeout: time.Duration(100 * time.Millisecond),
+		},
+		Config: &tcp.Server{Handler: h},
+	}
+}

proxy/tcp_integration_test.go

@@ -199,3 +199,194 @@ func testRoundtrip(t *testing.T, c net.Conn) {
 		t.Fatalf("got %q want %q", got, want)
 	}
 }
+
+var proxyHandler tcp.HandlerFunc = func(c net.Conn) error {
+	defer c.Close()
+	line, _, err := bufio.NewReader(c).ReadLine()
+	if err != nil {
+		return err
+	}
+
+	str := " " + c.RemoteAddr().String()
+	line = append(line, []byte(str)...)
+	_, err = c.Write(line)
+	return err
+}
+
+// TestTCPProxyWithProxyProtoEnables tests proxying an unencrypted TCP connection
+// to a TCP upstream server with proxy protocol enabed on upstream connection
+func TestTCPProxyWithProxyProto(t *testing.T) {
+	srv := tcptest.NewServerWithProxyProto(proxyHandler)
+	defer srv.Close()
+
+	// start proxy
+	proxyAddr := "127.0.0.1:57778"
+	go func() {
+		h := &tcp.Proxy{
+			Lookup: func(h string) *route.Target {
+				tbl, _ := route.NewTable("route add srv :57778 tcp://" + srv.Addr + " opts \"pxyproto=true\"")
+				tgt := tbl.LookupHost(h, route.Picker["rr"])
+				return tgt
+			},
+		}
+		l := config.Listen{Addr: proxyAddr, ProxyProto: true}
+		if err := ListenAndServeTCP(l, h, nil); err != nil {
+			t.Log("ListenAndServeTCP: ", err)
+		}
+	}()
+	defer Close()
+
+	// connect to proxy
+	dialer := tcptest.NewRetryDialer()
+	dialer.ProxyProto = true
+	out, err := dialer.Dial("tcp", proxyAddr)
+	if err != nil {
+		t.Fatalf("net.Dial: %#v", err)
+	}
+	defer out.Close()
+
+	testProxyProto(t, out)
+}
+
+// TestTCPProxyWithTLSWithProxyProto tests proxying an encrypted TCP connection
+// to an unencrypted upstream TCP server with proxy protocol enabled.
+// The proxy extract the proxy protocl header and terminates the TLS connection.
+func TestTCPProxyWithTLSWithProxyProto(t *testing.T) {
+	srv := tcptest.NewServerWithProxyProto(proxyHandler)
+	defer srv.Close()
+
+	// setup cert source
+	dir, err := ioutil.TempDir("", "fabio")
+	if err != nil {
+		t.Fatal("ioutil.TempDir", err)
+	}
+	defer os.RemoveAll(dir)
+
+	mustWrite := func(name string, data []byte) {
+		path := filepath.Join(dir, name)
+		if err := ioutil.WriteFile(path, data, 0644); err != nil {
+			t.Fatalf("ioutil.WriteFile: %s", err)
+		}
+	}
+	mustWrite("example.com-key.pem", internal.LocalhostKey)
+	mustWrite("example.com-cert.pem", internal.LocalhostCert)
+
+	// start tcp proxy
+	proxyAddr := "127.0.0.1:57779"
+	go func() {
+		cs := config.CertSource{Name: "cs", Type: "path", CertPath: dir}
+		src, err := cert.NewSource(cs)
+		if err != nil {
+			t.Fatal("cert.NewSource: ", err)
+		}
+		cfg, err := cert.TLSConfig(src, false, 0, 0, nil)
+		if err != nil {
+			t.Fatal("cert.TLSConfig: ", err)
+		}
+
+		h := &tcp.Proxy{
+			Lookup: func(string) *route.Target {
+				return &route.Target{URL: &url.URL{Host: srv.Addr}, ProxyProto: true}
+			},
+		}
+
+		l := config.Listen{Addr: proxyAddr, ProxyProto: true}
+		if err := ListenAndServeTCP(l, h, cfg); err != nil {
+			// closing the listener returns this error from the accept loop
+			// which we can ignore.
+			if err.Error() != "accept tcp 127.0.0.1:57779: use of closed network connection" {
+				t.Log("ListenAndServeTCP: ", err)
+			}
+		}
+	}()
+	defer Close()
+
+	// give cert store some time to pick up certs
+	time.Sleep(250 * time.Millisecond)
+
+	rootCAs := x509.NewCertPool()
+	if ok := rootCAs.AppendCertsFromPEM(internal.LocalhostCert); !ok {
+		t.Fatal("could not parse cert")
+	}
+	cfg := &tls.Config{
+		RootCAs:    rootCAs,
+		ServerName: "example.com",
+	}
+
+	// connect to proxy
+	dialer := tcptest.NewTLSRetryDialer(cfg)
+	dialer.ProxyProto = true
+	out, err := dialer.Dial("tcp", proxyAddr)
+	if err != nil {
+		t.Fatalf("tls.Dial: %#v", err)
+	}
+	defer out.Close()
+
+	testProxyProto(t, out)
+}
+
+// TestTCPSNIProxyWithProxyProto tests proxying an encrypted TCP connection adding
+// proxy protocol header to an upstream TCP service without decrypting the traffic.
+// The upstream server extracts the proxy protocol and terminates the TLS connection.
+func TestTCPSNIProxyWithProxyProto(t *testing.T) {
+	srv := tcptest.NewTLSServerWithProxyProto(proxyHandler)
+	defer srv.Close()
+
+	// start tcp proxy
+	proxyAddr := "127.0.0.1:57778"
+	go func() {
+		h := &tcp.SNIProxy{
+			Lookup: func(string) *route.Target {
+				return &route.Target{URL: &url.URL{Host: srv.Addr}, ProxyProto: true}
+			},
+		}
+		l := config.Listen{Addr: proxyAddr, ProxyProto: true}
+		if err := ListenAndServeTCP(l, h, nil); err != nil {
+			t.Log("ListenAndServeTCP: ", err)
+		}
+	}()
+	defer Close()
+
+	rootCAs := x509.NewCertPool()
+	if ok := rootCAs.AppendCertsFromPEM(internal.LocalhostCert); !ok {
+		t.Fatal("could not parse cert")
+	}
+	cfg := &tls.Config{
+		RootCAs:    rootCAs,
+		ServerName: "example.com",
+	}
+
+	// connect to proxy
+	dialer := tcptest.NewTLSRetryDialer(cfg)
+	dialer.ProxyProto = true
+	out, err := dialer.Dial("tcp", proxyAddr)
+	if err != nil {
+		t.Fatalf("tls.Dial: %#v", err)
+	}
+	defer out.Close()
+
+	testProxyProto(t, out)
+}
+
+func testProxyProto(t *testing.T, c net.Conn) {
+	// send data to server
+	_, err := c.Write([]byte("foo\n"))
+	if err != nil {
+		t.Fatal("out.Write: ", err)
+	}
+
+	// read response which should be
+	// PROXY proto header
+	line, _, err := bufio.NewReader(c).ReadLine()
+	if err != nil {
+		t.Fatal("readLine: ", err)
+	}
+
+	// remote := c.RemoteAddr().String()
+	// local := c.LocalAddr().String()
+
+	// compare
+	if got, want := line, []byte("foo 1.2.3.4:12345"); !bytes.Equal(got, want) {
+		t.Fatalf("got %q want %q", got, want)
+	}
+}

route/route.go

@@ -74,6 +74,7 @@ func (r *Route) addTarget(service string, targetURL *url.URL, fixedWeight float6
 		t.StripPath = opts["strip"]
 		t.TLSSkipVerify = opts["tlsskipverify"] == "true"
 		t.Host = opts["host"]
+		t.ProxyProto = opts["pxyproto"] == "true"
 
 		if opts["redirect"] != "" {
 			t.RedirectCode, err = strconv.Atoi(opts["redirect"])

route/target.go

@@ -60,6 +60,9 @@ type Target struct {
 
 	// name of the auth handler for this target
 	AuthScheme string
+
+	// ProxyProto enables PROXY Protocol on upstream connection
+	ProxyProto bool
 }
 
 func (t *Target) BuildRedirectURL(requestURL *url.URL) {