Support proxy protocol v2 in MySQL

lib/multiplexer/proxyline.go

@@ -60,6 +60,38 @@ func (p *ProxyLine) String() string {
 	return fmt.Sprintf("PROXY %s %s %s %d %d\r\n", p.Protocol, p.Source.IP.String(), p.Destination.IP.String(), p.Source.Port, p.Destination.Port)
 }
 
+// Bytes returns on-the wire bytes representation of proxy line conforming to the proxy v2 protocol
+func (p *ProxyLine) Bytes() []byte {
+	var b []byte
+	b = append(b, proxyV2Prefix...)
+	b = append(b, (Version2<<4)|ProxyCommand)
+
+	switch p.Protocol {
+	case TCP4:
+		b = append(b, ProtocolTCP4)
+		b = append(b, 0, 12)
+		b = append(b, p.Source.IP.To4()...)
+		b = append(b, portToBytes(p.Source)...)
+		b = append(b, p.Destination.IP.To4()...)
+		b = append(b, portToBytes(p.Destination)...)
+	case TCP6:
+		b = append(b, ProtocolTCP6)
+		b = append(b, 36)
+		b = append(b, p.Source.IP.To16()...)
+		b = append(b, portToBytes(p.Source)...)
+		b = append(b, p.Destination.IP.To16()...)
+		b = append(b, portToBytes(p.Destination)...)
+	}
+
+	return b
+}
+
+func portToBytes(addr net.TCPAddr) []byte {
+	b := make([]byte, 2)
+	binary.BigEndian.PutUint16(b, uint16(addr.Port))
+	return b
+}
+
 // ReadProxyLine reads proxy line protocol from the reader
 func ReadProxyLine(reader *bufio.Reader) (*ProxyLine, error) {
 	line, err := reader.ReadString('\n')

lib/multiplexer/testproxy.go

@@ -33,11 +33,12 @@ type TestProxy struct {
 	target   string
 	closeCh  chan (struct{})
 	log      logrus.FieldLogger
+	v2       bool
 }
 
 // NewTestProxy creates a new test proxy that sends a proxy-line when
 // proxying connections to the provided target address.
-func NewTestProxy(target string) (*TestProxy, error) {
+func NewTestProxy(target string, v2 bool) (*TestProxy, error) {
 	listener, err := net.Listen("tcp", "localhost:0")
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -47,6 +48,7 @@ func NewTestProxy(target string) (*TestProxy, error) {
 		target:   target,
 		closeCh:  make(chan struct{}),
 		log:      logrus.WithField(trace.Component, "test:proxy"),
+		v2:       v2,
 	}, nil
 }
 
@@ -128,7 +130,11 @@ func (p *TestProxy) sendProxyLine(clientConn, serverConn net.Conn) error {
 		Destination: net.TCPAddr{IP: net.ParseIP(serverAddr.Host()), Port: serverAddr.Port(0)},
 	}
 	p.log.Debugf("Sending %v to %v.", proxyLine.String(), serverConn.RemoteAddr().String())
-	_, err = serverConn.Write([]byte(proxyLine.String()))
+	if p.v2 {
+		_, err = serverConn.Write(proxyLine.Bytes())
+	} else {
+		_, err = serverConn.Write([]byte(proxyLine.String()))
+	}
 	if err != nil {
 		return trace.Wrap(err)
 	}

lib/multiplexer/wrappers.go

@@ -84,7 +84,16 @@ func (c *Conn) Detect() (Protocol, error) {
 
 // ReadProxyLine reads proxy-line from the connection.
 func (c *Conn) ReadProxyLine() (*ProxyLine, error) {
-	proxyLine, err := ReadProxyLine(c.reader)
+	var proxyLine *ProxyLine
+	protocol, err := c.Detect()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if protocol == ProtoProxyV2 {
+		proxyLine, err = ReadProxyLineV2(c.reader)
+	} else {
+		proxyLine, err = ReadProxyLine(c.reader)
+	}
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}

lib/srv/db/mysql/proxy.go

@@ -212,7 +212,7 @@ func (p *Proxy) maybeReadProxyLine(conn *multiplexer.Conn) error {
 	if err != nil {
 		return trace.Wrap(err)
 	}
-	if proto != multiplexer.ProtoProxy {
+	if proto != multiplexer.ProtoProxy && proto != multiplexer.ProtoProxyV2 {
 		return nil
 	}
 	proxyLine, err := conn.ReadProxyLine()

lib/srv/db/proxy_test.go

@@ -40,17 +40,22 @@ func TestProxyProtocolPostgres(t *testing.T) {
 
 	testCtx.createUserAndRole(ctx, t, "alice", "admin", []string{"postgres"}, []string{"postgres"})
 
-	// Point our proxy to the Teleport's db listener on the multiplexer.
-	proxy, err := multiplexer.NewTestProxy(testCtx.mux.DB().Addr().String())
-	require.NoError(t, err)
-	t.Cleanup(func() { proxy.Close() })
-	go proxy.Serve()
-
-	// Connect to the proxy instead of directly to Postgres listener and make
-	// sure the connection succeeds.
-	psql, err := testCtx.postgresClientWithAddr(ctx, proxy.Address(), "alice", "postgres", "postgres", "postgres")
-	require.NoError(t, err)
-	require.NoError(t, psql.Close(ctx))
+	for _, v2 := range []bool{false, true} {
+		v2 := v2
+		t.Run(name(v2), func(t *testing.T) {
+			// Point our proxy to the Teleport's db listener on the multiplexer.
+			proxy, err := multiplexer.NewTestProxy(testCtx.mux.DB().Addr().String(), v2)
+			require.NoError(t, err)
+			t.Cleanup(func() { proxy.Close() })
+			go proxy.Serve()
+
+			// Connect to the proxy instead of directly to Postgres listener and make
+			// sure the connection succeeds.
+			psql, err := testCtx.postgresClientWithAddr(ctx, proxy.Address(), "alice", "postgres", "postgres", "postgres")
+			require.NoError(t, err)
+			require.NoError(t, psql.Close(ctx))
+		})
+	}
 }
 
 // TestProxyProtocolMySQL ensures that clients can successfully connect to a
@@ -63,17 +68,22 @@ func TestProxyProtocolMySQL(t *testing.T) {
 
 	testCtx.createUserAndRole(ctx, t, "alice", "admin", []string{"root"}, []string{types.Wildcard})
 
-	// Point our proxy to the Teleport's MySQL listener.
-	proxy, err := multiplexer.NewTestProxy(testCtx.mysqlListener.Addr().String())
-	require.NoError(t, err)
-	t.Cleanup(func() { proxy.Close() })
-	go proxy.Serve()
-
-	// Connect to the proxy instead of directly to MySQL listener and make
-	// sure the connection succeeds.
-	mysql, err := testCtx.mysqlClientWithAddr(proxy.Address(), "alice", "mysql", "root")
-	require.NoError(t, err)
-	require.NoError(t, mysql.Close())
+	for _, v2 := range []bool{false, true} {
+		v2 := v2
+		t.Run(name(v2), func(t *testing.T) {
+			// Point our proxy to the Teleport's MySQL listener.
+			proxy, err := multiplexer.NewTestProxy(testCtx.mysqlListener.Addr().String(), v2)
+			require.NoError(t, err)
+			t.Cleanup(func() { proxy.Close() })
+			go proxy.Serve()
+
+			// Connect to the proxy instead of directly to MySQL listener and make
+			// sure the connection succeeds.
+			mysql, err := testCtx.mysqlClientWithAddr(proxy.Address(), "alice", "mysql", "root")
+			require.NoError(t, err)
+			require.NoError(t, mysql.Close())
+		})
+	}
 }
 
 // TestProxyProtocolMongo ensures that clients can successfully connect to a
@@ -86,17 +96,22 @@ func TestProxyProtocolMongo(t *testing.T) {
 
 	testCtx.createUserAndRole(ctx, t, "alice", "admin", []string{"admin"}, []string{types.Wildcard})
 
-	// Point our proxy to the Teleport's TLS listener.
-	proxy, err := multiplexer.NewTestProxy(testCtx.webListener.Addr().String())
-	require.NoError(t, err)
-	t.Cleanup(func() { proxy.Close() })
-	go proxy.Serve()
-
-	// Connect to the proxy instead of directly to Teleport listener and make
-	// sure the connection succeeds.
-	mongo, err := testCtx.mongoClientWithAddr(ctx, proxy.Address(), "alice", "mongo", "admin")
-	require.NoError(t, err)
-	require.NoError(t, mongo.Disconnect(ctx))
+	for _, v2 := range []bool{false, true} {
+		v2 := v2
+		t.Run(name(v2), func(t *testing.T) {
+			// Point our proxy to the Teleport's TLS listener.
+			proxy, err := multiplexer.NewTestProxy(testCtx.webListener.Addr().String(), false)
+			require.NoError(t, err)
+			t.Cleanup(func() { proxy.Close() })
+			go proxy.Serve()
+
+			// Connect to the proxy instead of directly to Teleport listener and make
+			// sure the connection succeeds.
+			mongo, err := testCtx.mongoClientWithAddr(ctx, proxy.Address(), "alice", "mongo", "admin")
+			require.NoError(t, err)
+			require.NoError(t, mongo.Disconnect(ctx))
+		})
+	}
 }
 
 func TestProxyProtocolRedis(t *testing.T) {
@@ -106,23 +121,28 @@ func TestProxyProtocolRedis(t *testing.T) {
 
 	testCtx.createUserAndRole(ctx, t, "alice", "admin", []string{"admin"}, []string{types.Wildcard})
 
-	// Point our proxy to the Teleport's TLS listener.
-	proxy, err := multiplexer.NewTestProxy(testCtx.webListener.Addr().String())
-	require.NoError(t, err)
-	t.Cleanup(func() { proxy.Close() })
-	go proxy.Serve()
-
-	// Connect to the proxy instead of directly to Teleport listener and make
-	// sure the connection succeeds.
-	redisClient, err := testCtx.redisClientWithAddr(ctx, proxy.Address(), "alice", "redis", "admin")
-	require.NoError(t, err)
-
-	// Send ECHO to Redis server and check if we get it back.
-	resp := redisClient.Echo(ctx, "hello")
-	require.NoError(t, resp.Err())
-	require.Equal(t, "hello", resp.Val())
-
-	require.NoError(t, redisClient.Close())
+	for _, v2 := range []bool{false, true} {
+		v2 := v2
+		t.Run(name(v2), func(t *testing.T) {
+			// Point our proxy to the Teleport's TLS listener.
+			proxy, err := multiplexer.NewTestProxy(testCtx.webListener.Addr().String(), false)
+			require.NoError(t, err)
+			t.Cleanup(func() { proxy.Close() })
+			go proxy.Serve()
+
+			// Connect to the proxy instead of directly to Teleport listener and make
+			// sure the connection succeeds.
+			redisClient, err := testCtx.redisClientWithAddr(ctx, proxy.Address(), "alice", "redis", "admin")
+			require.NoError(t, err)
+
+			// Send ECHO to Redis server and check if we get it back.
+			resp := redisClient.Echo(ctx, "hello")
+			require.NoError(t, resp.Err())
+			require.Equal(t, "hello", resp.Val())
+
+			require.NoError(t, redisClient.Close())
+		})
+	}
 }
 
 // TestProxyClientDisconnectDueToIdleConnection ensures that idle clients will be disconnected.
@@ -232,3 +252,10 @@ func TestExtractMySQLVersion(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, "8.0.25", version)
 }
+
+func name(v2 bool) string {
+	if v2 {
+		return "v2"
+	}
+	return "v1"
+}