JobSink: Reject unauthorized requests (#8169)

* Reject unauthorized requests in JobSink

* Add e2e test

cmd/jobsink/main.go

@@ -191,22 +191,19 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		Name:      parts[2],
 	}
 
-	logger.Debug("Handling POST request", zap.String("URI", r.RequestURI))
-
-	features := feature.FromContext(ctx)
-	logger.Debug("features", zap.Any("features", features))
-
-	if features.IsOIDCAuthentication() {
-		logger.Debug("OIDC authentication is enabled")
+	js, err := h.lister.JobSinks(ref.Namespace).Get(ref.Name)
+	if err != nil {
+		logger.Warn("Failed to retrieve jobsink", zap.String("ref", ref.String()), zap.Error(err))
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
 
-		audience := auth.GetAudienceDirect(sinksv.SchemeGroupVersion.WithKind("JobSink"), ref.Namespace, ref.Name)
+	logger.Debug("Handling POST request", zap.String("URI", r.RequestURI))
 
-		err := h.oidcTokenVerifier.VerifyJWTFromRequest(ctx, r, &audience, w)
-		if err != nil {
-			logger.Warn("Error when validating the JWT token in the request", zap.Error(err))
-			return
-		}
-		logger.Debug("Request contained a valid JWT. Continuing...")
+	err = h.oidcTokenVerifier.VerifyRequest(ctx, feature.FromContext(ctx), js.Status.Address.Audience, js.Namespace, js.Status.Policies, r, w)
+	if err != nil {
+		logger.Warn("Failed to verify AuthN and AuthZ.", zap.Error(err))
+		return
 	}
 
 	message := cehttp.NewMessageFromHttpRequest(r)
@@ -225,13 +222,6 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	js, err := h.lister.JobSinks(ref.Namespace).Get(ref.Name)
-	if err != nil {
-		logger.Warn("Failed to retrieve jobsink", zap.String("ref", ref.String()), zap.Error(err))
-		w.WriteHeader(http.StatusBadRequest)
-		return
-	}
-
 	id := toIdHashLabelValue(event.Source(), event.ID())
 	logger.Debug("Getting job for event", zap.String("URI", r.RequestURI), zap.String("id", id))
 

test/rekt/job_sink_test.go

@@ -22,13 +22,17 @@ package rekt
 import (
 	"testing"
 
+	"knative.dev/eventing/test/rekt/features/authz"
+
 	"knative.dev/pkg/system"
 	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
 
 	"knative.dev/eventing/test/rekt/features/jobsink"
+	jsresource "knative.dev/eventing/test/rekt/resources/jobsink"
 )
 
 func TestJobSinkSuccess(t *testing.T) {
@@ -74,3 +78,21 @@ func TestJobSinkOIDC(t *testing.T) {
 
 	env.Test(ctx, t, jobsink.OIDC())
 }
+
+func TestJobSinkSupportsAuthZ(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		eventshub.WithTLS(t),
+		environment.Managed(t),
+	)
+
+	name := feature.MakeRandomK8sName("jobsink")
+	env.Prerequisite(ctx, t, jsresource.GoesReadySimple(name))
+
+	env.TestSet(ctx, t, authz.AddressableAuthZConformance(jsresource.GVR(), "JobSink", name))
+}

test/rekt/resources/jobsink/jobsink.go

@@ -20,9 +20,12 @@ import (
 	"context"
 	"embed"
 	"encoding/json"
+	"fmt"
 	"strings"
 	"time"
 
+	"knative.dev/pkg/apis"
+
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -203,3 +206,20 @@ func AsKReference(name string) *duckv1.KReference {
 		APIVersion: GVR().GroupVersion().String(),
 	}
 }
+
+// GoesReadySimple returns a feature that will create a JobSink with a simple sink
+// and confirm it becomes ready with an address.
+func GoesReadySimple(name string) *feature.Feature {
+	f := new(feature.Feature)
+
+	sink := feature.MakeRandomK8sName("sink")
+	sinkURL := &apis.URL{Scheme: "http", Host: sink}
+
+	f.Setup("Install sink", eventshub.Install(sink, eventshub.StartReceiver))
+
+	f.Setup(fmt.Sprintf("install JobSink %q", name), Install(name, WithForwarderJob(sinkURL.String())))
+	f.Setup("JobSink is ready", IsReady(name))
+	f.Setup("JobSink is addressable", IsAddressable(name))
+
+	return f
+}