diff --git a/cmd/defaults.yaml b/cmd/defaults.yaml index afd4215ca..d46509599 100644 --- a/cmd/defaults.yaml +++ b/cmd/defaults.yaml @@ -625,8 +625,22 @@ OIDC: DefaultLogoutURLV2: "/ui/v2/login/logout?post_logout_redirect=" # ZITADEL_OIDC_DEFAULTLOGOUTURLV2 # Internal cache age for public keys to speed up validations (e.g. id_token_hints) on the authorization endpoint. PublicKeyCacheMaxAge: 24h # ZITADEL_OIDC_PUBLICKEYCACHEMAXAGE - # Lifetime of the token used to notify clients through OIDC back-channel logout when the feature is enabled and registered on the client. + # Lifetime of the token used to notify clients through OIDC back-channel logout. + # Deprecated: use BackChannelLogout.TokenLifetime instead DefaultBackChannelLogoutLifetime: 15m # ZITADEL_OIDC_DEFAULTBACKCHANNELLOGOUTLIFETIME + BackChannelLogout: + # The amount of workers processing the back-channel logout notifications. + # If set to 0, no notifications will be handled. This can be useful when running in + # multi binary / pod setup and allowing only certain executables to process the events. + Workers: 1 # ZITADEL_OIDC_BACKCHANNELLOGOUT_WORKERS + # The maximum duration a job can do it's work before it is considered as failed. + TransactionDuration: 10s # ZITADEL_OIDC_BACKCHANNELLOGOUT_TRANSACTIONDURATION + # Automatically cancel the notification after the amount of failed attempts + MaxAttempts: 3 # ZITADEL_OIDC_BACKCHANNELLOGOUT_MAXATTEMPTS + # Automatically cancel the notification if it cannot be handled within a specific time + MaxTtl: 5m # ZITADEL_OIDC_BACKCHANNELLOGOUT_MAXTTL + # Lifetime of the token used to notify clients through OIDC back-channel logout. + TokenLifetime: 15m # ZITADEL_OIDC_BACKCHANNELLOGOUT_TOKENLIFETIME SAML: DefaultLoginURLV2: "/ui/v2/login/login?samlRequest=" # ZITADEL_SAML_DEFAULTLOGINURLV2 diff --git a/cmd/mirror/projections.go b/cmd/mirror/projections.go index 6baa1b88f..5a2b8cda8 100644 --- a/cmd/mirror/projections.go +++ b/cmd/mirror/projections.go @@ -216,6 +216,7 @@ func projections( config.Projections.Customizations["backchannel"], config.Projections.Customizations["telemetry"], config.Notifications, + config.OIDC.BackChannelLogoutConfig(), *config.Telemetry, config.ExternalDomain, config.ExternalPort, @@ -228,8 +229,6 @@ func projections( keys.User, keys.SMTP, keys.SMS, - keys.OIDC, - config.OIDC.DefaultBackChannelLogoutLifetime, nil, ) diff --git a/cmd/setup/setup.go b/cmd/setup/setup.go index 9b0fe40ce..ca9196a2c 100644 --- a/cmd/setup/setup.go +++ b/cmd/setup/setup.go @@ -589,6 +589,7 @@ func startCommandsQueries( config.Projections.Customizations["backchannel"], config.Projections.Customizations["telemetry"], config.Notifications, + config.OIDC.BackChannelLogoutConfig(), *config.Telemetry, config.ExternalDomain, config.ExternalPort, @@ -601,8 +602,6 @@ func startCommandsQueries( keys.User, keys.SMTP, keys.SMS, - keys.OIDC, - config.OIDC.DefaultBackChannelLogoutLifetime, q, ) diff --git a/cmd/start/start.go b/cmd/start/start.go index 8b66bfed4..2102add32 100644 --- a/cmd/start/start.go +++ b/cmd/start/start.go @@ -324,6 +324,7 @@ func startZitadel(ctx context.Context, config *Config, masterKey string, server config.Projections.Customizations["backchannel"], config.Projections.Customizations["telemetry"], config.Notifications, + config.OIDC.BackChannelLogoutConfig(), *config.Telemetry, config.ExternalDomain, config.ExternalPort, @@ -336,8 +337,6 @@ func startZitadel(ctx context.Context, config *Config, masterKey string, server keys.User, keys.SMTP, keys.SMS, - keys.OIDC, - config.OIDC.DefaultBackChannelLogoutLifetime, q, ) notification.Start(ctx) diff --git a/internal/api/oidc/op.go b/internal/api/oidc/op.go index 504e2ba01..dae87964d 100644 --- a/internal/api/oidc/op.go +++ b/internal/api/oidc/op.go @@ -21,6 +21,7 @@ import ( "github.com/zitadel/zitadel/internal/crypto" "github.com/zitadel/zitadel/internal/domain/federatedlogout" "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/notification/handlers" "github.com/zitadel/zitadel/internal/query" "github.com/zitadel/zitadel/internal/zerrors" ) @@ -42,6 +43,16 @@ type Config struct { DefaultLogoutURLV2 string PublicKeyCacheMaxAge time.Duration DefaultBackChannelLogoutLifetime time.Duration + BackChannelLogout handlers.BackChannelLogoutWorkerConfig +} + +// BackChannelLogoutConfig returns the BackChannelLogoutWorkerConfig and takes the deprecated TokenLifetime into account. +func (c *Config) BackChannelLogoutConfig() *handlers.BackChannelLogoutWorkerConfig { + if c.DefaultBackChannelLogoutLifetime == 0 { + return &c.BackChannelLogout + } + c.BackChannelLogout.TokenLifetime = c.DefaultBackChannelLogoutLifetime + return &c.BackChannelLogout } type EndpointConfig struct { diff --git a/internal/notification/backchannel/back_channel_logout.go b/internal/notification/backchannel/back_channel_logout.go new file mode 100644 index 000000000..c519d35f3 --- /dev/null +++ b/internal/notification/backchannel/back_channel_logout.go @@ -0,0 +1,23 @@ +package backchannel + +import "github.com/zitadel/zitadel/internal/eventstore" + +const ( + QueueName = "back_channel_logout" +) + +type LogoutRequest struct { + Aggregate *eventstore.Aggregate + SessionID string + TriggeredAtOrigin string + TriggeringEventType eventstore.EventType + TokenID string + UserID string + OIDCSessionID string + ClientID string + BackChannelLogoutURI string +} + +func (l *LogoutRequest) Kind() string { + return "back_channel_logout_request" +} diff --git a/internal/notification/handlers/back_channel_logout.go b/internal/notification/handlers/back_channel_logout.go index 6f6847939..6437135e0 100644 --- a/internal/notification/handlers/back_channel_logout.go +++ b/internal/notification/handlers/back_channel_logout.go @@ -2,25 +2,15 @@ package handlers import ( "context" - "errors" "slices" - "sync" - "time" - - "github.com/zitadel/oidc/v3/pkg/crypto" - "github.com/zitadel/oidc/v3/pkg/oidc" "github.com/zitadel/zitadel/internal/api/authz" - http_utils "github.com/zitadel/zitadel/internal/api/http" - "github.com/zitadel/zitadel/internal/api/oidc/sign" - "github.com/zitadel/zitadel/internal/command" - zcrypto "github.com/zitadel/zitadel/internal/crypto" + http_util "github.com/zitadel/zitadel/internal/api/http" "github.com/zitadel/zitadel/internal/eventstore" "github.com/zitadel/zitadel/internal/eventstore/handler/v2" - "github.com/zitadel/zitadel/internal/id" - "github.com/zitadel/zitadel/internal/notification/channels/set" + "github.com/zitadel/zitadel/internal/notification/backchannel" _ "github.com/zitadel/zitadel/internal/notification/statik" - "github.com/zitadel/zitadel/internal/notification/types" + "github.com/zitadel/zitadel/internal/queue" "github.com/zitadel/zitadel/internal/repository/session" "github.com/zitadel/zitadel/internal/repository/sessionlogout" "github.com/zitadel/zitadel/internal/repository/user" @@ -32,33 +22,22 @@ const ( ) type backChannelLogoutNotifier struct { - commands *command.Commands - queries *NotificationQueries - eventstore *eventstore.Eventstore - keyEncryptionAlg zcrypto.EncryptionAlgorithm - channels types.ChannelChains - idGenerator id.Generator - tokenLifetime time.Duration + queries *NotificationQueries + queue Queue + maxAttempts uint8 } func NewBackChannelLogoutNotifier( ctx context.Context, config handler.Config, - commands *command.Commands, queries *NotificationQueries, - es *eventstore.Eventstore, - keyEncryptionAlg zcrypto.EncryptionAlgorithm, - channels types.ChannelChains, - tokenLifetime time.Duration, + queue Queue, + maxAttempts uint8, ) *handler.Handler { return handler.NewHandler(ctx, &config, &backChannelLogoutNotifier{ - commands: commands, - queries: queries, - eventstore: es, - keyEncryptionAlg: keyEncryptionAlg, - channels: channels, - tokenLifetime: tokenLifetime, - idGenerator: id.SonyFlakeGenerator(), + queries: queries, + queue: queue, + maxAttempts: maxAttempts, }) } @@ -106,7 +85,20 @@ func (u *backChannelLogoutNotifier) reduceUserSignedOut(event eventstore.Event) if e.SessionID == "" { return nil } - return u.terminateSession(ctx, e.SessionID, e) + ctx, err = u.queries.Origin(ctx, e) + if err != nil { + return err + } + return u.queue.Insert(ctx, + &backchannel.LogoutRequest{ + Aggregate: e.Aggregate(), + SessionID: e.SessionID, + TriggeredAtOrigin: http_util.DomainContext(ctx).Origin(), + TriggeringEventType: event.Type(), + }, + queue.WithQueueName(backchannel.QueueName), + queue.WithMaxAttempts(u.maxAttempts), + ) }), nil } @@ -124,7 +116,20 @@ func (u *backChannelLogoutNotifier) reduceSessionTerminated(event eventstore.Eve if !authz.GetFeatures(ctx).EnableBackChannelLogout { return nil } - return u.terminateSession(ctx, e.Aggregate().ID, e) + ctx, err = u.queries.Origin(ctx, e) + if err != nil { + return err + } + return u.queue.Insert(ctx, + &backchannel.LogoutRequest{ + Aggregate: e.Aggregate(), + SessionID: e.Aggregate().ID, + TriggeredAtOrigin: http_util.DomainContext(ctx).Origin(), + TriggeringEventType: event.Type(), + }, + queue.WithQueueName(backchannel.QueueName), + queue.WithMaxAttempts(u.maxAttempts), + ) }), nil } @@ -135,74 +140,6 @@ type backChannelLogoutSession struct { sessions []backChannelLogoutOIDCSessions } -func (u *backChannelLogoutNotifier) terminateSession(ctx context.Context, id string, e eventstore.Event) error { - sessions := &backChannelLogoutSession{sessionID: id} - err := u.eventstore.FilterToQueryReducer(ctx, sessions) - if err != nil { - return err - } - - ctx, err = u.queries.Origin(ctx, e) - if err != nil { - return err - } - - getSigner := sign.GetSignerOnce(u.queries.GetActiveSigningWebKey) - - var wg sync.WaitGroup - wg.Add(len(sessions.sessions)) - errs := make([]error, 0, len(sessions.sessions)) - for _, oidcSession := range sessions.sessions { - go func(oidcSession *backChannelLogoutOIDCSessions) { - defer wg.Done() - err := u.sendLogoutToken(ctx, oidcSession, e, getSigner) - if err != nil { - errs = append(errs, err) - return - } - err = u.commands.BackChannelLogoutSent(ctx, oidcSession.SessionID, oidcSession.OIDCSessionID, e.Aggregate().InstanceID) - if err != nil { - errs = append(errs, err) - } - }(&oidcSession) - } - wg.Wait() - return errors.Join(errs...) -} - -func (u *backChannelLogoutNotifier) sendLogoutToken(ctx context.Context, oidcSession *backChannelLogoutOIDCSessions, e eventstore.Event, getSigner sign.SignerFunc) error { - token, err := u.logoutToken(ctx, oidcSession, getSigner) - if err != nil { - return err - } - err = types.SendSecurityTokenEvent(ctx, set.Config{CallURL: oidcSession.BackChannelLogoutURI}, u.channels, &LogoutTokenMessage{LogoutToken: token}, e.Type()).WithoutTemplate() - if err != nil { - return err - } - return nil -} - -func (u *backChannelLogoutNotifier) logoutToken(ctx context.Context, oidcSession *backChannelLogoutOIDCSessions, getSigner sign.SignerFunc) (string, error) { - jwtID, err := u.idGenerator.Next() - if err != nil { - return "", err - } - token := oidc.NewLogoutTokenClaims( - http_utils.DomainContext(ctx).Origin(), - oidcSession.UserID, - oidc.Audience{oidcSession.ClientID}, - time.Now().Add(u.tokenLifetime), - jwtID, - oidcSession.SessionID, - time.Second, - ) - signer, _, err := getSigner(ctx) - if err != nil { - return "", err - } - return crypto.Sign(token, signer) -} - type LogoutTokenMessage struct { LogoutToken string `schema:"logout_token"` } diff --git a/internal/notification/handlers/back_channel_logout_worker.go b/internal/notification/handlers/back_channel_logout_worker.go new file mode 100644 index 000000000..bc469060f --- /dev/null +++ b/internal/notification/handlers/back_channel_logout_worker.go @@ -0,0 +1,160 @@ +package handlers + +import ( + "context" + "errors" + "time" + + "github.com/riverqueue/river" + "github.com/zitadel/oidc/v3/pkg/crypto" + "github.com/zitadel/oidc/v3/pkg/oidc" + + "github.com/zitadel/zitadel/internal/api/oidc/sign" + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/id" + "github.com/zitadel/zitadel/internal/notification/backchannel" + "github.com/zitadel/zitadel/internal/notification/channels" + "github.com/zitadel/zitadel/internal/notification/channels/set" + "github.com/zitadel/zitadel/internal/notification/types" + "github.com/zitadel/zitadel/internal/queue" +) + +type BackChannelLogoutWorker struct { + river.WorkerDefaults[*backchannel.LogoutRequest] + + commands Commands + queries *NotificationQueries + eventstore *eventstore.Eventstore + queue Queue + channels types.ChannelChains + config *BackChannelLogoutWorkerConfig + now nowFunc + idGenerator id.Generator +} + +// Timeout implements the Timeout-function of [river.Worker]. +// Maximum time a job can run before the context gets canceled. +func (w *BackChannelLogoutWorker) Timeout(*river.Job[*backchannel.LogoutRequest]) time.Duration { + return w.config.TransactionDuration +} + +// Work implements [river.Worker]. +func (w *BackChannelLogoutWorker) Work(ctx context.Context, job *river.Job[*backchannel.LogoutRequest]) error { + ctx = ContextWithNotifier(ctx, job.Args.Aggregate) + + ctx, err := enrichCtx(ctx, job.Args.TriggeredAtOrigin) + if err != nil { + return channels.NewCancelError(err) + } + + // if the notification is too old, we can directly cancel + if job.CreatedAt.Add(w.config.MaxTtl).Before(w.now()) { + return river.JobCancel(errors.New("back channel logout notification is too old")) + } + + if job.Args.OIDCSessionID == "" { + return w.createNotificationJobs(ctx, job.Args) + } + return w.sendLogoutRequest(ctx, job.Args) +} + +func (w *BackChannelLogoutWorker) createNotificationJobs(ctx context.Context, request *backchannel.LogoutRequest) error { + sessions := &backChannelLogoutSession{sessionID: request.SessionID} + err := w.eventstore.FilterToQueryReducer(ctx, sessions) + if err != nil { + return err + } + + for _, oidcSession := range sessions.sessions { + tokenID, err := w.idGenerator.Next() + if err != nil { + return err + } + logoutRequest := &backchannel.LogoutRequest{ + Aggregate: request.Aggregate, + SessionID: request.SessionID, + TriggeredAtOrigin: request.TriggeredAtOrigin, + TriggeringEventType: request.TriggeringEventType, + TokenID: tokenID, + UserID: oidcSession.UserID, + OIDCSessionID: oidcSession.OIDCSessionID, + ClientID: oidcSession.ClientID, + BackChannelLogoutURI: oidcSession.BackChannelLogoutURI, + } + err = w.queue.Insert(ctx, logoutRequest, + queue.WithQueueName(backchannel.QueueName), + queue.WithMaxAttempts(w.config.MaxAttempts)) + if err != nil { + return err + } + } + return nil +} + +func (w *BackChannelLogoutWorker) sendLogoutRequest(ctx context.Context, request *backchannel.LogoutRequest) error { + getSigner := sign.GetSignerOnce(w.queries.GetActiveSigningWebKey) + token, err := w.logoutToken(ctx, request, getSigner) + if err != nil { + return err + } + if err = types.SendSecurityTokenEvent(ctx, set.Config{CallURL: request.BackChannelLogoutURI}, w.channels, &LogoutTokenMessage{LogoutToken: token}, request.TriggeringEventType).WithoutTemplate(); err != nil { + return err + } + return w.commands.BackChannelLogoutSent(ctx, request.SessionID, request.OIDCSessionID, request.Aggregate.InstanceID) +} + +func (w *BackChannelLogoutWorker) logoutToken(ctx context.Context, request *backchannel.LogoutRequest, getSigner sign.SignerFunc) (string, error) { + token := oidc.NewLogoutTokenClaims( + request.TriggeredAtOrigin, + request.UserID, + oidc.Audience{request.ClientID}, + w.now().Add(w.config.TokenLifetime), + request.TokenID, + request.SessionID, + time.Second, + ) + signer, _, err := getSigner(ctx) + if err != nil { + return "", err + } + return crypto.Sign(token, signer) +} + +func NewBackChannelLogoutWorker( + commands Commands, + queries *NotificationQueries, + eventstore *eventstore.Eventstore, + queue Queue, + channels types.ChannelChains, + config *BackChannelLogoutWorkerConfig, + + idGenerator id.Generator, +) *BackChannelLogoutWorker { + return &BackChannelLogoutWorker{ + commands: commands, + queries: queries, + eventstore: eventstore, + queue: queue, + channels: channels, + config: config, + now: time.Now, + idGenerator: idGenerator, + } +} + +var _ river.Worker[*backchannel.LogoutRequest] = (*BackChannelLogoutWorker)(nil) + +func (w *BackChannelLogoutWorker) Register(workers *river.Workers, queues map[string]river.QueueConfig) { + river.AddWorker(workers, w) + queues[backchannel.QueueName] = river.QueueConfig{ + MaxWorkers: int(w.config.Workers), + } +} + +type BackChannelLogoutWorkerConfig struct { + Workers uint8 + TransactionDuration time.Duration + MaxAttempts uint8 + MaxTtl time.Duration + TokenLifetime time.Duration +} diff --git a/internal/notification/handlers/back_channel_logout_worker_test.go b/internal/notification/handlers/back_channel_logout_worker_test.go new file mode 100644 index 000000000..a994f6d0a --- /dev/null +++ b/internal/notification/handlers/back_channel_logout_worker_test.go @@ -0,0 +1,508 @@ +package handlers + +import ( + "context" + "crypto/rand" + "crypto/rsa" + "database/sql" + "encoding/json" + "reflect" + "testing" + "time" + + "github.com/go-jose/go-jose/v4" + "github.com/riverqueue/river" + "github.com/riverqueue/river/rivertype" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/zitadel/oidc/v3/pkg/oidc" + "go.uber.org/mock/gomock" + + "github.com/zitadel/zitadel/internal/api/authz" + "github.com/zitadel/zitadel/internal/eventstore" + "github.com/zitadel/zitadel/internal/eventstore/repository" + es_repo_mock "github.com/zitadel/zitadel/internal/eventstore/repository/mock" + "github.com/zitadel/zitadel/internal/id" + id_mock "github.com/zitadel/zitadel/internal/id/mock" + "github.com/zitadel/zitadel/internal/notification/backchannel" + "github.com/zitadel/zitadel/internal/notification/channels" + channel_mock "github.com/zitadel/zitadel/internal/notification/channels/mock" + "github.com/zitadel/zitadel/internal/notification/handlers/mock" + "github.com/zitadel/zitadel/internal/notification/messages" + "github.com/zitadel/zitadel/internal/notification/senders" + "github.com/zitadel/zitadel/internal/queue" + "github.com/zitadel/zitadel/internal/repository/sessionlogout" +) + +func Test_backChannelLogoutWorker_reduceNotificationRequested(t *testing.T) { + testNow := time.Now() + sessionLogoutAgg := &sessionlogout.NewAggregate(sessionID, instanceID).Aggregate + type fields struct { + es func(*testing.T) *eventstore.Eventstore + queue func(*gomock.Controller) Queue + commands func(*gomock.Controller) Commands + queries func(*gomock.Controller) Queries + channel func(*gomock.Controller) channels.NotificationChannel + idGenerator id.Generator + } + type args struct { + job *river.Job[*backchannel.LogoutRequest] + } + type want struct { + err error + } + tests := []struct { + name string + fields fields + args args + want want + }{ + { + name: "too old", + fields: fields{ + es: expectEventstore(), + queue: func(ctrl *gomock.Controller) Queue { + q := mock.NewMockQueue(ctrl) + return q + }, + commands: func(ctrl *gomock.Controller) Commands { + c := mock.NewMockCommands(ctrl) + return c + }, + queries: func(ctrl *gomock.Controller) Queries { + q := mock.NewMockQueries(ctrl) + return q + }, + channel: func(ctrl *gomock.Controller) channels.NotificationChannel { + c := channel_mock.NewMockNotificationChannel(ctrl) + return c + }, + }, + args: args{ + job: &river.Job[*backchannel.LogoutRequest]{ + JobRow: &rivertype.JobRow{ + CreatedAt: time.Now().Add(-1 * time.Hour), + }, + Args: &backchannel.LogoutRequest{ + Aggregate: sessionLogoutAgg, + }, + }, + }, + want: want{ + err: new(river.JobCancelError), + }, + }, + { + name: "no oidc sessions with back channel logout", + fields: fields{ + es: expectEventstore( + expectFilter(), + ), + queue: func(ctrl *gomock.Controller) Queue { + q := mock.NewMockQueue(ctrl) + return q + }, + commands: func(ctrl *gomock.Controller) Commands { + c := mock.NewMockCommands(ctrl) + return c + }, + queries: func(ctrl *gomock.Controller) Queries { + q := mock.NewMockQueries(ctrl) + return q + }, + channel: func(ctrl *gomock.Controller) channels.NotificationChannel { + c := channel_mock.NewMockNotificationChannel(ctrl) + return c + }, + }, + args: args{ + job: &river.Job[*backchannel.LogoutRequest]{ + JobRow: &rivertype.JobRow{ + CreatedAt: testNow, + }, + Args: &backchannel.LogoutRequest{ + Aggregate: sessionLogoutAgg, + SessionID: sessionID, + }, + }, + }, + want: want{ + err: nil, + }, + }, + { + name: "create jobs for oidc sessions with back channel logout", + fields: fields{ + es: expectEventstore( + expectFilter( + eventFromEventPusher( + sessionlogout.NewBackChannelLogoutRegisteredEvent( + context.Background(), + sessionLogoutAgg, + "oidc-session-id1", + "user-id", + "client-id1", + "back-channel-logout-uri1", + ), + ), + eventFromEventPusher( + sessionlogout.NewBackChannelLogoutRegisteredEvent( + context.Background(), + sessionLogoutAgg, + "oidc-session-id2", + "user-id", + "client-id2", + "back-channel-logout-uri2", + ), + ), + ), + ), + queue: func(ctrl *gomock.Controller) Queue { + q := mock.NewMockQueue(ctrl) + q.EXPECT().Insert(gomock.Any(), + &backchannel.LogoutRequest{ + Aggregate: sessionLogoutAgg, + SessionID: sessionID, + TriggeredAtOrigin: "", + TriggeringEventType: "", + TokenID: "id1", + UserID: "user-id", + OIDCSessionID: "oidc-session-id1", + ClientID: "client-id1", + BackChannelLogoutURI: "back-channel-logout-uri1", + }, + gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithQueueName(backchannel.QueueName))), + gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithMaxAttempts(1))), + ).Return(nil) + q.EXPECT().Insert(gomock.Any(), &backchannel.LogoutRequest{ + Aggregate: sessionLogoutAgg, + SessionID: sessionID, + TriggeredAtOrigin: "", + TriggeringEventType: "", + TokenID: "id2", + UserID: "user-id", + OIDCSessionID: "oidc-session-id2", + ClientID: "client-id2", + BackChannelLogoutURI: "back-channel-logout-uri2", + }, + gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithQueueName(backchannel.QueueName))), + gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithMaxAttempts(1))), + ).Return(nil) + return q + }, + commands: func(ctrl *gomock.Controller) Commands { + c := mock.NewMockCommands(ctrl) + return c + }, + queries: func(ctrl *gomock.Controller) Queries { + q := mock.NewMockQueries(ctrl) + return q + }, + channel: func(ctrl *gomock.Controller) channels.NotificationChannel { + c := channel_mock.NewMockNotificationChannel(ctrl) + return c + }, + idGenerator: id_mock.NewIDGeneratorExpectIDs(t, "id1", "id2"), + }, + args: args{ + job: &river.Job[*backchannel.LogoutRequest]{ + JobRow: &rivertype.JobRow{ + CreatedAt: testNow, + }, + Args: &backchannel.LogoutRequest{ + Aggregate: sessionLogoutAgg, + SessionID: sessionID, + }, + }, + }, + want: want{ + err: nil, + }, + }, + { + name: "send logout request", + fields: fields{ + es: expectEventstore(), + queue: func(ctrl *gomock.Controller) Queue { + q := mock.NewMockQueue(ctrl) + return q + }, + commands: func(ctrl *gomock.Controller) Commands { + c := mock.NewMockCommands(ctrl) + c.EXPECT().BackChannelLogoutSent(gomock.Any(), sessionID, "oidc-session-id", instanceID).Return(nil) + return c + }, + queries: func(ctrl *gomock.Controller) Queries { + q := mock.NewMockQueries(ctrl) + q.EXPECT().GetActiveSigningWebKey(gomock.Any()).Return( + &jose.JSONWebKey{ + Key: privateKey, + Algorithm: string(signingAlgorithm), + Use: "sig", + }, nil) + return q + }, + channel: func(ctrl *gomock.Controller) channels.NotificationChannel { + c := channel_mock.NewMockNotificationChannel(ctrl) + c.EXPECT().HandleMessage(gomock.Any()).DoAndReturn( + func(message channels.Message) error { + logoutMessage, ok := message.(*messages.Form) + if !ok { + ctrl.T.Errorf("unexpected message type: %T", message) + } + logoutToken, ok := logoutMessage.Serializable.(*LogoutTokenMessage) + if !ok { + ctrl.T.Errorf("unexpected serializable type: %T", logoutMessage.Serializable) + } + jws, err := jose.ParseSigned(logoutToken.LogoutToken, []jose.SignatureAlgorithm{jose.RS256}) + require.NoError(t, err) + payload, err := jws.Verify(privateKey.Public()) + require.NoError(t, err) + var claims oidc.LogoutTokenClaims + err = json.Unmarshal(payload, &claims) + require.NoError(t, err) + assert.Equal(t, "", claims.Issuer) + assert.Equal(t, "user-id", claims.Subject) + assert.Equal(t, "client-id", claims.Audience[0]) + assert.WithinRange(t, claims.IssuedAt.AsTime(), testNow.Add(-2*time.Second), time.Now()) + assert.WithinRange(t, claims.Expiration.AsTime(), testNow.Add(-time.Second).Add(time.Hour), time.Now().Add(time.Hour)) + assert.Equal(t, "id1", claims.JWTID) + assert.NotNil(t, claims.Events["http://schemas.openid.net/event/backchannel-logout"]) + assert.Equal(t, sessionID, claims.SessionID) + + return nil + }) + return c + }, + idGenerator: id_mock.NewIDGeneratorExpectIDs(t), + }, + args: args{ + job: &river.Job[*backchannel.LogoutRequest]{ + JobRow: &rivertype.JobRow{ + CreatedAt: testNow, + }, + Args: &backchannel.LogoutRequest{ + Aggregate: sessionLogoutAgg, + SessionID: sessionID, + TriggeredAtOrigin: "", + TriggeringEventType: "", + TokenID: "id1", + UserID: "user-id", + OIDCSessionID: "oidc-session-id", + ClientID: "client-id", + BackChannelLogoutURI: "back-channel-logout-uri", + }, + }, + }, + want: want{ + err: nil, + }, + }, + { + name: "job creation failed", + fields: fields{ + es: expectEventstore( + expectFilter( + eventFromEventPusher( + sessionlogout.NewBackChannelLogoutRegisteredEvent( + context.Background(), + sessionLogoutAgg, + "oidc-session-id1", + "user-id", + "client-id1", + "back-channel-logout-uri1", + ), + ), + ), + ), + queue: func(ctrl *gomock.Controller) Queue { + q := mock.NewMockQueue(ctrl) + q.EXPECT().Insert(gomock.Any(), + gomock.Any(), + gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithQueueName(backchannel.QueueName))), + gomock.AssignableToTypeOf(reflect.TypeOf(queue.WithMaxAttempts(1))), + ).Return(assert.AnError) + return q + }, + commands: func(ctrl *gomock.Controller) Commands { + c := mock.NewMockCommands(ctrl) + return c + }, + queries: func(ctrl *gomock.Controller) Queries { + q := mock.NewMockQueries(ctrl) + return q + }, + channel: func(ctrl *gomock.Controller) channels.NotificationChannel { + c := channel_mock.NewMockNotificationChannel(ctrl) + return c + }, + idGenerator: id_mock.NewIDGeneratorExpectIDs(t, "id1"), + }, + args: args{ + job: &river.Job[*backchannel.LogoutRequest]{ + JobRow: &rivertype.JobRow{ + CreatedAt: testNow, + }, + Args: &backchannel.LogoutRequest{ + Aggregate: sessionLogoutAgg, + SessionID: sessionID, + }, + }, + }, + want: want{ + err: assert.AnError, + }, + }, + + { + name: "send logout request failed", + fields: fields{ + es: expectEventstore(), + queue: func(ctrl *gomock.Controller) Queue { + q := mock.NewMockQueue(ctrl) + return q + }, + commands: func(ctrl *gomock.Controller) Commands { + c := mock.NewMockCommands(ctrl) + return c + }, + queries: func(ctrl *gomock.Controller) Queries { + q := mock.NewMockQueries(ctrl) + q.EXPECT().GetActiveSigningWebKey(gomock.Any()).Return( + &jose.JSONWebKey{ + Key: privateKey, + Algorithm: string(signingAlgorithm), + Use: "sig", + }, nil) + return q + }, + channel: func(ctrl *gomock.Controller) channels.NotificationChannel { + c := channel_mock.NewMockNotificationChannel(ctrl) + c.EXPECT().HandleMessage(gomock.Any()).Return(assert.AnError) + return c + }, + idGenerator: id_mock.NewIDGeneratorExpectIDs(t), + }, + args: args{ + job: &river.Job[*backchannel.LogoutRequest]{ + JobRow: &rivertype.JobRow{ + CreatedAt: testNow, + }, + Args: &backchannel.LogoutRequest{ + Aggregate: sessionLogoutAgg, + SessionID: sessionID, + TriggeredAtOrigin: "", + TriggeringEventType: "", + TokenID: "id1", + UserID: "user-id", + OIDCSessionID: "oidc-session-id", + ClientID: "client-id", + BackChannelLogoutURI: "back-channel-logout-uri", + }, + }, + }, + want: want{ + err: assert.AnError, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + err := newBackChannelLogoutWorker( + tt.fields.queries(ctrl), + tt.fields.commands(ctrl), + tt.fields.es(t), + tt.fields.queue(ctrl), + tt.fields.channel(ctrl), + tt.fields.idGenerator, + func() time.Time { return testNow }, + ).Work( + authz.WithInstanceID(context.Background(), instanceID), + tt.args.job, + ) + assert.ErrorIs(t, err, tt.want.err) + }) + } +} + +func newBackChannelLogoutWorker(queries Queries, commands Commands, es *eventstore.Eventstore, queue Queue, channel channels.NotificationChannel, idGenerator id.Generator, testNow func() time.Time) *BackChannelLogoutWorker { + return &BackChannelLogoutWorker{ + commands: commands, + queries: NewNotificationQueries( + queries, + es, + externalDomain, + externalPort, + externalSecure, + "", + nil, + nil, + nil, + ), + eventstore: es, + queue: queue, + channels: ¬ificationChannels{ + Chain: *senders.ChainChannels(channel), + }, + config: &BackChannelLogoutWorkerConfig{ + Workers: 1, + TransactionDuration: 5 * time.Second, + MaxTtl: 5 * time.Minute, + MaxAttempts: 1, + TokenLifetime: time.Hour, + }, + now: testNow, + idGenerator: idGenerator, + } +} + +type expect func(mockRepository *es_repo_mock.MockRepository) + +func expectEventstore(expects ...expect) func(*testing.T) *eventstore.Eventstore { + return func(t *testing.T) *eventstore.Eventstore { + m := es_repo_mock.NewRepo(t) + for _, e := range expects { + e(m) + } + es := eventstore.NewEventstore( + &eventstore.Config{ + Querier: m.MockQuerier, + Pusher: m.MockPusher, + }, + ) + return es + } +} + +func expectFilter(events ...eventstore.Event) expect { + return func(m *es_repo_mock.MockRepository) { + m.ExpectFilterEvents(events...) + } +} + +func eventFromEventPusher(event eventstore.Command) *repository.Event { + data, _ := eventstore.EventData(event) + return &repository.Event{ + InstanceID: event.Aggregate().InstanceID, + ID: "", + Seq: 0, + CreationDate: time.Time{}, + Typ: event.Type(), + Data: data, + EditorUser: event.Creator(), + Version: event.Aggregate().Version, + AggregateID: event.Aggregate().ID, + AggregateType: event.Aggregate().Type, + ResourceOwner: sql.NullString{String: event.Aggregate().ResourceOwner, Valid: event.Aggregate().ResourceOwner != ""}, + Constraints: event.UniqueConstraints(), + } +} + +var ( + privateKey = func() *rsa.PrivateKey { + privateKey, _ := rsa.GenerateKey(rand.Reader, 2048) + return privateKey + }() + signingAlgorithm = jose.RS256 +) diff --git a/internal/notification/handlers/commands.go b/internal/notification/handlers/commands.go index a14368298..7b1a33362 100644 --- a/internal/notification/handlers/commands.go +++ b/internal/notification/handlers/commands.go @@ -24,4 +24,5 @@ type Commands interface { InviteCodeSent(ctx context.Context, orgID, userID string) error UsageNotificationSent(ctx context.Context, dueEvent *quota.NotificationDueEvent) error MilestonePushed(ctx context.Context, instanceID string, msType milestone.Type, endpoints []string) error + BackChannelLogoutSent(ctx context.Context, id, oidcSessionID, instanceID string) (err error) } diff --git a/internal/notification/handlers/mock/commands.mock.go b/internal/notification/handlers/mock/commands.mock.go index 130032bc5..8cb525fa1 100644 --- a/internal/notification/handlers/mock/commands.mock.go +++ b/internal/notification/handlers/mock/commands.mock.go @@ -43,6 +43,20 @@ func (m *MockCommands) EXPECT() *MockCommandsMockRecorder { return m.recorder } +// BackChannelLogoutSent mocks base method. +func (m *MockCommands) BackChannelLogoutSent(arg0 context.Context, arg1, arg2, arg3 string) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "BackChannelLogoutSent", arg0, arg1, arg2, arg3) + ret0, _ := ret[0].(error) + return ret0 +} + +// BackChannelLogoutSent indicates an expected call of BackChannelLogoutSent. +func (mr *MockCommandsMockRecorder) BackChannelLogoutSent(arg0, arg1, arg2, arg3 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BackChannelLogoutSent", reflect.TypeOf((*MockCommands)(nil).BackChannelLogoutSent), arg0, arg1, arg2, arg3) +} + // HumanEmailVerificationCodeSent mocks base method. func (m *MockCommands) HumanEmailVerificationCodeSent(ctx context.Context, orgID, userID string) error { m.ctrl.T.Helper() diff --git a/internal/notification/projections.go b/internal/notification/projections.go index 7fedaaf30..832e9ebf9 100644 --- a/internal/notification/projections.go +++ b/internal/notification/projections.go @@ -3,7 +3,6 @@ package notification import ( "context" "fmt" - "time" "github.com/zitadel/logging" @@ -12,6 +11,7 @@ import ( "github.com/zitadel/zitadel/internal/crypto" "github.com/zitadel/zitadel/internal/eventstore" "github.com/zitadel/zitadel/internal/eventstore/handler/v2" + "github.com/zitadel/zitadel/internal/id" "github.com/zitadel/zitadel/internal/notification/handlers" _ "github.com/zitadel/zitadel/internal/notification/statik" "github.com/zitadel/zitadel/internal/query" @@ -27,6 +27,7 @@ func Register( ctx context.Context, userHandlerCustomConfig, quotaHandlerCustomConfig, telemetryHandlerCustomConfig, backChannelLogoutHandlerCustomConfig projection.CustomConfig, notificationWorkerConfig handlers.WorkerConfig, + backChannelLogoutWorkerConfig *handlers.BackChannelLogoutWorkerConfig, telemetryCfg handlers.TelemetryPusherConfig, externalDomain string, externalPort uint16, @@ -35,8 +36,7 @@ func Register( queries *query.Queries, es *eventstore.Eventstore, otpEmailTmpl, fileSystemPath string, - userEncryption, smtpEncryption, smsEncryption, keysEncryptionAlg crypto.EncryptionAlgorithm, - tokenLifetime time.Duration, + userEncryption, smtpEncryption, smsEncryption crypto.EncryptionAlgorithm, queue *queue.Queue, ) { if !notificationWorkerConfig.LegacyEnabled { @@ -53,13 +53,11 @@ func Register( projections = append(projections, handlers.NewBackChannelLogoutNotifier( ctx, projection.ApplyCustomConfig(backChannelLogoutHandlerCustomConfig), - commands, q, - es, - keysEncryptionAlg, - c, - tokenLifetime, + queue, + backChannelLogoutWorkerConfig.MaxAttempts, )) + queue.AddWorkers(handlers.NewBackChannelLogoutWorker(commands, q, es, queue, c, backChannelLogoutWorkerConfig, id.SonyFlakeGenerator())) if telemetryCfg.Enabled { projections = append(projections, handlers.NewTelemetryPusher(ctx, telemetryCfg, projection.ApplyCustomConfig(telemetryHandlerCustomConfig), commands, q, c)) }