Make verify_outgoing compatible with agents not configured for TLS

consul/client.go

@@ -115,12 +115,13 @@ func NewClient(config *Config) (*Client, error) {
 	// Create server
 	c := &Client{
 		config:     config,
-		connPool:   NewPool(config.LogOutput, clientRPCConnMaxIdle, clientMaxStreams, tlsWrap),
 		eventCh:    make(chan serf.Event, serfEventBacklog),
 		logger:     logger,
 		shutdownCh: make(chan struct{}),
 	}
 
+	c.connPool = NewPool(config.LogOutput, clientRPCConnMaxIdle, clientMaxStreams, tlsWrap, c.LANMembers)
+
 	// Start lan event handlers before lan Serf setup to prevent deadlock
 	go c.lanEventHandler()
 
@@ -150,6 +151,10 @@ func (c *Client) setupSerf(conf *serf.Config, ch chan serf.Event, path string) (
 	conf.Tags["vsn_min"] = fmt.Sprintf("%d", ProtocolVersionMin)
 	conf.Tags["vsn_max"] = fmt.Sprintf("%d", ProtocolVersionMax)
 	conf.Tags["build"] = c.config.Build
+	conf.Tags["rpc_port"] = fmt.Sprintf("%d", c.config.RPCAddr.Port)
+	if c.config.VerifyIncoming {
+		conf.Tags["tls_verify_incoming"] = "1"
+	}
 	conf.MemberlistConfig.LogOutput = c.config.LogOutput
 	conf.LogOutput = c.config.LogOutput
 	conf.EventCh = ch

consul/pool.go

@@ -6,13 +6,15 @@ import (
 	"io"
 	"net"
 	"net/rpc"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/hashicorp/consul/consul/agent"
 	"github.com/hashicorp/consul/tlsutil"
 	"github.com/hashicorp/net-rpc-msgpackrpc"
+	"github.com/hashicorp/serf/serf"
 	"github.com/hashicorp/yamux"
 )
 
@@ -129,6 +131,10 @@ type ConnPool struct {
 	// The maximum number of open streams to keep
 	maxStreams int
 
+	// memberFunc is a callback to get the serf members for determining
+	// when to initiate a TLS connection
+	memberFunc func() []serf.Member
+
 	// Pool maps an address to a open connection
 	pool map[string]*Conn
 
@@ -151,11 +157,13 @@ type ConnPool struct {
 // Set maxTime to 0 to disable reaping. maxStreams is used to control
 // the number of idle streams allowed.
 // If TLS settings are provided outgoing connections use TLS.
-func NewPool(logOutput io.Writer, maxTime time.Duration, maxStreams int, tlsWrap tlsutil.DCWrapper) *ConnPool {
+func NewPool(logOutput io.Writer, maxTime time.Duration, maxStreams int, tlsWrap tlsutil.DCWrapper,
+	memberFunc func() []serf.Member) *ConnPool {
 	pool := &ConnPool{
 		logOutput:  logOutput,
 		maxTime:    maxTime,
 		maxStreams: maxStreams,
+		memberFunc: memberFunc,
 		pool:       make(map[string]*Conn),
 		limiter:    make(map[string]chan struct{}),
 		tlsWrap:    tlsWrap,
@@ -280,24 +288,51 @@ func (p *ConnPool) DialTimeout(dc string, addr net.Addr, timeout time.Duration)
 
 	// Check if TLS is enabled
 	if p.tlsWrap != nil {
-		// Switch the connection into TLS mode
-		if _, err := conn.Write([]byte{byte(rpcTLS)}); err != nil {
-			conn.Close()
+		doWrap, err := memberExpectsTLS(addr.String(), p.memberFunc())
+		if err != nil {
 			return nil, nil, err
 		}
 
-		// Wrap the connection in a TLS client
-		tlsConn, err := p.tlsWrap(dc, conn)
-		if err != nil {
-			conn.Close()
-			return nil, nil, err
+		if doWrap {
+			// Switch the connection into TLS mode
+			if _, err := conn.Write([]byte{byte(rpcTLS)}); err != nil {
+				conn.Close()
+				return nil, nil, err
+			}
+
+			// Wrap the connection in a TLS client
+			tlsConn, err := p.tlsWrap(dc, conn)
+			if err != nil {
+				conn.Close()
+				return nil, nil, err
+			}
+			conn = tlsConn
 		}
-		conn = tlsConn
 	}
 
 	return conn, hc, nil
 }
 
+// memberExpectsTLS returns true if the member with the specified address is
+// verifying incoming TLS connections
+func memberExpectsTLS(addr string, members []serf.Member) (bool, error) {
+	for _, member := range members {
+		port, err := strconv.Atoi(member.Tags["rpc_port"])
+		if err != nil {
+			return false, err
+		}
+		memberAddr := (&net.TCPAddr{IP: member.Addr, Port: port}).String()
+		if memberAddr == addr {
+			if _, ok := member.Tags["tls_verify_incoming"]; !ok {
+				return false, nil
+			}
+			break
+		}
+	}
+
+	return true, nil
+}
+
 // getNewConn is used to return a new connection
 func (p *ConnPool) getNewConn(dc string, addr net.Addr, version int) (*Conn, error) {
 	// Get a new, raw connection.

consul/raft_rpc.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/hashicorp/consul/tlsutil"
 	"github.com/hashicorp/raft"
+	"github.com/hashicorp/serf/serf"
 )
 
 // RaftLayer implements the raft.StreamLayer interface,
@@ -26,17 +27,22 @@ type RaftLayer struct {
 	closed    bool
 	closeCh   chan struct{}
 	closeLock sync.Mutex
+
+	// memberFunc is a callback to get the serf members for determining
+	// when to initiate a TLS connection
+	memberFunc func() []serf.Member
 }
 
 // NewRaftLayer is used to initialize a new RaftLayer which can
 // be used as a StreamLayer for Raft. If a tlsConfig is provided,
 // then the connection will use TLS.
-func NewRaftLayer(addr net.Addr, tlsWrap tlsutil.Wrapper) *RaftLayer {
+func NewRaftLayer(addr net.Addr, tlsWrap tlsutil.Wrapper, memberFunc func() []serf.Member) *RaftLayer {
 	layer := &RaftLayer{
-		addr:    addr,
-		connCh:  make(chan net.Conn),
-		tlsWrap: tlsWrap,
-		closeCh: make(chan struct{}),
+		addr:       addr,
+		connCh:     make(chan net.Conn),
+		tlsWrap:    tlsWrap,
+		closeCh:    make(chan struct{}),
+		memberFunc: memberFunc,
 	}
 	return layer
 }
@@ -89,16 +95,23 @@ func (l *RaftLayer) Dial(address raft.ServerAddress, timeout time.Duration) (net
 
 	// Check for tls mode
 	if l.tlsWrap != nil {
-		// Switch the connection into TLS mode
-		if _, err := conn.Write([]byte{byte(rpcTLS)}); err != nil {
-			conn.Close()
+		doWrap, err := memberExpectsTLS(string(address), l.memberFunc())
+		if err != nil {
 			return nil, err
 		}
 
-		// Wrap the connection in a TLS client
-		conn, err = l.tlsWrap(conn)
-		if err != nil {
-			return nil, err
+		if doWrap {
+			// Switch the connection into TLS mode
+			if _, err := conn.Write([]byte{byte(rpcTLS)}); err != nil {
+				conn.Close()
+				return nil, err
+			}
+
+			// Wrap the connection in a TLS client
+			conn, err = l.tlsWrap(conn)
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
 

consul/server.go

@@ -257,7 +257,6 @@ func NewServer(config *Config) (*Server, error) {
 		autopilotRemoveDeadCh: make(chan struct{}),
 		autopilotShutdownCh:   make(chan struct{}),
 		config:                config,
-		connPool:              NewPool(config.LogOutput, serverRPCCache, serverMaxStreams, tlsWrap),
 		eventChLAN:            make(chan serf.Event, 256),
 		eventChWAN:            make(chan serf.Event, 256),
 		localConsuls:          make(map[raft.ServerAddress]*agent.Server),
@@ -270,6 +269,8 @@ func NewServer(config *Config) (*Server, error) {
 		shutdownCh:            make(chan struct{}),
 	}
 
+	s.connPool = NewPool(config.LogOutput, serverRPCCache, serverMaxStreams, tlsWrap, s.LANMembers)
+
 	// Set up the autopilot policy
 	s.autopilotPolicy = &BasicAutopilot{server: s}
 
@@ -379,6 +380,7 @@ func (s *Server) setupSerf(conf *serf.Config, ch chan serf.Event, path string, w
 	conf.Tags["raft_vsn"] = fmt.Sprintf("%d", s.config.RaftConfig.ProtocolVersion)
 	conf.Tags["build"] = s.config.Build
 	conf.Tags["port"] = fmt.Sprintf("%d", addr.Port)
+	conf.Tags["rpc_port"] = fmt.Sprintf("%d", s.config.RPCAddr.Port)
 	if s.config.Bootstrap {
 		conf.Tags["bootstrap"] = "1"
 	}
@@ -388,6 +390,9 @@ func (s *Server) setupSerf(conf *serf.Config, ch chan serf.Event, path string, w
 	if s.config.NonVoter {
 		conf.Tags["nonvoter"] = "1"
 	}
+	if s.config.VerifyIncoming {
+		conf.Tags["tls_verify_incoming"] = "1"
+	}
 	conf.MemberlistConfig.LogOutput = s.config.LogOutput
 	conf.LogOutput = s.config.LogOutput
 	conf.EventCh = ch
@@ -625,7 +630,7 @@ func (s *Server) setupRPC(tlsWrap tlsutil.DCWrapper) error {
 	// Provide a DC specific wrapper. Raft replication is only
 	// ever done in the same datacenter, so we can provide it as a constant.
 	wrapper := tlsutil.SpecificDC(s.config.Datacenter, tlsWrap)
-	s.raftLayer = NewRaftLayer(advertise, wrapper)
+	s.raftLayer = NewRaftLayer(advertise, wrapper, s.LANMembers)
 	return nil
 }