[WIP]: POC of triggers in different namespace than broker

cmd/webhook/main.go

@@ -145,9 +145,14 @@ func NewValidationAdmissionController(ctx context.Context, cmw configmap.Watcher
 	featureStore := feature.NewStore(logging.FromContext(ctx).Named("feature-config-store"))
 	featureStore.WatchConfigs(cmw)
 
+	cfg := injection.GetConfig(ctx)
+	if cfg == nil {
+		panic("cfg was nil")
+	}
+
 	// Decorate contexts with the current state of the config.
 	ctxFunc := func(ctx context.Context) context.Context {
-		return featureStore.ToContext(channelStore.ToContext(pingstore.ToContext(store.ToContext(ctx))))
+		return injection.WithConfig(featureStore.ToContext(channelStore.ToContext(pingstore.ToContext(store.ToContext(ctx)))), cfg)
 	}
 
 	return validation.NewAdmissionController(ctx,

config/core/resources/trigger.yaml

@@ -57,6 +57,9 @@ spec:
               broker:
                 description: Broker is the broker that this trigger receives events from.
                 type: string
+              brokerNamespace:
+                description: The namespace of the broker that this trigger receives events from.
+                type: string
               delivery:
                 description: Delivery contains the delivery spec for this specific trigger.
                 type: object

config/core/roles/webhook-clusterrole.yaml

@@ -29,6 +29,14 @@
       - "get"
       - "list"
       - "watch"
+
+  # for checking if user has permissions to make a cross namespace broker
+  - apiGroups:
+      - "authorization.k8s.io"
+    resources:
+      - "subjectaccessreviews"
+    verbs:
+      - "create"
 
   # For manipulating certs into secrets.
   - apiGroups:

docs/eventing-api.md

@@ -1869,6 +1869,16 @@ string
 </tr>
 <tr>
 <td>
+<code>brokerNamespace</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+</td>
+</tr>
+<tr>
+<td>
 <code>filter</code><br/>
 <em>
 <a href="#eventing.knative.dev/v1.TriggerFilter">
@@ -2274,6 +2284,16 @@ string
 </tr>
 <tr>
 <td>
+<code>brokerNamespace</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+</td>
+</tr>
+<tr>
+<td>
 <code>filter</code><br/>
 <em>
 <a href="#eventing.knative.dev/v1.TriggerFilter">

pkg/apis/eventing/v1/trigger_defaults.go

@@ -28,15 +28,18 @@ const (
 
 func (t *Trigger) SetDefaults(ctx context.Context) {
 	withNS := apis.WithinParent(ctx, t.ObjectMeta)
-	t.Spec.SetDefaults(withNS)
+	t.Spec.SetDefaults(withNS, t.GetNamespace())
 	setLabels(t)
 }
 
-func (ts *TriggerSpec) SetDefaults(ctx context.Context) {
+func (ts *TriggerSpec) SetDefaults(ctx context.Context, BrokerNamespace string) {
 	// Make a default filter that allows anything.
 	if ts.Filter == nil {
 		ts.Filter = &TriggerFilter{}
 	}
+	if ts.BrokerNamespace == "" {
+		ts.BrokerNamespace = BrokerNamespace
+	}
 	// Default the Subscriber namespace
 	ts.Subscriber.SetDefaults(ctx)
 	ts.Delivery.SetDefaults(ctx)

pkg/apis/eventing/v1/trigger_types.go

@@ -77,6 +77,8 @@ type TriggerSpec struct {
 	// Broker is the broker that this trigger receives events from.
 	Broker string `json:"broker,omitempty"`
 
+	BrokerNamespace string `json:"brokerNamespace,omitempty"`
+
 	// Filter is the filter to apply against all events from the Broker. Only events that pass this
 	// filter will be sent to the Subscriber. If not specified, will default to allowing all events.
 	//

pkg/apis/eventing/v1/trigger_validation.go

@@ -24,8 +24,14 @@ import (
 
 	cesqlparser "github.com/cloudevents/sdk-go/sql/v2/parser"
 	"go.uber.org/zap"
+	authv1 "k8s.io/api/authorization/v1"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+
+	//	"k8s.io/client-go/rest"
 	"knative.dev/pkg/apis"
+	"knative.dev/pkg/injection"
 	"knative.dev/pkg/kmp"
 	"knative.dev/pkg/logging"
 
@@ -42,6 +48,7 @@ func (t *Trigger) Validate(ctx context.Context) *apis.FieldError {
 	errs := t.Spec.Validate(apis.WithinSpec(ctx)).ViaField("spec")
 	errs = t.validateAnnotation(errs, DependencyAnnotation, t.validateDependencyAnnotation)
 	errs = t.validateAnnotation(errs, InjectionAnnotation, t.validateInjectionAnnotation)
+	errs = errs.Also(t.CheckBrokerNamespace(ctx))
 	if apis.IsInUpdate(ctx) {
 		original := apis.GetBaseline(ctx).(*Trigger)
 		errs = errs.Also(t.CheckImmutableFields(ctx, original))
@@ -66,6 +73,69 @@ func (ts *TriggerSpec) Validate(ctx context.Context) (errs *apis.FieldError) {
 	)
 }
 
+func (t *Trigger) CheckBrokerNamespace(ctx context.Context) *apis.FieldError {
+	if t.GetNamespace() == t.Spec.BrokerNamespace {
+		return nil
+	}
+
+	userInfo := apis.GetUserInfo(ctx)
+	if userInfo == nil {
+		return &apis.FieldError{
+			Paths:   []string{".spec.BrokerNamespace"},
+			Message: "failed to get userInfo, which is needed to validate triggers created with the namespace different than the broker's namespace",
+		}
+	}
+
+	config := injection.GetConfig(ctx)
+	logging.FromContext(ctx).Info("got config", zap.Any("config", config))
+	if config == nil {
+		return &apis.FieldError{
+			Paths:   []string{".spec.BrokerNamespace"},
+			Message: "failed to get config, which is needed to validate triggers created with the namespace different than the broker's namespace",
+		}
+	}
+
+	client, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return &apis.FieldError{
+			Paths:   []string{".spec.BrokerNamespace"},
+			Message: "failed to get k8s client, which is needed to validate triggers created with the namespace different than the broker's namespace",
+		}
+	}
+
+	action := authv1.ResourceAttributes{
+		Namespace: t.Spec.BrokerNamespace,
+		Verb:      "get",
+		Group:     "eventing.knative.dev",
+		Resource:  "brokers",
+	}
+
+	check := authv1.SubjectAccessReview{
+		Spec: authv1.SubjectAccessReviewSpec{
+			ResourceAttributes: &action,
+			User:               userInfo.Username,
+			Groups:             userInfo.Groups,
+		},
+	}
+
+	resp, err := client.AuthorizationV1().SubjectAccessReviews().Create(ctx, &check, metav1.CreateOptions{})
+	if err != nil {
+		return &apis.FieldError{
+			Paths:   []string{".spec.BrokerNamespace"},
+			Message: fmt.Sprintf("failed to make authorization request to see if user can get brokers in namespace: %s", err.Error()),
+		}
+	}
+
+	if !resp.Status.Allowed {
+		return &apis.FieldError{
+			Paths:   []string{".spec.BrokerNamespace"},
+			Message: fmt.Sprintf("user %s is not authorized to get brokers in namespace: %s", userInfo.Username, t.Spec.BrokerNamespace),
+		}
+	}
+
+	return nil
+}
+
 // CheckImmutableFields checks that any immutable fields were not changed.
 func (t *Trigger) CheckImmutableFields(ctx context.Context, original *Trigger) *apis.FieldError {
 	if original == nil {