Close proxy listener if session is no longer active

api/proxy/option.go

@@ -9,6 +9,7 @@ import (
 	"net/netip"
 	"time"
 
+	"github.com/hashicorp/boundary/api/sessions"
 	"github.com/hashicorp/boundary/api/targets"
 )
 
@@ -37,6 +38,7 @@ type Options struct {
 	WithSessionAuthorizationData *targets.SessionAuthorizationData
 	WithSkipSessionTeardown      bool
 	withSessionTeardownTimeout   time.Duration
+	withSessionsClient           *sessions.Client
 }
 
 // Option is a function that takes in an options struct and sets values or
@@ -129,3 +131,11 @@ func WithSessionTeardownTimeout(with time.Duration) Option {
 		return nil
 	}
 }
+
+// WithSessionsClient provides an optional sessions client
+func WithSessionsClient(with *sessions.Client) Option {
+	return func(o *Options) error {
+		o.withSessionsClient = with
+		return nil
+	}
+}

api/proxy/option_test.go

@@ -9,6 +9,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hashicorp/boundary/api"
+	"github.com/hashicorp/boundary/api/sessions"
 	"github.com/hashicorp/boundary/api/targets"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -96,4 +98,14 @@ func Test_GetOpts(t *testing.T) {
 		require.NoError(t, err)
 		assert.Equal(3*time.Millisecond, opts.withSessionTeardownTimeout)
 	})
+	t.Run("withSessionsClient", func(t *testing.T) {
+		assert := assert.New(t)
+		opts, err := getOpts()
+		require.NoError(t, err)
+		assert.Nil(opts.withSessionsClient)
+		client := sessions.NewClient(&api.Client{})
+		opts, err = getOpts(WithSessionsClient(client))
+		require.NoError(t, err)
+		assert.Equal(client, opts.withSessionTeardownTimeout)
+	})
 }

api/proxy/proxy.go

@@ -15,6 +15,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/hashicorp/boundary/api/sessions"
 	"github.com/hashicorp/boundary/api/targets"
 	cleanhttp "github.com/hashicorp/go-cleanhttp"
 	"github.com/hashicorp/go-secure-stdlib/base62"
@@ -37,6 +38,7 @@ type ClientProxy struct {
 	connectionsLeft         *atomic.Int32
 	connsLeftCh             chan int32
 	callerConnectionsLeftCh chan int32
+	sessionsClient          *sessions.Client
 	sessionAuthzData        *targets.SessionAuthorizationData
 	createTime              time.Time
 	expiration              time.Time
@@ -97,6 +99,7 @@ func New(ctx context.Context, authzToken string, opt ...Option) (*ClientProxy, e
 		callerConnectionsLeftCh: opts.WithConnectionsLeftCh,
 		started:                 new(atomic.Bool),
 		skipSessionTeardown:     opts.WithSkipSessionTeardown,
+		sessionsClient:          opts.withSessionsClient,
 	}
 
 	if opts.WithListener != nil {
@@ -260,6 +263,24 @@ func (p *ClientProxy) Start(opt ...Option) (retErr error) {
 					p.cancel()
 					return
 				}
+				if p.sessionsClient != nil {
+					// If we can tell that the session for the connection we just
+					// closed is terminated, we can close the listener, otherwise
+					// might as well leave it open so the next connection can be
+					// tried.
+					sess, err := p.sessionsClient.Read(p.ctx, p.sessionAuthzData.SessionId)
+					if err != nil || sess == nil || sess.Item == nil || sess.Item.TerminationReason == "" {
+						return
+					}
+
+					// We got a valid session response for the session we just
+					// closed a connection for. Since there is a termination reason
+					// we can treat the session as being terminated so no more
+					// connections will be able to be established.
+					fin <- fmt.Errorf("session no longer active")
+					listenerCloseFunc()
+					p.cancel()
+				}
 			}()
 		}
 	}()