fix(rbac): bound SSE change-authorizer with a per-connection SAR memo

internal/server/server.go

@@ -1245,23 +1245,44 @@
 			return v
 		}
 	}
+	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
+}
+
+// canReadUserSAR runs a single fresh SubjectAccessReview for (group, resource,
+// namespace, verb) against the current apiserver, bypassing the shared
+// 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
+// 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) (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] canReadUser: 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
+		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, 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)
-		return false
-	}
-	if perms != nil {
-		perms.SetCanI(verb, group, resource, namespace, allowed)
+		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, false
 	}
-	return allowed
+	return allowed, true
 }
 
 // filterNamespacesByCanRead returns the subset of `namespaces` where the
@@ -4451,7 +4472,7 @@
 	// namespace-restricted user doesn't learn namespace names via SSE.
 	if !s.canRead(r, "", "namespaces", "", "list") {
 		if deny == nil {
 			deny = map[topology.NodeKind]bool{}
 		}
 		deny[topology.KindNamespace] = true
 	}
@@ -4464,11 +4485,140 @@
 	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
+	// 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
+// 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, bool) {
+		sarCtx, cancel := context.WithTimeout(ctx, sseChangeAuthSARTimeout)
+		defer cancel()
+		return s.canReadUserSAR(sarCtx, user, group, resource, namespace, verb)
+	}
+	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):
+//
+//   - 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]authMemoEntry)
+	var lastSweep time.Time
+	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, authoritative := base(group, resource, namespace, verb)
+
+		// 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
 	}
-	s.broadcaster.HandleSSE(w, r, deny, authorize)
 }
 
 // Settings handlers

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
 }