functional connector POC that works with 'connString' credential

go.mod

@@ -51,6 +51,8 @@ require (
 
 replace github.com/denisenkom/go-mssqldb => ./third_party/go-mssqldb
 
+replace go.mongodb.org/mongo-driver => github.com/doodlesbykumbi/mongo-go-driver v1.4.0-beta2.0.20200923140903-dd0d9ce83942
+
 // 2/19/2019: cert on honnef.co -- one of grpc's dependencies -- expired.
 // This is our fix:
 replace honnef.co/go/tools => github.com/dominikh/go-tools v0.0.1-2019.2.3

go.sum

@@ -59,6 +59,8 @@ github.com/docker/go-units v0.4.0 h1:3uh0PgVws3nIA0Q+MwDC8yjEPf9zjRfZZWXZYDct3Tw
 github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/dominikh/go-tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
+github.com/doodlesbykumbi/mongo-go-driver v1.4.0-beta2.0.20200923140903-dd0d9ce83942 h1:bnAFmE/c8GuAHteBX+QO2yEaa5XbK1dYNo1p/Bf62Ew=
+github.com/doodlesbykumbi/mongo-go-driver v1.4.0-beta2.0.20200923140903-dd0d9ce83942/go.mod h1:llVBH2pkj9HywK0Dtdt6lDikOjFLbceHVu/Rc0iMKLs=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga6PJ7M=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=

internal/plugin/connectors/tcp/mongodb/connector.go

@@ -2,9 +2,12 @@ package mongodb
 
 import (
 	"context"
-	"fmt"
 	"net"
 
+	"go.mongodb.org/mongo-driver/mongo/address"
+	"go.mongodb.org/mongo-driver/x/mongo/driver/connstring"
+	"go.mongodb.org/mongo-driver/x/mongo/driver/topology"
+
 	"github.com/cyberark/secretless-broker/pkg/secretless/log"
 	"github.com/cyberark/secretless-broker/pkg/secretless/plugin/connector"
 )
@@ -21,14 +24,133 @@ func (connector *SingleUseConnector) Connect(
 	clientConn net.Conn,
 	credentialValuesByID connector.CredentialValuesByID,
 ) (net.Conn, error) {
-	connDetails, _ := NewConnectionDetails(credentialValuesByID)
+	//connDetails, _ := NewConnectionDetails(credentialValuesByID)
+
+	//host := net.JoinHostPort(connDetails.Host, fmt.Sprintf("%d", connDetails.Port))
+	//backendConn, err := dialer.DialContext(context.Background(), "tcp", host)
+	//if err != nil {
+	//	return nil, err
+	////}
+	//authenticator, err := auth.CreateAuthenticator("SCRAM-SHA-1", &auth.Cred{
+	//	Source:      "admin",
+	//	Username:    "user0",
+	//	Password:    "pass0",
+	//})
+	//if err != nil {
+	//	return nil, err
+	//}
+
+	var firstResponse []byte
+
+	connString, err := connstring.ParseAndValidate(
+		string(credentialValuesByID["connString"]),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	host, port, err := net.SplitHostPort(connString.Hosts[0])
+	if err != nil {
+		return nil, err
+	}
 
-	host := net.JoinHostPort(connDetails.Host, fmt.Sprintf("%d", connDetails.Port))
-	dialer := newProxyDialer()
-	backendConn, err := dialer.DialContext(context.Background(), "tcp", host)
+	conn, err := topology.NewConnection(address.Address(net.JoinHostPort(host, port)),
+		topology.WithDialer(func(topology.Dialer) topology.Dialer {
+			return newProxyDialer(func(bytes []byte) {
+				firstResponse = bytes
+			})
+		}),
+		topology.WithConnStringForConn(func(connstring.ConnString) connstring.ConnString {
+			return connString
+		}),
+		//topology.WithHandshaker(func(h topology.Handshaker) topology.Handshaker {
+		//	options := &auth.HandshakeOptions{
+		//		AppName:               "meow",
+		//		Authenticator:         authenticator,
+		//		PerformAuthentication: func(server description.Server) bool {
+		//			return true
+		//		},
+		//	}
+		//
+		//	return auth.Handshaker(h, options)
+		//}),
+	)
 	if err != nil {
 		return nil, err
 	}
 
-	return backendConn, nil
+	conn.Connect(context.Background())
+	err = conn.Wait()
+	if err != nil {
+		return nil, err
+	}
+
+	stolenConn := conn.StealConn()
+
+	if _stolenConn, ok := stolenConn.(*proxyConn); ok {
+		stolenConn = _stolenConn.NetConn()
+	}
+
+	bytes := make([]byte, 2048)
+	readBytes, err := clientConn.Read(bytes)
+	if err != nil {
+		return nil, err
+	}
+	//fmt.Println("Read from the client", readBytes)
+	clientFirstMessage, err := parseSentMessage(bytes[:readBytes])
+	if err != nil {
+		return nil, err
+	}
+
+	// Do something with the client first message
+	if clientFirstMessage == clientFirstMessage {}
+
+	err = backendTemp(
+		clientConn,
+		connString.Hosts[0],
+		bytes,
+		readBytes,
+	)
+	//if err != nil {
+	//	fmt.Println("Warning:", "Tried to use backend temp connection and got:", err)
+	//	fmt.Println("Will attempt to use first response");
+	//	_, err = clientConn.Write(firstResponse)
+	//}
+	if err != nil {
+		return nil, err
+	}
+
+	return stolenConn, nil
 }
+
+func backendTemp(
+	clientConn net.Conn,
+	host string,
+	bytes []byte,
+	readBytes int,
+) error  {
+	backendTmpConn, err := net.Dial("tcp", host)
+	if err != nil {
+		return err
+	}
+
+	_, err = backendTmpConn.Write(bytes[:readBytes])
+	if err != nil {
+		return err
+	}
+
+	readBytes, err = backendTmpConn.Read(bytes)
+	if err != nil {
+		return err
+	}
+
+	// writtenBytes, err := clientConn.Write(bytes[:readBytes])
+	_, err = clientConn.Write(bytes[:readBytes])
+	if err != nil {
+		return err
+	}
+
+	// fmt.Println("Wrote to the client", readBytes, writtenBytes)
+
+	return nil
+}

internal/plugin/connectors/tcp/mongodb/proxy_dialer.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
 )
 
 // ProxyMessage represents a sent/received pair of parsed wire messages.
@@ -33,13 +34,18 @@ type proxyDialer struct {
 	// differ. This can happen if a connection is dialed to a host name, in which case the reported remote address will
 	// be the resolved IP address.
 	addressTranslations sync.Map
+
+	onFirstResponse func([]byte)
 }
 
 var _ options.ContextDialer = (*proxyDialer)(nil)
 
-func newProxyDialer() *proxyDialer {
+func newProxyDialer(onFirstResponse func([]byte)) *proxyDialer {
 	return &proxyDialer{
-		Dialer: &net.Dialer{Timeout: 30 * time.Second},
+		Dialer: &net.Dialer{
+			Timeout:       2 * time.Second,
+		},
+		onFirstResponse: onFirstResponse,
 	}
 }
 
@@ -66,8 +72,10 @@ func (p *proxyDialer) DialContext(ctx context.Context, network, address string)
 	}
 
 	proxy := &proxyConn{
-		Conn:   netConn,
-		dialer: p,
+		processedFirstResponse: false,
+		onFirstResponse:        p.onFirstResponse,
+		Conn:                   netConn,
+		dialer:                 p,
 	}
 	return proxy, nil
 }
@@ -147,6 +155,8 @@ func (p *proxyDialer) Messages() []*ProxyMessage {
 // storing wire messages are wrapped to add context, while errors returned from the underlying network connection are
 // forwarded without wrapping.
 type proxyConn struct {
+	processedFirstResponse bool
+	onFirstResponse func ([]byte)
 	net.Conn
 	dialer *proxyDialer
 }
@@ -165,24 +175,31 @@ func (pc *proxyConn) Write(wm []byte) (n int, err error) {
 
 // Read reads the message from the server into the given buffer and stores the read message in the proxyDialer
 // associated with this connection.
-func (pc *proxyConn) Read(wm []byte) (int, error) {
-	n, err := pc.Conn.Read(wm)
+func (pc *proxyConn) Read(buffer []byte) (int, error) {
+	n, err := pc.Conn.Read(buffer)
 	if err != nil {
 		return n, err
 	}
+
 	//
 	//// The driver reads wire messages in two phases: a four-byte read to get the length of the incoming wire message
 	//// and a (length-4) byte read to get the message itself. There's nothing to be stored during the initial four-byte
 	//// read because we can calculate the length from the rest of the message.
-	//if len(buffer) == 4 {
-	//	return 4, nil
-	//}
+	if len(buffer) == 4 {
+		return 4, nil
+	}
+
 	//
 	//// The buffer contains the entire wire message except for the length bytes. Re-create the full message by appending
 	//// buffer to the end of a four-byte slice and using UpdateLength to set the length bytes.
-	//idx, wm := bsoncore.ReserveLength(nil)
-	//wm = append(wm, buffer...)
-	//wm = bsoncore.UpdateLength(wm, idx, int32(len(wm[idx:])))
+	idx, wm := bsoncore.ReserveLength(nil)
+	wm = append(wm, buffer...)
+	wm = bsoncore.UpdateLength(wm, idx, int32(len(wm[idx:])))
+
+	if !pc.processedFirstResponse {
+		pc.onFirstResponse(copyBytes(wm))
+		pc.processedFirstResponse = true
+	}
 
 	if err := pc.dialer.storeReceivedMessage(wm, pc.RemoteAddr().String()); err != nil {
 		wrapped := fmt.Errorf("error storing received message: %v", err)
@@ -192,3 +209,7 @@ func (pc *proxyConn) Read(wm []byte) (int, error) {
 
 	return n, nil
 }
+
+func (pc *proxyConn) NetConn() net.Conn {
+	return pc.Conn
+}

internal/plugin/connectors/tcp/mongodb/sent_message.go

@@ -1,4 +1,4 @@
-/package mongodb
+package mongodb
 
 import (
 	"errors"

test/connector/tcp/mongodb/client-cli

@@ -1,3 +1,4 @@
 #!/usr/bin/env bash
 
-mongo "mongodb://user0:pass0@localhost:27018/?authSource=admin&ssl=false"
+#mongo --authenticationMechanism SCRAM-SHA-1 "mongodb://user0:pass0@localhost:27018/?authSource=admin&ssl=false"
+mongo "mongodb://localhost:27018/meow?ssl=false" "$@"

test/connector/tcp/mongodb/local-run

@@ -1,3 +1,5 @@
 #!/usr/bin/env bash
 
-go run github.com/cyberark/secretless-broker/cmd/secretless-broker | tee log.txt
+set -x
+
+go run github.com/cyberark/secretless-broker/cmd/secretless-broker "$@" | tee log.txt

test/connector/tcp/mongodb/secretless.yml

@@ -7,7 +7,4 @@ services:
     listenOn: tcp://127.0.0.1:27018
     # Which credentials secretless should get for incoming connections, and their source
     credentials:
-      password: pass0
-      user: user0
-      host: 127.0.0.1
-      port: 27017
+      connString: "mongodb://user0:pass0@localhost:27017/app?authSource=admin&ssl=false"