validate only sa

Signed-off-by: Karol Szwaj <[email protected]>

Author: cnvergence

internal/xds/cache/snapshotcache.go

@@ -47,7 +47,6 @@ type SnapshotCacheWithCallbacks interface {
 	cachev3.SnapshotCache
 	serverv3.Callbacks
 	GenerateNewSnapshot(string, types.XdsResources) error
-	SnapshotHasIrKey(string) bool
 	GetIrKeys() []string
 }
 
@@ -366,19 +365,6 @@ 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,11 +6,9 @@
 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"
@@ -36,56 +34,33 @@ func NewJWTAuthInterceptor(clientset *kubernetes.Clientset, issuer, audience str
 	}
 }
 
-type wrappedStream struct {
-	grpc.ServerStream
-	ctx         context.Context
-	interceptor *JWTAuthInterceptor
-	validated   bool
+// 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)
+	}
 }
 
-func (w *wrappedStream) RecvMsg(m any) error {
-	err := w.ServerStream.RecvMsg(m)
-	if err != nil {
-		return err
+// 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")
 	}
 
-	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
-
-			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)
-			}
+	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 ")
 
-			w.validated = true
-		}
+	err := i.validateKubeJWT(ss.Context(), tokenStr)
+	if err != nil {
+		return fmt.Errorf("failed to validate token: %w", err)
 	}
 
 	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

@@ -11,13 +11,12 @@ import (
 	"slices"
 	"strings"
 
-	"github.com/envoyproxy/gateway/internal/envoygateway/config"
-	"github.com/envoyproxy/gateway/internal/utils"
 	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"
+
+	"github.com/envoyproxy/gateway/internal/infrastructure/kubernetes/proxy"
 )
 
 // GetKubernetesClient creates a Kubernetes client using in-cluster configuration.
@@ -35,7 +34,7 @@ func GetKubernetesClient() (*kubernetes.Clientset, error) {
 	return clientset, nil
 }
 
-func (i *JWTAuthInterceptor) validateKubeJWT(ctx context.Context, token, nodeID string) error {
+func (i *JWTAuthInterceptor) validateKubeJWT(ctx context.Context, token string) error {
 	tokenReview := &authenticationv1.TokenReview{
 		Spec: authenticationv1.TokenReviewSpec{
 			Token:     token,
@@ -56,36 +55,21 @@ func (i *JWTAuthInterceptor) validateKubeJWT(ctx context.Context, token, nodeID
 		return fmt.Errorf("token is not authenticated")
 	}
 
-	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 to verify that the token
 	// is valid for an Envoy proxy managed by Envoy Gateway.
 	// example: "system:serviceaccount:default:envoy-default-eg-e41e7b31"
-	parts:=strings.Split(tokenReview.Status.User.Username, ":")
+	parts := strings.Split(tokenReview.Status.User.Username, ":")
 	if len(parts) != 4 {
 		return fmt.Errorf("invalid username format: %s", tokenReview.Status.User.Username)
 	}
 	sa := parts[3]
 
-	irKeys:=i.cache.GetIrKeys()
+	irKeys := i.cache.GetIrKeys()
 	for _, irKey := range irKeys {
-		if irKey2ServiceAccountName(irKey) == sa {
+		if proxy.ExpectedResourceHashedName(irKey) == sa {
 			return nil
 		}
 	}
-	return fmt.Errorf("Envoy service account %s not found in the cache", sa)
-}
 
-func irKey2ServiceAccountName(irKey string) string {
-	hashedName := utils.GetHashedName(irKey, 48)
-	return fmt.Sprintf("%s-%s", config.EnvoyPrefix, hashedName)
+	return fmt.Errorf("envoy service account %s not found in the cache", sa)
 }