Implement HTTP Forwarded header policy API

This commit resolves NE-318.

https://issues.redhat.com/browse/NE-318

* pkg/operator/controller/ingress/deployment.go
(RouterForwardedHeadersPolicy): New constant with the name of the related
environment variable.
(desiredRouterDeployment): Set the ROUTER_SET_FORWARDED_HEADERS environment
variable as appropriate.
* pkg/operator/controller/ingress/deployment_test.go
(TestDesiredRouterDeployment): Verify that
spec.httpHeaders.forwardedHeaderPolicy has the expected effect.
* test/e2e/operator_test.go (TestForwardedHeaderPolicy): New test.

pkg/operator/controller/ingress/deployment.go

@@ -35,6 +35,8 @@ import (
 const (
 	WildcardRouteAdmissionPolicy = "ROUTER_ALLOW_WILDCARD_ROUTES"
 
+	RouterForwardedHeadersPolicy = "ROUTER_SET_FORWARDED_HEADERS"
+
 	RouterLogLevelEnvName       = "ROUTER_LOG_LEVEL"
 	RouterSyslogAddressEnvName  = "ROUTER_SYSLOG_ADDRESS"
 	RouterSyslogFormatEnvName   = "ROUTER_SYSLOG_FORMAT"
@@ -533,6 +535,12 @@ func desiredRouterDeployment(ci *operatorv1.IngressController, ingressController
 		env = append(env, corev1.EnvVar{Name: WildcardRouteAdmissionPolicy, Value: "false"})
 	}
 
+	forwardedHeaderPolicy := operatorv1.AppendHTTPHeaderPolicy
+	if ci.Spec.HTTPHeaders != nil && len(ci.Spec.HTTPHeaders.ForwardedHeaderPolicy) != 0 {
+		forwardedHeaderPolicy = ci.Spec.HTTPHeaders.ForwardedHeaderPolicy
+	}
+	env = append(env, corev1.EnvVar{Name: RouterForwardedHeadersPolicy, Value: string(forwardedHeaderPolicy)})
+
 	if HTTP2IsEnabled(ci, ingressConfig) {
 		env = append(env, corev1.EnvVar{Name: RouterDisableHTTP2EnvName, Value: "false"})
 	} else {

pkg/operator/controller/ingress/deployment_test.go

@@ -281,6 +281,8 @@ func TestDesiredRouterDeployment(t *testing.T) {
 	checkDeploymentHasEnvVar(t, deployment, "ROUTER_SYSLOG_ADDRESS", true, "/var/lib/rsyslog/rsyslog.sock")
 	checkDeploymentHasEnvVar(t, deployment, "ROUTER_SYSLOG_FORMAT", true, `"%ci:%cp [%t] %ft %b/%s %B %bq %HM %HU %HV"`)
 
+	checkDeploymentHasEnvVar(t, deployment, "ROUTER_SET_FORWARDED_HEADERS", true, "Append")
+
 	checkDeploymentHasEnvVar(t, deployment, "ROUTER_CIPHERS", true, "quux")
 
 	// TODO: Update when haproxy is built with an openssl version that supports tls v1.3.
@@ -304,6 +306,9 @@ func TestDesiredRouterDeployment(t *testing.T) {
 			},
 		},
 	}
+	ci.Spec.HTTPHeaders = &operatorv1.IngressControllerHTTPHeaders{
+		ForwardedHeaderPolicy: operatorv1.NeverHTTPHeaderPolicy,
+	}
 	ci.Spec.NodePlacement = &operatorv1.NodePlacement{
 		NodeSelector: &metav1.LabelSelector{
 			MatchLabels: map[string]string{
@@ -371,6 +376,8 @@ func TestDesiredRouterDeployment(t *testing.T) {
 	checkDeploymentHasEnvVar(t, deployment, "ROUTER_SYSLOG_ADDRESS", true, "1.2.3.4:12345")
 	checkDeploymentHasEnvVar(t, deployment, "ROUTER_SYSLOG_FORMAT", false, "")
 
+	checkDeploymentHasEnvVar(t, deployment, "ROUTER_SET_FORWARDED_HEADERS", true, "Never")
+
 	checkDeploymentHasEnvVar(t, deployment, "ROUTER_IP_V4_V6_MODE", true, "v6")
 	checkDeploymentHasEnvVar(t, deployment, RouterDisableHTTP2EnvName, true, "true")
 }

test/e2e/operator_test.go

@@ -41,6 +41,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 
@@ -1097,6 +1098,295 @@ func TestContainerLogging(t *testing.T) {
 	}
 }
 
+func TestForwardedHeaderPolicy(t *testing.T) {
+	icName := types.NamespacedName{Namespace: operatorNamespace, Name: "forwardedheader"}
+	domain := icName.Name + "." + dnsConfig.Spec.BaseDomain
+	ic := newPrivateController(icName, domain)
+	if err := kclient.Create(context.TODO(), ic); err != nil {
+		t.Fatalf("failed to create ingresscontroller %s: %v", icName, err)
+	}
+	defer assertIngressControllerDeleted(t, kclient, ic)
+	conditions := []operatorv1.OperatorCondition{
+		{Type: operatorv1.IngressControllerAvailableConditionType, Status: operatorv1.ConditionTrue},
+		{Type: operatorv1.LoadBalancerManagedIngressConditionType, Status: operatorv1.ConditionFalse},
+		{Type: operatorv1.DNSManagedIngressConditionType, Status: operatorv1.ConditionFalse},
+	}
+	if err := waitForIngressControllerCondition(kclient, 5*time.Minute, icName, conditions...); err != nil {
+		t.Fatalf("failed to observe expected conditions: %v", err)
+	}
+
+	deployment := &appsv1.Deployment{}
+	if err := kclient.Get(context.TODO(), controller.RouterDeploymentName(ic), deployment); err != nil {
+		t.Fatalf("failed to get ingresscontroller deployment: %v", err)
+	}
+	service := &corev1.Service{}
+	if err := kclient.Get(context.TODO(), controller.InternalIngressControllerServiceName(ic), service); err != nil {
+		t.Fatalf("failed to get ingresscontroller service: %v", err)
+	}
+
+	// Create a pod and route that echoes back the request.
+	echoPod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Labels: map[string]string{
+				"app": "echo",
+			},
+			Name:      "echo",
+			Namespace: deployment.Namespace,
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Args: []string{
+						"TCP4-LISTEN:8080,reuseaddr,fork",
+						`EXEC:'/bin/bash -c \"printf \\\"HTTP/1.0 200 OK\r\n\r\n\\\"; sed -e \\\"/^\r/q\\\"\"'`,
+					},
+					Command: []string{"/bin/socat"},
+					Image:   "openshift/origin-node",
+					Name:    "echo",
+					Ports: []corev1.ContainerPort{
+						{
+							ContainerPort: int32(8080),
+							Protocol:      corev1.ProtocolTCP,
+						},
+					},
+				},
+			},
+		},
+	}
+	if err := kclient.Create(context.TODO(), echoPod); err != nil {
+		t.Fatalf("failed to create pod %s/%s: %v", echoPod.Namespace, echoPod.Name, err)
+	}
+	defer func() {
+		if err := kclient.Delete(context.TODO(), echoPod); err != nil {
+			t.Fatalf("failed to delete pod %s/%s: %v", echoPod.Namespace, echoPod.Name, err)
+		}
+	}()
+
+	echoService := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: echoPod.Namespace,
+			Name:      echoPod.Name,
+		},
+		Spec: corev1.ServiceSpec{
+			Ports: []corev1.ServicePort{
+				{
+					Name:       "http",
+					Port:       int32(80),
+					Protocol:   corev1.ProtocolTCP,
+					TargetPort: intstr.FromInt(8080),
+				},
+			},
+			Selector: echoPod.ObjectMeta.Labels,
+		},
+	}
+	if err := kclient.Create(context.TODO(), echoService); err != nil {
+		t.Fatalf("failed to create service %s/%s: %v", echoService.Namespace, echoService.Name, err)
+	}
+	defer func() {
+		if err := kclient.Delete(context.TODO(), echoService); err != nil {
+			t.Fatalf("failed to delete service %s/%s: %v", echoService.Namespace, echoService.Name, err)
+		}
+	}()
+
+	echoRoute := &routev1.Route{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      echoPod.Name,
+			Namespace: echoPod.Namespace,
+		},
+		Spec: routev1.RouteSpec{
+			To: routev1.RouteTargetReference{
+				Kind: "Service",
+				Name: echoService.Name,
+			},
+		},
+	}
+	if err := kclient.Create(context.TODO(), echoRoute); err != nil {
+		t.Fatalf("failed to create route %s/%s: %v", echoRoute.Namespace, echoRoute.Name, err)
+	}
+	defer func() {
+		if err := kclient.Delete(context.TODO(), echoRoute); err != nil {
+			t.Fatalf("failed to delete route %s/%s: %v", echoRoute.Namespace, echoRoute.Name, err)
+		}
+	}()
+
+	// Create a client pod that sends a request to the echo route and checks
+	// whether it gets the expected number of X-Forwarded-For headers.
+	kubeConfig, err := config.GetConfig()
+	if err != nil {
+		t.Fatalf("failed to get kube config: %v", err)
+	}
+	client, err := kubernetes.NewForConfig(kubeConfig)
+	if err != nil {
+		t.Fatalf("failed to create kube client: %v", err)
+	}
+	var testPodCount int
+	testRoute := func(expectedMatches int, headers []string) {
+		t.Helper()
+		curlArgs := []string{
+			"-s",
+			"--retry", "15", "--retry-delay", "1",
+			"--max-time", "2",
+			"--resolve",
+			echoRoute.Spec.Host + ":80:" + service.Spec.ClusterIP,
+		}
+		for _, header := range headers {
+			curlArgs = append(curlArgs, "-H", header)
+		}
+		curlArgs = append(curlArgs, "http://"+echoRoute.Spec.Host)
+		testPodCount++
+		clientPod := &corev1.Pod{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      fmt.Sprintf("forwardedheader%d", testPodCount),
+				Namespace: echoRoute.Namespace,
+			},
+			Spec: corev1.PodSpec{
+				Containers: []corev1.Container{
+					{
+						Name:    "curl",
+						Image:   deployment.Spec.Template.Spec.Containers[0].Image,
+						Command: []string{"/bin/curl"},
+						Args:    curlArgs,
+					},
+				},
+				RestartPolicy: corev1.RestartPolicyNever,
+			},
+		}
+		if err := kclient.Create(context.TODO(), clientPod); err != nil {
+			t.Fatalf("failed to create pod %s/%s: %v", clientPod.Namespace, clientPod.Name, err)
+		}
+		defer func() {
+			if err := kclient.Delete(context.TODO(), clientPod); err != nil {
+				if !errors.IsNotFound(err) {
+					t.Fatalf("failed to delete pod %s/%s: %v", clientPod.Namespace, clientPod.Name, err)
+				}
+			}
+		}()
+		err = wait.PollImmediate(1*time.Second, 2*time.Minute, func() (bool, error) {
+			readCloser, err := client.CoreV1().Pods(clientPod.Namespace).GetLogs(clientPod.Name, &corev1.PodLogOptions{
+				Container: "curl",
+				Follow:    false,
+			}).Stream(context.TODO())
+			if err != nil {
+				t.Logf("failed to read output from pod %s: %v", clientPod.Name, err)
+				return false, nil
+			}
+			scanner := bufio.NewScanner(readCloser)
+			defer func() {
+				if err := readCloser.Close(); err != nil {
+					t.Errorf("failed to close reader for pod %s: %v", clientPod.Name, err)
+				}
+			}()
+			var numMatches int
+			for scanner.Scan() {
+				line := scanner.Text()
+				if strings.HasPrefix(line, "x-forwarded-for:") {
+					numMatches++
+					t.Logf("found match %d of %d expected: %s", numMatches, expectedMatches, line)
+				}
+			}
+			if numMatches > 0 && numMatches != expectedMatches {
+				return false, fmt.Errorf("got %d x-forwarded-for headers, expected %d", numMatches, expectedMatches)
+			}
+			return numMatches == expectedMatches, nil
+		})
+		if err != nil {
+			t.Fatalf("failed to observe the expected output: %v", err)
+		}
+	}
+
+	// The default policy is append.  If the client doesn't specify any
+	// X-Forwarded-For header in the request, the router should append 1
+	// X-Forwarded-For header.
+	testRoute(1, nil)
+	// If the client specifies 2 X-Forwarded-For headers, then the router
+	// should append a 3rd.
+	testRoute(3, []string{"x-forwarded-for:foo", "x-forwarded-for:bar"})
+
+	// Verify that we get the expected behavior if we set the policy to
+	// "append" explicitly.
+	if err := kclient.Get(context.TODO(), icName, ic); err != nil {
+		t.Fatalf("failed to get ingresscontroller: %v", err)
+	}
+	ic.Spec.HTTPHeaders = &operatorv1.IngressControllerHTTPHeaders{
+		ForwardedHeaderPolicy: operatorv1.AppendHTTPHeaderPolicy,
+	}
+	if err := kclient.Update(context.TODO(), ic); err != nil {
+		t.Fatalf("failed to update ingresscontroller: %v", err)
+	}
+	if err := waitForDeploymentEnvVar(t, kclient, deployment, 1*time.Minute, "ROUTER_SET_FORWARDED_HEADERS", "append"); err != nil {
+		t.Fatalf("failed to observe ROUTER_SET_FORWARDED_HEADERS=append: %v", err)
+	}
+	if err := waitForDeploymentComplete(t, kclient, deployment, 3*time.Minute); err != nil {
+		t.Fatalf("failed to observe expected conditions: %v", err)
+	}
+	testRoute(1, nil)
+	testRoute(3, []string{"x-forwarded-for:foo", "x-forwarded-for:bar"})
+
+	// Verify that we get the expected behavior if we set the policy to
+	// "replace".  We should always get 1 X-Forwarded-For header.
+	if err := kclient.Get(context.TODO(), icName, ic); err != nil {
+		t.Fatalf("failed to get ingresscontroller: %v", err)
+	}
+	ic.Spec.HTTPHeaders = &operatorv1.IngressControllerHTTPHeaders{
+		ForwardedHeaderPolicy: operatorv1.ReplaceHTTPHeaderPolicy,
+	}
+	if err := kclient.Update(context.TODO(), ic); err != nil {
+		t.Fatalf("failed to update ingresscontroller: %v", err)
+	}
+	if err := waitForDeploymentEnvVar(t, kclient, deployment, 1*time.Minute, "ROUTER_SET_FORWARDED_HEADERS", "replace"); err != nil {
+		t.Fatalf("failed to observe ROUTER_SET_FORWARDED_HEADERS=replace: %v", err)
+	}
+	if err := waitForDeploymentComplete(t, kclient, deployment, 3*time.Minute); err != nil {
+		t.Fatalf("failed to observe expected conditions: %v", err)
+	}
+	testRoute(1, nil)
+	testRoute(1, []string{"x-forwarded-for:foo", "x-forwarded-for:bar"})
+
+	// Verify that we get the expected behavior if we set the policy to
+	// "never".  We should always receive exactly as many X-Forwarded-For
+	// headers as we send.
+	if err := kclient.Get(context.TODO(), icName, ic); err != nil {
+		t.Fatalf("failed to get ingresscontroller: %v", err)
+	}
+	ic.Spec.HTTPHeaders = &operatorv1.IngressControllerHTTPHeaders{
+		ForwardedHeaderPolicy: operatorv1.NeverHTTPHeaderPolicy,
+	}
+	if err := kclient.Update(context.TODO(), ic); err != nil {
+		t.Fatalf("failed to update ingresscontroller: %v", err)
+	}
+	if err := waitForDeploymentEnvVar(t, kclient, deployment, 1*time.Minute, "ROUTER_SET_FORWARDED_HEADERS", "never"); err != nil {
+		t.Fatalf("failed to observe ROUTER_SET_FORWARDED_HEADERS=never: %v", err)
+	}
+	if err := waitForDeploymentComplete(t, kclient, deployment, 3*time.Minute); err != nil {
+		t.Fatalf("failed to observe expected conditions: %v", err)
+	}
+	testRoute(0, nil)
+	testRoute(2, []string{"x-forwarded-for:foo", "x-forwarded-for:bar"})
+
+	// Verify that we get the expected behavior if we set the policy to
+	// "if-none".  We should always receive at least 1 X-Forwarded-For
+	// header, and if the request specifies more than 1, we should receive
+	// exactly as many X-Forwarded-For headers as are in the request.
+	if err := kclient.Get(context.TODO(), icName, ic); err != nil {
+		t.Fatalf("failed to get ingresscontroller: %v", err)
+	}
+	ic.Spec.HTTPHeaders = &operatorv1.IngressControllerHTTPHeaders{
+		ForwardedHeaderPolicy: operatorv1.IfNoneHTTPHeaderPolicy,
+	}
+	if err := kclient.Update(context.TODO(), ic); err != nil {
+		t.Fatalf("failed to update ingresscontroller: %v", err)
+	}
+	if err := waitForDeploymentEnvVar(t, kclient, deployment, 1*time.Minute, "ROUTER_SET_FORWARDED_HEADERS", "if-none"); err != nil {
+		t.Fatalf("failed to observe ROUTER_SET_FORWARDED_HEADERS=if-none: %v", err)
+	}
+	if err := waitForDeploymentComplete(t, kclient, deployment, 3*time.Minute); err != nil {
+		t.Fatalf("failed to observe expected conditions: %v", err)
+	}
+	testRoute(1, nil)
+	testRoute(1, []string{"x-forwarded-for:foo"})
+	testRoute(2, []string{"x-forwarded-for:foo", "x-forwarded-for:bar"})
+}
+
 func newLoadBalancerController(name types.NamespacedName, domain string) *operatorv1.IngressController {
 	repl := int32(1)
 	return &operatorv1.IngressController{
@@ -1218,6 +1508,30 @@ func waitForDeploymentComplete(t *testing.T, cl client.Client, deployment *appsv
 	return nil
 }
 
+// Wait for the provided deployment to have the specified environment variable set.
+func waitForDeploymentEnvVar(t *testing.T, cl client.Client, deployment *appsv1.Deployment, timeout time.Duration, name, value string) error {
+	t.Helper()
+	deploymentName := types.NamespacedName{Namespace: deployment.Namespace, Name: deployment.Name}
+	err := wait.PollImmediate(1*time.Second, 1*time.Minute, func() (bool, error) {
+		deployment := &appsv1.Deployment{}
+		if err := kclient.Get(context.TODO(), deploymentName, deployment); err != nil {
+			t.Logf("error getting deployment %s: %v", name, err)
+			return false, nil
+		}
+		for _, container := range deployment.Spec.Template.Spec.Containers {
+			if container.Name == "router" {
+				for _, v := range container.Env {
+					if v.Name == name {
+						return v.Value == value, nil
+					}
+				}
+			}
+		}
+		return false, nil
+	})
+	return err
+}
+
 func clusterOperatorConditionMap(conditions ...configv1.ClusterOperatorStatusCondition) map[string]string {
 	conds := map[string]string{}
 	for _, cond := range conditions {