From d51d615e979561ad39fc74098643f9bb29f38854 Mon Sep 17 00:00:00 2001 From: Livio Spring Date: Tue, 3 Feb 2026 11:47:48 +0100 Subject: [PATCH] feat(oidc): use worker queue for OIDC Back-Channel Logout notifications (#11441) # Which Problems Are Solved The current back-channel logout notification still used a single projection handler for handling terminating sessions and notifying the necessary clients. This can lead to back pressure and delayed notifications. # How the Problems Are Solved The handler now only creates a job in the (river) worker queue to handle the notification. The worker will then search for all necessary clients to be informed and create a job for each. These jobs will then be picked up by the (same) worker(s) again, which will create, sign and send the logout token to the client. If one notification fails, this one job will be retired, but others can still be processed in parallel and ore not affected. Each successful job will still create an event on the session stating the successful notification of the client. The existing `OIDC.DefaultBackChannelLogoutLifetime` configuration has been deprecated in favor of making it part of the new `OIDC.BackChannelLogout` config. # Additional Changes None # Additional Context - feature check is still executed and will be removed seperately for easier review: https://github.com/zitadel/zitadel/issues/11277 - closes https://github.com/zitadel/zitadel/issues/9279 - requires backport to v4.x --- cmd/defaults.yaml | 16 +- cmd/mirror/projections.go | 3 +- cmd/setup/setup.go | 3 +- cmd/start/start.go | 3 +- internal/api/oidc/op.go | 11 + .../backchannel/back_channel_logout.go | 23 + .../handlers/back_channel_logout.go | 141 ++--- .../handlers/back_channel_logout_worker.go | 160 ++++++ .../back_channel_logout_worker_test.go | 508 ++++++++++++++++++ internal/notification/handlers/commands.go | 1 + .../handlers/mock/commands.mock.go | 14 + internal/notification/projections.go | 14 +- 12 files changed, 780 insertions(+), 117 deletions(-) create mode 100644 internal/notification/backchannel/back_channel_logout.go create mode 100644 internal/notification/handlers/back_channel_logout_worker.go create mode 100644 internal/notification/handlers/back_channel_logout_worker_test.go 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)) }