Extend authenticators with a TokenReview one

Extend the authenticators which the KFP apiserver applies on a request
with a TokenReview authenticator.

This authenticator expects a ServiceAccountToken in a header with the
format: 'Authorization: Bearer <token>'

Part of kubeflow#5138

backend/src/apiserver/auth/auth_util.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/kubeflow/pipelines/backend/src/apiserver/client"
 	"github.com/kubeflow/pipelines/backend/src/apiserver/common"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"github.com/pkg/errors"
@@ -38,9 +39,15 @@ var IdentityHeaderMissingError = util.NewUnauthenticatedError(
 // Make this public for tests to force its re-instantiation
 var Authenticators []Authenticator
 
-func GetAuthenticators() []Authenticator {
+func GetAuthenticators(tokenReviewClient client.TokenReviewInterface) []Authenticator {
 	if Authenticators == nil {
 		Authenticators = []Authenticator{
+			NewTokenReviewAuthenticator(
+				common.AuthorizationBearerTokenHeader,
+				common.AuthorizationBearerTokenPrefix,
+				[]string{common.TokenReviewAudience},
+				tokenReviewClient,
+			),
 			NewHTTPHeaderAuthenticator(common.GetKubeflowUserIDHeader(), common.GetKubeflowUserIDPrefix()),
 		}
 	}

backend/src/apiserver/resource/resource_manager.go

@@ -1239,7 +1239,7 @@ func (r *ResourceManager) IsRequestAuthenticated(ctx context.Context) (string, e
 	// If the request header contains the user identity, requests are authorized
 	// based on the namespace field in the request.
 	var errlist []error
-	for _, auth := range kfpauth.GetAuthenticators() {
+	for _, auth := range kfpauth.GetAuthenticators(r.tokenReviewClient) {
 		userIdentity, err := auth.GetUserIdentity(ctx)
 		if err == nil {
 			return userIdentity, nil