From b72cfadf5d0d29bed5f551751ca33dbd3b7acb0d Mon Sep 17 00:00:00 2001 From: zerox80 Date: Mon, 7 Sep 2026 09:36:26 +0200 Subject: [PATCH 1/7] refactor(proxy): support per-token logout records Add token-specific session keys and separate revocation keys. Keep the legacy subject/session key format readable while allowing multiple tokens to belong to one session. Match both subject and session when both are supplied and avoid truncating logout lookups after 1,000 cache entries. Include unit tests for key compatibility, multiple tokens, and subject isolation. Authentication starts using these records in the logout fix. --- .../backchannellogout/backchannellogout.go | 50 ++++++++++++--- .../backchannellogout_test.go | 63 +++++++++++++++++++ 2 files changed, 106 insertions(+), 7 deletions(-) diff --git a/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout.go b/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout.go index 1d906233cd..4890e1b379 100644 --- a/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout.go +++ b/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout.go @@ -6,6 +6,7 @@ package backchannellogout import ( "encoding/base64" "errors" + "math" "strings" microstore "go-micro.dev/v4/store" @@ -31,6 +32,38 @@ func NewKey(subject, session string) (string, error) { return subjectSession, nil } +// NewTokenKey associates a token with a subject and session without replacing +// tokens that were previously issued for the same session. +func NewTokenKey(subject, session, tokenKey string) (string, error) { + key, err := NewKey(subject, session) + if err != nil { + return "", err + } + if tokenKey == "" || strings.Contains(tokenKey, ".") { + return "", ErrInvalidKey + } + parts := strings.Split(key, ".") + return strings.Join([]string{parts[0], tokenKey, parts[1]}, "."), nil +} + +// RevokedTokenKey is separate from the claims key so that a concurrent cache +// write cannot overwrite a logout decision. +func RevokedTokenKey(tokenKey string) string { + return "revoked/" + tokenKey +} + +// IsTokenRevoked checks whether a token was invalidated by backchannel logout. +func IsTokenRevoked(tokenKey string, cache microstore.Store) (bool, error) { + records, err := cache.Read(RevokedTokenKey(tokenKey)) + if errors.Is(err, microstore.ErrNotFound) { + return false, nil + } + if err != nil { + return false, err + } + return len(records) > 0, nil +} + // LogoutMode defines the mode of backchannel logout, either by session or by subject type LogoutMode int @@ -99,6 +132,12 @@ func NewSuSe(key string) (SuSe, error) { case 2: suse.encodedSubject = keys[0] suse.encodedSession = keys[1] + case 3: + if keys[1] == "" { + return suse, ErrInvalidSubjectOrSession + } + suse.encodedSubject = keys[0] + suse.encodedSession = keys[2] default: return suse, ErrInvalidSubjectOrSession } @@ -148,7 +187,7 @@ func GetLogoutRecords(suse SuSe, store microstore.Store) ([]*microstore.Record, } // the go micro memory store requires a limit to work, why??? - records, err := store.Read(key, append(opts, microstore.ReadLimit(1000))...) + records, err := store.Read(key, append(opts, microstore.ReadLimit(math.MaxInt))...) if err != nil { return nil, err } @@ -157,10 +196,6 @@ func GetLogoutRecords(suse SuSe, store microstore.Store) ([]*microstore.Record, return nil, microstore.ErrNotFound } - if suse.Mode() == LogoutModeSession && len(records) > 1 { - return nil, errors.Join(errors.New("multiple session records found"), ErrSuspiciousCacheResult) - } - // double-check if the found records match the requested subject and or session id as well, // to prevent false positives. for _, record := range records { @@ -174,8 +209,9 @@ func GetLogoutRecords(suse SuSe, store microstore.Store) ([]*microstore.Record, // in subject mode, the subject must match, but the session id can be different case suse.Mode() == LogoutModeSubject && suse.encodedSubject == recordSuSe.encodedSubject: continue - // in session mode, the session id must match, but the subject can be different - case suse.Mode() == LogoutModeSession && suse.encodedSession == recordSuSe.encodedSession: + // In session mode, match the subject too when it was supplied. + case suse.Mode() == LogoutModeSession && suse.encodedSession == recordSuSe.encodedSession && + (suse.encodedSubject == "" || suse.encodedSubject == recordSuSe.encodedSubject): continue } diff --git a/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout_test.go b/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout_test.go index fead1c0310..fb9589747a 100644 --- a/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout_test.go +++ b/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout_test.go @@ -1,6 +1,7 @@ package backchannellogout import ( + "fmt" "slices" "strings" "testing" @@ -18,6 +19,68 @@ func mustNewKey(t *testing.T, subject, session string) string { return key } +func TestTokenLookupKeys(t *testing.T) { + for _, tt := range []struct{ subject, session string }{ + {"alice", "session"}, {"alice", ""}, {"", "session"}, + } { + t.Run(tt.subject+"/"+tt.session, func(t *testing.T) { + key, err := NewTokenKey(tt.subject, tt.session, "token-key") + require.NoError(t, err) + suse, err := NewSuSe(key) + require.NoError(t, err) + subject, err := suse.Subject() + require.NoError(t, err) + session, err := suse.Session() + require.NoError(t, err) + require.Equal(t, tt.subject, subject) + require.Equal(t, tt.session, session) + }) + } + _, err := NewTokenKey("", "", "token-key") + require.ErrorIs(t, err, ErrInvalidKey) + _, err = NewTokenKey("alice", "session", "") + require.ErrorIs(t, err, ErrInvalidKey) + _, err = NewTokenKey("alice", "session", "invalid.key") + require.ErrorIs(t, err, ErrInvalidKey) +} + +func TestGetLogoutRecordsIncludesAllTokens(t *testing.T) { + cache := store.NewMemoryStore() + // Unrelated entries must not make the lookup truncate before finding the + // session. The memory store applies its read limit before filtering keys. + for i := range 1100 { + require.NoError(t, cache.Write(&store.Record{Key: fmt.Sprintf("unrelated-%d", i)})) + } + want := []string{} + for _, token := range []string{"first", "second"} { + key, err := NewTokenKey("alice", "session", token) + require.NoError(t, err) + require.NoError(t, cache.Write(&store.Record{Key: key, Value: []byte(token)})) + want = append(want, key) + } + legacy := mustNewKey(t, "alice", "session") + require.NoError(t, cache.Write(&store.Record{Key: legacy, Value: []byte("legacy")})) + want = append(want, legacy) + for _, request := range []SuSe{mustNewSuSe(t, "alice", ""), mustNewSuSe(t, "", "session"), mustNewSuSe(t, "alice", "session")} { + records, err := GetLogoutRecords(request, cache) + require.NoError(t, err) + keys := make([]string, 0, len(records)) + for _, record := range records { + keys = append(keys, record.Key) + } + require.ElementsMatch(t, want, keys) + } +} + +func TestGetLogoutRecordsRejectsMismatchedSubject(t *testing.T) { + cache := store.NewMemoryStore() + key, err := NewTokenKey("bob", "session", "token") + require.NoError(t, err) + require.NoError(t, cache.Write(&store.Record{Key: key})) + _, err = GetLogoutRecords(mustNewSuSe(t, "alice", "session"), cache) + require.ErrorIs(t, err, ErrSuspiciousCacheResult) +} + func mustNewSuSe(t *testing.T, subject, session string) SuSe { suse, err := NewSuSe(mustNewKey(t, subject, session)) require.NoError(t, err) From 812350e8cb0ec5a9f52f3a34c979a18e94033e2f Mon Sep 17 00:00:00 2001 From: zerox80 Date: Mon, 7 Sep 2026 09:36:27 +0200 Subject: [PATCH 2/7] fix(proxy): preserve OIDC cache expiry across storage backends Store absolute record expiry with the value so Redis and NATS retain the same lifetime information as memory stores. Use a dedicated OIDC namespace without a bucket-wide TTL, migrate legacy cached tokens before startup, and clean up expired records with the proxy context. Keep logout state when claims caching is disabled. Add expiry, migration, cleanup, and memory/Redis/NATS backend tests alongside the cache changes. --- services/proxy/pkg/command/oidc_cache.go | 54 +++ services/proxy/pkg/command/oidc_cache_test.go | 111 ++++++ services/proxy/pkg/command/server.go | 19 +- .../staticroutes/backchannellogout/cache.go | 317 ++++++++++++++++++ .../backchannellogout/cache_test.go | 179 ++++++++++ 5 files changed, 668 insertions(+), 12 deletions(-) create mode 100644 services/proxy/pkg/command/oidc_cache.go create mode 100644 services/proxy/pkg/command/oidc_cache_test.go create mode 100644 services/proxy/pkg/staticroutes/backchannellogout/cache.go create mode 100644 services/proxy/pkg/staticroutes/backchannellogout/cache_test.go diff --git a/services/proxy/pkg/command/oidc_cache.go b/services/proxy/pkg/command/oidc_cache.go new file mode 100644 index 0000000000..f7df22d6cf --- /dev/null +++ b/services/proxy/pkg/command/oidc_cache.go @@ -0,0 +1,54 @@ +package command + +import ( + "context" + "time" + + "github.com/opencloud-eu/opencloud/services/proxy/pkg/config" + bcl "github.com/opencloud-eu/opencloud/services/proxy/pkg/staticroutes/backchannellogout" + "github.com/opencloud-eu/reva/v2/pkg/store" + microstore "go-micro.dev/v4/store" +) + +func newUserInfoCache(cfg *config.Cache) *bcl.Cache { + storeType := cfg.Store + cacheClaims := storeType != store.TypeNoop + if !cacheClaims || storeType == store.TypeOCMem { + // Security records must not use the shared, capacity-evicted ocmem + // cache, which also ignores the configured database namespace. + storeType = store.TypeMemory + } + database := cfg.Database + if database == "" { + database = "cache-userinfo" + } + // Keep the no-TTL bucket separate from other services and older proxies + // which may use the configured database with a bucket-wide TTL. + // Redis ignores Database, so give its table a dedicated prefix as well. + return bcl.NewCache(newUserInfoStore(cfg, storeType, database+"-oidc-v2", cfg.Table+"/oidc-v2/", 0), cacheClaims) +} + +func migrateUserInfoCache(ctx context.Context, cache *bcl.Cache, cfg *config.Cache) error { + // Memory and noop stores cannot contain entries from a previous process. + if cfg.Store == "" || cfg.Store == "mem" || cfg.Store == store.TypeMemory || cfg.Store == store.TypeNoop || cfg.Store == store.TypeOCMem { + return nil + } + legacy := newUserInfoStore(cfg, cfg.Store, cfg.Database, cfg.Table, cfg.TTL) + defer legacy.Close() + return cache.IndexLegacyTokens(ctx, legacy) +} + +func newUserInfoStore(cfg *config.Cache, storeType, database, table string, ttl time.Duration) microstore.Store { + return store.Create( + store.Store(storeType), + store.TTL(ttl), + microstore.Nodes(cfg.Nodes...), + microstore.Database(database), + microstore.Table(table), + store.DisablePersistence(cfg.DisablePersistence), + store.Authentication(cfg.AuthUsername, cfg.AuthPassword), + store.TLSEnabled(cfg.EnableTLS), + store.TLSInsecure(cfg.TLSInsecure), + store.TLSRootCA(cfg.TLSRootCACertificate), + ) +} diff --git a/services/proxy/pkg/command/oidc_cache_test.go b/services/proxy/pkg/command/oidc_cache_test.go new file mode 100644 index 0000000000..9725398e3a --- /dev/null +++ b/services/proxy/pkg/command/oidc_cache_test.go @@ -0,0 +1,111 @@ +package command + +import ( + "context" + "encoding/base64" + "os" + "os/exec" + "path/filepath" + "testing" + "time" + + nserver "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/opencloud-eu/opencloud/services/proxy/pkg/config" + bcl "github.com/opencloud-eu/opencloud/services/proxy/pkg/staticroutes/backchannellogout" + "github.com/stretchr/testify/require" + "github.com/vmihailenco/msgpack/v5" + "go-micro.dev/v4/store" +) + +func TestOIDCCacheBackends(t *testing.T) { + for _, backend := range []string{"memory", "nats-js-kv", "redis"} { + t.Run(backend, func(t *testing.T) { + cfg := &config.Cache{Store: backend, Database: "cache-userinfo", TTL: time.Second} + var natsURL string + switch backend { + case "nats-js-kv": + server, err := nserver.NewServer(&nserver.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir()}) + require.NoError(t, err) + go server.Start() + t.Cleanup(func() { server.Shutdown(); server.WaitForShutdown() }) + require.True(t, server.ReadyForConnections(5*time.Second)) + natsURL = server.ClientURL() + cfg.Nodes = []string{natsURL} + case "redis": + binary, err := exec.LookPath("redis-server") + if err != nil { + t.Skip("redis-server is required for the Redis backend integration test") + } + socket := filepath.Join(t.TempDir(), "redis.sock") + cmd := exec.Command(binary, "--port", "0", "--unixsocket", socket, "--save", "", "--appendonly", "no") + require.NoError(t, cmd.Start()) + t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }) + require.Eventually(t, func() bool { _, err := os.Stat(socket); return err == nil }, 5*time.Second, 10*time.Millisecond) + cfg.Nodes = []string{"unix://" + socket} + } + + legacy := newUserInfoStore(cfg, cfg.Store, cfg.Database, cfg.Table, cfg.TTL) + t.Cleanup(func() { _ = legacy.Close() }) + cache := newUserInfoCache(cfg) + defer cache.Close() + var tokenKeys []string + data, err := msgpack.Marshal(map[string]any{"sub": "alice", "sid": "session", "exp": time.Now().Add(time.Hour).Unix()}) + require.NoError(t, err) + for i := range 2 { + bytes := make([]byte, 64) + bytes[0] = byte(i) + key := base64.URLEncoding.EncodeToString(bytes) + tokenKeys = append(tokenKeys, key) + require.NoError(t, legacy.Write(&store.Record{Key: key, Value: data, Expiry: time.Hour})) + lookup, err := bcl.NewKey("alice", "session") + require.NoError(t, err) + require.NoError(t, legacy.Write(&store.Record{Key: lookup, Value: []byte(key), Expiry: time.Hour})) + } + // Persistent backends import all old claims, not just the last lookup. + require.NoError(t, cache.IndexLegacyTokens(context.Background(), legacy)) + lookup, err := bcl.NewKey("alice", "session") + require.NoError(t, err) + suse, err := bcl.NewSuSe(lookup) + require.NoError(t, err) + records, err := bcl.GetLogoutRecords(suse, cache) + require.NoError(t, err) + require.Len(t, records, 2) + for _, record := range records { + require.NoError(t, bcl.RevokeToken(record, cache)) + require.NoError(t, cache.Delete(record.Key)) + require.NoError(t, cache.Delete(string(record.Value))) + } + // Exercise known token lifetimes independently of migration's unknown TTL. + require.NoError(t, bcl.RevokeToken(&store.Record{Value: []byte("known-expiry"), Expiry: time.Hour}, cache)) + if backend != "memory" { + peer := newUserInfoCache(cfg) + t.Cleanup(func() { _ = peer.Close() }) + require.NoError(t, migrateUserInfoCache(context.Background(), peer, cfg)) + cache = peer + } + if natsURL != "" { + conn, err := nats.Connect(natsURL) + require.NoError(t, err) + defer conn.Close() + js, err := conn.JetStream() + require.NoError(t, err) + old, err := js.StreamInfo("KV_cache-userinfo") + require.NoError(t, err) + require.Equal(t, cfg.TTL, old.Config.MaxAge) + current, err := js.StreamInfo("KV_cache-userinfo-oidc-v2") + require.NoError(t, err) + require.Zero(t, current.Config.MaxAge, "bucket TTL must not discard revocations") + require.Eventually(t, func() bool { + records, err := legacy.Read(tokenKeys[0]) + return (err == nil || err == store.ErrNotFound) && len(records) == 0 + }, 5*time.Second, 20*time.Millisecond) + } + for _, key := range append(tokenKeys, "known-expiry") { + revoked, err := bcl.IsTokenRevoked(key, cache) + require.NoError(t, err) + require.True(t, revoked, "revocations must survive migration, another proxy, and the legacy bucket TTL") + } + }) + } +} diff --git a/services/proxy/pkg/command/server.go b/services/proxy/pkg/command/server.go index 2a9a5cc0ba..6c38ad38f8 100644 --- a/services/proxy/pkg/command/server.go +++ b/services/proxy/pkg/command/server.go @@ -58,18 +58,7 @@ func Server(cfg *config.Config) *cobra.Command { return configlog.ReturnFatal(parser.ParseConfig(cfg)) }, RunE: func(cmd *cobra.Command, args []string) error { - userInfoCache := store.Create( - store.Store(cfg.OIDC.UserinfoCache.Store), - store.TTL(cfg.OIDC.UserinfoCache.TTL), - microstore.Nodes(cfg.OIDC.UserinfoCache.Nodes...), - microstore.Database(cfg.OIDC.UserinfoCache.Database), - microstore.Table(cfg.OIDC.UserinfoCache.Table), - store.DisablePersistence(cfg.OIDC.UserinfoCache.DisablePersistence), - store.Authentication(cfg.OIDC.UserinfoCache.AuthUsername, cfg.OIDC.UserinfoCache.AuthPassword), - store.TLSEnabled(cfg.OIDC.UserinfoCache.EnableTLS), - store.TLSInsecure(cfg.OIDC.UserinfoCache.TLSInsecure), - store.TLSRootCA(cfg.OIDC.UserinfoCache.TLSRootCACertificate), - ) + userInfoCache := newUserInfoCache(cfg.OIDC.UserinfoCache) signingKeyStore := store.Create( store.Store(cfg.PreSignedURL.SigningKeys.Store), @@ -121,6 +110,12 @@ func Server(cfg *config.Config) *cobra.Command { cfg.Context, cancel = signal.NotifyContext(context.Background(), runner.StopSignals...) defer cancel() } + if err := migrateUserInfoCache(cfg.Context, userInfoCache, cfg.OIDC.UserinfoCache); err != nil { + return fmt.Errorf("failed to migrate OIDC cache: %w", err) + } + cacheContext, stopCacheCleanup := context.WithCancel(cfg.Context) + defer stopCacheCleanup() + go userInfoCache.CollectExpired(cacheContext, logger) m := metrics.New() m.BuildInfo.WithLabelValues(version.GetString()).Set(1) diff --git a/services/proxy/pkg/staticroutes/backchannellogout/cache.go b/services/proxy/pkg/staticroutes/backchannellogout/cache.go new file mode 100644 index 0000000000..100064a841 --- /dev/null +++ b/services/proxy/pkg/staticroutes/backchannellogout/cache.go @@ -0,0 +1,317 @@ +package backchannellogout + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/vmihailenco/msgpack/v5" + "go-micro.dev/v4/store" +) + +const expiryMetadataKey = "opencloud-oidc-expires-at" + +// Cache preserves per-record expiry even on stores, such as NATS, that only +// support a bucket-wide TTL. The underlying store must have no bucket-wide TTL. +type Cache struct { + store.Store + cacheClaims bool + now func() time.Time +} + +// NewCache wraps the dedicated OIDC cache. Disabling claims caching still keeps +// the session index and revocations needed to process logout requests. +func NewCache(underlying store.Store, cacheClaims bool) *Cache { + return &Cache{Store: underlying, cacheClaims: cacheClaims, now: time.Now} +} + +func isClaimsKey(key string) bool { + decoded, err := keyEncoding.DecodeString(key) + return err == nil && len(decoded) == 64 +} + +// Write records an absolute expiry alongside the data without modifying the +// caller's record. Native per-record expiry remains enabled where supported. +func (c *Cache) Write(record *store.Record, opts ...store.WriteOption) error { + if !c.cacheClaims && isClaimsKey(record.Key) { + return nil + } + r := *record + r.Metadata = make(map[string]any, len(record.Metadata)+1) + for key, value := range record.Metadata { + r.Metadata[key] = value + } + options := store.WriteOptions{} + for _, opt := range opts { + opt(&options) + } + expiresAt := "0" + if options.TTL != 0 { + expiresAt = strconv.FormatInt(c.now().Add(options.TTL).UnixNano(), 10) + r.Expiry = options.TTL + } else if !options.Expiry.IsZero() { + expiresAt = strconv.FormatInt(options.Expiry.UnixNano(), 10) + r.Expiry = options.Expiry.Sub(c.now()) + } else if record.Expiry != 0 { + expiresAt = strconv.FormatInt(c.now().Add(record.Expiry).UnixNano(), 10) + } + r.Metadata[expiryMetadataKey] = expiresAt + // Redis stores only Value and ignores Metadata and WriteExpiry. Preserve + // the complete record in Value and also supply its native relative TTL. + data, err := msgpack.Marshal(&r) + if err != nil { + return err + } + r.Value = data + return c.Store.Write(&r, opts...) +} + +// RevokeToken preserves the token's original absolute expiry. Security records +// for a given token must not acquire a new lifetime on each write. +func RevokeToken(record *store.Record, cache store.Store) error { + var opts []store.WriteOption + if value, ok := record.Metadata[expiryMetadataKey]; ok { + nanos, err := strconv.ParseInt(fmt.Sprint(value), 10, 64) + if err != nil { + return fmt.Errorf("invalid token expiry: %w", err) + } + if nanos != 0 { + opts = append(opts, store.WriteExpiry(time.Unix(0, nanos))) + } + } + return cache.Write(&store.Record{ + Key: RevokedTokenKey(string(record.Value)), Value: []byte{1}, Expiry: record.Expiry, + }, opts...) +} + +// Read excludes expired records and returns their remaining lifetime. +func (c *Cache) Read(key string, opts ...store.ReadOption) ([]*store.Record, error) { + if !c.cacheClaims && isClaimsKey(key) { + return nil, store.ErrNotFound + } + records, err := c.readRecords(key, opts...) + if err != nil { + return nil, err + } + active := make([]*store.Record, 0, len(records)) + for _, stored := range records { + record := &store.Record{} + if err := msgpack.Unmarshal(stored.Value, record); err != nil { + return nil, fmt.Errorf("invalid OIDC cache record: %w", err) + } + expiry, err := c.expiry(record) + if err != nil { + return nil, err + } + if !expiry.IsZero() { + remaining := expiry.Sub(c.now()) + if remaining <= 0 { + // A failed cleanup must never make expired data usable again. + _ = c.Store.Delete(record.Key) + continue + } + record.Expiry = remaining + } + active = append(active, record) + } + return active, nil +} + +func (c *Cache) readRecords(key string, opts ...store.ReadOption) ([]*store.Record, error) { + options := store.ReadOptions{} + for _, opt := range opts { + opt(&options) + } + if !options.Prefix && !options.Suffix { + return c.Store.Read(key, opts...) + } + // Read each listed key individually: the Redis plugin returns the search + // prefix as every record's key, and the memory plugin can abort a prefix + // read when any matching entry expires between listing and reading it. + from := c.Store.Options() + if options.Database != "" { + from.Database = options.Database + } + if options.Table != "" { + from.Table = options.Table + } + keys, err := c.Store.List(store.ListFrom(from.Database, from.Table)) + if err != nil { + return nil, err + } + var records []*store.Record + for _, found := range keys { + if (options.Prefix && !strings.HasPrefix(found, key)) || (options.Suffix && !strings.HasSuffix(found, key)) { + continue + } + read, err := c.Store.Read(found, store.ReadFrom(from.Database, from.Table)) + if errors.Is(err, store.ErrNotFound) { + continue + } + if err != nil { + return nil, err + } + records = append(records, read...) + } + return records, nil +} + +func (c *Cache) expiry(record *store.Record) (time.Time, error) { + if value, ok := record.Metadata[expiryMetadataKey]; ok { + nanos, err := strconv.ParseInt(fmt.Sprint(value), 10, 64) + if err != nil { + return time.Time{}, fmt.Errorf("invalid OIDC cache expiry: %w", err) + } + if nanos == 0 { + return time.Time{}, nil + } + return time.Unix(0, nanos), nil + } + // Entries written by an older proxy have no expiry metadata. Memory and + // Redis supply their remaining TTL; legacy NATS claims contain exp instead. + if record.Expiry > 0 { + return c.now().Add(record.Expiry), nil + } + if isClaimsKey(record.Key) { + var claims map[string]any + if err := msgpack.Unmarshal(record.Value, &claims); err != nil { + return time.Time{}, err + } + seconds, err := strconv.ParseInt(fmt.Sprint(claims["exp"]), 10, 64) + if err != nil { + return time.Time{}, err + } + return time.Unix(seconds, 0), nil + } + return time.Time{}, nil +} + +// IndexLegacyTokens migrates claims cached by an older proxy before serving +// requests. A legacy session lookup contains only the last token's hash, so it +// is not sufficient to invalidate every token already present in the cache. +func (c *Cache) IndexLegacyTokens(ctx context.Context, legacy store.Store) error { + keys, err := legacy.List() + if err != nil { + return err + } + for _, key := range keys { + if err := ctx.Err(); err != nil { + return err + } + if !isClaimsKey(key) { + continue + } + revoked, err := IsTokenRevoked(key, c) + if err != nil { + return err + } + if revoked { + continue + } + records, err := legacy.Read(key) + if errors.Is(err, store.ErrNotFound) { + continue + } + if err != nil { + return err + } + for _, record := range records { + if _, current := record.Metadata[expiryMetadataKey]; current { + continue + } + var claims map[string]any + err := msgpack.Unmarshal(record.Value, &claims) + expiresAt, expiryErr := c.expiry(record) + if err != nil || expiryErr != nil || (!expiresAt.IsZero() && !c.now().Before(expiresAt)) { + continue + } + subject, _ := claims["sub"].(string) + session, _ := claims["sid"].(string) + lookupKey, err := NewTokenKey(subject, session, key) + if err == nil { + existing, err := c.Read(lookupKey) + if err != nil && !errors.Is(err, store.ErrNotFound) { + return err + } + if len(existing) > 0 { + continue + } + // Legacy exp may have been a fallback claims-cache TTL rather + // than a token expiry. Do not guess when its revocation can end. + if err := c.Write(&store.Record{Key: lookupKey, Value: []byte(key)}); err != nil { + return err + } + } else { + continue + } + var opts []store.WriteOption + if !expiresAt.IsZero() { + opts = append(opts, store.WriteExpiry(expiresAt)) + } + if err := c.Write(record, opts...); err != nil { + return err + } + } + } + return nil +} + +// CollectExpired removes records that are no longer needed, including on stores +// without native per-record TTL support. It stops with the proxy's context. +func (c *Cache) CollectExpired(ctx context.Context, logger log.Logger) { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := c.collectExpired(ctx); err != nil && ctx.Err() == nil { + logger.Error().Err(err).Msg("failed to clean up OIDC cache") + } + } + } +} + +func (c *Cache) collectExpired(ctx context.Context) error { + keys, err := c.Store.List() + if err != nil { + return err + } + var cleanupErr error + for _, key := range keys { + if err := ctx.Err(); err != nil { + return err + } + records, err := c.Store.Read(key) + if errors.Is(err, store.ErrNotFound) { + continue + } + if err != nil { + cleanupErr = errors.Join(cleanupErr, err) + continue + } + for _, stored := range records { + record := &store.Record{} + if err := msgpack.Unmarshal(stored.Value, record); err != nil { + cleanupErr = errors.Join(cleanupErr, err) + continue + } + expiresAt, err := c.expiry(record) + if err != nil { + cleanupErr = errors.Join(cleanupErr, err) + continue + } + if !expiresAt.IsZero() && !c.now().Before(expiresAt) { + if err := c.Store.Delete(record.Key); err != nil && !errors.Is(err, store.ErrNotFound) { + cleanupErr = errors.Join(cleanupErr, err) + } + } + } + } + return cleanupErr +} diff --git a/services/proxy/pkg/staticroutes/backchannellogout/cache_test.go b/services/proxy/pkg/staticroutes/backchannellogout/cache_test.go new file mode 100644 index 0000000000..af499b2bcd --- /dev/null +++ b/services/proxy/pkg/staticroutes/backchannellogout/cache_test.go @@ -0,0 +1,179 @@ +package backchannellogout + +import ( + "context" + "encoding/base64" + "errors" + "strconv" + "testing" + "time" + + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/stretchr/testify/require" + "github.com/vmihailenco/msgpack/v5" + "go-micro.dev/v4/store" +) + +// Like a NATS bucket with no bucket-wide TTL, this store preserves metadata but +// does not implement expiry supplied with individual writes. +type cacheWithoutTTL struct{ store.Store } + +func (s cacheWithoutTTL) Write(record *store.Record, _ ...store.WriteOption) error { + r := *record + r.Expiry = 0 + return s.Store.Write(&r) +} + +func TestCacheEnforcesExpiryWithoutNativeTTLSupport(t *testing.T) { + backing := cacheWithoutTTL{store.NewMemoryStore()} + cache := NewCache(backing, true) + now := time.Now() + cache.now = func() time.Time { return now } + expiresAt := now.Add(time.Hour) + record := &store.Record{Key: "record", Value: []byte("value"), Metadata: map[string]any{"original": "unchanged"}} + require.NoError(t, cache.Write(record, store.WriteExpiry(expiresAt))) + require.Equal(t, map[string]any{"original": "unchanged"}, record.Metadata) + + for range 3 { + now = now.Add(10 * time.Minute) + read, err := cache.Read(record.Key) + require.NoError(t, err) + require.Len(t, read, 1) + require.Equal(t, expiresAt.Sub(now), read[0].Expiry, "reads must not extend expiry") + } + now = expiresAt + read, err := cache.Read(record.Key) + require.NoError(t, err) + require.Empty(t, read) + _, err = backing.Read(record.Key) + require.ErrorIs(t, err, store.ErrNotFound) +} + +func TestCachePreservesTokenRevocationLifetime(t *testing.T) { + backing := cacheWithoutTTL{store.NewMemoryStore()} + cache := NewCache(backing, true) + now := time.Now() + cache.now = func() time.Time { return now } + expiresAt := now.Add(time.Hour) + key, err := NewTokenKey("alice", "session", "token") + require.NoError(t, err) + require.NoError(t, cache.Write(&store.Record{Key: key, Value: []byte("token")}, store.WriteExpiry(expiresAt))) + now = now.Add(time.Minute) + records, err := cache.Read(key) + require.NoError(t, err) + require.Len(t, records, 1) + now = now.Add(time.Minute) + require.NoError(t, RevokeToken(records[0], cache)) + revoked, err := IsTokenRevoked("token", cache) + require.NoError(t, err) + require.True(t, revoked) + + now = expiresAt.Add(-time.Second) + revoked, err = IsTokenRevoked("token", cache) + require.NoError(t, err) + require.True(t, revoked) + now = expiresAt + require.NoError(t, cache.collectExpired(context.Background())) + revoked, err = IsTokenRevoked("token", cache) + require.NoError(t, err) + require.False(t, revoked) +} + +func TestCacheRetainsRevocationsWithoutKnownExpiry(t *testing.T) { + cache := NewCache(cacheWithoutTTL{store.NewMemoryStore()}, true) + now := time.Now() + cache.now = func() time.Time { return now } + require.NoError(t, RevokeToken(&store.Record{Value: []byte("token")}, cache)) + now = now.AddDate(1, 0, 0) + require.NoError(t, cache.collectExpired(context.Background())) + revoked, err := IsTokenRevoked("token", cache) + require.NoError(t, err) + require.True(t, revoked) +} + +func TestCacheCleanupContinuesAfterCorruptEntry(t *testing.T) { + backing := cacheWithoutTTL{store.NewMemoryStore()} + cache := NewCache(backing, true) + now := time.Now() + cache.now = func() time.Time { return now } + require.NoError(t, cache.Write(&store.Record{Key: "expired", Expiry: time.Minute})) + require.NoError(t, backing.Write(&store.Record{Key: "corrupt", Metadata: map[string]any{expiryMetadataKey: "bad expiry"}})) + now = now.Add(time.Hour) + require.Error(t, cache.collectExpired(context.Background())) + _, err := backing.Read("expired") + require.ErrorIs(t, err, store.ErrNotFound) + _, err = backing.Read("corrupt") + require.NoError(t, err, "corrupt security data must not be silently dropped") +} + +func TestCacheCleanupStopsOnCancellation(t *testing.T) { + cache := NewCache(store.NewMemoryStore(), true) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { cache.CollectExpired(ctx, log.NopLogger()); close(done) }() + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("cache cleanup did not stop") + } +} + +func TestCacheCanDisableClaimsWithoutDisablingLogout(t *testing.T) { + cache := NewCache(store.NewMemoryStore(), false) + key := base64.URLEncoding.EncodeToString(make([]byte, 64)) + require.NoError(t, cache.Write(&store.Record{Key: key, Value: []byte("claims")})) + _, err := cache.Read(key) + require.ErrorIs(t, err, store.ErrNotFound) + require.NoError(t, RevokeToken(&store.Record{Value: []byte(key), Expiry: time.Hour}, cache)) + revoked, err := IsTokenRevoked(key, cache) + require.NoError(t, err) + require.True(t, revoked) +} + +func TestCacheMigratesEveryLegacyToken(t *testing.T) { + legacy := cacheWithoutTTL{store.NewMemoryStore()} + cache := NewCache(cacheWithoutTTL{store.NewMemoryStore()}, true) + now := time.Now() + cache.now = func() time.Time { return now } + var tokenKeys []string + for i := range 2 { + bytes := make([]byte, 64) + bytes[0] = byte(i) + key := base64.URLEncoding.EncodeToString(bytes) + tokenKeys = append(tokenKeys, key) + data, err := msgpack.Marshal(map[string]any{"sub": "alice", "sid": "session", "exp": now.Add(time.Hour).Unix()}) + require.NoError(t, err) + require.NoError(t, legacy.Write(&store.Record{Key: key, Value: data})) + lookupKey, err := NewKey("alice", "session") + require.NoError(t, err) + require.NoError(t, legacy.Write(&store.Record{Key: lookupKey, Value: []byte(key)})) + } + require.NoError(t, cache.IndexLegacyTokens(context.Background(), legacy)) + records, err := GetLogoutRecords(mustNewSuSe(t, "alice", "session"), cache) + require.NoError(t, err) + require.Len(t, records, 2) + for _, record := range records { + require.NoError(t, RevokeToken(record, cache)) + require.NoError(t, cache.Delete(record.Key)) + require.NoError(t, cache.Delete(string(record.Value))) + } + // Another proxy starting with the same old cache must not restore a logout. + require.NoError(t, cache.IndexLegacyTokens(context.Background(), legacy)) + for _, key := range tokenKeys { + revoked, err := IsTokenRevoked(key, cache) + require.NoError(t, err) + require.True(t, revoked) + _, err = cache.Read(key) + require.True(t, errors.Is(err, store.ErrNotFound)) + } + legacyKeys, err := legacy.List() + require.NoError(t, err) + require.Len(t, legacyKeys, 3, "migration must not modify a shared legacy bucket") +} + +func TestRevocationRejectsMalformedExpiry(t *testing.T) { + cache := NewCache(store.NewMemoryStore(), true) + err := RevokeToken(&store.Record{Value: []byte("token"), Metadata: map[string]any{expiryMetadataKey: strconv.Itoa(1) + "x"}}, cache) + require.Error(t, err) +} From 6055d34f2be44b85c728bcade46ffe03ef977c4c Mon Sep 17 00:00:00 2001 From: zerox80 Date: Mon, 7 Sep 2026 09:36:27 +0200 Subject: [PATCH 3/7] fix(proxy): revoke every tracked token on backchannel logout Register tokens synchronously before caching claims and check revocation before authentication and after registration. Revoke every token matching the logout subject or session independently of its cached claims, preserving the original token expiry. Notification failures must not skip invalidation. Add HTTP regressions for refreshes and userinfo bypass, concurrent claims writes, disabled claims caching, missing expiry, and storage failures. --- services/proxy/pkg/middleware/oidc_auth.go | 85 ++++---- .../pkg/staticroutes/backchannellogout.go | 30 ++- .../backchannellogout_concurrency_test.go | 148 +++++++++++++ .../backchannellogout_integration_test.go | 199 ++++++++++++++++++ 4 files changed, 421 insertions(+), 41 deletions(-) create mode 100644 services/proxy/pkg/staticroutes/backchannellogout_concurrency_test.go create mode 100644 services/proxy/pkg/staticroutes/backchannellogout_integration_test.go diff --git a/services/proxy/pkg/middleware/oidc_auth.go b/services/proxy/pkg/middleware/oidc_auth.go index b80f7fcb7a..33b62be1b1 100644 --- a/services/proxy/pkg/middleware/oidc_auth.go +++ b/services/proxy/pkg/middleware/oidc_auth.go @@ -16,7 +16,6 @@ import ( "github.com/opencloud-eu/opencloud/pkg/log" "github.com/opencloud-eu/opencloud/pkg/oidc" - "github.com/opencloud-eu/opencloud/services/proxy/pkg/config" bcl "github.com/opencloud-eu/opencloud/services/proxy/pkg/staticroutes/backchannellogout" ) @@ -62,6 +61,9 @@ func (m *OIDCAuthenticator) getClaims(token string, req *http.Request) (map[stri hash := make([]byte, 64) sha3.ShakeSum256(hash, []byte(token)) encodedHash := base64.URLEncoding.EncodeToString(hash) + if err := m.checkRevocation(encodedHash); err != nil { + return nil, false, err + } record, err := m.userInfoCache.Read(encodedHash) if err != nil && err != store.ErrNotFound { @@ -112,46 +114,46 @@ func (m *OIDCAuthenticator) getClaims(token string, req *http.Request) (map[stri return claims, true, nil } - go func() { - err = m.userInfoCache.Write(&store.Record{ - Key: encodedHash, - Value: d, - Expiry: time.Until(expiration), - }) - if err != nil { - m.Logger.Error().Err(err).Msg("failed to write to userinfo cache") - } - - // fail if creating the storage key fails, - // it means there is no subject and no session. - // - // ok: {key: ".sessionId"} - // ok: {key: "subject."} - // ok: {key: "subject.sessionId"} - // fail: {key: "."} - subjectSessionKey, err := bcl.NewKey(aClaims.Subject, aClaims.SessionID) - switch { - // fails if the verify method is set to `none`, in that case the oidc client verification returns - // an empty oidcclient.RegClaimsWithSID but no err. - // - // revisit once: - // - Authelia OpenID Connect Back-Channel Logout 1.0 is implemented, - // e.g. https://www.authelia.com/roadmap/active/openid-connect-1.0-provider/#beta-9 - case m.AccessTokenVerifyMethod == config.AccessTokenVerificationNone && errors.Is(err, bcl.ErrInvalidKey): - return - case err != nil: - m.Logger.Error().Err(err).Msg("failed to build subject.session") - return + // Register the token before making the claims reusable. Every token needs its + // own entry, including tokens issued by a refresh of an existing session. + subject := aClaims.Subject + if subject == "" { + subject, _ = claims["sub"].(string) + } + subjectSessionKey, err := bcl.NewTokenKey(subject, aClaims.SessionID, encodedHash) + if err != nil { + // Providers without a subject or session cannot support backchannel + // logout, but their existing authentication behavior remains unchanged. + m.Logger.Debug().Err(err).Msg("could not build session lookup key") + } + if err == nil { + var tokenTTL time.Duration + var writeOptions []store.WriteOption + if aClaims.ExpiresAt != nil { + tokenTTL = time.Until(aClaims.ExpiresAt.Time) + writeOptions = append(writeOptions, store.WriteExpiry(aClaims.ExpiresAt.Time)) } - + // A token with no trusted expiration can outlive the claims cache. + // Its logout record must therefore not use the claims-cache TTL. if err := m.userInfoCache.Write(&store.Record{ Key: subjectSessionKey, Value: []byte(encodedHash), - Expiry: time.Until(expiration), - }); err != nil { - m.Logger.Error().Err(err).Msg("failed to write session lookup cache") + Expiry: tokenTTL, + }, writeOptions...); err != nil { + return nil, false, errors.Wrap(err, "failed to write session lookup cache") } - }() + } + if err := m.userInfoCache.Write(&store.Record{ + Key: encodedHash, + Value: d, + Expiry: time.Until(expiration), + }, store.WriteExpiry(expiration)); err != nil { + m.Logger.Error().Err(err).Msg("failed to write to userinfo cache") + } + // A logout may have arrived while the token was being verified or cached. + if err := m.checkRevocation(encodedHash); err != nil { + return nil, false, err + } // If we get here this was a new login (or a renewal of the token) // add a flag about that to the claims, to be able to distinguish @@ -161,6 +163,17 @@ func (m *OIDCAuthenticator) getClaims(token string, req *http.Request) (map[stri return claims, true, nil } +func (m *OIDCAuthenticator) checkRevocation(tokenKey string) error { + revoked, err := bcl.IsTokenRevoked(tokenKey, m.userInfoCache) + if err != nil { + return errors.Wrap(err, "failed to read token revocation") + } + if revoked { + return errors.New("access token has been logged out") + } + return nil +} + // extractExpiration tries to extract the expriration time from the access token // If the access token does not have an exp claim it will fallback to the configured // default expiration diff --git a/services/proxy/pkg/staticroutes/backchannellogout.go b/services/proxy/pkg/staticroutes/backchannellogout.go index 83d864f03e..a12d4a00f7 100644 --- a/services/proxy/pkg/staticroutes/backchannellogout.go +++ b/services/proxy/pkg/staticroutes/backchannellogout.go @@ -74,7 +74,7 @@ func (s *StaticRouteHandler) backchannelLogout(w http.ResponseWriter, r *http.Re } lookupRecords, err := bcl.GetLogoutRecords(requestSubjectAndSession, s.UserInfoCache) - if errors.Is(err, microstore.ErrNotFound) || len(lookupRecords) == 0 { + if errors.Is(err, microstore.ErrNotFound) { render.Status(r, http.StatusOK) render.JSON(w, r, nil) return @@ -87,9 +87,28 @@ func (s *StaticRouteHandler) backchannelLogout(w http.ResponseWriter, r *http.Re return } + // A legacy lookup and a token-specific lookup can refer to the same token. + // Preserve the longest lifetime; zero means that no expiration is known. + tokens := make(map[string]*microstore.Record, len(lookupRecords)) for _, record := range lookupRecords { - // the record key is in the format "subject.session" or ".session" - // the record value is the key of the record that contains the claim in its value + previous := tokens[string(record.Value)] + if previous == nil || record.Expiry == 0 || (previous.Expiry != 0 && record.Expiry > previous.Expiry) { + tokens[string(record.Value)] = record + } + } + // Establish all revocations before sending notifications or clearing claims. + for _, record := range tokens { + if err := bcl.RevokeToken(record, s.UserInfoCache); err != nil { + msg := "failed to revoke token" + logger.Error().Err(err).Msg(msg) + render.Status(r, http.StatusBadRequest) + render.JSON(w, r, jse{Error: "invalid_request", ErrorDescription: msg}) + return + } + } + logoutEventPublished := false + for _, record := range lookupRecords { + // The lookup value refers to the claims cache record. key, value := record.Key, string(record.Value) subjectSession, err := bcl.NewSuSe(key) @@ -105,10 +124,11 @@ func (s *StaticRouteHandler) backchannelLogout(w http.ResponseWriter, r *http.Re continue } - if requestSubjectAndSession.Mode() == bcl.LogoutModeSession { + if requestSubjectAndSession.Mode() == bcl.LogoutModeSession && !logoutEventPublished { if err := s.publishBackchannelLogoutEvent(r.Context(), session, value); err != nil { s.Logger.Warn().Err(err).Msgf("failed to publish backchannel logout event for: %s", key) - continue + } else { + logoutEventPublished = true } } diff --git a/services/proxy/pkg/staticroutes/backchannellogout_concurrency_test.go b/services/proxy/pkg/staticroutes/backchannellogout_concurrency_test.go new file mode 100644 index 0000000000..3955027e19 --- /dev/null +++ b/services/proxy/pkg/staticroutes/backchannellogout_concurrency_test.go @@ -0,0 +1,148 @@ +package staticroutes_test + +import ( + "encoding/base64" + "errors" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + bcl "github.com/opencloud-eu/opencloud/services/proxy/pkg/staticroutes/backchannellogout" + "github.com/stretchr/testify/require" + "go-micro.dev/v4/store" + "golang.org/x/crypto/sha3" +) + +func TestBackchannelLogoutDuringClaimsWrite(t *testing.T) { + idp := newLogoutTestIDP(t) + token := idp.accessToken(t, "alice", "session", "token", time.Hour) + hash := make([]byte, 64) + sha3.ShakeSum256(hash, []byte(token)) + key := base64.URLEncoding.EncodeToString(hash) + started, release := make(chan struct{}), make(chan struct{}) + var unblock sync.Once + t.Cleanup(func() { unblock.Do(func() { close(release) }) }) + cache := &logoutInterceptStore{Store: bcl.NewCache(store.NewMemoryStore(), true)} + cache.write = func(record *store.Record) error { + if record.Key == key { + close(started) + <-release + } + return nil + } + auth, routes := idp.handlers(cache, true) + result := make(chan int, 1) + go func() { result <- logoutTestRequest(auth, token) }() + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("claims write did not start") + } + require.Equal(t, http.StatusOK, logoutTestLogout(routes, idp.logoutToken(t, "alice", "session")).Code) + unblock.Do(func() { close(release) }) + select { + case status := <-result: + require.Equal(t, http.StatusUnauthorized, status) + case <-time.After(5 * time.Second): + t.Fatal("authentication did not finish") + } + require.Equal(t, http.StatusUnauthorized, logoutTestRequest(auth, token), "the late claims write must not restore the token") +} + +func TestBackchannelLogoutWithoutClaimsCacheOrTokenExpiry(t *testing.T) { + for _, cacheClaims := range []bool{false, true} { + t.Run(map[bool]string{false: "claims disabled", true: "claims enabled"}[cacheClaims], func(t *testing.T) { + idp := newLogoutTestIDP(t) + cache := bcl.NewCache(store.NewMemoryStore(), cacheClaims) + auth, routes := idp.handlers(cache, true) + token := idp.sign(t, jwt.MapClaims{"iss": idp.server.URL, "sub": "alice", "sid": "session", "aud": "opencloud"}) + require.Equal(t, http.StatusOK, logoutTestRequest(auth, token)) + require.Equal(t, http.StatusOK, logoutTestLogout(routes, idp.logoutToken(t, "alice", "session")).Code) + require.Equal(t, http.StatusUnauthorized, logoutTestRequest(auth, token)) + }) + } +} + +func TestBackchannelLogoutStorageFailures(t *testing.T) { + idp := newLogoutTestIDP(t) + token := idp.accessToken(t, "alice", "session", "token", time.Hour) + logout := idp.logoutToken(t, "alice", "session") + failure := errors.New("store unavailable") + + t.Run("cannot register session", func(t *testing.T) { + cache := &logoutInterceptStore{Store: bcl.NewCache(store.NewMemoryStore(), true)} + cache.write = func(record *store.Record) error { + if strings.Contains(record.Key, ".") { + return failure + } + return nil + } + auth, _ := idp.handlers(cache, true) + require.Equal(t, http.StatusUnauthorized, logoutTestRequest(auth, token)) + }) + t.Run("cannot check revocation", func(t *testing.T) { + cache := &logoutInterceptStore{Store: bcl.NewCache(store.NewMemoryStore(), true)} + auth, _ := idp.handlers(cache, true) + require.Equal(t, http.StatusOK, logoutTestRequest(auth, token)) + cache.read = func(key string) error { + if strings.HasPrefix(key, "revoked/") { + return failure + } + return nil + } + require.Equal(t, http.StatusUnauthorized, logoutTestRequest(auth, token)) + }) + for _, operation := range []string{"lookup", "revoke"} { + t.Run(operation, func(t *testing.T) { + cache := &logoutInterceptStore{Store: bcl.NewCache(store.NewMemoryStore(), true)} + auth, routes := idp.handlers(cache, true) + require.Equal(t, http.StatusOK, logoutTestRequest(auth, token)) + if operation == "lookup" { + cache.read = func(key string) error { + if strings.Contains(key, ".") { + return failure + } + return nil + } + } else { + cache.write = func(record *store.Record) error { + if strings.HasPrefix(record.Key, "revoked/") { + return failure + } + return nil + } + } + require.Equal(t, http.StatusBadRequest, logoutTestLogout(routes, logout).Code, "a storage failure must not report a successful logout") + cache.read, cache.write = nil, nil + require.Equal(t, http.StatusOK, logoutTestLogout(routes, logout).Code, "the lookup must survive so the logout can be retried") + require.Equal(t, http.StatusUnauthorized, logoutTestRequest(auth, token)) + }) + } +} + +type logoutInterceptStore struct { + store.Store + read func(string) error + write func(*store.Record) error +} + +func (s *logoutInterceptStore) Read(key string, opts ...store.ReadOption) ([]*store.Record, error) { + if s.read != nil { + if err := s.read(key); err != nil { + return nil, err + } + } + return s.Store.Read(key, opts...) +} + +func (s *logoutInterceptStore) Write(record *store.Record, opts ...store.WriteOption) error { + if s.write != nil { + if err := s.write(record); err != nil { + return err + } + } + return s.Store.Write(record, opts...) +} diff --git a/services/proxy/pkg/staticroutes/backchannellogout_integration_test.go b/services/proxy/pkg/staticroutes/backchannellogout_integration_test.go new file mode 100644 index 0000000000..885f56c8b4 --- /dev/null +++ b/services/proxy/pkg/staticroutes/backchannellogout_integration_test.go @@ -0,0 +1,199 @@ +package staticroutes_test + +import ( + "crypto/rand" + "crypto/rsa" + "encoding/base64" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/pkg/oidc" + "github.com/opencloud-eu/opencloud/services/proxy/pkg/config/defaults" + "github.com/opencloud-eu/opencloud/services/proxy/pkg/middleware" + "github.com/opencloud-eu/opencloud/services/proxy/pkg/router" + "github.com/opencloud-eu/opencloud/services/proxy/pkg/staticroutes" + "github.com/stretchr/testify/require" + "go-micro.dev/v4/store" +) + +func TestBackchannelLogoutInvalidatesEveryCachedToken(t *testing.T) { + for _, skipUserInfo := range []bool{false, true} { + for _, mode := range []string{"subject", "session", "subject and session"} { + t.Run(fmt.Sprintf("skip_user_info=%t/%s", skipUserInfo, mode), func(t *testing.T) { + idp := newLogoutTestIDP(t) + cache := &logoutTestCache{Store: store.NewMemoryStore(), writes: make(chan string, 32)} + auth, routes := idp.handlers(cache, skipUserInfo) + first := idp.accessToken(t, "alice", "session-a", "first", time.Hour) + second := idp.accessToken(t, "alice", "session-a", "second", 5*time.Minute) + otherSubject := "alice" + if mode == "subject" { + otherSubject = "bob" + } + unrelated := idp.accessToken(t, otherSubject, "session-b", "unrelated", time.Hour) + for _, token := range []string{first, second, unrelated} { + require.Equal(t, http.StatusOK, logoutTestRequest(auth, token)) + cache.waitForSession(t) + } + subject, session := "alice", "session-a" + if mode == "subject" { + session = "" + } else if mode == "session" { + subject = "" + } + logout := idp.logoutToken(t, subject, session) + require.Equal(t, http.StatusOK, logoutTestLogout(routes, logout).Code) + before := idp.userinfoRequests.Load() + for _, token := range []string{first, second} { + require.Equal(t, http.StatusUnauthorized, logoutTestRequest(auth, token), "all previously accepted tokens of the logged-out session must be rejected") + } + require.Equal(t, before, idp.userinfoRequests.Load(), "revoked tokens must be rejected before requesting userinfo") + require.Equal(t, http.StatusOK, logoutTestRequest(auth, unrelated), "unrelated sessions must remain authenticated") + // Repeated logout callbacks must succeed without re-enabling tokens. + require.Equal(t, http.StatusOK, logoutTestLogout(routes, logout).Code) + require.Equal(t, http.StatusUnauthorized, logoutTestRequest(auth, first)) + // Sharing a persistent store with another proxy must retain revocations. + otherAuth, _ := idp.handlers(cache, skipUserInfo) + require.Equal(t, http.StatusUnauthorized, logoutTestRequest(otherAuth, second)) + }) + } + } +} + +type logoutTestIDP struct { + server *httptest.Server + key *rsa.PrivateKey + userinfoRequests atomic.Int64 +} + +func newLogoutTestIDP(t *testing.T) *logoutTestIDP { + t.Helper() + key, err := rsa.GenerateKey(rand.Reader, 2048) + require.NoError(t, err) + idp := &logoutTestIDP{key: key} + mux := http.NewServeMux() + idp.server = httptest.NewServer(mux) + t.Cleanup(idp.server.Close) + mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "issuer": idp.server.URL, "jwks_uri": idp.server.URL + "/jwks", "userinfo_endpoint": idp.server.URL + "/userinfo", + "id_token_signing_alg_values_supported": []string{"RS256"}, + }) + }) + mux.HandleFunc("/jwks", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"keys": []any{map[string]any{ + "kty": "RSA", "kid": "test", "alg": "RS256", "use": "sig", + "n": base64.RawURLEncoding.EncodeToString(key.N.Bytes()), "e": "AQAB", + }}}) + }) + mux.HandleFunc("/userinfo", func(w http.ResponseWriter, r *http.Request) { + idp.userinfoRequests.Add(1) + claims := jwt.MapClaims{} + _, _, err := new(jwt.Parser).ParseUnverified(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer "), claims) + if err != nil { + w.WriteHeader(http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"sub": claims["sub"], "preferred_username": claims["sub"]}) + }) + return idp +} + +func (idp *logoutTestIDP) sign(t *testing.T, claims jwt.MapClaims) string { + t.Helper() + token := jwt.NewWithClaims(jwt.SigningMethodRS256, claims) + token.Header["kid"] = "test" + signed, err := token.SignedString(idp.key) + require.NoError(t, err) + return signed +} + +func (idp *logoutTestIDP) accessToken(t *testing.T, subject, session, id string, lifetime time.Duration) string { + return idp.sign(t, jwt.MapClaims{ + "iss": idp.server.URL, "sub": subject, "sid": session, "aud": "opencloud", "jti": id, + "iat": time.Now().Unix(), "exp": time.Now().Add(lifetime).Unix(), + }) +} + +func (idp *logoutTestIDP) logoutToken(t *testing.T, subject, session string) string { + return idp.sign(t, jwt.MapClaims{ + "iss": idp.server.URL, "sub": subject, "sid": session, "aud": "opencloud", "jti": "logout", + "iat": time.Now().Unix(), "exp": time.Now().Add(time.Minute).Unix(), + "events": map[string]any{"http://schemas.openid.net/event/backchannel-logout": map[string]any{}}, + }) +} + +func (idp *logoutTestIDP) handlers(cache store.Store, skipUserInfo bool) (middleware.Authenticator, http.Handler) { + cfg := defaults.FullDefaultConfig() + cfg.OIDC.Issuer = idp.server.URL + cfg.OIDC.SkipUserInfo = skipUserInfo + client := oidc.NewOIDCClient(oidc.WithOidcIssuer(idp.server.URL), oidc.WithHTTPClient(idp.server.Client()), oidc.WithAccessTokenVerifyMethod("jwt"), oidc.WithLogger(log.NopLogger())) + auth := middleware.NewOIDCAuthenticator( + middleware.Logger(log.NopLogger()), middleware.UserInfoCache(cache), middleware.OIDCClient(client), + middleware.OIDCIss(idp.server.URL), middleware.SkipUserInfo(skipUserInfo), + middleware.AccessTokenVerifyMethod("jwt"), middleware.DefaultAccessTokenTTL(time.Minute), + middleware.HTTPClient(idp.server.Client()), + ) + routes := &staticroutes.StaticRouteHandler{ + Prefix: "/", Config: *cfg, Logger: log.NopLogger(), OidcClient: client, UserInfoCache: cache, Proxy: http.NotFoundHandler(), + } + return auth, routes.Handler() +} + +func logoutTestRequest(auth middleware.Authenticator, token string) int { + handler := middleware.Authentication([]middleware.Authenticator{auth})(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) + req := httptest.NewRequest(http.MethodGet, "/protected", http.NoBody) + req = req.WithContext(router.SetRoutingInfo(req.Context(), router.RoutingInfo{})) + req.Header.Set("Authorization", "Bearer "+token) + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, req) + return recorder.Code +} + +func logoutTestLogout(routes http.Handler, token string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, "/backchannel_logout", strings.NewReader(url.Values{"logout_token": {token}}.Encode())) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + recorder := httptest.NewRecorder() + routes.ServeHTTP(recorder, req) + return recorder +} + +type logoutTestCache struct { + store.Store + writes chan string +} + +func (cache *logoutTestCache) Write(record *store.Record, opts ...store.WriteOption) error { + err := cache.Store.Write(record, opts...) + if err == nil { + cache.writes <- record.Key + } + return err +} + +func (cache *logoutTestCache) waitForSession(t *testing.T) { + t.Helper() + timer := time.NewTimer(5 * time.Second) + defer timer.Stop() + for { + select { + case key := <-cache.writes: + if strings.Contains(key, ".") { + return + } + case <-timer.C: + t.Fatal("timed out waiting for the session lookup to be written") + } + } +} From 6c4da8b389aaf63331e62f72ceaff2d963f5e405 Mon Sep 17 00:00:00 2001 From: zerox80 Date: Mon, 7 Sep 2026 09:36:28 +0200 Subject: [PATCH 4/7] docs(proxy): explain logout persistence and cache upgrades [docs-only] Document the dedicated OIDC namespace, legacy cache migration, shared-store requirements, expiry cleanup, and retention for tokens without verified expiry. Add the changelog entry for the complete backchannel logout fix. --- .../unreleased/fix-backchannel-logout-all-tokens.md | 12 ++++++++++++ services/proxy/README.md | 8 +++++++- 2 files changed, 19 insertions(+), 1 deletion(-) create mode 100644 changelog/unreleased/fix-backchannel-logout-all-tokens.md diff --git a/changelog/unreleased/fix-backchannel-logout-all-tokens.md b/changelog/unreleased/fix-backchannel-logout-all-tokens.md new file mode 100644 index 0000000000..b7bcc5db60 --- /dev/null +++ b/changelog/unreleased/fix-backchannel-logout-all-tokens.md @@ -0,0 +1,12 @@ +Bugfix: Revoke every cached token during backchannel logout + +Backchannel logout now invalidates all accepted access tokens associated with +the requested session or subject, including tokens issued before a refresh. +Revoked tokens cannot authenticate again through local JWT verification when +userinfo lookup is disabled, and concurrent claims writes cannot overwrite a +revocation. Notification failures no longer prevent token invalidation. + +OIDC logout state uses a dedicated cache namespace with per-record expiry and +migrates existing persistent claims on startup. Tokens without a verified +expiry retain their logout state indefinitely. See the proxy caching +documentation for persistence and upgrade details. diff --git a/services/proxy/README.md b/services/proxy/README.md index 16030f7530..5c2740dbdc 100644 --- a/services/proxy/README.md +++ b/services/proxy/README.md @@ -247,7 +247,7 @@ The `proxy` service can use a configured store via `PROXY_OIDC_USERINFO_CACHE_ST - `memory`: Basic in-memory store and the default. - `redis-sentinel`: Stores data in a configured Redis Sentinel cluster. - `nats-js-kv`: Stores data using key-value-store feature of [nats jetstream](https://docs.nats.io/nats-concepts/jetstream/key-value-store) - - `noop`: Stores nothing. Useful for testing. Not recommended in production environments. + - `noop`: Disables claims caching. Logout state is kept in memory. Useful for testing. Not recommended in production environments. Other store types may work but are not supported currently. @@ -260,6 +260,12 @@ Store specific notes: - When using `nats-js-kv` it is recommended to set `OC_CACHE_STORE_NODES` to the same value as `OC_EVENTS_ENDPOINT`. That way the cache uses the same nats instance as the event bus. - When using the `nats-js-kv` store, it is possible to set `OC_CACHE_DISABLE_PERSISTENCE` to instruct nats to not persist cache data on disc. +Backchannel logout tracks each accepted access token separately and rejects revoked tokens even when `PROXY_OIDC_SKIP_USER_INFO` is enabled. Revocations are kept until the verified token expiry. Tokens without a verified expiry, including tokens imported from the legacy claims cache, require logout state with no automatic expiry. The claims cache continues to use the configured TTL when the token has no expiry. + +OIDC records use the configured database with the suffix `-oidc-v2` and the configured table with the suffix `/oidc-v2/`. This separates logout state from legacy caches and their bucket-wide TTL. On startup, persistent stores import unexpired legacy claims into the new session index. All proxy instances must use the updated code and the same persistent store to share logout decisions; memory-backed logout state is lost when the process stops. Keep the persistent OIDC namespace when clearing ordinary caches. + +For NATS, the new bucket has no bucket-wide TTL. The proxy enforces each record's expiry and removes expired records every minute. The deprecated `ocmem` option uses a dedicated memory store for OIDC state to prevent capacity eviction of revocations. + ## Presigned Urls From 147336c326bb25c9964ba32155a91bc537e4a4e2 Mon Sep 17 00:00:00 2001 From: zerox80 Date: Mon, 7 Sep 2026 10:08:32 +0200 Subject: [PATCH 5/7] fix(proxy): accept empty NATS caches during logout migration --- .../proxy/pkg/command/oidc_cache_nats_test.go | 64 +++++++++++++++++++ .../staticroutes/backchannellogout/cache.go | 21 +++++- .../backchannellogout/cache_test.go | 34 ++++++++++ 3 files changed, 116 insertions(+), 3 deletions(-) create mode 100644 services/proxy/pkg/command/oidc_cache_nats_test.go diff --git a/services/proxy/pkg/command/oidc_cache_nats_test.go b/services/proxy/pkg/command/oidc_cache_nats_test.go new file mode 100644 index 0000000000..b3dcf2b443 --- /dev/null +++ b/services/proxy/pkg/command/oidc_cache_nats_test.go @@ -0,0 +1,64 @@ +package command + +import ( + "context" + "errors" + "testing" + "time" + + nserver "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/opencloud-eu/opencloud/services/proxy/pkg/config" + bcl "github.com/opencloud-eu/opencloud/services/proxy/pkg/staticroutes/backchannellogout" + "github.com/stretchr/testify/require" + "go-micro.dev/v4/store" +) + +func newOIDCNATSTestConfig(t *testing.T) (*config.Cache, nats.JetStreamContext) { + t.Helper() + server, err := nserver.NewServer(&nserver.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir()}) + require.NoError(t, err) + go server.Start() + t.Cleanup(func() { server.Shutdown(); server.WaitForShutdown() }) + require.True(t, server.ReadyForConnections(5*time.Second)) + conn, err := nats.Connect(server.ClientURL()) + require.NoError(t, err) + t.Cleanup(conn.Close) + js, err := conn.JetStream() + require.NoError(t, err) + return &config.Cache{Store: "nats-js-kv", Database: "cache-userinfo", Nodes: []string{server.ClientURL()}, TTL: time.Minute}, js +} + +func TestOIDCCacheMigratesEmptyNATS(t *testing.T) { + for _, state := range []string{"new bucket", "expired entries", "delete markers only"} { + t.Run(state, func(t *testing.T) { + cfg, js := newOIDCNATSTestConfig(t) + if state == "expired entries" { + cfg.TTL = 100 * time.Millisecond + } + legacy := newUserInfoStore(cfg, cfg.Store, cfg.Database, cfg.Table, cfg.TTL) + t.Cleanup(func() { _ = legacy.Close() }) + if state != "new bucket" { + require.NoError(t, legacy.Write(&store.Record{Key: "old", Value: []byte("old claims")})) + if state == "delete markers only" { + require.NoError(t, legacy.Delete("old")) + } + bucket, err := js.KeyValue(cfg.Database) + require.NoError(t, err) + require.Eventually(t, func() bool { + _, err := bucket.Keys() + return errors.Is(err, nats.ErrNoKeysFound) + }, 3*time.Second, 10*time.Millisecond) + } + cache := newUserInfoCache(cfg) + t.Cleanup(func() { _ = cache.Close() }) + require.NoError(t, migrateUserInfoCache(context.Background(), cache, cfg), "empty legacy caches must not prevent proxy startup") + key, err := bcl.NewKey("alice", "session") + require.NoError(t, err) + session, err := bcl.NewSuSe(key) + require.NoError(t, err) + _, err = bcl.GetLogoutRecords(session, cache) + require.ErrorIs(t, err, store.ErrNotFound, "an empty current cache must behave as an already logged-out session") + }) + } +} diff --git a/services/proxy/pkg/staticroutes/backchannellogout/cache.go b/services/proxy/pkg/staticroutes/backchannellogout/cache.go index 100064a841..af79494d3e 100644 --- a/services/proxy/pkg/staticroutes/backchannellogout/cache.go +++ b/services/proxy/pkg/staticroutes/backchannellogout/cache.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "github.com/nats-io/nats.go" "github.com/opencloud-eu/opencloud/pkg/log" "github.com/vmihailenco/msgpack/v5" "go-micro.dev/v4/store" @@ -29,6 +30,20 @@ func NewCache(underlying store.Store, cacheClaims bool) *Cache { return &Cache{Store: underlying, cacheClaims: cacheClaims, now: time.Now} } +// List treats an empty NATS bucket like an empty store. The NATS adapter wraps +// ErrNoKeysFound when the bucket is new, expired, or contains only delete markers. +func (c *Cache) List(opts ...store.ListOption) ([]string, error) { + return listCacheKeys(c.Store, opts...) +} + +func listCacheKeys(cache store.Store, opts ...store.ListOption) ([]string, error) { + keys, err := cache.List(opts...) + if errors.Is(err, nats.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + func isClaimsKey(key string) bool { decoded, err := keyEncoding.DecodeString(key) return err == nil && len(decoded) == 64 @@ -139,7 +154,7 @@ func (c *Cache) readRecords(key string, opts ...store.ReadOption) ([]*store.Reco if options.Table != "" { from.Table = options.Table } - keys, err := c.Store.List(store.ListFrom(from.Database, from.Table)) + keys, err := c.List(store.ListFrom(from.Database, from.Table)) if err != nil { return nil, err } @@ -194,7 +209,7 @@ func (c *Cache) expiry(record *store.Record) (time.Time, error) { // requests. A legacy session lookup contains only the last token's hash, so it // is not sufficient to invalidate every token already present in the cache. func (c *Cache) IndexLegacyTokens(ctx context.Context, legacy store.Store) error { - keys, err := legacy.List() + keys, err := listCacheKeys(legacy) if err != nil { return err } @@ -278,7 +293,7 @@ func (c *Cache) CollectExpired(ctx context.Context, logger log.Logger) { } func (c *Cache) collectExpired(ctx context.Context) error { - keys, err := c.Store.List() + keys, err := c.List() if err != nil { return err } diff --git a/services/proxy/pkg/staticroutes/backchannellogout/cache_test.go b/services/proxy/pkg/staticroutes/backchannellogout/cache_test.go index af499b2bcd..15fe21c9ca 100644 --- a/services/proxy/pkg/staticroutes/backchannellogout/cache_test.go +++ b/services/proxy/pkg/staticroutes/backchannellogout/cache_test.go @@ -4,10 +4,12 @@ import ( "context" "encoding/base64" "errors" + "fmt" "strconv" "testing" "time" + "github.com/nats-io/nats.go" "github.com/opencloud-eu/opencloud/pkg/log" "github.com/stretchr/testify/require" "github.com/vmihailenco/msgpack/v5" @@ -177,3 +179,35 @@ func TestRevocationRejectsMalformedExpiry(t *testing.T) { err := RevokeToken(&store.Record{Value: []byte("token"), Metadata: map[string]any{expiryMetadataKey: strconv.Itoa(1) + "x"}}, cache) require.Error(t, err) } + +func TestCacheListErrors(t *testing.T) { + for _, tt := range []struct { + name string + err error + }{ + {"empty NATS bucket", nats.ErrNoKeysFound}, + {"connection failure", errors.New("connection unavailable")}, + } { + t.Run(tt.name, func(t *testing.T) { + backing := cacheListFailure{Store: store.NewMemoryStore(), err: fmt.Errorf("list failed: %w", tt.err)} + cache := NewCache(backing, true) + _, listErr := cache.List() + migrationErr := cache.IndexLegacyTokens(context.Background(), backing) + cleanupErr := cache.collectExpired(context.Background()) + for _, err := range []error{listErr, migrationErr, cleanupErr} { + if tt.err == nats.ErrNoKeysFound { + require.NoError(t, err) + } else { + require.ErrorIs(t, err, tt.err, "real storage failures must remain visible") + } + } + }) + } +} + +type cacheListFailure struct { + store.Store + err error +} + +func (s cacheListFailure) List(...store.ListOption) ([]string, error) { return nil, s.err } From 4653dbaf775eff3943d277cd9d17f6819f4b65f0 Mon Sep 17 00:00:00 2001 From: zerox80 Date: Mon, 7 Sep 2026 10:17:07 +0200 Subject: [PATCH 6/7] fix(proxy): purge expired NATS cache delete markers safely --- .../fix-backchannel-logout-all-tokens.md | 6 +- services/proxy/README.md | 2 +- services/proxy/pkg/command/oidc_cache.go | 11 +- services/proxy/pkg/command/oidc_cache_nats.go | 31 ++++ .../proxy/pkg/command/oidc_cache_nats_test.go | 33 +++- .../staticroutes/backchannellogout/cache.go | 3 + .../staticroutes/backchannellogout/nats.go | 98 ++++++++++++ .../backchannellogout/nats_test.go | 144 ++++++++++++++++++ 8 files changed, 321 insertions(+), 7 deletions(-) create mode 100644 services/proxy/pkg/command/oidc_cache_nats.go create mode 100644 services/proxy/pkg/staticroutes/backchannellogout/nats.go create mode 100644 services/proxy/pkg/staticroutes/backchannellogout/nats_test.go diff --git a/changelog/unreleased/fix-backchannel-logout-all-tokens.md b/changelog/unreleased/fix-backchannel-logout-all-tokens.md index b7bcc5db60..e6321c6684 100644 --- a/changelog/unreleased/fix-backchannel-logout-all-tokens.md +++ b/changelog/unreleased/fix-backchannel-logout-all-tokens.md @@ -7,6 +7,8 @@ userinfo lookup is disabled, and concurrent claims writes cannot overwrite a revocation. Notification failures no longer prevent token invalidation. OIDC logout state uses a dedicated cache namespace with per-record expiry and -migrates existing persistent claims on startup. Tokens without a verified -expiry retain their logout state indefinitely. See the proxy caching +migrates existing persistent claims on startup, accepting empty legacy caches. +NATS delete markers are physically removed without deleting concurrent writes. +Tokens without a verified expiry retain their logout state indefinitely. +See the proxy caching documentation for persistence and upgrade details. diff --git a/services/proxy/README.md b/services/proxy/README.md index 5c2740dbdc..3ae54a1b3e 100644 --- a/services/proxy/README.md +++ b/services/proxy/README.md @@ -264,7 +264,7 @@ Backchannel logout tracks each accepted access token separately and rejects revo OIDC records use the configured database with the suffix `-oidc-v2` and the configured table with the suffix `/oidc-v2/`. This separates logout state from legacy caches and their bucket-wide TTL. On startup, persistent stores import unexpired legacy claims into the new session index. All proxy instances must use the updated code and the same persistent store to share logout decisions; memory-backed logout state is lost when the process stops. Keep the persistent OIDC namespace when clearing ordinary caches. -For NATS, the new bucket has no bucket-wide TTL. The proxy enforces each record's expiry and removes expired records every minute. The deprecated `ocmem` option uses a dedicated memory store for OIDC state to prevent capacity eviction of revocations. +For NATS, the new bucket has no bucket-wide TTL. The proxy enforces each record's expiry and removes expired records and their delete markers every minute. Marker cleanup is limited to the proxy's table and the observed revisions, preserving concurrent writes. An empty legacy bucket requires no migration. The deprecated `ocmem` option uses a dedicated memory store for OIDC state to prevent capacity eviction of revocations. ## Presigned Urls diff --git a/services/proxy/pkg/command/oidc_cache.go b/services/proxy/pkg/command/oidc_cache.go index f7df22d6cf..4bad175107 100644 --- a/services/proxy/pkg/command/oidc_cache.go +++ b/services/proxy/pkg/command/oidc_cache.go @@ -4,6 +4,7 @@ import ( "context" "time" + "github.com/nats-io/nats.go" "github.com/opencloud-eu/opencloud/services/proxy/pkg/config" bcl "github.com/opencloud-eu/opencloud/services/proxy/pkg/staticroutes/backchannellogout" "github.com/opencloud-eu/reva/v2/pkg/store" @@ -25,7 +26,15 @@ func newUserInfoCache(cfg *config.Cache) *bcl.Cache { // Keep the no-TTL bucket separate from other services and older proxies // which may use the configured database with a bucket-wide TTL. // Redis ignores Database, so give its table a dedicated prefix as well. - return bcl.NewCache(newUserInfoStore(cfg, storeType, database+"-oidc-v2", cfg.Table+"/oidc-v2/", 0), cacheClaims) + backend := newUserInfoStore(cfg, storeType, database+"-oidc-v2", cfg.Table+"/oidc-v2/", 0) + if storeType == store.TypeNatsJSKV { + cleanupConfig := *cfg + cleanupConfig.Nodes = append([]string(nil), cfg.Nodes...) + backend = bcl.WithNATSCleanup(backend, func() (*nats.Conn, error) { + return connectOIDCNATSCache(&cleanupConfig) + }) + } + return bcl.NewCache(backend, cacheClaims) } func migrateUserInfoCache(ctx context.Context, cache *bcl.Cache, cfg *config.Cache) error { diff --git a/services/proxy/pkg/command/oidc_cache_nats.go b/services/proxy/pkg/command/oidc_cache_nats.go new file mode 100644 index 0000000000..9990dc4cf9 --- /dev/null +++ b/services/proxy/pkg/command/oidc_cache_nats.go @@ -0,0 +1,31 @@ +package command + +import ( + "crypto/tls" + + "github.com/nats-io/nats.go" + "github.com/opencloud-eu/opencloud/services/proxy/pkg/config" +) + +// connectOIDCNATSCache mirrors the NATS authentication and TLS options used by +// reva's store factory. Cleanup needs the same access as ordinary cache writes. +func connectOIDCNATSCache(cfg *config.Cache) (*nats.Conn, error) { + opts := nats.GetDefaultOptions() + opts.Name = "opencloud-proxy-oidc-cleanup" + opts.Servers = cfg.Nodes + opts.User, opts.Password = cfg.AuthUsername, cfg.AuthPassword + if cfg.EnableTLS { + if cfg.TLSRootCACertificate != "" { + if err := nats.RootCAs(cfg.TLSRootCACertificate)(&opts); err != nil { + return nil, err + } + } else { + if err := nats.Secure(&tls.Config{ + MinVersion: tls.VersionTLS12, InsecureSkipVerify: cfg.TLSInsecure, //nolint:gosec + })(&opts); err != nil { + return nil, err + } + } + } + return opts.Connect() +} diff --git a/services/proxy/pkg/command/oidc_cache_nats_test.go b/services/proxy/pkg/command/oidc_cache_nats_test.go index b3dcf2b443..802c7f5a06 100644 --- a/services/proxy/pkg/command/oidc_cache_nats_test.go +++ b/services/proxy/pkg/command/oidc_cache_nats_test.go @@ -16,17 +16,23 @@ import ( func newOIDCNATSTestConfig(t *testing.T) (*config.Cache, nats.JetStreamContext) { t.Helper() - server, err := nserver.NewServer(&nserver.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir()}) + username, password := "oidc-test", t.Name() + server, err := nserver.NewServer(&nserver.Options{ + Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), Username: username, Password: password, + }) require.NoError(t, err) go server.Start() t.Cleanup(func() { server.Shutdown(); server.WaitForShutdown() }) require.True(t, server.ReadyForConnections(5*time.Second)) - conn, err := nats.Connect(server.ClientURL()) + conn, err := nats.Connect(server.ClientURL(), nats.UserInfo(username, password)) require.NoError(t, err) t.Cleanup(conn.Close) js, err := conn.JetStream() require.NoError(t, err) - return &config.Cache{Store: "nats-js-kv", Database: "cache-userinfo", Nodes: []string{server.ClientURL()}, TTL: time.Minute}, js + return &config.Cache{ + Store: "nats-js-kv", Database: "cache-userinfo", Nodes: []string{server.ClientURL()}, TTL: time.Minute, + AuthUsername: username, AuthPassword: password, + }, js } func TestOIDCCacheMigratesEmptyNATS(t *testing.T) { @@ -62,3 +68,24 @@ func TestOIDCCacheMigratesEmptyNATS(t *testing.T) { }) } } + +func TestOIDCCacheWiresNATSMarkerCleanup(t *testing.T) { + cfg, js := newOIDCNATSTestConfig(t) + cache := newUserInfoCache(cfg) + t.Cleanup(func() { _ = cache.Close() }) + require.NoError(t, cache.Write(&store.Record{Key: "deleted"})) + require.NoError(t, cache.Delete("deleted")) + legacy := newUserInfoStore(cfg, cfg.Store, cfg.Database, cfg.Table, cfg.TTL) + t.Cleanup(func() { _ = legacy.Close() }) + require.NoError(t, legacy.Write(&store.Record{Key: "legacy"})) + require.NoError(t, legacy.Delete("legacy")) + cleaner, ok := cache.Store.(interface{ PurgeDeleted(context.Context) error }) + require.True(t, ok, "the configured NATS store must support marker maintenance") + require.NoError(t, cleaner.PurgeDeleted(context.Background()), "cleanup must use the configured authentication") + info, err := js.StreamInfo("KV_cache-userinfo-oidc-v2") + require.NoError(t, err) + require.Zero(t, info.State.Msgs) + info, err = js.StreamInfo("KV_cache-userinfo") + require.NoError(t, err) + require.EqualValues(t, 1, info.State.Msgs, "cleanup must leave the legacy bucket untouched") +} diff --git a/services/proxy/pkg/staticroutes/backchannellogout/cache.go b/services/proxy/pkg/staticroutes/backchannellogout/cache.go index af79494d3e..f8892e3e6f 100644 --- a/services/proxy/pkg/staticroutes/backchannellogout/cache.go +++ b/services/proxy/pkg/staticroutes/backchannellogout/cache.go @@ -328,5 +328,8 @@ func (c *Cache) collectExpired(ctx context.Context) error { } } } + if cleaner, ok := c.Store.(interface{ PurgeDeleted(context.Context) error }); ok { + cleanupErr = errors.Join(cleanupErr, cleaner.PurgeDeleted(ctx)) + } return cleanupErr } diff --git a/services/proxy/pkg/staticroutes/backchannellogout/nats.go b/services/proxy/pkg/staticroutes/backchannellogout/nats.go new file mode 100644 index 0000000000..b2692ee9f0 --- /dev/null +++ b/services/proxy/pkg/staticroutes/backchannellogout/nats.go @@ -0,0 +1,98 @@ +package backchannellogout + +import ( + "context" + "errors" + "time" + + "github.com/nats-io/nats.go" + "go-micro.dev/v4/store" +) + +type natsCacheStore struct { + store.Store + connect func() (*nats.Conn, error) +} + +// WithNATSCleanup adds delete-marker maintenance to a NATS KV store. The +// connection factory must use the same endpoints, authentication, and TLS +// settings as the underlying store. Each maintenance connection is closed. +func WithNATSCleanup(underlying store.Store, connect func() (*nats.Conn, error)) store.Store { + return &natsCacheStore{Store: underlying, connect: connect} +} + +// PurgeDeleted removes tombstones from this store's table. They otherwise remain +// forever in the OIDC bucket, which deliberately has no bucket-wide TTL. +func (s *natsCacheStore) PurgeDeleted(ctx context.Context) error { + ctx, cancel := context.WithTimeout(ctx, time.Minute) + defer cancel() + if err := ctx.Err(); err != nil { + return err + } + conn, err := s.connect() + if err != nil { + return err + } + defer conn.Close() + js, err := conn.JetStream(nats.Context(ctx)) + if err != nil { + return err + } + return s.purgeDeleted(ctx, js) +} + +func (s *natsCacheStore) purgeDeleted(ctx context.Context, js nats.JetStreamContext) error { + // Use the adapter's decoder so cleanup respects its key encoding and table + // namespace without duplicating the encoding implementation. + keys, ok := s.Store.(interface { + MicroKeyFilter(table, natsKey, prefix, suffix string) (string, bool) + }) + if !ok { + return errors.New("NATS cache store does not expose its key decoder") + } + opts := s.Options() + bucket, err := js.KeyValue(opts.Database) + if err != nil { + return err + } + watcher, err := bucket.WatchAll(nats.MetaOnly(), nats.Context(ctx)) + if err != nil { + return err + } + defer watcher.Stop() + var markers []nats.KeyValueEntry +snapshot: + for { + select { + case <-ctx.Done(): + return ctx.Err() + case entry, open := <-watcher.Updates(): + if !open { + return errors.New("NATS cache marker snapshot ended before completion") + } + if entry == nil { + break snapshot + } + if entry.Operation() != nats.KeyValueDelete && entry.Operation() != nats.KeyValuePurge { + continue + } + if _, matches := keys.MicroKeyFilter(opts.Table, entry.Key(), "", ""); matches { + markers = append(markers, entry) + } + } + } + _ = watcher.Stop() + for _, marker := range markers { + if err := ctx.Err(); err != nil { + return err + } + // Bound the purge by the observed revision. A new value written after + // the snapshot must survive, even if it reuses a deleted token's key. + if err := js.PurgeStream("KV_"+opts.Database, &nats.StreamPurgeRequest{ + Subject: "$KV." + opts.Database + "." + marker.Key(), Sequence: marker.Revision() + 1, + }); err != nil { + return err + } + } + return nil +} diff --git a/services/proxy/pkg/staticroutes/backchannellogout/nats_test.go b/services/proxy/pkg/staticroutes/backchannellogout/nats_test.go new file mode 100644 index 0000000000..8ab19124e2 --- /dev/null +++ b/services/proxy/pkg/staticroutes/backchannellogout/nats_test.go @@ -0,0 +1,144 @@ +package backchannellogout + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + natsjskv "github.com/go-micro/plugins/v4/store/nats-js-kv" + nserver "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/require" + "go-micro.dev/v4/store" +) + +func newNATSCleanupTestCache(t *testing.T) (*Cache, nats.JetStreamContext, nats.KeyValue) { + t.Helper() + server, err := nserver.NewServer(&nserver.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir()}) + require.NoError(t, err) + go server.Start() + t.Cleanup(func() { server.Shutdown(); server.WaitForShutdown() }) + require.True(t, server.ReadyForConnections(5*time.Second)) + options := nats.GetDefaultOptions() + options.Servers = []string{server.ClientURL()} + conn, err := options.Connect() + require.NoError(t, err) + t.Cleanup(conn.Close) + js, err := conn.JetStream() + require.NoError(t, err) + backing := natsjskv.NewStore(store.Nodes(server.ClientURL()), store.Database("oidc-cleanup"), store.Table("oidc"), natsjskv.EncodeKeys(), natsjskv.DefaultTTL(0)) + cache := NewCache(WithNATSCleanup(backing, options.Connect), true) + t.Cleanup(func() { _ = cache.Close() }) + _, err = cache.List() + require.NoError(t, err) + bucket, err := js.KeyValue("oidc-cleanup") + require.NoError(t, err) + return cache, js, bucket +} + +func TestNATSCleanupRemovesDeleteMarkers(t *testing.T) { + cache, js, _ := newNATSCleanupTestCache(t) + now := time.Now() + cache.now = func() time.Time { return now } + require.NoError(t, cache.collectExpired(context.Background()), "an empty bucket needs no cleanup") + for i := range 20 { + require.NoError(t, cache.Write(&store.Record{Key: fmt.Sprintf("expired-%d", i), Value: []byte("claims"), Expiry: time.Minute})) + } + require.NoError(t, RevokeToken(&store.Record{Value: []byte("active"), Expiry: time.Hour}, cache)) + now = now.Add(2 * time.Minute) + require.NoError(t, cache.collectExpired(context.Background())) + info, err := js.StreamInfo("KV_oidc-cleanup") + require.NoError(t, err) + require.EqualValues(t, 1, info.State.Msgs, "expired values and their markers must be physically removed") + revoked, err := IsTokenRevoked("active", cache) + require.NoError(t, err) + require.True(t, revoked, "valid revocations must remain stored") + + require.NoError(t, cache.Delete(RevokedTokenKey("active"))) + keys, err := cache.List() + require.NoError(t, err) + require.Empty(t, keys) + require.NoError(t, cache.collectExpired(context.Background()), "markers must be collected even when List returns no live keys") + info, err = js.StreamInfo("KV_oidc-cleanup") + require.NoError(t, err) + require.Zero(t, info.State.Msgs) + + // Lazy expiry during Read also creates a marker, collected on the next pass. + require.NoError(t, cache.Write(&store.Record{Key: "lazy", Expiry: time.Minute})) + now = now.Add(2 * time.Minute) + read, err := cache.Read("lazy") + require.NoError(t, err) + require.Empty(t, read) + require.NoError(t, cache.collectExpired(context.Background())) + info, err = js.StreamInfo("KV_oidc-cleanup") + require.NoError(t, err) + require.Zero(t, info.State.Msgs) +} + +func TestNATSCleanupPreservesConcurrentWritesAndOtherTables(t *testing.T) { + cache, js, bucket := newNATSCleanupTestCache(t) + backend := cache.Store.(*natsCacheStore) + encoder := backend.Store.(interface{ NatsKey(string, string) string }) + reusedKey := "reused" + require.NoError(t, cache.Write(&store.Record{Key: reusedKey, Value: []byte("old")})) + require.NoError(t, cache.Delete(reusedKey)) + otherKey := encoder.NatsKey("other-table", "deleted") + _, err := bucket.Put(otherKey, []byte("other table")) + require.NoError(t, err) + require.NoError(t, bucket.Delete(otherKey)) + proxy := &beforeNATSPurge{JetStreamContext: js, before: func() { + // Reuse the key after the snapshot, immediately before the purge. + require.NoError(t, cache.Write(&store.Record{Key: reusedKey, Value: []byte("new"), Expiry: time.Hour})) + }} + require.NoError(t, backend.purgeDeleted(context.Background(), proxy)) + read, err := cache.Read(reusedKey) + require.NoError(t, err) + require.Len(t, read, 1) + require.Equal(t, []byte("new"), read[0].Value) + info, err := js.StreamInfo("KV_oidc-cleanup") + require.NoError(t, err) + require.EqualValues(t, 2, info.State.Msgs, "keep the concurrent value and the other table's marker") +} + +func TestNATSCleanupReportsFailureAndRetries(t *testing.T) { + cache, js, _ := newNATSCleanupTestCache(t) + backend := cache.Store.(*natsCacheStore) + require.NoError(t, cache.Write(&store.Record{Key: "deleted"})) + require.NoError(t, cache.Delete("deleted")) + failure := errors.New("purge unavailable") + proxy := &beforeNATSPurge{JetStreamContext: js, err: failure} + require.ErrorIs(t, backend.purgeDeleted(context.Background(), proxy), failure) + info, err := js.StreamInfo("KV_oidc-cleanup") + require.NoError(t, err) + require.EqualValues(t, 1, info.State.Msgs, "keep the marker so cleanup can retry") + connect := backend.connect + backend.connect = func() (*nats.Conn, error) { return nil, failure } + require.ErrorIs(t, cache.collectExpired(context.Background()), failure, "maintenance failures must reach the cleanup caller") + backend.connect = connect + require.NoError(t, backend.PurgeDeleted(context.Background())) + info, err = js.StreamInfo("KV_oidc-cleanup") + require.NoError(t, err) + require.Zero(t, info.State.Msgs) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + require.ErrorIs(t, backend.PurgeDeleted(ctx), context.Canceled) +} + +type beforeNATSPurge struct { + nats.JetStreamContext + before func() + err error +} + +func (p *beforeNATSPurge) PurgeStream(name string, opts ...nats.JSOpt) error { + if p.before != nil { + p.before() + p.before = nil + } + if p.err != nil { + return p.err + } + return p.JetStreamContext.PurgeStream(name, opts...) +} From 0fe98de3e5ea860bf299f3458693d7c77df023f5 Mon Sep 17 00:00:00 2001 From: zerox80 Date: Mon, 7 Sep 2026 10:33:52 +0200 Subject: [PATCH 7/7] fix(proxy): match session logout without a cached subject --- .../backchannellogout/backchannellogout.go | 5 ++-- .../backchannellogout_test.go | 26 +++++++++++++++++++ .../backchannellogout_integration_test.go | 23 ++++++++++++++++ 3 files changed, 52 insertions(+), 2 deletions(-) diff --git a/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout.go b/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout.go index 4890e1b379..9a7ab463ce 100644 --- a/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout.go +++ b/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout.go @@ -209,9 +209,10 @@ func GetLogoutRecords(suse SuSe, store microstore.Store) ([]*microstore.Record, // in subject mode, the subject must match, but the session id can be different case suse.Mode() == LogoutModeSubject && suse.encodedSubject == recordSuSe.encodedSubject: continue - // In session mode, match the subject too when it was supplied. + // In session mode, compare subjects only when both are known. Access + // tokens without a subject can still be identified by their session ID. case suse.Mode() == LogoutModeSession && suse.encodedSession == recordSuSe.encodedSession && - (suse.encodedSubject == "" || suse.encodedSubject == recordSuSe.encodedSubject): + (suse.encodedSubject == "" || recordSuSe.encodedSubject == "" || suse.encodedSubject == recordSuSe.encodedSubject): continue } diff --git a/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout_test.go b/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout_test.go index fb9589747a..c4bd982601 100644 --- a/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout_test.go +++ b/services/proxy/pkg/staticroutes/backchannellogout/backchannellogout_test.go @@ -81,6 +81,32 @@ func TestGetLogoutRecordsRejectsMismatchedSubject(t *testing.T) { require.ErrorIs(t, err, ErrSuspiciousCacheResult) } +func TestGetLogoutRecordsMatchesSessionWithoutSubject(t *testing.T) { + cache := store.NewMemoryStore() + legacy := mustNewKey(t, "", "session") + token, err := NewTokenKey("", "session", "token") + require.NoError(t, err) + knownSubject, err := NewTokenKey("alice", "session", "known-subject") + require.NoError(t, err) + for _, key := range []string{legacy, token, knownSubject} { + require.NoError(t, cache.Write(&store.Record{Key: key})) + } + + records, err := GetLogoutRecords(mustNewSuSe(t, "alice", "session"), cache) + require.NoError(t, err) + keys := make([]string, 0, len(records)) + for _, record := range records { + keys = append(keys, record.Key) + } + require.ElementsMatch(t, []string{legacy, token, knownSubject}, keys) + + // Without a session ID, a logout must still match the stored subject. + records, err = GetLogoutRecords(mustNewSuSe(t, "alice", ""), cache) + require.NoError(t, err) + require.Len(t, records, 1) + require.Equal(t, knownSubject, records[0].Key) +} + func mustNewSuSe(t *testing.T, subject, session string) SuSe { suse, err := NewSuSe(mustNewKey(t, subject, session)) require.NoError(t, err) diff --git a/services/proxy/pkg/staticroutes/backchannellogout_integration_test.go b/services/proxy/pkg/staticroutes/backchannellogout_integration_test.go index 885f56c8b4..035b133d04 100644 --- a/services/proxy/pkg/staticroutes/backchannellogout_integration_test.go +++ b/services/proxy/pkg/staticroutes/backchannellogout_integration_test.go @@ -21,6 +21,7 @@ import ( "github.com/opencloud-eu/opencloud/services/proxy/pkg/middleware" "github.com/opencloud-eu/opencloud/services/proxy/pkg/router" "github.com/opencloud-eu/opencloud/services/proxy/pkg/staticroutes" + bcl "github.com/opencloud-eu/opencloud/services/proxy/pkg/staticroutes/backchannellogout" "github.com/stretchr/testify/require" "go-micro.dev/v4/store" ) @@ -68,6 +69,28 @@ func TestBackchannelLogoutInvalidatesEveryCachedToken(t *testing.T) { } } +func TestBackchannelLogoutWithSubjectForSessionOnlyAccessToken(t *testing.T) { + idp := newLogoutTestIDP(t) + cache := bcl.NewCache(store.NewMemoryStore(), true) + auth, routes := idp.handlers(cache, true) + token := idp.sign(t, jwt.MapClaims{ + "iss": idp.server.URL, "sid": "session", "aud": "opencloud", "preferred_username": "alice", + "iat": time.Now().Unix(), "exp": time.Now().Add(time.Hour).Unix(), + }) + unrelated := idp.sign(t, jwt.MapClaims{ + "iss": idp.server.URL, "sid": "other-session", "aud": "opencloud", "preferred_username": "alice", + "iat": time.Now().Unix(), "exp": time.Now().Add(time.Hour).Unix(), + }) + for _, token := range []string{token, unrelated} { + require.Equal(t, http.StatusOK, logoutTestRequest(auth, token)) + } + + response := logoutTestLogout(routes, idp.logoutToken(t, "alice", "session")) + require.Equal(t, http.StatusOK, response.Code) + require.Equal(t, http.StatusUnauthorized, logoutTestRequest(auth, token)) + require.Equal(t, http.StatusOK, logoutTestRequest(auth, unrelated)) +} + type logoutTestIDP struct { server *httptest.Server key *rsa.PrivateKey