diff --git a/CHANGELOG.md b/CHANGELOG.md index a0c3874ef..363e73d20 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/internal/daemon/worker/session/manager.go b/internal/daemon/worker/session/manager.go index 8d3f2b5e1..9c86d34d2 100644 --- a/internal/daemon/worker/session/manager.go +++ b/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 } diff --git a/internal/daemon/worker/session/manager_test.go b/internal/daemon/worker/session/manager_test.go index 46a5c1416..24ad73fbc 100644 --- a/internal/daemon/worker/session/manager_test.go +++ b/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 diff --git a/internal/daemon/worker/session/session.go b/internal/daemon/worker/session/session.go index 68be87eb0..5a94199aa 100644 --- a/internal/daemon/worker/session/session.go +++ b/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 { diff --git a/internal/tests/cluster/worker_proxy_test.go b/internal/tests/cluster/worker_proxy_test.go new file mode 100644 index 000000000..6ded71440 --- /dev/null +++ b/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) +}