bug(sessions): Fix authorizeSession race conditions in handleProxy (#2795)

* initial commit

CHANGELOG.md

@@ -23,6 +23,7 @@ Canonical reference for changes, improvements, and bugfixes for Boundary.
   plugin to exit ([PR](https://github.com/hashicorp/boundary/pull/2677))
 * data warehouse: Fix bug that caused credential dimensions to not get
     associated with session facts ([PR](https://github.com/hashicorp/boundary/pull/2787)).
+* sessions: Fix two authorizeSession race conditions in handleProxy. ([PR](https://github.com/hashicorp/boundary/pull/2795))
 
 ## 0.11.2 (2022/12/09)
 

internal/daemon/worker/session/session.go

@@ -128,6 +128,7 @@ type sess struct {
 	status      pbs.SESSIONSTATUS
 	cert        *x509.Certificate
 	sessionId   string
+	tofuToken   string
 }
 
 func newSess(client pbs.SessionServiceClient, resp *pbs.LookupSessionResponse) (*sess, error) {
@@ -187,6 +188,12 @@ func (s *sess) ApplyLocalStatus(st pbs.SESSIONSTATUS) {
 	s.status = st
 }
 
+func (s *sess) ApplyLocalTofuToken(tt string) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.tofuToken = tt
+}
+
 func (s *sess) GetLocalConnections() map[string]ConnInfo {
 	s.lock.RLock()
 	defer s.lock.RUnlock()
@@ -207,7 +214,7 @@ func (s *sess) GetLocalConnections() map[string]ConnInfo {
 func (s *sess) GetTofuToken() string {
 	s.lock.RLock()
 	defer s.lock.RUnlock()
-	return s.resp.GetTofuToken()
+	return s.tofuToken
 }
 
 func (s *sess) GetConnectionLimit() int32 {
@@ -292,6 +299,7 @@ func (s *sess) RequestActivate(ctx context.Context, tofu string) error {
 	if err != nil {
 		return err
 	}
+	s.ApplyLocalTofuToken(tofu)
 	s.ApplyLocalStatus(st)
 	return nil
 }

internal/session/repository_session.go

@@ -12,6 +12,7 @@ import (
 	"github.com/hashicorp/boundary/internal/db/timestamp"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/kms"
+	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/util"
 	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
 )
@@ -470,11 +471,75 @@ func (r *Repository) sessionAuthzSummary(ctx context.Context, sessionId string)
 	return info, nil
 }
 
+// Lookup an activated session. Must run in a transaction.
+func (r *Repository) lookupActivatedSessionTx(ctx context.Context, reader db.Reader, writer db.Writer, sessionId string,
+	tofuToken []byte, activatedSession *Session,
+) error {
+	const op = "session.(Repository).lookupActivatedSessionTx"
+	var txErr error
+	if txErr = reader.LookupById(ctx, activatedSession); txErr != nil {
+		return errors.Wrap(ctx, txErr, op, errors.WithMsg(fmt.Sprintf("failed for %s", sessionId)))
+	}
+	if txErr = decryptAndMaybeUpdateSession(ctx, r.kms, activatedSession, writer); txErr != nil {
+		return errors.Wrap(ctx, txErr, op)
+	}
+	if len(activatedSession.TofuToken) > 0 && subtle.ConstantTimeCompare(activatedSession.TofuToken, tofuToken) != 1 {
+		return errors.New(ctx, errors.TokenMismatch, op, "tofu token mismatch")
+	}
+
+	return nil
+}
+
+// Return states for an activated session. Must run in a transaction.
+func (r *Repository) fetchActivatedSessionStatesTx(ctx context.Context, reader db.Reader, sessionId string) ([]*State, error) {
+	const op = "session.(Repository).fetchActivatedSessionStatesTx"
+	var txErr error
+
+	var returnedStates []*State
+	returnedStates, txErr = fetchStates(ctx, reader, sessionId, db.WithOrder("start_time desc"))
+	if txErr != nil {
+		return nil, errors.Wrap(ctx, txErr, op)
+	}
+	return returnedStates, nil
+}
+
+// getActivatedSession is called if there was a duplicate attempt to activate a session
+// It validates the tofu token matches and returns the session
+func (r *Repository) getActivatedSession(ctx context.Context, sessionId string, tofuToken []byte) (*Session, []*State, error) {
+	const op = "session.(Repository).getActivatedSession"
+
+	activatedSession := AllocSession()
+	activatedSession.PublicId = sessionId
+	var returnedStates []*State
+	_, err := r.writer.DoTx(
+		ctx,
+		db.StdRetryCnt,
+		db.ExpBackoff{},
+		func(reader db.Reader, w db.Writer) error {
+			err := r.lookupActivatedSessionTx(ctx, reader, w, sessionId, tofuToken, &activatedSession)
+			if err != nil {
+				return errors.Wrap(ctx, err, op)
+			}
+			returnedStates, err = r.fetchActivatedSessionStatesTx(ctx, reader, sessionId)
+			if err != nil {
+				return errors.Wrap(ctx, err, op)
+			}
+			return nil
+		},
+	)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+	return &activatedSession, returnedStates, nil
+}
+
 // ActivateSession will activate the session and is called by a worker after
 // authenticating the session. The session must be in a "pending" state to be
 // activated. States are ordered by start time descending. Returns an
 // InvalidSessionState error code if a connection cannot be made because the session
 // was canceled or terminated.
+// If ActivateSession receives duplicate requests for the same session, it will return the
+// already active session if the tofu token is correct
 func (r *Repository) ActivateSession(ctx context.Context, sessionId string, sessionVersion uint32, tofuToken []byte) (*Session, []*State, error) {
 	const op = "session.(Repository).ActivateSession"
 	if sessionId == "" {
@@ -507,15 +572,10 @@ func (r *Repository) ActivateSession(ctx context.Context, sessionId string, sess
 			}
 			foundSession := AllocSession()
 			foundSession.PublicId = sessionId
-			if err := reader.LookupById(ctx, &foundSession); err != nil {
-				return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("failed for %s", sessionId)))
-			}
-			if err := decryptAndMaybeUpdateSession(ctx, r.kms, &foundSession, w); err != nil {
+			err = r.lookupActivatedSessionTx(ctx, reader, w, sessionId, tofuToken, &foundSession)
+			if err != nil {
 				return errors.Wrap(ctx, err, op)
 			}
-			if len(foundSession.TofuToken) > 0 && subtle.ConstantTimeCompare(foundSession.TofuToken, tofuToken) != 1 {
-				return errors.New(ctx, errors.TokenMismatch, op, "tofu token mismatch")
-			}
 
 			updatedSession.TofuToken = tofuToken
 			sessionWrapper, err := r.kms.GetWrapper(ctx, foundSession.ProjectId, kms.KeyPurposeSessions)
@@ -534,7 +594,7 @@ func (r *Repository) ActivateSession(ctx context.Context, sessionId string, sess
 				return errors.New(ctx, errors.MultipleRecords, op, "more than 1 resource would have been updated")
 			}
 
-			returnedStates, err = fetchStates(ctx, reader, sessionId, db.WithOrder("start_time desc"))
+			returnedStates, err = r.fetchActivatedSessionStatesTx(ctx, reader, sessionId)
 			if err != nil {
 				return errors.Wrap(ctx, err, op)
 			}
@@ -542,6 +602,11 @@ func (r *Repository) ActivateSession(ctx context.Context, sessionId string, sess
 		},
 	)
 	if err != nil {
+		// If this was a duplicate activation attempt, return existing session if the tofu token matches
+		if errors.IsUniqueError(err) {
+			event.WriteSysEvent(ctx, op, fmt.Sprintf("ignoring duplicate session activation attempt for session %v", sessionId))
+			return r.getActivatedSession(ctx, sessionId, tofuToken)
+		}
 		return nil, nil, errors.Wrap(ctx, err, op)
 	}
 	return &updatedSession, returnedStates, nil