diff --git a/go.mod b/go.mod index a4c53f82a..bea2b2d3a 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,6 @@ require ( github.com/grpc-ecosystem/grpc-gateway/v2 v2.0.0-beta.4.0.20200813083929-9bdca3d79b97 github.com/hashicorp/dbassert v0.0.0-20200602142727-a6709b436ad6 github.com/hashicorp/errwrap v1.0.0 - github.com/hashicorp/go-alpnmux v0.0.0-20200822022353-14527cfc58e5 github.com/hashicorp/go-cleanhttp v0.5.1 github.com/hashicorp/go-hclog v0.14.1 github.com/hashicorp/go-kms-wrapping v0.5.14 diff --git a/go.sum b/go.sum index 7204821b9..63a2b9782 100644 --- a/go.sum +++ b/go.sum @@ -537,8 +537,6 @@ github.com/hashicorp/dbassert v0.0.0-20200602142727-a6709b436ad6 h1:Nejn3Kli0VXW github.com/hashicorp/dbassert v0.0.0-20200602142727-a6709b436ad6/go.mod h1:+B5eZq7vXqN4Gxkgjm4ub1bwG6OTcYzFL0r+bMvWorg= github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA= github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= -github.com/hashicorp/go-alpnmux v0.0.0-20200822022353-14527cfc58e5 h1:2YnwbJbobkTULUOhJ77seiSGqd7tD3V7zFXn7u5666k= -github.com/hashicorp/go-alpnmux v0.0.0-20200822022353-14527cfc58e5/go.mod h1:KvpteZzIafT4tRAuQ9vVRBgZyqeVCS2B2177fNAyEZc= github.com/hashicorp/go-cleanhttp v0.5.0/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80= github.com/hashicorp/go-cleanhttp v0.5.1 h1:dH3aiDG9Jvb5r5+bYHsikaOUIpcM0xvgMXVoDkXMzJM= github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80= @@ -1095,7 +1093,6 @@ go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= -go.uber.org/atomic v1.5.1/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= go.uber.org/atomic v1.6.0 h1:Ezj3JGmsOnG1MoRWQkPBsKLe9DwWD9QeXzTRzzldNVk= go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw= diff --git a/internal/cmd/base/listener.go b/internal/cmd/base/listener.go index 3b6b4957b..9ecaf28c2 100644 --- a/internal/cmd/base/listener.go +++ b/internal/cmd/base/listener.go @@ -13,7 +13,7 @@ import ( // certificates that use it can be parsed. _ "crypto/sha512" - "github.com/hashicorp/go-alpnmux" + "github.com/hashicorp/boundary/internal/libs/alpnmux" "github.com/hashicorp/go-hclog" "github.com/hashicorp/shared-secure-libs/configutil" "github.com/hashicorp/shared-secure-libs/listenerutil" diff --git a/internal/libs/alpnmux/mux.go b/internal/libs/alpnmux/mux.go new file mode 100644 index 000000000..cb54319c4 --- /dev/null +++ b/internal/libs/alpnmux/mux.go @@ -0,0 +1,315 @@ +package alpnmux + +import ( + "bufio" + "context" + "crypto/tls" + "errors" + "fmt" + "net" + "strings" + "sync" + + "github.com/hashicorp/go-hclog" +) + +const ( + // NoProto is used when the connection isn't actually TLS + NoProto = "(none)" + + // DefaultProto is used when there is an ALPN we don't actually know about. + // If no protos are specified on an incoming TLS connection we will first + // look for a proto of ""; if not found, will use DefaultProto. On a + // connection that has protos defined, we will look for that proto first, + // then DefaultProto. + DefaultProto = "(*)" +) + +type bufferedConn struct { + net.Conn + buffer *bufio.Reader +} + +func (b *bufferedConn) Read(p []byte) (int, error) { + return b.buffer.Read(p) +} + +type muxedListener struct { + connMutex *sync.RWMutex + ctx context.Context + addr net.Addr + proto string + tlsConf *tls.Config + connCh chan net.Conn + closed bool + closeFunc func() + closeOnce *sync.Once +} + +type ALPNMux struct { + ctx context.Context + baseLn net.Listener + log hclog.Logger + cancel context.CancelFunc + muxMap *sync.Map +} + +func New(baseLn net.Listener, log hclog.Logger) *ALPNMux { + ctx, cancel := context.WithCancel(context.Background()) + ret := &ALPNMux{ + ctx: ctx, + log: log, + cancel: cancel, + muxMap: new(sync.Map), + baseLn: baseLn, + } + go ret.accept() + return ret +} + +func (l *ALPNMux) Addr() net.Addr { + return l.baseLn.Addr() +} + +func (l *ALPNMux) Close() error { + return l.baseLn.Close() +} + +func (l *ALPNMux) SetLogger(log hclog.Logger) { + l.log = log +} + +func (l *ALPNMux) RegisterProto(proto string, tlsConf *tls.Config) (net.Listener, error) { + switch proto { + case NoProto: + if tlsConf != nil { + return nil, errors.New("tls config cannot be non-nil when using NoProto") + } + default: + if tlsConf == nil { + return nil, errors.New("nil tls config given") + } + } + sub := &muxedListener{ + connMutex: new(sync.RWMutex), + ctx: l.ctx, + addr: l.baseLn.Addr(), + proto: proto, + tlsConf: tlsConf, + connCh: make(chan net.Conn), + closeOnce: new(sync.Once), + } + _, loaded := l.muxMap.LoadOrStore(proto, sub) + if loaded { + close(sub.connCh) + return nil, fmt.Errorf("proto %q already registered", proto) + } + + sub.closeFunc = func() { + go l.UnregisterProto(proto) + } + + if l.log != nil && l.log.IsDebug() { + l.log.Debug("registered", "proto", proto) + } + + return sub, nil +} + +func (l *ALPNMux) UnregisterProto(proto string) { + val, ok := l.muxMap.Load(proto) + if !ok { + return + } + ml := val.(*muxedListener) + ml.closeOnce.Do(func() { + ml.connMutex.Lock() + defer ml.connMutex.Unlock() + ml.closed = true + close(ml.connCh) + }) + l.muxMap.Delete(proto) + if l.log != nil && l.log.IsDebug() { + l.log.Debug("unregistered", "proto", proto) + } +} + +func (l *ALPNMux) GetListener(proto string) net.Listener { + val, ok := l.muxMap.Load(proto) + if !ok || val == nil { + val, ok = l.muxMap.Load(DefaultProto) + if !ok || val == nil { + return nil + } + } + return val.(*muxedListener) +} + +func (l *ALPNMux) getConfigForClient(hello *tls.ClientHelloInfo) (*tls.Config, error) { + var ret *tls.Config + + if l.log != nil && l.log.IsTrace() { + l.log.Trace("getting config for client", "supported_protos", hello.SupportedProtos, "server_name", hello.ServerName) + } + + supportedProtos := hello.SupportedProtos + if len(hello.SupportedProtos) == 0 { + supportedProtos = append(supportedProtos, "") + } + for _, proto := range supportedProtos { + val, ok := l.muxMap.Load(proto) + if !ok { + continue + } + ret = val.(*muxedListener).tlsConf + } + if ret == nil { + val, ok := l.muxMap.Load(DefaultProto) + if ok && val != nil { + ret = val.(*muxedListener).tlsConf + } + } + if ret == nil { + return nil, errors.New("no tls configuration available for any client protos") + } + + // If the TLS config we found has its own lookup function, chain to it + if ret.GetConfigForClient != nil { + return ret.GetConfigForClient(hello) + } + + return ret, nil +} + +func (l *ALPNMux) accept() { + baseTLSConf := &tls.Config{ + GetConfigForClient: l.getConfigForClient, + } + for { + conn, err := l.baseLn.Accept() + if err != nil { + if strings.Contains(err.Error(), "use of closed network connection") { + l.cancel() + return + } + } + if conn == nil { + continue + } + if l.log != nil && l.log.IsTrace() { + l.log.Trace("got connection", "addr", conn.RemoteAddr()) + } + + // Do the rest in a goroutine so that a timeout in e.g. handshaking + // doesn't block acceptance of the next connection + go func() { + bufConn := &bufferedConn{ + Conn: conn, + buffer: bufio.NewReader(conn), + } + peeked, err := bufConn.buffer.Peek(3) + if err != nil { + if l.log != nil && l.log.IsDebug() { + l.log.Debug("error peeking connection", "addr", conn.RemoteAddr(), "error", err) + } + bufConn.Close() + return + } + switch { + // First byte should always be a handshake, second byte a 3, and + // third can be 3 or 1 depending on the implementation + case peeked[0] != 0x16 || peeked[1] != 0x03 || (peeked[2] != 0x03 && peeked[2] != 0x01): + if l.log != nil && l.log.IsTrace() { + l.log.Trace("connection is not tls", "addr", conn.RemoteAddr()) + } + val, ok := l.muxMap.Load(NoProto) + if !ok { + if l.log != nil && l.log.IsTrace() { + l.log.Trace("no non-tls registration found", "addr", conn.RemoteAddr()) + } + bufConn.Close() + return + } + if l.log != nil && l.log.IsTrace() { + l.log.Trace("found noproto handler", "addr", conn.RemoteAddr()) + } + ml := val.(*muxedListener) + ml.connMutex.RLock() + if !ml.closed { + ml.connCh <- bufConn + } + ml.connMutex.RUnlock() + + default: + if l.log != nil && l.log.IsTrace() { + l.log.Trace("connection is tls", "addr", conn.RemoteAddr()) + } + tlsConn := tls.Server(bufConn, baseTLSConf) + if l.log != nil && l.log.IsTrace() { + l.log.Trace("handshaking", "addr", conn.RemoteAddr()) + } + if err := tlsConn.Handshake(); err != nil { + closeErr := tlsConn.Close() + if l.log != nil && l.log.IsDebug() { + l.log.Debug("error handshaking connection", "addr", conn.RemoteAddr(), "error", err, "close_error", closeErr) + } + return + } + negProto := tlsConn.ConnectionState().NegotiatedProtocol + if l.log != nil && l.log.IsTrace() { + l.log.Trace("tls negotiated", "addr", conn.RemoteAddr(), "proto", negProto) + } + val, ok := l.muxMap.Load(negProto) + if !ok { + val, ok = l.muxMap.Load(DefaultProto) + if !ok { + if l.log != nil && l.log.IsTrace() { + l.log.Trace("no handler found", "addr", conn.RemoteAddr(), "proto", negProto) + } + tlsConn.Close() + return + } + } + if l.log != nil && l.log.IsTrace() { + l.log.Trace("found tls handler", "addr", conn.RemoteAddr(), "proto", negProto) + } + ml := val.(*muxedListener) + ml.connMutex.RLock() + if !ml.closed { + ml.connCh <- tlsConn + } + ml.connMutex.RUnlock() + } + }() + } +} + +func (m *muxedListener) Accept() (net.Conn, error) { + for { + select { + case <-m.ctx.Done(): + // Wouldn't it be so much better if this error was an exported + // const from Go... + m.closeFunc() + return nil, fmt.Errorf("accept proto %s: use of closed network connection", m.proto) + case conn, ok := <-m.connCh: + if !ok { + // Channel closed + return nil, fmt.Errorf("accept proto %s: use of closed network connection", m.proto) + } + if conn == nil { + return nil, fmt.Errorf("accept proto %s: nil connection received", m.proto) + } + return conn, nil + } + } +} + +func (m *muxedListener) Close() error { + m.closeFunc() + return nil +} + +func (m *muxedListener) Addr() net.Addr { + return m.addr +} diff --git a/internal/libs/alpnmux/mux_test.go b/internal/libs/alpnmux/mux_test.go new file mode 100644 index 000000000..d5835792a --- /dev/null +++ b/internal/libs/alpnmux/mux_test.go @@ -0,0 +1,193 @@ +package alpnmux + +import ( + "crypto/tls" + "fmt" + "log" + "net" + "strings" + "sync" + "testing" + "time" + + "github.com/hashicorp/go-hclog" + "go.uber.org/atomic" +) + +func TestListenCloseErrMsg(t *testing.T) { + listener := getListener(t) + listener.Close() + _, err := listener.Accept() + if !strings.Contains(err.Error(), "use of closed network connection") { + t.Fatal(err) + } +} + +func TestRegistrationErrors(t *testing.T) { + listener := getListener(t) + defer listener.Close() + mux := New(listener, nil) + p1config := getTestTLS(t, []string{"p1"}) + if _, err := mux.RegisterProto("p1", nil); err.Error() != "nil tls config given" { + t.Fatal(err) + } + l, err := mux.RegisterProto("p1", p1config) + if err != nil { + t.Fatal(err) + } + if _, err := mux.RegisterProto("p1", p1config); err.Error() != `proto "p1" already registered` { + t.Fatal(err) + } + l.Close() + // Unregister is not sync, so need to wait for it to actually be removed + var unregistered bool + for i := 0; i < 5; i++ { + _, ok := mux.muxMap.Load("p1") + if !ok { + unregistered = true + break + } + time.Sleep(100 * time.Millisecond) + } + if !unregistered { + t.Fatal("failed to unregister proto") + } + l, err = mux.RegisterProto("p1", p1config) + if err != nil { + t.Fatal(err) + } + l.Close() + l, err = mux.RegisterProto(NoProto, nil) + if err != nil { + t.Fatal(err) + } + l.Close() +} + +func TestListening(t *testing.T) { + listener := getListener(t) + + logger := hclog.Default() + logger.SetLevel(hclog.Trace) + //log.SetOutput(logger.StandardWriter(new(hclog.StandardLoggerOptions))) + mux := New(listener, logger) + defer mux.Close() + + emptyconns := atomic.NewUint32(0) + noneconns := atomic.NewUint32(0) + l1conns := atomic.NewUint32(0) + l2conns := atomic.NewUint32(0) + l3conns := atomic.NewUint32(0) + defconns := atomic.NewUint32(0) + clientCountTracker := atomic.NewUint32(0) + + baseconfig := getTestTLS(t, nil) + noneconfig := baseconfig.Clone() + p1config := baseconfig.Clone() + p1config.NextProtos = []string{"p1"} + p2p3config := getTestTLS(t, []string{"p2", "p3"}) + p3config := p2p3config.Clone() + p3config.NextProtos = []string{"p3"} + defconfig := baseconfig.Clone() + defconfig.GetConfigForClient = func(hello *tls.ClientHelloInfo) (*tls.Config, error) { + ret := baseconfig.Clone() + ret.NextProtos = []string{fmt.Sprintf("%d", clientCountTracker.Load())} + log.Printf("returning def config with next protos = %v\n", ret.NextProtos) + clientCountTracker.Inc() + return ret, nil + } + + lempty, err := mux.RegisterProto("", noneconfig) + if err != nil { + t.Fatal(err) + } + l1, err := mux.RegisterProto("p1", p1config) + if err != nil { + t.Fatal(err) + } + l2, err := mux.RegisterProto("p2", p2p3config) + if err != nil { + t.Fatal(err) + } + l3, err := mux.RegisterProto("p3", p2p3config) + if err != nil { + t.Fatal(err) + } + lnone, err := mux.RegisterProto(NoProto, nil) + if err != nil { + t.Fatal(err) + } + ldef, err := mux.RegisterProto(DefaultProto, defconfig) + if err != nil { + t.Fatal(err) + } + + addr := listener.Addr().String() + wg := new(sync.WaitGroup) + wg.Add(6) + connWatchFunc := func(l net.Listener, connCounter *atomic.Uint32, tlsConf *tls.Config, numConns int) { + defer wg.Done() + tlsToUse := tlsConf + go func() { + for i := 0; i < numConns; i++ { + var err error + var conn net.Conn + switch tlsToUse { + case nil: + conn, err = net.Dial("tcp4", addr) + if err != nil { + t.Fatal(err) + } + // We need to send some data here because we won't have any + // from just the TLS handshake + log.Println("defconn") + n, err := conn.Write([]byte("GET ")) + if err != nil { + t.Fatal(err) + } + if n != 4 { + t.Fatal(n) + } + log.Println("defconn done") + + default: + if connCounter == defconns { + tlsToUse = baseconfig.Clone() + log.Println("FOUND CURR") + tlsToUse.NextProtos = []string{fmt.Sprintf("%d", i)} + } + log.Println(fmt.Sprintf("dialing on %d, counter = %d, protos = %v", numConns, i, tlsToUse.NextProtos)) + conn, err = tls.Dial("tcp4", addr, tlsToUse) + if err != nil { + t.Fatal(err) + } + log.Println(fmt.Sprintf("dialing done on %d, counter = %d, protos = %v", numConns, i, tlsToUse.NextProtos)) + } + conn.Close() + } + }() + for i := 0; i < numConns; i++ { + log.Println(fmt.Sprintf("accepting on %d, counter = %d", numConns, connCounter.Load())) + conn, err := l.Accept() + if err == nil && conn != nil { + conn.Close() + } else { + t.Fatal(err) + } + log.Println(fmt.Sprintf("done accepting on %d, counter = %d", numConns, connCounter.Load())) + connCounter.Inc() + } + return + } + go connWatchFunc(lempty, emptyconns, noneconfig, 4) + go connWatchFunc(l1, l1conns, p1config, 5) + go connWatchFunc(l2, l2conns, p2p3config, 6) + go connWatchFunc(l3, l3conns, p3config, 7) + go connWatchFunc(lnone, noneconns, nil, 8) + go connWatchFunc(ldef, defconns, defconfig, 9) + wg.Wait() + + if emptyconns.Load() != 4 || l1conns.Load() != 5 || l2conns.Load() != 6 || l3conns.Load() != 7 || noneconns.Load() != 8 || defconns.Load() != 9 { + t.Fatal("wrong number of conns") + } +} diff --git a/internal/libs/alpnmux/testing.go b/internal/libs/alpnmux/testing.go new file mode 100644 index 000000000..a7fde9109 --- /dev/null +++ b/internal/libs/alpnmux/testing.go @@ -0,0 +1,119 @@ +package alpnmux + +import ( + "crypto/ed25519" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + mathrand "math/rand" + "net" + "testing" + "time" +) + +func getListener(t *testing.T) net.Listener { + addr, err := net.ResolveTCPAddr("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + listener, err := net.ListenTCP("tcp", addr) + if err != nil { + t.Fatal(err) + } + return listener +} + +func getTestTLS(t *testing.T, protos []string) *tls.Config { + certIPs := []net.IP{ + net.IPv6loopback, + net.ParseIP("127.0.0.1"), + } + + _, caKey, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + caCertTemplate := &x509.Certificate{ + Subject: pkix.Name{ + CommonName: "localhost", + }, + DNSNames: []string{"localhost"}, + IPAddresses: certIPs, + KeyUsage: x509.KeyUsage(x509.KeyUsageCertSign | x509.KeyUsageCRLSign), + SerialNumber: big.NewInt(mathrand.Int63()), + NotBefore: time.Now().Add(-30 * time.Second), + NotAfter: time.Now().Add(262980 * time.Hour), + BasicConstraintsValid: true, + IsCA: true, + } + caBytes, err := x509.CreateCertificate(rand.Reader, caCertTemplate, caCertTemplate, caKey.Public(), caKey) + if err != nil { + t.Fatal(err) + } + caCert, err := x509.ParseCertificate(caBytes) + if err != nil { + t.Fatal(err) + } + rootCAs := x509.NewCertPool() + rootCAs.AddCert(caCert) + + // + // Certs generation + // + _, key, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + certTemplate := &x509.Certificate{ + Subject: pkix.Name{ + CommonName: "localhost", + }, + DNSNames: []string{"localhost"}, + IPAddresses: certIPs, + ExtKeyUsage: []x509.ExtKeyUsage{ + x509.ExtKeyUsageServerAuth, + x509.ExtKeyUsageClientAuth, + }, + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageKeyAgreement, + SerialNumber: big.NewInt(mathrand.Int63()), + NotBefore: time.Now().Add(-30 * time.Second), + NotAfter: time.Now().Add(262980 * time.Hour), + } + certBytes, err := x509.CreateCertificate(rand.Reader, certTemplate, caCert, key.Public(), caKey) + if err != nil { + t.Fatal(err) + } + certPEMBlock := &pem.Block{ + Type: "CERTIFICATE", + Bytes: certBytes, + } + certPEM := pem.EncodeToMemory(certPEMBlock) + marshaledKey, err := x509.MarshalPKCS8PrivateKey(key) + if err != nil { + t.Fatal(err) + } + keyPEMBlock := &pem.Block{ + Type: "PRIVATE KEY", + Bytes: marshaledKey, + } + keyPEM := pem.EncodeToMemory(keyPEMBlock) + + tlsCert, err := tls.X509KeyPair(certPEM, keyPEM) + if err != nil { + t.Fatal(err) + } + tlsConfig := &tls.Config{ + Certificates: []tls.Certificate{tlsCert}, + RootCAs: rootCAs, + ClientCAs: rootCAs, + ClientAuth: tls.RequestClientCert, + NextProtos: protos, + MinVersion: tls.VersionTLS12, + MaxVersion: tls.VersionTLS13, + } + + return tlsConfig +} diff --git a/internal/servers/controller/listeners.go b/internal/servers/controller/listeners.go index cd4b56a25..ac63dc6cb 100644 --- a/internal/servers/controller/listeners.go +++ b/internal/servers/controller/listeners.go @@ -13,8 +13,8 @@ import ( "github.com/hashicorp/boundary/internal/cmd/base" pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" + "github.com/hashicorp/boundary/internal/libs/alpnmux" "github.com/hashicorp/boundary/internal/servers/controller/handlers/workers" - "github.com/hashicorp/go-alpnmux" "github.com/hashicorp/go-multierror" "google.golang.org/grpc" ) diff --git a/internal/servers/worker/listeners.go b/internal/servers/worker/listeners.go index 63dcdcf98..9ea9f9613 100644 --- a/internal/servers/worker/listeners.go +++ b/internal/servers/worker/listeners.go @@ -10,7 +10,7 @@ import ( "sync" "time" - "github.com/hashicorp/go-alpnmux" + "github.com/hashicorp/boundary/internal/libs/alpnmux" "github.com/hashicorp/go-multierror" )