diff --git a/api/go.mod b/api/go.mod index e2d8d5146..1ea7641bb 100644 --- a/api/go.mod +++ b/api/go.mod @@ -10,14 +10,19 @@ require ( github.com/hashicorp/go-kms-wrapping/v2 v2.0.14 github.com/hashicorp/go-retryablehttp v0.7.4 github.com/hashicorp/go-rootcerts v1.0.2 + github.com/hashicorp/go-secure-stdlib/base62 v0.1.2 github.com/hashicorp/go-secure-stdlib/parseutil v0.1.8 + github.com/hashicorp/go-secure-stdlib/temperror v0.1.1 github.com/hashicorp/go-uuid v1.0.3 + github.com/mitchellh/copystructure v1.2.0 github.com/mitchellh/mapstructure v1.5.0 github.com/mr-tron/base58 v1.2.0 github.com/stretchr/testify v1.8.4 + go.uber.org/atomic v1.11.0 golang.org/x/time v0.3.0 google.golang.org/grpc v1.59.0 google.golang.org/protobuf v1.31.0 + nhooyr.io/websocket v1.8.10 ) require ( @@ -29,11 +34,11 @@ require ( github.com/hashicorp/go-multierror v1.1.1 // indirect github.com/hashicorp/go-secure-stdlib/strutil v0.1.2 // indirect github.com/hashicorp/go-sockaddr v1.0.5 // indirect - github.com/mitchellh/copystructure v1.2.0 // indirect github.com/mitchellh/go-homedir v1.1.0 // indirect github.com/mitchellh/pointerstructure v1.2.1 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/rogpeppe/go-internal v1.8.1 // indirect github.com/ryanuber/go-glob v1.0.0 // indirect golang.org/x/crypto v0.14.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20231030173426-d783a09b4405 // indirect diff --git a/api/go.sum b/api/go.sum index 9bd3f1856..de5b03072 100644 --- a/api/go.sum +++ b/api/go.sum @@ -67,6 +67,8 @@ github.com/hashicorp/go-secure-stdlib/reloadutil v0.1.1/go.mod h1:Ch/bf00Qnx77MZ github.com/hashicorp/go-secure-stdlib/strutil v0.1.1/go.mod h1:gKOamz3EwoIoJq7mlMIRBpVTAUn8qPCrEclOKKWhD3U= github.com/hashicorp/go-secure-stdlib/strutil v0.1.2 h1:kes8mmyCpxJsI7FTwtzRqEy9CdjCtrXrXGuOpxEA7Ts= github.com/hashicorp/go-secure-stdlib/strutil v0.1.2/go.mod h1:Gou2R9+il93BqX25LAKCLuM+y9U2T4hlwvT1yprcna4= +github.com/hashicorp/go-secure-stdlib/temperror v0.1.1 h1:WkyqHb9NZWMEbB2rypsadIlbFOK4UOQPrZK5lPRDDrc= +github.com/hashicorp/go-secure-stdlib/temperror v0.1.1/go.mod h1:BkcKjSGVPPVa9VBEbyYBhkaSEp8dqq97m8TbdcW+Y3U= github.com/hashicorp/go-secure-stdlib/tlsutil v0.1.3 h1:xbrxd0U9XQW8qL1BAz2XrAjAF/P2vcqUTAues9c24B8= github.com/hashicorp/go-secure-stdlib/tlsutil v0.1.3/go.mod h1:LWq2Sy8UoKKuK4lFuCNWSjJj57MhNNf2zzBWMtkAIX4= github.com/hashicorp/go-sockaddr v1.0.5 h1:dvk7TIXCZpmfOlM+9mlcrWmWjw/wlKT+VDq2wMvfPJU= @@ -115,12 +117,14 @@ github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o= github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc= github.com/oklog/run v1.1.0 h1:GEenZ1cK0+q0+wsJew9qUg/DyD8k3JzYsZAi5gYi2mA= github.com/oklog/run v1.1.0/go.mod h1:sVPdnTZT1zYwAJeCMu2Th4T21pA3FPOQRfWjQlk7DVU= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/posener/complete v1.2.3 h1:NP0eAhjcjImqslEwo/1hq7gpajME0fTLTezBKDqfXqo= github.com/posener/complete v1.2.3/go.mod h1:WZIdtGGp+qx0sLrYKtIRAruyNpv6hFCicSgv7Sy7s/s= -github.com/rogpeppe/go-internal v1.6.2 h1:aIihoIOHCiLZHxyoNQ+ABL4NKhFTgKLBdMLyEAh98m0= github.com/rogpeppe/go-internal v1.6.2/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc= +github.com/rogpeppe/go-internal v1.8.1 h1:geMPLpDpQOgVyCg5z5GoRwLHepNdb71NXb67XFkP+Eg= +github.com/rogpeppe/go-internal v1.8.1/go.mod h1:JeRgkft04UBgHMgCIwADu4Pn6Mtm5d4nPKWu0nJ5d+o= github.com/ryanuber/go-glob v1.0.0 h1:iQh3xXAumdQ+4Ufa5b25cRpC5TYKlno6hsv6Cb3pkBk= github.com/ryanuber/go-glob v1.0.0/go.mod h1:807d1WSdnB0XRJzKNil9Om6lcp/3a0v4qIHxIXzX/Yc= github.com/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5gKV8= @@ -134,6 +138,8 @@ github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/ github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.0.0/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A= go.uber.org/goleak v1.1.12 h1:gZAh5/EyT/HQwlpkCy6wTpqfH9H8Lz8zbm3dZh+OyzA= go.uber.org/goleak v1.1.12/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ= @@ -198,3 +204,5 @@ gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= mvdan.cc/gofumpt v0.1.1/go.mod h1:yXG1r1WqZVKWbVRtBWKWX9+CxGYfA51nSomhM0woR48= +nhooyr.io/websocket v1.8.10 h1:mv4p+MnGrLDcPlBoWsvPP7XCzTYMXP9F9eIGoKbgx7Q= +nhooyr.io/websocket v1.8.10/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c= diff --git a/api/proxy/option.go b/api/proxy/option.go new file mode 100644 index 000000000..c22f1961e --- /dev/null +++ b/api/proxy/option.go @@ -0,0 +1,120 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package proxy + +import ( + "errors" + "net" + "net/netip" + + "github.com/hashicorp/boundary/api/targets" +) + +// getOpts iterates the inbound Options and returns a struct and any errors +func getOpts(opt ...Option) (*Options, error) { + opts := getDefaultOptions() + for _, o := range opt { + if o == nil { + continue + } + if err := o(opts); err != nil { + return nil, err + } + + } + return opts, nil +} + +// Options contains various options. The values are exported since the options +// are parsed in various other packages. +type Options struct { + WithListener net.Listener + WithListenAddrPort netip.AddrPort + WithConnectionsLeftCh chan int32 + WithWorkerHost string + WithSessionAuthorizationData *targets.SessionAuthorizationData + WithSkipSessionTeardown bool +} + +// Option is a function that takes in an options struct and sets values or +// returns an error +type Option func(*Options) error + +func getDefaultOptions() *Options { + return &Options{ + WithListenAddrPort: netip.MustParseAddrPort("127.0.0.1:0"), + } +} + +// WithListener allows passing a listener on which to accept connections. If +// this and WithListenAddrPort are both specified, this will take precedence. +func WithListener(with net.Listener) Option { + return func(o *Options) error { + if with == nil { + return errors.New("nil listener passed to WithListener") + } + o.WithListener = with + return nil + } +} + +// WithListenAddrPort allows overriding an address to listen on. Mutually +// exclusive with WithListener; that option will take precedence. If you do not +// want a TCP connection you must use WithListener. +func WithListenAddrPort(with netip.AddrPort) Option { + return func(o *Options) error { + if !with.IsValid() { + return errors.New("invalid addr/port passed to WithListenAddrPort") + } + o.WithListenAddrPort = with + return nil + } +} + +// WithConnectionsLeftCh allows providing a channel to receive updates about how +// many connections are left. It is the caller's responsibility to ensure that +// this is drained and does not block. +func WithConnectionsLeftCh(with chan int32) Option { + return func(o *Options) error { + if with == nil { + return errors.New("channel passed to WithConnectionsLeftCh is nil") + } + o.WithConnectionsLeftCh = with + return nil + } +} + +// WithWorkerHost can be used to override the worker host read from the session +// authorization data. This can be used to override the SNI value in the client +// TLS configuration and is mostly useful for tests. +func WithWorkerHost(with string) Option { + return func(o *Options) error { + o.WithWorkerHost = with + return nil + } +} + +// WithSessionAuthorizationData can be used to provide already-unmarshaled +// session authorization instead of a string token. +func WithSessionAuthorizationData(with *targets.SessionAuthorizationData) Option { + return func(o *Options) error { + if with == nil { + return errors.New("data passed to WithSessionAuthorizationData is nil") + } + o.WithSessionAuthorizationData = with + return nil + } +} + +// WithSkipSessionTeardown can be used to override the normal behavior of the +// session sending a teardown request to the worker on completion. This is +// useful if you know that this will result in an error (for instance, if the +// worker is going to be offline) and want to avoid the attempted connection or +// avoid the error rather than ignore it. +func WithSkipSessionTeardown(with bool) Option { + return func(o *Options) error { + o.WithSkipSessionTeardown = with + return nil + } +} diff --git a/api/proxy/option_test.go b/api/proxy/option_test.go new file mode 100644 index 000000000..f982c2887 --- /dev/null +++ b/api/proxy/option_test.go @@ -0,0 +1,89 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package proxy + +import ( + "net" + "net/netip" + "testing" + + "github.com/hashicorp/boundary/api/targets" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func Test_GetOpts(t *testing.T) { + t.Parallel() + t.Run("nil", func(t *testing.T) { + opts, err := getOpts(nil) + require.NoError(t, err) + assert.NotNil(t, opts) + }) + t.Run("with-listener", func(t *testing.T) { + assert := assert.New(t) + opts, err := getOpts() + require.NoError(t, err) + assert.Nil(opts.WithListener) + _, err = getOpts(WithListener(nil)) + require.Error(t, err) + l := &net.TCPListener{} + opts, err = getOpts(WithListener(l)) + require.NoError(t, err) + assert.Equal(l, opts.WithListener) + }) + t.Run("with-listen-addr-port", func(t *testing.T) { + assert := assert.New(t) + opts, err := getOpts() + require.NoError(t, err) + assert.EqualValues(netip.MustParseAddrPort("127.0.0.1:0"), opts.WithListenAddrPort) + _, err = getOpts(WithListenAddrPort(netip.AddrPort{})) + require.Error(t, err) + l := netip.AddrPortFrom(netip.IPv6LinkLocalAllNodes(), 22) + opts, err = getOpts(WithListenAddrPort(l)) + require.NoError(t, err) + assert.Equal(l, opts.WithListenAddrPort) + }) + t.Run("with-connections-left-ch", func(t *testing.T) { + assert := assert.New(t) + opts, err := getOpts() + require.NoError(t, err) + assert.Nil(opts.WithConnectionsLeftCh) + _, err = getOpts(WithConnectionsLeftCh(nil)) + require.Error(t, err) + l := make(chan int32) + opts, err = getOpts(WithConnectionsLeftCh(l)) + require.NoError(t, err) + assert.Equal(l, opts.WithConnectionsLeftCh) + }) + t.Run("with-worker-host", func(t *testing.T) { + assert := assert.New(t) + opts, err := getOpts() + require.NoError(t, err) + assert.Empty(opts.WithWorkerHost) + opts, err = getOpts(WithWorkerHost("foo")) + require.NoError(t, err) + assert.Equal("foo", opts.WithWorkerHost) + }) + t.Run("with-session-authorization-data", func(t *testing.T) { + assert := assert.New(t) + opts, err := getOpts() + require.NoError(t, err) + assert.Nil(opts.WithSessionAuthorizationData) + _, err = getOpts(WithSessionAuthorizationData(nil)) + require.Error(t, err) + l := &targets.SessionAuthorizationData{} + opts, err = getOpts(WithSessionAuthorizationData(l)) + require.NoError(t, err) + assert.Equal(l, opts.WithSessionAuthorizationData) + }) + t.Run("with-skip-session-teardown", func(t *testing.T) { + assert := assert.New(t) + opts, err := getOpts() + require.NoError(t, err) + assert.Empty(opts.WithSkipSessionTeardown) + opts, err = getOpts(WithSkipSessionTeardown(true)) + require.NoError(t, err) + assert.True(opts.WithSkipSessionTeardown) + }) +} diff --git a/api/proxy/proxy.go b/api/proxy/proxy.go new file mode 100644 index 000000000..a565d12b8 --- /dev/null +++ b/api/proxy/proxy.go @@ -0,0 +1,390 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package proxy + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "net" + "net/http" + "net/netip" + "sync" + "sync/atomic" + "time" + + "github.com/hashicorp/boundary/api/targets" + cleanhttp "github.com/hashicorp/go-cleanhttp" + "github.com/hashicorp/go-secure-stdlib/base62" + "github.com/hashicorp/go-secure-stdlib/temperror" + ua "go.uber.org/atomic" +) + +// Note: Much of the proxying behavior is tested in various tests within +// internal/tests but those cannot be included in this package as they would +// create a dependency on the main module from this module. + +// This can take more time than you might expect, especially if a lot of these +// are sent at once, so the timeout is quite long. We could allow a custom +// timeout to be an option if we wish. +const sessionCancelTimeout = 30 * time.Second + +type ClientProxy struct { + tofuToken string + cachedListenerAddress *ua.String + connectionsLeft *atomic.Int32 + connsLeftCh chan int32 + callerConnectionsLeftCh chan int32 + sessionAuthzData *targets.SessionAuthorizationData + createTime time.Time + expiration time.Time + ctx context.Context + cancel context.CancelFunc + transport *http.Transport + workerAddr string + listenAddrPort netip.AddrPort + listener *atomic.Value + listenerCloseOnce *sync.Once + clientTlsConf *tls.Config + connWg *sync.WaitGroup + started *atomic.Bool + skipSessionTeardown bool +} + +// New creates a new client proxy. The given context should be cancelable; once +// the proxy is started, cancel the context to stop the proxy. The proxy may +// also cancel on its own if the session expires or there are no connections +// left. +// +// Supported options: +// +// * WithListenAddrPort - Specify a TCP address and port on which to listen +// +// * WithListener - Specify a custom listener on which to accept connections; +// overrides WithListenAddrPort if both are set +// +// * WithSessionAuthorizationData - Specify an already-unmarshaled session +// authorization object. If set, authzToken can be empty. +// +// * WithConnectionsLeftCh - Specify a channel on which to send the number of +// remaining connections as they are consumed +// +// * WithWorkerHost - If set, use this host name as the SNI host when making the +// TLS connection to the worker +// +// EXPERIMENTAL: While this API is not expected to change, it is new and +// feedback from users may necessitate changes. +func New(ctx context.Context, authzToken string, opt ...Option) (*ClientProxy, error) { + opts, err := getOpts(opt...) + if err != nil { + return nil, fmt.Errorf("could not parse options: %w", err) + } + + if authzToken == "" && opts.WithSessionAuthorizationData == nil { + return nil, fmt.Errorf("empty session authorization token and object") + } + + p := &ClientProxy{ + cachedListenerAddress: ua.NewString(""), + connsLeftCh: make(chan int32), + connectionsLeft: new(atomic.Int32), + listener: new(atomic.Value), + listenerCloseOnce: new(sync.Once), + connWg: new(sync.WaitGroup), + listenAddrPort: opts.WithListenAddrPort, + callerConnectionsLeftCh: opts.WithConnectionsLeftCh, + started: new(atomic.Bool), + skipSessionTeardown: opts.WithSkipSessionTeardown, + } + + if opts.WithListener != nil { + p.listener.Store(opts.WithListener) + } + + p.tofuToken, err = base62.Random(20) + if err != nil { + return nil, fmt.Errorf("could not derive random bytes for tofu token: %w", err) + } + + p.sessionAuthzData = opts.WithSessionAuthorizationData + if p.sessionAuthzData == nil { + p.sessionAuthzData, err = targets.SessionAuthorization{AuthorizationToken: authzToken}.GetSessionAuthorizationData() + if err != nil { + return nil, fmt.Errorf("error turning authz token into authorization data: %w", err) + } + } + + if len(p.sessionAuthzData.WorkerInfo) == 0 { + return nil, errors.New("no workers found in authorization data") + } + + if opts.WithListener == nil { + if p.listenAddrPort.Port() == 0 { + p.listenAddrPort = netip.AddrPortFrom(p.listenAddrPort.Addr(), uint16(p.sessionAuthzData.DefaultClientPort)) + } + } + p.connectionsLeft.Store(p.sessionAuthzData.ConnectionLimit) + p.workerAddr = p.sessionAuthzData.WorkerInfo[0].Address + + tlsConf, err := p.clientTlsConfig(opt...) + if err != nil { + return nil, fmt.Errorf("error creating TLS configuration: %w", err) + } + p.createTime = p.sessionAuthzData.CreatedTime + p.expiration = p.sessionAuthzData.Expiration + + // We don't _rely_ on client-side timeout verification but this prevents us + // seeming to be ready for a connection that will immediately fail when we + // try to actually make it + p.ctx, p.cancel = context.WithDeadline(ctx, p.expiration) + + transport := cleanhttp.DefaultTransport() + transport.DisableKeepAlives = false + // This isn't/shouldn't be used anyways really because the connection is + // hijacked, just setting for completeness + transport.IdleConnTimeout = 0 + transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + dialer := &tls.Dialer{Config: tlsConf} + return dialer.DialContext(ctx, network, addr) + } + p.transport = transport + + return p, nil +} + +// Start starts the listener for client proxying. It ends, with any errors, when +// the listener is closed and no connections are left. Cancel the client's proxy +// to force this to happen early. It is not safe to call Start twice, including +// once it has exited, and will immediately error in this case; create a new +// ClientProxy with New(). +// +// Note: if a custom listener implementation is used and the implementation can +// return a Temporary error, the listener will not be closed on that condition +// and no feedback will be given. It is up to the listener implementation to +// inform the client, if needed, of any status causing a Temporary error to be +// returned on accept. +// +// EXPERIMENTAL: While this API is not expected to change, it is new and +// feedback from users may necessitate changes. +func (p *ClientProxy) Start() (retErr error) { + if !p.started.CompareAndSwap(false, true) { + return errors.New("proxy was already started") + } + + defer p.cancel() + + if p.listener.Load() == nil { + var err error + ln, err := net.ListenTCP("tcp", &net.TCPAddr{ + IP: p.listenAddrPort.Addr().AsSlice(), + Port: int(p.listenAddrPort.Port()), + }) + if err != nil { + return fmt.Errorf("unable to start listening: %w", err) + } + p.listener.Store(ln) + } + + listenerCloseFunc := func() { + p.listenerCloseOnce.Do(func() { + // Forces the for loop to exit instead of spinning on errors + p.cancel() + p.connectionsLeft.Store(0) + if err := p.listener.Load().(net.Listener).Close(); err != nil && err != net.ErrClosed { + retErr = errors.Join(retErr, fmt.Errorf("error closing proxy listener: %w", err)) + } + }) + } + + // Ensure closing the listener runs on any other return condition + defer listenerCloseFunc() + + fin := make(chan error, 10) + p.connWg.Add(1) + go func() { + defer p.connWg.Done() + for { + listeningConn, err := p.listener.Load().(net.Listener).Accept() + if err != nil { + select { + case <-p.ctx.Done(): + return + default: + if errors.Is(err, net.ErrClosed) { + // Generally this will be because we canceled the + // context or ran out of session connections and are + // winding down. This will never revert, so return. + return + } + // If the upstream listener indicates that this is an error + // with e.g. just this connection, don't close, just + // continue + if temperror.IsTempError(err) { + continue + } + // No reason to think we can successfully handle the next + // connection that comes our way, so cancel the proxy + fin <- fmt.Errorf("error from accept: %w", err) + listenerCloseFunc() + return + } + } + p.connWg.Add(1) + go func() { + defer listeningConn.Close() + defer p.connWg.Done() + wsConn, err := p.getWsConn(p.ctx) + if err != nil { + fin <- fmt.Errorf("error from getWsConn: %w", err) + // No reason to think we can successfully handle the next + // connection that comes our way, so cancel the proxy + listenerCloseFunc() + return + } + if err := p.runTcpProxyV1(wsConn, listeningConn); err != nil { + fin <- fmt.Errorf("error from runTcpProxyV1: %w", err) + // No reason to think we can successfully handle the next + // connection that comes our way, so cancel the proxy + listenerCloseFunc() + return + } + }() + } + }() + + p.connWg.Add(1) + go func() { + defer func() { + // Run a function (last, after connwg is done) just to ensure that + // we drain from this in case any connections starting as this + // number changes are trying to send the information down + for { + select { + case <-p.connsLeftCh: + default: + return + } + } + }() + defer p.connWg.Done() + defer listenerCloseFunc() + + for { + select { + case <-p.ctx.Done(): + return + case connsLeft := <-p.connsLeftCh: + p.connectionsLeft.Store(connsLeft) + if p.callerConnectionsLeftCh != nil { + p.callerConnectionsLeftCh <- connsLeft + } + if connsLeft == 0 { + // Close the listener as we can't authorize any more + // connections + return + } + } + } + }() + + p.connWg.Wait() + + { + // the go funcs are done, so we can safely close the chan and range over any errors + close(fin) + var finErrors []error + for err := range fin { + finErrors = append(finErrors, err) + } + if len(finErrors) > 0 { + return errors.Join(finErrors...) + } + } + + var sendSessionCancel bool + // If we're not after expiration, ensure there is a bit of buffer in + // case clocks are not quite the same between worker and this machine + if time.Now().Before(p.expiration.Add(-5 * time.Minute)) { + sendSessionCancel = true + } + + if !sendSessionCancel || p.skipSessionTeardown { + return nil + } + + ctx, cancel := context.WithTimeout(context.Background(), sessionCancelTimeout) + defer cancel() + if err := p.sendSessionTeardown(ctx); err != nil { + return fmt.Errorf("error sending session teardown request to worker: %w", err) + } + + return nil +} + +// ListenerAddress returns the address of the client proxy listener. Because the +// listener is started with Start(), this could be called before listening +// occurs. To avoid returning until we have a valid value, pass a context; +// canceling the context will cause the function to return an empty AddrPort if +// it's not yet known. Otherwise the function will return when the address is +// available. In either case, test the result to ensure it's not empty. +// +// Warning: a non-cancelable context will cause this call to block forever until +// the listener's address can be determined. +// +// EXPERIMENTAL: While this API is not expected to change, it is new and +// feedback from users may necessitate changes. +func (p *ClientProxy) ListenerAddress(ctx context.Context) string { + switch { + case p.cachedListenerAddress.Load() != "": + return p.cachedListenerAddress.Load() + case p.listener.Load() != nil: + addr := p.listener.Load().(net.Listener).Addr().String() + p.cachedListenerAddress.Store(addr) + return addr + case ctx == nil: + return "" + } + timer := time.NewTimer(0) + for { + select { + case <-ctx.Done(): + timer.Stop() + return "" + case <-timer.C: + if p.listener.Load() != nil { + timer.Stop() + addr := p.listener.Load().(net.Listener).Addr().String() + p.cachedListenerAddress.Store(addr) + return addr + } + timer.Reset(10 * time.Millisecond) + } + } +} + +// SessionCreatedTime returns the creation time of the session +// +// EXPERIMENTAL: While this API is not expected to change, it is new and +// feedback from users may necessitate changes. +func (p *ClientProxy) SessionCreatedTime() time.Time { + return p.createTime +} + +// SessionExpiration returns the expiration time of the session +// +// EXPERIMENTAL: While this API is not expected to change, it is new and +// feedback from users may necessitate changes. +func (p *ClientProxy) SessionExpiration() time.Time { + return p.expiration +} + +// ConnectionsLeft returns the number of connections left in the session, or -1 +// if unlimited. +// +// EXPERIMENTAL: While this API is not expected to change, it is new and +// feedback from users may necessitate changes. +func (p *ClientProxy) ConnectionsLeft() int32 { + return p.connectionsLeft.Load() +} diff --git a/api/proxy/proxy_test.go b/api/proxy/proxy_test.go new file mode 100644 index 000000000..6bcebb0a2 --- /dev/null +++ b/api/proxy/proxy_test.go @@ -0,0 +1,195 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package proxy + +import ( + "context" + "crypto/ed25519" + "crypto/rand" + "crypto/x509" + "math/big" + mathrand "math/rand" + "net" + "sync/atomic" + "testing" + "time" + + "github.com/hashicorp/boundary/api/scopes" + "github.com/hashicorp/boundary/api/targets" + "github.com/mitchellh/copystructure" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNew(t *testing.T) { + t.Parallel() + + sessionAuth := testSessionAuth(t) + l := &net.TCPListener{} + + tc := []struct { + name string + authzString string + transformFn func(*targets.SessionAuthorizationData) *targets.SessionAuthorizationData + opts []Option + errContains string + badOpts bool + }{ + { + name: "base", + opts: []Option{WithSessionAuthorizationData(sessionAuth)}, + }, + { + name: "bad options", + opts: []Option{WithSessionAuthorizationData(nil)}, + errContains: "data passed to WithSessionAuthorizationData is nil", + }, + { + name: "no authz string", + errContains: "empty session authorization token and object", + }, + { + name: "bad authz string", + authzString: "bad", + errContains: "error turning authz token into authorization data", + }, + { + name: "no workers", + transformFn: func(in *targets.SessionAuthorizationData) *targets.SessionAuthorizationData { + in.WorkerInfo = []*targets.WorkerInfo{} + return in + }, + opts: []Option{WithSessionAuthorizationData(sessionAuth)}, + errContains: "no workers found in authorization data", + }, + { + name: "with listener", + opts: []Option{WithSessionAuthorizationData(sessionAuth), WithListener(l)}, + }, + } + for _, tt := range tc { + t.Run(tt.name, func(t *testing.T) { + require, assert := require.New(t), assert.New(t) + cpRaw, err := copystructure.Copy(sessionAuth) + require.NoError(err) + cp := sessionAuth + if tt.transformFn != nil { + cp = tt.transformFn(cpRaw.(*targets.SessionAuthorizationData)) + tt.opts = append(tt.opts, WithSessionAuthorizationData(cp)) + } + p, err := New(context.Background(), tt.authzString, tt.opts...) + if tt.errContains != "" { + require.ErrorContains(err, tt.errContains) + return + } + require.NoError(err) + assert.Len(p.tofuToken, 20) + assert.Equal(p.sessionAuthzData, sessionAuth) + assert.True(p.listenAddrPort.IsValid()) + + opts, err := getOpts(tt.opts...) + require.NoError(err) + if opts.WithListener == nil { + assert.EqualValues(getDefaultOptions().WithListenAddrPort.Addr(), p.listenAddrPort.Addr()) + } else { + assert.Equal(p.listener.Load(), l) + } + assert.Equal(p.connectionsLeft.Load(), cp.ConnectionLimit) + assert.Equal(p.ConnectionsLeft(), cp.ConnectionLimit) + assert.Equal(p.createTime, sessionAuth.CreatedTime) + assert.Equal(p.SessionCreatedTime(), sessionAuth.CreatedTime) + assert.Equal(p.expiration, sessionAuth.Expiration) + assert.Equal(p.SessionExpiration(), sessionAuth.Expiration) + assert.NotNil(p.ctx) + assert.NotNil(p.cancel) + assert.NotNil(p.transport) + }) + } +} + +func TestListenerAddr(t *testing.T) { + t.Parallel() + require, assert := require.New(t), assert.New(t) + + sessionAuth := testSessionAuth(t) + + l, err := net.ListenTCP("tcp", &net.TCPAddr{Port: 0}) + require.NoError(err) + defer l.Close() + expAddr := l.Addr().String() + + p, err := New(context.Background(), "", WithSessionAuthorizationData(sessionAuth), WithListener(l)) + require.NoError(err) + + // Listener should be cached so we should get the address of the listener + assert.Equal(expAddr, p.ListenerAddress(context.Background())) + assert.NotNil(p.cachedListenerAddress.Load()) + assert.Equal(expAddr, p.cachedListenerAddress.Load()) + // This should be returning the cached value now, which we already verified + // above, but this will provide coverage + assert.Equal(expAddr, p.ListenerAddress(context.Background())) + + // Now test the context cases + p.cachedListenerAddress.Store("") + p.listener = new(atomic.Value) + + // Check a canceled context + ctx, cancel := context.WithCancel(context.Background()) + cancel() + assert.Equal("", p.ListenerAddress(ctx)) + + // Pass in a new listener after a few seconds and ensure we block until + // then, then receive a valid address + start := time.Now() + time.AfterFunc(3*time.Second, func() { + p.listener.Store(l) + }) + assert.Equal(expAddr, p.ListenerAddress(context.Background())) + assert.WithinDuration(start.Add(3*time.Second), time.Now(), 500*time.Millisecond) +} + +func testSessionAuth(t *testing.T) *targets.SessionAuthorizationData { + sessionAuth := &targets.SessionAuthorizationData{ + SessionId: "s_1234567890", + TargetId: "ttcp_1234567890", + Scope: &scopes.ScopeInfo{ + Id: "p_1234567890", + Type: "project", + ParentScopeId: "o_1234567890", + }, + CreatedTime: time.Now().Add(-time.Minute), + Type: "tcp", + ConnectionLimit: 4, + EndpointPort: 22, + Expiration: time.Now().Add(time.Minute), + HostId: "h_1234567890", + Endpoint: "localhost", + WorkerInfo: []*targets.WorkerInfo{ + { + Address: "localhost:9202", + }, + }, + } + + pubKey, privKey, err := ed25519.GenerateKey(nil) + require.NoError(t, err) + template := &x509.Certificate{ + ExtKeyUsage: []x509.ExtKeyUsage{ + x509.ExtKeyUsageServerAuth, + x509.ExtKeyUsageClientAuth, + }, + DNSNames: []string{"s_1234567890"}, + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageKeyAgreement | x509.KeyUsageCertSign, + SerialNumber: big.NewInt(mathrand.Int63()), + NotBefore: sessionAuth.CreatedTime.Add(-1 * time.Minute), + NotAfter: sessionAuth.Expiration, + BasicConstraintsValid: true, + IsCA: true, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + } + certBytes, err := x509.CreateCertificate(rand.Reader, template, template, pubKey, privKey) + require.NoError(t, err) + sessionAuth.Certificate = certBytes + return sessionAuth +} diff --git a/api/proxy/testing.go b/api/proxy/testing.go new file mode 100644 index 000000000..27f26b79f --- /dev/null +++ b/api/proxy/testing.go @@ -0,0 +1,23 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package proxy + +import ( + "context" + "crypto/tls" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestClientTlsConfig is designed to allow tests to obtain the TLS +// configuration that would be used by the proxy in order to make specific calls +// to workers during tests. +func TestClientTlsConfig(t *testing.T, authzToken string, opt ...Option) *tls.Config { + proxyClient, err := New(context.Background(), authzToken, opt...) + require.NoError(t, err) + tlsConf, err := proxyClient.clientTlsConfig(opt...) + require.NoError(t, err) + return tlsConf +} diff --git a/api/proxy/tls.go b/api/proxy/tls.go new file mode 100644 index 000000000..a33b625a0 --- /dev/null +++ b/api/proxy/tls.go @@ -0,0 +1,100 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package proxy + +import ( + "crypto/ed25519" + "crypto/tls" + "crypto/x509" + "fmt" + "net" + "strings" +) + +// clientTlsConfig creates a TLS configuration to connect to a worker proxy, or +// returns a cached config +// +// Supported options: WithWorkerHost. If provided will use that host (minus +// port) for the SNI header. Otherwise, it will use the first worker host +// provided in the session authorization data. +func (p *ClientProxy) clientTlsConfig(opt ...Option) (*tls.Config, error) { + if p.clientTlsConf != nil { + return p.clientTlsConf, nil + } + + const op = "proxy.clientTlsConfig" + if p.sessionAuthzData == nil { + return nil, fmt.Errorf("%s: nil session authorization data", op) + } + if len(p.sessionAuthzData.WorkerInfo) == 0 { + return nil, fmt.Errorf("%s: no worker info", op) + } + + opts, err := getOpts(opt...) + if err != nil { + return nil, fmt.Errorf("%s: error getting options: %w", op, err) + } + + parsedCert, err := x509.ParseCertificate(p.sessionAuthzData.Certificate) + if err != nil { + return nil, fmt.Errorf("%s: unable to decode mTLS certificate: %w", op, err) + } + + workerHostRaw := opts.WithWorkerHost + if workerHostRaw == "" { + workerHostRaw = p.sessionAuthzData.WorkerInfo[0].Address + } + workerHost, _, err := net.SplitHostPort(workerHostRaw) + if err != nil { + if strings.Contains(err.Error(), "missing port") { + workerHost = workerHostRaw + } else { + return nil, fmt.Errorf("%s: error splitting worker host/port: %w", op, err) + } + } + + certPool := x509.NewCertPool() + certPool.AddCert(parsedCert) + + tlsConf := &tls.Config{ + Certificates: []tls.Certificate{ + { + Certificate: [][]byte{p.sessionAuthzData.Certificate}, + PrivateKey: ed25519.PrivateKey(p.sessionAuthzData.PrivateKey), + Leaf: parsedCert, + }, + }, + ServerName: workerHost, + MinVersion: tls.VersionTLS13, + NextProtos: []string{"http/1.1", p.sessionAuthzData.SessionId}, + + // This is set this way so we can make use of VerifyConnection, which we + // set on this TLS config below. We are not skipping verification! + InsecureSkipVerify: true, + } + + // We disable normal DNS SAN behavior as we don't rely on DNS or IP + // addresses for security and want to avoid issues with including localhost + // etc. + verifyOpts := x509.VerifyOptions{ + DNSName: p.sessionAuthzData.SessionId, + Roots: certPool, + KeyUsages: []x509.ExtKeyUsage{ + x509.ExtKeyUsageClientAuth, + x509.ExtKeyUsageServerAuth, + }, + } + tlsConf.VerifyConnection = func(cs tls.ConnectionState) error { + // Go will not run this without at least one peer certificate, but + // doesn't hurt to check + if len(cs.PeerCertificates) == 0 { + return fmt.Errorf("%s: no peer certificates provided", op) + } + _, err := cs.PeerCertificates[0].Verify(verifyOpts) + return err + } + + p.clientTlsConf = tlsConf + return tlsConf, nil +} diff --git a/api/proxy/websocket.go b/api/proxy/websocket.go new file mode 100644 index 000000000..0774b5147 --- /dev/null +++ b/api/proxy/websocket.go @@ -0,0 +1,125 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package proxy + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "strings" + "sync" + + "github.com/hashicorp/boundary/api/consts" + pb "github.com/hashicorp/boundary/sdk/pbs/proxy" + "github.com/hashicorp/boundary/sdk/wspb" + "nhooyr.io/websocket" +) + +func (p *ClientProxy) getWsConn(ctx context.Context) (*websocket.Conn, error) { + conn, resp, err := websocket.Dial( + ctx, + fmt.Sprintf("ws://%s/v1/proxy", p.workerAddr), + &websocket.DialOptions{ + HTTPClient: &http.Client{ + Transport: p.transport, + }, + Subprotocols: []string{consts.WebsocketProtocolTcpProxyV1}, + }, + ) + if err != nil { + switch { + case strings.Contains(err.Error(), "tls: internal error"): + return nil, errors.New("session credentials were not accepted, or session is unauthorized") + case strings.Contains(err.Error(), "connect: connection refused"): + return nil, fmt.Errorf("unable to connect to worker at %s", p.workerAddr) + default: + return nil, fmt.Errorf("error dialing the worker: %w", err) + } + } + + if resp == nil { + return nil, errors.New("response from worker is nil") + } + if resp.Header == nil { + return nil, errors.New("response header is nil") + } + negProto := resp.Header.Get("Sec-WebSocket-Protocol") + if negProto != consts.WebsocketProtocolTcpProxyV1 { + return nil, fmt.Errorf("unexpected negotiated protocol: %s", negProto) + } + return conn, nil +} + +func (p *ClientProxy) sendSessionTeardown(ctx context.Context) error { + handshake := pb.ClientHandshake{ + TofuToken: p.tofuToken, + Command: pb.HANDSHAKECOMMAND_HANDSHAKECOMMAND_SESSION_CANCEL, + } + wsConn, err := p.getWsConn(ctx) + if err != nil { + return fmt.Errorf("error fetching connection to send session teardown request to worker: %w", err) + } + if err := wspb.Write(ctx, wsConn, &handshake); err != nil { + return fmt.Errorf("error sending teardown handshake to worker: %w", err) + } + + return nil +} + +func (p *ClientProxy) runTcpProxyV1(wsConn *websocket.Conn, listeningConn net.Conn) error { + handshake := pb.ClientHandshake{TofuToken: p.tofuToken} + if err := wspb.Write(p.ctx, wsConn, &handshake); err != nil { + return fmt.Errorf("error sending handshake to worker: %w", err) + } + var handshakeResult pb.HandshakeResult + if err := wspb.Read(p.ctx, wsConn, &handshakeResult); err != nil { + switch { + case strings.Contains(err.Error(), "unable to authorize connection"): + // There's no reason to think we'd be able to authorize any more + // connections after the first has failed. We don't cancel the + // context here as existing connections may be fine. + p.connsLeftCh <- 0 + return errors.New("unable to authorize connection") + } + switch { + case strings.Contains(err.Error(), "tofu token not allowed"): + // If our tofu token is not allowed something is wrong, and we + // should cancel anything we have going. + p.cancel() + return errors.New("session is already in use") + default: + // If we can't handshake we can't do anything, so quit out + p.cancel() + return fmt.Errorf("error reading handshake result: %w", err) + } + } + + if handshakeResult.GetConnectionsLeft() != -1 { + p.connsLeftCh <- handshakeResult.GetConnectionsLeft() + } + + // Get a wrapped net.Conn so we can use io.Copy + netConn := websocket.NetConn(p.ctx, wsConn, websocket.MessageBinary) + + localWg := new(sync.WaitGroup) + localWg.Add(2) + go func() { + defer localWg.Done() + io.Copy(netConn, listeningConn) + netConn.Close() + listeningConn.Close() + }() + go func() { + defer localWg.Done() + io.Copy(listeningConn, netConn) + listeningConn.Close() + netConn.Close() + }() + localWg.Wait() + + return nil +} diff --git a/api/targets/session_authorization.gen.go b/api/targets/session_authorization.gen.go index a0c1634d0..7b7d4f43c 100644 --- a/api/targets/session_authorization.gen.go +++ b/api/targets/session_authorization.gen.go @@ -20,6 +20,7 @@ type SessionAuthorization struct { HostId string `json:"host_id,omitempty"` Type string `json:"type,omitempty"` AuthorizationToken string `json:"authorization_token,omitempty"` + ConnectionLimit int32 `json:"connection_limit,omitempty"` Endpoint string `json:"endpoint,omitempty"` EndpointPort uint32 `json:"endpoint_port,omitempty"` Expiration time.Time `json:"expiration,omitempty"` diff --git a/go.mod b/go.mod index 8a06a5fa0..1d183c384 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/hashicorp/boundary -go 1.21 +go 1.21.1 replace github.com/hashicorp/boundary/api => ./api @@ -164,6 +164,7 @@ require ( github.com/hashicorp/errwrap v1.1.0 // indirect github.com/hashicorp/go-kms-wrapping/plugin/v2 v2.0.5 // indirect github.com/hashicorp/go-plugin v1.5.2 // indirect + github.com/hashicorp/go-secure-stdlib/temperror v0.1.1 // indirect github.com/hashicorp/go-secure-stdlib/tlsutil v0.1.3 // indirect github.com/hashicorp/mql v0.1.3 github.com/hashicorp/vault/sdk v0.3.0 // indirect diff --git a/go.sum b/go.sum index 639e17a75..76287f535 100644 --- a/go.sum +++ b/go.sum @@ -304,6 +304,8 @@ github.com/hashicorp/go-secure-stdlib/reloadutil v0.1.1/go.mod h1:Ch/bf00Qnx77MZ github.com/hashicorp/go-secure-stdlib/strutil v0.1.1/go.mod h1:gKOamz3EwoIoJq7mlMIRBpVTAUn8qPCrEclOKKWhD3U= github.com/hashicorp/go-secure-stdlib/strutil v0.1.2 h1:kes8mmyCpxJsI7FTwtzRqEy9CdjCtrXrXGuOpxEA7Ts= github.com/hashicorp/go-secure-stdlib/strutil v0.1.2/go.mod h1:Gou2R9+il93BqX25LAKCLuM+y9U2T4hlwvT1yprcna4= +github.com/hashicorp/go-secure-stdlib/temperror v0.1.1 h1:WkyqHb9NZWMEbB2rypsadIlbFOK4UOQPrZK5lPRDDrc= +github.com/hashicorp/go-secure-stdlib/temperror v0.1.1/go.mod h1:BkcKjSGVPPVa9VBEbyYBhkaSEp8dqq97m8TbdcW+Y3U= github.com/hashicorp/go-secure-stdlib/tlsutil v0.1.1/go.mod h1:l8slYwnJA26yBz+ErHpp2IRCLr0vuOMGBORIz4rRiAs= github.com/hashicorp/go-secure-stdlib/tlsutil v0.1.3 h1:xbrxd0U9XQW8qL1BAz2XrAjAF/P2vcqUTAues9c24B8= github.com/hashicorp/go-secure-stdlib/tlsutil v0.1.3/go.mod h1:LWq2Sy8UoKKuK4lFuCNWSjJj57MhNNf2zzBWMtkAIX4= diff --git a/internal/cmd/commands/connect/connect.go b/internal/cmd/commands/connect/connect.go index 5fe8e4123..e840506ba 100644 --- a/internal/cmd/commands/connect/connect.go +++ b/internal/cmd/commands/connect/connect.go @@ -5,38 +5,27 @@ package connect import ( "context" - "crypto/tls" "encoding/json" "errors" "fmt" "io" - "io/ioutil" + "math" "net" - "net/http" + "net/netip" "os" "strconv" "strings" - "sync" "syscall" "time" "github.com/hashicorp/boundary/api" apiproxy "github.com/hashicorp/boundary/api/proxy" "github.com/hashicorp/boundary/api/targets" - "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/cmd/base" - targetspb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets" - "github.com/hashicorp/boundary/sdk/pbs/proxy" - "github.com/hashicorp/boundary/sdk/wspb" - "github.com/hashicorp/go-cleanhttp" - "github.com/hashicorp/go-secure-stdlib/base62" "github.com/mitchellh/cli" - "github.com/mr-tron/base58" "github.com/posener/complete" "go.uber.org/atomic" exec "golang.org/x/sys/execabs" - "google.golang.org/protobuf/proto" - "nhooyr.io/websocket" ) const sessionCancelTimeout = 10 * time.Second @@ -48,6 +37,10 @@ type SessionInfo struct { Expiration time.Time `json:"expiration"` ConnectionLimit int32 `json:"connection_limit"` SessionId string `json:"session_id"` + Endpoint string `json:"-"` + Type string `json:"-"` + TargetId string `json:"-"` + HostId string `json:"-"` Credentials []*targets.SessionCredential `json:"credentials,omitempty"` } @@ -94,16 +87,8 @@ type Command struct { Func string - sessionAuthz *targets.SessionAuthorization - sessionAuthzData *targetspb.SessionAuthorizationData + sessInfo SessionInfo - connWg *sync.WaitGroup - listenerCloseOnce sync.Once - listener *net.TCPListener - listenerAddr *net.TCPAddr - connsLeftCh chan int32 - connectionsLeft *atomic.Int32 - expiration time.Time execCmdReturnValue *atomic.Int32 proxyCtx context.Context proxyCancel context.CancelFunc @@ -277,6 +262,14 @@ func (c *Command) Run(args []string) (retCode int) { return base.CommandUserError } + if c.flagListenPort < 0 || c.flagListenPort > math.MaxUint16 { + c.PrintCliError(errors.New("Invalid listen port supplied")) + return base.CommandCliError + } + + c.proxyCtx, c.proxyCancel = context.WithCancel(c.Context) + defer c.proxyCancel() + switch { case c.flagAuthzToken != "": switch { @@ -316,29 +309,11 @@ func (c *Command) Run(args []string) (retCode int) { } } - tofuToken, err := base62.Random(20) - if err != nil { - c.PrintCliError(fmt.Errorf("Could not derive random bytes for tofu token: %w", err)) - return base.CommandCliError - } - - c.connectionsLeft = atomic.NewInt32(0) - c.connsLeftCh = make(chan int32) - - if c.flagListenAddr == "" { - c.flagListenAddr = "127.0.0.1" - } - listenAddr := net.ParseIP(c.flagListenAddr) - if listenAddr == nil { - c.PrintCliError(fmt.Errorf("Could not successfully parse listen address of %s", c.flagListenAddr)) - return base.CommandUserError - } - authzString := c.flagAuthzToken switch { case authzString != "": if authzString == "-" { - authBytes, err := ioutil.ReadAll(os.Stdin) + authBytes, err := io.ReadAll(os.Stdin) if err != nil { c.PrintCliError(fmt.Errorf("No authorization string was provided and encountered the following error attempting to read it from stdin: %w", err)) return base.CommandUserError @@ -358,12 +333,27 @@ func (c *Command) Run(args []string) (retCode int) { if authzString[0] == '{' { // Attempt to decode the JSON output of an authorize-session call // and pull the token out of there - c.sessionAuthz = new(targets.SessionAuthorization) - if err := json.Unmarshal([]byte(authzString), c.sessionAuthz); err == nil { - authzString = c.sessionAuthz.AuthorizationToken + sa := new(targets.SessionAuthorization) + if err := json.Unmarshal([]byte(authzString), sa); err == nil { + authzString = sa.AuthorizationToken } } + sad, err := targets.SessionAuthorization{AuthorizationToken: authzString}.GetSessionAuthorizationData() + if err != nil { + c.PrintCliError(fmt.Errorf("Error decoding session authorization data: %w", err)) + return base.CommandUserError + } + c.sessInfo = SessionInfo{ + Protocol: sad.Type, + ConnectionLimit: sad.ConnectionLimit, + SessionId: sad.SessionId, + Endpoint: sad.Endpoint, + Type: sad.Type, + TargetId: sad.TargetId, + HostId: sad.HostId, + } + default: client, err := c.Client() if c.WrapperCleanupFunc != nil { @@ -402,186 +392,66 @@ func (c *Command) Run(args []string) (retCode int) { c.PrintCliError(fmt.Errorf("Error trying to authorize a session against target: %w", err)) return base.CommandCliError } - c.sessionAuthz = sar.GetItem().(*targets.SessionAuthorization) - authzString = c.sessionAuthz.AuthorizationToken - } - marshaled, err := base58.FastBase58Decoding(authzString) - if err != nil { - c.PrintCliError(fmt.Errorf("Unable to base58-decode authorization data: %w", err)) - return base.CommandUserError - } - if len(marshaled) == 0 { - c.PrintCliError(errors.New("Zero length authorization information after decoding")) - return base.CommandUserError - } - - c.sessionAuthzData = new(targetspb.SessionAuthorizationData) - if err := proto.Unmarshal(marshaled, c.sessionAuthzData); err != nil { - c.PrintCliError(fmt.Errorf("Unable to proto-decode authorization data: %w", err)) - return base.CommandUserError - } - - if len(c.sessionAuthzData.GetWorkerInfo()) == 0 { - c.PrintCliError(errors.New("No workers found in authorization string")) - return base.CommandUserError - } - - if c.flagListenPort == 0 { - c.flagListenPort = int(c.sessionAuthzData.DefaultClientPort) - } - - c.connectionsLeft.Store(c.sessionAuthzData.ConnectionLimit) - workerAddr := c.sessionAuthzData.GetWorkerInfo()[0].GetAddress() - workerHost, _, err := net.SplitHostPort(workerAddr) - if err != nil { - if strings.Contains(err.Error(), "missing port") { - workerHost = workerAddr - } else { - c.PrintCliError(fmt.Errorf("Error splitting worker adddress host/port: %w", err)) - return base.CommandUserError + sa := sar.GetItem().(*targets.SessionAuthorization) + c.sessInfo = SessionInfo{ + Protocol: sa.Type, + ConnectionLimit: sa.ConnectionLimit, + SessionId: sa.SessionId, + Endpoint: sa.Endpoint, + Type: sa.Type, + TargetId: sa.TargetId, + HostId: sa.HostId, + Credentials: sa.Credentials, } + authzString = sa.AuthorizationToken } - tlsConf, err := ClientTlsConfig(c.sessionAuthzData, workerHost) + var listenAddr netip.AddrPort + var addr netip.Addr + if c.flagListenAddr == "" { + c.flagListenAddr = "127.0.0.1" + } + addr, err := netip.ParseAddr(c.flagListenAddr) if err != nil { - c.PrintCliError(fmt.Errorf("Error creating TLS configuration: %w", err)) + c.PrintCliError(fmt.Errorf("Error parsing listen address: %w", err)) return base.CommandCliError } - c.expiration = tlsConf.Certificates[0].Leaf.NotAfter + listenAddr = netip.AddrPortFrom(addr, uint16(c.flagListenPort)) - // We don't _rely_ on client-side timeout verification but this prevents us - // seeming to be ready for a connection that will immediately fail when we - // try to actually make it - c.proxyCtx, c.proxyCancel = context.WithDeadline(c.Context, c.expiration) - defer c.proxyCancel() - - transport := cleanhttp.DefaultTransport() - transport.DisableKeepAlives = false - // This isn't/shouldn't used anyways really because the connection is - // hijacked, just setting for completeness - transport.IdleConnTimeout = 0 - transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { - dialer := &tls.Dialer{Config: tlsConf} - return dialer.DialContext(ctx, network, addr) + connsLeftCh := make(chan int32) + apiProxyOpts := []apiproxy.Option{apiproxy.WithConnectionsLeftCh(connsLeftCh)} + if listenAddr.IsValid() { + apiProxyOpts = append(apiProxyOpts, apiproxy.WithListenAddrPort(listenAddr)) } - - c.listener, err = net.ListenTCP("tcp", &net.TCPAddr{ - IP: listenAddr, - Port: c.flagListenPort, - }) + clientProxy, err := apiproxy.New( + c.proxyCtx, + authzString, + apiProxyOpts..., + ) if err != nil { - c.PrintCliError(fmt.Errorf("Error starting listening port: %w", err)) + c.PrintCliError(fmt.Errorf("Could not create client proxy: %w", err)) return base.CommandCliError } + c.sessInfo.Expiration = clientProxy.SessionExpiration() - listenerCloseFunc := func() { - // Forces the for loop to exist instead of spinning on errors - c.connectionsLeft.Store(0) - if err := c.listener.Close(); err != nil { - c.PrintCliError(fmt.Errorf("Error closing listener on shutdown: %w", err)) - retCode = 2 - } - } + clientProxyCloseCh := make(chan struct{}) + connCountCloseCh := make(chan struct{}) - // Ensure it runs on any other return condition - defer func() { - c.listenerCloseOnce.Do(listenerCloseFunc) - }() - - c.listenerAddr = c.listener.Addr().(*net.TCPAddr) - - if c.Func == "connect" { - // "connect" indicates there is no subcommand to the connect function. - // The only way a user will be able to connect to the session is by - // connecting directly to the port and address we report to them here. - - var creds []*targets.SessionCredential - if c.sessionAuthz != nil && len(c.sessionAuthz.Credentials) > 0 { - creds = c.sessionAuthz.Credentials - } - - sessInfo := SessionInfo{ - Protocol: c.sessionAuthzData.GetType(), - Address: c.listenerAddr.IP.String(), - Port: c.listenerAddr.Port, - Expiration: c.expiration, - ConnectionLimit: c.sessionAuthzData.GetConnectionLimit(), - SessionId: c.sessionAuthzData.GetSessionId(), - Credentials: creds, - } - switch base.Format(c.UI) { - case "table": - c.UI.Output(generateSessionInfoTableOutput(sessInfo)) - case "json": - out, err := json.Marshal(&sessInfo) - if err != nil { - c.PrintCliError(fmt.Errorf("error marshaling session information: %w", err)) - return base.CommandCliError - } - c.UI.Output(string(out)) - } - } - - c.connWg = new(sync.WaitGroup) - - c.connWg.Add(1) + proxyError := new(atomic.Error) go func() { - defer c.connWg.Done() - for { - listeningConn, err := c.listener.AcceptTCP() - if err != nil { - select { - case <-c.proxyCtx.Done(): - return - case <-c.Context.Done(): - return - default: - // When this hits zero we trigger listener close so this - // isn't actually an error condition - if c.connectionsLeft.Load() == 0 { - return - } - c.PrintCliError(fmt.Errorf("Error accepting connection: %w", err)) - continue - } - } - c.connWg.Add(1) - go func() { - defer listeningConn.Close() - defer c.connWg.Done() - wsConn, err := c.getWsConn( - c.proxyCtx, - workerAddr, - transport) - if err != nil { - c.PrintCliError(err) - } else { - if err := c.runTcpProxyV1(wsConn, listeningConn, tofuToken); err != nil { - c.PrintCliError(err) - } - } - }() - } + defer close(clientProxyCloseCh) + proxyError.Store(clientProxy.Start()) }() - - timer := time.NewTimer(time.Until(c.expiration)) - c.connWg.Add(1) go func() { - defer c.connWg.Done() - defer c.listenerCloseOnce.Do(listenerCloseFunc) - + defer close(connCountCloseCh) for { select { case <-c.proxyCtx.Done(): - timer.Stop() + // When the proxy exits it will cancel this even if we haven't + // done it manually return - case <-c.Context.Done(): - timer.Stop() - return - case <-timer.C: - return - case connsLeft := <-c.connsLeftCh: + case connsLeft := <-connsLeftCh: c.updateConnsLeft(connsLeft) if connsLeft == 0 { return @@ -590,54 +460,73 @@ func (c *Command) Run(args []string) (retCode int) { } }() - if c.flagExec != "" { - c.connWg.Add(1) - c.execCmdReturnValue = atomic.NewInt32(0) - go c.handleExec(passthroughArgs) + if c.Func == "connect" { + // "connect" indicates there is no subcommand to the connect function. + // The only way a user will be able to connect to the session is by + // connecting directly to the port and address we report to them here. + + proxyAddr := clientProxy.ListenerAddress(context.Background()) + var clientProxyHost, clientProxyPort string + clientProxyHost, clientProxyPort, err = net.SplitHostPort(proxyAddr) + if err != nil { + if strings.Contains(err.Error(), "missing port") { + clientProxyHost = proxyAddr + } else { + c.PrintCliError(fmt.Errorf("error splitting listener addr: %w", err)) + return base.CommandCliError + } + } + c.sessInfo.Address = clientProxyHost + + if clientProxyPort != "" { + c.sessInfo.Port, err = strconv.Atoi(clientProxyPort) + if err != nil { + c.PrintCliError(fmt.Errorf("error parsing listener port: %w", err)) + return base.CommandCliError + } + } + switch base.Format(c.UI) { + case "table": + c.UI.Output(generateSessionInfoTableOutput(c.sessInfo)) + case "json": + out, err := json.Marshal(&c.sessInfo) + if err != nil { + c.PrintCliError(fmt.Errorf("error marshaling session information: %w", err)) + return base.CommandCliError + } + c.UI.Output(string(out)) + } } - c.connWg.Wait() + if c.flagExec != "" { + c.execCmdReturnValue = atomic.NewInt32(0) + c.handleExec(clientProxy, passthroughArgs) + } + + <-connCountCloseCh + <-clientProxyCloseCh if c.execCmdReturnValue != nil { retCode = int(c.execCmdReturnValue.Load()) } termInfo := TerminationInfo{Reason: "Unknown"} - sendSessionCancel := false select { case <-c.Context.Done(): termInfo.Reason = "Received shutdown signal" - sendSessionCancel = true default: if c.execCmdReturnValue != nil { // Don't print out in this case, so ensure we clear it termInfo.Reason = "" - sendSessionCancel = true - } else if !timer.Stop() { + } else if time.Now().After(clientProxy.SessionExpiration()) { termInfo.Reason = "Session has expired" - } else { - if c.connectionsLeft.Load() == 0 { - termInfo.Reason = "No connections left in session" - } + } else if clientProxy.ConnectionsLeft() == 0 { + termInfo.Reason = "No connections left in session" + } else if err := proxyError.Load(); err != nil { + termInfo.Reason = "Error from proxy client: " + err.Error() } } - // Only send it if we should, and also if we're not after expiration, with a - // bit of buffer in case clocks are not quite the same between worker and - // this machine. - if sendSessionCancel && time.Now().Before(c.expiration.Add(-5*time.Minute)) { - ctx, cancel := context.WithTimeout(context.Background(), sessionCancelTimeout) - wsConn, err := c.getWsConn(ctx, workerAddr, transport) - if err != nil { - c.PrintCliError(fmt.Errorf("Error fetching connection to send session teardown request to worker: %w", err)) - } else { - if err := c.sendSessionTeardown(ctx, wsConn, tofuToken); err != nil { - c.PrintCliError(fmt.Errorf("Error sending session teardown request to worker: %w", err)) - } - } - cancel() - } - for _, f := range c.cleanupFuncs { if err := f(); err != nil { c.PrintCliError(err) @@ -682,121 +571,7 @@ func (c *Command) printCredentials(creds []*targets.SessionCredential) error { return nil } -func (c *Command) getWsConn( - ctx context.Context, - workerAddr string, - transport *http.Transport, -) (*websocket.Conn, error) { - conn, resp, err := websocket.Dial( - ctx, - fmt.Sprintf("ws://%s/v1/proxy", workerAddr), - &websocket.DialOptions{ - HTTPClient: &http.Client{ - Transport: transport, - }, - Subprotocols: []string{globals.TcpProxyV1}, - }, - ) - if err != nil { - switch { - case strings.Contains(err.Error(), "tls: internal error"): - return nil, errors.New("Session credentials were not accepted, or session is unauthorized") - case strings.Contains(err.Error(), "connect: connection refused"): - return nil, fmt.Errorf("Unable to connect to worker at %s", workerAddr) - default: - return nil, fmt.Errorf("Error dialing the worker: %w", err) - } - } - - if resp == nil { - return nil, errors.New("Response from worker is nil") - } - if resp.Header == nil { - return nil, errors.New("Response header is nil") - } - negProto := resp.Header.Get("Sec-WebSocket-Protocol") - if negProto != globals.TcpProxyV1 { - return nil, fmt.Errorf("Unexpected negotiated protocol: %s", negProto) - } - return conn, nil -} - -func (c *Command) sendSessionTeardown( - ctx context.Context, - wsConn *websocket.Conn, - tofuToken string, -) error { - handshake := proxy.ClientHandshake{ - TofuToken: tofuToken, - Command: proxy.HANDSHAKECOMMAND_HANDSHAKECOMMAND_SESSION_CANCEL, - } - if err := wspb.Write(ctx, wsConn, &handshake); err != nil { - return fmt.Errorf("error sending teardown handshake to worker: %w", err) - } - - return nil -} - -func (c *Command) runTcpProxyV1( - wsConn *websocket.Conn, - listeningConn *net.TCPConn, - tofuToken string, -) error { - handshake := proxy.ClientHandshake{TofuToken: tofuToken} - if err := wspb.Write(c.proxyCtx, wsConn, &handshake); err != nil { - return fmt.Errorf("Error sending handshake to worker: %w", err) - } - var handshakeResult proxy.HandshakeResult - if err := wspb.Read(c.proxyCtx, wsConn, &handshakeResult); err != nil { - switch { - case strings.Contains(err.Error(), "unable to authorize connection"): - // There's no reason to think we'd be able to authorize any more - // connections after the first has failed - c.connsLeftCh <- 0 - return errors.New("Unable to authorize connection") - } - switch { - case strings.Contains(err.Error(), "tofu token not allowed"): - // Nothing will be able to be done here, so cancel the context too - c.proxyCancel() - return errors.New("Session token has already been used") - default: - // If we can't handshake we can't do anything, so quit out - c.proxyCancel() - return fmt.Errorf("Error reading handshake result: %w", err) - } - } - - if handshakeResult.GetConnectionsLeft() != -1 { - c.connsLeftCh <- handshakeResult.GetConnectionsLeft() - } - - // Get a wrapped net.Conn so we can use io.Copy - netConn := websocket.NetConn(c.proxyCtx, wsConn, websocket.MessageBinary) - - localWg := new(sync.WaitGroup) - localWg.Add(2) - - go func() { - defer localWg.Done() - io.Copy(netConn, listeningConn) - netConn.Close() - listeningConn.Close() - }() - go func() { - defer localWg.Done() - io.Copy(listeningConn, netConn) - listeningConn.Close() - netConn.Close() - }() - localWg.Wait() - - return nil -} - func (c *Command) updateConnsLeft(connsLeft int32) { - c.connectionsLeft.Store(connsLeft) - connInfo := ConnectionInfo{ ConnectionsLeft: connsLeft, } @@ -808,29 +583,38 @@ func (c *Command) updateConnsLeft(connsLeft int32) { case "json": out, err := json.Marshal(&connInfo) if err != nil { - c.PrintCliError(fmt.Errorf("error marshaling connection information: %w", err)) + c.PrintCliError(fmt.Errorf("Error marshaling connection information: %w", err)) } c.UI.Output(string(out)) } } } -func (c *Command) handleExec(passthroughArgs []string) { - defer c.connWg.Done() +func (c *Command) handleExec(clientProxy *apiproxy.ClientProxy, passthroughArgs []string) { defer c.proxyCancel() - port := strconv.Itoa(c.listenerAddr.Port) - ip := c.listenerAddr.IP.String() - addr := c.listenerAddr.String() + addr := clientProxy.ListenerAddress(context.Background()) + var host, port string + var err error + host, port, err = net.SplitHostPort(addr) + if err != nil { + if strings.Contains(err.Error(), "missing port") { + host = addr + } else { + c.PrintCliError(fmt.Errorf("Error splitting listener addr: %w", err)) + c.execCmdReturnValue.Store(int32(3)) + return + } + } var args []string var envs []string var argsErr error var creds apiproxy.Credentials - if c.sessionAuthz != nil { + if len(c.sessInfo.Credentials) > 0 { var err error - creds, err = apiproxy.ParseCredentials(c.sessionAuthz.Credentials) + creds, err = apiproxy.ParseCredentials(c.sessInfo.Credentials) if err != nil { c.PrintCliError(fmt.Errorf("Error interpreting secret: %w", err)) c.execCmdReturnValue.Store(int32(3)) @@ -840,7 +624,7 @@ func (c *Command) handleExec(passthroughArgs []string) { switch c.Func { case "http": - httpArgs, err := c.httpFlags.buildArgs(c, port, ip, addr) + httpArgs, err := c.httpFlags.buildArgs(c, port, host, addr) if err != nil { c.PrintCliError(fmt.Errorf("Error parsing session args: %w", err)) c.execCmdReturnValue.Store(int32(3)) @@ -849,7 +633,7 @@ func (c *Command) handleExec(passthroughArgs []string) { args = append(args, httpArgs...) case "postgres": - pgArgs, pgEnvs, pgCreds, pgErr := c.postgresFlags.buildArgs(c, port, ip, addr, creds) + pgArgs, pgEnvs, pgCreds, pgErr := c.postgresFlags.buildArgs(c, port, host, addr, creds) if pgErr != nil { argsErr = pgErr break @@ -859,10 +643,10 @@ func (c *Command) handleExec(passthroughArgs []string) { creds = pgCreds case "rdp": - args = append(args, c.rdpFlags.buildArgs(c, port, ip, addr)...) + args = append(args, c.rdpFlags.buildArgs(c, port, host, addr)...) case "ssh": - sshArgs, sshEnvs, sshCreds, sshErr := c.sshFlags.buildArgs(c, port, ip, addr, creds) + sshArgs, sshEnvs, sshCreds, sshErr := c.sshFlags.buildArgs(c, port, host, addr, creds) if sshErr != nil { argsErr = sshErr break @@ -872,7 +656,7 @@ func (c *Command) handleExec(passthroughArgs []string) { creds = sshCreds case "kube": - kubeArgs, err := c.kubeFlags.buildArgs(c, port, ip, addr) + kubeArgs, err := c.kubeFlags.buildArgs(c, port, host, addr) if err != nil { c.PrintCliError(fmt.Errorf("Error parsing session args: %w", err)) c.execCmdReturnValue.Store(int32(3)) @@ -909,7 +693,7 @@ func (c *Command) handleExec(passthroughArgs []string) { for i := range args { args[i] = stringReplacer(args[i], "port", port) - args[i] = stringReplacer(args[i], "ip", ip) + args[i] = stringReplacer(args[i], "ip", host) args[i] = stringReplacer(args[i], "addr", addr) } @@ -920,7 +704,7 @@ func (c *Command) handleExec(passthroughArgs []string) { // Add original and network related envs here cmd.Env = append(os.Environ(), fmt.Sprintf("BOUNDARY_PROXIED_PORT=%s", port), - fmt.Sprintf("BOUNDARY_PROXIED_IP=%s", ip), + fmt.Sprintf("BOUNDARY_PROXIED_IP=%s", host), fmt.Sprintf("BOUNDARY_PROXIED_ADDR=%s", addr), ) // Envs that came from subcommand handling diff --git a/internal/cmd/commands/connect/http.go b/internal/cmd/commands/connect/http.go index eccb460bc..eac4e9915 100644 --- a/internal/cmd/commands/connect/http.go +++ b/internal/cmd/commands/connect/http.go @@ -77,8 +77,8 @@ func (h *httpFlags) defaultExec() string { func (h *httpFlags) buildArgs(c *Command, port, ip, addr string) ([]string, error) { var args []string host := h.flagHttpHost - if host == "" && c.sessionAuthzData.GetEndpoint() != "" { - hostUrl := c.sessionAuthzData.GetEndpoint() + if host == "" && c.sessInfo.Endpoint != "" { + hostUrl := c.sessInfo.Endpoint u, err := url.Parse(hostUrl) if err != nil { return nil, fmt.Errorf("error parsing endpoint URL: %w", err) @@ -94,8 +94,14 @@ func (h *httpFlags) buildArgs(c *Command, port, ip, addr string) ([]string, erro if host != "" { host = strings.TrimSuffix(host, "/") args = append(args, "-H", fmt.Sprintf("Host: %s", host)) - args = append(args, "--resolve", fmt.Sprintf("%s:%s:%s", host, port, ip)) - uri = fmt.Sprintf("%s://%s:%s", h.flagHttpScheme, host, port) + switch port == "" { + case false: + args = append(args, "--resolve", fmt.Sprintf("%s:%s:%s", host, port, ip)) + uri = fmt.Sprintf("%s://%s:%s", h.flagHttpScheme, host, port) + default: + args = append(args, "--resolve", fmt.Sprintf("%s:%s", host, ip)) + uri = fmt.Sprintf("%s://%s", h.flagHttpScheme, host) + } } else { uri = fmt.Sprintf("%s://%s", h.flagHttpScheme, addr) } diff --git a/internal/cmd/commands/connect/kube.go b/internal/cmd/commands/connect/kube.go index dbbc460f3..be7fd463a 100644 --- a/internal/cmd/commands/connect/kube.go +++ b/internal/cmd/commands/connect/kube.go @@ -59,8 +59,8 @@ func (f *kubeFlags) defaultExec() string { func (f *kubeFlags) buildArgs(c *Command, port, ip, addr string) ([]string, error) { var args []string host := f.flagKubeHost - if host == "" && c.sessionAuthzData.GetEndpoint() != "" { - hostUrl := c.sessionAuthzData.GetEndpoint() + if host == "" && c.sessInfo.Endpoint != "" { + hostUrl := c.sessInfo.Endpoint u, err := url.Parse(hostUrl) if err != nil { return nil, fmt.Errorf("error parsing endpoint URL: %w", err) diff --git a/internal/cmd/commands/connect/postgres.go b/internal/cmd/commands/connect/postgres.go index 7415b2929..69e15e97b 100644 --- a/internal/cmd/commands/connect/postgres.go +++ b/internal/cmd/commands/connect/postgres.go @@ -69,7 +69,10 @@ func (p *postgresFlags) buildArgs(c *Command, port, ip, _ string, creds proxy.Cr switch p.flagPostgresStyle { case "psql": - args = append(args, "-p", port, "-h", ip) + if port != "" { + args = append(args, "-p", port) + } + args = append(args, "-h", ip) if c.flagDbname != "" { args = append(args, "-d", c.flagDbname) diff --git a/internal/cmd/commands/connect/ssh.go b/internal/cmd/commands/connect/ssh.go index fa601973a..2cec2e808 100644 --- a/internal/cmd/commands/connect/ssh.go +++ b/internal/cmd/commands/connect/ssh.go @@ -60,7 +60,7 @@ func (s *sshFlags) buildArgs(c *Command, port, ip, _ string, creds apiproxy.Cred retCreds = creds var tryConsume bool - switch string(target.SubtypeFromId(c.sessionAuthzData.GetTargetId())) { + switch string(target.SubtypeFromId(c.sessInfo.TargetId)) { case "tcp": tryConsume = true } @@ -68,21 +68,23 @@ func (s *sshFlags) buildArgs(c *Command, port, ip, _ string, creds apiproxy.Cred switch strings.ToLower(s.flagSshStyle) { case "ssh": // Might want -t for ssh or -tt but seems fine without it for now... - args = append(args, "-p", port) + if port != "" { + args = append(args, "-p", port) + } - switch c.sessionAuthzData.GetType() { + switch c.sessInfo.Type { case "tcp": // SSH detects a host key change when the localhost proxy port changes // This uses the host ID instead of 'localhost:port'. - if len(c.sessionAuthzData.GetHostId()) > 0 { - args = append(args, "-o", fmt.Sprintf("HostKeyAlias=%s", c.sessionAuthzData.GetHostId())) + if len(c.sessInfo.HostId) > 0 { + args = append(args, "-o", fmt.Sprintf("HostKeyAlias=%s", c.sessInfo.HostId)) } else { // In cases where the Target has no host sources and has an // address directly attached to it, we have no Host Id. Use // Target Id instead. Only one address can ever be present on a // target, and no other host sources may be present at the same // time, so this is a reasonable alternative. - args = append(args, "-o", fmt.Sprintf("HostKeyAlias=%s", c.sessionAuthzData.GetTargetId())) + args = append(args, "-o", fmt.Sprintf("HostKeyAlias=%s", c.sessInfo.TargetId)) } case "ssh": args = append(args, "-o", "NoHostAuthenticationForLocalhost=yes") @@ -110,14 +112,18 @@ func (s *sshFlags) buildArgs(c *Command, port, ip, _ string, creds apiproxy.Cred // when the using env-vars. envs = append(envs, fmt.Sprintf("SSHPASS=%s", password)) args = append(args, "-e", "ssh") - args = append(args, "-p", port) + if port != "" { + args = append(args, "-p", port) + } // sshpass cannot handle host key checking, disable localhost key verification // to avoid error: 'SSHPASS detected host authentication prompt. Exiting.' args = append(args, "-o", "NoHostAuthenticationForLocalhost=yes") case "putty": - args = append(args, "-P", port) + if port != "" { + args = append(args, "-P", port) + } } // Check if we got credentials to attempt to use for ssh or putty, diff --git a/internal/cmd/commands/server/worker_shutdown_reload_test.go b/internal/cmd/commands/server/worker_shutdown_reload_test.go index 3f2089e00..63ab344bf 100644 --- a/internal/cmd/commands/server/worker_shutdown_reload_test.go +++ b/internal/cmd/commands/server/worker_shutdown_reload_test.go @@ -57,7 +57,7 @@ func TestServer_ShutdownWorker(t *testing.T) { recoveryWrapper, _ := wrapperWithKey(t) workerAuthWrapper, key := wrapperWithKey(t) testController := controller.NewTestController(t, controller.WithWorkerAuthKms(workerAuthWrapper), controller.WithRootKms(rootWrapper), controller.WithRecoveryKms(recoveryWrapper)) - defer testController.Shutdown() + t.Cleanup(testController.Shutdown) // Start the worker workerCmd := testServerCommand(t, testServerCommandOpts{}) @@ -81,7 +81,7 @@ func TestServer_ShutdownWorker(t *testing.T) { tcl := targets.NewClient(client) tgtL, err := tcl.List(ctx, scope.Global.String(), targets.WithRecursive(true)) require.NoError(err) - require.Len(tgtL.Items, 2) + require.LessOrEqual(2, len(tgtL.Items)) tgt := tgtL.Items[0] require.NotNil(tgt) require.NotNil(tgtL.GetItems()[1]) @@ -89,7 +89,7 @@ func TestServer_ShutdownWorker(t *testing.T) { // Create test server, update default port on target ts := helper.NewTestTcpServer(t) require.NotNil(ts) - defer ts.Close() + t.Cleanup(ts.Close) tgtR, err := tcl.Update(ctx, tgt.Id, tgt.Version, targets.WithTcpTargetDefaultPort(ts.Port())) require.NoError(err) require.NotNil(tgtR) @@ -97,7 +97,7 @@ func TestServer_ShutdownWorker(t *testing.T) { // Authorize and connect // This prevents us from running tests in parallel. tg.SetupSuiteTargetFilters(t) - sess := helper.NewTestSession(ctx, t, tcl, tgt.Id) + sess := helper.NewTestSession(ctx, t, tcl, tgt.Id, helper.WithSkipSessionTeardown(true)) sConn := sess.Connect(ctx, t) // Run initial send/receive test, make sure things are working diff --git a/internal/gen/controller.swagger.json b/internal/gen/controller.swagger.json index 4b32c37ec..55be1f0c0 100644 --- a/internal/gen/controller.swagger.json +++ b/internal/gen/controller.swagger.json @@ -7073,6 +7073,12 @@ "description": "Output only. The marshaled SessionAuthorizationData message containing all information that the proxy needs.", "readOnly": true }, + "connection_limit": { + "type": "integer", + "format": "int32", + "description": "Output only. The connection limit being applied to this session. -1 means unlimited. This is not actually enforced on the client side but it provides for better listener handling by including it.", + "readOnly": true + }, "endpoint": { "type": "string", "description": "Output only. The endpoint address that the worker will connect to, useful for setting TLS parameters.", diff --git a/internal/proto/controller/api/resources/targets/v1/target.proto b/internal/proto/controller/api/resources/targets/v1/target.proto index f7a181c49..8b4f6cf45 100644 --- a/internal/proto/controller/api/resources/targets/v1/target.proto +++ b/internal/proto/controller/api/resources/targets/v1/target.proto @@ -371,6 +371,9 @@ message SessionAuthorization { // Output only. The marshaled SessionAuthorizationData message containing all information that the proxy needs. string authorization_token = 90 [json_name = "authorization_token"]; // @gotags: `class:"secret"` + // Output only. The connection limit being applied to this session. -1 means unlimited. This is not actually enforced on the client side but it provides for better listener handling by including it. + int32 connection_limit = 95 [json_name = "connection_limit"]; + // Output only. The endpoint address that the worker will connect to, useful for setting TLS parameters. string endpoint = 100; // @gotags: `class:"public"` diff --git a/internal/tests/api/proxy/proxy_test.go b/internal/tests/api/proxy/proxy_test.go new file mode 100644 index 000000000..461bf248b --- /dev/null +++ b/internal/tests/api/proxy/proxy_test.go @@ -0,0 +1,137 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package proxy + +import ( + "context" + "net" + "net/netip" + "sync" + "testing" + "time" + + "github.com/hashicorp/boundary/api/proxy" + "github.com/hashicorp/boundary/api/targets" + "github.com/hashicorp/boundary/internal/cmd/config" + "github.com/hashicorp/boundary/internal/daemon/controller" + "github.com/hashicorp/boundary/internal/daemon/worker" + "github.com/hashicorp/boundary/internal/tests/helper" + "github.com/hashicorp/go-hclog" + "github.com/stretchr/testify/require" + + _ "github.com/hashicorp/boundary/internal/daemon/controller/handlers/targets/tcp" +) + +// TestConnectionsLeft tests general proxy functionality, as well as the +// mechanism to notify the caller of the number of connections left on the +// caller's channel and in the ConnectionsLeft function. +func TestConnectionsLeft(t *testing.T) { + require := require.New(t) + logger := hclog.New(&hclog.LoggerOptions{ + Name: t.Name(), + Level: hclog.Trace, + }) + + // Create controller and worker + conf, err := config.DevController() + require.NoError(err) + c1 := controller.NewTestController(t, &controller.TestControllerOpts{ + Config: conf, + InitialResourcesSuffix: "1234567890", + Logger: logger.Named("c1"), + WorkerStatusGracePeriodDuration: helper.DefaultWorkerStatusGracePeriod, + }) + defer c1.Shutdown() + helper.ExpectWorkers(t, c1) + + w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + Logger: logger.Named("w1"), + SuccessfulStatusGracePeriodDuration: helper.DefaultWorkerStatusGracePeriod, + }) + defer w1.Shutdown() + err = w1.Worker().WaitForNextSuccessfulStatusUpdate() + require.NoError(err) + err = c1.WaitForNextWorkerStatusUpdate(w1.Name()) + require.NoError(err) + helper.ExpectWorkers(t, c1, w1) + + // Connect target + client := c1.Client() + client.SetToken(c1.Token().Token) + tcl := targets.NewClient(client) + tgt, err := tcl.Read(c1.Context(), "ttcp_1234567890") + require.NoError(err) + require.NotNil(tgt) + + // Create test server, update default port on target + ts := helper.NewTestTcpServer(t) + require.NotNil(t, ts) + defer ts.Close() + var sessionConnsLimit int32 = 4 + tgt, err = tcl.Update(c1.Context(), tgt.Item.Id, tgt.Item.Version, targets.WithTcpTargetDefaultPort(ts.Port()), targets.WithSessionConnectionLimit(sessionConnsLimit)) + require.NoError(err) + require.NotNil(tgt) + + // Authorize session to get authorization data + sess, err := tcl.AuthorizeSession(c1.Context(), tgt.Item.Id) + require.NoError(err) + sessAuthz, err := sess.GetSessionAuthorization() + require.NoError(err) + + // Create a context we can cancel to stop the proxy, a channel for conns + // left, and a waitgroup to ensure cleanup + pxyCtx, pxyCancel := context.WithCancel(c1.Context()) + defer pxyCancel() + connsLeftCh := make(chan int32) + wg := new(sync.WaitGroup) + + pxy, err := proxy.New(pxyCtx, sessAuthz.AuthorizationToken, proxy.WithConnectionsLeftCh(connsLeftCh)) + require.NoError(err) + wg.Add(1) + go func() { + defer wg.Done() + require.NoError(pxy.Start()) + // Ensure that starting it again does not actually result in it starting + // again + require.ErrorContains(pxy.Start(), "already started") + }() + + addr := pxy.ListenerAddress(context.Background()) + require.NotEmpty(addr) + addrPort, err := netip.ParseAddrPort(addr) + require.NoError(err) + + echo := []byte("echo") + readBuf := make([]byte, len(echo)) + // While we have sessions left, expect no error and to read and write + // through the proxy, and to read the conns left from the channel. Once we + // have hit the connection limit, we expect an error on dial. + for i := sessionConnsLimit; i >= 0; i-- { + // Give time for the listener to be closed. The information about conns + // left comes from upstream responses and we can circle around too fast + // to have the listener be closed, leading to indeterminate results. + time.Sleep(time.Second) + conn, err := net.DialTCP("tcp", nil, net.TCPAddrFromAddrPort(addrPort)) + if i == 0 { + require.Error(err) + break + } + require.NoError(err) + written, err := conn.Write(echo) + require.NoError(err) + require.Equal(written, len(echo)) + read, err := conn.Read(readBuf) + require.NoError(err) + require.Equal(read, len(echo)) + connsLeft := <-connsLeftCh + require.Equal(i-1, connsLeft) + require.Equal(i-1, pxy.ConnectionsLeft()) + } + + pxyCancel() + // Wait to ensure cleanup and that the second-start logic works + wg.Wait() +} diff --git a/internal/tests/cluster/session_cleanup_test.go b/internal/tests/cluster/session_cleanup_test.go index f3fa64572..6e6a2a784 100644 --- a/internal/tests/cluster/session_cleanup_test.go +++ b/internal/tests/cluster/session_cleanup_test.go @@ -106,7 +106,9 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin dawdle.WithWbufSize(256), ) require.NoError(err) - defer proxy.Close() + t.Cleanup(func() { + proxy.Close() + }) require.NotEmpty(t, proxy.ListenerAddr()) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ @@ -138,7 +140,7 @@ func testWorkerSessionCleanupSingle(burdenCase timeoutBurdenType) func(t *testin // Create test server, update default port on target ts := helper.NewTestTcpServer(t) require.NotNil(t, ts) - defer ts.Close() + t.Cleanup(ts.Close) tgt, err = tcl.Update(ctx, tgt.Item.Id, tgt.Item.Version, targets.WithTcpTargetDefaultPort(ts.Port()), targets.WithSessionConnectionLimit(-1)) require.NoError(err) require.NotNil(tgt) @@ -248,7 +250,9 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing dawdle.WithWbufSize(256), ) require.NoError(err) - defer p1.Close() + t.Cleanup(func() { + p1.Close() + }) require.NotEmpty(t, p1.ListenerAddr()) // ************* @@ -261,7 +265,9 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing dawdle.WithWbufSize(256), ) require.NoError(err) - defer p2.Close() + t.Cleanup(func() { + p2.Close() + }) require.NotEmpty(t, p2.ListenerAddr()) // ************ @@ -303,7 +309,7 @@ func testWorkerSessionCleanupMulti(burdenCase timeoutBurdenType) func(t *testing // Create test server, update default port on target ts := helper.NewTestTcpServer(t) require.NotNil(ts) - defer ts.Close() + t.Cleanup(ts.Close) tgt, err = tcl.Update(ctx, tgt.Item.Id, tgt.Item.Version, targets.WithTcpTargetDefaultPort(ts.Port()), targets.WithSessionConnectionLimit(-1)) require.NoError(err) require.NotNil(tgt) diff --git a/internal/tests/cluster/worker_proxy_test.go b/internal/tests/cluster/worker_proxy_test.go index 8c876a7db..585d6452f 100644 --- a/internal/tests/cluster/worker_proxy_test.go +++ b/internal/tests/cluster/worker_proxy_test.go @@ -44,7 +44,7 @@ func TestWorkerSessionProxyMultipleConnections(t *testing.T) { PublicClusterAddr: pl.Addr().String(), WorkerStatusGracePeriodDuration: helper.DefaultWorkerStatusGracePeriod, }) - defer c1.Shutdown() + t.Cleanup(c1.Shutdown) helper.ExpectWorkers(t, c1) @@ -56,7 +56,9 @@ func TestWorkerSessionProxyMultipleConnections(t *testing.T) { dawdle.WithWbufSize(256), ) require.NoError(err) - defer proxy.Close() + t.Cleanup(func() { + proxy.Close() + }) require.NotEmpty(t, proxy.ListenerAddr()) w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ @@ -65,7 +67,7 @@ func TestWorkerSessionProxyMultipleConnections(t *testing.T) { Logger: logger.Named("w1"), SuccessfulStatusGracePeriodDuration: helper.DefaultWorkerStatusGracePeriod, }) - defer w1.Shutdown() + t.Cleanup(w1.Shutdown) err = w1.Worker().WaitForNextSuccessfulStatusUpdate() require.NoError(err) @@ -89,7 +91,9 @@ func TestWorkerSessionProxyMultipleConnections(t *testing.T) { // Create test server, update default port on target ts := helper.NewTestTcpServer(t) require.NotNil(t, ts) - defer ts.Close() + t.Cleanup(ts.Close) + t.Logf("test server listening on port %d", ts.Port()) + tgt, err = tcl.Update(ctx, tgt.Item.Id, tgt.Item.Version, targets.WithTcpTargetDefaultPort(ts.Port()), targets.WithSessionConnectionLimit(-1)) require.NoError(err) require.NotNil(tgt) diff --git a/internal/tests/cluster/x509_verification_test.go b/internal/tests/cluster/x509_verification_test.go index 06ac011aa..655bf80dc 100644 --- a/internal/tests/cluster/x509_verification_test.go +++ b/internal/tests/cluster/x509_verification_test.go @@ -10,11 +10,13 @@ import ( "crypto/x509" "errors" "fmt" + "net" "net/http" "sync" "testing" "time" + apiproxy "github.com/hashicorp/boundary/api/proxy" "github.com/hashicorp/boundary/api/targets" "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/cmd/config" @@ -23,6 +25,7 @@ import ( "github.com/hashicorp/boundary/internal/daemon/worker" "github.com/hashicorp/boundary/internal/event" "github.com/hashicorp/boundary/internal/tests/helper" + "github.com/hashicorp/go-cleanhttp" "github.com/hashicorp/go-hclog" "github.com/stretchr/testify/require" "nhooyr.io/websocket" @@ -64,6 +67,7 @@ func TestCustomX509Verification_Client(t *testing.T) { err = w1.Worker().WaitForNextSuccessfulStatusUpdate() req.NoError(err) + helper.ExpectWorkers(t, c1, w1) // Connect target client := c1.Client() @@ -104,17 +108,26 @@ func TestCustomX509Verification_Client(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { require := require.New(t) - sess := helper.NewTestSession(ctx, t, tcl, "ttcp_1234567890") + sar, err := tcl.AuthorizeSession(ctx, "ttcp_1234567890") + require.NoError(err) + require.NotNil(sar) + + sessAuth, err := sar.GetSessionAuthorization() + require.NoError(err) + sessAuthzData, err := sessAuth.GetSessionAuthorizationData() + require.NoError(err) + require.LessOrEqual(1, len(sessAuthzData.WorkerInfo)) + tlsConf := apiproxy.TestClientTlsConfig(t, sessAuth.AuthorizationToken, apiproxy.WithWorkerHost(sessAuth.SessionId)) certPool := tc.certPool if certPool == nil { - parsedCert, err := x509.ParseCertificate(sess.SessionAuthzData.Certificate) + parsedCert, err := x509.ParseCertificate(sessAuthzData.Certificate) require.NoError(err) certPool = x509.NewCertPool() certPool.AddCert(parsedCert) } - dnsName := sess.SessionId + dnsName := sessAuth.SessionId if tc.dnsName != nil { dnsName = tc.dnsName.String() } @@ -127,7 +140,7 @@ func TestCustomX509Verification_Client(t *testing.T) { x509.ExtKeyUsageServerAuth, }, } - sess.Transport.TLSClientConfig.VerifyConnection = func(cs tls.ConnectionState) error { + tlsConf.VerifyConnection = func(cs tls.ConnectionState) error { // Go will not run this without at least one peer certificate, but // doesn't hurt to check if len(cs.PeerCertificates) == 0 { @@ -136,12 +149,24 @@ func TestCustomX509Verification_Client(t *testing.T) { _, err := cs.PeerCertificates[0].Verify(verifyOpts) return err } + + transport := cleanhttp.DefaultTransport() + transport.DisableKeepAlives = false + // This isn't/shouldn't used anyways really because the connection is + // hijacked, just setting for completeness + transport.IdleConnTimeout = 0 + transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + // time.Sleep(1000 * time.Second) + dialer := &tls.Dialer{Config: tlsConf} + return dialer.DialContext(ctx, network, addr) + } + conn, _, err := websocket.Dial( ctx, - fmt.Sprintf("wss://%s/v1/proxy", sess.WorkerAddr), + fmt.Sprintf("ws://%s/v1/proxy", sessAuthzData.WorkerInfo[0].Address), &websocket.DialOptions{ HTTPClient: &http.Client{ - Transport: sess.Transport, + Transport: transport, }, Subprotocols: []string{globals.TcpProxyV1}, }, @@ -220,14 +245,33 @@ func testCustomX509Verification_Server(ec event.TestConfig, certPool *x509.CertP req.NoError(err) req.NotNil(tgt) - sess := helper.NewTestSession(ctx, t, tcl, "ttcp_1234567890") + sar, err := tcl.AuthorizeSession(ctx, "ttcp_1234567890") + req.NoError(err) + req.NotNil(sar) + sessAuth, err := sar.GetSessionAuthorization() + req.NoError(err) + sessAuthzData, err := sessAuth.GetSessionAuthorizationData() + req.NoError(err) + req.LessOrEqual(1, len(sessAuthzData.WorkerInfo)) + tlsConf := apiproxy.TestClientTlsConfig(t, sessAuth.AuthorizationToken, apiproxy.WithWorkerHost(sessAuth.SessionId)) + + transport := cleanhttp.DefaultTransport() + transport.DisableKeepAlives = false + // This isn't/shouldn't used anyways really because the connection is + // hijacked, just setting for completeness + transport.IdleConnTimeout = 0 + transport.DialContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + // time.Sleep(1000 * time.Second) + dialer := &tls.Dialer{Config: tlsConf} + return dialer.DialContext(ctx, network, addr) + } conn, _, err := websocket.Dial( ctx, - fmt.Sprintf("wss://%s/v1/proxy", sess.WorkerAddr), + fmt.Sprintf("wss://%s/v1/proxy", sessAuthzData.WorkerInfo[0].Address), &websocket.DialOptions{ HTTPClient: &http.Client{ - Transport: sess.Transport, + Transport: transport, }, Subprotocols: []string{globals.TcpProxyV1}, }, diff --git a/internal/tests/helper/option.go b/internal/tests/helper/option.go new file mode 100644 index 000000000..28e9b8be5 --- /dev/null +++ b/internal/tests/helper/option.go @@ -0,0 +1,45 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package helper + +// getOpts iterates the inbound Options and returns a struct and any errors +func getOpts(opt ...Option) (*Options, error) { + opts := getDefaultOptions() + for _, o := range opt { + if o == nil { + continue + } + if err := o(opts); err != nil { + return nil, err + } + + } + return opts, nil +} + +// Options contains various options. The values are exported since the options +// are parsed in various other packages. +type Options struct { + WithSkipSessionTeardown bool +} + +// Option is a function that takes in an options struct and sets values or +// returns an error +type Option func(*Options) error + +func getDefaultOptions() *Options { + return &Options{} +} + +// WithSkipSessionTeardown can be used to override the normal behavior of the +// session sending a teardown request to the worker on completion. This is +// useful if you know that this will result in an error (for instance, if the +// worker is going to be offline) and want to avoid the attempted connection or +// avoid the error rather than ignore it. +func WithSkipSessionTeardown(with bool) Option { + return func(o *Options) error { + o.WithSkipSessionTeardown = with + return nil + } +} diff --git a/internal/tests/helper/testing_helper.go b/internal/tests/helper/testing_helper.go index c4128a250..a5066cb8d 100644 --- a/internal/tests/helper/testing_helper.go +++ b/internal/tests/helper/testing_helper.go @@ -7,32 +7,21 @@ import ( "context" "encoding/binary" "errors" - "fmt" "io" "net" - "net/http" "reflect" "strconv" "testing" "time" - "github.com/cenkalti/backoff/v4" + apiproxy "github.com/hashicorp/boundary/api/proxy" "github.com/hashicorp/boundary/api/targets" - "github.com/hashicorp/boundary/globals" - "github.com/hashicorp/boundary/internal/cmd/commands/connect" "github.com/hashicorp/boundary/internal/daemon/controller" "github.com/hashicorp/boundary/internal/daemon/controller/common" "github.com/hashicorp/boundary/internal/daemon/worker" "github.com/hashicorp/boundary/internal/session" - targetspb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/targets" - "github.com/hashicorp/boundary/sdk/pbs/proxy" - "github.com/hashicorp/boundary/sdk/wspb" - "github.com/hashicorp/go-cleanhttp" - "github.com/hashicorp/go-secure-stdlib/base62" - "github.com/mr-tron/base58" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "google.golang.org/protobuf/proto" "nhooyr.io/websocket" ) @@ -58,124 +47,67 @@ const ( // TestSession represents an authorized session. type TestSession struct { - SessionId string - WorkerAddr string - Transport *http.Transport - tofuToken string - connectionsLeft int32 - SessionAuthzData *targetspb.SessionAuthorizationData + proxy *apiproxy.ClientProxy + SessionId string } -// NewTestSession authorizes a session and creates all of the data -// necessary to initialize. +// NewTestSession authorizes a session and starts a client proxy. Connections +// through the proxy can be established via the `connect` function. func NewTestSession( ctx context.Context, t *testing.T, tcl *targets.Client, targetId string, + opt ...Option, ) *TestSession { t.Helper() - require := require.New(t) + + ctx, cancel := context.WithCancel(ctx) + + require, assert := require.New(t), assert.New(t) + + opts, err := getOpts(opt...) + require.NoError(err) + sar, err := tcl.AuthorizeSession(ctx, targetId) require.NoError(err) require.NotNil(sar) - s := &TestSession{ - SessionId: sar.Item.SessionId, - } - authzString := sar.GetItem().(*targets.SessionAuthorization).AuthorizationToken - marshaled, err := base58.FastBase58Decoding(authzString) - require.NoError(err) - require.NotZero(marshaled) - - s.SessionAuthzData = new(targetspb.SessionAuthorizationData) - err = proto.Unmarshal(marshaled, s.SessionAuthzData) - require.NoError(err) - require.NotZero(s.SessionAuthzData.GetWorkerInfo()) - - s.WorkerAddr = s.SessionAuthzData.GetWorkerInfo()[0].GetAddress() - - tlsConf, err := connect.ClientTlsConfig(s.SessionAuthzData, "") + sessAuth, err := sar.GetSessionAuthorization() require.NoError(err) - s.Transport = cleanhttp.DefaultTransport() - s.Transport.DisableKeepAlives = false - s.Transport.TLSClientConfig = tlsConf - s.Transport.IdleConnTimeout = 0 - - return s -} - -// connect returns a connected websocket for the stored session, -// connecting to the stored workerAddr with the configured transport. -// -// The returned (wrapped) net.Conn should be ready for communication. -func (s *TestSession) connect(ctx context.Context, t *testing.T) net.Conn { - t.Helper() - require := require.New(t) - - var conn *websocket.Conn - var resp *http.Response - var handshakeResult proxy.HandshakeResult - // A retry was added here to mitigate some flakiness. - // Occasionally, `wspb.Read` would throw an error due to "received header with unexpected - // rsv bits set: false:false:true". It was unclear what the cause of this was, so we - // resorted to retrying the connection. - err := backoff.RetryNotify( - func() error { - var err error - conn, resp, err = websocket.Dial( - ctx, - fmt.Sprintf("wss://%s/v1/proxy", s.WorkerAddr), - &websocket.DialOptions{ - HTTPClient: &http.Client{ - Transport: s.Transport, - }, - Subprotocols: []string{globals.TcpProxyV1}, - }, - ) - if err != nil { - return backoff.Permanent(err) - } - if resp.Header.Get("Sec-WebSocket-Protocol") != globals.TcpProxyV1 { - return backoff.Permanent(errors.New( - fmt.Sprintf("Unexpected header. Expected: %s, Actual: %s", - globals.TcpProxyV1, - resp.Header.Get("Sec-WebSocket-Protocol"), - ))) - } - - // Send and receive/check the handshake. - if s.tofuToken == "" { - s.tofuToken, err = base62.Random(20) - if err != nil { - return backoff.Permanent(errors.New("Error creating token")) - } - } - handshake := proxy.ClientHandshake{TofuToken: s.tofuToken} - t.Logf("Using token: %s", s.tofuToken) - - err = wspb.Write(ctx, conn, &handshake) - if err != nil { - return backoff.Permanent(err) - } - - return wspb.Read(ctx, conn, &handshakeResult) - }, - backoff.WithMaxRetries(backoff.NewConstantBackOff(time.Second), 1), - func(err error, td time.Duration) { - t.Logf("Issue with reading websocket: %s. Retrying...", err) - }, + proxy, err := apiproxy.New( + ctx, + sessAuth.AuthorizationToken, + apiproxy.WithWorkerHost(sessAuth.SessionId), + apiproxy.WithSkipSessionTeardown(opts.WithSkipSessionTeardown), ) require.NoError(err) - // This is just a cursory check to make sure that the handshake is - // populated. We could check connections remaining too, but that - // could legitimately be a trivial (zero) value. - require.NotNil(handshakeResult.GetExpiration()) - s.connectionsLeft = handshakeResult.GetConnectionsLeft() + cleanupDone := make(chan struct{}, 1) + go func() { + proxyErr := proxy.Start() + assert.NoError(proxyErr) + close(cleanupDone) + }() + t.Cleanup(func() { + cancel() + <-cleanupDone + }) - return websocket.NetConn(ctx, conn, websocket.MessageBinary) + return &TestSession{ + proxy: proxy, + SessionId: sessAuth.SessionId, + } +} + +// connect returns a connected dialed through the client proxy. The returned +// net.Conn should be ready for communications. +func (s *TestSession) connect(ctx context.Context, t *testing.T) net.Conn { + t.Helper() + conn, err := net.Dial("tcp", s.proxy.ListenerAddress(ctx)) + require.NoError(t, err) + return conn } // ExpectConnectionStateOnController waits until all connections in a diff --git a/sdk/pbs/controller/api/resources/targets/target.pb.go b/sdk/pbs/controller/api/resources/targets/target.pb.go index 6f93a4553..75f56f561 100644 --- a/sdk/pbs/controller/api/resources/targets/target.pb.go +++ b/sdk/pbs/controller/api/resources/targets/target.pb.go @@ -983,6 +983,8 @@ type SessionAuthorization struct { Type string `protobuf:"bytes,80,opt,name=type,proto3" json:"type,omitempty" class:"public" eventstream:"observation"` // @gotags: `class:"public" eventstream:"observation"` // Output only. The marshaled SessionAuthorizationData message containing all information that the proxy needs. AuthorizationToken string `protobuf:"bytes,90,opt,name=authorization_token,proto3" json:"authorization_token,omitempty" class:"secret"` // @gotags: `class:"secret"` + // Output only. The connection limit being applied to this session. -1 means unlimited. This is not actually enforced on the client side but it provides for better listener handling by including it. + ConnectionLimit int32 `protobuf:"varint,95,opt,name=connection_limit,proto3" json:"connection_limit,omitempty"` // Output only. The endpoint address that the worker will connect to, useful for setting TLS parameters. Endpoint string `protobuf:"bytes,100,opt,name=endpoint,proto3" json:"endpoint,omitempty" class:"public"` // @gotags: `class:"public"` // Output only. The endpoint port that will be connected to, either from a @@ -1089,6 +1091,13 @@ func (x *SessionAuthorization) GetAuthorizationToken() string { return "" } +func (x *SessionAuthorization) GetConnectionLimit() int32 { + if x != nil { + return x.ConnectionLimit + } + return 0 +} + func (x *SessionAuthorization) GetEndpoint() string { if x != nil { return x.Endpoint @@ -1555,7 +1564,7 @@ var file_controller_api_resources_targets_v1_target_proto_rawDesc = []byte{ 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x12, 0x31, 0x0a, 0x13, 0x64, 0x65, 0x66, 0x61, 0x75, 0x6c, 0x74, 0x5f, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0xa0, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x13, 0x64, 0x65, 0x66, 0x61, 0x75, - 0x6c, 0x74, 0x5f, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x22, 0xcd, + 0x6c, 0x74, 0x5f, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x22, 0xf9, 0x04, 0x0a, 0x14, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x41, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1e, 0x0a, 0x0a, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73, 0x65, 0x73, @@ -1579,40 +1588,43 @@ var file_controller_api_resources_targets_v1_target_proto_rawDesc = []byte{ 0x74, 0x79, 0x70, 0x65, 0x12, 0x30, 0x0a, 0x13, 0x61, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x18, 0x5a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x13, 0x61, 0x75, 0x74, 0x68, 0x6f, 0x72, 0x69, 0x7a, 0x61, 0x74, 0x69, 0x6f, 0x6e, - 0x5f, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, - 0x6e, 0x74, 0x18, 0x64, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, - 0x6e, 0x74, 0x12, 0x24, 0x0a, 0x0d, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x5f, 0x70, - 0x6f, 0x72, 0x74, 0x18, 0x66, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0d, 0x65, 0x6e, 0x64, 0x70, 0x6f, - 0x69, 0x6e, 0x74, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x3a, 0x0a, 0x0a, 0x65, 0x78, 0x70, 0x69, - 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x69, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, - 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, - 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x65, 0x78, 0x70, 0x69, 0x72, 0x61, - 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x58, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, - 0x61, 0x6c, 0x73, 0x18, 0x6e, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x36, 0x2e, 0x63, 0x6f, 0x6e, 0x74, - 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, - 0x72, 0x63, 0x65, 0x73, 0x2e, 0x74, 0x61, 0x72, 0x67, 0x65, 0x74, 0x73, 0x2e, 0x76, 0x31, 0x2e, - 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x43, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, - 0x6c, 0x52, 0x0b, 0x63, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x73, 0x22, 0x54, - 0x0a, 0x1a, 0x55, 0x73, 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x50, 0x61, 0x73, 0x73, 0x77, 0x6f, - 0x72, 0x64, 0x43, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x12, 0x1a, 0x0a, 0x08, + 0x5f, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x12, 0x2a, 0x0a, 0x10, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, + 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x18, 0x5f, 0x20, 0x01, 0x28, 0x05, + 0x52, 0x10, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x6c, 0x69, 0x6d, + 0x69, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x18, 0x64, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x12, 0x24, + 0x0a, 0x0d, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, + 0x66, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0d, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x5f, + 0x70, 0x6f, 0x72, 0x74, 0x12, 0x3a, 0x0a, 0x0a, 0x65, 0x78, 0x70, 0x69, 0x72, 0x61, 0x74, 0x69, + 0x6f, 0x6e, 0x18, 0x69, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, + 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, + 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x65, 0x78, 0x70, 0x69, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, + 0x12, 0x58, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x73, 0x18, + 0x6e, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x36, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, + 0x65, 0x72, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, + 0x2e, 0x74, 0x61, 0x72, 0x67, 0x65, 0x74, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x65, 0x73, 0x73, + 0x69, 0x6f, 0x6e, 0x43, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x52, 0x0b, 0x63, + 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x73, 0x22, 0x54, 0x0a, 0x1a, 0x55, 0x73, + 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x50, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x43, 0x72, + 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x12, 0x1a, 0x0a, 0x08, 0x75, 0x73, 0x65, 0x72, + 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x75, 0x73, 0x65, 0x72, + 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, + 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, + 0x22, 0x8c, 0x01, 0x0a, 0x17, 0x53, 0x73, 0x68, 0x50, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x4b, + 0x65, 0x79, 0x43, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, 0x12, 0x1a, 0x0a, 0x08, 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, - 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x61, 0x73, 0x73, - 0x77, 0x6f, 0x72, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x61, 0x73, 0x73, - 0x77, 0x6f, 0x72, 0x64, 0x22, 0x8c, 0x01, 0x0a, 0x17, 0x53, 0x73, 0x68, 0x50, 0x72, 0x69, 0x76, - 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x43, 0x72, 0x65, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x61, 0x6c, - 0x12, 0x1a, 0x0a, 0x08, 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, - 0x28, 0x09, 0x52, 0x08, 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1f, 0x0a, 0x0b, - 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, - 0x09, 0x52, 0x0a, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x12, 0x34, 0x0a, - 0x16, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x70, 0x61, 0x73, - 0x73, 0x70, 0x68, 0x72, 0x61, 0x73, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x14, 0x70, - 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x50, 0x61, 0x73, 0x73, 0x70, 0x68, 0x72, - 0x61, 0x73, 0x65, 0x42, 0x50, 0x5a, 0x4e, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, - 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, - 0x64, 0x61, 0x72, 0x79, 0x2f, 0x73, 0x64, 0x6b, 0x2f, 0x70, 0x62, 0x73, 0x2f, 0x63, 0x6f, 0x6e, - 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x72, 0x65, 0x73, 0x6f, - 0x75, 0x72, 0x63, 0x65, 0x73, 0x2f, 0x74, 0x61, 0x72, 0x67, 0x65, 0x74, 0x73, 0x3b, 0x74, 0x61, - 0x72, 0x67, 0x65, 0x74, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1f, 0x0a, 0x0b, 0x70, 0x72, 0x69, 0x76, + 0x61, 0x74, 0x65, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x70, + 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x12, 0x34, 0x0a, 0x16, 0x70, 0x72, 0x69, + 0x76, 0x61, 0x74, 0x65, 0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x70, 0x61, 0x73, 0x73, 0x70, 0x68, 0x72, + 0x61, 0x73, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x14, 0x70, 0x72, 0x69, 0x76, 0x61, + 0x74, 0x65, 0x4b, 0x65, 0x79, 0x50, 0x61, 0x73, 0x73, 0x70, 0x68, 0x72, 0x61, 0x73, 0x65, 0x42, + 0x50, 0x5a, 0x4e, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, + 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, + 0x2f, 0x73, 0x64, 0x6b, 0x2f, 0x70, 0x62, 0x73, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, + 0x6c, 0x65, 0x72, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, + 0x73, 0x2f, 0x74, 0x61, 0x72, 0x67, 0x65, 0x74, 0x73, 0x3b, 0x74, 0x61, 0x72, 0x67, 0x65, 0x74, + 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var (