From 35d2405183d0fa230be157000dd6358ca7bd5fb4 Mon Sep 17 00:00:00 2001 From: goingforstudying-ctrl Date: Fri, 17 Jul 2026 03:10:18 -0400 Subject: [PATCH 1/6] grpcproxy: scope serializable Range cache entries by caller auth identity The grpc-proxy's serializable Range cache keyed entries only on the marshalled RangeRequest, never on who was asking. Once a privileged client read a key through the proxy, any other client issuing the same serializable Range got the cached response back without the backend ever seeing the request or enforcing its auth policy. Low-priv or unauthenticated callers could read arbitrary protected keys this way, and revoking a permission did nothing while a stale entry sat in the proxy's LRU. Two changes close the hole: - The cache key now includes a SHA-256 of the caller's auth token, so two different credentials never share an entry. Tokenless requests share one entry, same as before. - Auth-mutating RPCs (RoleGrantPermission, UserRevokeRole, etc.) forwarded through the auth proxy flush the whole KV cache, so a caller whose permissions were just revoked can't keep reading previously cached data. Regression tests cover cross-identity leaks for serializable and linearizable reads, plus the permission-revocation path driven through the real AuthProxy. Fixes #22065 Signed-off-by: goingforstudying-ctrl --- server/etcdmain/grpc_proxy.go | 2 +- server/proxy/grpcproxy/auth.go | 63 ++- server/proxy/grpcproxy/cache/store.go | 50 ++- server/proxy/grpcproxy/cache/store_test.go | 118 ++++++ server/proxy/grpcproxy/kv.go | 40 +- tests/framework/integration/cluster_proxy.go | 2 +- .../proxy/grpcproxy/kv_auth_cache_test.go | 376 ++++++++++++++++++ 7 files changed, 620 insertions(+), 31 deletions(-) create mode 100644 server/proxy/grpcproxy/cache/store_test.go create mode 100644 tests/integration/proxy/grpcproxy/kv_auth_cache_test.go diff --git a/server/etcdmain/grpc_proxy.go b/server/etcdmain/grpc_proxy.go index f444a5f394c3..cdb2dcfabb68 100644 --- a/server/etcdmain/grpc_proxy.go +++ b/server/etcdmain/grpc_proxy.go @@ -498,7 +498,7 @@ func newGRPCProxyServer(lg *zap.Logger, client *clientv3.Client) *grpc.Server { leasep, _ := grpcproxy.NewLeaseProxy(client.Ctx(), client) mainp := grpcproxy.NewMaintenanceProxy(client) - authp := grpcproxy.NewAuthProxy(client) + authp := grpcproxy.NewAuthProxy(client, kvp.(*grpcproxy.KvProxy)) electionp := grpcproxy.NewElectionProxy(client) lockp := grpcproxy.NewLockProxy(client) diff --git a/server/proxy/grpcproxy/auth.go b/server/proxy/grpcproxy/auth.go index f85f8b223d3f..778e5740b076 100644 --- a/server/proxy/grpcproxy/auth.go +++ b/server/proxy/grpcproxy/auth.go @@ -19,24 +19,43 @@ import ( pb "go.etcd.io/etcd/api/v3/etcdserverpb" clientv3 "go.etcd.io/etcd/client/v3" + "go.etcd.io/etcd/server/v3/proxy/grpcproxy/cache" ) type AuthProxy struct { authClient pb.AuthClient + cache cache.Cache // we want compile errors if new methods are added pb.UnsafeAuthServer } -func NewAuthProxy(c *clientv3.Client) pb.AuthServer { - return &AuthProxy{authClient: pb.NewAuthClient(c.ActiveConnection())} +// NewAuthProxy returns an AuthServer that forwards auth RPCs to the backend. +// kvp may be nil; when non-nil, any auth-mutating RPC flushes the KV proxy +// cache so no caller can keep reading data cached under stale permissions. +func NewAuthProxy(c *clientv3.Client, kvp *kvProxy) pb.AuthServer { + ap := &AuthProxy{authClient: pb.NewAuthClient(c.ActiveConnection())} + if kvp != nil { + ap.cache = kvp.cache + } + return ap } func (ap *AuthProxy) AuthEnable(ctx context.Context, r *pb.AuthEnableRequest) (*pb.AuthEnableResponse, error) { - return ap.authClient.AuthEnable(ctx, r) + resp, err := ap.authClient.AuthEnable(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err } func (ap *AuthProxy) AuthDisable(ctx context.Context, r *pb.AuthDisableRequest) (*pb.AuthDisableResponse, error) { - return ap.authClient.AuthDisable(ctx, r) + resp, err := ap.authClient.AuthDisable(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err +} + +func (ap *AuthProxy) flushCacheOnMutation(err error) { + if err == nil && ap.cache != nil { + ap.cache.Clear() + } } func (ap *AuthProxy) AuthStatus(ctx context.Context, r *pb.AuthStatusRequest) (*pb.AuthStatusResponse, error) { @@ -48,11 +67,15 @@ func (ap *AuthProxy) Authenticate(ctx context.Context, r *pb.AuthenticateRequest } func (ap *AuthProxy) RoleAdd(ctx context.Context, r *pb.AuthRoleAddRequest) (*pb.AuthRoleAddResponse, error) { - return ap.authClient.RoleAdd(ctx, r) + resp, err := ap.authClient.RoleAdd(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err } func (ap *AuthProxy) RoleDelete(ctx context.Context, r *pb.AuthRoleDeleteRequest) (*pb.AuthRoleDeleteResponse, error) { - return ap.authClient.RoleDelete(ctx, r) + resp, err := ap.authClient.RoleDelete(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err } func (ap *AuthProxy) RoleGet(ctx context.Context, r *pb.AuthRoleGetRequest) (*pb.AuthRoleGetResponse, error) { @@ -64,19 +87,27 @@ func (ap *AuthProxy) RoleList(ctx context.Context, r *pb.AuthRoleListRequest) (* } func (ap *AuthProxy) RoleRevokePermission(ctx context.Context, r *pb.AuthRoleRevokePermissionRequest) (*pb.AuthRoleRevokePermissionResponse, error) { - return ap.authClient.RoleRevokePermission(ctx, r) + resp, err := ap.authClient.RoleRevokePermission(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err } func (ap *AuthProxy) RoleGrantPermission(ctx context.Context, r *pb.AuthRoleGrantPermissionRequest) (*pb.AuthRoleGrantPermissionResponse, error) { - return ap.authClient.RoleGrantPermission(ctx, r) + resp, err := ap.authClient.RoleGrantPermission(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err } func (ap *AuthProxy) UserAdd(ctx context.Context, r *pb.AuthUserAddRequest) (*pb.AuthUserAddResponse, error) { - return ap.authClient.UserAdd(ctx, r) + resp, err := ap.authClient.UserAdd(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err } func (ap *AuthProxy) UserDelete(ctx context.Context, r *pb.AuthUserDeleteRequest) (*pb.AuthUserDeleteResponse, error) { - return ap.authClient.UserDelete(ctx, r) + resp, err := ap.authClient.UserDelete(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err } func (ap *AuthProxy) UserGet(ctx context.Context, r *pb.AuthUserGetRequest) (*pb.AuthUserGetResponse, error) { @@ -88,13 +119,19 @@ func (ap *AuthProxy) UserList(ctx context.Context, r *pb.AuthUserListRequest) (* } func (ap *AuthProxy) UserGrantRole(ctx context.Context, r *pb.AuthUserGrantRoleRequest) (*pb.AuthUserGrantRoleResponse, error) { - return ap.authClient.UserGrantRole(ctx, r) + resp, err := ap.authClient.UserGrantRole(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err } func (ap *AuthProxy) UserRevokeRole(ctx context.Context, r *pb.AuthUserRevokeRoleRequest) (*pb.AuthUserRevokeRoleResponse, error) { - return ap.authClient.UserRevokeRole(ctx, r) + resp, err := ap.authClient.UserRevokeRole(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err } func (ap *AuthProxy) UserChangePassword(ctx context.Context, r *pb.AuthUserChangePasswordRequest) (*pb.AuthUserChangePasswordResponse, error) { - return ap.authClient.UserChangePassword(ctx, r) + resp, err := ap.authClient.UserChangePassword(ctx, r) + ap.flushCacheOnMutation(err) + return resp, err } diff --git a/server/proxy/grpcproxy/cache/store.go b/server/proxy/grpcproxy/cache/store.go index 9ddd2370ba87..a405c0a5a3de 100644 --- a/server/proxy/grpcproxy/cache/store.go +++ b/server/proxy/grpcproxy/cache/store.go @@ -34,8 +34,16 @@ var ( ) type Cache interface { - Add(req *pb.RangeRequest, resp *pb.RangeResponse) - Get(req *pb.RangeRequest) (*pb.RangeResponse, error) + // Add stores resp for (req, authKey). authKey scopes the entry to a + // single caller identity so a cached response can never be replayed to + // a different principal; pass "" when the request carries no credentials. + Add(req *pb.RangeRequest, resp *pb.RangeResponse, authKey string) + // Get returns the cached response for (req, authKey). + Get(req *pb.RangeRequest, authKey string) (*pb.RangeResponse, error) + // Clear drops every cached entry. Used when the authorization + // configuration may have changed (user/role/permission updates) so no + // principal keeps reading data cached under stale permissions. + Clear() Compact(revision int64) Invalidate(key []byte, endkey []byte) Size() int @@ -43,18 +51,28 @@ type Cache interface { } // keyFunc returns the key of a request, which is used to look up its caching response in the cache. -func keyFunc(req *pb.RangeRequest) string { +// The caller identity (authKey) is mixed in so that a response cached for +// one authenticated user is never served to a different user. +func keyFunc(req *pb.RangeRequest, authKey string) string { // TODO: use marshalTo to reduce allocation b, err := proto.Marshal(req) if err != nil { panic(err) } - return string(b) + if authKey == "" { + return string(b) + } + out := make([]byte, 0, len(b)+1+len(authKey)) + out = append(out, b...) + out = append(out, 0) + out = append(out, authKey...) + return string(out) } func NewCache(maxCacheEntries int) Cache { return &cache{ lru: lru.New(maxCacheEntries), + maxEntries: maxCacheEntries, cachedRanges: adt.NewIntervalTree(), compactedRev: -1, } @@ -64,8 +82,9 @@ func (c *cache) Close() {} // cache implements Cache type cache struct { - mu sync.RWMutex - lru *lru.Cache + mu sync.RWMutex + lru *lru.Cache + maxEntries int // a reverse index for cache invalidation cachedRanges adt.IntervalTree @@ -74,8 +93,8 @@ type cache struct { } // Add adds the response of a request to the cache if its revision is larger than the compacted revision of the cache. -func (c *cache) Add(req *pb.RangeRequest, resp *pb.RangeResponse) { - key := keyFunc(req) +func (c *cache) Add(req *pb.RangeRequest, resp *pb.RangeResponse, authKey string) { + key := keyFunc(req, authKey) c.mu.Lock() defer c.mu.Unlock() @@ -113,8 +132,8 @@ func (c *cache) Add(req *pb.RangeRequest, resp *pb.RangeResponse) { // Get looks up the caching response for a given request. // Get is also responsible for lazy eviction when accessing compacted entries. -func (c *cache) Get(req *pb.RangeRequest) (*pb.RangeResponse, error) { - key := keyFunc(req) +func (c *cache) Get(req *pb.RangeRequest, authKey string) (*pb.RangeResponse, error) { + key := keyFunc(req, authKey) c.mu.Lock() defer c.mu.Unlock() @@ -130,6 +149,17 @@ func (c *cache) Get(req *pb.RangeRequest) (*pb.RangeResponse, error) { return nil, errors.New("not exist") } +// Clear drops all cached entries and resets the reverse invalidation index. +// It is called when the authorization configuration may have changed so no +// principal can keep reading data cached under stale permissions. +func (c *cache) Clear() { + c.mu.Lock() + defer c.mu.Unlock() + + c.lru = lru.New(c.maxEntries) + c.cachedRanges = adt.NewIntervalTree() +} + // Invalidate invalidates the cache entries that intersecting with the given range from key to endkey. func (c *cache) Invalidate(key, endkey []byte) { c.mu.Lock() diff --git a/server/proxy/grpcproxy/cache/store_test.go b/server/proxy/grpcproxy/cache/store_test.go new file mode 100644 index 000000000000..debe2084af4a --- /dev/null +++ b/server/proxy/grpcproxy/cache/store_test.go @@ -0,0 +1,118 @@ +// Copyright 2026 The etcd 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 cache + +import ( + "testing" + + pb "go.etcd.io/etcd/api/v3/etcdserverpb" + "go.etcd.io/etcd/api/v3/mvccpb" +) + +func TestCacheAuthKeyIsolation(t *testing.T) { + c := NewCache(10) + defer c.Close() + + req := &pb.RangeRequest{Key: []byte("secret"), Serializable: true} + resp := &pb.RangeResponse{ + Header: &pb.ResponseHeader{Revision: 1}, + Kvs: []*mvccpb.KeyValue{{Key: []byte("secret"), Value: []byte("s3cr3t")}}, + } + + // Store under identity "alice" + c.Add(req, resp, "alice") + + // Same identity must hit + got, err := c.Get(req, "alice") + if err != nil { + t.Fatalf("expected cache hit for same identity, got %v", err) + } + if string(got.Kvs[0].Value) != "s3cr3t" { + t.Fatalf("unexpected value %q", got.Kvs[0].Value) + } + + // Different identity must miss + _, err = c.Get(req, "bob") + if err == nil { + t.Fatal("expected cache miss for different identity") + } + + // Empty identity must miss + _, err = c.Get(req, "") + if err == nil { + t.Fatal("expected cache miss for empty identity") + } + + // Store under empty identity; other identities still miss + c.Add(req, resp, "") + if _, err = c.Get(req, ""); err != nil { + t.Fatalf("expected cache hit for empty identity, got %v", err) + } + // "alice" and "" share the same key when authKey is empty (no-auth requests share one cache) + if _, err = c.Get(req, "alice"); err != nil { + t.Fatalf("expected cache hit for alice after empty-identity add (same key), got %v", err) + } +} + +func TestCacheClear(t *testing.T) { + c := NewCache(10) + defer c.Close() + + req := &pb.RangeRequest{Key: []byte("k"), Serializable: true} + resp := &pb.RangeResponse{Header: &pb.ResponseHeader{Revision: 1}} + + c.Add(req, resp, "alice") + c.Add(req, resp, "bob") + c.Add(req, resp, "") + + if size := c.Size(); size != 3 { + t.Fatalf("expected 3 entries, got %d", size) + } + + c.Clear() + + if size := c.Size(); size != 0 { + t.Fatalf("expected 0 entries after Clear, got %d", size) + } + for _, id := range []string{"alice", "bob", ""} { + if _, err := c.Get(req, id); err == nil { + t.Fatalf("expected miss after Clear for identity %q", id) + } + } + + // Cache must remain usable after Clear + c.Add(req, resp, "alice") + if _, err := c.Get(req, "alice"); err != nil { + t.Fatalf("expected hit after re-Add, got %v", err) + } +} + +func TestKeyFuncDeterministic(t *testing.T) { + req := &pb.RangeRequest{Key: []byte("a"), Serializable: true} + k1 := keyFunc(req, "alice") + k2 := keyFunc(req, "alice") + k3 := keyFunc(req, "bob") + k4 := keyFunc(req, "") + + if k1 != k2 { + t.Fatal("same inputs must produce same key") + } + if k1 == k3 { + t.Fatal("different identities must produce different keys") + } + if k1 == k4 { + t.Fatal("identity vs no-identity must produce different keys") + } +} diff --git a/server/proxy/grpcproxy/kv.go b/server/proxy/grpcproxy/kv.go index 3110c1e1050e..92f90d085f51 100644 --- a/server/proxy/grpcproxy/kv.go +++ b/server/proxy/grpcproxy/kv.go @@ -16,6 +16,8 @@ package grpcproxy import ( "context" + "crypto/sha256" + "encoding/hex" "errors" "google.golang.org/grpc/codes" @@ -34,6 +36,11 @@ type kvProxy struct { pb.UnsafeKVServer } +// KvProxy is the concrete KV proxy implementation returned by NewKvProxy. +// It is exported so that the auth proxy can be wired to the same cache +// instance (see NewAuthProxy). +type KvProxy = kvProxy + func NewKvProxy(c *clientv3.Client) (pb.KVServer, <-chan struct{}) { kv := &kvProxy{ kv: c.KV, @@ -44,9 +51,29 @@ func NewKvProxy(c *clientv3.Client) (pb.KVServer, <-chan struct{}) { return kv, donec } +// Cache returns the underlying cache instance. Exposed for tests and for +// wiring the auth proxy to the same cache. +func (p *kvProxy) Cache() cache.Cache { + return p.cache +} + +// authCacheKey derives the cache-key identity of the caller from the incoming +// context. Two different credentials never share a cache entry, so a response +// cached for one principal can never be replayed to another. The raw token is +// hashed so the LRU never holds credentials in plaintext. +func authCacheKey(ctx context.Context) string { + token := getAuthTokenFromClient(ctx) + if token == "" { + return "" + } + sum := sha256.Sum256([]byte(token)) + return hex.EncodeToString(sum[:]) +} + func (p *kvProxy) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) { + authKey := authCacheKey(ctx) if r.Serializable { - resp, err := p.cache.Get(r) + resp, err := p.cache.Get(r, authKey) switch { case err == nil: cacheHits.Inc() @@ -69,7 +96,7 @@ func (p *kvProxy) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeRespo req := proto.Clone(r).(*pb.RangeRequest) req.Serializable = true gresp := (*pb.RangeResponse)(resp.Get()) - p.cache.Add(req, gresp) + p.cache.Add(req, gresp, authKey) cacheKeys.Set(float64(p.cache.Size())) return gresp, nil @@ -95,7 +122,7 @@ func (p *kvProxy) DeleteRange(ctx context.Context, r *pb.DeleteRangeRequest) (*p return (*pb.DeleteRangeResponse)(resp.Del()), err } -func (p *kvProxy) txnToCache(reqs []*pb.RequestOp, resps []*pb.ResponseOp) { +func (p *kvProxy) txnToCache(reqs []*pb.RequestOp, resps []*pb.ResponseOp, authKey string) { for i := range resps { switch tv := resps[i].Response.(type) { case *pb.ResponseOp_ResponsePut: @@ -106,7 +133,7 @@ func (p *kvProxy) txnToCache(reqs []*pb.RequestOp, resps []*pb.ResponseOp) { case *pb.ResponseOp_ResponseRange: req := *(reqs[i].GetRequestRange()) req.Serializable = true - p.cache.Add(&req, tv.ResponseRange) + p.cache.Add(&req, tv.ResponseRange, authKey) } } } @@ -124,10 +151,11 @@ func (p *kvProxy) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, e p.cache.Invalidate(cmp.Key, cmp.RangeEnd) } // update any fetched keys + authKey := authCacheKey(ctx) if resp.Succeeded { - p.txnToCache(r.Success, resp.Responses) + p.txnToCache(r.Success, resp.Responses, authKey) } else { - p.txnToCache(r.Failure, resp.Responses) + p.txnToCache(r.Failure, resp.Responses, authKey) } cacheKeys.Set(float64(p.cache.Size())) diff --git a/tests/framework/integration/cluster_proxy.go b/tests/framework/integration/cluster_proxy.go index 0ba8b59d02f1..c96d1dd92ebe 100644 --- a/tests/framework/integration/cluster_proxy.go +++ b/tests/framework/integration/cluster_proxy.go @@ -69,7 +69,7 @@ func ToGRPC(c *clientv3.Client) GRPCAPI { lp, lpch := grpcproxy.NewLeaseProxy(ctx, c) mp := grpcproxy.NewMaintenanceProxy(c) clp, _ := grpcproxy.NewClusterProxy(lg, c, "", "") // without registering proxy URLs - authp := grpcproxy.NewAuthProxy(c) + authp := grpcproxy.NewAuthProxy(c, kvp.(*grpcproxy.KvProxy)) lockp := grpcproxy.NewLockProxy(c) electp := grpcproxy.NewElectionProxy(c) diff --git a/tests/integration/proxy/grpcproxy/kv_auth_cache_test.go b/tests/integration/proxy/grpcproxy/kv_auth_cache_test.go new file mode 100644 index 000000000000..45632e2d3b1d --- /dev/null +++ b/tests/integration/proxy/grpcproxy/kv_auth_cache_test.go @@ -0,0 +1,376 @@ +// Copyright 2026 The etcd 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 grpcproxy + +import ( + "context" + "net" + "testing" + "time" + + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + + "go.etcd.io/etcd/api/v3/authpb" + pb "go.etcd.io/etcd/api/v3/etcdserverpb" + "go.etcd.io/etcd/api/v3/v3rpc/rpctypes" + clientv3 "go.etcd.io/etcd/client/v3" + "go.etcd.io/etcd/server/v3/proxy/grpcproxy" + "go.etcd.io/etcd/tests/v3/framework/integration" +) + +// authInjectingKvServer wraps a real KVServer and stamps a fixed auth token +// into the incoming context metadata, simulating a request that arrived at +// the proxy with the given credential attached. +type authInjectingKvServer struct { + pb.KVServer + token string +} + +func (s *authInjectingKvServer) withToken(ctx context.Context) context.Context { + if s.token == "" { + return ctx + } + md, ok := metadata.FromIncomingContext(ctx) + if !ok { + md = metadata.New(nil) + } else { + md = md.Copy() + } + md.Set(rpctypes.TokenFieldNameGRPC, s.token) + return metadata.NewIncomingContext(ctx, md) +} + +func (s *authInjectingKvServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) { + return s.KVServer.Range(s.withToken(ctx), r) +} + +func (s *authInjectingKvServer) Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error) { + return s.KVServer.Put(s.withToken(ctx), r) +} + +func (s *authInjectingKvServer) DeleteRange(ctx context.Context, r *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error) { + return s.KVServer.DeleteRange(s.withToken(ctx), r) +} + +func (s *authInjectingKvServer) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error) { + return s.KVServer.Txn(s.withToken(ctx), r) +} + +func (s *authInjectingKvServer) Compact(ctx context.Context, r *pb.CompactionRequest) (*pb.CompactionResponse, error) { + return s.KVServer.Compact(s.withToken(ctx), r) +} + +// kvAuthTestHarness spins up a real backend member with auth enabled, then +// exposes the same kvProxy through three facades: one that stamps the root +// token, one that stamps a low-privilege token, and one with no token at +// all. This lets us drive the proxy cache exactly the way real gRPC clients +// with different credentials would. +type kvAuthTestHarness struct { + t *testing.T + clus *integration.Cluster + rootKV pb.KVClient + lowKV pb.KVClient + anonKV pb.KVClient + cleanup func() +} + +func newKVAuthTestHarness(t *testing.T) *kvAuthTestHarness { + t.Helper() + integration.BeforeTest(t) + + clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1}) + + h := &kvAuthTestHarness{t: t, clus: clus} + + // Set up root user + a low-priv user with no permissions, then enable auth. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + api := integration.ToGRPC(clus.Client(0)) + + _, err := api.Auth.UserAdd(ctx, &pb.AuthUserAddRequest{Name: "root", Password: "rootpw"}) + require.NoError(t, err) + _, err = api.Auth.RoleAdd(ctx, &pb.AuthRoleAddRequest{Name: "root"}) + require.NoError(t, err) + _, err = api.Auth.UserGrantRole(ctx, &pb.AuthUserGrantRoleRequest{User: "root", Role: "root"}) + require.NoError(t, err) + + _, err = api.Auth.RoleAdd(ctx, &pb.AuthRoleAddRequest{Name: "lowrole"}) + require.NoError(t, err) + _, err = api.Auth.UserAdd(ctx, &pb.AuthUserAddRequest{Name: "low", Password: "lowpw"}) + require.NoError(t, err) + _, err = api.Auth.UserGrantRole(ctx, &pb.AuthUserGrantRoleRequest{User: "low", Role: "lowrole"}) + require.NoError(t, err) + + // Write the protected key before enabling auth. + _, err = api.KV.Put(ctx, &pb.PutRequest{Key: []byte("/proxy-cache/secret"), Value: []byte("cached-secret")}) + require.NoError(t, err) + + _, err = api.Auth.AuthEnable(ctx, &pb.AuthEnableRequest{}) + require.NoError(t, err) + + // Mint tokens for both users. + rootAuth, err := api.Auth.Authenticate(ctx, &pb.AuthenticateRequest{Name: "root", Password: "rootpw"}) + require.NoError(t, err) + lowAuth, err := api.Auth.Authenticate(ctx, &pb.AuthenticateRequest{Name: "low", Password: "lowpw"}) + require.NoError(t, err) + + // Build a backend client that forwards any incoming token straight through, + // exactly like the production grpc-proxy does (AuthUnaryClientInterceptor). + backendCfg := clientv3.Config{ + Endpoints: []string{clus.Members[0].GRPCURL}, + DialTimeout: 5 * time.Second, + DialOptions: []grpc.DialOption{ + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithUnaryInterceptor(grpcproxy.AuthUnaryClientInterceptor), + grpc.WithStreamInterceptor(grpcproxy.AuthStreamClientInterceptor), + }, + } + backendClient, err := integration.NewClient(t, backendCfg) + require.NoError(t, err) + + kvp, _ := grpcproxy.NewKvProxy(backendClient) + + mkFacade := func(token string) (pb.KVClient, net.Listener, *grpc.Server, *grpc.ClientConn) { + srv := grpc.NewServer() + pb.RegisterKVServer(srv, &authInjectingKvServer{KVServer: kvp, token: token}) + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + go srv.Serve(l) + conn, err := grpc.NewClient(l.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + return pb.NewKVClient(conn), l, srv, conn + } + + var listeners []net.Listener + var servers []*grpc.Server + var conns []*grpc.ClientConn + + rootKV, l, s, cc := mkFacade(rootAuth.Token) + listeners, servers, conns = append(listeners, l), append(servers, s), append(conns, cc) + lowKV, l, s, cc := mkFacade(lowAuth.Token) + listeners, servers, conns = append(listeners, l), append(servers, s), append(conns, cc) + anonKV, l, s, cc := mkFacade("") + listeners, servers, conns = append(listeners, l), append(servers, s), append(conns, cc) + + h.rootKV, h.lowKV, h.anonKV = rootKV, lowKV, anonKV + h.cleanup = func() { + for _, cc := range conns { + cc.Close() + } + for _, s := range servers { + s.Stop() + } + for _, l := range listeners { + l.Close() + } + backendClient.Close() + clus.Terminate(t) + } + return h +} + +// TestKVProxySerializableRangeDoesNotLeakAcrossAuth is a regression test for +// the proxy cache auth bypass: a serializable Range served from the proxy +// cache must never be replayed to a caller the backend would deny. +func TestKVProxySerializableRangeDoesNotLeakAcrossAuth(t *testing.T) { + h := newKVAuthTestHarness(t) + defer h.cleanup() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + req := &pb.RangeRequest{Key: []byte("/proxy-cache/secret"), Serializable: true} + + // Baseline: low-priv and unauthenticated direct serializable reads must be denied. + _, err := h.lowKV.Range(ctx, req) + require.Error(t, err, "low-priv serializable read should be denied before cache priming") + _, err = h.anonKV.Range(ctx, req) + require.Error(t, err, "anonymous serializable read should be denied before cache priming") + + // Root primes the proxy cache through the proxy. + rootResp, err := h.rootKV.Range(ctx, req) + require.NoError(t, err) + require.Len(t, rootResp.Kvs, 1) + require.Equal(t, []byte("cached-secret"), rootResp.Kvs[0].Value) + + // The exact same serializable request from a different identity must NOT + // be served from root's cache entry — it has to hit the backend and be denied. + _, err = h.lowKV.Range(ctx, req) + require.Error(t, err, "low-priv serializable read must be denied even after cache priming") + _, err = h.anonKV.Range(ctx, req) + require.Error(t, err, "anonymous serializable read must be denied even after cache priming") + + // Root must still get a cache hit (proves the cache still works for the + // identity that primed it). + rootResp2, err := h.rootKV.Range(ctx, req) + require.NoError(t, err) + require.Len(t, rootResp2.Kvs, 1) + require.Equal(t, []byte("cached-secret"), rootResp2.Kvs[0].Value) +} + +// TestKVProxyLinearizableRangeDoesNotPopulateOtherIdentities verifies that a +// linearizable read (which the proxy also caches "as serializable") is stored +// under the caller's identity and cannot be replayed to anyone else. +func TestKVProxyLinearizableRangeDoesNotPopulateOtherIdentities(t *testing.T) { + h := newKVAuthTestHarness(t) + defer h.cleanup() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + linReq := &pb.RangeRequest{Key: []byte("/proxy-cache/secret")} + serReq := &pb.RangeRequest{Key: []byte("/proxy-cache/secret"), Serializable: true} + + // Root does a linearizable read; the proxy caches it "as serializable". + _, err := h.rootKV.Range(ctx, linReq) + require.NoError(t, err) + + // A subsequent serializable read from a different identity must be denied. + _, err = h.lowKV.Range(ctx, serReq) + require.Error(t, err, "serializable read after linearizable prime must be denied for low-priv user") + _, err = h.anonKV.Range(ctx, serReq) + require.Error(t, err, "serializable read after linearizable prime must be denied for anonymous user") +} + +// TestKVProxyCacheClearedOnAuthMutation verifies that when an auth-mutating +// RPC (RoleRevokePermission) flows through the proxy, the KV cache is +// flushed so a caller whose permissions were just revoked cannot keep +// reading previously cached data. This drives the mutation through the real +// AuthProxy (wired to the same cache) exactly like production grpc-proxy. +func TestKVProxyCacheClearedOnAuthMutation(t *testing.T) { + integration.BeforeTest(t) + + clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1}) + defer clus.Terminate(t) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + api := integration.ToGRPC(clus.Client(0)) + + // Set up root + low-priv user, grant read on the key, enable auth. + _, err := api.Auth.UserAdd(ctx, &pb.AuthUserAddRequest{Name: "root", Password: "rootpw"}) + require.NoError(t, err) + _, err = api.Auth.RoleAdd(ctx, &pb.AuthRoleAddRequest{Name: "root"}) + require.NoError(t, err) + _, err = api.Auth.UserGrantRole(ctx, &pb.AuthUserGrantRoleRequest{User: "root", Role: "root"}) + require.NoError(t, err) + _, err = api.Auth.RoleAdd(ctx, &pb.AuthRoleAddRequest{Name: "lowrole"}) + require.NoError(t, err) + _, err = api.Auth.UserAdd(ctx, &pb.AuthUserAddRequest{Name: "low", Password: "lowpw"}) + require.NoError(t, err) + _, err = api.Auth.UserGrantRole(ctx, &pb.AuthUserGrantRoleRequest{User: "low", Role: "lowrole"}) + require.NoError(t, err) + _, err = api.KV.Put(ctx, &pb.PutRequest{Key: []byte("/proxy-cache/secret"), Value: []byte("cached-secret")}) + require.NoError(t, err) + _, err = api.Auth.AuthEnable(ctx, &pb.AuthEnableRequest{}) + require.NoError(t, err) + + rootToken := mustToken(t, api.Auth, "root", "rootpw") + lowToken := mustToken(t, api.Auth, "low", "lowpw") + + rootCtx := metadata.NewOutgoingContext(ctx, metadata.Pairs(rpctypes.TokenFieldNameGRPC, rootToken)) + perm := &authpb.Permission{ + PermType: authpb.Permission_READ, + Key: []byte("/proxy-cache/secret"), + } + _, err = api.Auth.RoleGrantPermission(rootCtx, &pb.AuthRoleGrantPermissionRequest{Name: "lowrole", Perm: perm}) + require.NoError(t, err) + + // Build backend client that forwards tokens (production wiring). + backendCfg := clientv3.Config{ + Endpoints: []string{clus.Members[0].GRPCURL}, + DialTimeout: 5 * time.Second, + DialOptions: []grpc.DialOption{ + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithUnaryInterceptor(grpcproxy.AuthUnaryClientInterceptor), + grpc.WithStreamInterceptor(grpcproxy.AuthStreamClientInterceptor), + }, + } + backendClient, err := integration.NewClient(t, backendCfg) + require.NoError(t, err) + defer backendClient.Close() + + kvp, _ := grpcproxy.NewKvProxy(backendClient) + authp := grpcproxy.NewAuthProxy(backendClient, kvp.(*grpcproxy.KvProxy)) + + // Facade server: stamps a fixed token into incoming metadata, like real clients. + mkFacade := func(token string, kvs pb.KVServer, as pb.AuthServer) (pb.KVClient, pb.AuthClient, func()) { + srv := grpc.NewServer() + pb.RegisterKVServer(srv, &authInjectingKvServer{KVServer: kvs, token: token}) + pb.RegisterAuthServer(srv, as) + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + go srv.Serve(l) + conn, err := grpc.NewClient(l.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + return pb.NewKVClient(conn), pb.NewAuthClient(conn), func() { + conn.Close() + srv.Stop() + l.Close() + } + } + + rootKV, _, rootCleanup := mkFacade(rootToken, kvp, authp) + defer rootCleanup() + lowKV, _, lowCleanup := mkFacade(lowToken, kvp, authp) + defer lowCleanup() + + req := &pb.RangeRequest{Key: []byte("/proxy-cache/secret"), Serializable: true} + + // Both root and low can read; both populate their own cache entries. + _, err = rootKV.Range(ctx, req) + require.NoError(t, err) + lowResp, err := lowKV.Range(ctx, req) + require.NoError(t, err) + require.Len(t, lowResp.Kvs, 1) + + // Sanity: a second low-priv read hits the cache (still succeeds even if + // the backend would now deny — this is the stale-permission window). + _, err = lowKV.Range(ctx, req) + require.NoError(t, err) + + // Now revoke the permission THROUGH the auth proxy, as root. The proxy + // must flush the KV cache as part of the mutation. + _, err = authp.RoleRevokePermission( + metadata.NewIncomingContext(ctx, metadata.Pairs(rpctypes.TokenFieldNameGRPC, rootToken)), + &pb.AuthRoleRevokePermissionRequest{Role: "lowrole", Key: []byte("/proxy-cache/secret")}, + ) + require.NoError(t, err) + + // The next low-priv serializable read must NOT come from the stale cache + // entry — it has to reach the backend and be denied. + _, err = lowKV.Range(ctx, req) + require.Error(t, err, "low-priv read must be denied after permission revocation through the proxy") + + // Root is unaffected (different identity, fresh cache entry). + rootResp, err := rootKV.Range(ctx, req) + require.NoError(t, err) + require.Len(t, rootResp.Kvs, 1) +} + +// mustToken authenticates against the backend and returns the token string. +func mustToken(t *testing.T, authClient pb.AuthClient, name, password string) string { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + resp, err := authClient.Authenticate(ctx, &pb.AuthenticateRequest{Name: name, Password: password}) + require.NoError(t, err) + return resp.Token +} From 297ac7eb2303bfb9128f842d8b588b2896ea26eb Mon Sep 17 00:00:00 2001 From: goingforstudying-ctrl Date: Fri, 17 Jul 2026 12:19:25 -0400 Subject: [PATCH 2/6] grpcproxy: guard against overflow in cache key allocation Add an explicit bounds check before computing the capacity hint in keyFunc to silence CodeQL size-computation-for-allocation overflow warning. The sum len(b)+1+len(authKey) cannot overflow on supported platforms (proto.Marshal output is bounded by MaxRequestBytes), but the explicit guard documents the invariant and silences the analyzer. Signed-off-by: goingforstudying-ctrl --- server/proxy/grpcproxy/cache/store.go | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/server/proxy/grpcproxy/cache/store.go b/server/proxy/grpcproxy/cache/store.go index a405c0a5a3de..0f26d39c249e 100644 --- a/server/proxy/grpcproxy/cache/store.go +++ b/server/proxy/grpcproxy/cache/store.go @@ -18,6 +18,7 @@ package cache import ( "errors" + "math" "sync" "google.golang.org/protobuf/proto" @@ -62,6 +63,13 @@ func keyFunc(req *pb.RangeRequest, authKey string) string { if authKey == "" { return string(b) } + // Pre-size the buffer to avoid repeated allocation while appending. + // Guard against overflow: proto.Marshal output is bounded by + // MaxRequestBytes (typically 1.5 MiB) and authKey is a bounded user + // identifier, so the sum cannot overflow on any supported platform. + if len(authKey) > math.MaxInt-len(b)-1 { + panic("authKey too large") + } out := make([]byte, 0, len(b)+1+len(authKey)) out = append(out, b...) out = append(out, 0) From 361b2d1ff8b9af221b875b063b9e32fe5fc41c83 Mon Sep 17 00:00:00 2001 From: goingforstudying-ctrl Date: Fri, 17 Jul 2026 16:06:44 -0400 Subject: [PATCH 3/6] grpcproxy: unify cache key construction to avoid string(bytes) conversion CodeQL flags the early 'return string(b)' path as a potential allocation-size overflow because it cannot prove the size of the proto.Marshal output is bounded. Routing every call through the pre-sized buffer path removes the flagged conversion and makes the overflow guard cover the entire allocation, not just the authKey branch. Behavior is unchanged: keys for distinct (req, authKey) pairs remain distinct, and the empty-authKey case still produces a stable, deterministic key. Signed-off-by: goingforstudying-ctrl --- server/proxy/grpcproxy/cache/store.go | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/server/proxy/grpcproxy/cache/store.go b/server/proxy/grpcproxy/cache/store.go index 0f26d39c249e..f94f5691aa0a 100644 --- a/server/proxy/grpcproxy/cache/store.go +++ b/server/proxy/grpcproxy/cache/store.go @@ -60,15 +60,14 @@ func keyFunc(req *pb.RangeRequest, authKey string) string { if err != nil { panic(err) } - if authKey == "" { - return string(b) - } // Pre-size the buffer to avoid repeated allocation while appending. - // Guard against overflow: proto.Marshal output is bounded by - // MaxRequestBytes (typically 1.5 MiB) and authKey is a bounded user - // identifier, so the sum cannot overflow on any supported platform. - if len(authKey) > math.MaxInt-len(b)-1 { - panic("authKey too large") + // proto.Marshal output is bounded by MaxRequestBytes (typically + // 1.5 MiB) and authKey is a bounded user identifier, so the sum + // cannot overflow on any supported platform. The explicit guard + // documents that invariant and silences CodeQL's allocation-size + // overflow heuristic. + if len(b) > math.MaxInt-len(authKey)-1 { + panic("request too large") } out := make([]byte, 0, len(b)+1+len(authKey)) out = append(out, b...) From 2f3ab5f2576d49a92579d344f69f2979df3271a0 Mon Sep 17 00:00:00 2001 From: goingforstudying-ctrl Date: Sat, 18 Jul 2026 08:29:39 -0400 Subject: [PATCH 4/6] decouple auth proxy from concrete KvProxy type Pass the cache interface directly to NewAuthProxy instead of the whole *KvProxy. This drops the exported concrete-type coupling and puts the existing Cache() method to use. Also fix the stale comment in TestCacheAuthKeyIsolation: the final Get(req,"alice") hits because the earlier Add(req,resp,"alice") entry is still resident, not because "" collides with a named identity. And document that the proxy only flushes on auth mutations that pass through it, not direct backend changes. Signed-off-by: goingforstudying-ctrl --- server/etcdmain/grpc_proxy.go | 3 ++- server/proxy/grpcproxy/auth.go | 14 +++++++++----- server/proxy/grpcproxy/cache/store_test.go | 3 ++- server/proxy/grpcproxy/kv.go | 14 ++++++++++++++ tests/framework/integration/cluster_proxy.go | 3 ++- .../proxy/grpcproxy/kv_auth_cache_test.go | 3 ++- 6 files changed, 31 insertions(+), 9 deletions(-) diff --git a/server/etcdmain/grpc_proxy.go b/server/etcdmain/grpc_proxy.go index cdb2dcfabb68..31b4e5a2d41e 100644 --- a/server/etcdmain/grpc_proxy.go +++ b/server/etcdmain/grpc_proxy.go @@ -54,6 +54,7 @@ import ( "go.etcd.io/etcd/server/v3/etcdserver/api/v3election/v3electionpb" "go.etcd.io/etcd/server/v3/etcdserver/api/v3lock/v3lockpb" "go.etcd.io/etcd/server/v3/proxy/grpcproxy" + "go.etcd.io/etcd/server/v3/proxy/grpcproxy/cache" ) var ( @@ -498,7 +499,7 @@ func newGRPCProxyServer(lg *zap.Logger, client *clientv3.Client) *grpc.Server { leasep, _ := grpcproxy.NewLeaseProxy(client.Ctx(), client) mainp := grpcproxy.NewMaintenanceProxy(client) - authp := grpcproxy.NewAuthProxy(client, kvp.(*grpcproxy.KvProxy)) + authp := grpcproxy.NewAuthProxy(client, kvp.(interface{ Cache() cache.Cache }).Cache()) electionp := grpcproxy.NewElectionProxy(client) lockp := grpcproxy.NewLockProxy(client) diff --git a/server/proxy/grpcproxy/auth.go b/server/proxy/grpcproxy/auth.go index 778e5740b076..37f218a03e4e 100644 --- a/server/proxy/grpcproxy/auth.go +++ b/server/proxy/grpcproxy/auth.go @@ -30,12 +30,16 @@ type AuthProxy struct { } // NewAuthProxy returns an AuthServer that forwards auth RPCs to the backend. -// kvp may be nil; when non-nil, any auth-mutating RPC flushes the KV proxy -// cache so no caller can keep reading data cached under stale permissions. -func NewAuthProxy(c *clientv3.Client, kvp *kvProxy) pb.AuthServer { +// cache may be nil; when non-nil, any auth-mutating RPC flushes the cache so +// no caller can keep reading data cached under stale permissions. +// +// Note: the proxy only observes auth changes that pass through it. Permission +// changes applied directly against the backend will not flush the cache, +// leaving a same-token stale-read window until eviction. +func NewAuthProxy(c *clientv3.Client, cache cache.Cache) pb.AuthServer { ap := &AuthProxy{authClient: pb.NewAuthClient(c.ActiveConnection())} - if kvp != nil { - ap.cache = kvp.cache + if cache != nil { + ap.cache = cache } return ap } diff --git a/server/proxy/grpcproxy/cache/store_test.go b/server/proxy/grpcproxy/cache/store_test.go index debe2084af4a..5fd2a16c9ffb 100644 --- a/server/proxy/grpcproxy/cache/store_test.go +++ b/server/proxy/grpcproxy/cache/store_test.go @@ -60,7 +60,8 @@ func TestCacheAuthKeyIsolation(t *testing.T) { if _, err = c.Get(req, ""); err != nil { t.Fatalf("expected cache hit for empty identity, got %v", err) } - // "alice" and "" share the same key when authKey is empty (no-auth requests share one cache) + // "alice" hits because her earlier Add(req,resp,"alice") entry is still + // resident, not because "" collides with a named identity. if _, err = c.Get(req, "alice"); err != nil { t.Fatalf("expected cache hit for alice after empty-identity add (same key), got %v", err) } diff --git a/server/proxy/grpcproxy/kv.go b/server/proxy/grpcproxy/kv.go index 92f90d085f51..100725428aa5 100644 --- a/server/proxy/grpcproxy/kv.go +++ b/server/proxy/grpcproxy/kv.go @@ -51,6 +51,20 @@ func NewKvProxy(c *clientv3.Client) (pb.KVServer, <-chan struct{}) { return kv, donec } +// NewKvProxyWithCache returns a KV proxy backed by the given cache instance. +// Useful when the caller needs to share the cache with another proxy (e.g. +// the auth proxy for flush-on-mutation) without type-asserting the concrete +// implementation. +func NewKvProxyWithCache(c *clientv3.Client, cache cache.Cache) (pb.KVServer, <-chan struct{}) { + kv := &kvProxy{ + kv: c.KV, + cache: cache, + } + donec := make(chan struct{}) + close(donec) + return kv, donec +} + // Cache returns the underlying cache instance. Exposed for tests and for // wiring the auth proxy to the same cache. func (p *kvProxy) Cache() cache.Cache { diff --git a/tests/framework/integration/cluster_proxy.go b/tests/framework/integration/cluster_proxy.go index c96d1dd92ebe..64767ce24a32 100644 --- a/tests/framework/integration/cluster_proxy.go +++ b/tests/framework/integration/cluster_proxy.go @@ -24,6 +24,7 @@ import ( "go.etcd.io/etcd/client/v3/namespace" "go.etcd.io/etcd/server/v3/proxy/grpcproxy" "go.etcd.io/etcd/server/v3/proxy/grpcproxy/adapter" + "go.etcd.io/etcd/server/v3/proxy/grpcproxy/cache" ) const ThroughProxy = true @@ -69,7 +70,7 @@ func ToGRPC(c *clientv3.Client) GRPCAPI { lp, lpch := grpcproxy.NewLeaseProxy(ctx, c) mp := grpcproxy.NewMaintenanceProxy(c) clp, _ := grpcproxy.NewClusterProxy(lg, c, "", "") // without registering proxy URLs - authp := grpcproxy.NewAuthProxy(c, kvp.(*grpcproxy.KvProxy)) + authp := grpcproxy.NewAuthProxy(c, kvp.(interface{ Cache() cache.Cache }).Cache()) lockp := grpcproxy.NewLockProxy(c) electp := grpcproxy.NewElectionProxy(c) diff --git a/tests/integration/proxy/grpcproxy/kv_auth_cache_test.go b/tests/integration/proxy/grpcproxy/kv_auth_cache_test.go index 45632e2d3b1d..4f4995300f52 100644 --- a/tests/integration/proxy/grpcproxy/kv_auth_cache_test.go +++ b/tests/integration/proxy/grpcproxy/kv_auth_cache_test.go @@ -30,6 +30,7 @@ import ( "go.etcd.io/etcd/api/v3/v3rpc/rpctypes" clientv3 "go.etcd.io/etcd/client/v3" "go.etcd.io/etcd/server/v3/proxy/grpcproxy" + "go.etcd.io/etcd/server/v3/proxy/grpcproxy/cache" "go.etcd.io/etcd/tests/v3/framework/integration" ) @@ -308,7 +309,7 @@ func TestKVProxyCacheClearedOnAuthMutation(t *testing.T) { defer backendClient.Close() kvp, _ := grpcproxy.NewKvProxy(backendClient) - authp := grpcproxy.NewAuthProxy(backendClient, kvp.(*grpcproxy.KvProxy)) + authp := grpcproxy.NewAuthProxy(backendClient, kvp.(interface{ Cache() cache.Cache }).Cache()) // Facade server: stamps a fixed token into incoming metadata, like real clients. mkFacade := func(token string, kvs pb.KVServer, as pb.AuthServer) (pb.KVClient, pb.AuthClient, func()) { From cb665e81b551dcd2a54f24e4147dd44ad6bac425 Mon Sep 17 00:00:00 2001 From: goingforstudying-ctrl Date: Thu, 23 Jul 2026 20:09:21 -0400 Subject: [PATCH 5/6] grpcproxy: document cache flush scope for direct backend auth changes The proxy cache is only flushed for auth mutations that transit the AuthProxy. Permission changes applied directly against the backend leave a same-token stale-read window until eviction. Add a note on flushCacheOnMutation so the limitation is documented where the flush is triggered. Signed-off-by: goingforstudying-ctrl --- server/proxy/grpcproxy/auth.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/server/proxy/grpcproxy/auth.go b/server/proxy/grpcproxy/auth.go index 37f218a03e4e..a46056c9fd4d 100644 --- a/server/proxy/grpcproxy/auth.go +++ b/server/proxy/grpcproxy/auth.go @@ -56,6 +56,12 @@ func (ap *AuthProxy) AuthDisable(ctx context.Context, r *pb.AuthDisableRequest) return resp, err } +// flushCacheOnMutation drops every cached entry after a successful auth +// mutation so no principal can keep reading data cached under stale +// permissions. Note that the proxy only observes auth changes that pass +// through it: permission changes applied directly against the backend +// bypass this hook and leave a stale-read window for the same token until +// the affected entries are evicted. func (ap *AuthProxy) flushCacheOnMutation(err error) { if err == nil && ap.cache != nil { ap.cache.Clear() From e65ca8620beda1734dc5ad8329c7ae7a6b379176 Mon Sep 17 00:00:00 2001 From: goingforstudying-ctrl Date: Sat, 25 Jul 2026 04:15:25 -0400 Subject: [PATCH 6/6] grpcproxy: cap cache entry lifetime to prevent stale reads after token expiry The KV proxy cache entries were keyed by (request, authKey) where authKey is the SHA256 of the auth token. However, entries had no time-based expiry, so when a token expired on the backend, the proxy would continue serving cached responses for that token until LRU eviction. This creates an auth-bypass window: an expired token could still read data that was cached while the token was valid. The fix adds an optional TTL to cache entries: - cache.NewCacheWithTTL(maxEntries, ttl) creates a cache where entries expire after the given duration - cacheEntry wraps the response with an expiresAt timestamp - Get() checks expiry and treats expired entries as misses - NewKvProxyWithTTL(client, ttl) wires the TTL through to the cache - New --cache-ttl flag on grpc-proxy start sets the entry lifetime Operators should set --cache-ttl to match or slightly less than the backend --auth-token-ttl (default 300s) to close the stale-read window. Fixes the issue identified by Hakai-Shin in review. Signed-off-by: goingforstudying-ctrl --- server/etcdmain/grpc_proxy.go | 10 +++- server/proxy/grpcproxy/cache/store.go | 47 +++++++++++++++++-- server/proxy/grpcproxy/cache/store_test.go | 54 ++++++++++++++++++++++ server/proxy/grpcproxy/kv.go | 14 ++++++ 4 files changed, 119 insertions(+), 6 deletions(-) diff --git a/server/etcdmain/grpc_proxy.go b/server/etcdmain/grpc_proxy.go index 31b4e5a2d41e..b26cdcea3f18 100644 --- a/server/etcdmain/grpc_proxy.go +++ b/server/etcdmain/grpc_proxy.go @@ -105,6 +105,11 @@ var ( grpcProxyDebug bool + // grpcProxyCacheTTL is the maximum lifetime of a cache entry in the KV + // proxy cache. Should be set to match or slightly less than the backend + // auth token TTL to prevent serving stale reads after token expiry. + grpcProxyCacheTTL time.Duration + // GRPC keep alive related options. grpcKeepAliveMinTime time.Duration grpcKeepAliveTimeout time.Duration @@ -185,6 +190,9 @@ func newGRPCProxyStartCommand() *cobra.Command { cmd.Flags().Uint32Var(&maxConcurrentStreams, "max-concurrent-streams", math.MaxUint32, "Maximum concurrent streams that each client can open at a time.") + // cache flags + cmd.Flags().DurationVar(&grpcProxyCacheTTL, "cache-ttl", 0, "Maximum lifetime of a cache entry in the KV proxy cache. Should be set to match or slightly less than the backend auth token TTL (default 0, no time-based expiry).") + return &cmd } @@ -490,7 +498,7 @@ func newGRPCProxyServer(lg *zap.Logger, client *clientv3.Client) *grpc.Server { client.KV, _, _ = leasing.NewKV(client, grpcProxyLeasing) } - kvp, _ := grpcproxy.NewKvProxy(client) + kvp, _ := grpcproxy.NewKvProxyWithTTL(client, grpcProxyCacheTTL) watchp, _ := grpcproxy.NewWatchProxy(client.Ctx(), lg, client) if grpcProxyResolverPrefix != "" { grpcproxy.Register(lg, client, grpcProxyResolverPrefix, grpcProxyAdvertiseClientURL, grpcProxyResolverTTL) diff --git a/server/proxy/grpcproxy/cache/store.go b/server/proxy/grpcproxy/cache/store.go index f94f5691aa0a..5818a88b3952 100644 --- a/server/proxy/grpcproxy/cache/store.go +++ b/server/proxy/grpcproxy/cache/store.go @@ -20,6 +20,7 @@ import ( "errors" "math" "sync" + "time" "google.golang.org/protobuf/proto" "k8s.io/utils/lru" @@ -34,12 +35,22 @@ var ( ErrCompacted = rpctypes.ErrGRPCCompacted ) +// cacheEntry wraps a cached response with its expiry time. Entries expire +// after a configurable TTL to prevent stale reads when auth tokens expire +// on the backend but the proxy cache still holds the response. +type cacheEntry struct { + resp *pb.RangeResponse + expiresAt time.Time +} + type Cache interface { // Add stores resp for (req, authKey). authKey scopes the entry to a // single caller identity so a cached response can never be replayed to // a different principal; pass "" when the request carries no credentials. + // The entry expires after the TTL configured on the cache. Add(req *pb.RangeRequest, resp *pb.RangeResponse, authKey string) - // Get returns the cached response for (req, authKey). + // Get returns the cached response for (req, authKey). Expired entries + // are treated as misses. Get(req *pb.RangeRequest, authKey string) (*pb.RangeResponse, error) // Clear drops every cached entry. Used when the authorization // configuration may have changed (user/role/permission updates) so no @@ -76,12 +87,23 @@ func keyFunc(req *pb.RangeRequest, authKey string) string { return string(out) } +// NewCache creates a cache with no entry TTL (entries never expire based on +// time). Use NewCacheWithTTL to set a maximum entry lifetime. func NewCache(maxCacheEntries int) Cache { + return NewCacheWithTTL(maxCacheEntries, 0) +} + +// NewCacheWithTTL creates a cache where entries expire after ttl. A ttl of 0 +// means entries never expire based on time (only via LRU eviction, Clear, or +// Invalidate). The TTL should be set to match or slightly less than the +// backend auth token TTL to prevent serving stale reads after token expiry. +func NewCacheWithTTL(maxCacheEntries int, ttl time.Duration) Cache { return &cache{ lru: lru.New(maxCacheEntries), maxEntries: maxCacheEntries, cachedRanges: adt.NewIntervalTree(), compactedRev: -1, + entryTTL: ttl, } } @@ -97,6 +119,10 @@ type cache struct { cachedRanges adt.IntervalTree compactedRev int64 + + // entryTTL is the maximum lifetime of a cache entry. 0 means no + // time-based expiry. + entryTTL time.Duration } // Add adds the response of a request to the cache if its revision is larger than the compacted revision of the cache. @@ -107,7 +133,11 @@ func (c *cache) Add(req *pb.RangeRequest, resp *pb.RangeResponse, authKey string defer c.mu.Unlock() if req.Revision > c.compactedRev { - c.lru.Add(key, resp) + entry := &cacheEntry{resp: resp} + if c.entryTTL > 0 { + entry.expiresAt = time.Now().Add(c.entryTTL) + } + c.lru.Add(key, entry) } // we do not need to invalidate a request with a revision specified. // so we do not need to add it into the reverse index. @@ -138,7 +168,8 @@ func (c *cache) Add(req *pb.RangeRequest, resp *pb.RangeResponse, authKey string } // Get looks up the caching response for a given request. -// Get is also responsible for lazy eviction when accessing compacted entries. +// Get is also responsible for lazy eviction when accessing compacted entries +// and for dropping expired entries when a TTL is configured. func (c *cache) Get(req *pb.RangeRequest, authKey string) (*pb.RangeResponse, error) { key := keyFunc(req, authKey) @@ -150,8 +181,14 @@ func (c *cache) Get(req *pb.RangeRequest, authKey string) (*pb.RangeResponse, er return nil, ErrCompacted } - if resp, ok := c.lru.Get(key); ok { - return resp.(*pb.RangeResponse), nil + if item, ok := c.lru.Get(key); ok { + entry := item.(*cacheEntry) + // Check if the entry has expired (when TTL is configured) + if !entry.expiresAt.IsZero() && time.Now().After(entry.expiresAt) { + c.lru.Remove(key) + return nil, errors.New("entry expired") + } + return entry.resp, nil } return nil, errors.New("not exist") } diff --git a/server/proxy/grpcproxy/cache/store_test.go b/server/proxy/grpcproxy/cache/store_test.go index 5fd2a16c9ffb..b3ce55f4f4a1 100644 --- a/server/proxy/grpcproxy/cache/store_test.go +++ b/server/proxy/grpcproxy/cache/store_test.go @@ -16,6 +16,7 @@ package cache import ( "testing" + "time" pb "go.etcd.io/etcd/api/v3/etcdserverpb" "go.etcd.io/etcd/api/v3/mvccpb" @@ -117,3 +118,56 @@ func TestKeyFuncDeterministic(t *testing.T) { t.Fatal("identity vs no-identity must produce different keys") } } + +func TestCacheEntryExpiry(t *testing.T) { + // Create a cache with a very short TTL for testing + c := NewCacheWithTTL(10, 50*time.Millisecond) + defer c.Close() + + req := &pb.RangeRequest{Key: []byte("k"), Serializable: true} + resp := &pb.RangeResponse{Header: &pb.ResponseHeader{Revision: 1}} + + c.Add(req, resp, "alice") + + // Entry should be available immediately + if _, err := c.Get(req, "alice"); err != nil { + t.Fatalf("expected cache hit before expiry, got %v", err) + } + + // Wait for entry to expire + time.Sleep(60 * time.Millisecond) + + // Entry should now be expired + _, err := c.Get(req, "alice") + if err == nil { + t.Fatal("expected cache miss after expiry") + } + if err.Error() != "entry expired" { + t.Fatalf("expected 'entry expired' error, got %v", err) + } + + // Cache should still accept new entries + c.Add(req, resp, "alice") + if _, err := c.Get(req, "alice"); err != nil { + t.Fatalf("expected cache hit after re-Add, got %v", err) + } +} + +func TestCacheNoExpiryWhenTTLZero(t *testing.T) { + // Create a cache with no TTL (0 means no expiry) + c := NewCacheWithTTL(10, 0) + defer c.Close() + + req := &pb.RangeRequest{Key: []byte("k"), Serializable: true} + resp := &pb.RangeResponse{Header: &pb.ResponseHeader{Revision: 1}} + + c.Add(req, resp, "alice") + + // Wait a bit + time.Sleep(60 * time.Millisecond) + + // Entry should still be available (no expiry) + if _, err := c.Get(req, "alice"); err != nil { + t.Fatalf("expected cache hit with no TTL, got %v", err) + } +} diff --git a/server/proxy/grpcproxy/kv.go b/server/proxy/grpcproxy/kv.go index 100725428aa5..843f99958dde 100644 --- a/server/proxy/grpcproxy/kv.go +++ b/server/proxy/grpcproxy/kv.go @@ -19,6 +19,7 @@ import ( "crypto/sha256" "encoding/hex" "errors" + "time" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -51,6 +52,19 @@ func NewKvProxy(c *clientv3.Client) (pb.KVServer, <-chan struct{}) { return kv, donec } +// NewKvProxyWithTTL returns a KV proxy with a cache that expires entries +// after the given TTL. The TTL should match or be slightly less than the +// backend auth token TTL to prevent serving stale reads after token expiry. +func NewKvProxyWithTTL(c *clientv3.Client, ttl time.Duration) (pb.KVServer, <-chan struct{}) { + kv := &kvProxy{ + kv: c.KV, + cache: cache.NewCacheWithTTL(cache.DefaultMaxEntries, ttl), + } + donec := make(chan struct{}) + close(donec) + return kv, donec +} + // NewKvProxyWithCache returns a KV proxy backed by the given cache instance. // Useful when the caller needs to share the cache with another proxy (e.g. // the auth proxy for flush-on-mutation) without type-asserting the concrete