Connect to Kafka through SOCKS5 Proxy

Author: everesio

Gopkg.lock

@@ -121,6 +121,9 @@ See:
           --sasl-jaas-config-file string                   Location of JAAS config file with SASL username and password
           --sasl-password string                           SASL user password
           --sasl-username string                           SASL user name
+          --socks5-address string                          Address of SOCKS5 proxy to connect through when connecting to kafka brokers
+          --socks5-password string                         Password for SOCKS5 proxy Username/Password Authentication
+          --socks5-username string                         Username for SOCKS5 proxy Username/Password Authentication
           --tls-ca-chain-cert-file string                  PEM encoded CA's certificate file
           --tls-client-cert-file string                    PEM encoded file with client certificate
           --tls-client-key-file string                     PEM encoded file with private key for the client certificate
@@ -323,7 +326,8 @@ spec:
 * [X] Set TLS server CipherSuites and CurvePreferences
 * [ ] Optional ApiVersionsRequest before Local SASL Authentication Sequence
 * [ ] SaslHandshakeRequest v1 - Kafka 1.1.0
-* [ ] Socks5 proxy and http proxy (googleid) for outgoing connections
+* [X] Connect to Kafka through SOCKS5 Proxy
+* [ ] Proxy support for outgoing HTTP/HTTPS connections (googleid)
 * [ ] Performance tests and tuning
 * [ ] Socket buffer sizing e.g. SO_RCVBUF = 32768, SO_SNDBUF = 131072
 * [ ] Kafka connect tests

README.md

@@ -148,6 +148,11 @@ func init() {
 	// Logging
 	Server.Flags().StringVar(&c.Log.Format, "log-format", "text", "Log format text or json")
 	Server.Flags().StringVar(&c.Log.Level, "log-level", "info", "Log level debug, info, warning, error, fatal or panic")
+
+	// Socks5 to Kafka
+	Server.Flags().StringVar(&c.Socks5.ProxyAddress, "socks5-address", "", "Address of SOCKS5 proxy to connect through when connecting to kafka brokers")
+	Server.Flags().StringVar(&c.Socks5.Username, "socks5-username", "", "Username for SOCKS5 proxy Username/Password Authentication")
+	Server.Flags().StringVar(&c.Socks5.Password, "socks5-password", "", "Password for SOCKS5 proxy Username/Password Authentication")
 }
 
 func Run(_ *cobra.Command, _ []string) {

cmd/kafka-proxy/server.go

@@ -120,6 +120,11 @@ type Config struct {
 			JaasConfigFile string
 		}
 	}
+	Socks5 struct {
+		ProxyAddress string
+		Username     string
+		Password     string
+	}
 }
 
 func (c *Config) InitBootstrapServers(bootstrapServersMapping []string) (err error) {
@@ -261,5 +266,21 @@ func (c *Config) Validate() error {
 	if c.Auth.Gateway.Server.Enable && c.Auth.Gateway.Server.Timeout <= 0 {
 		return errors.New("Auth.Gateway.Server.Timeout must be greater than 0")
 	}
+	if c.Socks5.ProxyAddress == "" && (c.Socks5.Username != "" || c.Socks5.Password != "") {
+		return errors.New("Socks5.ProxyAddress must not be empty when Socks5 Username/Password is provided")
+	}
+	if (c.Socks5.Username != "" && c.Socks5.Password == "") || (c.Socks5.Username == "" && c.Socks5.Password != "") {
+		return errors.New("Both Socks5 Username and Password must be provided provided")
+	}
+	if len(c.Socks5.Username) > 255 || len(c.Socks5.Password) > 255 {
+		// RFC1929
+		return errors.New("Max length of Socks5 Username/Password is 255 chars")
+	}
+	if c.Socks5.ProxyAddress != "" {
+		if _, _, err := util.SplitHostPort(c.Socks5.ProxyAddress); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }

config/config.go

@@ -28,7 +28,7 @@ type Client struct {
 	// Config of Proxy request-response processor (instance p)
 	processorConfig ProcessorConfig
 
-	tlsConfig      *tls.Config
+	dialer         Dialer
 	tcpConnOptions TCPConnOptions
 
 	stopRun  chan struct{}
@@ -43,6 +43,10 @@ func NewClient(conns *ConnSet, c *config.Config, netAddressMappingFunc config.Ne
 	if err != nil {
 		return nil, err
 	}
+	dialer, err := newDialer(c, tlsConfig)
+	if err != nil {
+		return nil, err
+	}
 	tcpConnOptions := TCPConnOptions{
 		KeepAlive:       c.Kafka.KeepAlive,
 		WriteBufferSize: c.Kafka.ConnectionWriteBufferSize,
@@ -67,7 +71,7 @@ func NewClient(conns *ConnSet, c *config.Config, netAddressMappingFunc config.Ne
 		return nil, errors.New("Auth.Gateway.Server.Enable is enabled but tokenInfo is nil")
 	}
 
-	return &Client{conns: conns, config: c, tlsConfig: tlsConfig, tcpConnOptions: tcpConnOptions, stopRun: make(chan struct{}, 1),
+	return &Client{conns: conns, config: c, dialer: dialer, tcpConnOptions: tcpConnOptions, stopRun: make(chan struct{}, 1),
 		saslPlainAuth: &SASLPlainAuth{
 			clientID:     c.Kafka.ClientID,
 			writeTimeout: c.Kafka.WriteTimeout,
@@ -104,6 +108,40 @@ func NewClient(conns *ConnSet, c *config.Config, netAddressMappingFunc config.Ne
 		}}, nil
 }
 
+func newDialer(c *config.Config, tlsConfig *tls.Config) (Dialer, error) {
+	var rawDialer Dialer
+	if c.Socks5.ProxyAddress != "" {
+		logrus.Infof("Kafka clients will connect through the SOCKS5 proxy %s", c.Socks5.ProxyAddress)
+		rawDialer = &socks5Dialer{
+			directDialer: directDialer{
+				dialTimeout: c.Kafka.DialTimeout,
+				keepAlive:   c.Kafka.KeepAlive,
+			},
+			proxyNetwork: "tcp",
+			proxyAddr:    c.Socks5.ProxyAddress,
+			username:     c.Socks5.Username,
+			password:     c.Socks5.Password,
+		}
+	} else {
+		rawDialer = directDialer{
+			dialTimeout: c.Kafka.DialTimeout,
+			keepAlive:   c.Kafka.KeepAlive,
+		}
+	}
+	if c.Kafka.TLS.Enable {
+		if tlsConfig == nil {
+			return nil, errors.New("tlsConfig must not be nil")
+		}
+		tlsDialer := tlsDialer{
+			timeout:   c.Kafka.DialTimeout,
+			rawDialer: rawDialer,
+			config:    tlsConfig,
+		}
+		return tlsDialer, nil
+	}
+	return rawDialer, nil
+}
+
 // Run causes the client to start waiting for new connections to connSrc and
 // proxy them to the destination instance. It blocks until connSrc is closed.
 func (c *Client) Run(connSrc <-chan Conn) error {
@@ -136,7 +174,7 @@ func (c *Client) Close() {
 func (c *Client) handleConn(conn Conn) {
 	proxyConnectionsTotal.WithLabelValues(conn.BrokerAddress).Inc()
 
-	server, err := c.DialAndAuth(conn.BrokerAddress, c.tlsConfig)
+	server, err := c.DialAndAuth(conn.BrokerAddress)
 	if err != nil {
 		logrus.Infof("couldn't connect to %s: %v", conn.BrokerAddress, err)
 		conn.LocalConnection.Close()
@@ -148,15 +186,15 @@ func (c *Client) handleConn(conn Conn) {
 		}
 	}
 	c.conns.Add(conn.BrokerAddress, conn.LocalConnection)
-	localDesc := "local connection on " + conn.LocalConnection.LocalAddr().String() + " from " + conn.LocalConnection.RemoteAddr().String()
+	localDesc := "local connection on " + conn.LocalConnection.LocalAddr().String() + " from " + conn.LocalConnection.RemoteAddr().String() + " (" + conn.BrokerAddress + ")"
 	copyThenClose(c.processorConfig, server, conn.LocalConnection, conn.BrokerAddress, conn.BrokerAddress, localDesc)
 	if err := c.conns.Remove(conn.BrokerAddress, conn.LocalConnection); err != nil {
 		logrus.Info(err)
 	}
 }
 
-func (c *Client) DialAndAuth(brokerAddress string, tlsConfig *tls.Config) (net.Conn, error) {
-	conn, err := c.dial(brokerAddress, tlsConfig)
+func (c *Client) DialAndAuth(brokerAddress string) (net.Conn, error) {
+	conn, err := c.dialer.Dial("tcp", brokerAddress)
 	if err != nil {
 		return nil, err
 	}
@@ -167,20 +205,6 @@ func (c *Client) DialAndAuth(brokerAddress string, tlsConfig *tls.Config) (net.C
 	return conn, nil
 }
 
-func (c *Client) dial(brokerAddress string, tlsConfig *tls.Config) (net.Conn, error) {
-	dialer := net.Dialer{
-		Timeout:   c.config.Kafka.DialTimeout,
-		KeepAlive: c.config.Kafka.KeepAlive,
-	}
-	if c.config.Kafka.TLS.Enable {
-		if tlsConfig == nil {
-			return nil, errors.New("tlsConfig must not be nil")
-		}
-		return tls.DialWithDialer(&dialer, "tcp", brokerAddress, tlsConfig)
-	}
-	return dialer.Dial("tcp", brokerAddress)
-}
-
 func (c *Client) auth(conn net.Conn) error {
 	if c.config.Auth.Gateway.Client.Enable {
 		if err := c.authClient.sendAndReceiveGatewayAuth(conn); err != nil {

proxy/client.go

@@ -0,0 +1,124 @@
+package proxy
+
+import (
+	"crypto/tls"
+	"github.com/pkg/errors"
+	"golang.org/x/net/proxy"
+	"net"
+	"strings"
+	"time"
+)
+
+type Dialer interface {
+	Dial(network, addr string) (c net.Conn, err error)
+}
+
+type directDialer struct {
+	dialTimeout time.Duration
+	keepAlive   time.Duration
+}
+
+func (d directDialer) Dial(network, addr string) (net.Conn, error) {
+	dialer := net.Dialer{
+		Timeout:   d.dialTimeout,
+		KeepAlive: d.keepAlive,
+	}
+	return dialer.Dial(network, addr)
+}
+
+type socks5Dialer struct {
+	directDialer            directDialer
+	proxyNetwork, proxyAddr string
+	username, password      string
+}
+
+func (d socks5Dialer) Dial(network, addr string) (net.Conn, error) {
+	if d.proxyNetwork == "" || d.proxyAddr == "" {
+		return nil, errors.New("socks5 proxy network and addr must be not empty")
+	}
+	var auth *proxy.Auth
+	if d.username != "" && d.password != "" {
+		auth = &proxy.Auth{
+			User:     d.username,
+			Password: d.password,
+		}
+	}
+	socks5Dialer, err := proxy.SOCKS5(d.proxyNetwork, d.proxyAddr, auth, d.directDialer)
+	if err != nil {
+		return nil, err
+	}
+	conn, err := socks5Dialer.Dial(network, addr)
+	if err != nil {
+		return nil, err
+	}
+	return conn, nil
+}
+
+type tlsDialer struct {
+	timeout   time.Duration
+	rawDialer Dialer
+	config    *tls.Config
+}
+
+// see tls.DialWithDialer
+func (d tlsDialer) Dial(network, addr string) (net.Conn, error) {
+	if d.config == nil {
+		return nil, errors.New("tlsConfig must not be nil")
+	}
+	if d.rawDialer == nil {
+		return nil, errors.New("rawDialer must not be nil")
+	}
+
+	timeout := d.timeout
+
+	var errChannel chan error
+
+	if timeout != 0 {
+		errChannel = make(chan error, 2)
+		timer := time.AfterFunc(timeout, func() {
+			errChannel <- errors.Errorf("Handshake timeout to %s after %v", addr, timeout)
+		})
+		defer timer.Stop()
+	}
+
+	rawConn, err := d.rawDialer.Dial(network, addr)
+	if err != nil {
+		return nil, err
+	}
+
+	colonPos := strings.LastIndex(addr, ":")
+	if colonPos == -1 {
+		colonPos = len(addr)
+	}
+	hostname := addr[:colonPos]
+
+	config := d.config
+
+	// If no ServerName is set, infer the ServerName
+	// from the hostname we're connecting to.
+	if config.ServerName == "" {
+		// Make a copy to avoid polluting argument or default.
+		c := config.Clone()
+		c.ServerName = hostname
+		config = c
+	}
+
+	conn := tls.Client(rawConn, config)
+
+	if timeout == 0 {
+		err = conn.Handshake()
+	} else {
+		go func() {
+			errChannel <- conn.Handshake()
+		}()
+
+		err = <-errChannel
+	}
+
+	if err != nil {
+		rawConn.Close()
+		return nil, err
+	}
+
+	return conn, nil
+}