diff --git a/internal/server/server.go b/internal/server/server.go index 32f18da92..985ef220f 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1245,23 +1245,44 @@ func (s *Server) canReadUser(ctx context.Context, user *auth.User, group, resour 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 @@ -4464,11 +4485,140 @@ 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 + // 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 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..ef8884486 --- /dev/null +++ b/internal/server/sse_authorizer_test.go @@ -0,0 +1,191 @@ +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. All verdicts here are + // authoritative (a real apiserver answer). + base := func(group, resource, namespace, verb string) (bool, bool) { + calls++ + return verb == "list" && resource == "secrets", true + } + clock := time.Now() + ctxName := "cluster-a" + 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") + } + // 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, bool) { + calls++ + return ctxName == "cluster-a", true + } + clock := time.Now() + 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") { + 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. 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_FailsClosedAndNoCacheWhenContextChangesDuringSAR(t *testing.T) { + calls := 0 + ctxName := "cluster-a" + 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, true + } + clock := time.Now() + 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. + // 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" + authz("", "secrets", "team-a", "list") + if calls != 2 { + t.Fatalf("switch-straddling SAR must not be cached; want 2 base calls, got %d", calls) + } +} + +// 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, cache only briefly + } + return true, true // recovered: authoritative allow + } + clock := time.Now() + ctxName := "cluster-a" + 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 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("after negativeTTL the recovered apiserver should allow") + } + if calls != 2 { + t.Fatalf("want a re-check after negativeTTL, got %d base calls", calls) + } + // 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 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) + } +} + +// 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") + } +}