Update request data in session manager (#2369)

* bug(session): Update request data in session manager

CHANGELOG.md

@@ -4,6 +4,10 @@ Canonical reference for changes, improvements, and bugfixes for Boundary.
 
 ## Next
 
+### Bug Fixes
+* Sessions: Fix an issue where sessions could not have more than one connection 
+([Issue](https://github.com/hashicorp/boundary/issues/2362)), ([PR](https://github.com/hashicorp/boundary/pull/2369))
+
 ## 0.10.2 (2022/08/23)
 
 ### Security

internal/daemon/worker/session/manager.go

@@ -113,7 +113,7 @@ func (m *manager) LoadLocalSession(ctx context.Context, id string, workerId stri
 	}
 	// Update the response to the latest
 	actualSess := actualSessRaw.(*sess)
-	actualSess.ApplyLocalStatus(s.GetStatus())
+	actualSess.ApplySessionUpdate(s.resp)
 	return actualSess, nil
 }
 

internal/daemon/worker/session/manager_test.go

@@ -406,11 +406,13 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 					connInfoMap: map[string]*ConnInfo{
 						"foo": {Id: "foo"},
 					},
+					sessionId: "one",
 				})
 				m.Store("two", &sess{
 					resp: &pbs.LookupSessionResponse{Authorization: &targets.SessionAuthorizationData{
 						SessionId: "two",
 					}},
+					sessionId: "two",
 				})
 
 				return m

internal/daemon/worker/session/session.go

@@ -107,6 +107,7 @@ type sess struct {
 	resp        *pbs.LookupSessionResponse
 	status      pbs.SESSIONSTATUS
 	cert        *x509.Certificate
+	sessionId   string
 }
 
 func newSess(client pbs.SessionServiceClient, resp *pbs.LookupSessionResponse) (*sess, error) {
@@ -128,6 +129,7 @@ func newSess(client pbs.SessionServiceClient, resp *pbs.LookupSessionResponse) (
 		resp:        resp,
 		status:      resp.GetStatus(),
 		cert:        parsedCert,
+		sessionId:   resp.GetAuthorization().GetSessionId(),
 	}
 	return s, nil
 }
@@ -152,6 +154,13 @@ func (s *sess) ApplyLocalConnectionStatus(connId string, status pbs.CONNECTIONST
 	return nil
 }
 
+func (s *sess) ApplySessionUpdate(r *pbs.LookupSessionResponse) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.resp = r
+	s.status = r.Status
+}
+
 func (s *sess) ApplyLocalStatus(st pbs.SESSIONSTATUS) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
@@ -174,18 +183,26 @@ func (s *sess) GetLocalConnections() map[string]ConnInfo {
 }
 
 func (s *sess) GetTofuToken() string {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
 	return s.resp.GetTofuToken()
 }
 
 func (s *sess) GetConnectionLimit() int32 {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
 	return s.resp.GetConnectionLimit()
 }
 
 func (s *sess) GetEndpoint() string {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
 	return s.resp.GetEndpoint()
 }
 
 func (s *sess) GetCredentials() []*pbs.Credential {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
 	return s.resp.GetCredentials()
 }
 
@@ -196,19 +213,25 @@ func (s *sess) GetStatus() pbs.SESSIONSTATUS {
 }
 
 func (s *sess) GetExpiration() time.Time {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
 	return s.resp.GetExpiration().AsTime()
 }
 
 func (s *sess) GetCertificate() *x509.Certificate {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
 	return s.cert
 }
 
 func (s *sess) GetPrivateKey() []byte {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
 	return s.resp.GetAuthorization().GetPrivateKey()
 }
 
 func (s *sess) GetId() string {
-	return s.resp.GetAuthorization().GetSessionId()
+	return s.sessionId
 }
 
 func (s *sess) RequestCancel(ctx context.Context) error {

internal/tests/cluster/worker_proxy_test.go

@@ -0,0 +1,109 @@
+package cluster
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"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/observability/event"
+	"github.com/hashicorp/boundary/internal/tests/helper"
+	"github.com/hashicorp/dawdle"
+	"github.com/hashicorp/go-hclog"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWorkerSessionProxyMultipleConnections(t *testing.T) {
+	const op = "cluster.TestWorkerSessionMultipleConnections"
+	t.Parallel()
+	require := require.New(t)
+	logger := hclog.New(&hclog.LoggerOptions{
+		Name:  t.Name(),
+		Level: hclog.Trace,
+	})
+
+	conf, err := config.DevController()
+	require.NoError(err)
+
+	pl, err := net.Listen("tcp", "localhost:0")
+	require.NoError(err)
+	c1 := controller.NewTestController(t, &controller.TestControllerOpts{
+		Config:                    conf,
+		InitialResourcesSuffix:    "1234567890",
+		Logger:                    logger.Named("c1"),
+		PublicClusterAddr:         pl.Addr().String(),
+		StatusGracePeriodDuration: helper.DefaultGracePeriod,
+	})
+	defer c1.Shutdown()
+
+	expectWorkers(t, c1)
+
+	// Wire up the testing proxies
+	require.Len(c1.ClusterAddrs(), 1)
+	proxy, err := dawdle.NewProxy("tcp", "", c1.ClusterAddrs()[0],
+		dawdle.WithListener(pl),
+		dawdle.WithRbufSize(256),
+		dawdle.WithWbufSize(256),
+	)
+	require.NoError(err)
+	defer proxy.Close()
+	require.NotEmpty(t, proxy.ListenerAddr())
+
+	w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{
+		WorkerAuthKms:             c1.Config().WorkerAuthKms,
+		InitialUpstreams:          []string{proxy.ListenerAddr()},
+		Logger:                    logger.Named("w1"),
+		StatusGracePeriodDuration: helper.DefaultGracePeriod,
+	})
+	defer w1.Shutdown()
+
+	err = w1.Worker().WaitForNextSuccessfulStatusUpdate()
+	require.NoError(err)
+	err = c1.WaitForNextWorkerStatusUpdate(w1.Name())
+	require.NoError(err)
+	expectWorkers(t, c1, w1)
+
+	// Use an independent context for test things that take a context so
+	// that we aren't tied to any timeouts in the controller, etc. This
+	// can interfere with some of the test operations.
+	ctx := context.Background()
+
+	// Connect target
+	client := c1.Client()
+	client.SetToken(c1.Token().Token)
+	tcl := targets.NewClient(client)
+	tgt, err := tcl.Read(ctx, "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()
+	tgt, err = tcl.Update(ctx, tgt.Item.Id, tgt.Item.Version, targets.WithTcpTargetDefaultPort(ts.Port()), targets.WithSessionConnectionLimit(-1))
+	require.NoError(err)
+	require.NotNil(tgt)
+
+	// Authorize and connect
+	sess := helper.NewTestSession(ctx, t, tcl, "ttcp_1234567890")
+	sConn := sess.Connect(ctx, t)
+
+	// Run initial send/receive test, make sure things are working
+	event.WriteSysEvent(ctx, op, "running initial send/recv test for initial connection")
+	sConn.TestSendRecvAll(t)
+
+	// Create another connection for this session
+	sConn2 := sess.Connect(ctx, t)
+	// Ensure second connection works
+	event.WriteSysEvent(ctx, op, "running initial send/recv test for second connection")
+	sConn2.TestSendRecvAll(t)
+
+	// Create another connection for this session
+	sConn3 := sess.Connect(ctx, t)
+	// Ensure second connection works
+	event.WriteSysEvent(ctx, op, "running initial send/recv test for third connection")
+	sConn3.TestSendRecvAll(t)
+}