Skip SASL negotiation with datanodes in one special case

When data.transfer.protection is set but not dfs.encrypt.data.transfer,
hadoop expects us to check if the datanode is running on a privileged port. If
it is, then no SASL negatiation should take place. Don't ask me why.

client.go

@@ -91,6 +91,12 @@ type ClientOptions struct {
 	// has dfs.encrypt.data.transfer enabled, this setting is ignored and
 	// a level of "privacy" is used.
 	DataTransferProtection string
+	// skipSaslForPrivilegedDatanodePorts implements a strange edge case present
+	// in the official java client. If data.transfer.protection is set but not
+	// dfs.encrypt.data.transfer, and the datanode is running on a privileged
+	// port, the client connects without doing a SASL handshake. This field is
+	// only set by ClientOptionsFromConf.
+	skipSaslForPrivilegedDatanodePorts bool
 }
 
 // ClientOptionsFromConf attempts to load any relevant configuration options
@@ -163,6 +169,9 @@ func ClientOptionsFromConf(conf hadoopconf.HadoopConf) ClientOptions {
 
 	if strings.ToLower(conf["dfs.encrypt.data.transfer"]) == "true" {
 		options.DataTransferProtection = "privacy"
+	} else {
+		// See the comment for this property above.
+		options.skipSaslForPrivilegedDatanodePorts = true
 	}
 
 	return options
@@ -352,10 +361,11 @@ func (c *Client) wrapDatanodeDial(dc dialContext, token *hadoop.TokenProto) (dia
 		}
 
 		return (&transfer.SaslDialer{
-			DialFunc:   dc,
-			Key:        key,
-			Token:      token,
-			EnforceQop: c.options.DataTransferProtection,
+			DialFunc:                  dc,
+			Key:                       key,
+			Token:                     token,
+			EnforceQop:                c.options.DataTransferProtection,
+			SkipSaslOnPrivilegedPorts: c.options.skipSaslForPrivilegedDatanodePorts,
 		}).DialContext, nil
 	}
 

internal/transfer/sasl_dialer.go

@@ -26,10 +26,11 @@ const (
 // data protection level is specified by the server, whether it be wire
 // encryption or integrity checks.
 type SaslDialer struct {
-	DialFunc   func(ctx context.Context, network, addr string) (net.Conn, error)
-	Key        *hdfs.DataEncryptionKeyProto
-	Token      *hadoop.TokenProto
-	EnforceQop string
+	DialFunc                  func(ctx context.Context, network, addr string) (net.Conn, error)
+	Key                       *hdfs.DataEncryptionKeyProto
+	Token                     *hadoop.TokenProto
+	EnforceQop                string
+	SkipSaslOnPrivilegedPorts bool
 }
 
 func (d *SaslDialer) DialContext(ctx context.Context, network, addr string) (net.Conn, error) {
@@ -42,6 +43,16 @@ func (d *SaslDialer) DialContext(ctx context.Context, network, addr string) (net
 		return nil, err
 	}
 
+	// If the port is privileged, and a certain combination of configuration
+	// variables are set, hadoop expects us to skip SASL negotiation. See the
+	// documentation for ClientOptions in the top-level package for more detail.
+	if d.SkipSaslOnPrivilegedPorts {
+		if addr, ok := conn.RemoteAddr().(*net.TCPAddr); ok && addr.Port < 1024 {
+			return conn, nil
+		}
+
+	}
+
 	return d.wrapDatanodeConn(conn)
 }