Implement reach command

cmd/npv/app/list.go

@@ -2,6 +2,7 @@ package app
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"sort"
@@ -102,6 +103,15 @@ func runList(ctx context.Context, w io.Writer, name string) error {
 	if err != nil {
 		return fmt.Errorf("failed to get endpoint information: %w", err)
 	}
+	if response.Payload == nil ||
+		response.Payload.Status == nil ||
+		response.Payload.Status.Policy == nil ||
+		response.Payload.Status.Policy.Realized == nil ||
+		response.Payload.Status.Policy.Realized.L4 == nil ||
+		response.Payload.Status.Policy.Realized.L4.Ingress == nil ||
+		response.Payload.Status.Policy.Realized.L4.Egress == nil {
+		return errors.New("api response is insufficient")
+	}
 
 	// The same rule appears multiple times in the response, so we need to dedup it
 	policySet := make(map[derivedFromEntry]struct{})

cmd/npv/app/reach.go

@@ -0,0 +1,159 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"io"
+	"strconv"
+
+	"github.com/cilium/cilium/pkg/u8proto"
+	"github.com/spf13/cobra"
+)
+
+var reachOptions struct {
+	from string
+	to   string
+}
+
+func init() {
+	reachCmd.Flags().StringVar(&reachOptions.from, "from", "", "egress pod")
+	reachCmd.Flags().StringVar(&reachOptions.to, "to", "", "ingress pod")
+	reachCmd.RegisterFlagCompletionFunc("from", completeNamespacePods)
+	reachCmd.RegisterFlagCompletionFunc("to", completeNamespacePods)
+	rootCmd.AddCommand(reachCmd)
+}
+
+var reachCmd = &cobra.Command{
+	Use:   "reach",
+	Short: "List traffic policies between pod pair",
+	Long:  `List traffic policies between pod pair`,
+
+	Args: cobra.ExactArgs(0),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		return runReach(context.Background(), cmd.OutOrStdout())
+	},
+}
+
+type reachEntry struct {
+	Namespace        string `json:"namespace"`
+	Name             string `json:"name"`
+	Direction        string `json:"direction"`
+	Policy           string `json:"policy"`
+	Identity         int    `json:"identity"`
+	WildcardProtocol bool   `json:"wildcard_protocol"`
+	WildcardPort     bool   `json:"wildcard_port"`
+	Protocol         int    `json:"protocol"`
+	Port             int    `json:"port"`
+	Bytes            int    `json:"bytes"`
+	Packets          int    `json:"packets"`
+}
+
+func runReach(ctx context.Context, w io.Writer) error {
+	if reachOptions.from == "" || reachOptions.to == "" {
+		return errors.New("--from and --to options are required")
+	}
+
+	from, err := parseNamespacedName(reachOptions.from)
+	if err != nil {
+		return errors.New("--from and --to should be specified as NAMESPACE/POD")
+	}
+
+	to, err := parseNamespacedName(reachOptions.to)
+	if err != nil {
+		return errors.New("--from and --to should be specified as NAMESPACE/POD")
+	}
+
+	clientset, dynamicClient, err := createK8sClients()
+	if err != nil {
+		return err
+	}
+
+	fromIdentity, err := getPodIdentity(ctx, dynamicClient, from.Namespace, from.Name)
+	if err != nil {
+		return err
+	}
+
+	toIdentity, err := getPodIdentity(ctx, dynamicClient, to.Namespace, to.Name)
+	if err != nil {
+		return err
+	}
+
+	fromPolicies, err := queryPolicyMap(ctx, clientset, dynamicClient, from.Namespace, from.Name)
+	if err != nil {
+		return err
+	}
+
+	toPolicies, err := queryPolicyMap(ctx, clientset, dynamicClient, to.Namespace, to.Name)
+	if err != nil {
+		return err
+	}
+
+	arr := make([]reachEntry, 0)
+	for _, p := range fromPolicies {
+		if (p.Key.Identity != 0) && (p.Key.Identity != int(toIdentity)) {
+			continue
+		}
+		if !p.IsEgressRule() {
+			continue
+		}
+		var entry reachEntry
+		entry.Namespace = from.Namespace
+		entry.Name = from.Name
+		entry.Direction = directionEgress
+		if p.IsDenyRule() {
+			entry.Policy = policyDeny
+		} else {
+			entry.Policy = policyAllow
+		}
+		entry.Identity = p.Key.Identity
+		entry.WildcardProtocol = p.IsWildcardProtocol()
+		entry.WildcardPort = p.IsWildcardPort()
+		entry.Protocol = p.Key.Protocol
+		entry.Port = p.Key.Port()
+		entry.Bytes = p.Bytes
+		entry.Packets = p.Packets
+		arr = append(arr, entry)
+	}
+	for _, p := range toPolicies {
+		if (p.Key.Identity != 0) && (p.Key.Identity != int(fromIdentity)) {
+			continue
+		}
+		if p.IsEgressRule() {
+			continue
+		}
+		var entry reachEntry
+		entry.Namespace = to.Namespace
+		entry.Name = to.Name
+		entry.Direction = directionIngress
+		if p.IsDenyRule() {
+			entry.Policy = policyDeny
+		} else {
+			entry.Policy = policyAllow
+		}
+		entry.Identity = p.Key.Identity
+		entry.WildcardProtocol = p.IsWildcardProtocol()
+		entry.WildcardPort = p.IsWildcardPort()
+		entry.Protocol = p.Key.Protocol
+		entry.Port = p.Key.Port()
+		entry.Bytes = p.Bytes
+		entry.Packets = p.Packets
+		arr = append(arr, entry)
+	}
+
+	header := []string{"NAMESPACE", "NAME", "DIRECTION", "POLICY", "IDENTITY", "PROTOCOL", "PORT", "BYTES", "PACKETS"}
+	return writeSimpleOrJson(w, arr, header, len(arr), func(index int) []any {
+		p := arr[index]
+		var protocol, port string
+		if p.WildcardProtocol {
+			protocol = "ANY"
+		} else {
+			protocol = u8proto.U8proto(p.Protocol).String()
+		}
+		if p.WildcardPort {
+			port = "ANY"
+		} else {
+			port = strconv.Itoa(p.Port)
+		}
+		return []any{p.Namespace, p.Name, p.Direction, p.Policy, p.Identity, protocol, port, p.Bytes, p.Packets}
+	})
+}

e2e/Makefile

@@ -51,7 +51,7 @@ run-test-pod-%:
 .PHONY: install-test-pod
 install-test-pod:
 	$(MAKE) --no-print-directory run-test-pod-self
-	$(MAKE) --no-print-directory DEPLOYMENT_REPLICAS=2 run-test-pod-l3-ingress-explicit-allow-all
+	$(MAKE) --no-print-directory run-test-pod-l3-ingress-explicit-allow-all
 	$(MAKE) --no-print-directory run-test-pod-l3-ingress-implicit-deny-all
 	$(MAKE) --no-print-directory run-test-pod-l3-ingress-explicit-deny-all
 	$(MAKE) --no-print-directory run-test-pod-l3-egress-implicit-deny-all
@@ -63,6 +63,12 @@ install-test-pod:
 	$(MAKE) --no-print-directory run-test-pod-l4-ingress-explicit-deny-udp
 	$(MAKE) --no-print-directory run-test-pod-l4-egress-explicit-deny-any
 	$(MAKE) --no-print-directory run-test-pod-l4-egress-explicit-deny-tcp
+	$(MAKE) --no-print-directory run-test-pod-l4-ingress-all-allow-tcp
+	$(MAKE) --no-print-directory wait-for-workloads
+
+	# Cilium-agents on different nodes may simultaneously create multiple CiliumIdentities for a same set of labels.
+	# To enforce the following test deployment to use a same CiliumIdentity, we first create it with replicas=1 and then upscale.
+	$(MAKE) --no-print-directory DEPLOYMENT_REPLICAS=2 run-test-pod-l3-ingress-explicit-allow-all
 	$(MAKE) --no-print-directory wait-for-workloads
 
 	kubectl apply -f testdata/policy/l3.yaml

e2e/id_test.go

@@ -1,6 +1,9 @@
 package e2e
 
 import (
+	"fmt"
+	"strconv"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 )
@@ -21,6 +24,7 @@ func testIdLabel() {
     "l3-ingress-implicit-deny-all",
     "l4-egress-explicit-deny-any",
     "l4-egress-explicit-deny-tcp",
+    "l4-ingress-all-allow-tcp",
     "l4-ingress-explicit-allow-any",
     "l4-ingress-explicit-allow-tcp",
     "l4-ingress-explicit-deny-any",
@@ -36,10 +40,38 @@ func testIdLabel() {
 }
 
 func testIdSummary() {
-	expected := `{"default":1,"kube-system":2,"local-path-storage":1,"test":12}`
+	cases := []struct {
+		Namespace string
+		Count     int
+	}{
+		{
+			Namespace: "default",
+			Count:     1,
+		},
+		{
+			Namespace: "kube-system",
+			Count:     2,
+		},
+		{
+			Namespace: "local-path-storage",
+			Count:     1,
+		},
+		{
+			Namespace: "test",
+			Count:     13,
+		},
+	}
 	It("should show ID summary", func() {
-		result := runViewerSafe(Default, nil, "id", "summary", "-o=json")
-		result = jqSafe(Default, result, "-c")
-		Expect(string(result)).To(Equal(expected), "compare failed.\nactual: %s\nexpected: %s", string(result), expected)
+		for _, c := range cases {
+			resultData := runViewerSafe(Default, nil, "id", "summary", "-o=json")
+			resultData = jqSafe(Default, resultData, "-r", fmt.Sprintf(`."%s"`, c.Namespace))
+			result, err := strconv.Atoi(string(resultData))
+			Expect(err).NotTo(HaveOccurred())
+
+			expected := c.Count
+
+			// Multiple CiliumIdentities may be generated for a same set of security-relevant labels
+			Expect(result).To(BeNumerically(">=", expected), "compare failed. namespace: %s\nactual: %d\nexpected: %d", result, expected)
+		}
 	})
 }

e2e/inspect_test.go

@@ -92,6 +92,12 @@ Deny,Ingress,self,false,false,17,161`,
 			Selector: "test=l4-egress-explicit-deny-tcp",
 			Expected: `Allow,Ingress,reserved:host,true,true,0,0`,
 		},
+		{
+			Selector: "test=l4-ingress-all-allow-tcp",
+			Expected: `Allow,Ingress,reserved:host,false,false,6,8080
+Allow,Ingress,reserved:host,true,true,0,0
+Allow,Ingress,reserved:unknown,false,false,6,8080`,
+		},
 	}
 
 	It("should inspect policy configuration", func() {

e2e/list_test.go

@@ -80,6 +80,12 @@ Ingress,CiliumClusterwideNetworkPolicy,-,l3-baseline`,
 			Expected: `Egress,CiliumClusterwideNetworkPolicy,-,l3-baseline
 Ingress,CiliumClusterwideNetworkPolicy,-,l3-baseline`,
 		},
+		{
+			Selector: "test=l4-ingress-all-allow-tcp",
+			Expected: `Egress,CiliumClusterwideNetworkPolicy,-,l3-baseline
+Ingress,CiliumClusterwideNetworkPolicy,-,l3-baseline
+Ingress,CiliumNetworkPolicy,test,l4-ingress-all-allow-tcp`,
+		},
 	}
 
 	It("should list applied policies", func() {

e2e/reach_test.go

@@ -0,0 +1,96 @@
+package e2e
+
+import (
+	"strings"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func testReach() {
+	cases := []struct {
+		Selector string
+		Expected string
+	}{
+		{
+			Selector: "test=l3-ingress-explicit-allow-all",
+			Expected: `test,l3-ingress-explicit-allow-all,Ingress,Allow,true,true,0,0
+test,self,Egress,Allow,true,true,0,0`,
+		},
+		{
+			Selector: "test=l3-ingress-implicit-deny-all",
+			Expected: `test,self,Egress,Allow,true,true,0,0`,
+		},
+		{
+			Selector: "test=l3-ingress-explicit-deny-all",
+			Expected: `test,l3-ingress-explicit-deny-all,Ingress,Deny,true,true,0,0
+test,self,Egress,Allow,true,true,0,0`,
+		},
+		{
+			Selector: "test=l3-egress-implicit-deny-all",
+			Expected: ``,
+		},
+		{
+			Selector: "test=l3-egress-explicit-deny-all",
+			Expected: `test,self,Egress,Deny,true,true,0,0`,
+		},
+		{
+			Selector: "test=l4-ingress-explicit-allow-any",
+			Expected: `test,l4-ingress-explicit-allow-any,Ingress,Allow,false,false,6,53
+test,l4-ingress-explicit-allow-any,Ingress,Allow,false,false,17,53
+test,l4-ingress-explicit-allow-any,Ingress,Allow,false,false,132,53
+test,self,Egress,Allow,false,false,6,53
+test,self,Egress,Allow,false,false,17,53
+test,self,Egress,Allow,false,false,132,53`,
+		},
+		{
+			Selector: "test=l4-ingress-explicit-allow-tcp",
+			Expected: `test,l4-ingress-explicit-allow-tcp,Ingress,Allow,false,false,6,8080
+test,self,Egress,Allow,false,false,6,8080`,
+		},
+		{
+			Selector: "test=l4-ingress-explicit-deny-any",
+			Expected: `test,l4-ingress-explicit-deny-any,Ingress,Deny,false,false,6,53
+test,l4-ingress-explicit-deny-any,Ingress,Deny,false,false,17,53
+test,l4-ingress-explicit-deny-any,Ingress,Deny,false,false,132,53
+test,self,Egress,Allow,false,false,6,53
+test,self,Egress,Allow,false,false,17,53
+test,self,Egress,Allow,false,false,132,53`,
+		},
+		{
+			Selector: "test=l4-ingress-explicit-deny-udp",
+			Expected: `test,l4-ingress-explicit-deny-udp,Ingress,Deny,false,false,17,161
+test,self,Egress,Allow,false,false,17,161`,
+		},
+		{
+			Selector: "test=l4-egress-explicit-deny-any",
+			Expected: `test,self,Egress,Deny,false,false,6,53
+test,self,Egress,Deny,false,false,17,53
+test,self,Egress,Deny,false,false,132,53`,
+		},
+		{
+			Selector: "test=l4-egress-explicit-deny-tcp",
+			Expected: `test,self,Egress,Deny,false,false,6,8080`,
+		},
+		{
+			Selector: "test=l4-ingress-all-allow-tcp",
+			Expected: `test,l4-ingress-all-allow-tcp,Ingress,Allow,false,false,6,8080`,
+		},
+	}
+
+	It("should list traffic policy", func() {
+		for _, c := range cases {
+			fromOption := "--from=test/" + onePodByLabelSelector(Default, "test", "test=self")
+			toOption := "--to=test/" + onePodByLabelSelector(Default, "test", c.Selector)
+
+			result := runViewerSafe(Default, nil, "reach", "-o=json", fromOption, toOption)
+			result = jqSafe(Default, result, "-r", `[.[] | .name = (.name | split("-") | .[0:5] | join("-"))]`)
+			result = jqSafe(Default, result, "-r", `[.[] | .name = (.name | if startswith("self") then "self" else . end)]`)
+			// "npv reach" returns a unstable result, so we need to sort it in test
+			result = jqSafe(Default, result, "-r", `sort_by(.namespace, .name, .direction, .policy, .wildcard_protocol, .wildcard_port, .protocol, .port)`)
+			result = jqSafe(Default, result, "-r", `.[] | [.namespace, .name, .direction, .policy, .wildcard_protocol, .wildcard_port, .protocol, .port] | @csv`)
+			resultString := strings.Replace(string(result), `"`, "", -1)
+			Expect(resultString).To(Equal(c.Expected), "compare failed. selector: %s\nactual: %s\nexpected: %s", c.Selector, resultString, c.Expected)
+		}
+	})
+}

e2e/suite_test.go

@@ -32,4 +32,5 @@ func runTest() {
 	Context("summary", testSummary)
 	Context("manifest-generate", testManifestGenerate)
 	Context("manifest-range", testManifestRange)
+	Context("reach", testReach)
 }