From a403b3904cdcc5f229fb841884880abbe61486cd Mon Sep 17 00:00:00 2001 From: Irena Rindos Date: Mon, 9 Jan 2023 12:35:13 -0500 Subject: [PATCH] initial commit (#2735) --- internal/daemon/worker/handler.go | 2 +- internal/daemon/worker/proxy/proxy.go | 4 ++- internal/daemon/worker/proxy/proxydialer.go | 34 ++++++++++++------ internal/daemon/worker/proxy/tcp/tcp.go | 4 +-- internal/daemon/worker/proxy/tcp/tcp_test.go | 4 +-- internal/daemon/worker/testing.go | 36 ++++++++++++++++++++ internal/daemon/worker/testing_test.go | 22 ++++++++++++ 7 files changed, 88 insertions(+), 18 deletions(-) diff --git a/internal/daemon/worker/handler.go b/internal/daemon/worker/handler.go index aceb0edac..1da3342cc 100644 --- a/internal/daemon/worker/handler.go +++ b/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) diff --git a/internal/daemon/worker/proxy/proxy.go b/internal/daemon/worker/proxy/proxy.go index ea8fb0e65..ef58f8d46 100644 --- a/internal/daemon/worker/proxy/proxy.go +++ b/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 { diff --git a/internal/daemon/worker/proxy/proxydialer.go b/internal/daemon/worker/proxy/proxydialer.go index 1fe2cb5a2..fb5899903 100644 --- a/internal/daemon/worker/proxy/proxydialer.go +++ b/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") diff --git a/internal/daemon/worker/proxy/tcp/tcp.go b/internal/daemon/worker/proxy/tcp/tcp.go index e31f839fb..96117c10f 100644 --- a/internal/daemon/worker/proxy/tcp/tcp.go +++ b/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 { diff --git a/internal/daemon/worker/proxy/tcp/tcp_test.go b/internal/daemon/worker/proxy/tcp/tcp_test.go index 2b2a83033..b25ef634f 100644 --- a/internal/daemon/worker/proxy/tcp/tcp_test.go +++ b/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 { diff --git a/internal/daemon/worker/testing.go b/internal/daemon/worker/testing.go index c90f7d66b..f2b97cf0e 100644 --- a/internal/daemon/worker/testing.go +++ b/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() +} diff --git a/internal/daemon/worker/testing_test.go b/internal/daemon/worker/testing_test.go index 4f32544ba..1802769a6 100644 --- a/internal/daemon/worker/testing_test.go +++ b/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{