CVE-2024-23820 (Go, published 2024-01-26T17:15Z)
Description: value: OpenFGA, an authorization/permission engine, is vulnerable to a denial of service attack in versions prior to 1.4.3. In some scenarios that depend on the model and tuples used, a call to `ListObjects` may not  release memory properly. So when a sufficiently high number of those calls are executed, the OpenFGA server can create an `out of memory` error and terminate. Version 1.4.3 contains a patch for this issue.
Severity: nan (CVSS v3 base score: 6.5)
Weakness: CWE-401 \u2014 Missing Release of Memory after Effective Lifetime
The product does not sufficiently track and release allocated memory after it has been used, which slowly consumes remaining memory.
Repository: https://github.com/openfga/openfga
Files changed:
pkg/server/commands/list_objects.go
pkg/server/commands/reverseexpand/reverse_expand.go
pkg/server/commands/reverseexpand/reverse_expand_test.go
pkg/server/server_test.go
pkg/server/test/list_objects.go
pkg/server/test/reverse_expand.go
pkg/testfixtures/storage/mysql.go
pkg/testfixtures/storage/postgres.go
tests/listobjects/listobjects_test.go
Commit message: fix: handle ReverseExpand channel closure correctly (#1315)



Vulnerable code:
		reverseExpandQuery := reverseexpand.NewReverseExpandQuery(q.datastore, typesys,
			reverseExpandQuery.Execute(cancelCtx, &reverseexpand.ReverseExpandRequest{
		for res := range reverseExpandResultsChan {
			if res.Err != nil {
				err := res.Err

				if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
					err = serverErrors.AuthorizationModelResolutionTooComplex
				resultsChan <- ListObjectsResult{Err: err}
				break
			}

			if !(maxResults == 0) && objectsFound.Load() >= maxResults {
				break
			}
			if res.ResultStatus == reverseexpand.NoFurtherEvalStatus {
				noFurtherEvalRequiredCounter.Inc()
				trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
				continue
			}
			furtherEvalRequiredCounter.Inc()

			wg.Add(1)
			go func(res *reverseexpand.ReverseExpandResult) {
				defer func() {
					<-concurrencyLimiterCh
					wg.Done()
				}()

				concurrencyLimiterCh <- struct{}{}

				resp, err := checkResolver.ResolveCheck(ctx, &graph.ResolveCheckRequest{
					StoreID:              req.GetStoreId(),
					AuthorizationModelID: req.GetAuthorizationModelId(),
					TupleKey:             tuple.NewTupleKey(res.Object, req.GetRelation(), req.GetUser()),
					ContextualTuples:     req.GetContextualTuples().GetTupleKeys(),
					Context:              req.GetContext(),
					ResolutionMetadata: &graph.ResolutionMetadata{
						Depth: q.resolveNodeLimit,
					},
				})
				if err != nil {
					if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
						resultsChan <- ListObjectsResult{Err: serverErrors.AuthorizationModelResolutionTooComplex}
					resultsChan <- ListObjectsResult{Err: err}
					return
				}
				atomic.AddUint32(resolutionMetadata.QueryCount, resp.GetResolutionMetadata().DatastoreQueryCount)
				if resp.Allowed {
					trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
			}(res)
	for {
		select {
		case <-timeoutCtx.Done():
			q.logger.WarnWithContext(
				ctx, fmt.Sprintf("list objects timeout after %s", q.listObjectsDeadline.String()),
			)
			return &ListObjectsResponse{
				Objects:            objects,
				ResolutionMetadata: *resolutionMetadata,
			}, nil

		case result, channelOpen := <-resultsChan:
			if result.Err != nil {
				if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
					return nil, result.Err
				}
				if errors.Is(result.Err, condition.ErrEvaluationFailed) {
					errs = multierror.Append(errs, result.Err)
					continue
				}

				if errors.Is(result.Err, context.Canceled) || errors.Is(result.Err, context.DeadlineExceeded) {
					continue
				}

				return nil, serverErrors.HandleError("", result.Err)
			if !channelOpen {
				if len(objects) < int(maxResults) && errs.ErrorOrNil() != nil {
					return nil, errs
				}
				return &ListObjectsResponse{
					Objects:            objects,
					ResolutionMetadata: *resolutionMetadata,
				}, nil
			objects = append(objects, result.ObjectID)
	for {
		select {
		case <-timeoutCtx.Done():
			q.logger.WarnWithContext(
				ctx, fmt.Sprintf("list objects timeout after %s", q.listObjectsDeadline.String()),
			)
			return resolutionMetadata, nil

		case result, channelOpen := <-resultsChan:
			if !channelOpen {
				// Channel closed! No more results.
				return resolutionMetadata, nil
			if result.Err != nil {
				if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
					return nil, result.Err
				}

				if errors.Is(result.Err, condition.ErrEvaluationFailed) {
					return nil, serverErrors.ValidationError(result.Err)
				}

				return nil, serverErrors.HandleError("", result.Err)
			if err := srv.Send(&openfgav1.StreamedListObjectsResponse{
				Object: result.ObjectID,
			}); err != nil {
				return nil, serverErrors.NewInternalError("", err)
			}
	Err          error
// Execute yields all the objects of the provided objectType that the given user has, possibly, a specific relation with
// and sends those objects to resultChan. It MUST guarantee no duplicate objects sent.
// If an error is encountered before resolving all objects: the provided channel will NOT be closed and
// - if the error is context cancellation or deadline: Execute may send the error through the channel
// - otherwise: Execute will send the error through the channel
// If no errors, Execute will yield all of the objects on the provided channel and then close the channel
// to signal that it is done.
) {
		select {
		case <-ctx.Done():
			return
		case resultChan <- &ReverseExpandResult{Err: err}:
			return
		}

	done := make(chan struct{})
		reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
		done <- struct{}{}
		require.NoError(t, res.Err)
	case <-done:
		t.Log("OK!")
		return
		require.FailNow(t, "timed out")
	done := make(chan struct{})
		reverseExpandQuery.Execute(timeoutCtx, &ReverseExpandRequest{
		done <- struct{}{}
	case res, open := <-resultChan:
		if open {
			require.Error(t, res.Err)
		} else {
			require.Nil(t, res)
		<-done
	case <-done:
		// OK!

	done := make(chan struct{})
		reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
		done <- struct{}{}
	go func() {
		<-resultChan
		// We want to read resultChan twice because Next() will fail after first read
		<-resultChan
		cancelFunc()
	}()
	select {
	case <-done:
		return
	case <-time.After(30 * time.Millisecond):
		require.FailNow(t, "timed out")
		//goleak.VerifyNone(t)
		//goleak.VerifyNone(t)
		//goleak.VerifyNone(t)
		//goleak.VerifyNone(t)
		//goleak.VerifyNone(t)
					for x := range server.channel {
						streamedObjectIds = append(streamedObjectIds, x)

					done <- struct{}{}
			require := require.New(t)

			require.NoError(err)
			require.NoError(err)
				reverseExpandQuery.Execute(timeoutCtx, test.request, resultChan, resolutionMetadata)
			reverseExpandErrCh := make(chan error)
			go func() {
				for result := range resultChan {
					if result.Err != nil {
						reverseExpandErrCh <- result.Err

					results = append(results, result)

				reverseExpandErrCh <- nil
			}()

			select {
			case <-timeoutCtx.Done():
				require.FailNow("timed out waiting for response")
			case err := <-reverseExpandErrCh:
				require.ErrorIs(err, test.expectedError)
			}

			if test.expectedError == nil {
				require.ElementsMatch(test.expectedResult, results)
				require.Equal(test.expectedDSQueryCount, *resolutionMetadata.QueryCount)
	"database/sql"
	var db *sql.DB
			db, err = goose.OpenDBWithDriver("mysql", uri)
			if err != nil {
				return err
			}
	err = db.Close()
	require.NoError(t, err)

	err = db.Close()
	require.NoError(t, err)

	// uncomment in https://github.com/openfga/openfga/pull/1315
	// defer goleak.VerifyNone(t)

Fixed code:
// evaluate fires of evaluation of the ListObjects query by delegating to
// [[reverseexpand.ReverseExpand#Execute]] and resolving the results yielded
// from it. If any results yielded by reverse expansion require further eval,
// then these results get dispatched to Check to resolve the residual outcome.
//
// The resultsChan is **always** closed by evaluate when it is done with its work,
// which is either when all results have been yielded, the deadline has been met,
// or some other terminal error case has occurred.
		reverseExpandQuery := reverseexpand.NewReverseExpandQuery(
			q.datastore,
			typesys,
		errChan := make(chan error, 1)

			err := reverseExpandQuery.Execute(cancelCtx, &reverseexpand.ReverseExpandRequest{
			if err != nil {
				errChan <- err
			}
	ConsumerReadLoop:
		for {
			select {
			case <-ctx.Done():
				break ConsumerReadLoop
			case res, channelOpen := <-reverseExpandResultsChan:
				if !channelOpen {
					break ConsumerReadLoop
				if !(maxResults == 0) && objectsFound.Load() >= maxResults {
					break ConsumerReadLoop
				}
				if res.ResultStatus == reverseexpand.NoFurtherEvalStatus {
					noFurtherEvalRequiredCounter.Inc()
					trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
					continue
				}
				furtherEvalRequiredCounter.Inc()

				wg.Add(1)
				go func(res *reverseexpand.ReverseExpandResult) {
					defer func() {
						<-concurrencyLimiterCh
						wg.Done()
					}()

					concurrencyLimiterCh <- struct{}{}

					resp, err := checkResolver.ResolveCheck(ctx, &graph.ResolveCheckRequest{
						StoreID:              req.GetStoreId(),
						AuthorizationModelID: req.GetAuthorizationModelId(),
						TupleKey:             tuple.NewTupleKey(res.Object, req.GetRelation(), req.GetUser()),
						ContextualTuples:     req.GetContextualTuples().GetTupleKeys(),
						Context:              req.GetContext(),
						ResolutionMetadata: &graph.ResolutionMetadata{
							Depth: q.resolveNodeLimit,
						},
					})
					if err != nil {
						if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
							resultsChan <- ListObjectsResult{Err: serverErrors.AuthorizationModelResolutionTooComplex}
							return
						}

						resultsChan <- ListObjectsResult{Err: err}
					atomic.AddUint32(resolutionMetadata.QueryCount, resp.GetResolutionMetadata().DatastoreQueryCount)
					if resp.Allowed {
						trySendObject(res.Object, &objectsFound, maxResults, resultsChan)
					}
				}(res)
			case err := <-errChan:
				if errors.Is(err, graph.ErrResolutionDepthExceeded) || errors.Is(err, graph.ErrCycleDetected) {
					err = serverErrors.AuthorizationModelResolutionTooComplex

				resultsChan <- ListObjectsResult{Err: err}
				break ConsumerReadLoop
			}
	for result := range resultsChan {
		if result.Err != nil {
			if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
				return nil, result.Err
			if errors.Is(result.Err, condition.ErrEvaluationFailed) {
				errs = multierror.Append(errs, result.Err)
				continue
			}
			if errors.Is(result.Err, context.Canceled) || errors.Is(result.Err, context.DeadlineExceeded) {
				continue

			return nil, serverErrors.HandleError("", result.Err)

		objects = append(objects, result.ObjectID)

	if len(objects) < int(maxResults) && errs.ErrorOrNil() != nil {
		return nil, errs
	}

	return &ListObjectsResponse{
		Objects:            objects,
		ResolutionMetadata: *resolutionMetadata,
	}, nil
	for result := range resultsChan {
		if result.Err != nil {
			if errors.Is(result.Err, serverErrors.AuthorizationModelResolutionTooComplex) {
				return nil, result.Err
			if errors.Is(result.Err, condition.ErrEvaluationFailed) {
				return nil, serverErrors.ValidationError(result.Err)
			return nil, serverErrors.HandleError("", result.Err)
		}

		if err := srv.Send(&openfgav1.StreamedListObjectsResponse{
			Object: result.ObjectID,
		}); err != nil {
			return nil, serverErrors.NewInternalError("", err)

	return resolutionMetadata, nil
// Execute yields all the objects of the provided objectType that the
// given user possibly has, a specific relation with and sends those
// objects to resultChan. It MUST guarantee no duplicate objects sent.
// This function respects context timeouts and cancellations. If an
// error is encountered (e.g. context timeout) before resolving all
// objects, then the provided channel will NOT be closed, and it will
// send the error through the channel.
//
// If no errors occur, then Execute will yield all of the objects on
// the provided channel and then close the channel to signal that it
// is done.
) error {
		return err
	return nil
	"github.com/openfga/openfga/pkg/storage/memory"
func TestReverseExpandResultChannelClosed(t *testing.T) {
	defer goleak.VerifyNone(t)

	store := ulid.Make().String()

	model := testutils.MustTransformDSLToProtoWithID(`model
  schema 1.1
type user
type document
  relations
	define viewer: [user]`)

	typeSystem := typesystem.New(model)
	mockController := gomock.NewController(t)
	defer mockController.Finish()

	var tuples []*openfgav1.Tuple

	mockDatastore := mocks.NewMockOpenFGADatastore(mockController)
	mockDatastore.EXPECT().ReadStartingWithUser(gomock.Any(), store, gomock.Any()).
		Times(1).
		DoAndReturn(func(_ context.Context, _ string, _ storage.ReadStartingWithUserFilter) (storage.TupleIterator, error) {
			iterator := storage.NewStaticTupleIterator(tuples)
			return iterator, nil
		})

	ctx := context.Background()

	resultChan := make(chan *ReverseExpandResult)
	errChan := make(chan error, 1)

	// process query in one goroutine, but it will be cancelled almost right away
	go func() {
		reverseExpandQuery := NewReverseExpandQuery(mockDatastore, typeSystem)
		t.Logf("before execute reverse expand")
		err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
			StoreID:    store,
			ObjectType: "document",
			Relation:   "viewer",
			User: &UserRefObject{
				Object: &openfgav1.Object{
					Type: "user",
					Id:   "maria",
				},
			},
			ContextualTuples: []*openfgav1.TupleKey{},
		}, resultChan, NewResolutionMetadata())
		t.Logf("after execute reverse expand")

		if err != nil {
			errChan <- err
		}
	}()

	select {
	case _, open := <-resultChan:
		if open {
			require.FailNow(t, "expected immediate closure of result channel")
		}
	case err := <-errChan:
		require.FailNow(t, "unexpected error received on error channel :%v", err)
	case <-time.After(30 * time.Millisecond):
		require.FailNow(t, "unexpected timeout on channel receive, expected receive on error channel")
	}
}

	errChan := make(chan error, 1)
		err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{

		if err != nil {
			errChan <- err
		}
	case err := <-errChan:
		require.Error(t, err)
		require.FailNow(t, "unexpected timeout on channel receive, expected receive on error channel")
	errChan := make(chan error, 1)
		err := reverseExpandQuery.Execute(timeoutCtx, &ReverseExpandRequest{

		if err != nil {
			errChan <- err
		}
	case _, open := <-resultChan:
		if !open {
			require.FailNow(t, "unexpected closure of result channel")
	case err := <-errChan:
		require.Error(t, err)
	case <-time.After(1 * time.Second):
		require.FailNow(t, "unexpected timeout encountered, expected other receive")

	defer cancelFunc()
	errChan := make(chan error, 1)
		err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
		if err != nil {
			errChan <- err
		}
ConsumerLoop:
	for {
		select {
		case _, open := <-resultChan:
			if !open {
				require.FailNow(t, "unexpected closure of result channel")
			}

			cancelFunc()
		case err := <-errChan:
			require.Error(t, err)
			break ConsumerLoop
		case <-time.After(30 * time.Millisecond):
			require.FailNow(t, "unexpected timeout waiting for channel receive, expected an error on the error channel")
		}
	}
}
func TestReverseExpandSendsAllErrorsThroughChannel(t *testing.T) {
	defer goleak.VerifyNone(t)

	store := ulid.Make().String()

	model := testutils.MustTransformDSLToProtoWithID(`model
  schema 1.1
type user
type document
  relations
    define viewer: [user]`)

	mockDatastore := mocks.NewMockSlowDataStorage(memory.New(), 1*time.Second)

	for i := 0; i < 50; i++ {
		t.Logf("iteration %d", i)
		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(1*time.Nanosecond))
		t.Cleanup(func() {
			cancel()
		})

		resultChan := make(chan *ReverseExpandResult)
		errChan := make(chan error, 1)

		go func() {
			reverseExpandQuery := NewReverseExpandQuery(mockDatastore, typesystem.New(model))
			t.Logf("before produce")
			err := reverseExpandQuery.Execute(ctx, &ReverseExpandRequest{
				StoreID:    store,
				ObjectType: "document",
				Relation:   "viewer",
				User: &UserRefObject{
					Object: &openfgav1.Object{
						Type: "user",
						Id:   "maria",
					},
				},
				ContextualTuples: []*openfgav1.TupleKey{},
			}, resultChan, NewResolutionMetadata())
			t.Logf("after produce")

			if err != nil {
				errChan <- err
			}
		}()

		select {
		case _, channelOpen := <-resultChan:
			if !channelOpen {
				require.FailNow(t, "unexpected closure of result channel")
			}
		case err := <-errChan:
			require.Error(t, err)
		case <-time.After(3 * time.Second):
			require.FailNow(t, "unexpected timeout waiting for channel receive, expected an error on the error channel")
		}
	"go.uber.org/goleak"
		goleak.VerifyNone(t)
		goleak.VerifyNone(t)
		goleak.VerifyNone(t)
		goleak.VerifyNone(t)
		goleak.VerifyNone(t)
					for {
						select {
						case objectID, open := <-server.channel:
							if !open {
								done <- struct{}{}
								return
							}

							streamedObjectIds = append(streamedObjectIds, objectID)

						// for tests whose deadline is sooner than the latency of the storage layer
						case <-time.After(test.readTuplesDelay + 1*time.Second):
							done <- struct{}{}
							return
						}
			require.NoError(t, err)
			require.NoError(t, err)
			reverseExpandErrCh := make(chan error, 1)
				err := reverseExpandQuery.Execute(timeoutCtx, test.request, resultChan, resolutionMetadata)
				if err != nil {
					reverseExpandErrCh <- err
					t.Logf("sent err %s", err)
				}

			for {
				select {
				case err := <-reverseExpandErrCh:
					require.ErrorIs(t, err, test.expectedError)
					return
				case res, channelOpen := <-resultChan:
					if !channelOpen {
						t.Log("channel closed")
						if test.expectedError == nil {
							require.ElementsMatch(t, test.expectedResult, results)
							require.Equal(t, test.expectedDSQueryCount, *resolutionMetadata.QueryCount)
						} else {
							require.FailNow(t, "expected an error, got none")
						}
					} else {
						t.Logf("appending result %s", res.Object)
						results = append(results, res)
	db, err := goose.OpenDBWithDriver("mysql", uri)
	require.NoError(t, err)
	defer db.Close()
	defer db.Close()
	"go.uber.org/goleak"
	defer goleak.VerifyNone(t)