perf(oidc): optimize client verification (#6999)

* fix some spelling errors

* client credential auth

* implementation of client auth

* improve error handling

* unit test command package

* unit test database package

* unit test query package

* cleanup unused tracing func

* fix integration tests

* errz to zerrors

* fix linting and import issues

* fix another linting error

* integration test with client secret

* Revert "integration test with client secret"

This reverts commit 0814ba522f.

* add integration tests

* client credentials integration test

* resolve comments

* pin oidc v3.5.0

internal/command/command.go

@@ -9,8 +9,11 @@ import (
 	"math/big"
 	"net/http"
 	"strconv"
+	"sync"
 	"time"
 
+	"github.com/zitadel/logging"
+
 	"github.com/zitadel/zitadel/internal/api/authz"
 	api_http "github.com/zitadel/zitadel/internal/api/http"
 	"github.com/zitadel/zitadel/internal/command/preparation"
@@ -37,12 +40,15 @@ import (
 	usr_repo "github.com/zitadel/zitadel/internal/repository/user"
 	usr_grant_repo "github.com/zitadel/zitadel/internal/repository/usergrant"
 	"github.com/zitadel/zitadel/internal/static"
+	"github.com/zitadel/zitadel/internal/telemetry/tracing"
 	webauthn_helper "github.com/zitadel/zitadel/internal/webauthn"
 )
 
 type Commands struct {
 	httpClient *http.Client
 
+	jobs sync.WaitGroup
+
 	checkPermission    domain.PermissionCheck
 	newCode            cryptoCodeFunc
 	newCodeWithDefault cryptoCodeWithDefaultFunc
@@ -257,3 +263,54 @@ func samlCertificateAndKeyGenerator(keySize int) func(id string) ([]byte, []byte
 		return pem.EncodeToMemory(keyBlock), pem.EncodeToMemory(certBlock), nil
 	}
 }
+
+// Close blocks until all async jobs are finished,
+// the context expires or after eventstore.PushTimeout.
+func (c *Commands) Close(ctx context.Context) error {
+	if c.eventstore.PushTimeout != 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, c.eventstore.PushTimeout)
+		defer cancel()
+	}
+
+	done := make(chan struct{})
+	go func() {
+		c.jobs.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// asyncPush attempts to push events to the eventstore in a separate Go routine.
+// This can be used to speed up request times when the outcome of the push is
+// not important for business logic but have a pure logging function.
+// For example this can be used for Secret Check Success and Failed events.
+// On push error, a log line describing the error will be emitted.
+func (c *Commands) asyncPush(ctx context.Context, cmds ...eventstore.Command) {
+	// Create a new context, as the request scoped context might get
+	// canceled before we where able to push.
+	// The eventstore has its own PushTimeout setting,
+	// so we don't need to have a context with timeout here.
+	ctx = context.WithoutCancel(ctx)
+
+	c.jobs.Add(1)
+
+	go func() {
+		defer c.jobs.Done()
+		localCtx, span := tracing.NewSpan(ctx)
+
+		_, err := c.eventstore.Push(localCtx, cmds...)
+		if err != nil {
+			for _, cmd := range cmds {
+				logging.WithError(err).Errorf("could not push event %q", cmd.Type())
+			}
+		}
+
+		span.EndWithError(err)
+	}()
+}

internal/command/command_test.go

@@ -1,11 +1,19 @@
 package command
 
 import (
+	"context"
+	"io"
+	"os"
 	"testing"
+	"time"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"golang.org/x/text/language"
 
+	"github.com/zitadel/zitadel/internal/eventstore"
 	"github.com/zitadel/zitadel/internal/i18n"
+	"github.com/zitadel/zitadel/internal/repository/user"
 )
 
 var (
@@ -18,5 +26,121 @@ var (
 
 func TestMain(m *testing.M) {
 	i18n.SupportLanguages(SupportedLanguages...)
-	m.Run()
+	os.Exit(m.Run())
+}
+
+func TestCommands_asyncPush(t *testing.T) {
+	// make sure the test terminates on deadlock
+	background := context.Background()
+	agg := user.NewAggregate("userID", "orgID")
+	cmd := user.NewMachineSecretCheckFailedEvent(background, &agg.Aggregate)
+
+	tests := []struct {
+		name         string
+		pushCtx      func() (context.Context, context.CancelFunc)
+		eventstore   func(*testing.T) *eventstore.Eventstore
+		closeCtx     func() (context.Context, context.CancelFunc)
+		wantCloseErr bool
+	}{
+		{
+			name: "push error",
+			pushCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithCancel(background)
+			},
+			eventstore: expectEventstore(
+				expectPushFailed(io.ErrClosedPipe, cmd),
+			),
+			closeCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithTimeout(background, time.Second)
+			},
+			wantCloseErr: false,
+		},
+		{
+			name: "success",
+			pushCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithCancel(background)
+			},
+			eventstore: expectEventstore(
+				expectPushSlow(time.Second/10, cmd),
+			),
+			closeCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithTimeout(background, time.Second)
+			},
+			wantCloseErr: false,
+		},
+		{
+			name: "success after push context cancels",
+			pushCtx: func() (context.Context, context.CancelFunc) {
+				ctx, cancel := context.WithCancel(background)
+				cancel()
+				return ctx, cancel
+			},
+			eventstore: expectEventstore(
+				expectPushSlow(time.Second/10, cmd),
+			),
+			closeCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithTimeout(background, time.Second)
+			},
+			wantCloseErr: false,
+		},
+		{
+			name: "success after push context timeout",
+			pushCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithTimeout(background, time.Second/100)
+			},
+			eventstore: expectEventstore(
+				expectPushSlow(time.Second/10, cmd),
+			),
+			closeCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithTimeout(background, time.Second)
+			},
+			wantCloseErr: false,
+		},
+		{
+			name: "success after push context timeout",
+			pushCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithTimeout(background, time.Second/100)
+			},
+			eventstore: expectEventstore(
+				expectPushSlow(time.Second/10, cmd),
+			),
+			closeCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithTimeout(background, time.Second)
+			},
+			wantCloseErr: false,
+		},
+		{
+			name: "close timeout error",
+			pushCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithCancel(background)
+			},
+			eventstore: expectEventstore(
+				expectPushSlow(time.Second/10, cmd),
+			),
+			closeCtx: func() (context.Context, context.CancelFunc) {
+				return context.WithTimeout(background, time.Second/100)
+			},
+			wantCloseErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := &Commands{
+				eventstore: tt.eventstore(t),
+			}
+			c.eventstore.PushTimeout = 10 * time.Second
+			pushCtx, cancel := tt.pushCtx()
+			c.asyncPush(pushCtx, cmd)
+			cancel()
+
+			closeCtx, cancel := tt.closeCtx()
+			defer cancel()
+			err := c.Close(closeCtx)
+			if tt.wantCloseErr {
+				assert.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+		})
+	}
 }

internal/command/main_test.go

@@ -95,7 +95,13 @@ func eventPusherToEvents(eventsPushes ...eventstore.Command) []*repository.Event
 
 func expectPush(commands ...eventstore.Command) expect {
 	return func(m *mock.MockRepository) {
-		m.ExpectPush(commands)
+		m.ExpectPush(commands, 0)
+	}
+}
+
+func expectPushSlow(sleep time.Duration, commands ...eventstore.Command) expect {
+	return func(m *mock.MockRepository) {
+		m.ExpectPush(commands, sleep)
 	}
 }
 

internal/command/user_machine_secret.go

@@ -3,8 +3,6 @@ package command
 import (
 	"context"
 
-	"github.com/zitadel/logging"
-
 	"github.com/zitadel/zitadel/internal/command/preparation"
 	"github.com/zitadel/zitadel/internal/crypto"
 	"github.com/zitadel/zitadel/internal/domain"
@@ -111,59 +109,12 @@ func prepareRemoveMachineSecret(a *user.Aggregate) preparation.Validation {
 	}
 }
 
-func (c *Commands) VerifyMachineSecret(ctx context.Context, userID string, resourceOwner string, secret string) (*domain.ObjectDetails, error) {
+func (c *Commands) MachineSecretCheckSucceeded(ctx context.Context, userID, resourceOwner string) {
 	agg := user.NewAggregate(userID, resourceOwner)
-	cmds, err := preparation.PrepareCommands(ctx, c.eventstore.Filter, prepareVerifyMachineSecret(agg, secret, c.codeAlg))
-	if err != nil {
-		return nil, err
-	}
-
-	events, err := c.eventstore.Push(ctx, cmds...)
-	for _, cmd := range cmds {
-		if cmd.Type() == user.MachineSecretCheckFailedType {
-			logging.OnError(err).Error("could not push event MachineSecretCheckFailed")
-			return nil, caos_errs.ThrowInvalidArgument(nil, "COMMAND-3kjh", "Errors.User.Machine.Secret.Invalid")
-		}
-	}
-	if err != nil {
-		return nil, err
-	}
-
-	return &domain.ObjectDetails{
-		Sequence:      events[len(events)-1].Sequence(),
-		EventDate:     events[len(events)-1].CreatedAt(),
-		ResourceOwner: events[len(events)-1].Aggregate().ResourceOwner,
-	}, nil
+	c.asyncPush(ctx, user.NewMachineSecretCheckSucceededEvent(ctx, &agg.Aggregate))
 }
 
-func prepareVerifyMachineSecret(a *user.Aggregate, secret string, algorithm crypto.HashAlgorithm) preparation.Validation {
-	return func() (_ preparation.CreateCommands, err error) {
-		if a.ResourceOwner == "" {
-			return nil, caos_errs.ThrowInvalidArgument(nil, "COMMAND-0qp2hus", "Errors.ResourceOwnerMissing")
-		}
-		if a.ID == "" {
-			return nil, caos_errs.ThrowInvalidArgument(nil, "COMMAND-bzosjs", "Errors.User.UserIDMissing")
-		}
-		return func(ctx context.Context, filter preparation.FilterToQueryReducer) ([]eventstore.Command, error) {
-			writeModel, err := getMachineWriteModel(ctx, a.ID, a.ResourceOwner, filter)
-			if err != nil {
-				return nil, err
-			}
-			if !isUserStateExists(writeModel.UserState) {
-				return nil, caos_errs.ThrowPreconditionFailed(nil, "COMMAND-569sh2o", "Errors.User.NotExisting")
-			}
-			if writeModel.ClientSecret == nil {
-				return nil, caos_errs.ThrowPreconditionFailed(nil, "COMMAND-x8910n", "Errors.User.Machine.Secret.NotExisting")
-			}
-			err = crypto.CompareHash(writeModel.ClientSecret, []byte(secret), algorithm)
-			if err == nil {
-				return []eventstore.Command{
-					user.NewMachineSecretCheckSucceededEvent(ctx, &a.Aggregate),
-				}, nil
-			}
-			return []eventstore.Command{
-				user.NewMachineSecretCheckFailedEvent(ctx, &a.Aggregate),
-			}, nil
-		}, nil
-	}
+func (c *Commands) MachineSecretCheckFailed(ctx context.Context, userID, resourceOwner string) {
+	agg := user.NewAggregate(userID, resourceOwner)
+	c.asyncPush(ctx, user.NewMachineSecretCheckFailedEvent(ctx, &agg.Aggregate))
 }

internal/command/user_machine_secret_test.go

@@ -3,8 +3,10 @@ package command
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/zitadel/zitadel/internal/crypto"
 	"github.com/zitadel/zitadel/internal/domain"
@@ -319,212 +321,34 @@ func TestCommandSide_RemoveMachineSecret(t *testing.T) {
 	}
 }
 
-func TestCommandSide_VerifyMachineSecret(t *testing.T) {
-	type fields struct {
-		eventstore *eventstore.Eventstore
-	}
-	type args struct {
-		ctx           context.Context
-		userID        string
-		resourceOwner string
-		secret        string
-	}
-	type res struct {
-		want *domain.ObjectDetails
-		err  func(error) bool
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		args   args
-		res    res
-	}{
-		{
-			name: "user invalid, invalid argument error userID",
-			fields: fields{
-				eventstore: eventstoreExpect(
-					t,
-				),
-			},
-			args: args{
-				ctx:           context.Background(),
-				userID:        "",
-				resourceOwner: "org1",
-			},
-			res: res{
-				err: caos_errs.IsErrorInvalidArgument,
-			},
-		},
-		{
-			name: "user invalid, invalid argument error resourceowner",
-			fields: fields{
-				eventstore: eventstoreExpect(
-					t,
-				),
-			},
-			args: args{
-				ctx:           context.Background(),
-				userID:        "user1",
-				resourceOwner: "",
-			},
-			res: res{
-				err: caos_errs.IsErrorInvalidArgument,
-			},
-		},
-		{
-			name: "user not existing, precondition error",
-			fields: fields{
-				eventstore: eventstoreExpect(
-					t,
-					expectFilter(),
-				),
-			},
-			args: args{
-				ctx:           context.Background(),
-				userID:        "user1",
-				resourceOwner: "org1",
-			},
-			res: res{
-				err: caos_errs.IsPreconditionFailed,
-			},
-		},
-		{
-			name: "user existing without secret, precondition error",
-			fields: fields{
-				eventstore: eventstoreExpect(
-					t,
-					expectFilter(
-						eventFromEventPusher(
-							user.NewMachineAddedEvent(context.Background(),
-								&user.NewAggregate("user1", "org1").Aggregate,
-								"user1",
-								"username",
-								"user",
-								false,
-								domain.OIDCTokenTypeBearer,
-							),
-						),
-					),
-				),
-			},
-			args: args{
-				ctx:           context.Background(),
-				userID:        "user1",
-				resourceOwner: "org1",
-			},
-			res: res{
-				err: caos_errs.IsPreconditionFailed,
-			},
-		},
-		{
-			name: "verify machine secret, ok",
-			fields: fields{
-				eventstore: eventstoreExpect(
-					t,
-					expectFilter(
-						eventFromEventPusher(
-							user.NewMachineAddedEvent(context.Background(),
-								&user.NewAggregate("user1", "org1").Aggregate,
-								"user1",
-								"username",
-								"user",
-								false,
-								domain.OIDCTokenTypeBearer,
-							),
-						),
-						eventFromEventPusher(
-							user.NewMachineSecretSetEvent(context.Background(),
-								&user.NewAggregate("user1", "org1").Aggregate,
-								&crypto.CryptoValue{
-									CryptoType: crypto.TypeEncryption,
-									Algorithm:  "bcrypt",
-									KeyID:      "id",
-									Crypted:    []byte("$2a$14$HxC7TAXMeowdqHdSBUfsjOUc0IGajYeApxdYl9lAYC0duZmSkgFia"),
-								},
-							),
-						),
-					),
-					expectPush(
-						user.NewMachineSecretCheckSucceededEvent(context.Background(),
-							&user.NewAggregate("user1", "org1").Aggregate,
-						),
-					),
-				),
-			},
-			args: args{
-				ctx:           context.Background(),
-				userID:        "user1",
-				resourceOwner: "org1",
-				secret:        "test",
-			},
-			res: res{
-				want: &domain.ObjectDetails{
-					ResourceOwner: "org1",
-				},
-			},
-		},
-		{
-			name: "verify machine secret, failed",
-			fields: fields{
-				eventstore: eventstoreExpect(
-					t,
-					expectFilter(
-						eventFromEventPusher(
-							user.NewMachineAddedEvent(context.Background(),
-								&user.NewAggregate("user1", "org1").Aggregate,
-								"user1",
-								"username",
-								"user",
-								false,
-								domain.OIDCTokenTypeBearer,
-							),
-						),
-						eventFromEventPusher(
-							user.NewMachineSecretSetEvent(context.Background(),
-								&user.NewAggregate("user1", "org1").Aggregate,
-								&crypto.CryptoValue{
-									CryptoType: crypto.TypeEncryption,
-									Algorithm:  "bcrypt",
-									KeyID:      "id",
-									Crypted:    []byte("$2a$14$HxC7TAXMeowdqHdSBUfsjOUc0IGajYeApxdYl9lAYC0duZmSkgFia"),
-								},
-							),
-						),
-					),
-					expectPush(
-						user.NewMachineSecretCheckFailedEvent(context.Background(),
-							&user.NewAggregate("user1", "org1").Aggregate,
-						),
-					),
-				),
-			},
-			args: args{
-				ctx:           context.Background(),
-				userID:        "user1",
-				resourceOwner: "org1",
-				secret:        "wrong",
-			},
-			res: res{
-				err: caos_errs.IsErrorInvalidArgument,
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			r := &Commands{
-				eventstore: tt.fields.eventstore,
-				codeAlg:    crypto.NewBCrypt(14),
-			}
-			got, err := r.VerifyMachineSecret(tt.args.ctx, tt.args.userID, tt.args.resourceOwner, tt.args.secret)
-			if tt.res.err == nil {
-				assert.NoError(t, err)
-			}
-			if tt.res.err != nil && !tt.res.err(err) {
-				t.Errorf("got wrong err: %v ", err)
-			}
-			if tt.res.err == nil {
-				assert.Equal(t, tt.res.want, got)
-			}
-		})
+func TestCommands_MachineSecretCheckSucceeded(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	agg := user.NewAggregate("userID", "orgID")
+	cmd := user.NewMachineSecretCheckSucceededEvent(ctx, &agg.Aggregate)
+
+	c := &Commands{
+		eventstore: eventstoreExpect(t,
+			expectPushSlow(time.Second/100, cmd),
+		),
 	}
+	c.MachineSecretCheckSucceeded(ctx, "userID", "orgID")
+	require.NoError(t, c.Close(ctx))
+}
+
+func TestCommands_MachineSecretCheckFailed(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	agg := user.NewAggregate("userID", "orgID")
+	cmd := user.NewMachineSecretCheckFailedEvent(ctx, &agg.Aggregate)
+
+	c := &Commands{
+		eventstore: eventstoreExpect(t,
+			expectPushSlow(time.Second/100, cmd),
+		),
+	}
+	c.MachineSecretCheckFailed(ctx, "userID", "orgID")
+	require.NoError(t, c.Close(ctx))
 }