From fff5f9506ff4fcf7a7ce79927e55cb02e31f3b23 Mon Sep 17 00:00:00 2001 From: Jeff Mitchell Date: Sun, 21 Jan 2024 19:34:33 -0800 Subject: [PATCH] Migrate proxy logic to the API package (#4216) This allows arbitrary programs to have an easy mechanism to make connections through Boundary using Go. Internal cluster tests and test helpers have been migrated to use this package, as well as the `connect` command. --------- Co-authored-by: Johan Brandhorst-Satzkorn --- api/go.mod | 7 +- api/go.sum | 10 +- api/proxy/option.go | 120 +++++ api/proxy/option_test.go | 89 +++ api/proxy/proxy.go | 390 ++++++++++++++ api/proxy/proxy_test.go | 195 +++++++ api/proxy/testing.go | 23 + api/proxy/tls.go | 100 ++++ api/proxy/websocket.go | 125 +++++ api/targets/session_authorization.gen.go | 1 + go.mod | 3 +- go.sum | 2 + internal/cmd/commands/connect/connect.go | 506 +++++------------- internal/cmd/commands/connect/http.go | 14 +- internal/cmd/commands/connect/kube.go | 4 +- internal/cmd/commands/connect/postgres.go | 5 +- internal/cmd/commands/connect/ssh.go | 22 +- .../server/worker_shutdown_reload_test.go | 8 +- internal/gen/controller.swagger.json | 6 + .../api/resources/targets/v1/target.proto | 3 + internal/tests/api/proxy/proxy_test.go | 137 +++++ .../tests/cluster/session_cleanup_test.go | 16 +- internal/tests/cluster/worker_proxy_test.go | 12 +- .../tests/cluster/x509_verification_test.go | 62 ++- internal/tests/helper/option.go | 45 ++ internal/tests/helper/testing_helper.go | 154 ++---- .../api/resources/targets/target.pb.go | 80 +-- 27 files changed, 1593 insertions(+), 546 deletions(-) create mode 100644 api/proxy/option.go create mode 100644 api/proxy/option_test.go create mode 100644 api/proxy/proxy.go create mode 100644 api/proxy/proxy_test.go create mode 100644 api/proxy/testing.go create mode 100644 api/proxy/tls.go create mode 100644 api/proxy/websocket.go create mode 100644 internal/tests/api/proxy/proxy_test.go create mode 100644 internal/tests/helper/option.go 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 (