Split into Provider and Verifier

cmd/broker/filter/main.go

@@ -120,10 +120,10 @@ func main() {
 
 	reporter := filter.NewStatsReporter(env.ContainerName, kmeta.ChildName(env.PodName, uuid.New().String()))
 
-	oidcTokenHandler := auth.NewOIDCTokenHandler(ctx)
+	oidcTokenProvider := auth.NewOIDCTokenProvider(ctx)
 	// We are running both the receiver (takes messages in from the Broker) and the dispatcher (send
 	// the messages to the triggers' subscribers) in this binary.
-	handler, err := filter.NewHandler(logger, oidcTokenHandler, triggerinformer.Get(ctx), reporter, ctxFunc)
+	handler, err := filter.NewHandler(logger, oidcTokenProvider, triggerinformer.Get(ctx), reporter, ctxFunc)
 	if err != nil {
 		logger.Fatal("Error creating Handler", zap.Error(err))
 	}

cmd/broker/ingress/main.go

@@ -152,9 +152,9 @@ func main() {
 
 	reporter := ingress.NewStatsReporter(env.ContainerName, kmeta.ChildName(env.PodName, uuid.New().String()))
 
-	oidcTokenHandler := auth.NewOIDCTokenHandler(ctx)
+	oidcTokenProvider := auth.NewOIDCTokenProvider(ctx)
 
-	handler, err = ingress.NewHandler(logger, reporter, broker.TTLDefaulter(logger, int32(env.MaxTTL)), brokerInformer, oidcTokenHandler)
+	handler, err = ingress.NewHandler(logger, reporter, broker.TTLDefaulter(logger, int32(env.MaxTTL)), brokerInformer, oidcTokenProvider)
 	if err != nil {
 		logger.Fatal("Error creating Handler", zap.Error(err))
 	}

pkg/auth/token_provider.go

@@ -0,0 +1,67 @@
+/*
+Copyright 2023 The Knative 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 auth
+
+import (
+	"context"
+	"fmt"
+
+	"go.uber.org/zap"
+	authv1 "k8s.io/api/authentication/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/logging"
+)
+
+type OIDCTokenProvider struct {
+	logger     *zap.SugaredLogger
+	kubeClient kubernetes.Interface
+}
+
+func NewOIDCTokenProvider(ctx context.Context) *OIDCTokenProvider {
+	tokenProvider := &OIDCTokenProvider{
+		logger:     logging.FromContext(ctx).With("component", "oidc-token-provider"),
+		kubeClient: kubeclient.Get(ctx),
+	}
+
+	return tokenProvider
+}
+
+// GetJWT returns a JWT from the given service account for the given audience.
+func (c *OIDCTokenProvider) GetJWT(serviceAccount types.NamespacedName, audience string) (string, error) {
+	// TODO: check the cache
+
+	// if not found in cache: request new token
+	tokenRequest := authv1.TokenRequest{
+		Spec: authv1.TokenRequestSpec{
+			Audiences: []string{audience},
+		},
+	}
+
+	tokenRequestResponse, err := c.kubeClient.
+		CoreV1().
+		ServiceAccounts(serviceAccount.Namespace).
+		CreateToken(context.TODO(), serviceAccount.Name, &tokenRequest, metav1.CreateOptions{})
+
+	if err != nil {
+		return "", fmt.Errorf("could not request a token for %s: %w", serviceAccount, err)
+	}
+
+	return tokenRequestResponse.Status.Token, nil
+}

pkg/auth/token_handler.go → pkg/auth/token_verifier.go

@@ -26,13 +26,8 @@ import (
 
 	"github.com/coreos/go-oidc/v3/oidc"
 	"go.uber.org/zap"
-	authv1 "k8s.io/api/authentication/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"knative.dev/eventing/pkg/apis/feature"
-	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/logging"
 )
@@ -41,29 +36,61 @@ const (
 	kubernetesOIDCDiscoveryBaseURL = "https://kubernetes.default.svc"
 )
 
-type OIDCTokenHandler struct {
+type OIDCTokenVerifier struct {
 	logger     *zap.SugaredLogger
-	kubeClient kubernetes.Interface
 	restConfig *rest.Config
 	provider   *oidc.Provider
 }
 
+type IDToken struct {
+	Issuer          string
+	Audience        []string
+	Subject         string
+	Expiry          time.Time
+	IssuedAt        time.Time
+	AccessTokenHash string
+}
+
 // TODO: rename OIDCTokenProvider and OIDCTokenVerifier
-func NewOIDCTokenHandler(ctx context.Context) *OIDCTokenHandler {
-	tokenHandler := &OIDCTokenHandler{
+func NewOIDCTokenVerifier(ctx context.Context) *OIDCTokenVerifier {
+	tokenHandler := &OIDCTokenVerifier{
 		logger:     logging.FromContext(ctx).With("component", "oidc-token-handler"),
-		kubeClient: kubeclient.Get(ctx),
 		restConfig: injection.GetConfig(ctx),
 	}
 
-	if err := tokenHandler.initProvider(ctx); err != nil {
+	if err := tokenHandler.initOIDCProvider(ctx); err != nil {
 		tokenHandler.logger.Error(fmt.Sprintf("could not initialize provider. You can ignore this message, when the %s feature is disabled", feature.OIDCAuthentication), zap.Error(err))
 	}
 
 	return tokenHandler
 }
 
-func (c *OIDCTokenHandler) initProvider(ctx context.Context) error {
+// VerifyJWT verifies the given JWT for the expected audience and returns the parsed ID token.
+func (c *OIDCTokenVerifier) VerifyJWT(ctx context.Context, jwt, audience string) (*IDToken, error) {
+	if c.provider == nil {
+		return nil, fmt.Errorf("provider is nil. Is the OIDC provider config correct?")
+	}
+
+	verifier := c.provider.Verifier(&oidc.Config{
+		ClientID: audience,
+	})
+
+	token, err := verifier.Verify(ctx, jwt)
+	if err != nil {
+		return nil, fmt.Errorf("could not verify JWT: %w", err)
+	}
+
+	return &IDToken{
+		Issuer:          token.Issuer,
+		Audience:        token.Audience,
+		Subject:         token.Subject,
+		Expiry:          token.Expiry,
+		IssuedAt:        token.IssuedAt,
+		AccessTokenHash: token.AccessTokenHash,
+	}, nil
+}
+
+func (c *OIDCTokenVerifier) initOIDCProvider(ctx context.Context) error {
 	discovery, err := c.getKubernetesOIDCDiscovery()
 	if err != nil {
 		return fmt.Errorf("could not load Kubernetes OIDC discovery information: %w", err)
@@ -91,7 +118,7 @@ func (c *OIDCTokenHandler) initProvider(ctx context.Context) error {
 	return nil
 }
 
-func (c *OIDCTokenHandler) getHTTPClientForKubeAPIServer() (*http.Client, error) {
+func (c *OIDCTokenVerifier) getHTTPClientForKubeAPIServer() (*http.Client, error) {
 	client, err := rest.HTTPClientFor(c.restConfig)
 	if err != nil {
 		return nil, fmt.Errorf("could not create HTTP client from rest config: %w", err)
@@ -100,7 +127,7 @@ func (c *OIDCTokenHandler) getHTTPClientForKubeAPIServer() (*http.Client, error)
 	return client, nil
 }
 
-func (c *OIDCTokenHandler) getKubernetesOIDCDiscovery() (*openIDMetadata, error) {
+func (c *OIDCTokenVerifier) getKubernetesOIDCDiscovery() (*openIDMetadata, error) {
 	client, err := c.getHTTPClientForKubeAPIServer()
 	if err != nil {
 		return nil, fmt.Errorf("could not get HTTP client for API server: %w", err)
@@ -125,63 +152,6 @@ func (c *OIDCTokenHandler) getKubernetesOIDCDiscovery() (*openIDMetadata, error)
 	return openIdConfig, nil
 }
 
-// GetJWT returns a JWT from the given service account for the given audience.
-func (c *OIDCTokenHandler) GetJWT(serviceAccount types.NamespacedName, audience string) (string, error) {
-	// TODO: check the cache
-
-	// if not found in cache: request new token
-	tokenRequest := authv1.TokenRequest{
-		Spec: authv1.TokenRequestSpec{
-			Audiences: []string{audience},
-		},
-	}
-
-	tokenRequestResponse, err := c.kubeClient.
-		CoreV1().
-		ServiceAccounts(serviceAccount.Namespace).
-		CreateToken(context.TODO(), serviceAccount.Name, &tokenRequest, metav1.CreateOptions{})
-
-	if err != nil {
-		return "", fmt.Errorf("could not request a token for %s: %w", serviceAccount, err)
-	}
-
-	return tokenRequestResponse.Status.Token, nil
-}
-
-// VerifyJWT verifies the given JWT for the expected audience and returns the parsed ID token.
-func (c *OIDCTokenHandler) VerifyJWT(ctx context.Context, jwt, audience string) (*IDToken, error) {
-	if c.provider == nil {
-		return nil, fmt.Errorf("provider is nil. Is the OIDC provider config correct?")
-	}
-
-	verifier := c.provider.Verifier(&oidc.Config{
-		ClientID: audience,
-	})
-
-	token, err := verifier.Verify(ctx, jwt)
-	if err != nil {
-		return nil, fmt.Errorf("could not verify JWT: %w", err)
-	}
-
-	return &IDToken{
-		Issuer:          token.Issuer,
-		Audience:        token.Audience,
-		Subject:         token.Subject,
-		Expiry:          token.Expiry,
-		IssuedAt:        token.IssuedAt,
-		AccessTokenHash: token.AccessTokenHash,
-	}, nil
-}
-
-type IDToken struct {
-	Issuer          string
-	Audience        []string
-	Subject         string
-	Expiry          time.Time
-	IssuedAt        time.Time
-	AccessTokenHash string
-}
-
 type openIDMetadata struct {
 	Issuer        string   `json:"issuer"`
 	JWKSURI       string   `json:"jwks_uri"`

pkg/broker/filter/filter_handler.go

@@ -79,7 +79,7 @@ type Handler struct {
 }
 
 // NewHandler creates a new Handler and its associated EventReceiver.
-func NewHandler(logger *zap.Logger, oidcTokenHandler *auth.OIDCTokenHandler, triggerInformer v1.TriggerInformer, reporter StatsReporter, wc func(ctx context.Context) context.Context) (*Handler, error) {
+func NewHandler(logger *zap.Logger, oidcTokenProvider *auth.OIDCTokenProvider, triggerInformer v1.TriggerInformer, reporter StatsReporter, wc func(ctx context.Context) context.Context) (*Handler, error) {
 	kncloudevents.ConfigureConnectionArgs(&kncloudevents.ConnectionArgs{
 		MaxIdleConns:        defaultMaxIdleConnections,
 		MaxIdleConnsPerHost: defaultMaxIdleConnectionsPerHost,
@@ -128,7 +128,7 @@ func NewHandler(logger *zap.Logger, oidcTokenHandler *auth.OIDCTokenHandler, tri
 
 	return &Handler{
 		reporter:        reporter,
-		eventDispatcher: kncloudevents.NewDispatcher(oidcTokenHandler),
+		eventDispatcher: kncloudevents.NewDispatcher(oidcTokenProvider),
 		triggerLister:   triggerInformer.Lister(),
 		logger:          logger,
 		withContext:     wc,

pkg/broker/filter/filter_handler_test.go

@@ -430,7 +430,7 @@ func TestReceiver(t *testing.T) {
 			defer s.Close()
 
 			logger := zaptest.NewLogger(t, zaptest.WrapOptions(zap.AddCaller()))
-			oidcTokenHandler := auth.NewOIDCTokenHandler(ctx)
+			oidcTokenProvider := auth.NewOIDCTokenProvider(ctx)
 
 			// Replace the SubscriberURI to point at our fake server.
 			for _, trig := range tc.triggers {
@@ -447,7 +447,7 @@ func TestReceiver(t *testing.T) {
 			reporter := &mockReporter{}
 			r, err := NewHandler(
 				logger,
-				oidcTokenHandler,
+				oidcTokenProvider,
 				triggerinformerfake.Get(ctx),
 				reporter,
 				func(ctx context.Context) context.Context {
@@ -615,7 +615,7 @@ func TestReceiver_WithSubscriptionsAPI(t *testing.T) {
 			filtersMap := subscriptionsapi.NewFiltersMap()
 
 			logger := zaptest.NewLogger(t, zaptest.WrapOptions(zap.AddCaller()))
-			oidcTokenHandler := auth.NewOIDCTokenHandler(ctx)
+			oidcTokenProvider := auth.NewOIDCTokenProvider(ctx)
 
 			// Replace the SubscriberURI to point at our fake server.
 			for _, trig := range tc.triggers {
@@ -633,7 +633,7 @@ func TestReceiver_WithSubscriptionsAPI(t *testing.T) {
 			reporter := &mockReporter{}
 			r, err := NewHandler(
 				logger,
-				oidcTokenHandler,
+				oidcTokenProvider,
 				triggerinformerfake.Get(ctx),
 				reporter,
 				func(ctx context.Context) context.Context {

pkg/broker/ingress/ingress_handler.go

@@ -66,12 +66,12 @@ type Handler struct {
 
 	Logger *zap.Logger
 
-	oidcTokenHandler *auth.OIDCTokenHandler
+	oidcTokenProvider *auth.OIDCTokenProvider
 
 	eventDispatcher *kncloudevents.Dispatcher
 }
 
-func NewHandler(logger *zap.Logger, reporter StatsReporter, defaulter client.EventDefaulter, brokerInformer v1.BrokerInformer, oidcTokenHandler *auth.OIDCTokenHandler) (*Handler, error) {
+func NewHandler(logger *zap.Logger, reporter StatsReporter, defaulter client.EventDefaulter, brokerInformer v1.BrokerInformer, oidcTokenProvider *auth.OIDCTokenProvider) (*Handler, error) {
 	connectionArgs := kncloudevents.ConnectionArgs{
 		MaxIdleConns:        defaultMaxIdleConnections,
 		MaxIdleConnsPerHost: defaultMaxIdleConnectionsPerHost,
@@ -112,12 +112,12 @@ func NewHandler(logger *zap.Logger, reporter StatsReporter, defaulter client.Eve
 	})
 
 	return &Handler{
-		Defaulter:        defaulter,
-		Reporter:         reporter,
-		Logger:           logger,
-		BrokerLister:     brokerInformer.Lister(),
-		oidcTokenHandler: oidcTokenHandler,
-		eventDispatcher:  kncloudevents.NewDispatcher(oidcTokenHandler),
+		Defaulter:         defaulter,
+		Reporter:          reporter,
+		Logger:            logger,
+		BrokerLister:      brokerInformer.Lister(),
+		oidcTokenProvider: oidcTokenProvider,
+		eventDispatcher:   kncloudevents.NewDispatcher(oidcTokenProvider),
 	}, nil
 }
 

pkg/broker/ingress/ingress_handler_test.go

@@ -285,9 +285,9 @@ func TestHandler_ServeHTTP(t *testing.T) {
 				brokerinformerfake.Get(ctx).Informer().GetStore().Add(b)
 			}
 
-			oidcTokenHandler := auth.NewOIDCTokenHandler(ctx)
+			oidcTokenProvider := auth.NewOIDCTokenProvider(ctx)
 
-			h, err := NewHandler(logger, &mockReporter{}, tc.defaulter, brokerinformerfake.Get(ctx), oidcTokenHandler)
+			h, err := NewHandler(logger, &mockReporter{}, tc.defaulter, brokerinformerfake.Get(ctx), oidcTokenProvider)
 			if err != nil {
 				t.Fatal("Unable to create receiver:", err)
 			}

pkg/channel/fanout/fanout_event_handler_test.go

@@ -371,8 +371,8 @@ func testFanoutEventHandler(t *testing.T, async bool, receiverFunc channel.Event
 		t.Fatal(err)
 	}
 
-	oidcTokenHandler := auth.NewOIDCTokenHandler(ctx)
-	dispatcher := kncloudevents.NewDispatcher(oidcTokenHandler)
+	oidcTokenProvider := auth.NewOIDCTokenProvider(ctx)
+	dispatcher := kncloudevents.NewDispatcher(oidcTokenProvider)
 
 	calledChan := make(chan bool, 1)
 	recvOptionFunc := func(*channel.EventReceiver) error {