Verify pod name for authz

This reverts commit b0748a0.

Author: zhaohuabing

internal/xds/cache/snapshotcache.go

@@ -47,6 +47,7 @@ type SnapshotCacheWithCallbacks interface {
 	cachev3.SnapshotCache
 	serverv3.Callbacks
 	GenerateNewSnapshot(string, types.XdsResources) error
+	SnapshotHasIrKey(string) bool
 	GetIrKeys() []string
 }
 
@@ -365,6 +366,19 @@ func (s *snapshotCache) OnFetchRequest(_ context.Context, _ *discoveryv3.Discove
 func (s *snapshotCache) OnFetchResponse(_ *discoveryv3.DiscoveryRequest, _ *discoveryv3.DiscoveryResponse) {
 }
 
+func (s *snapshotCache) SnapshotHasIrKey(irKey string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	for key, snapshot := range s.lastSnapshot {
+		if snapshot != nil && key == irKey {
+			return true
+		}
+	}
+
+	return false
+}
+
 func (s *snapshotCache) GetIrKeys() []string {
 	s.mu.Lock()
 	defer s.mu.Unlock()

internal/xds/server/kubejwt/jwtinterceptor.go

@@ -6,9 +6,11 @@
 package kubejwt
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
+	discoveryv3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
 	"k8s.io/client-go/kubernetes"
@@ -34,33 +36,56 @@ func NewJWTAuthInterceptor(clientset *kubernetes.Clientset, issuer, audience str
 	}
 }
 
-// Stream intercepts streaming gRPC calls for authentication.
-func (i *JWTAuthInterceptor) Stream() grpc.StreamServerInterceptor {
-	return func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
-		if err := i.authorize(ss); err != nil {
-			return err
-		}
-		return handler(srv, ss)
-	}
+type wrappedStream struct {
+	grpc.ServerStream
+	ctx         context.Context
+	interceptor *JWTAuthInterceptor
+	validated   bool
 }
 
-// authorize validates the Kubernetes Service Account JWT token from the metadata.
-func (i *JWTAuthInterceptor) authorize(ss grpc.ServerStream) error {
-	md, ok := metadata.FromIncomingContext(ss.Context())
-	if !ok {
-		return fmt.Errorf("missing metadata")
+func (w *wrappedStream) RecvMsg(m any) error {
+	err := w.ServerStream.RecvMsg(m)
+	if err != nil {
+		return err
 	}
 
-	authHeader, exists := md["authorization"]
-	if !exists || len(authHeader) == 0 {
-		return fmt.Errorf("missing authorization token in metadata: %s", md)
-	}
-	tokenStr := strings.TrimPrefix(authHeader[0], "Bearer ")
+	if !w.validated {
+		if req, ok := m.(*discoveryv3.DeltaDiscoveryRequest); ok {
+			if req.Node == nil || req.Node.Id == "" {
+				return fmt.Errorf("missing node ID in request")
+			}
+			nodeID := req.Node.Id
 
-	err := i.validateKubeJWT(ss.Context(), tokenStr)
-	if err != nil {
-		return fmt.Errorf("failed to validate token: %w", err)
+			md, ok := metadata.FromIncomingContext(w.ctx)
+			if !ok {
+				return fmt.Errorf("missing metadata")
+			}
+
+			authHeader := md.Get("authorization")
+			if len(authHeader) == 0 {
+				return fmt.Errorf("missing authorization token in metadata: %s", md)
+			}
+			token := strings.TrimPrefix(authHeader[0], "Bearer ")
+
+			if err := w.interceptor.validateKubeJWT(w.ctx, token, nodeID); err != nil {
+				return fmt.Errorf("failed to validate token: %w", err)
+			}
+
+			w.validated = true
+		}
 	}
 
 	return nil
 }
+
+func newWrappedStream(s grpc.ServerStream, ctx context.Context, interceptor *JWTAuthInterceptor) grpc.ServerStream {
+	return &wrappedStream{s, ctx, interceptor, false}
+}
+
+// Stream intercepts streaming gRPC calls for authorization.
+func (i *JWTAuthInterceptor) Stream() grpc.StreamServerInterceptor {
+	return func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		wrapped := newWrappedStream(ss, ss.Context(), i)
+		return handler(srv, wrapped)
+	}
+}

internal/xds/server/kubejwt/tokenreview.go

@@ -13,6 +13,7 @@ import (
 
 	authenticationv1 "k8s.io/api/authentication/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apiserver/pkg/authentication/serviceaccount"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 
@@ -35,7 +36,7 @@ func GetKubernetesClient() (*kubernetes.Clientset, error) {
 	return clientset, nil
 }
 
-func (i *JWTAuthInterceptor) validateKubeJWT(ctx context.Context, token string) error {
+func (i *JWTAuthInterceptor) validateKubeJWT(ctx context.Context, token, nodeID string) error {
 	tokenReview := &authenticationv1.TokenReview{
 		Spec: authenticationv1.TokenReviewSpec{
 			Token:     token,
@@ -56,8 +57,21 @@ func (i *JWTAuthInterceptor) validateKubeJWT(ctx context.Context, token string)
 		return fmt.Errorf("token is not authenticated")
 	}
 
-	// Check if the service account name in the JWT token exists in the cache to verify that the token
-	// is valid for an Envoy proxy managed by Envoy Gateway.
+	// Check if the node ID in the request matches the pod name in the token review response.
+	// This is used to prevent a client from accessing the xDS resource of another one.
+	if tokenReview.Status.User.Extra != nil {
+		podName := tokenReview.Status.User.Extra[serviceaccount.PodNameKey]
+		if podName[0] == "" {
+			return fmt.Errorf("pod name not found in token review response")
+		}
+
+		if podName[0] != nodeID {
+			return fmt.Errorf("pod name mismatch: expected %s, got %s", nodeID, podName[0])
+		}
+	}
+
+	// Check if the service account name in the JWT token exists in the cache.
+	// This is used to verify that the token belongs to a valid Enovy managed by Envoy Gateway.
 	// example: "system:serviceaccount:default:envoy-default-eg-e41e7b31"
 	parts := strings.Split(tokenReview.Status.User.Username, ":")
 	if len(parts) != 4 {
@@ -71,8 +85,7 @@ func (i *JWTAuthInterceptor) validateKubeJWT(ctx context.Context, token string)
 			return nil
 		}
 	}
-
-	return fmt.Errorf("envoy service account %s not found in the cache", sa)
+	return fmt.Errorf("Envoy service account %s not found in the cache", sa)
 }
 
 // this is the same logic used in infra pkg func ExpectedResourceHashedName to generate the resource name.