Authenticate Requests from ApiServerSources

SECURITY.md

@@ -1,7 +1,9 @@
 # Knative Security Policy
 
-We're extremely grateful for security researchers and users that report vulnerabilities to the Knative Open Source Community. All reports are thoroughly investigated by a set of community volunteers.
+We're extremely grateful for security researchers and users that report vulnerabilities to the Knative Open Source
+Community. All reports are thoroughly investigated by a set of community volunteers.
 
-To make a report, please email the private security@knative.team list with the security details and the details expected for all Knative bug reports.
+To make a report, please email the private security@knative.team list with the security details and the details expected
+for all Knative bug reports.
 
 See [Knative Security and Disclosure Information](https://knative.dev/docs/reference/security/) for more details.

pkg/adapter/apiserver/delegate.go

@@ -1,4 +1,4 @@
 /*
 Copyright 2020 The Knative Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
@@ -18,7 +18,6 @@
 
 import (
 	"context"
-
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/google/uuid"
 	"go.uber.org/zap"
@@ -75,6 +74,32 @@
 	subject := event.Context.GetSubject()
 	a.logger.Debugf("sending cloudevent id: %s, source: %s, subject: %s", event.ID(), source, subject)
 
+	// Decide whether to request the JWT token or not
+	// Condition: if the sink has audience or not
+	// ?? Question: where can we get the sink audience? As we don't specify the destination in the cloudevent
+
+	// If the sink has audience, then we need to request the JWT token
+	// In order to request the JWT token, we need to get the service account name and namespace from the sink
+	// And also need to pass in OIDC token provider
+	// ?? Question again: where can we get the sink audience? And how to pass in OIDC token provider?
+
+	// If the sink doesn't have audience, then we don't need to request the JWT token
+
+	// If the sink has audience, and we have the JWT token, then we need to add the JWT token to the cloudevent
+	// Easy to do this, just add the JWT token as the bearer auth header to the cloudevent header
+
+	// Discovery:
+	// ReceiveAdapter -> ResourceDelegate -> MakeAddEvent -> MakeEvent -> MakeCloudEvent -> SendCloudEvent
+	// Receive adapter is the entry point of the adapter, it receives the k8s api event
+	// ResourceDelegate is the cache.Store, it receives the k8s api event from the receive adapter
+
+	//ApiServerSource will listen to the k8s api event, and then send the cloudevent to the sink when the k8s api event is created, updated or deleted
+
+	// Prepare the headers
+	//headers := http.HeaderFrom(ctx)
+	//jwt := auth.GetJWT(ctx)
+	//headers.Set("Authentication", fmt.Print("Bearer %s", jwt))
+
 	if result := a.ce.Send(ctx, event); !cloudevents.IsACK(result) {
 		a.logger.Errorw("failed to send cloudevent", zap.Error(result), zap.String("source", source),
 			zap.String("subject", subject), zap.String("id", event.ID()))

pkg/adapter/v2/cloudevents.go

@@ -1,4 +1,4 @@
 /*
 Copyright 2020 The Knative Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
@@ -20,6 +20,8 @@
 	"context"
 	"errors"
 	"fmt"
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/auth"
 	nethttp "net/http"
 	"net/url"
 	"time"
@@ -105,6 +107,7 @@
 }
 
 type ClientConfig struct {
+	Context             context.Context
 	Env                 EnvConfigAccessor
 	CeOverrides         *duckv1.CloudEventOverrides
 	Reporter            source.StatsReporter
@@ -195,6 +198,8 @@
 		ceOverrides:         ceOverrides,
 		reporter:            cfg.Reporter,
 		crStatusEventClient: cfg.CrStatusEventClient,
+		oidcTokenProvider:   auth.NewOIDCTokenProvider(cfg.Context),
+		audience:            cfg.Env.GetAudience(),
 	}, nil
 }
 
@@ -217,6 +222,10 @@
 	reporter            source.StatsReporter
 	crStatusEventClient *crstatusevent.CRStatusEventClient
 	closeIdler          closeIdler
+
+	oidcTokenProvider  *auth.OIDCTokenProvider
+	audience           *string
+	serviceAccountName types.NamespacedName
 }
 
 func (c *client) CloseIdleConnections() {
@@ -228,6 +237,22 @@
 // Send implements client.Send
 func (c *client) Send(ctx context.Context, out event.Event) protocol.Result {
 	c.applyOverrides(&out)
+
+	// If the sink has audience, then we need to request the JWT token
+	if c.audience != nil {
+		// Request the JWT token for the given service account
+		jwt, err := c.oidcTokenProvider.GetJWT(c.serviceAccountName, *c.audience)
+
+		if err != nil {
+			return protocol.NewResult("%w", err)
+		}
+
+		// Appending the auth token to the outgoing request
+		headers := http.HeaderFrom(ctx)
+		headers.Set("Authentication", fmt.Sprintf("Bearer %s", jwt))
+		ctx = http.WithCustomHeader(ctx, headers)
+	}
+
 	res := c.ceClient.Send(ctx, out)
 	c.reportMetrics(ctx, out, res)
 	return res
@@ -236,6 +261,22 @@
 // Request implements client.Request
 func (c *client) Request(ctx context.Context, out event.Event) (*event.Event, protocol.Result) {
 	c.applyOverrides(&out)
+
+	// If the sink has audience, then we need to request the JWT token
+	if c.audience != nil {
+		// Request the JWT token for the given service account
+		jwt, err := c.oidcTokenProvider.GetJWT(c.serviceAccountName, *c.audience)
+
+		if err != nil {
+			return nil, protocol.NewResult("%w", err)
+		}
+
+		// Appending the auth token to the outgoing request
+		headers := http.HeaderFrom(ctx)
+		headers.Set("Authentication", fmt.Sprintf("Bearer %s", jwt))
+		ctx = http.WithCustomHeader(ctx, headers)
+	}
+
 	resp, res := c.ceClient.Request(ctx, out)
 	c.reportMetrics(ctx, out, res)
 	return resp, res

pkg/adapter/v2/config.go

@@ -66,6 +66,9 @@ type EnvConfig struct {
 	// Sink is the URI messages will be sent.
 	Sink string `envconfig:"K_SINK"`
 
+	// Audience is the audience of the target sink.
+	Audience *string `envconfig:"K_AUDIENCE"`
+
 	// CACerts are the Certification Authority (CA) certificates in PEM format
 	// according to https://www.rfc-editor.org/rfc/rfc7468.
 	// +optional
@@ -113,6 +116,9 @@ type EnvConfigAccessor interface {
 	// GetCACerts gets the CACerts of the Sink.
 	GetCACerts() *string
 
+	// Get the audience of the target sink.
+	GetAudience() *string
+
 	// Get the namespace of the adapter.
 	GetNamespace() string
 
@@ -176,6 +182,10 @@ func (e *EnvConfig) GetCACerts() *string {
 	return e.CACerts
 }
 
+func (e *EnvConfig) GetAudience() *string {
+	return e.Audience
+}
+
 func (e *EnvConfig) GetNamespace() string {
 	return e.Namespace
 }

pkg/adapter/v2/main.go

@@ -215,6 +215,7 @@ func MainWithInformers(ctx context.Context, component string, env EnvConfigAcces
 	}
 
 	clientConfig := ClientConfig{
+		Context:             ctx,
 		Env:                 env,
 		Reporter:            reporter,
 		CrStatusEventClient: crStatusEventClient,

pkg/auth/serviceaccount.go

@@ -1,4 +1,4 @@
 /*
 Copyright 2023 The Knative Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
@@ -19,6 +19,7 @@
 import (
 	"context"
 	"fmt"
+	rbacv1 "k8s.io/api/rbac/v1"
 	"strings"
 
 	"go.uber.org/zap"
@@ -94,3 +95,102 @@
 
 	return nil
 }
+
+// EnsureOIDCServiceAccountRoleBindingExistsForResource
+// makes sure the given resource has an OIDC service account role binding with
+// an owner reference to the resource set.
+func EnsureOIDCServiceAccountRoleBindingExistsForResource(ctx context.Context, kubeclient kubernetes.Interface, gvk schema.GroupVersionKind, objectMeta metav1.ObjectMeta) error {
+	roleName := fmt.Sprintf("create-oidc-token")
+	roleBindingName := fmt.Sprintf("create-oidc-token", objectMeta.GetName())
+	roleBinding, err := kubeclient.RbacV1().RoleBindings(objectMeta.Namespace).Get(ctx, roleBindingName, metav1.GetOptions{})
+
+	// If the resource doesn't exist, we'll create it.
+	if apierrs.IsNotFound(err) {
+		logging.FromContext(ctx).Debugw("Creating OIDC service account role binding", zap.Error(err))
+
+		// Create the "create-oidc-token" role
+		CreateRoleForServiceAccount(ctx, kubeclient, gvk, objectMeta)
+
+		roleBinding := &rbacv1.RoleBinding{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      roleBindingName,
+				Namespace: objectMeta.GetNamespace(),
+				Annotations: map[string]string{
+					"description": fmt.Sprintf("Role Binding for OIDC Authentication for %s %q", gvk.GroupKind().Kind, objectMeta.Name),
+				},
+			},
+			RoleRef: rbacv1.RoleRef{
+				APIGroup: "rbac.authorization.k8s.io",
+				Kind:     "Role",
+				Name:     roleName,
+			},
+			Subjects: []rbacv1.Subject{
+				{
+					Kind:      "ServiceAccount",
+					Namespace: objectMeta.GetNamespace(),
+					Name:      GetOIDCServiceAccountNameForResource(gvk, objectMeta),
+				},
+			},
+		}
+
+		_, err = kubeclient.RbacV1().RoleBindings(objectMeta.Namespace).Create(ctx, roleBinding, metav1.CreateOptions{})
+		if err != nil {
+			return fmt.Errorf("could not create OIDC service account role binding %s/%s for %s: %w", objectMeta.Name, objectMeta.Namespace, gvk.Kind, err)
+		}
+
+		return nil
+	}
+
+	if err != nil {
+		return fmt.Errorf("could not get OIDC service account role binding %s/%s for %s: %w", objectMeta.Name, objectMeta.Namespace, gvk.Kind, err)
+
+	}
+
+	if !metav1.IsControlledBy(&roleBinding.ObjectMeta, &objectMeta) {
+		return fmt.Errorf("role binding %s not owned by %s %s", roleBinding.Name, gvk.Kind, objectMeta.Name)
+	}
+
+	return nil
+}
+
+// Create the create-oidc-token role for the service account
+func CreateRoleForServiceAccount(ctx context.Context, kubeclient kubernetes.Interface, gvk schema.GroupVersionKind, objectMeta metav1.ObjectMeta) error {
+	roleName := fmt.Sprintf("create-oidc-token")
+	role, err := kubeclient.RbacV1().Roles(objectMeta.Namespace).Get(ctx, roleName, metav1.GetOptions{})
+
+	// If the resource doesn't exist, we'll create it.
+	if apierrs.IsNotFound(err) && role == nil {
+		logging.FromContext(ctx).Debugw("Creating OIDC service account role", zap.Error(err))
+
+		role := &rbacv1.Role{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      roleName,
+				Namespace: objectMeta.GetNamespace(),
+				Annotations: map[string]string{
+					"description": fmt.Sprintf("Role for OIDC Authentication for %s %q", gvk.GroupKind().Kind, objectMeta.Name),
+				},
+			},
+			Rules: []rbacv1.PolicyRule{
+				rbacv1.PolicyRule{
+					APIGroups:     []string{""},
+					ResourceNames: []string{objectMeta.Name},
+					Resources:     []string{"serviceaccounts/token"},
+					Verbs:         []string{"create"},
+				},
+			},
+		}
+
+		_, err = kubeclient.RbacV1().Roles(objectMeta.Namespace).Create(ctx, role, metav1.CreateOptions{})
+		if err != nil {
+			return fmt.Errorf("could not create OIDC service account role %s/%s for %s: %w", objectMeta.Name, objectMeta.Namespace, gvk.Kind, err)
+		}
+
+		return nil
+	}
+
+	if err != nil {
+		return fmt.Errorf("could not get OIDC service account role %s/%s for %s: %w", objectMeta.Name, objectMeta.Namespace, gvk.Kind, err)
+
+	}
+	return nil
+}

pkg/kncloudevents/event_dispatcher.go

@@ -343,6 +343,7 @@ func (d *Dispatcher) createRequest(ctx context.Context, message binding.Message,
 		request.Header[key] = val
 	}
 
+	// saw something interesting here, it is similar to what we want to do in this PR
 	if oidcServiceAccount != nil {
 		if target.Audience != nil && *target.Audience != "" {
 			jwt, err := d.oidcTokenProvider.GetJWT(*oidcServiceAccount, *target.Audience)

pkg/reconciler/apiserversource/apiserversource.go

@@ -106,11 +106,20 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, source *v1.ApiServerSour
 			ServiceAccountName: &saName,
 		}
 
+		// set the role for the service account
+
 		if err := auth.EnsureOIDCServiceAccountExistsForResource(ctx, r.serviceAccountLister, r.kubeClientSet, v1.SchemeGroupVersion.WithKind("ApiServerSource"), source.ObjectMeta); err != nil {
 			source.Status.MarkOIDCIdentityCreatedFailed("Unable to resolve service account for OIDC authentication", "%v", err)
 			return err
 		}
 		source.Status.MarkOIDCIdentityCreatedSucceeded()
+
+		// TODO: add the role binding
+		if err := auth.EnsureOIDCServiceAccountRoleBindingExistsForResource(ctx, r.kubeClientSet, v1.SchemeGroupVersion.WithKind("ApiServerSource"), source.ObjectMeta); err != nil {
+			source.Status.MarkOIDCIdentityCreatedFailed("Unable to resolve role binding for OIDC authentication", "%v", err)
+			return err
+		}
+
 	} else {
 		source.Status.Auth = nil
 		source.Status.MarkOIDCIdentityCreatedSucceededWithReason(fmt.Sprintf("%s feature disabled", feature.OIDCAuthentication), "")
@@ -204,12 +213,18 @@ func (r *Reconciler) createReceiveAdapter(ctx context.Context, src *v1.ApiServer
 	// 	return nil, err
 	// }
 
+	fmt.Printf("haha sinkAddr: %v\n", sinkAddr)
+	fmt.Printf("haha sinkAddr.URL: %v\n", sinkAddr.URL)
+	fmt.Printf("haha sinkAddr.URL.String(): %v\n", sinkAddr.URL.String())
+	fmt.Printf("haha sinkAddr.audience: %v\n", sinkAddr.Audience)
+
 	adapterArgs := resources.ReceiveAdapterArgs{
 		Image:         r.receiveAdapterImage,
 		Source:        src,
 		Labels:        resources.Labels(src.Name),
 		CACerts:       sinkAddr.CACerts,
 		SinkURI:       sinkAddr.URL.String(),
+		Audience:      sinkAddr.Audience,
 		Configs:       r.configs,
 		Namespaces:    namespaces,
 		AllNamespaces: allNamespaces,
@@ -348,3 +363,5 @@ func (r *Reconciler) createCloudEventAttributes(src *v1.ApiServerSource) ([]duck
 	}
 	return ceAttributes, nil
 }
+
+// TODO: adding the rolebinding function to the resource folder and the role to the auth folder

pkg/reconciler/apiserversource/resources/receive_adapter.go

@@ -1,4 +1,4 @@
 /*
 Copyright 2020 The Knative Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
@@ -19,7 +19,6 @@
 import (
 	"encoding/json"
 	"fmt"
-
 	"knative.dev/eventing/pkg/adapter/v2"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -44,6 +43,7 @@
 	Image         string
 	Source        *v1.ApiServerSource
 	Labels        map[string]string
+	Audience      *string
 	SinkURI       string
 	CACerts       *string
 	Configs       reconcilersource.ConfigAccessor
@@ -120,6 +120,7 @@
 }
 
 func makeEnv(args *ReceiveAdapterArgs) ([]corev1.EnvVar, error) {
+	fmt.Printf("haha geting started make Env")
 	cfg := &apiserver.Config{
 		Namespaces:    args.Namespaces,
 		Resources:     make([]apiserver.ResourceWatch, 0, len(args.Source.Spec.Resources)),
@@ -181,6 +182,20 @@
 		})
 	}
 
+	fmt.Printf("haha receive_adapter: trying to add the k_audience env var\n")
+	if args.Audience != nil {
+		fmt.Printf("haha receive_adapter: adding the k_audience env var\n")
+		envs = append(envs, corev1.EnvVar{
+			Name:  "K_AUDIENCE",
+			Value: *args.Audience,
+		})
+	} else {
+		envs = append(envs, corev1.EnvVar{
+			Name:  "K_AUDIENCE",
+			Value: "0000",
+		})
+	}
+
 	envs = append(envs, args.Configs.ToEnvVars()...)
 
 	if args.Source.Spec.CloudEventOverrides != nil {