initial commit (#2735)

internal/daemon/worker/handler.go

@@ -265,7 +265,7 @@ func (w *Worker) handleProxy(listenerCfg *listenerutil.ListenerConfig, sessionMa
 			}
 		}
 
-		pDialer, err := proxyHandlers.GetEndpointDialer(ctx, endpointUrl, ci)
+		pDialer, err := proxyHandlers.GetEndpointDialer(ctx, endpointUrl.Host, workerId, ci, w.downstreamRoutes)
 		if err != nil {
 			conn.Close(proxyHandlers.WebsocketStatusProtocolSetupError, "unable to get endpoint dialer")
 			event.WriteError(ctx, op, err)

internal/daemon/worker/proxy/proxy.go

@@ -26,7 +26,9 @@ type ProxyConnFn func()
 
 // Handler is the type that all proxies need to implement to be called by the worker
 // when a new client connection is created.  If there is an error ProxyConnFn must
-// be nil. If there is no error ProxyConnFn must be set.
+// be nil. If there is no error ProxyConnFn must be set.  When Handler has
+// returned, it is expected that the initial connection to the endpoint has been
+// established.
 type Handler func(context.Context, net.Conn, *ProxyDialer, string, *anypb.Any) (ProxyConnFn, error)
 
 func RegisterHandler(protocol string, handler Handler) error {

internal/daemon/worker/proxy/proxydialer.go

@@ -3,11 +3,10 @@ package proxy
 import (
 	"context"
 	"net"
-	"net/url"
 	"sync/atomic"
 
 	"github.com/hashicorp/boundary/internal/errors"
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
+	"google.golang.org/protobuf/proto"
 )
 
 // GetEndpointDialer returns a ProxyDialer which, when Dial() is called
@@ -16,13 +15,13 @@ var GetEndpointDialer = directDialer
 
 // directDialer returns a ProxyDialer which tcp dials directly to the provided
 // endpoint.
-func directDialer(ctx context.Context, endpoint *url.URL, _ *pbs.AuthorizeConnectionResponse) (*ProxyDialer, error) {
+func directDialer(ctx context.Context, endpoint string, _ string, _ proto.Message, _ interface{}) (*ProxyDialer, error) {
 	const op = "proxy.directDialer"
-	if endpoint == nil {
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "endpoint is nil")
+	if len(endpoint) == 0 {
+		return nil, errors.New(ctx, errors.InvalidParameter, op, "endpoint is empty")
 	}
 	d, err := NewProxyDialer(ctx, func() (net.Conn, error) {
-		remoteConn, err := net.Dial("tcp", endpoint.Host)
+		remoteConn, err := net.Dial("tcp", endpoint)
 		if err != nil {
 			return nil, errors.Wrap(ctx, err, op)
 		}
@@ -36,19 +35,19 @@ func directDialer(ctx context.Context, endpoint *url.URL, _ *pbs.AuthorizeConnec
 
 // proxyAddr is a net.Addr with IP and port information for the endpoint being
 // proxied to.
-// TODO: Support returning the IP and port information that gets sent back up through grpc.
 type proxyAddr struct {
-	wrapped *net.TCPAddr
+	ip   string
+	port uint32
 }
 
 // Ip returns the ip address of the dialed endpoint
 func (p *proxyAddr) Ip() string {
-	return p.wrapped.IP.String()
+	return p.ip
 }
 
 // Port returns the tcp port of the endpoint connected to
 func (p *proxyAddr) Port() uint32 {
-	return uint32(p.wrapped.Port)
+	return p.port
 }
 
 // ProxyDialer dials downstream to eventually get to the target host.
@@ -75,6 +74,15 @@ func (d *ProxyDialer) LastConnectionAddr() *proxyAddr {
 	return d.latestAddr.Load()
 }
 
+// portAndIpGetter allows a dialing function to return a connection that can
+// provide it's ip address and port through the GetIp and GetPort methods
+// instead of providing directly a *net.TCPConn.  This might be helpful if the
+// net.Conn embeds a protobuf message with Ip and Port fields for example.
+type portAndIpGetter interface {
+	GetIp() string
+	GetPort() uint32
+}
+
 // Dial uses the provided dial function to get a net.Conn and record its
 // net.Addr information.  The returned net.Addr should contain the information
 // for the endpoint that is being proxied to.
@@ -86,7 +94,11 @@ func (d *ProxyDialer) Dial(ctx context.Context) (net.Conn, error) {
 	}
 	switch v := c.RemoteAddr().(type) {
 	case *net.TCPAddr:
-		d.latestAddr.Store(&proxyAddr{v})
+		ip := v.IP.String()
+		port := uint32(v.Port)
+		d.latestAddr.Store(&proxyAddr{ip: ip, port: port})
+	case portAndIpGetter:
+		d.latestAddr.Store(&proxyAddr{ip: v.GetIp(), port: v.GetPort()})
 	default:
 		c.Close()
 		return nil, errors.New(ctx, errors.Internal, op, "connection type unexpected")

internal/daemon/worker/proxy/tcp/tcp.go

@@ -26,7 +26,7 @@ func init() {
 // connection.
 //
 // All options are ignored.
-func handleProxy(ctx context.Context, conn net.Conn, out *proxy.ProxyDialer, connId string, pi *anypb.Any) (proxy.ProxyConnFn, error) {
+func handleProxy(ctx context.Context, conn net.Conn, out *proxy.ProxyDialer, connId string, _ *anypb.Any) (proxy.ProxyConnFn, error) {
 	const op = "tcp.HandleProxy"
 	switch {
 	case conn == nil:
@@ -35,8 +35,6 @@ func handleProxy(ctx context.Context, conn net.Conn, out *proxy.ProxyDialer, con
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "proxy dialer is nil")
 	case len(connId) == 0:
 		return nil, errors.New(ctx, errors.InvalidParameter, op, "connection id is empty")
-	case pi != nil:
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "protocol context Any proto is not nil")
 	}
 	remoteConn, err := out.Dial(ctx)
 	if err != nil {

internal/daemon/worker/proxy/tcp/tcp_test.go

@@ -72,7 +72,7 @@ func TestHandleProxy_Errors(t *testing.T) {
 			wantError:   true,
 		},
 		{
-			name:   "specified protocol context",
+			name:   "specified protocol context passed on without validation",
 			conn:   c,
 			dialer: dialer,
 			connId: "someconnectionid",
@@ -80,7 +80,7 @@ func TestHandleProxy_Errors(t *testing.T) {
 				TypeUrl: "some.type.information",
 				Value:   []byte("this is just for this test"),
 			},
-			wantError: true,
+			wantError: false,
 		},
 	}
 	for _, tc := range cases {

internal/daemon/worker/testing.go

@@ -457,3 +457,39 @@ func NewTestMultihopWorkers(t testing.TB, logger hclog.Logger, controllerContext
 
 	return kmsWorker, pkiWorker, childPkiWorker
 }
+
+// NewAuthorizedPkiTestWorker creates a new test worker with the provided upstreams
+// and creates it in the provided repo as an authroized worker. It returns
+// The TestWorker and it's boundary id.
+func NewAuthorizedPkiTestWorker(t *testing.T, repo *server.Repository, name string, upstreams []string) (*TestWorker, string) {
+	t.Helper()
+	logger := hclog.New(&hclog.LoggerOptions{
+		Level: hclog.Trace,
+	})
+	wcfg, err := config.DevWorker()
+	require.NoError(t, err)
+	wcfg.Worker.Name = ""
+	wcfg.Worker.InitialUpstreams = upstreams
+	w := NewTestWorker(t, &TestWorkerOpts{
+		InitialUpstreams: upstreams,
+		Logger:           logger.Named(name),
+		Config:           wcfg,
+	})
+	t.Cleanup(w.Shutdown)
+
+	// Perform initial authentication of worker to controller
+	reqBytes, err := base58.FastBase58Decoding(w.Worker().WorkerAuthRegistrationRequest)
+	require.NoError(t, err)
+
+	// Decode the proto into the request and create the worker
+	pkiWorkerReq := new(types.FetchNodeCredentialsRequest)
+	require.NoError(t, proto.Unmarshal(reqBytes, pkiWorkerReq))
+	wr, err := repo.CreateWorker(context.Background(), &server.Worker{
+		Worker: &store.Worker{
+			Name:    name,
+			ScopeId: scope.Global.String(),
+		},
+	}, server.WithFetchNodeCredentialsRequest(pkiWorkerReq))
+	require.NoError(t, err)
+	return w, wr.GetPublicId()
+}

internal/daemon/worker/testing_test.go

@@ -126,6 +126,28 @@ func TestTestWorker_WorkerAuthStorageKms(t *testing.T) {
 	}
 }
 
+func TestNewAuthorizedPkiTestWorker(t *testing.T) {
+	logger := hclog.New(&hclog.LoggerOptions{
+		Level: hclog.Trace,
+	})
+	conf, err := config.DevController()
+	require.NoError(t, err)
+	c := controller.NewTestController(t, &controller.TestControllerOpts{
+		Config: conf,
+		Logger: logger.Named("controller"),
+	})
+	t.Cleanup(c.Shutdown)
+	tw, id := NewAuthorizedPkiTestWorker(t, c.ServersRepo(), "test", c.ClusterAddrs())
+	assert.NotNil(t, tw)
+	assert.NotEmpty(t, id)
+
+	w, err := c.ServersRepo().LookupWorker(context.Background(), id)
+	assert.NoError(t, err)
+	assert.NotNil(t, w)
+	assert.Equal(t, "pki", w.GetType())
+	assert.Equal(t, "test", w.GetName())
+}
+
 func TestNewTestMultihopWorkers(t *testing.T) {
 	ctx := context.Background()
 	logger := hclog.New(&hclog.LoggerOptions{