From 448cd2b5bd23c45671734884cffaa0384cd703cc Mon Sep 17 00:00:00 2001 From: ZverGuy Date: Wed, 4 Nov 2026 04:13:08 +0700 Subject: [PATCH] feat(observability-proxy): authenticate and authorize with cached reviews The proxy holds no credentials of its own: kube-apiserver answers who a bearer token belongs to and whether that identity may read a scope, the same way an aggregated API server would ask through delegated auth. Both answers are cached briefly so a dashboard polling several panels does not turn into a review per request, while transport errors are left uncached so an apiserver blip does not lock users out for the TTL. Assisted-by: LLM Signed-off-by: ZverGuy --- internal/observabilityproxy/auth.go | 164 +++++++++++++++++++++ internal/observabilityproxy/auth_test.go | 173 +++++++++++++++++++++++ 2 files changed, 337 insertions(+) create mode 100644 internal/observabilityproxy/auth.go create mode 100644 internal/observabilityproxy/auth_test.go diff --git a/internal/observabilityproxy/auth.go b/internal/observabilityproxy/auth.go new file mode 100644 index 000000000..90756bdad --- /dev/null +++ b/internal/observabilityproxy/auth.go @@ -0,0 +1,164 @@ +/* +Copyright 2026 The Cozystack Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package observabilityproxy + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sync" + "time" + + authenticationv1 "k8s.io/api/authentication/v1" + authorizationv1 "k8s.io/api/authorization/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" +) + +// ErrUnauthenticated means the token is missing or kube-apiserver rejected it. +var ErrUnauthenticated = errors.New("unauthenticated") + +// Authorizer asks kube-apiserver who a token belongs to and whether that +// identity may read a scope. Answers, including negative ones, are cached +// for the TTL; transport errors are not, so an apiserver blip is retried on +// the next request instead of locking a user out for the TTL. +type Authorizer struct { + client kubernetes.Interface + now func() time.Time + tokens *ttlCache[*authenticationv1.UserInfo] + decisions *ttlCache[bool] +} + +func NewAuthorizer(client kubernetes.Interface, ttl time.Duration) *Authorizer { + return &Authorizer{ + client: client, + now: time.Now, + tokens: newTTLCache[*authenticationv1.UserInfo](ttl), + decisions: newTTLCache[bool](ttl), + } +} + +func (a *Authorizer) Authenticate(ctx context.Context, token string) (*authenticationv1.UserInfo, error) { + if token == "" { + return nil, ErrUnauthenticated + } + sum := sha256.Sum256([]byte(token)) + key := hex.EncodeToString(sum[:]) + if u, ok := a.tokens.get(key, a.now()); ok { + if u == nil { + return nil, ErrUnauthenticated + } + return u, nil + } + tr, err := a.client.AuthenticationV1().TokenReviews().Create(ctx, + &authenticationv1.TokenReview{Spec: authenticationv1.TokenReviewSpec{Token: token}}, metav1.CreateOptions{}) + if err != nil { + return nil, fmt.Errorf("token review: %w", err) + } + var u *authenticationv1.UserInfo + if tr.Status.Authenticated && tr.Status.User.Username != "" { + u = tr.Status.User.DeepCopy() + } + a.tokens.set(key, u, a.now()) + if u == nil { + return nil, ErrUnauthenticated + } + return u, nil +} + +func (a *Authorizer) Authorize(ctx context.Context, user *authenticationv1.UserInfo, attrs authorizationv1.ResourceAttributes) (bool, error) { + extra := make(map[string]authorizationv1.ExtraValue, len(user.Extra)) + for k, v := range user.Extra { + extra[k] = authorizationv1.ExtraValue(v) + } + spec := authorizationv1.SubjectAccessReviewSpec{ + ResourceAttributes: &attrs, + User: user.Username, + UID: user.UID, + Groups: user.Groups, + Extra: extra, + } + raw, err := json.Marshal(spec) + if err != nil { + return false, err + } + key := string(raw) + if allowed, ok := a.decisions.get(key, a.now()); ok { + return allowed, nil + } + sar, err := a.client.AuthorizationV1().SubjectAccessReviews().Create(ctx, + &authorizationv1.SubjectAccessReview{Spec: spec}, metav1.CreateOptions{}) + if err != nil { + return false, fmt.Errorf("subject access review: %w", err) + } + allowed := sar.Status.Allowed && !sar.Status.Denied + a.decisions.set(key, allowed, a.now()) + return allowed, nil +} + +// ttlCacheMaxEntries bounds memory against a flood of distinct tokens; once +// reached, expired entries are swept and, if that is not enough, the cache +// starts over, which costs only extra reviews. +const ttlCacheMaxEntries = 10000 + +type ttlEntry[V any] struct { + value V + expires time.Time +} + +type ttlCache[V any] struct { + mu sync.Mutex + ttl time.Duration + entries map[string]ttlEntry[V] +} + +func newTTLCache[V any](ttl time.Duration) *ttlCache[V] { + return &ttlCache[V]{ttl: ttl, entries: map[string]ttlEntry[V]{}} +} + +func (c *ttlCache[V]) get(key string, now time.Time) (V, bool) { + c.mu.Lock() + defer c.mu.Unlock() + e, ok := c.entries[key] + if !ok || !now.Before(e.expires) { + var zero V + return zero, false + } + return e.value, true +} + +func (c *ttlCache[V]) set(key string, v V, now time.Time) { + if c.ttl <= 0 { + return + } + c.mu.Lock() + defer c.mu.Unlock() + if len(c.entries) >= ttlCacheMaxEntries { + for k, e := range c.entries { + if !now.Before(e.expires) { + delete(c.entries, k) + } + } + if len(c.entries) >= ttlCacheMaxEntries { + c.entries = map[string]ttlEntry[V]{} + } + } + c.entries[key] = ttlEntry[V]{value: v, expires: now.Add(c.ttl)} +} diff --git a/internal/observabilityproxy/auth_test.go b/internal/observabilityproxy/auth_test.go new file mode 100644 index 000000000..b57b80cf1 --- /dev/null +++ b/internal/observabilityproxy/auth_test.go @@ -0,0 +1,173 @@ +/* +Copyright 2026 The Cozystack Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package observabilityproxy + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + authenticationv1 "k8s.io/api/authentication/v1" + authorizationv1 "k8s.io/api/authorization/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" +) + +// fakeKube answers TokenReviews for the tokens in users and allows a +// SubjectAccessReview when allow returns true; it counts both calls. +type fakeKube struct { + client *fake.Clientset + reviews atomic.Int32 + sars atomic.Int32 + lastSAR atomic.Pointer[authorizationv1.SubjectAccessReviewSpec] + reviewFn func() error +} + +func newFakeKube(users map[string]string, allow func(user string, attrs *authorizationv1.ResourceAttributes) bool) *fakeKube { + f := &fakeKube{client: fake.NewClientset()} + f.client.PrependReactor("create", "tokenreviews", func(action k8stesting.Action) (bool, runtime.Object, error) { + f.reviews.Add(1) + if f.reviewFn != nil { + if err := f.reviewFn(); err != nil { + return true, nil, err + } + } + tr := action.(k8stesting.CreateAction).GetObject().(*authenticationv1.TokenReview).DeepCopy() + if name, ok := users[tr.Spec.Token]; ok { + tr.Status = authenticationv1.TokenReviewStatus{Authenticated: true, User: authenticationv1.UserInfo{Username: name, Groups: []string{"tenant-a-view"}}} + } + return true, tr, nil + }) + f.client.PrependReactor("create", "subjectaccessreviews", func(action k8stesting.Action) (bool, runtime.Object, error) { + f.sars.Add(1) + sar := action.(k8stesting.CreateAction).GetObject().(*authorizationv1.SubjectAccessReview).DeepCopy() + spec := sar.Spec + f.lastSAR.Store(&spec) + sar.Status.Allowed = allow(sar.Spec.User, sar.Spec.ResourceAttributes) + return true, sar, nil + }) + return f +} + +func allowNamespace(ns string) func(string, *authorizationv1.ResourceAttributes) bool { + return func(_ string, a *authorizationv1.ResourceAttributes) bool { return a != nil && a.Namespace == ns } +} + +func TestAuthenticate(t *testing.T) { + kube := newFakeKube(map[string]string{"good": "alice"}, allowNamespace("tenant-a")) + a := NewAuthorizer(kube.client, 10*time.Second) + ctx := context.Background() + + u, err := a.Authenticate(ctx, "good") + if err != nil || u.Username != "alice" { + t.Fatalf("Authenticate(good) = %+v, %v", u, err) + } + if _, err := a.Authenticate(ctx, "good"); err != nil { + t.Fatal(err) + } + if n := kube.reviews.Load(); n != 1 { + t.Fatalf("TokenReviews = %d, want 1 (second call cached)", n) + } + if _, err := a.Authenticate(ctx, "bad"); !errors.Is(err, ErrUnauthenticated) { + t.Fatalf("Authenticate(bad) err = %v, want ErrUnauthenticated", err) + } + if _, err := a.Authenticate(ctx, ""); !errors.Is(err, ErrUnauthenticated) { + t.Fatalf("Authenticate(\"\") err = %v, want ErrUnauthenticated", err) + } +} + +func TestAuthenticateCacheExpires(t *testing.T) { + kube := newFakeKube(map[string]string{"good": "alice"}, allowNamespace("tenant-a")) + a := NewAuthorizer(kube.client, 10*time.Second) + now := time.Unix(1000, 0) + a.now = func() time.Time { return now } + ctx := context.Background() + if _, err := a.Authenticate(ctx, "good"); err != nil { + t.Fatal(err) + } + now = now.Add(11 * time.Second) + if _, err := a.Authenticate(ctx, "good"); err != nil { + t.Fatal(err) + } + if n := kube.reviews.Load(); n != 2 { + t.Fatalf("TokenReviews = %d, want 2 after TTL", n) + } +} + +func TestAuthenticateErrorNotCached(t *testing.T) { + kube := newFakeKube(map[string]string{"good": "alice"}, allowNamespace("tenant-a")) + kube.reviewFn = func() error { return errors.New("apiserver down") } + a := NewAuthorizer(kube.client, 10*time.Second) + ctx := context.Background() + if _, err := a.Authenticate(ctx, "good"); err == nil || errors.Is(err, ErrUnauthenticated) { + t.Fatalf("err = %v, want a transport error", err) + } + kube.reviewFn = nil + if _, err := a.Authenticate(ctx, "good"); err != nil { + t.Fatalf("err = %v after recovery", err) + } +} + +func TestAuthorize(t *testing.T) { + kube := newFakeKube(map[string]string{"good": "alice"}, allowNamespace("tenant-a")) + a := NewAuthorizer(kube.client, 10*time.Second) + ctx := context.Background() + user := &authenticationv1.UserInfo{Username: "alice", UID: "u1", Groups: []string{"g"}, Extra: map[string]authenticationv1.ExtraValue{"scope": {"x"}}} + + allowed := Route{Kind: NamespaceLogs, Namespace: "tenant-a"} + denied := Route{Kind: NamespaceLogs, Namespace: "tenant-b"} + + ok, err := a.Authorize(ctx, user, allowed.ResourceAttributes()) + if err != nil || !ok { + t.Fatalf("allow: %v, %v", ok, err) + } + spec := kube.lastSAR.Load() + if spec.User != "alice" || spec.UID != "u1" || len(spec.Groups) != 1 || spec.Extra["scope"][0] != "x" { + t.Fatalf("SAR identity not forwarded: %+v", spec) + } + if ra := spec.ResourceAttributes; ra.Verb != "list" || ra.Group != "cozystack.io" || ra.Resource != "logs" || ra.Namespace != "tenant-a" { + t.Fatalf("SAR attributes = %+v", ra) + } + + ok, err = a.Authorize(ctx, user, denied.ResourceAttributes()) + if err != nil || ok { + t.Fatalf("deny: %v, %v", ok, err) + } + + for range 3 { + if ok, _ := a.Authorize(ctx, user, allowed.ResourceAttributes()); !ok { + t.Fatal("cached allow flipped") + } + if ok, _ := a.Authorize(ctx, user, denied.ResourceAttributes()); ok { + t.Fatal("cached deny flipped") + } + } + if n := kube.sars.Load(); n != 2 { + t.Fatalf("SubjectAccessReviews = %d, want 2 (rest cached)", n) + } + + other := &authenticationv1.UserInfo{Username: "alice", UID: "u1", Groups: []string{"g", "tenant-b-view"}} + if _, err := a.Authorize(ctx, other, denied.ResourceAttributes()); err != nil { + t.Fatal(err) + } + if n := kube.sars.Load(); n != 3 { + t.Fatalf("a different group set must not hit another identity's cache entry, SARs = %d", n) + } +}