gateway: policy filter implementation

Watch our custom PolicyFilter objects. Update a "Valid" status condition
to report on whether the PPL could be parsed successfully. Add logic to
apply policies to the generated Pomerium routes.

cmd/common.go

@@ -20,6 +20,7 @@ import (
 	gateway_v1 "sigs.k8s.io/gateway-api/apis/v1"
 	gateway_v1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
 
+	icgv1alpha1 "github.com/pomerium/ingress-controller/apis/gateway/v1alpha1"
 	icsv1 "github.com/pomerium/ingress-controller/apis/ingress/v1"
 )
 
@@ -59,6 +60,7 @@ func getScheme() (*runtime.Scheme, error) {
 		{"settings", icsv1.AddToScheme},
 		{"gateway_v1", gateway_v1.Install},
 		{"gateway_v1beta1", gateway_v1beta1.Install},
+		{"gateway.pomerium.io", icgv1alpha1.AddToScheme},
 	} {
 		if err := apply.fn(scheme); err != nil {
 			return nil, fmt.Errorf("%s: %w", apply.name, err)

config/crd/kustomization.yaml

@@ -5,6 +5,7 @@ apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
 resources:
 - bases/ingress.pomerium.io_pomerium.yaml
+- bases/gateway.pomerium.io_policyfilters.yaml
 #+kubebuilder:scaffold:crdkustomizeresource
 
 # the following config is for teaching kustomize how to do kustomization for CRDs.

controllers/gateway/controller.go

@@ -16,6 +16,7 @@ import (
 	gateway_v1 "sigs.k8s.io/gateway-api/apis/v1"
 	gateway_v1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
 
+	icgv1alpha1 "github.com/pomerium/ingress-controller/apis/gateway/v1alpha1"
 	"github.com/pomerium/ingress-controller/pomerium"
 )
 
@@ -50,6 +51,8 @@ type gatewayController struct {
 	client.Client
 	pomerium.GatewayReconciler
 	ControllerConfig
+
+	extensionFilters map[refKey]objectAndFilter
 }
 
 // NewGatewayController creates and registers a new controller for Gateway objects.
@@ -63,6 +66,7 @@ func NewGatewayController(
 		Client:            mgr.GetClient(),
 		GatewayReconciler: pgr,
 		ControllerConfig:  config,
+		extensionFilters:  make(map[refKey]objectAndFilter),
 	}
 
 	err := mgr.GetFieldIndexer().IndexField(ctx, &corev1.Secret{}, "type",
@@ -97,6 +101,7 @@ func NewGatewayController(
 		Watches(&corev1.Namespace{}, enqueueRequest).
 		Watches(&corev1.Service{}, enqueueRequest).
 		Watches(&gateway_v1beta1.ReferenceGrant{}, enqueueRequest).
+		Watches(&icgv1alpha1.PolicyFilter{}, enqueueRequest).
 		Complete(gtc)
 	if err != nil {
 		return fmt.Errorf("build controller: %w", err)

controllers/gateway/extensionfilters.go

@@ -0,0 +1,73 @@
+package gateway
+
+import (
+	context "context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	icgv1alpha1 "github.com/pomerium/ingress-controller/apis/gateway/v1alpha1"
+	"github.com/pomerium/ingress-controller/model"
+	"github.com/pomerium/ingress-controller/pomerium/gateway"
+)
+
+func (c *gatewayController) processExtensionFilters(
+	ctx context.Context,
+	config *model.GatewayConfig,
+	o *objects,
+) {
+	for _, pf := range o.PolicyFilters {
+		c.processPolicyFilter(ctx, pf)
+	}
+	config.ExtensionFilters = makeExtensionFilterMap(c.extensionFilters)
+}
+
+func (c *gatewayController) processPolicyFilter(ctx context.Context, pf *icgv1alpha1.PolicyFilter) {
+	logger := log.FromContext(ctx)
+
+	// Check to see if we already have a parsed representation of this filter.
+	k := refKeyForObject(pf)
+	f := c.extensionFilters[k]
+	if f.object != nil && f.object.GetGeneration() == pf.Generation {
+		return
+	}
+
+	filter, err := gateway.NewPolicyFilter(pf)
+
+	// Set a "Valid" condition with information about whether the policy could be parsed.
+	validCondition := metav1.Condition{
+		Type: "Valid",
+	}
+	if err == nil {
+		validCondition.Status = metav1.ConditionTrue
+		validCondition.Reason = "Valid"
+	} else {
+		validCondition.Status = metav1.ConditionFalse
+		validCondition.Reason = "Invalid"
+		validCondition.Message = err.Error()
+	}
+	if upsertCondition(&pf.Status.Conditions, pf.Generation, validCondition) {
+		if err := c.Status().Update(ctx, pf); err != nil {
+			logger.Error(err, "couldn't update PolicyFilter status", "name", pf.Name)
+		}
+	}
+
+	c.extensionFilters[k] = objectAndFilter{pf, filter}
+}
+
+type objectAndFilter struct {
+	object client.Object
+	filter model.ExtensionFilter
+}
+
+func makeExtensionFilterMap(
+	extensionFilters map[refKey]objectAndFilter,
+) map[model.ExtensionFilterKey]model.ExtensionFilter {
+	m := make(map[model.ExtensionFilterKey]model.ExtensionFilter)
+	for k, f := range extensionFilters {
+		key := model.ExtensionFilterKey{Kind: k.Kind, Namespace: k.Namespace, Name: k.Name}
+		m[key] = f.filter
+	}
+	return m
+}

controllers/gateway/fetch.go

@@ -10,6 +10,7 @@ import (
 	gateway_v1 "sigs.k8s.io/gateway-api/apis/v1"
 	gateway_v1beta1 "sigs.k8s.io/gateway-api/apis/v1beta1"
 
+	icgv1alpha1 "github.com/pomerium/ingress-controller/apis/gateway/v1alpha1"
 	"github.com/pomerium/ingress-controller/util"
 )
 
@@ -22,6 +23,7 @@ type objects struct {
 	ReferenceGrants         referenceGrantMap
 	TLSSecrets              map[refKey]*corev1.Secret
 	Services                map[types.NamespacedName]*corev1.Service
+	PolicyFilters           map[types.NamespacedName]*icgv1alpha1.PolicyFilter
 }
 
 type httpRouteAndOriginalStatus struct {
@@ -120,6 +122,17 @@ func (c *gatewayController) fetchObjects(ctx context.Context) (*objects, error)
 		o.Services[util.GetNamespacedName(s)] = s
 	}
 
+	// Fetch all PolicyFilters.
+	var pfl icgv1alpha1.PolicyFilterList
+	if err := c.List(ctx, &pfl); err != nil {
+		return nil, err
+	}
+	o.PolicyFilters = make(map[types.NamespacedName]*icgv1alpha1.PolicyFilter, 0)
+	for i := range pfl.Items {
+		pf := &pfl.Items[i]
+		o.PolicyFilters[util.GetNamespacedName(pf)] = pf
+	}
+
 	return &o, nil
 }
 

controllers/gateway/gateway.go

@@ -21,6 +21,8 @@ func (c *gatewayController) processGateways(
 ) *model.GatewayConfig {
 	var config model.GatewayConfig
 
+	c.processExtensionFilters(ctx, &config, o)
+
 	for key := range o.Gateways {
 		c.processGateway(ctx, &config, o, key)
 	}

model/gateway_config.go

@@ -6,12 +6,15 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	gateway_v1 "sigs.k8s.io/gateway-api/apis/v1"
+
+	pb "github.com/pomerium/pomerium/pkg/grpc/config"
 )
 
 // GatewayConfig represents the entirety of the Gateway-defined configuration.
 type GatewayConfig struct {
-	Routes       []GatewayHTTPRouteConfig
-	Certificates []*corev1.Secret
+	Routes           []GatewayHTTPRouteConfig
+	Certificates     []*corev1.Secret
+	ExtensionFilters map[ExtensionFilterKey]ExtensionFilter
 }
 
 // GatewayHTTPRouteConfig represents a single Gateway-defined route together
@@ -34,3 +37,15 @@ type GatewayHTTPRouteConfig struct {
 type BackendRefChecker interface {
 	Valid(obj client.Object, r *gateway_v1.BackendRef) bool
 }
+
+// ExtensionFilter represents a custom Pomerium route filter.
+type ExtensionFilter interface {
+	ApplyToRoute(*pb.Route)
+}
+
+// ExtensionFilterKey is a look-up key for available custom filters.
+type ExtensionFilterKey struct {
+	Kind      string
+	Namespace string
+	Name      string
+}

pomerium/gateway/filters.go

@@ -1,23 +1,42 @@
 package gateway
 
 import (
+	"fmt"
+	"strings"
+
+	"github.com/open-policy-agent/opa/ast"
 	gateway_v1 "sigs.k8s.io/gateway-api/apis/v1"
 
+	icgv1alpha1 "github.com/pomerium/ingress-controller/apis/gateway/v1alpha1"
+	"github.com/pomerium/ingress-controller/model"
 	pb "github.com/pomerium/pomerium/pkg/grpc/config"
+	"github.com/pomerium/pomerium/pkg/policy"
 )
 
-func applyFilters(route *pb.Route, filters []gateway_v1.HTTPRouteFilter) {
+func applyFilters(
+	route *pb.Route,
+	config *model.GatewayConfig,
+	routeConfig *model.GatewayHTTPRouteConfig,
+	filters []gateway_v1.HTTPRouteFilter,
+) {
 	for i := range filters {
-		applyFilter(route, &filters[i])
+		applyFilter(route, config, routeConfig, &filters[i])
 	}
 }
 
-func applyFilter(route *pb.Route, filter *gateway_v1.HTTPRouteFilter) {
+func applyFilter(
+	route *pb.Route,
+	config *model.GatewayConfig,
+	routeConfig *model.GatewayHTTPRouteConfig,
+	filter *gateway_v1.HTTPRouteFilter,
+) {
 	switch filter.Type {
 	case gateway_v1.HTTPRouteFilterRequestHeaderModifier:
 		applyRequestHeaderFilter(route, filter.RequestHeaderModifier)
 	case gateway_v1.HTTPRouteFilterRequestRedirect:
 		applyRedirectFilter(route, filter.RequestRedirect)
+	case gateway_v1.HTTPRouteFilterExtensionRef:
+		applyExtensionFilter(route, config, routeConfig, filter.ExtensionRef)
 	}
 }
 
@@ -54,3 +73,53 @@ func applyRedirectFilter(route *pb.Route, filter *gateway_v1.HTTPRequestRedirect
 	}
 	route.Redirect = &rr
 }
+
+func applyExtensionFilter(
+	route *pb.Route,
+	config *model.GatewayConfig,
+	routeConfig *model.GatewayHTTPRouteConfig,
+	filter *gateway_v1.LocalObjectReference,
+) {
+	// Make sure the API group is the one we expect.
+	if filter.Group != gateway_v1.Group(icgv1alpha1.GroupVersion.Group) {
+		return
+	}
+
+	k := model.ExtensionFilterKey{
+		Kind:      string(filter.Kind),
+		Namespace: routeConfig.Namespace,
+		Name:      string(filter.Name),
+	}
+	f := config.ExtensionFilters[k]
+	if f == nil {
+		return
+	}
+
+	f.ApplyToRoute(route)
+}
+
+// PolicyFilter applies a Pomerium policy defined by the PolicyFilter CRD.
+type PolicyFilter struct {
+	rego string
+}
+
+// NewPolicyFilter parses a PolicyFilter CRD object, returning an error if the object is not valid.
+func NewPolicyFilter(obj *icgv1alpha1.PolicyFilter) (*PolicyFilter, error) {
+	src, err := policy.GenerateRegoFromReader(strings.NewReader(obj.Spec.PPL))
+	if err != nil {
+		return nil, fmt.Errorf("couldn't parse policy: %w", err)
+	}
+
+	_, err = ast.ParseModule("policy.rego", src)
+	if err != nil && strings.Contains(err.Error(), "package expected") {
+		_, err = ast.ParseModule("policy.rego", "package pomerium.policy\n\n"+src)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("internal error: %w", err)
+	}
+	return &PolicyFilter{src}, nil
+}
+
+func (f *PolicyFilter) ApplyToRoute(r *pb.Route) {
+	r.Policies = append(r.Policies, &pb.Policy{Rego: []string{f.rego}})
+}

pomerium/gateway/translate.go

@@ -13,17 +13,20 @@ import (
 )
 
 // TranslateRoutes converts from Gateway-defined routes to Pomerium route configuration protos.
-func TranslateRoutes(gc *model.GatewayHTTPRouteConfig) []*pb.Route {
+func TranslateRoutes(
+	gatewayConfig *model.GatewayConfig,
+	routeConfig *model.GatewayHTTPRouteConfig,
+) []*pb.Route {
 	// A single HTTPRoute may need to be represented using many Pomerium routes:
 	//  - An HTTPRoute may have multiple hostnames.
 	//  - An HTTPRoute may have multiple HTTPRouteRules.
 	//  - An HTTPRouteRule may have multiple HTTPRouteMatches.
 	// First we'll expand all HTTPRouteRules into "template" Pomerium routes, and then we'll
 	// repeat each "template" route once per hostname.
-	trs := templateRoutes(gc)
+	trs := templateRoutes(gatewayConfig, routeConfig)
 
-	prs := make([]*pb.Route, 0, len(gc.Hostnames)*len(trs))
-	for _, h := range gc.Hostnames {
+	prs := make([]*pb.Route, 0, len(routeConfig.Hostnames)*len(trs))
+	for _, h := range routeConfig.Hostnames {
 		from := (&url.URL{
 			Scheme: "https",
 			Host:   string(h),
@@ -46,10 +49,13 @@ func TranslateRoutes(gc *model.GatewayHTTPRouteConfig) []*pb.Route {
 }
 
 // templateRoutes converts an HTTPRoute into zero or more Pomerium routes, ignoring hostname.
-func templateRoutes(gc *model.GatewayHTTPRouteConfig) []*pb.Route {
+func templateRoutes(
+	gatewayConfig *model.GatewayConfig,
+	routeConfig *model.GatewayHTTPRouteConfig,
+) []*pb.Route {
 	var prs []*pb.Route
 
-	rules := gc.Spec.Rules
+	rules := routeConfig.Spec.Rules
 	for i := range rules {
 		rule := &rules[i]
 		pr := &pb.Route{}
@@ -60,8 +66,8 @@ func templateRoutes(gc *model.GatewayHTTPRouteConfig) []*pb.Route {
 		// forward this header unmodified to the backend."
 		pr.PreserveHostHeader = true
 
-		applyFilters(pr, rule.Filters)
-		applyBackendRefs(pr, gc, rule.BackendRefs)
+		applyFilters(pr, gatewayConfig, routeConfig, rule.Filters)
+		applyBackendRefs(pr, routeConfig, rule.BackendRefs)
 
 		if len(rule.Matches) == 0 {
 			prs = append(prs, pr)

pomerium/sync.go

@@ -140,7 +140,7 @@ func (r *DataBrokerReconciler) SetGatewayConfig(
 
 	for i := range config.Routes {
 		r := &config.Routes[i]
-		next.Routes = append(next.Routes, gateway.TranslateRoutes(r)...)
+		next.Routes = append(next.Routes, gateway.TranslateRoutes(config, r)...)
 	}
 	next.Settings = new(pb.Settings)
 	for _, cert := range config.Certificates {