Add matcher for used OIDC user name (#639)

* Add matcher for used OIDC user name

* run goimports

* Add matcher for OIDC username based on GVK of sender

* Remove claims info again

* Make argument name clearer

* Add doc comment

* Run goimports

* Addressed review comments and made error messages clearer

pkg/eventshub/assert/event_info_matchers.go

@@ -17,9 +17,17 @@ limitations under the License.
 package assert
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/injection/clients/dynamicclient"
+	"knative.dev/reconciler-test/pkg/environment"
+
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	cetest "github.com/cloudevents/sdk-go/v2/test"
 
@@ -114,6 +122,59 @@ func MatchStatusCode(statusCode int) eventshub.EventInfoMatcher {
 	}
 }
 
+// MatchOIDCUser matches the OIDC username used for the request
+func MatchOIDCUser(username string) eventshub.EventInfoMatcher {
+	return func(info eventshub.EventInfo) error {
+		if info.OIDCUserInfo == nil {
+			return fmt.Errorf("event OIDC usernames don't match: Expected %q, but no OIDC user info in the event", username)
+		}
+		if info.OIDCUserInfo.Username != username {
+			return fmt.Errorf("event OIDC usernames don't match. Expected: %q, Actual: %q", username, info.OIDCUserInfo.Username)
+		}
+
+		return nil
+	}
+}
+
+// MatchOIDCUserFromResource matches the given resources OIDC identifier
+func MatchOIDCUserFromResource(gvr schema.GroupVersionResource, resourceName string) eventshub.EventInfoMatcherCtx {
+
+	type AuthenticatableType struct {
+		metav1.TypeMeta   `json:",inline"`
+		metav1.ObjectMeta `json:"metadata,omitempty"`
+
+		Status struct {
+			Auth *duckv1.AuthStatus `json:"auth,omitempty"`
+		} `json:"status"`
+	}
+
+	return func(ctx context.Context, info eventshub.EventInfo) error {
+
+		env := environment.FromContext(ctx)
+
+		us, err := dynamicclient.Get(ctx).Resource(gvr).Namespace(env.Namespace()).Get(ctx, resourceName, metav1.GetOptions{})
+		if err != nil {
+			return fmt.Errorf("error getting resource: %w", err)
+		}
+
+		obj := &AuthenticatableType{}
+		if err = runtime.DefaultUnstructuredConverter.FromUnstructured(us.Object, obj); err != nil {
+			return fmt.Errorf("error from DefaultUnstructured.Dynamiconverter: %w", err)
+		}
+
+		if obj.Status.Auth == nil || obj.Status.Auth.ServiceAccountName == nil {
+			return fmt.Errorf("resource does not have an OIDC service account set")
+		}
+
+		objFullSAName := fmt.Sprintf("system:serviceaccount:%s:%s", obj.GetNamespace(), *obj.Status.Auth.ServiceAccountName)
+		if objFullSAName != info.OIDCUserInfo.Username {
+			return fmt.Errorf("OIDC identity in event does not match identity of resource. Event: %q, resource: %q", info.OIDCUserInfo.Username, objFullSAName)
+		}
+
+		return nil
+	}
+}
+
 // MatchHeartBeatsImageMessage matches that the data field of the event, in the format of the heartbeats image, contains the following msg field
 func MatchHeartBeatsImageMessage(expectedMsg string) cetest.EventMatcher {
 	return cetest.AllOf(

pkg/eventshub/event_info.go

@@ -24,6 +24,8 @@ import (
 	"strings"
 	"time"
 
+	v1 "k8s.io/api/authentication/v1"
+
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 )
 
@@ -87,6 +89,9 @@ type EventInfo struct {
 
 	// AdditionalInfo can be used by event generator implementations to add more event details
 	AdditionalInfo map[string]interface{} `json:"additionalInfo"`
+
+	// OIDCUserInfo is the user info of the subject of the OIDC token used in the request
+	OIDCUserInfo *v1.UserInfo `json:"oidcUserInfo,omitempty"`
 }
 
 // Pretty print the event. Meant for debugging.

pkg/eventshub/receiver/receiver.go

@@ -225,8 +225,11 @@ func (o *Receiver) ServeHTTP(writer http.ResponseWriter, request *http.Request)
 		statusCode = http.StatusBadRequest
 	}
 
+	var oidcUser *authv1.UserInfo
 	if o.oidcAudience != "" {
-		if err := o.validateJWT(request); err != nil {
+		var err error
+		oidcUser, err = o.verifyJWT(request)
+		if err != nil {
 			rejectErr = err
 			statusCode = http.StatusUnauthorized
 		}
@@ -270,16 +273,17 @@ func (o *Receiver) ServeHTTP(writer http.ResponseWriter, request *http.Request)
 	}
 
 	eventInfo := eventshub.EventInfo{
-		Error:       errString,
-		Event:       event,
-		HTTPHeaders: headers,
-		Origin:      request.RemoteAddr,
-		Observer:    o.Name,
-		Time:        time.Now(),
-		Sequence:    s,
-		Kind:        kind,
-		Connection:  eventshub.TLSConnectionStateToConnection(request.TLS),
-		StatusCode:  statusCode,
+		Error:        errString,
+		Event:        event,
+		HTTPHeaders:  headers,
+		Origin:       request.RemoteAddr,
+		Observer:     o.Name,
+		Time:         time.Now(),
+		Sequence:     s,
+		Kind:         kind,
+		Connection:   eventshub.TLSConnectionStateToConnection(request.TLS),
+		StatusCode:   statusCode,
+		OIDCUserInfo: oidcUser,
 	}
 
 	if err := o.EventLogs.Vent(eventInfo); err != nil {
@@ -310,15 +314,24 @@ func (o *Receiver) ServeHTTP(writer http.ResponseWriter, request *http.Request)
 	}
 }
 
-func (o *Receiver) validateJWT(request *http.Request) error {
+func (o *Receiver) getJWTFromRequest(request *http.Request) (string, error) {
 	authHeader := request.Header.Get("Authorization")
 	if authHeader == "" {
-		return fmt.Errorf("could not get Authorization header")
+		return "", fmt.Errorf("could not get Authorization header")
 	}
 
 	token := strings.TrimPrefix(authHeader, "Bearer ")
 	if len(token) == len(authHeader) {
-		return fmt.Errorf("could not get Bearer token from header")
+		return "", fmt.Errorf("could not get Bearer token from header")
+	}
+
+	return token, nil
+}
+
+func (o *Receiver) verifyJWT(request *http.Request) (*authv1.UserInfo, error) {
+	token, err := o.getJWTFromRequest(request)
+	if err != nil {
+		return nil, err
 	}
 
 	tokenReview, err := o.kubeclient.AuthenticationV1().TokenReviews().Create(o.ctx, &authv1.TokenReview{
@@ -331,18 +344,18 @@ func (o *Receiver) validateJWT(request *http.Request) error {
 	}, metav1.CreateOptions{})
 
 	if err != nil {
-		return fmt.Errorf("could not get token review: %w", err)
+		return nil, fmt.Errorf("could not get token review: %w", err)
 	}
 
 	if err := tokenReview.Status.Error; err != "" {
-		return fmt.Errorf(err)
+		return nil, fmt.Errorf(err)
 	}
 
 	if !tokenReview.Status.Authenticated {
-		return fmt.Errorf("user not authenticated")
+		return nil, fmt.Errorf("user not authenticated")
 	}
 
-	return nil
+	return &tokenReview.Status.User, nil
 }
 
 func isTLS(request *http.Request) bool {