From 49b86264dad0ac76dfe1d081eac72b6448e3cb2e Mon Sep 17 00:00:00 2001 From: hisco <39222286+hisco@users.noreply.github.com> Date: Sat, 1 Aug 2026 02:08:20 +0300 Subject: [PATCH 1/3] fix(rbac): bound SSE change-authorizer with a per-connection SAR memo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SSE live-stream change authorizer built a plain closure that called canReadUser per frame. canReadUser only writes a SubjectAccessReview result back to the shared permission cache when that user's entry still exists; the SSE path primes the entry once at subscribe and it TTLs out after ~2min. From then on every qualifying change frame for a long-lived client ran a fresh, UNCACHED SAR, serially inside the single broadcast goroutine, with no timeout — one slow apiserver call stalled broadcasts for all clients, and sustained change traffic multiplied SAR load by client × (kind, namespace). Only affects auth-enabled multi-user deployments; auth-off (local kubeconfig) is unchanged. Fix: newSSEChangeAuthorizer wraps the decision in a connection-lived TTL memo (memoizedAuthorizer), so each (context, verb, group, resource, namespace) is resolved at most once per TTL instead of once per frame. Details: - The memo's base runs canReadUserSAR — a fresh, bounded SAR that BYPASSES the shared permission cache — so SSE staleness is bounded to the memo's own 2min TTL rather than stacking with the shared-cache TTL (~4min worst case). - The memo key includes the current context name. A kubeconfig context switch leaves SSE connections open (they receive a context_changed frame, not a disconnect), so without this a still-open stream could authorize new-cluster frames with the previous cluster's decisions; post-switch keys now miss and re-run against the new apiserver, mirroring the shared cache's own stamping. - If the context changes while a SAR is in flight, the result is returned but not cached, so a switch-back within the TTL can't serve a wrong-cluster decision. - The SAR runs under a bounded 5s context so a hung apiserver call can't wedge the broadcast loop. canReadUser is refactored to delegate its SAR tail to canReadUserSAR; behavior is unchanged (reads shared cache, writes back only when the entry exists). Tests: memoizedAuthorizer within-TTL memoization + past-TTL re-check, context-switch decision isolation, no-cache-on-mid-SAR-switch, and auth-off passthrough. Verified end-to-end against a kind cluster in --auth-mode proxy: a single long-lived viewer stream (no secrets RBAC) received zero Secret frames across the 2min cache-expiry boundary while still receiving new Deployment frames post-expiry. Claude-Session: https://claude.ai/code/session_01XNhMe6Zdqwj5EoBazNDnmW --- internal/server/server.go | 107 ++++++++++++++++++++-- internal/server/sse.go | 14 +-- internal/server/sse_authorizer_test.go | 117 +++++++++++++++++++++++++ 3 files changed, 223 insertions(+), 15 deletions(-) create mode 100644 internal/server/sse_authorizer_test.go diff --git a/internal/server/server.go b/internal/server/server.go index 32f18da92..5d6e396b8 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1245,22 +1245,37 @@ func (s *Server) canReadUser(ctx context.Context, user *auth.User, group, resour return v } } + allowed := s.canReadUserSAR(ctx, user, group, resource, namespace, verb) + if perms != nil { + perms.SetCanI(verb, group, resource, namespace, allowed) + } + return allowed +} + +// canReadUserSAR runs a single fresh SubjectAccessReview for (group, resource, +// namespace, verb) against the current apiserver, bypassing the shared +// permission cache entirely. Fail-closed on a missing client or SAR error. +// +// canReadUser wraps this behind the shared cache. The SSE change authorizer +// calls it directly instead: reusing the shared cache there would let a decision +// already up to the cache TTL old be re-cached under the SSE memo's own TTL, +// stacking staleness — and the shared entry's context stamping wouldn't help, +// because the SSE memo, not the shared cache, is what a long-lived stream reads. +// A fresh SAR keeps the SSE staleness bounded to that memo's TTL alone. +func (s *Server) canReadUserSAR(ctx context.Context, user *auth.User, group, resource, namespace, verb string) bool { client := k8s.GetClient() if client == nil { // Fail-closed: no apiserver to ask, refuse rather than quietly // serving from the cache. - log.Printf("[auth] canReadUser: K8s client unavailable, denying %s on %s/%s for %s", k8s.SanitizeForLog(verb), k8s.SanitizeForLog(group), k8s.SanitizeForLog(resource), k8s.SanitizeForLog(user.Username)) + log.Printf("[auth] canReadUserSAR: K8s client unavailable, denying %s on %s/%s for %s", k8s.SanitizeForLog(verb), k8s.SanitizeForLog(group), k8s.SanitizeForLog(resource), k8s.SanitizeForLog(user.Username)) return false } allowed, err := auth.SubjectCanI(ctx, client, user.Username, user.Groups, namespace, group, resource, verb) if err != nil { // Fail-closed on SAR error — apiserver said something we don't trust. - log.Printf("[auth] canReadUser SAR failed for %s on %s/%s in ns=%q: %v", k8s.SanitizeForLog(user.Username), k8s.SanitizeForLog(group), k8s.SanitizeForLog(resource), k8s.SanitizeForLog(namespace), err) + log.Printf("[auth] canReadUserSAR failed for %s on %s/%s in ns=%q: %v", k8s.SanitizeForLog(user.Username), k8s.SanitizeForLog(group), k8s.SanitizeForLog(resource), k8s.SanitizeForLog(namespace), err) return false } - if perms != nil { - perms.SetCanI(verb, group, resource, namespace, allowed) - } return allowed } @@ -4464,11 +4479,85 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { if user != nil && s.permCache != nil && s.permCache.Get(user.Username) == nil { _ = s.getUserNamespaces(r, []string{}) } - ctx := r.Context() - authorize := func(group, resource, namespace, verb string) bool { - return s.canReadUser(ctx, user, group, resource, namespace, verb) + s.broadcaster.HandleSSE(w, r, deny, s.newSSEChangeAuthorizer(r.Context(), user)) +} + +const ( + // sseChangeAuthTTL bounds how long an SSE client's per-frame authorization + // decision is cached before re-checking, so a revoked grant propagates within + // the window (matching the REST permission cache's cadence). + sseChangeAuthTTL = 2 * time.Minute + // sseChangeAuthSARTimeout caps a single authorization SAR issued from the + // broadcast goroutine, so one hung apiserver call can't stall broadcasts. + sseChangeAuthSARTimeout = 5 * time.Second +) + +// newSSEChangeAuthorizer returns the per-kind authorizer for one SSE client's +// change frames, backed by a connection-lived memo. +// +// Without the memo, every qualifying change frame for a long-lived client would +// run a fresh, UNCACHED SubjectAccessReview serially inside the single broadcast +// goroutine (canReadUser only writes back to the shared permission cache when +// that entry exists, and the SSE path primes it only once at subscribe, so it +// TTLs out): stalling every client and multiplying apiserver SAR load by +// client × (kind, namespace). The memo survives that cache expiry; its own TTL +// preserves RBAC-change propagation; and the bounded SAR context stops a hung +// apiserver call from wedging the broadcast loop. +// +// The memo keys on the current context name so a kubeconfig context switch — +// which leaves SSE connections open (they receive a context_changed frame, not +// a disconnect) — can't authorize new-cluster frames with the previous +// cluster's decisions; post-switch keys miss and re-run against the new +// apiserver, mirroring the shared cache's own context stamping. nil user +// (auth off) is a passthrough. +func (s *Server) newSSEChangeAuthorizer(ctx context.Context, user *auth.User) func(group, resource, namespace, verb string) bool { + if user == nil || s.permCache == nil { + return func(_, _, _, _ string) bool { return true } + } + base := func(group, resource, namespace, verb string) bool { + sarCtx, cancel := context.WithTimeout(ctx, sseChangeAuthSARTimeout) + defer cancel() + return s.canReadUserSAR(sarCtx, user, group, resource, namespace, verb) + } + return memoizedAuthorizer(base, sseChangeAuthTTL, k8s.GetContextName, time.Now) +} + +// memoizedAuthorizer wraps an authorization predicate with a per-(context, verb, +// group, resource, namespace) TTL memo so repeated lookups don't re-issue the +// SAR. Keying on contextName scopes decisions to the cluster they were made +// against. contextName and now are injectable for tests. +func memoizedAuthorizer(base func(group, resource, namespace, verb string) bool, ttl time.Duration, contextName func() string, now func() time.Time) func(group, resource, namespace, verb string) bool { + type entry struct { + allowed bool + expires time.Time + } + var mu sync.Mutex + memo := make(map[string]entry) + return func(group, resource, namespace, verb string) bool { + ctxName := "" + if contextName != nil { + ctxName = contextName() + } + key := ctxName + "\x00" + verb + "\x00" + group + "\x00" + resource + "\x00" + namespace + t := now() + mu.Lock() + if e, ok := memo[key]; ok && t.Before(e.expires) { + mu.Unlock() + return e.allowed + } + mu.Unlock() + allowed := base(group, resource, namespace, verb) + // If the cluster context changed while base() was in flight, the result + // may have been decided against a different apiserver than `key` names — + // caching it would let a later switch-back within the TTL serve the wrong + // cluster's decision. Drop it; the next frame re-evaluates cleanly. + if contextName == nil || contextName() == ctxName { + mu.Lock() + memo[key] = entry{allowed: allowed, expires: t.Add(ttl)} + mu.Unlock() + } + return allowed } - s.broadcaster.HandleSSE(w, r, deny, authorize) } // Settings handlers diff --git a/internal/server/sse.go b/internal/server/sse.go index 9a2e19988..7720ad26f 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -70,12 +70,14 @@ type ClientInfo struct { // broadcast loop never runs a SAR. nil/empty for users with full access. DeniedKinds map[topology.NodeKind]bool // Authorize authorizes a per-resource change frame for this client's user - // via SubjectAccessReview (memoized on the user's permission cache). Bound - // at subscribe time to the request's user + a connection-lived context, so - // the broadcast goroutine can gate diff-bearing k8s_event frames per kind - // without holding a request. nil when no authorizer was wired (defensive / - // tests) — clientCanSeeChange then falls back to the namespace + denied-kind - // gate. When auth is disabled the closure is still set and returns true. + // via SubjectAccessReview, memoized in a connection-lived TTL cache + // (context-scoped, independent of the shared permission cache) so a + // long-lived stream doesn't re-SAR every frame. Bound at subscribe time to + // the request's user + a connection-lived context, so the broadcast + // goroutine can gate diff-bearing k8s_event frames per kind without holding a + // request. nil when no authorizer was wired (defensive / tests) — + // clientCanSeeChange then falls back to the namespace + denied-kind gate. + // When auth is disabled the closure is still set and returns true. Authorize func(group, resource, namespace, verb string) bool } diff --git a/internal/server/sse_authorizer_test.go b/internal/server/sse_authorizer_test.go new file mode 100644 index 000000000..f4bff9538 --- /dev/null +++ b/internal/server/sse_authorizer_test.go @@ -0,0 +1,117 @@ +package server + +import ( + "testing" + "time" +) + +// The SSE change authorizer must not re-issue a SAR for a repeated (verb, group, +// resource, namespace) within the TTL — otherwise a long-lived stream re-SARs +// every frame once the shared permission cache expires, serially in the single +// broadcast goroutine. It must re-check after the TTL so RBAC changes propagate. +func TestMemoizedAuthorizer(t *testing.T) { + calls := 0 + // base allows only "list secrets"; counts invocations. + base := func(group, resource, namespace, verb string) bool { + calls++ + return verb == "list" && resource == "secrets" + } + clock := time.Now() + ctxName := "cluster-a" + authz := memoizedAuthorizer(base, 2*time.Minute, func() string { return ctxName }, func() time.Time { return clock }) + + if !authz("", "secrets", "team-a", "list") { + t.Fatal("secrets/list should be allowed") + } + // Repeat within TTL → served from memo, no new base call. + authz("", "secrets", "team-a", "list") + authz("", "secrets", "team-a", "list") + if calls != 1 { + t.Fatalf("want 1 base call within TTL, got %d", calls) + } + + // A different tuple is a distinct decision → one more base call. + if authz("", "pods", "team-a", "list") { + t.Fatal("pods/list should be denied by base") + } + if calls != 2 { + t.Fatalf("want 2 base calls for a new tuple, got %d", calls) + } + + // Past the TTL, the same tuple re-checks (propagates RBAC changes). + clock = clock.Add(2*time.Minute + time.Second) + authz("", "secrets", "team-a", "list") + if calls != 3 { + t.Fatalf("want a re-check after the TTL, got %d base calls", calls) + } +} + +// A kubeconfig context switch leaves SSE connections open, so the memo must not +// serve the previous cluster's decision for the new one — a changed context +// name is a cache miss even within the TTL, re-running the SAR against the new +// apiserver. +func TestMemoizedAuthorizer_ContextSwitchIsolatesDecisions(t *testing.T) { + calls := 0 + // Same tuple resolves differently per cluster: allowed on cluster-a, denied + // on cluster-b. + var ctxName string + base := func(group, resource, namespace, verb string) bool { + calls++ + return ctxName == "cluster-a" + } + clock := time.Now() + authz := memoizedAuthorizer(base, 2*time.Minute, func() string { return ctxName }, func() time.Time { return clock }) + + ctxName = "cluster-a" + if !authz("", "secrets", "team-a", "list") { + t.Fatal("secrets/list should be allowed on cluster-a") + } + // Switch context well within the TTL — the old allow must NOT leak. + ctxName = "cluster-b" + if authz("", "secrets", "team-a", "list") { + t.Fatal("cluster-a's allow leaked into cluster-b within the TTL") + } + if calls != 2 { + t.Fatalf("want a fresh base call after context switch, got %d", calls) + } +} + +// If a context switch lands while the SAR is in flight, the fresh result was +// decided against a different apiserver than the key names — it must NOT be +// cached, or a switch-back within the TTL would serve the wrong cluster's +// decision. +func TestMemoizedAuthorizer_NoCacheWhenContextChangesDuringSAR(t *testing.T) { + calls := 0 + ctxName := "cluster-a" + base := func(group, resource, namespace, verb string) bool { + calls++ + if calls == 1 { + // simulate a context switch landing mid-SAR + ctxName = "cluster-b" + } + return true + } + clock := time.Now() + authz := memoizedAuthorizer(base, 2*time.Minute, func() string { return ctxName }, func() time.Time { return clock }) + + // Key computed under cluster-a, but context flips to cluster-b during base → + // result must not be cached under the cluster-a key. + authz("", "secrets", "team-a", "list") + // Back on cluster-a within the TTL: a cached (poisoned) entry would be served + // with no new base call. A fresh call proves it wasn't cached. + ctxName = "cluster-a" + authz("", "secrets", "team-a", "list") + if calls != 2 { + t.Fatalf("switch-straddling SAR must not be cached; want 2 base calls, got %d", calls) + } +} + +// nil user (auth off) is a strict passthrough — every frame is allowed and no +// SAR is ever issued. +func TestNewSSEChangeAuthorizer_AuthOff(t *testing.T) { + s := &Server{} + authz := s.newSSEChangeAuthorizer(nil, nil) + if !authz("", "secrets", "team-a", "list") { + t.Fatal("auth-off authorizer must allow everything") + } +} From 8ec8bbcd6e2cdeaf0511931ed983341b81cc643e Mon Sep 17 00:00:00 2001 From: hisco <39222286+hisco@users.noreply.github.com> Date: Sat, 1 Aug 2026 02:15:17 +0300 Subject: [PATCH 2/3] fix(rbac): don't cache transient SAR failures in the SSE memo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit canReadUserSAR collapsed a missing client, SAR error, or the 5s timeout into the same false as a real deny, and both the SSE memo and canReadUser cached it for the full TTL — so a momentary apiserver blip would deny a (kind, namespace) on the live stream for up to 2min (and, via canReadUser's write-back, in the shared permission cache too). The pre-refactor canReadUser returned the error-false BEFORE SetCanI, so it never cached failures; the SSE refactor had inadvertently dropped that distinction. canReadUserSAR now returns (allowed, authoritative). A non-authoritative result is a fail-closed false that callers return but must not cache: canReadUser skips SetCanI, and memoizedAuthorizer skips the memo store. The next frame retries, so a transient failure drops at most the frames in flight during the blip, not a whole TTL window. Real allow/deny verdicts cache as before. Adds TestMemoizedAuthorizer_TransientFailureNotCached (failure not cached → retry re-runs base and succeeds; the subsequent authoritative allow is cached). Claude-Session: https://claude.ai/code/session_01XNhMe6Zdqwj5EoBazNDnmW --- internal/server/server.go | 43 +++++++++++++-------- internal/server/sse_authorizer_test.go | 52 ++++++++++++++++++++++---- 2 files changed, 72 insertions(+), 23 deletions(-) diff --git a/internal/server/server.go b/internal/server/server.go index 5d6e396b8..a26c7a930 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1245,8 +1245,11 @@ func (s *Server) canReadUser(ctx context.Context, user *auth.User, group, resour return v } } - allowed := s.canReadUserSAR(ctx, user, group, resource, namespace, verb) - if perms != nil { + allowed, authoritative := s.canReadUserSAR(ctx, user, group, resource, namespace, verb) + // Cache only a real apiserver verdict. A transient failure (no client, SAR + // error, timeout) fails closed for this call but must NOT be memoized, or a + // momentary blip would deny the tuple for the whole cache TTL. + if authoritative && perms != nil { perms.SetCanI(verb, group, resource, namespace, allowed) } return allowed @@ -1254,7 +1257,10 @@ func (s *Server) canReadUser(ctx context.Context, user *auth.User, group, resour // canReadUserSAR runs a single fresh SubjectAccessReview for (group, resource, // namespace, verb) against the current apiserver, bypassing the shared -// permission cache entirely. Fail-closed on a missing client or SAR error. +// permission cache entirely. It returns (allowed, authoritative): authoritative +// is false when the apiserver couldn't be consulted (no client, SAR error, +// timeout), in which case allowed is a fail-closed false that callers must not +// cache — the next call retries. // // canReadUser wraps this behind the shared cache. The SSE change authorizer // calls it directly instead: reusing the shared cache there would let a decision @@ -1262,21 +1268,21 @@ func (s *Server) canReadUser(ctx context.Context, user *auth.User, group, resour // stacking staleness — and the shared entry's context stamping wouldn't help, // because the SSE memo, not the shared cache, is what a long-lived stream reads. // A fresh SAR keeps the SSE staleness bounded to that memo's TTL alone. -func (s *Server) canReadUserSAR(ctx context.Context, user *auth.User, group, resource, namespace, verb string) bool { +func (s *Server) canReadUserSAR(ctx context.Context, user *auth.User, group, resource, namespace, verb string) (allowed bool, authoritative bool) { client := k8s.GetClient() if client == nil { // Fail-closed: no apiserver to ask, refuse rather than quietly // serving from the cache. log.Printf("[auth] canReadUserSAR: K8s client unavailable, denying %s on %s/%s for %s", k8s.SanitizeForLog(verb), k8s.SanitizeForLog(group), k8s.SanitizeForLog(resource), k8s.SanitizeForLog(user.Username)) - return false + return false, false } allowed, err := auth.SubjectCanI(ctx, client, user.Username, user.Groups, namespace, group, resource, verb) if err != nil { // Fail-closed on SAR error — apiserver said something we don't trust. log.Printf("[auth] canReadUserSAR failed for %s on %s/%s in ns=%q: %v", k8s.SanitizeForLog(user.Username), k8s.SanitizeForLog(group), k8s.SanitizeForLog(resource), k8s.SanitizeForLog(namespace), err) - return false + return false, false } - return allowed + return allowed, true } // filterNamespacesByCanRead returns the subset of `namespaces` where the @@ -4514,7 +4520,7 @@ func (s *Server) newSSEChangeAuthorizer(ctx context.Context, user *auth.User) fu if user == nil || s.permCache == nil { return func(_, _, _, _ string) bool { return true } } - base := func(group, resource, namespace, verb string) bool { + base := func(group, resource, namespace, verb string) (bool, bool) { sarCtx, cancel := context.WithTimeout(ctx, sseChangeAuthSARTimeout) defer cancel() return s.canReadUserSAR(sarCtx, user, group, resource, namespace, verb) @@ -4525,8 +4531,11 @@ func (s *Server) newSSEChangeAuthorizer(ctx context.Context, user *auth.User) fu // memoizedAuthorizer wraps an authorization predicate with a per-(context, verb, // group, resource, namespace) TTL memo so repeated lookups don't re-issue the // SAR. Keying on contextName scopes decisions to the cluster they were made -// against. contextName and now are injectable for tests. -func memoizedAuthorizer(base func(group, resource, namespace, verb string) bool, ttl time.Duration, contextName func() string, now func() time.Time) func(group, resource, namespace, verb string) bool { +// against. base returns (allowed, authoritative); a non-authoritative result +// (transient SAR failure) is returned to the caller fail-closed but not cached, +// so a momentary apiserver blip can't deny a tuple for the whole TTL. +// contextName and now are injectable for tests. +func memoizedAuthorizer(base func(group, resource, namespace, verb string) (bool, bool), ttl time.Duration, contextName func() string, now func() time.Time) func(group, resource, namespace, verb string) bool { type entry struct { allowed bool expires time.Time @@ -4546,12 +4555,14 @@ func memoizedAuthorizer(base func(group, resource, namespace, verb string) bool, return e.allowed } mu.Unlock() - allowed := base(group, resource, namespace, verb) - // If the cluster context changed while base() was in flight, the result - // may have been decided against a different apiserver than `key` names — - // caching it would let a later switch-back within the TTL serve the wrong - // cluster's decision. Drop it; the next frame re-evaluates cleanly. - if contextName == nil || contextName() == ctxName { + allowed, authoritative := base(group, resource, namespace, verb) + // Cache only an authoritative verdict, and only if the cluster context + // didn't change while base() was in flight — otherwise the result may + // reflect a different apiserver than `key` names, and a later switch-back + // within the TTL would serve the wrong cluster's decision. In either + // skip case the fail-closed result is still returned; the next frame + // re-evaluates cleanly. + if authoritative && (contextName == nil || contextName() == ctxName) { mu.Lock() memo[key] = entry{allowed: allowed, expires: t.Add(ttl)} mu.Unlock() diff --git a/internal/server/sse_authorizer_test.go b/internal/server/sse_authorizer_test.go index f4bff9538..9ba3cda5d 100644 --- a/internal/server/sse_authorizer_test.go +++ b/internal/server/sse_authorizer_test.go @@ -11,10 +11,11 @@ import ( // broadcast goroutine. It must re-check after the TTL so RBAC changes propagate. func TestMemoizedAuthorizer(t *testing.T) { calls := 0 - // base allows only "list secrets"; counts invocations. - base := func(group, resource, namespace, verb string) bool { + // base allows only "list secrets"; counts invocations. All verdicts here are + // authoritative (a real apiserver answer). + base := func(group, resource, namespace, verb string) (bool, bool) { calls++ - return verb == "list" && resource == "secrets" + return verb == "list" && resource == "secrets", true } clock := time.Now() ctxName := "cluster-a" @@ -55,9 +56,9 @@ func TestMemoizedAuthorizer_ContextSwitchIsolatesDecisions(t *testing.T) { // Same tuple resolves differently per cluster: allowed on cluster-a, denied // on cluster-b. var ctxName string - base := func(group, resource, namespace, verb string) bool { + base := func(group, resource, namespace, verb string) (bool, bool) { calls++ - return ctxName == "cluster-a" + return ctxName == "cluster-a", true } clock := time.Now() authz := memoizedAuthorizer(base, 2*time.Minute, func() string { return ctxName }, func() time.Time { return clock }) @@ -83,13 +84,13 @@ func TestMemoizedAuthorizer_ContextSwitchIsolatesDecisions(t *testing.T) { func TestMemoizedAuthorizer_NoCacheWhenContextChangesDuringSAR(t *testing.T) { calls := 0 ctxName := "cluster-a" - base := func(group, resource, namespace, verb string) bool { + base := func(group, resource, namespace, verb string) (bool, bool) { calls++ if calls == 1 { // simulate a context switch landing mid-SAR ctxName = "cluster-b" } - return true + return true, true } clock := time.Now() authz := memoizedAuthorizer(base, 2*time.Minute, func() string { return ctxName }, func() time.Time { return clock }) @@ -106,6 +107,43 @@ func TestMemoizedAuthorizer_NoCacheWhenContextChangesDuringSAR(t *testing.T) { } } +// A transient SAR failure (authoritative=false) fails closed for that frame but +// must NOT be cached — otherwise a momentary apiserver blip would deny the tuple +// for the whole TTL. The next frame must retry, and a subsequent success must be +// served correctly. +func TestMemoizedAuthorizer_TransientFailureNotCached(t *testing.T) { + calls := 0 + // First call is a transient failure (deny, non-authoritative); afterwards the + // apiserver recovers and authoritatively allows. + base := func(group, resource, namespace, verb string) (bool, bool) { + calls++ + if calls == 1 { + return false, false // transient failure: fail-closed, don't cache + } + return true, true // recovered: authoritative allow + } + clock := time.Now() + ctxName := "cluster-a" + authz := memoizedAuthorizer(base, 2*time.Minute, func() string { return ctxName }, func() time.Time { return clock }) + + if authz("", "secrets", "team-a", "list") { + t.Fatal("transient failure must fail closed (deny) for this frame") + } + // Same tuple, still within TTL: the failure must not have been cached, so the + // retry runs base again and now succeeds. + if !authz("", "secrets", "team-a", "list") { + t.Fatal("transient failure was cached — retry should have re-run base and allowed") + } + if calls != 2 { + t.Fatalf("want 2 base calls (failure not cached, retried), got %d", calls) + } + // The authoritative allow IS cached now — a third call is served from memo. + authz("", "secrets", "team-a", "list") + if calls != 2 { + t.Fatalf("authoritative allow should be cached; want still 2 base calls, got %d", calls) + } +} + // nil user (auth off) is a strict passthrough — every frame is allowed and no // SAR is ever issued. func TestNewSSEChangeAuthorizer_AuthOff(t *testing.T) { From 47485b329af0813d37e17683bf202e5310ddedab Mon Sep 17 00:00:00 2001 From: hisco <39222286+hisco@users.noreply.github.com> Date: Thu, 6 Aug 2026 13:47:29 +0300 Subject: [PATCH 3/3] fix(rbac): harden SSE change-authorizer memo (review follow-ups) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three follow-ups from the #1313 review of the SSE change-authorizer memo. All are in memoizedAuthorizer; none changes REST or auth-off behavior. 1. Fail closed on a mid-SAR context switch. When the kubeconfig context changed while base() was in flight, the memo already skipped caching the result (decided against a different apiserver than the key names) but still returned it, so a wrong-cluster verdict could release one frame. It now returns a fail-closed deny instead; the next frame re-evaluates against the new cluster. Dropping one frame during a switch is harmless — the client is about to receive a context_changed frame anyway. 2. Cache a transient SAR failure briefly instead of not at all. This REFINES 8ec8bbcd, it does not revert it. That commit fixed caching failures for the full 2min TTL (which denied a readable tuple for the whole window on a blip) by never caching them. But never caching means a degraded-but-alive apiserver re-pays the 5s SAR timeout on every frame for the same tuple, serially in the single broadcast goroutine, stalling all clients. A non-authoritative result is now cached as a fail-closed deny for a short negativeTTL (10s): short enough that a momentary blip can't deny a readable tuple for the full TTL (Bugbot's concern), non-zero so a sustained outage doesn't re-SAR every frame (the stall concern). Authoritative allow/deny still caches for the full TTL. 3. Bound the per-connection memo. Entries expired but were never removed, so a long-lived all-namespace stream on a CRD-heavy cluster accumulated one entry per observed (context, verb, group, resource, namespace) tuple for the life of the connection. Past a soft cap (sseChangeAuthMemoCap), expired entries are swept before the next insert; the sweep is time-gated so a genuinely large live working set doesn't trigger an O(n) pass on every frame. Eviction logic extracted to sweepExpiredAuthMemo for direct unit testing. Tests: assert fail-closed on mid-SAR switch; rewrite the transient-failure test for brief caching (served from the negative cache within negativeTTL, re-checks and caches the recovered allow after it); add TestSweepExpiredAuthMemo. Full internal/server suite green under -race. Claude-Session: https://claude.ai/code/session_01Eyyu9N6tEELU1YgFd51J4D --- internal/server/server.go | 92 ++++++++++++++++++++------ internal/server/sse_authorizer_test.go | 78 ++++++++++++++++------ 2 files changed, 128 insertions(+), 42 deletions(-) diff --git a/internal/server/server.go b/internal/server/server.go index a26c7a930..985ef220f 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -4496,6 +4496,17 @@ const ( // sseChangeAuthSARTimeout caps a single authorization SAR issued from the // broadcast goroutine, so one hung apiserver call can't stall broadcasts. sseChangeAuthSARTimeout = 5 * time.Second + // sseChangeAuthNegativeTTL caps how long a transient SAR failure (apiserver + // unreachable, error, or timeout) is remembered as a fail-closed deny. Short + // so a momentary blip clears within seconds, but non-zero so a degraded + // apiserver doesn't re-pay the SAR timeout on every frame for the same tuple + // in the single broadcast goroutine. + sseChangeAuthNegativeTTL = 10 * time.Second + // sseChangeAuthMemoCap bounds one connection's authorization memo. Past it, + // expired entries are swept before the next insert so a long-lived + // all-namespace stream can't accumulate them without bound. Soft: a + // legitimately large live working set may exceed it. + sseChangeAuthMemoCap = 8192 ) // newSSEChangeAuthorizer returns the per-kind authorizer for one SSE client's @@ -4525,23 +4536,51 @@ func (s *Server) newSSEChangeAuthorizer(ctx context.Context, user *auth.User) fu defer cancel() return s.canReadUserSAR(sarCtx, user, group, resource, namespace, verb) } - return memoizedAuthorizer(base, sseChangeAuthTTL, k8s.GetContextName, time.Now) + return memoizedAuthorizer(base, sseChangeAuthTTL, sseChangeAuthNegativeTTL, sseChangeAuthMemoCap, k8s.GetContextName, time.Now) +} + +// authMemoEntry is one cached authorization decision in an SSE connection's memo. +type authMemoEntry struct { + allowed bool + expires time.Time +} + +// sweepExpiredAuthMemo deletes every entry whose TTL has elapsed as of now, +// reclaiming space in a long-lived connection's authorization memo, and returns +// the number removed. The caller holds the memo's lock. +func sweepExpiredAuthMemo(memo map[string]authMemoEntry, now time.Time) int { + removed := 0 + for k, e := range memo { + if !now.Before(e.expires) { + delete(memo, k) + removed++ + } + } + return removed } // memoizedAuthorizer wraps an authorization predicate with a per-(context, verb, // group, resource, namespace) TTL memo so repeated lookups don't re-issue the // SAR. Keying on contextName scopes decisions to the cluster they were made -// against. base returns (allowed, authoritative); a non-authoritative result -// (transient SAR failure) is returned to the caller fail-closed but not cached, -// so a momentary apiserver blip can't deny a tuple for the whole TTL. -// contextName and now are injectable for tests. -func memoizedAuthorizer(base func(group, resource, namespace, verb string) (bool, bool), ttl time.Duration, contextName func() string, now func() time.Time) func(group, resource, namespace, verb string) bool { - type entry struct { - allowed bool - expires time.Time - } +// against. base returns (allowed, authoritative): +// +// - An authoritative allow/deny is cached for the full ttl. +// - A non-authoritative result (transient SAR failure: no client, error, or +// timeout) is a fail-closed deny cached only for the short negativeTTL — long +// enough that a degraded apiserver doesn't re-pay the SAR timeout on every +// frame for the same tuple in the single broadcast goroutine, short enough +// that a momentary blip can't deny a readable tuple for the whole ttl. +// - If the cluster context changes while base() is in flight, the verdict was +// decided against a different apiserver than key names: the frame fails +// closed and nothing is cached, so the next frame re-evaluates cleanly. +// +// maxEntries soft-bounds the memo: past it, expired entries are swept (time-gated +// so a large live working set doesn't trigger an O(n) sweep every frame) before +// the next insert. contextName and now are injectable for tests. +func memoizedAuthorizer(base func(group, resource, namespace, verb string) (bool, bool), ttl, negativeTTL time.Duration, maxEntries int, contextName func() string, now func() time.Time) func(group, resource, namespace, verb string) bool { var mu sync.Mutex - memo := make(map[string]entry) + memo := make(map[string]authMemoEntry) + var lastSweep time.Time return func(group, resource, namespace, verb string) bool { ctxName := "" if contextName != nil { @@ -4549,24 +4588,35 @@ func memoizedAuthorizer(base func(group, resource, namespace, verb string) (bool } key := ctxName + "\x00" + verb + "\x00" + group + "\x00" + resource + "\x00" + namespace t := now() + mu.Lock() if e, ok := memo[key]; ok && t.Before(e.expires) { mu.Unlock() return e.allowed } mu.Unlock() + allowed, authoritative := base(group, resource, namespace, verb) - // Cache only an authoritative verdict, and only if the cluster context - // didn't change while base() was in flight — otherwise the result may - // reflect a different apiserver than `key` names, and a later switch-back - // within the TTL would serve the wrong cluster's decision. In either - // skip case the fail-closed result is still returned; the next frame - // re-evaluates cleanly. - if authoritative && (contextName == nil || contextName() == ctxName) { - mu.Lock() - memo[key] = entry{allowed: allowed, expires: t.Add(ttl)} - mu.Unlock() + + // A context switch that landed while base() ran decided this verdict + // against a different apiserver than key names. Fail closed for the frame + // and don't cache; the next frame re-evaluates against the new cluster. + if contextName != nil && contextName() != ctxName { + return false + } + + expiry := ttl + if !authoritative { + expiry = negativeTTL } + + mu.Lock() + if maxEntries > 0 && len(memo) >= maxEntries && (lastSweep.IsZero() || t.Sub(lastSweep) >= negativeTTL) { + sweepExpiredAuthMemo(memo, t) + lastSweep = t + } + memo[key] = authMemoEntry{allowed: allowed, expires: t.Add(expiry)} + mu.Unlock() return allowed } } diff --git a/internal/server/sse_authorizer_test.go b/internal/server/sse_authorizer_test.go index 9ba3cda5d..ef8884486 100644 --- a/internal/server/sse_authorizer_test.go +++ b/internal/server/sse_authorizer_test.go @@ -19,7 +19,7 @@ func TestMemoizedAuthorizer(t *testing.T) { } clock := time.Now() ctxName := "cluster-a" - authz := memoizedAuthorizer(base, 2*time.Minute, func() string { return ctxName }, func() time.Time { return clock }) + authz := memoizedAuthorizer(base, 2*time.Minute, 10*time.Second, 0, func() string { return ctxName }, func() time.Time { return clock }) if !authz("", "secrets", "team-a", "list") { t.Fatal("secrets/list should be allowed") @@ -61,7 +61,7 @@ func TestMemoizedAuthorizer_ContextSwitchIsolatesDecisions(t *testing.T) { return ctxName == "cluster-a", true } clock := time.Now() - authz := memoizedAuthorizer(base, 2*time.Minute, func() string { return ctxName }, func() time.Time { return clock }) + authz := memoizedAuthorizer(base, 2*time.Minute, 10*time.Second, 0, func() string { return ctxName }, func() time.Time { return clock }) ctxName = "cluster-a" if !authz("", "secrets", "team-a", "list") { @@ -78,10 +78,11 @@ func TestMemoizedAuthorizer_ContextSwitchIsolatesDecisions(t *testing.T) { } // If a context switch lands while the SAR is in flight, the fresh result was -// decided against a different apiserver than the key names — it must NOT be +// decided against a different apiserver than the key names. The frame must fail +// closed (the verdict may reflect the wrong cluster), and the result must NOT be // cached, or a switch-back within the TTL would serve the wrong cluster's // decision. -func TestMemoizedAuthorizer_NoCacheWhenContextChangesDuringSAR(t *testing.T) { +func TestMemoizedAuthorizer_FailsClosedAndNoCacheWhenContextChangesDuringSAR(t *testing.T) { calls := 0 ctxName := "cluster-a" base := func(group, resource, namespace, verb string) (bool, bool) { @@ -93,11 +94,13 @@ func TestMemoizedAuthorizer_NoCacheWhenContextChangesDuringSAR(t *testing.T) { return true, true } clock := time.Now() - authz := memoizedAuthorizer(base, 2*time.Minute, func() string { return ctxName }, func() time.Time { return clock }) + authz := memoizedAuthorizer(base, 2*time.Minute, 10*time.Second, 0, func() string { return ctxName }, func() time.Time { return clock }) - // Key computed under cluster-a, but context flips to cluster-b during base → - // result must not be cached under the cluster-a key. - authz("", "secrets", "team-a", "list") + // Key computed under cluster-a, but context flips to cluster-b during base. + // The (possibly wrong-cluster) allow must not release the frame — fail closed. + if authz("", "secrets", "team-a", "list") { + t.Fatal("switch-straddling SAR must fail closed, not release the frame on the wrong cluster's verdict") + } // Back on cluster-a within the TTL: a cached (poisoned) entry would be served // with no new base call. A fresh call proves it wasn't cached. ctxName = "cluster-a" @@ -107,40 +110,73 @@ func TestMemoizedAuthorizer_NoCacheWhenContextChangesDuringSAR(t *testing.T) { } } -// A transient SAR failure (authoritative=false) fails closed for that frame but -// must NOT be cached — otherwise a momentary apiserver blip would deny the tuple -// for the whole TTL. The next frame must retry, and a subsequent success must be -// served correctly. -func TestMemoizedAuthorizer_TransientFailureNotCached(t *testing.T) { +// A transient SAR failure (authoritative=false) fails closed for that frame and +// is cached only for the short negativeTTL — long enough that a degraded +// apiserver isn't re-SAR'd on every frame for the same tuple (which would stall +// the single broadcast goroutine), short enough that a momentary blip can't deny +// a readable tuple for the full TTL. Once negativeTTL elapses the tuple re-checks +// and a recovered apiserver's authoritative allow is cached for the full TTL. +func TestMemoizedAuthorizer_TransientFailureCachedBriefly(t *testing.T) { calls := 0 // First call is a transient failure (deny, non-authoritative); afterwards the // apiserver recovers and authoritatively allows. base := func(group, resource, namespace, verb string) (bool, bool) { calls++ if calls == 1 { - return false, false // transient failure: fail-closed, don't cache + return false, false // transient failure: fail-closed, cache only briefly } return true, true // recovered: authoritative allow } clock := time.Now() ctxName := "cluster-a" - authz := memoizedAuthorizer(base, 2*time.Minute, func() string { return ctxName }, func() time.Time { return clock }) + negativeTTL := 10 * time.Second + authz := memoizedAuthorizer(base, 2*time.Minute, negativeTTL, 0, func() string { return ctxName }, func() time.Time { return clock }) if authz("", "secrets", "team-a", "list") { t.Fatal("transient failure must fail closed (deny) for this frame") } - // Same tuple, still within TTL: the failure must not have been cached, so the - // retry runs base again and now succeeds. + // Same tuple within the negativeTTL: served from the brief negative cache, so + // the degraded apiserver is NOT re-SAR'd — no new base call. + if authz("", "secrets", "team-a", "list") { + t.Fatal("negative cache must keep failing closed within negativeTTL") + } + if calls != 1 { + t.Fatalf("transient failure should be cached for negativeTTL; want 1 base call, got %d", calls) + } + + // Past the negativeTTL the tuple re-checks; the apiserver has recovered. + clock = clock.Add(negativeTTL + time.Second) if !authz("", "secrets", "team-a", "list") { - t.Fatal("transient failure was cached — retry should have re-run base and allowed") + t.Fatal("after negativeTTL the recovered apiserver should allow") } if calls != 2 { - t.Fatalf("want 2 base calls (failure not cached, retried), got %d", calls) + t.Fatalf("want a re-check after negativeTTL, got %d base calls", calls) } - // The authoritative allow IS cached now — a third call is served from memo. + // The authoritative allow IS cached for the full TTL — a further call within it + // is served from memo. + clock = clock.Add(negativeTTL + time.Second) // still well within the 2m TTL authz("", "secrets", "team-a", "list") if calls != 2 { - t.Fatalf("authoritative allow should be cached; want still 2 base calls, got %d", calls) + t.Fatalf("authoritative allow should be cached for the full TTL; want still 2 base calls, got %d", calls) + } +} + +// sweepExpiredAuthMemo reclaims only the entries whose TTL has elapsed, leaving +// live ones intact — the memory bound for a long-lived all-namespace stream that +// would otherwise accumulate expired entries for every observed tuple. +func TestSweepExpiredAuthMemo(t *testing.T) { + base := time.Now() + memo := map[string]authMemoEntry{ + "expired-a": {allowed: true, expires: base.Add(-time.Second)}, + "expired-b": {allowed: false, expires: base}, // expires == now → expired + "live": {allowed: true, expires: base.Add(time.Minute)}, + } + removed := sweepExpiredAuthMemo(memo, base) + if removed != 2 { + t.Fatalf("want 2 expired entries removed, got %d", removed) + } + if _, ok := memo["live"]; !ok || len(memo) != 1 { + t.Fatalf("sweep must keep only the live entry, got %v", memo) } }