quorate/services/short/tunnel/admission_test.go

package tunnel

import (
	"encoding/json"
	"os"
	"strings"
	"testing"

	"github.com/google/cel-go/cel"
	admission "k8s.io/api/admissionregistration/v1"
	core "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/yaml"
)

func TestElectionPolicy(t *testing.T) {
	data, err := os.ReadFile("../k8s/tunnel/admission.yaml")
	if err != nil {
		t.Fatal(err)
	}
	var policy admission.ValidatingAdmissionPolicy
	if err := yaml.Unmarshal([]byte(strings.Split(string(data), "\n---")[0]), &policy); err != nil {
		t.Fatal(err)
	}
	env, err := cel.NewEnv(
		cel.Variable("object", cel.DynType), cel.Variable("oldObject", cel.DynType),
		cel.Variable("request", cel.DynType), cel.Variable("variables", cel.DynType),
	)
	if err != nil {
		t.Fatal(err)
	}
	eval := func(expression string, values map[string]any) any {
		t.Helper()
		ast, issues := env.Compile(expression)
		if issues.Err() != nil {
			t.Fatal(issues.Err())
		}
		program, err := env.Program(ast)
		if err != nil {
			t.Fatal(err)
		}
		result, _, err := program.Eval(values)
		if err != nil {
			t.Fatal(err)
		}
		return result.Value()
	}
	object := func(p *core.Pod) map[string]any {
		t.Helper()
		data, err := json.Marshal(p)
		if err != nil {
			t.Fatal(err)
		}
		var result map[string]any
		if err := json.Unmarshal(data, &result); err != nil {
			t.Fatal(err)
		}
		return result
	}
	pod := &core.Pod{
		ObjectMeta: metav1.ObjectMeta{Name: "gateway-1", Namespace: "short", Labels: map[string]string{
			"app.kubernetes.io/name": "tunnel", "role": "active",
		}},
		Spec: core.PodSpec{ServiceAccountName: "tunnel", Containers: []core.Container{{Name: "tunnel", Image: "synthetic:v1"}}},
	}
	tests := []struct {
		name   string
		change func(*core.Pod, *core.Pod)
		allow  bool
	}{
		{"claim", func(old, next *core.Pod) { delete(old.Labels, "role") }, true},
		{"release", func(old, next *core.Pod) { delete(next.Labels, "role") }, true},
		{"unchanged", func(old, next *core.Pod) {}, true},
		{"controller", func(old, next *core.Pod) { old.Spec.ServiceAccountName = "short-controller"; next.Spec = old.Spec }, false},
		{"forged-label", func(old, next *core.Pod) { delete(old.Labels, "app.kubernetes.io/name") }, false},
		{"other-namespace", func(old, next *core.Pod) { old.Namespace = "other"; next.Namespace = "other" }, false},
		{"image", func(old, next *core.Pod) { next.Spec.Containers[0].Image = "synthetic:v2" }, false},
		{"label-added", func(old, next *core.Pod) { next.Labels["extra"] = "value" }, false},
		{"label-removed", func(old, next *core.Pod) { delete(next.Labels, "app.kubernetes.io/name") }, false},
		{"label-changed", func(old, next *core.Pod) { next.Labels["app.kubernetes.io/name"] = "other" }, false},
		{"role-value", func(old, next *core.Pod) { next.Labels["role"] = "other" }, false},
		{"annotation", func(old, next *core.Pod) { next.Annotations = map[string]string{"extra": "value"} }, false},
		{"finalizer", func(old, next *core.Pod) { next.Finalizers = []string{"example.test/hold"} }, false},
		{"owner", func(old, next *core.Pod) { next.OwnerReferences = []metav1.OwnerReference{{Name: "other"}} }, false},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			old, next := pod.DeepCopy(), pod.DeepCopy()
			test.change(old, next)
			variables := map[string]any{}
			values := map[string]any{"object": object(next), "oldObject": object(old), "variables": variables,
				"request": map[string]any{"userInfo": map[string]any{"username": "system:serviceaccount:short:tunnel"}}}
			for _, match := range policy.Spec.MatchConditions {
				if eval(match.Expression, values) != true {
					t.Fatal("tunnel admission was skipped")
				}
			}
			for _, variable := range policy.Spec.Variables {
				variables[variable.Name] = eval(variable.Expression, values)
			}
			allowed := true
			for _, validation := range policy.Spec.Validations {
				if eval(validation.Expression, values) != true {
					allowed = false
				}
			}
			if allowed != test.allow {
				t.Fatalf("allowed=%v, want %v", allowed, test.allow)
			}
		})
	}
	for _, actor := range []string{"system:serviceaccount:flux-system:kustomize-controller", "system:kube-controller-manager"} {
		values := map[string]any{"request": map[string]any{"userInfo": map[string]any{"username": actor}}}
		if eval(policy.Spec.MatchConditions[0].Expression, values) != false {
			t.Fatalf("policy restricts unrelated controller %s", actor)
		}
	}
}