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 <johan.brandhorst@gmail.com>

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

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=

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
+	}
+}

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)
+	})
+}

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()
+}

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
+}

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
+}

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
+}

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
+}

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"`