From 283098d8070f9e0fc194c698b8619f32cbfb4816 Mon Sep 17 00:00:00 2001 From: tanzee Date: Sun, 23 Aug 2026 20:35:53 +0200 Subject: [PATCH 1/5] feat(cel/network): serviceRef/serviceSelector/host neighbor resolution Let a ContainerProfile allowlist cluster-infrastructure egress/ingress by Service name, Service label selector, or host entity instead of a broad ipAddresses serviceCIDR that blinds R0011/R0012 to lateral movement. Each serviceRef/serviceSelector/entity neighbor resolves at projection time to the concrete ClusterIP + backing-endpoint (or node/gateway) IPs it stands for, carrying its own ports, and is appended as an ordinary selector-free ipAddresses neighbor. The existing port-sensitive address matcher enforces it unchanged; unresolved selectors contribute nothing (never a match-all). - pkg/networkpeer: Resolve/Matches/ResolveIPs + Lister over Service, EndpointSlice and Node informers, with a generation counter so a profile projected before the informers synced re-projects once the view changes. - objectcache/reconciler: mark profiles that use service resolution and re-project them when the lister generation advances; plain profiles keep the identical old fast-skip path. - cmd/main.go: cluster-wide Service/EndpointSlice informers + a node-scoped Node informer, started non-blocking (no WaitForCacheSync on the hot path). - fail closed on ServiceSelector MatchExpressions / empty matchLabels and on any namespaceSelector other than kubernetes.io/metadata.name=. - Test_50 component test (serviceRef egress allowed, external egress still fires R0011) + resolve/expand/lister unit tests + fixture-lint R-NN-12 extended to accept the new target fields. Depends on the storage schema fields ServiceRefNamespace/ServiceRefName/ ServiceSelector/Entity; go.mod pins the fork's storage until the companion upstream storage PR lands. Signed-off-by: tanzee --- .github/workflows/component-tests.yaml | 3 +- cmd/main.go | 47 ++++ go.mod | 2 + go.sum | 4 +- pkg/networkpeer/expand.go | 146 ++++++++++++ pkg/networkpeer/expand_test.go | 209 ++++++++++++++++++ pkg/networkpeer/lister.go | 167 ++++++++++++++ pkg/networkpeer/lister_test.go | 147 ++++++++++++ pkg/networkpeer/resolve.go | 197 +++++++++++++++++ pkg/networkpeer/resolve_test.go | 204 +++++++++++++++++ .../containerprofilecache.go | 39 +++- .../containerprofilecache/reconciler.go | 40 ++-- tests/component_test.go | 89 ++++++++ .../containerprofile-serviceref-network.yaml | 48 ++++ tests/resources/network_fixture_lint_test.go | 12 +- .../serviceref-client-deployment.yaml | 21 ++ 16 files changed, 1351 insertions(+), 24 deletions(-) create mode 100644 pkg/networkpeer/expand.go create mode 100644 pkg/networkpeer/expand_test.go create mode 100644 pkg/networkpeer/lister.go create mode 100644 pkg/networkpeer/lister_test.go create mode 100644 pkg/networkpeer/resolve.go create mode 100644 pkg/networkpeer/resolve_test.go create mode 100644 tests/resources/containerprofile-serviceref-network.yaml create mode 100644 tests/resources/serviceref-client-deployment.yaml diff --git a/.github/workflows/component-tests.yaml b/.github/workflows/component-tests.yaml index f3325394eb..f89cfb5639 100644 --- a/.github/workflows/component-tests.yaml +++ b/.github/workflows/component-tests.yaml @@ -107,7 +107,8 @@ jobs: Test_36_MultiContainerPerContainerBinding, Test_43_RelativeOpenPathResolution, Test_48_MultiSubtypeGroupedProfileDocument, - Test_49_EphemeralContainerFullTreatment + Test_49_EphemeralContainerFullTreatment, + Test_50_ServiceRefNetworkNeighbor ] steps: - name: Checkout code diff --git a/cmd/main.go b/cmd/main.go index e41b256727..92fd665d24 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -39,11 +39,13 @@ import ( "github.com/kubescape/node-agent/pkg/malwaremanager" malwaremanagerv1 "github.com/kubescape/node-agent/pkg/malwaremanager/v1" otelmetrics "github.com/kubescape/node-agent/pkg/metricsmanager/otel" + "github.com/kubescape/node-agent/pkg/networkpeer" "github.com/kubescape/node-agent/pkg/networkstream" networkstreamv1 "github.com/kubescape/node-agent/pkg/networkstream/v1" "github.com/kubescape/node-agent/pkg/nodeprofilemanager" nodeprofilemanagerv1 "github.com/kubescape/node-agent/pkg/nodeprofilemanager/v1" "github.com/kubescape/node-agent/pkg/objectcache" + "github.com/kubescape/node-agent/pkg/objectcache/containerprofilecache" "github.com/kubescape/node-agent/pkg/objectcache/dnscache" "github.com/kubescape/node-agent/pkg/objectcache/k8scache" @@ -72,6 +74,8 @@ import ( "github.com/kubescape/node-agent/pkg/watcher/seccompprofilewatcher" goruntime "go.opentelemetry.io/contrib/instrumentation/runtime" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/informers" + toolscache "k8s.io/client-go/tools/cache" ) func main() { @@ -321,6 +325,49 @@ func main() { ruleBindingCache.AddNotifier(&ruleBindingNotify) cpc := containerprofilecache.NewContainerProfileCache(cfg, storageClient, k8sObjectCache, metricsProvider) + // Resolve serviceRef/serviceSelector/entity network neighbors against + // live cluster state (Service ClusterIPs + endpoints, Node IPs + CNI + // gateway) at projection time. Services and EndpointSlices are watched + // cluster-wide (a profile may reference any namespace's Service); the + // Node informer is field-selected to this agent's own node — the "host" + // entity is local, and a cluster-wide Node watch on every DaemonSet pod + // is O(nodes^2) traffic for no benefit. + svcInformers := informers.NewSharedInformerFactory(k8sClient.GetKubernetesClient(), 0) + nodeInformers := informers.NewSharedInformerFactoryWithOptions( + k8sClient.GetKubernetesClient(), 0, + informers.WithTweakListOptions(func(o *metav1.ListOptions) { + o.FieldSelector = "metadata.name=" + cfg.NodeName + }), + ) + serviceLister := networkpeer.NewInformerLister( + svcInformers.Core().V1().Services().Lister(), + svcInformers.Discovery().V1().EndpointSlices().Lister(), + nodeInformers.Core().V1().Nodes().Lister(), + cfg.NodeName, + ) + // Advance the lister generation on any Service/EndpointSlice/Node change, + // so the reconciler re-projects serviceRef/entity profiles when the + // cluster view moves (endpoint churn, or caches that fill after startup). + // Per-event cost is a single atomic increment; the re-projection itself + // is coalesced onto the reconcile tick, and only serviceRef-using + // profiles are eligible. + bump := toolscache.ResourceEventHandlerFuncs{ + AddFunc: func(interface{}) { serviceLister.Bump() }, + UpdateFunc: func(_, _ interface{}) { serviceLister.Bump() }, + DeleteFunc: func(interface{}) { serviceLister.Bump() }, + } + _, _ = svcInformers.Core().V1().Services().Informer().AddEventHandler(bump) + _, _ = svcInformers.Discovery().V1().EndpointSlices().Informer().AddEventHandler(bump) + _, _ = nodeInformers.Core().V1().Nodes().Informer().AddEventHandler(bump) + // Start the informers and hand the lister over WITHOUT blocking on cache + // sync: node-agent's core startup (container watcher, profiling) must not + // wait on these, and a bounded wait here previously delayed learning + // enough to trip the tight completion budget of Test_22. The caches fill + // in the background; serviceRef/entity neighbors resolve on the next + // reconcile once populated. nil-until-set is a no-op in projection. + svcInformers.Start(ctx.Done()) + nodeInformers.Start(ctx.Done()) + cpc.SetServiceLister(serviceLister) cpc.Start(ctx) if cpm, ok := containerProfileManager.(*containerprofilemanagerv1.ContainerProfileManager); ok { cpm.SetCompletionNotifier(cpc) diff --git a/go.mod b/go.mod index 808f11127a..4013bc3b28 100644 --- a/go.mod +++ b/go.mod @@ -479,3 +479,5 @@ replace github.com/anchore/syft => github.com/kubescape/syft v1.32.0-ks.2 replace github.com/anchore/stereoscope => github.com/anchore/stereoscope v0.1.9 replace github.com/opencontainers/runtime-spec => github.com/opencontainers/runtime-spec v1.2.1 + +replace github.com/kubescape/storage => github.com/k8sstormcenter/storage v0.0.240-0.20260823123818-6f3a2ee6385d diff --git a/go.sum b/go.sum index b239690414..0c78ed4da6 100644 --- a/go.sum +++ b/go.sum @@ -859,6 +859,8 @@ github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHm github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU= github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk= github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w= +github.com/k8sstormcenter/storage v0.0.240-0.20260823123818-6f3a2ee6385d h1:4d7cpcoXpp8ZJXV2W/ubX5WW78gn9diy9qvLFbROvV0= +github.com/k8sstormcenter/storage v0.0.240-0.20260823123818-6f3a2ee6385d/go.mod h1:d/1hqWPda2clsjx2wmQgysnB5dThIo3rDKP7RWx+v+M= github.com/kastenhq/goversion v0.0.0-20230811215019-93b2f8823953 h1:WdAeg/imY2JFPc/9CST4bZ80nNJbiBFCAdSZCSgrS5Y= github.com/kastenhq/goversion v0.0.0-20230811215019-93b2f8823953/go.mod h1:6o+UrvuZWc4UTyBhQf0LGjW9Ld7qJxLz/OqvSOWWlEc= github.com/kevinburke/ssh_config v1.2.0 h1:x584FjTGwHzMwvHx18PXxbBVzfnxogHaAReU4gf13a4= @@ -893,8 +895,6 @@ github.com/kubescape/go-logger v0.0.32 h1:4mI+XJOV8VFCMewrEE9VIFEIOhzXokYT3nFpNf github.com/kubescape/go-logger v0.0.32/go.mod h1:Alj7JBQ8/WCxbXe8Ura6ZheSRK45E0p21M3xeqedX90= github.com/kubescape/k8s-interface v0.0.214 h1:j7KP0/5VvYOoQdBGV2+gRM3qnR8PWLAGF8RM/k/DmJ0= github.com/kubescape/k8s-interface v0.0.214/go.mod h1:WNYUG93aZ5kDmuaRKFLtVhp18Yc6EfaHdD1gLYtVTN4= -github.com/kubescape/storage v0.0.303 h1:0nXI6E07lbWsg7iEH04vR4kwiekj//uCQl/La+8j4aM= -github.com/kubescape/storage v0.0.303/go.mod h1:d/1hqWPda2clsjx2wmQgysnB5dThIo3rDKP7RWx+v+M= github.com/kubescape/syft v1.32.0-ks.2 h1:xdUksUmKEyyVKsTfJDYW8Z5HawVJtelsUolPOsWtDx0= github.com/kubescape/syft v1.32.0-ks.2/go.mod h1:E6Kd4iBM2ljUOUQvSt7hVK6vBwaHkMXwcvBZmGMSY5o= github.com/kubescape/workerpool v0.0.0-20250526074519-0e4a4e7f44cf h1:hI0jVwrB6fT4GJWvuUjzObfci1CUknrZdRHfnRVtKM0= diff --git a/pkg/networkpeer/expand.go b/pkg/networkpeer/expand.go new file mode 100644 index 0000000000..2f76c97897 --- /dev/null +++ b/pkg/networkpeer/expand.go @@ -0,0 +1,146 @@ +package networkpeer + +import ( + "strings" + + "github.com/kubescape/storage/pkg/apis/softwarecomposition/v1beta1" +) + +// ExpandServiceNeighbors resolves every serviceRef / serviceSelector / entity +// neighbor in the list against the cluster view and returns equivalent +// synthesized ipAddresses neighbors (one per source neighbor, carrying the +// resolved IPs and the source neighbor's own ports). +// +// The synthesized neighbors are ordinary selector-free ipAddresses entries, so +// the existing port-sensitive address matcher handles them with no further +// change — a serviceRef/host neighbor becomes exactly the narrow, resolved +// ipAddresses entry it stands for. Neighbors that resolve to nothing (unknown +// Service, selector matching nothing, unknown entity) contribute nothing — +// never a match-all. Callers append the result to the same direction (egress +// or ingress) before projecting the profile. +func ExpandServiceNeighbors(neighbors []v1beta1.NetworkNeighbor, l Lister) []v1beta1.NetworkNeighbor { + if l == nil { + return nil + } + var out []v1beta1.NetworkNeighbor + for i := range neighbors { + n := &neighbors[i] + spec, ok := specFromNeighbor(n) + if !ok { + continue + } + ips := ResolveIPs(spec, l) + if len(ips) == 0 { + continue + } + out = append(out, v1beta1.NetworkNeighbor{ + Identifier: n.Identifier + "-resolved", + Type: n.Type, + IPAddresses: ips, + Ports: n.Ports, + }) + } + return out +} + +// WithResolvedServiceNeighbors returns cp with every serviceRef/serviceSelector/ +// entity neighbor expanded into equivalent selector-free ipAddresses neighbors +// (appended to the same direction), so the projection's existing address +// surface enforces them. It is a no-op — returning cp unchanged — when the +// lister is nil or nothing resolves, and it never mutates the input: a copy is +// made only when there is something to add. Call it immediately before +// projecting a ContainerProfile. +func WithResolvedServiceNeighbors(cp *v1beta1.ContainerProfile, l Lister) *v1beta1.ContainerProfile { + if cp == nil || l == nil { + return cp + } + egExtra := ExpandServiceNeighbors(cp.Spec.Egress, l) + inExtra := ExpandServiceNeighbors(cp.Spec.Ingress, l) + if len(egExtra) == 0 && len(inExtra) == 0 { + return cp + } + out := cp.DeepCopy() + out.Spec.Egress = append(out.Spec.Egress, egExtra...) + out.Spec.Ingress = append(out.Spec.Ingress, inExtra...) + return out +} + +// HasServiceNeighbors reports whether any egress/ingress neighbor declares a +// serviceRef / serviceSelector / entity — i.e. whether this profile's +// projection depends on the live cluster view (Service/EndpointSlice/Node) and +// must be re-projected when that view changes. Keyed on the raw fields, not on +// whether they currently resolve, so a profile projected before the informers +// synced is still marked and re-projects once they do. +func HasServiceNeighbors(cp *v1beta1.ContainerProfile) bool { + if cp == nil { + return false + } + for i := range cp.Spec.Egress { + if hasServiceFields(&cp.Spec.Egress[i]) { + return true + } + } + for i := range cp.Spec.Ingress { + if hasServiceFields(&cp.Spec.Ingress[i]) { + return true + } + } + return false +} + +func hasServiceFields(n *v1beta1.NetworkNeighbor) bool { + return n.ServiceRefNamespace != "" || n.ServiceRefName != "" || n.ServiceSelector != nil || n.Entity != "" +} + +// specFromNeighbor extracts a PeerSpec from a NetworkNeighbor, reporting false +// if the neighbor declares none of the service/entity selectors (a plain +// ipAddresses / dnsNames / podSelector neighbor is left untouched). +func specFromNeighbor(n *v1beta1.NetworkNeighbor) (PeerSpec, bool) { + spec := PeerSpec{Ports: portsFromNeighbor(n.Ports)} + switch { + case n.Entity != "": + spec.Entity = n.Entity + case n.ServiceRefName != "": + spec.ServiceRef = &ServiceRef{Namespace: n.ServiceRefNamespace, Name: n.ServiceRefName} + case n.ServiceSelector != nil: + // Only equality (matchLabels) is honored. A MatchExpressions clause or + // an empty matchLabels would either be silently ignored (broadening the + // match) or resolve to every Service — fail closed instead. + if len(n.ServiceSelector.MatchExpressions) > 0 || len(n.ServiceSelector.MatchLabels) == 0 { + return PeerSpec{}, false + } + spec.ServiceSelector = n.ServiceSelector.MatchLabels + // A namespaceSelector is honored only as the single equality + // kubernetes.io/metadata.name= (the only key the lister scopes on). + // Any other form — MatchExpressions, extra keys, or a different key — + // would be silently dropped and broaden the match cluster-wide, so fail + // closed. A nil namespaceSelector is cluster-wide by design. + if n.NamespaceSelector != nil { + nsl := n.NamespaceSelector + if len(nsl.MatchExpressions) > 0 || len(nsl.MatchLabels) != 1 || + nsl.MatchLabels["kubernetes.io/metadata.name"] == "" { + return PeerSpec{}, false + } + spec.NamespaceLabels = nsl.MatchLabels + } + default: + return PeerSpec{}, false + } + return spec, true +} + +func portsFromNeighbor(ports []v1beta1.NetworkPort) []PortProto { + if len(ports) == 0 { + return nil + } + out := make([]PortProto, 0, len(ports)) + for i := range ports { + p := &ports[i] + var port int32 + if p.Port != nil { + port = *p.Port + } + out = append(out, PortProto{Port: port, Protocol: strings.ToUpper(string(p.Protocol))}) + } + return out +} diff --git a/pkg/networkpeer/expand_test.go b/pkg/networkpeer/expand_test.go new file mode 100644 index 0000000000..dfb397b879 --- /dev/null +++ b/pkg/networkpeer/expand_test.go @@ -0,0 +1,209 @@ +package networkpeer + +import ( + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/kubescape/storage/pkg/apis/softwarecomposition/v1beta1" +) + +func port(name string, p int32) v1beta1.NetworkPort { + return v1beta1.NetworkPort{Name: name, Protocol: "TCP", Port: &p} +} + +// TestExpandServiceNeighbors_Egress: a serviceRef (alertmanager) + a host +// entity neighbor expand into selector-free ipAddresses neighbors carrying the +// resolved IPs and the original ports; a plain ipAddresses neighbor and an +// unresolvable serviceRef contribute nothing. +func TestExpandServiceNeighbors_Egress(t *testing.T) { + l := realFluxTopology() + in := []v1beta1.NetworkNeighbor{ + {Identifier: "am", Type: "internal", ServiceRefNamespace: "honey", ServiceRefName: "alertmanager", Ports: []v1beta1.NetworkPort{port("TCP-9093", 9093)}}, + {Identifier: "plain", Type: "internal", IPAddresses: []string{"10.43.0.0/16"}, Ports: []v1beta1.NetworkPort{port("TCP-443", 443)}}, + {Identifier: "ghost", Type: "internal", ServiceRefNamespace: "honey", ServiceRefName: "missing", Ports: []v1beta1.NetworkPort{port("TCP-1", 1)}}, + } + out := ExpandServiceNeighbors(in, l) + + if len(out) != 1 { + t.Fatalf("expected 1 synthesized neighbor (alertmanager only), got %d", len(out)) + } + got := out[0] + if got.Identifier != "am-resolved" { + t.Errorf("identifier: got %q", got.Identifier) + } + // ClusterIP + both endpoints, port carried over. + wantIPs := map[string]bool{"10.43.54.190": false, "10.42.0.44": false, "10.42.0.84": false} + for _, ip := range got.IPAddresses { + if _, ok := wantIPs[ip]; !ok { + t.Errorf("unexpected resolved IP %s", ip) + } + wantIPs[ip] = true + } + for ip, seen := range wantIPs { + if !seen { + t.Errorf("missing resolved IP %s", ip) + } + } + if len(got.Ports) != 1 || got.Ports[0].Port == nil || *got.Ports[0].Port != 9093 { + t.Errorf("ports not carried over: %+v", got.Ports) + } +} + +// TestExpandServiceNeighbors_HostEntity: fromEntity host resolves to node + +// gateway IPs on the health port. +func TestExpandServiceNeighbors_HostEntity(t *testing.T) { + l := realFluxTopology() + in := []v1beta1.NetworkNeighbor{ + {Identifier: "probes", Type: "internal", Entity: "host", Ports: []v1beta1.NetworkPort{port("TCP-9440", 9440)}}, + } + out := ExpandServiceNeighbors(in, l) + if len(out) != 1 { + t.Fatalf("expected 1 synthesized host neighbor, got %d", len(out)) + } + // Feed the synthesized entry through the address matcher the same way the + // projection would, to prove end-to-end intent. + tuples := Resolve(PeerSpec{Entity: "host", Ports: []PortProto{{Port: 9440, Protocol: "TCP"}}}, l) + if !Matches(tuples, "10.42.0.1", 9440, "TCP") { + t.Errorf("gateway kubelet probe should match") + } + if Matches(tuples, "10.42.0.1", 9090, "TCP") { + t.Errorf("wrong port must not match") + } +} + +// TestExpandServiceNeighbors_Selector: serviceSelector expands across all +// matching Services in the scoped namespace. +func TestExpandServiceNeighbors_Selector(t *testing.T) { + l := realFluxTopology() + in := []v1beta1.NetworkNeighbor{{ + Identifier: "guestbooks", + Type: "internal", + ServiceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "guestbook"}}, + NamespaceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"kubernetes.io/metadata.name": "gitops-demo"}}, + Ports: []v1beta1.NetworkPort{port("TCP-80", 80)}, + }} + out := ExpandServiceNeighbors(in, l) + if len(out) != 1 { + t.Fatalf("expected 1 synthesized neighbor, got %d", len(out)) + } + if len(out[0].IPAddresses) != 2 { + t.Errorf("expected 2 guestbook ClusterIPs, got %v", out[0].IPAddresses) + } +} + +// TestExpandServiceNeighbors_NilLister: no cluster view, no expansion. +func TestExpandServiceNeighbors_NilLister(t *testing.T) { + in := []v1beta1.NetworkNeighbor{{Identifier: "am", Entity: "host"}} + if out := ExpandServiceNeighbors(in, nil); out != nil { + t.Errorf("nil lister must expand to nil, got %v", out) + } +} + +// TestExpandServiceNeighbors_SelectorFailClosed: a serviceSelector carrying +// MatchExpressions (unsupported) or an empty matchLabels must NOT broaden the +// allowlist — it resolves to nothing. +func TestExpandServiceNeighbors_SelectorFailClosed(t *testing.T) { + l := realFluxTopology() + cases := []v1beta1.NetworkNeighbor{ + {Identifier: "expr", ServiceSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"app": "guestbook"}, + MatchExpressions: []metav1.LabelSelectorRequirement{{Key: "tier", Operator: metav1.LabelSelectorOpExists}}, + }, Ports: []v1beta1.NetworkPort{port("TCP-80", 80)}}, + {Identifier: "empty", ServiceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{}}, Ports: []v1beta1.NetworkPort{port("TCP-80", 80)}}, + } + for _, n := range cases { + if out := ExpandServiceNeighbors([]v1beta1.NetworkNeighbor{n}, l); len(out) != 0 { + t.Errorf("%s: selector must fail closed, got %d", n.Identifier, len(out)) + } + } +} + +// TestWithResolvedServiceNeighbors: the CP-level wrapper appends resolved +// neighbors without mutating the input, and is a no-op when nothing resolves. +func TestWithResolvedServiceNeighbors(t *testing.T) { + l := realFluxTopology() + cp := &v1beta1.ContainerProfile{} + cp.Spec.Egress = []v1beta1.NetworkNeighbor{ + {Identifier: "am", ServiceRefNamespace: "honey", ServiceRefName: "alertmanager", Ports: []v1beta1.NetworkPort{port("TCP-9093", 9093)}}, + } + cp.Spec.Ingress = []v1beta1.NetworkNeighbor{ + {Identifier: "probes", Entity: "host", Ports: []v1beta1.NetworkPort{port("TCP-9440", 9440)}}, + } + out := WithResolvedServiceNeighbors(cp, l) + + if len(cp.Spec.Egress) != 1 || len(cp.Spec.Ingress) != 1 { + t.Fatalf("input CP must not be mutated: eg=%d in=%d", len(cp.Spec.Egress), len(cp.Spec.Ingress)) + } + if len(out.Spec.Egress) != 2 { + t.Errorf("egress: want original + 1 resolved, got %d", len(out.Spec.Egress)) + } + if len(out.Spec.Ingress) != 2 { + t.Errorf("ingress: want original + 1 resolved, got %d", len(out.Spec.Ingress)) + } + + // No-op cases. + if got := WithResolvedServiceNeighbors(cp, nil); got != cp { + t.Error("nil lister must return the same CP unchanged") + } + plain := &v1beta1.ContainerProfile{} + plain.Spec.Egress = []v1beta1.NetworkNeighbor{{Identifier: "ip", IPAddresses: []string{"10.0.0.0/8"}}} + if got := WithResolvedServiceNeighbors(plain, l); got != plain { + t.Error("a CP with no service/entity neighbors must return unchanged (same pointer)") + } +} + +// TestExpandServiceNeighbors_NamespaceSelectorFailClosed: a namespaceSelector +// is honored only as the single equality kubernetes.io/metadata.name=; any +// other form must fail closed rather than silently broaden cluster-wide. +func TestExpandServiceNeighbors_NamespaceSelectorFailClosed(t *testing.T) { + l := realFluxTopology() + withNS := func(nsSel *metav1.LabelSelector) v1beta1.NetworkNeighbor { + return v1beta1.NetworkNeighbor{ + Identifier: "svc", + ServiceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "guestbook"}}, + NamespaceSelector: nsSel, + Ports: []v1beta1.NetworkPort{port("TCP-80", 80)}, + } + } + bad := []*metav1.LabelSelector{ + {MatchExpressions: []metav1.LabelSelectorRequirement{{Key: "kubernetes.io/metadata.name", Operator: metav1.LabelSelectorOpExists}}}, + {MatchLabels: map[string]string{"env": "prod"}}, // wrong key + {MatchLabels: map[string]string{"kubernetes.io/metadata.name": "gitops-demo", "x": "y"}}, // extra key + {MatchLabels: map[string]string{}}, // empty + } + for i, ns := range bad { + if out := ExpandServiceNeighbors([]v1beta1.NetworkNeighbor{withNS(ns)}, l); len(out) != 0 { + t.Errorf("bad namespaceSelector[%d] must fail closed, got %d", i, len(out)) + } + } + good := withNS(&metav1.LabelSelector{MatchLabels: map[string]string{"kubernetes.io/metadata.name": "gitops-demo"}}) + if out := ExpandServiceNeighbors([]v1beta1.NetworkNeighbor{good}, l); len(out) != 1 { + t.Errorf("metadata.name namespaceSelector should resolve, got %d", len(out)) + } +} + +// TestHasServiceNeighbors: a profile with a serviceRef/serviceSelector/entity +// neighbor is flagged as depending on the live cluster view; a plain +// ipAddresses/dnsNames profile is not. +func TestHasServiceNeighbors(t *testing.T) { + if HasServiceNeighbors(nil) { + t.Error("nil CP must be false") + } + plain := &v1beta1.ContainerProfile{} + plain.Spec.Egress = []v1beta1.NetworkNeighbor{{Identifier: "ip", IPAddresses: []string{"10.0.0.0/8"}}} + if HasServiceNeighbors(plain) { + t.Error("plain ipAddresses profile must not use service resolution") + } + for _, n := range []v1beta1.NetworkNeighbor{ + {ServiceRefName: "alertmanager", ServiceRefNamespace: "honey"}, + {ServiceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "x"}}}, + {Entity: "host"}, + } { + cp := &v1beta1.ContainerProfile{} + cp.Spec.Ingress = []v1beta1.NetworkNeighbor{n} + if !HasServiceNeighbors(cp) { + t.Errorf("neighbor %+v should be flagged", n) + } + } +} diff --git a/pkg/networkpeer/lister.go b/pkg/networkpeer/lister.go new file mode 100644 index 0000000000..eff6d349db --- /dev/null +++ b/pkg/networkpeer/lister.go @@ -0,0 +1,167 @@ +package networkpeer + +import ( + "net" + "sync/atomic" + + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + "k8s.io/apimachinery/pkg/labels" + corelisters "k8s.io/client-go/listers/core/v1" + discoverylisters "k8s.io/client-go/listers/discovery/v1" +) + +// InformerLister is the production Lister, backed by Service / EndpointSlice / +// Node informer listers. It resolves a serviceRef/serviceSelector to the +// Service's ClusterIP(s) ∪ its EndpointSlice addresses, and the "host" entity +// to every node's InternalIP(s) plus the CNI gateway derived from its PodCIDR. +type InformerLister struct { + services corelisters.ServiceLister + slices discoverylisters.EndpointSliceLister + nodes corelisters.NodeLister + // nodeName scopes the "host" entity to the local node. Empty means every + // node (used by tests); production passes the agent's own node so kubelet + // probes from this node's gateway match without broadening "host" to the + // whole cluster. + nodeName string + // generation advances on every observed Service/EndpointSlice/Node change + // (bumped from informer event handlers wired in cmd/main.go). + generation atomic.Int64 +} + +// Generation returns the current cluster-view generation. +func (l *InformerLister) Generation() int64 { return l.generation.Load() } + +// Bump advances the generation; wire it to the informer event handlers. +func (l *InformerLister) Bump() { l.generation.Add(1) } + +func NewInformerLister(services corelisters.ServiceLister, slices discoverylisters.EndpointSliceLister, nodes corelisters.NodeLister, nodeName string) *InformerLister { + return &InformerLister{services: services, slices: slices, nodes: nodes, nodeName: nodeName} +} + +var _ Lister = (*InformerLister)(nil) + +func (l *InformerLister) ServiceByName(namespace, name string) (*ServiceInfo, bool) { + svc, err := l.services.Services(namespace).Get(name) + if err != nil { + return nil, false + } + return l.serviceInfo(svc), true +} + +func (l *InformerLister) ServicesByLabels(serviceSelector, namespaceLabels map[string]string) []*ServiceInfo { + // Never resolve an empty selector to labels.Everything() — that would + // allowlist every Service in the cluster. Fail closed. + if len(serviceSelector) == 0 { + return nil + } + svcs, err := l.services.List(labels.SelectorFromSet(serviceSelector)) + if err != nil { + return nil + } + wantNS := "" + if namespaceLabels != nil { + wantNS = namespaceLabels["kubernetes.io/metadata.name"] + } + var out []*ServiceInfo + for _, svc := range svcs { + if wantNS != "" && svc.Namespace != wantNS { + continue + } + out = append(out, l.serviceInfo(svc)) + } + return out +} + +func (l *InformerLister) HostIPs() []string { + nodes, err := l.nodes.List(labels.Everything()) + if err != nil { + return nil + } + var ips []string + for _, n := range nodes { + if l.nodeName != "" && n.Name != l.nodeName { + continue + } + for _, addr := range n.Status.Addresses { + if addr.Type == corev1.NodeInternalIP { + ips = append(ips, addr.Address) + } + } + for _, cidr := range podCIDRs(n) { + if gw := gatewayIP(cidr); gw != "" { + ips = append(ips, gw) + } + } + } + return dedupe(ips) +} + +func (l *InformerLister) serviceInfo(svc *corev1.Service) *ServiceInfo { + info := &ServiceInfo{Namespace: svc.Namespace, Name: svc.Name, Labels: svc.Labels} + for _, ip := range svc.Spec.ClusterIPs { + if ip != "" && ip != corev1.ClusterIPNone { + info.ClusterIPs = append(info.ClusterIPs, ip) + } + } + if len(info.ClusterIPs) == 0 && svc.Spec.ClusterIP != "" && svc.Spec.ClusterIP != corev1.ClusterIPNone { + info.ClusterIPs = append(info.ClusterIPs, svc.Spec.ClusterIP) + } + info.EndpointIPs = l.endpointIPs(svc.Namespace, svc.Name) + return info +} + +func (l *InformerLister) endpointIPs(namespace, service string) []string { + sel := labels.SelectorFromSet(labels.Set{discoveryv1.LabelServiceName: service}) + slices, err := l.slices.EndpointSlices(namespace).List(sel) + if err != nil { + return nil + } + var ips []string + for _, es := range slices { + for i := range es.Endpoints { + ips = append(ips, es.Endpoints[i].Addresses...) + } + } + return dedupe(ips) +} + +func podCIDRs(n *corev1.Node) []string { + if len(n.Spec.PodCIDRs) > 0 { + return n.Spec.PodCIDRs + } + if n.Spec.PodCIDR != "" { + return []string{n.Spec.PodCIDR} + } + return nil +} + +// gatewayIP returns the conventional CNI gateway for a pod CIDR: the network +// address + 1 (e.g. 10.42.0.0/24 -> 10.42.0.1). Masqueraded node-sourced +// traffic (kubelet health probes) appears from this address. IPv6 CIDRs yield +// no gateway (the .1 convention is IPv4). +func gatewayIP(cidr string) string { + _, ipNet, err := net.ParseCIDR(cidr) + if err != nil { + return "" + } + ip := ipNet.IP.To4() + if ip == nil { + return "" + } + gw := make(net.IP, len(ip)) + copy(gw, ip) + for i := len(gw) - 1; i >= 0; i-- { + gw[i]++ + if gw[i] != 0 { + break + } + } + // A /31 or /32 (or a network address ending in .255 that overflows) yields a + // gateway outside the CIDR — never allowlist an IP the pod network doesn't + // actually contain. + if !ipNet.Contains(gw) { + return "" + } + return gw.String() +} diff --git a/pkg/networkpeer/lister_test.go b/pkg/networkpeer/lister_test.go new file mode 100644 index 0000000000..98d0143b10 --- /dev/null +++ b/pkg/networkpeer/lister_test.go @@ -0,0 +1,147 @@ +package networkpeer + +import ( + "testing" + + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes/fake" +) + +// newTestLister builds an InformerLister over a fake cluster seeded from the +// real flux/kubescape topology, exercising the production Service/EndpointSlice/ +// Node -> Lister path (not the hand-written fake used by the resolver tests). +func newTestLister(t *testing.T) *InformerLister { + t.Helper() + client := fake.NewClientset( + &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{Namespace: "honey", Name: "alertmanager", Labels: map[string]string{"app": "alertmanager"}}, + Spec: corev1.ServiceSpec{ClusterIP: "10.43.54.190", ClusterIPs: []string{"10.43.54.190"}}, + }, + &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{Namespace: "honey", Name: "alertmanager-x1", Labels: map[string]string{discoveryv1.LabelServiceName: "alertmanager"}}, + Endpoints: []discoveryv1.Endpoint{{Addresses: []string{"10.42.0.44", "10.42.0.84"}}}, + }, + &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitops-demo", Name: "guestbook-ui", Labels: map[string]string{"app": "guestbook"}}, + Spec: corev1.ServiceSpec{ClusterIP: "10.43.111.192", ClusterIPs: []string{"10.43.111.192"}}, + }, + &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "tanzee"}, + Spec: corev1.NodeSpec{PodCIDR: "10.42.0.0/24", PodCIDRs: []string{"10.42.0.0/24"}}, + Status: corev1.NodeStatus{Addresses: []corev1.NodeAddress{{Type: corev1.NodeInternalIP, Address: "192.168.0.191"}}}, + }, + &corev1.Node{ // a second node whose IPs must NOT leak into this node's "host" + ObjectMeta: metav1.ObjectMeta{Name: "other"}, + Spec: corev1.NodeSpec{PodCIDR: "10.42.9.0/24", PodCIDRs: []string{"10.42.9.0/24"}}, + Status: corev1.NodeStatus{Addresses: []corev1.NodeAddress{{Type: corev1.NodeInternalIP, Address: "192.168.0.99"}}}, + }, + ) + factory := informers.NewSharedInformerFactory(client, 0) + l := NewInformerLister( + factory.Core().V1().Services().Lister(), + factory.Discovery().V1().EndpointSlices().Lister(), + factory.Core().V1().Nodes().Lister(), + "tanzee", + ) + stop := make(chan struct{}) + t.Cleanup(func() { close(stop) }) + factory.Start(stop) + factory.WaitForCacheSync(stop) + return l +} + +func TestInformerLister_ServiceByName(t *testing.T) { + l := newTestLister(t) + svc, ok := l.ServiceByName("honey", "alertmanager") + if !ok { + t.Fatal("alertmanager Service should resolve") + } + if len(svc.ClusterIPs) != 1 || svc.ClusterIPs[0] != "10.43.54.190" { + t.Errorf("ClusterIPs: got %v", svc.ClusterIPs) + } + if len(svc.EndpointIPs) != 2 { + t.Errorf("EndpointIPs: got %v (want 2 from the EndpointSlice)", svc.EndpointIPs) + } + // End-to-end through the resolver, like the projection will. + tuples := Resolve(PeerSpec{ServiceRef: &ServiceRef{"honey", "alertmanager"}, Ports: []PortProto{{Port: 9093, Protocol: "TCP"}}}, l) + if !Matches(tuples, "10.43.54.190", 9093, "TCP") || !Matches(tuples, "10.42.0.44", 9093, "TCP") { + t.Errorf("resolver over informer lister should match ClusterIP + endpoints") + } + if _, ok := l.ServiceByName("honey", "nope"); ok { + t.Error("unknown Service must not resolve") + } +} + +func TestInformerLister_HostIPs(t *testing.T) { + l := newTestLister(t) + ips := l.HostIPs() + want := map[string]bool{"192.168.0.191": false, "10.42.0.1": false} + for _, ip := range ips { + if _, ok := want[ip]; ok { + want[ip] = true + } + } + for ip, seen := range want { + if !seen { + t.Errorf("HostIPs missing %s (got %v)", ip, ips) + } + } + for _, ip := range ips { + if ip == "192.168.0.99" || ip == "10.42.9.1" { + t.Errorf("HostIPs must be scoped to the local node; leaked other node's %s", ip) + } + } +} + +// TestInformerLister_EmptySelectorFailsClosed: an empty serviceSelector must +// NOT resolve to every Service in the cluster. +func TestInformerLister_EmptySelectorFailsClosed(t *testing.T) { + l := newTestLister(t) + if got := l.ServicesByLabels(map[string]string{}, nil); len(got) != 0 { + t.Errorf("empty selector must fail closed, got %d services", len(got)) + } + if got := Resolve(PeerSpec{ServiceSelector: map[string]string{}, Ports: tcp(80)}, l); len(got) != 0 { + t.Errorf("Resolve with empty selector must yield nothing, got %v", got) + } +} + +func TestInformerLister_ServicesByLabels(t *testing.T) { + l := newTestLister(t) + svcs := l.ServicesByLabels(map[string]string{"app": "guestbook"}, map[string]string{"kubernetes.io/metadata.name": "gitops-demo"}) + if len(svcs) != 1 || svcs[0].Name != "guestbook-ui" { + t.Fatalf("expected guestbook-ui, got %v", svcs) + } + // Namespace scoping excludes a same-label service elsewhere (none here) and + // a wrong-namespace filter yields nothing. + if got := l.ServicesByLabels(map[string]string{"app": "guestbook"}, map[string]string{"kubernetes.io/metadata.name": "other"}); len(got) != 0 { + t.Errorf("namespace filter should exclude, got %v", got) + } +} + +func TestGatewayIP(t *testing.T) { + cases := map[string]string{ + "10.42.0.0/24": "10.42.0.1", + "10.244.5.0/24": "10.244.5.1", + "2001:db8::/64": "", + "10.42.0.5/32": "", // /32 host: incremented gateway is outside the CIDR + "255.255.255.255/32": "", // overflow to 0.0.0.0, out of CIDR + } + for cidr, want := range cases { + if got := gatewayIP(cidr); got != want { + t.Errorf("gatewayIP(%s)=%q want %q", cidr, got, want) + } + } +} + +func TestInformerLister_Generation(t *testing.T) { + l := newTestLister(t) + g0 := l.Generation() + l.Bump() + l.Bump() + if l.Generation() != g0+2 { + t.Errorf("Generation must advance on Bump: got %d want %d", l.Generation(), g0+2) + } +} diff --git a/pkg/networkpeer/resolve.go b/pkg/networkpeer/resolve.go new file mode 100644 index 0000000000..0467c14b50 --- /dev/null +++ b/pkg/networkpeer/resolve.go @@ -0,0 +1,197 @@ +// Package networkpeer resolves Kubernetes-native network-neighbor selectors +// (a Service reference, a Service label selector, or a reserved entity such as +// "host") into the concrete (IP, port, protocol) tuples an egress/ingress +// allowlist should match. +// +// It exists so a ContainerProfile can express cluster-infrastructure peers — +// Service ClusterIPs (alertmanager, kube-apiserver via default/kubernetes, +// kube-dns, ...) and host/kubelet traffic — portably (by name, resolved +// per-cluster) and narrowly (only the referenced Service/entity), instead of a +// broad ipAddresses CIDR over the whole service network. See +// k8sstormcenter/node-agent#92. +// +// Resolution is intentionally decoupled from live matching: callers resolve a +// PeerSpec to []AllowTuple once (e.g. at projection time) via a Lister backed +// by Service/EndpointSlice/Node informers, then match observed connections +// against the tuples with Matches. The Lister interface keeps the resolver +// unit-testable against a fake cluster view. +package networkpeer + +import "strings" + +// EntityHost is the reserved entity naming the local node: its InternalIP(s) +// and the CNI gateway address. It is the one peer class no Service object can +// represent (kubelet health probes, node-sourced / masqueraded traffic). +const EntityHost = "host" + +// PortProto is a single allowed destination port/protocol. Protocol is +// upper-case ("TCP"/"UDP"); an empty Protocol matches any protocol. +type PortProto struct { + Port int32 + Protocol string +} + +// ServiceRef names a single Service by namespace and name. +type ServiceRef struct { + Namespace string + Name string +} + +// PeerSpec is the storage-agnostic form of one serviceRef / serviceSelector / +// entity network-neighbor entry. Exactly one of ServiceRef, ServiceSelector, +// or Entity is expected to be set; Ports scopes the resolved tuples. +type PeerSpec struct { + ServiceRef *ServiceRef + ServiceSelector map[string]string + NamespaceLabels map[string]string + Entity string + Ports []PortProto +} + +// ServiceInfo is the resolver's view of one Service. +type ServiceInfo struct { + Namespace string + Name string + Labels map[string]string + ClusterIPs []string + EndpointIPs []string +} + +// Lister is the read-only cluster view the resolver needs. Production wires it +// to Service/EndpointSlice/Node informer listers; tests use a fake. +type Lister interface { + ServiceByName(namespace, name string) (*ServiceInfo, bool) + ServicesByLabels(serviceSelector, namespaceLabels map[string]string) []*ServiceInfo + HostIPs() []string + // Generation increments whenever the underlying cluster view changes (any + // Service/EndpointSlice/Node event). Callers store it alongside a projected + // profile and re-project when it advances, so resolved IPs don't go stale on + // endpoint churn or caches that filled after projection. + Generation() int64 +} + +// AllowTuple is one concrete (IP, port, protocol) a resolved PeerSpec permits. +type AllowTuple struct { + IP string + Port int32 + Protocol string +} + +// Resolve expands spec into the concrete tuples it authorises, using l for the +// current cluster view. A spec with no resolvable target (unknown Service, +// selector matching nothing, unknown entity) yields no tuples — never a +// match-all. A spec with no Ports yields one tuple per IP with Port 0 / +// Protocol "" (any-port), so callers that ignore ports still work; callers +// that enforce ports should treat Port 0 as "unspecified". +func Resolve(spec PeerSpec, l Lister) []AllowTuple { + if l == nil { + return nil + } + ips := resolveIPs(spec, l) + if len(ips) == 0 { + return nil + } + return expand(ips, spec.Ports) +} + +// ResolveIPs returns just the IPs a spec resolves to, ignoring ports. Used by +// the projection-time expansion, which pairs them with the neighbor's own +// ports. +func ResolveIPs(spec PeerSpec, l Lister) []string { + if l == nil { + return nil + } + return resolveIPs(spec, l) +} + +func resolveIPs(spec PeerSpec, l Lister) []string { + switch { + case spec.Entity != "": + if strings.EqualFold(spec.Entity, EntityHost) { + return dedupe(l.HostIPs()) + } + return nil + case spec.ServiceRef != nil: + svc, ok := l.ServiceByName(spec.ServiceRef.Namespace, spec.ServiceRef.Name) + if !ok || svc == nil { + return nil + } + return serviceIPs(svc) + case spec.ServiceSelector != nil: + // An empty selector is NOT a cluster-wide match-all: fail closed. + if len(spec.ServiceSelector) == 0 { + return nil + } + var ips []string + for _, svc := range l.ServicesByLabels(spec.ServiceSelector, spec.NamespaceLabels) { + ips = append(ips, serviceIPs(svc)...) + } + return dedupe(ips) + default: + return nil + } +} + +func serviceIPs(svc *ServiceInfo) []string { + out := make([]string, 0, len(svc.ClusterIPs)+len(svc.EndpointIPs)) + out = append(out, svc.ClusterIPs...) + out = append(out, svc.EndpointIPs...) + return dedupe(out) +} + +func expand(ips []string, ports []PortProto) []AllowTuple { + if len(ports) == 0 { + out := make([]AllowTuple, 0, len(ips)) + for _, ip := range ips { + out = append(out, AllowTuple{IP: ip}) + } + return out + } + out := make([]AllowTuple, 0, len(ips)*len(ports)) + for _, ip := range ips { + for _, p := range ports { + out = append(out, AllowTuple{IP: ip, Port: p.Port, Protocol: strings.ToUpper(p.Protocol)}) + } + } + return out +} + +// Matches reports whether the observed (ip, port, protocol) connection is +// permitted by any tuple. Matching is port-sensitive: a tuple with Port 0 +// (any-port) matches any observed port; otherwise the port must be equal. An +// empty tuple Protocol matches any protocol. +func Matches(tuples []AllowTuple, ip string, port int32, protocol string) bool { + protocol = strings.ToUpper(protocol) + for _, t := range tuples { + if t.IP != ip { + continue + } + if t.Port != 0 && t.Port != port { + continue + } + if t.Protocol != "" && t.Protocol != protocol { + continue + } + return true + } + return false +} + +func dedupe(in []string) []string { + if len(in) == 0 { + return nil + } + seen := make(map[string]struct{}, len(in)) + out := make([]string, 0, len(in)) + for _, s := range in { + if s == "" { + continue + } + if _, ok := seen[s]; ok { + continue + } + seen[s] = struct{}{} + out = append(out, s) + } + return out +} diff --git a/pkg/networkpeer/resolve_test.go b/pkg/networkpeer/resolve_test.go new file mode 100644 index 0000000000..f131f774fe --- /dev/null +++ b/pkg/networkpeer/resolve_test.go @@ -0,0 +1,204 @@ +package networkpeer + +import ( + "sort" + "testing" +) + +// fakeLister is a static cluster view seeded from the real Flux/kubescape +// topology observed on the k3s dev cluster (issue #92). It lets the resolver +// tests assert against genuine ClusterIPs/endpoints without a live cluster. +type fakeLister struct { + services map[string]*ServiceInfo // key "ns/name" + hostIPs []string +} + +func (f *fakeLister) ServiceByName(ns, name string) (*ServiceInfo, bool) { + s, ok := f.services[ns+"/"+name] + return s, ok +} + +func (f *fakeLister) ServicesByLabels(sel, nsLabels map[string]string) []*ServiceInfo { + var out []*ServiceInfo + for _, s := range f.services { + if nsLabels != nil { + if s.Labels["__ns__"] != nsLabels["kubernetes.io/metadata.name"] { + continue + } + } + if labelsSubset(sel, s.Labels) { + out = append(out, s) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) + return out +} + +func (f *fakeLister) HostIPs() []string { return f.hostIPs } + +func (f *fakeLister) Generation() int64 { return 0 } + +func labelsSubset(want, have map[string]string) bool { + for k, v := range want { + if have[k] != v { + return false + } + } + return true +} + +// realFluxTopology mirrors what `kubectl get svc/endpoints` returned live. +func realFluxTopology() *fakeLister { + return &fakeLister{ + hostIPs: []string{"192.168.0.191", "10.42.0.1"}, // node InternalIP + CNI gateway + services: map[string]*ServiceInfo{ + "honey/alertmanager": { + Namespace: "honey", Name: "alertmanager", + Labels: map[string]string{"app": "alertmanager", "__ns__": "honey"}, + ClusterIPs: []string{"10.43.54.190"}, + EndpointIPs: []string{"10.42.0.44", "10.42.0.84"}, + }, + "honey/storage": { + Namespace: "honey", Name: "storage", + Labels: map[string]string{"app": "storage", "__ns__": "honey"}, + ClusterIPs: []string{"10.43.70.156"}, + }, + "default/kubernetes": { // k3s: apiserver endpoint is the node IP (Kind: Host) + Namespace: "default", Name: "kubernetes", + Labels: map[string]string{"__ns__": "default"}, + ClusterIPs: []string{"10.43.0.1"}, + EndpointIPs: []string{"192.168.0.191"}, + }, + "argocd/argocd-server": { + Namespace: "argocd", Name: "argocd-server", + Labels: map[string]string{"app.kubernetes.io/name": "argocd-server", "__ns__": "argocd"}, + ClusterIPs: []string{"10.43.173.14"}, + }, + "gitops-demo/guestbook-ui": { + Namespace: "gitops-demo", Name: "guestbook-ui", + Labels: map[string]string{"app": "guestbook", "__ns__": "gitops-demo"}, + ClusterIPs: []string{"10.43.111.192"}, + }, + "gitops-demo/helm-guestbook": { + Namespace: "gitops-demo", Name: "helm-guestbook", + Labels: map[string]string{"app": "guestbook", "__ns__": "gitops-demo"}, + ClusterIPs: []string{"10.43.3.63"}, + }, + }, + } +} + +func tcp(port int32) []PortProto { return []PortProto{{Port: port, Protocol: "TCP"}} } + +// Test A1 — serviceRef egress (alertmanager) is narrow AND port-sensitive: +// matches its ClusterIP and endpoints on 9093 only; a sibling service on the +// same port stays visible to R0011 (the whole point vs a /16). +func TestServiceRef_Alertmanager(t *testing.T) { + l := realFluxTopology() + tuples := Resolve(PeerSpec{ServiceRef: &ServiceRef{"honey", "alertmanager"}, Ports: tcp(9093)}, l) + + cases := []struct { + ip string + port int32 + proto string + want bool + why string + }{ + {"10.43.54.190", 9093, "TCP", true, "ClusterIP + port match"}, + {"10.42.0.44", 9093, "TCP", true, "backing endpoint IP"}, + {"10.42.0.84", 9093, "TCP", true, "backing endpoint IP"}, + {"10.43.54.190", 8080, "TCP", false, "wrong port (port-sensitive)"}, + {"10.43.54.190", 9093, "UDP", false, "wrong protocol"}, + {"10.43.173.14", 9093, "TCP", false, "argocd-server — different service, detection preserved"}, + {"10.43.70.156", 9093, "TCP", false, "storage — different service, detection preserved"}, + } + for _, c := range cases { + if got := Matches(tuples, c.ip, c.port, c.proto); got != c.want { + t.Errorf("Matches(%s:%d/%s)=%v want %v (%s)", c.ip, c.port, c.proto, got, c.want, c.why) + } + } +} + +// Test A2 — kube-apiserver needs no dedicated entity: toService default/kubernetes +// resolves to the ClusterIP AND the node-IP endpoint (k3s embeds the apiserver). +func TestServiceRef_KubeApiserver(t *testing.T) { + l := realFluxTopology() + tuples := Resolve(PeerSpec{ServiceRef: &ServiceRef{"default", "kubernetes"}, Ports: tcp(443)}, l) + for _, ip := range []string{"10.43.0.1", "192.168.0.191"} { + if !Matches(tuples, ip, 443, "TCP") { + t.Errorf("apiserver egress %s:443 should match via default/kubernetes", ip) + } + } + if Matches(tuples, "10.43.0.1", 6443, "TCP") { + t.Errorf("apiserver :6443 must not match (port 443 only)") + } +} + +// Test A3 — host entity: kubelet probe from the node/gateway matches on the +// health port only; a pod-CIDR source or wrong port does not. +func TestEntityHost(t *testing.T) { + l := realFluxTopology() + tuples := Resolve(PeerSpec{Entity: EntityHost, Ports: tcp(9440)}, l) + if !Matches(tuples, "10.42.0.1", 9440, "TCP") { + t.Errorf("kubelet probe 10.42.0.1:9440 should match fromEntity host") + } + if !Matches(tuples, "192.168.0.191", 9440, "TCP") { + t.Errorf("node InternalIP :9440 should match fromEntity host") + } + if Matches(tuples, "10.42.0.1", 9090, "TCP") { + t.Errorf("host :9090 must not match (port 9440 only)") + } + if Matches(tuples, "10.42.0.55", 9440, "TCP") { + t.Errorf("a pod IP must not match fromEntity host") + } +} + +// Test A4 — serviceSelector fans out across all matching Services (the two +// gitops-demo guestbook services share app=guestbook), scoped by namespace. +func TestServiceSelector_GuestbookFanout(t *testing.T) { + l := realFluxTopology() + tuples := Resolve(PeerSpec{ + ServiceSelector: map[string]string{"app": "guestbook"}, + NamespaceLabels: map[string]string{"kubernetes.io/metadata.name": "gitops-demo"}, + Ports: tcp(80), + }, l) + for _, ip := range []string{"10.43.111.192", "10.43.3.63"} { + if !Matches(tuples, ip, 80, "TCP") { + t.Errorf("guestbook service %s:80 should match app=guestbook selector", ip) + } + } + if Matches(tuples, "10.43.173.14", 80, "TCP") { + t.Errorf("argocd-server must not match app=guestbook selector") + } +} + +// Test A5 — no accidental match-all: unknown service, unknown entity, and a +// selector matching nothing all resolve to zero tuples. +func TestResolve_NoMatchAll(t *testing.T) { + l := realFluxTopology() + specs := []PeerSpec{ + {ServiceRef: &ServiceRef{"honey", "does-not-exist"}, Ports: tcp(443)}, + {Entity: "world", Ports: tcp(443)}, + {ServiceSelector: map[string]string{"app": "nope"}, Ports: tcp(443)}, + } + for i, s := range specs { + if tuples := Resolve(s, l); len(tuples) != 0 { + t.Errorf("spec[%d] should resolve to no tuples, got %d", i, len(tuples)) + } + } + if Matches(nil, "10.43.0.1", 443, "TCP") { + t.Errorf("nil tuples must never match") + } +} + +// Test A6 — a nil Lister and any-port (no Ports) behave safely. +func TestResolve_Edges(t *testing.T) { + if got := Resolve(PeerSpec{Entity: EntityHost}, nil); got != nil { + t.Errorf("nil lister must resolve to nil, got %v", got) + } + l := realFluxTopology() + anyPort := Resolve(PeerSpec{ServiceRef: &ServiceRef{"honey", "storage"}}, l) + if !Matches(anyPort, "10.43.70.156", 443, "TCP") || !Matches(anyPort, "10.43.70.156", 8443, "TCP") { + t.Errorf("a serviceRef with no Ports should match any observed port on its IP") + } +} diff --git a/pkg/objectcache/containerprofilecache/containerprofilecache.go b/pkg/objectcache/containerprofilecache/containerprofilecache.go index 2a5394d18e..410b39ec47 100644 --- a/pkg/objectcache/containerprofilecache/containerprofilecache.go +++ b/pkg/objectcache/containerprofilecache/containerprofilecache.go @@ -16,6 +16,7 @@ import ( helpersv1 "github.com/kubescape/k8s-interface/instanceidhandler/v1/helpers" "github.com/kubescape/node-agent/pkg/config" "github.com/kubescape/node-agent/pkg/metricsmanager" + "github.com/kubescape/node-agent/pkg/networkpeer" "github.com/kubescape/node-agent/pkg/objectcache" "github.com/kubescape/node-agent/pkg/objectcache/callstackcache" "github.com/kubescape/node-agent/pkg/resourcelocks" @@ -58,6 +59,15 @@ type CachedContainerProfile struct { State *objectcache.ProfileState CallStackTree *callstackcache.CallStackSearchTree + // UsesServiceResolution is true when the profile declares serviceRef/ + // serviceSelector/entity neighbors, so its projected addresses depend on the + // live cluster view. ListerGen is the service-lister generation captured at + // projection time; the reconciler re-projects such entries when the + // generation advances (endpoint churn, or caches that filled after the + // initial projection). Non-resolving profiles keep the RV/spec fast-skip. + UsesServiceResolution bool + ListerGen int64 + ContainerName string PodName string Namespace string @@ -111,6 +121,10 @@ type ContainerProfileCacheImpl struct { containerLocks *resourcelocks.ResourceLocks storageClient storage.ProfileClient k8sObjectCache objectcache.K8sObjectCache + // serviceLister resolves serviceRef/serviceSelector/entity network + // neighbors to concrete IPs at projection time. nil = feature off (the + // profile projects unchanged), which is what every unit test leaves it as. + serviceLister networkpeer.Lister metricsManager metricsmanager.MetricsManager reconcileEvery time.Duration @@ -174,6 +188,22 @@ func NewContainerProfileCache(cfg config.Config, storageClient storage.ProfileCl return c } +// SetServiceLister installs the cluster view used to resolve +// serviceRef/serviceSelector/entity network neighbors at projection time. It +// is optional: when unset the projection leaves such neighbors unresolved. +func (c *ContainerProfileCacheImpl) SetServiceLister(l networkpeer.Lister) { + c.serviceLister = l +} + +// listerGen returns the current service-lister generation, or 0 when no lister +// is installed (unit tests, or the feature is off). +func (c *ContainerProfileCacheImpl) listerGen() int64 { + if c.serviceLister == nil { + return 0 + } + return c.serviceLister.Generation() +} + // refreshRPC calls fn with a context bounded by c.rpcBudget, enforcing a // per-call SLO so a slow API server cannot stall a full reconciler burst. func (c *ContainerProfileCacheImpl) refreshRPC(ctx context.Context, fn func(context.Context) error) error { @@ -579,9 +609,14 @@ func (c *ContainerProfileCacheImpl) buildEntry( } entry.CallStackTree = tree - // Project under the current spec. + // Project under the current spec, resolving any serviceRef/entity network + // neighbors to concrete IPs first. Record whether this profile depends on + // the live cluster view and the lister generation it was resolved against, + // so the reconciler re-projects it when that view changes. spec := c.snapshotSpec() - projected := Apply(spec, userMerged, tree) + entry.UsesServiceResolution = networkpeer.HasServiceNeighbors(userMerged) + entry.ListerGen = c.listerGen() + projected := Apply(spec, networkpeer.WithResolvedServiceNeighbors(userMerged, c.serviceLister), tree) entry.Projected = projected entry.SpecHash = projected.SpecHash diff --git a/pkg/objectcache/containerprofilecache/reconciler.go b/pkg/objectcache/containerprofilecache/reconciler.go index bd7517c728..85c4943aaf 100644 --- a/pkg/objectcache/containerprofilecache/reconciler.go +++ b/pkg/objectcache/containerprofilecache/reconciler.go @@ -23,6 +23,7 @@ import ( "github.com/kubescape/go-logger" "github.com/kubescape/go-logger/helpers" helpersv1 "github.com/kubescape/k8s-interface/instanceidhandler/v1/helpers" + "github.com/kubescape/node-agent/pkg/networkpeer" "github.com/kubescape/node-agent/pkg/objectcache" "github.com/kubescape/node-agent/pkg/objectcache/callstackcache" "github.com/kubescape/node-agent/pkg/utils" @@ -418,9 +419,14 @@ func (c *ContainerProfileCacheImpl) refreshOneEntry(ctx context.Context, id stri if spec := c.snapshotSpec(); spec != nil { currentSpecHash = spec.Hash } + // serviceRef/entity profiles must also re-project when the cluster view + // changed since they were resolved (endpoint churn, or caches that filled + // after projection). Non-resolving profiles ignore the lister generation and + // keep the cheap RV/spec fast-skip. if rvsMatchCP(cp, e.RV) && rvsMatchCP(userDefinedCP, e.UserCPRV) && - e.SpecHash == currentSpecHash { + e.SpecHash == currentSpecHash && + (!e.UsesServiceResolution || e.ListerGen == c.listerGen()) { return } @@ -490,27 +496,29 @@ func (c *ContainerProfileCacheImpl) rebuildEntryFromSources( // Project under the current spec. spec := c.snapshotSpec() applyStart := time.Now() - projectedCP := Apply(spec, projected, tree) + projectedCP := Apply(spec, networkpeer.WithResolvedServiceNeighbors(projected, c.serviceLister), tree) if c.cfg.ProfileProjection.DetailedMetricsEnabled { c.metricsManager.ObserveProjectionApplyDuration(time.Since(applyStart)) c.observeMemoryMetrics(projected, projectedCP) } newEntry := &CachedContainerProfile{ - Projected: projectedCP, - SpecHash: projectedCP.SpecHash, - State: &objectcache.ProfileState{Completion: effectiveCP.Annotations[helpersv1.CompletionMetadataKey], Status: effectiveCP.Annotations[helpersv1.StatusMetadataKey], Name: effectiveCP.Name}, - CallStackTree: tree, - ContainerName: prev.ContainerName, - PodName: prev.PodName, - Namespace: prev.Namespace, - PodUID: podUID, - WorkloadID: prev.WorkloadID, - CPName: prev.CPName, - WorkloadName: prev.WorkloadName, - RV: rvOfCP(cp), - UserCPRV: rvOfCP(userDefinedCP), - terminatedSeenAt: prev.terminatedSeenAt, + Projected: projectedCP, + SpecHash: projectedCP.SpecHash, + UsesServiceResolution: networkpeer.HasServiceNeighbors(projected), + ListerGen: c.listerGen(), + State: &objectcache.ProfileState{Completion: effectiveCP.Annotations[helpersv1.CompletionMetadataKey], Status: effectiveCP.Annotations[helpersv1.StatusMetadataKey], Name: effectiveCP.Name}, + CallStackTree: tree, + ContainerName: prev.ContainerName, + PodName: prev.PodName, + Namespace: prev.Namespace, + PodUID: podUID, + WorkloadID: prev.WorkloadID, + CPName: prev.CPName, + WorkloadName: prev.WorkloadName, + RV: rvOfCP(cp), + UserCPRV: rvOfCP(userDefinedCP), + terminatedSeenAt: prev.terminatedSeenAt, } if userDefinedCP != nil { // The user-authored CP is authoritative and complete by definition (no diff --git a/tests/component_test.go b/tests/component_test.go index 154e441e8b..fa37964a44 100644 --- a/tests/component_test.go +++ b/tests/component_test.go @@ -3906,3 +3906,92 @@ func Test_49_EphemeralContainerFullTreatment(t *testing.T) { return countRuleAlerts(t, ns.Name, "R0001", "ephcon", "id") > 0 }, 2*time.Minute, 10*time.Second, "id was not in the ephemeral container's learned profile — it must fire R0001 (detected + alerted like any other container)") } + +// Test_50_ServiceRefNetworkNeighbor validates the serviceRef selector end to +// end (k8sstormcenter/node-agent#92): a workload whose ContainerProfile +// allowlists egress by Service NAME (default/kubernetes — the apiserver, a +// service every workload legitimately reaches) must NOT fire R0011 for that +// egress, while egress to a real but UNLISTED in-cluster Service (kube-dns) +// MUST still fire R0011. That contrast is the whole point of serviceRef over a +// broad serviceCIDR ipAddresses entry: a narrow, portable allowlist that does +// not blind R0011 to lateral movement. No toy target manifests — the peers are +// the cluster's own infrastructure Services. +func Test_50_ServiceRefNetworkNeighbor(t *testing.T) { + start := time.Now() + defer tearDownTest(t, start) + + getClusterIP := func(t *testing.T, ns, name string) string { + t.Helper() + k := k8sinterface.NewKubernetesApi() + svc, err := k.KubernetesClient.CoreV1().Services(ns).Get(context.TODO(), name, metav1.GetOptions{}) + require.NoError(t, err, "must read %s/%s ClusterIP", ns, name) + require.NotEmpty(t, svc.Spec.ClusterIP) + return svc.Spec.ClusterIP + } + countR0011 := func(alerts []testutils.Alert) int { + n := 0 + for _, a := range alerts { + if a.Labels["rule_id"] == "R0011" { + n++ + } + } + return n + } + waitAlerts := func(t *testing.T, ns string) []testutils.Alert { + t.Helper() + var alerts []testutils.Alert + require.Eventually(t, func() bool { + var err error + alerts, err = testutils.GetAlerts(ns) + return err == nil + }, 60*time.Second, 5*time.Second, "must be able to fetch alerts") + time.Sleep(10 * time.Second) + alerts, _ = testutils.GetAlerts(ns) + return alerts + } + + ns := testutils.NewRandomNamespace() + _ = applyUserDefinedContainerProfile(t, ns.Name, "resources/containerprofile-serviceref-network.yaml") + + wl, err := testutils.NewTestWorkload(ns.Name, + path.Join(utils.CurrentDir(), "resources/serviceref-client-deployment.yaml")) + require.NoError(t, err) + require.NoError(t, wl.WaitForReady(80)) + // Let node-agent load the bound profile AND sync its Service informer + // (serviceRef resolves against live cluster state) before generating traffic. + time.Sleep(40 * time.Second) + + apiserverIP := getClusterIP(t, "default", "kubernetes") + t.Logf("apiserver ClusterIP=%s (serviceRef-allowed); unlisted egress target=1.1.1.1:80", apiserverIP) + + // Phase 1 — egress to the apiserver, allowlisted by serviceRef + // default/kubernetes. The TCP connect is what R0011 evaluates; -k so curl + // attempts it despite the self-signed cert. + t.Run("serviceref_allowed_no_r0011", func(t *testing.T) { + for i := 0; i < 3; i++ { + so, se, e := wl.ExecIntoPod([]string{"curl", "-skm", "5", fmt.Sprintf("https://%s:443/healthz", apiserverIP)}, "curl") + t.Logf("curl apiserver → err=%v out=%q stderr=%q", e, so, se) + } + alerts := waitAlerts(t, wl.Namespace) + assert.Equal(t, 0, countR0011(alerts), + "apiserver egress is allowlisted by serviceRef default/kubernetes — R0011 must NOT fire") + }) + + // Phase 2 — egress NOT covered by the serviceRef must still fire R0011, + // proving serviceRef is a NARROW allowlist (only default/kubernetes), not a + // blanket that suppresses everything. Raw-IP egress to 1.1.1.1:80 is the + // proven R0011 trigger in this suite (mirrors Test_28c) and is the faithful + // analog of the flux RCA, where R0011 fired for the external github egress + // the named-service allowlist did not cover. + t.Run("uncovered_egress_fires_r0011", func(t *testing.T) { + before := countR0011(waitAlerts(t, wl.Namespace)) + for i := 0; i < 3; i++ { + so, se, e := wl.ExecIntoPod([]string{"curl", "-sm", "5", "http://1.1.1.1:80"}, "curl") + t.Logf("curl 1.1.1.1 (uncovered) → err=%v out=%q stderr=%q", e, so, se) + } + require.Eventually(t, func() bool { + return countR0011(waitAlerts(t, wl.Namespace)) > before + }, 3*time.Minute, 15*time.Second, + "egress uncovered by serviceRef MUST fire R0011 — serviceRef is narrow, not a blanket allow") + }) +} diff --git a/tests/resources/containerprofile-serviceref-network.yaml b/tests/resources/containerprofile-serviceref-network.yaml new file mode 100644 index 0000000000..4db29ba9b2 --- /dev/null +++ b/tests/resources/containerprofile-serviceref-network.yaml @@ -0,0 +1,48 @@ +apiVersion: spdx.softwarecomposition.kubescape.io/v1beta1 +kind: ContainerProfile +metadata: + name: serviceref-overlay +spec: + execs: + - path: /bin/sleep + - path: /usr/bin/curl + syscalls: + - socket + - connect + - sendto + - recvfrom + - read + - write + - close + - openat + - mmap + - mprotect + - munmap + - fcntl + - ioctl + - poll + - epoll_create1 + - epoll_ctl + - epoll_wait + - bind + - listen + - accept4 + - getsockopt + - setsockopt + - getsockname + - getpid + - fstat + - rt_sigaction + - rt_sigprocmask + - writev + matchLabels: + app: serviceref-client + egress: + - identifier: apiserver-serviceref + type: internal + serviceRefNamespace: default + serviceRefName: kubernetes + ports: + - name: TCP-443 + protocol: TCP + port: 443 diff --git a/tests/resources/network_fixture_lint_test.go b/tests/resources/network_fixture_lint_test.go index a7ec6cd0e2..376af7a878 100644 --- a/tests/resources/network_fixture_lint_test.go +++ b/tests/resources/network_fixture_lint_test.go @@ -74,6 +74,11 @@ type netEndpoint struct { // declared target for R-NN-12. PodSelector json.RawMessage `json:"podSelector"` NamespaceSelector json.RawMessage `json:"namespaceSelector"` + // Service/entity targets resolved to concrete IPs at projection time. + // Presence of any counts as a declared target for R-NN-12. + ServiceRefName string `json:"serviceRefName"` + ServiceSelector json.RawMessage `json:"serviceSelector"` + Entity string `json:"entity"` } // hasSelector reports whether a raw selector field was set to a real object @@ -114,7 +119,7 @@ func (v NetViolation) String() string { // R-NN-02 — at least one endpoint (egress or ingress) declared // R-NN-10 — endpoint identifier non-empty // R-NN-11 — endpoint type in {internal, external} (or unset) -// R-NN-12 — endpoint declares at least one target (dnsNames/ipAddresses/dns/ipAddress) +// R-NN-12 — endpoint declares at least one target (dnsNames/ipAddresses/dns/ipAddress/selector/serviceRef/entity) // R-NN-13 — dnsNames wildcard tokens are whole-label; no recursive "**", no ascii "..." // R-NN-14 — an entry MUST NOT set both singular ipAddress and plural ipAddresses // R-NN-15 — ipAddresses entries are a literal IP, a CIDR, or the "*" sentinel @@ -176,8 +181,9 @@ func lintEndpoint(dir string, e netEndpoint, add func(rule, msg string)) { add("R-NN-11", where(fmt.Sprintf("type %q is not internal|external", e.Type))) } if len(e.DNSNames) == 0 && len(e.IPAddresses) == 0 && e.DNS == "" && e.IPAddress == "" && - !hasSelector(e.PodSelector) && !hasSelector(e.NamespaceSelector) { - add("R-NN-12", where("endpoint declares no target (dnsNames/ipAddresses/dns/ipAddress/selector)")) + !hasSelector(e.PodSelector) && !hasSelector(e.NamespaceSelector) && + e.ServiceRefName == "" && !hasSelector(e.ServiceSelector) && e.Entity == "" { + add("R-NN-12", where("endpoint declares no target (dnsNames/ipAddresses/dns/ipAddress/selector/serviceRef/entity)")) } if e.IPAddress != "" && len(e.IPAddresses) > 0 { add("R-NN-14", where("sets both singular ipAddress and plural ipAddresses — pick one")) diff --git a/tests/resources/serviceref-client-deployment.yaml b/tests/resources/serviceref-client-deployment.yaml new file mode 100644 index 0000000000..7459155909 --- /dev/null +++ b/tests/resources/serviceref-client-deployment.yaml @@ -0,0 +1,21 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + labels: + app: serviceref-client + name: serviceref-client +spec: + replicas: 1 + selector: + matchLabels: + app: serviceref-client + template: + metadata: + labels: + app: serviceref-client + kubescape.io/user-defined-profile: serviceref-overlay + spec: + containers: + - name: curl + image: docker.io/curlimages/curl@sha256:08e466006f0860e54fc299378de998935333e0e130a15f6f98482e9f8dab3058 + command: ["sleep", "infinity"] From 6f30d65a4b78cc21b4e7ea6d2be57862a4c98d0c Mon Sep 17 00:00:00 2001 From: tanzee Date: Sun, 23 Aug 2026 23:11:11 +0200 Subject: [PATCH 2/5] feat(cel/network): real-Flux component test, RBAC + perf fixes for serviceRef MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Component test Test_50 now generates its traffic from a real Flux source-controller reconciling HelmRepository CRs instead of exec'ing curl, and its ContainerProfile is network-only (no syscalls/execs, which only add false-positive surface to a network test). The profile names every peer as a Kubernetes object: serviceRef default/kubernetes for the apiserver, serviceRef kube-system/kube-dns for resolution, and a serviceSelector role=helm-repo fanning across the two repo Services. The negative is the lateral move a serviceCIDR entry hides: the HelmRepository URL is repointed at a sibling Service on the same port that the selector does not cover, and the controller fetches it itself. Verified on kind: 0 alerts for the named peers, R0011 within 15s for the sibling. Fixes found while validating end to end: - ClusterRole was missing discovery.k8s.io/endpointslices, so the informer was forbidden and Service endpoint IPs never resolved — the feature silently degraded to ClusterIP-only. - Service/EndpointSlice informers are now gated behind networkServiceResolutionEnabled and strip managedFields/annotations (and per-endpoint fields beyond Addresses) via SetTransform, so agents that do not use the feature pay no cluster-wide list+watch and the cache stays small on those that do. - serviceRef/serviceSelector now also imply the Service cluster FQDN as a dnsName, so a client dialling the Service by name is allowlisted without a parallel dnsNames entry. - specFromNeighbor no longer allocates a discarded port slice for every plain ipAddresses neighbor. - R0011 no longer excludes private destinations: in-cluster lateral movement is exactly what this feature exists to expose. Signed-off-by: tanzee --- cmd/main.go | 95 +++--- pkg/config/config.go | 1 + pkg/networkpeer/expand.go | 18 +- pkg/networkpeer/expand_test.go | 8 + pkg/networkpeer/lister.go | 25 ++ pkg/networkpeer/perf_bench_test.go | 283 ++++++++++++++++++ pkg/networkpeer/resolve.go | 49 ++- pkg/networkpeer/resolve_test.go | 24 ++ .../templates/node-agent/clusterrole.yaml | 3 + .../chart/templates/node-agent/configmap.yaml | 1 + .../templates/node-agent/default-rules.yaml | 2 +- tests/component_test.go | 204 +++++++++---- .../containerprofile-serviceref-network.yaml | 48 --- .../serviceref-client-deployment.yaml | 21 -- .../serviceref-suite/00-flux-source-crds.yaml | 132 ++++++++ .../serviceref-suite/10-helm-repo.yaml | 108 +++++++ .../20-source-controller.yaml | 117 ++++++++ tests/testutils/k8s.go | 75 +++++ 18 files changed, 1024 insertions(+), 190 deletions(-) create mode 100644 pkg/networkpeer/perf_bench_test.go delete mode 100644 tests/resources/containerprofile-serviceref-network.yaml delete mode 100644 tests/resources/serviceref-client-deployment.yaml create mode 100644 tests/resources/serviceref-suite/00-flux-source-crds.yaml create mode 100644 tests/resources/serviceref-suite/10-helm-repo.yaml create mode 100644 tests/resources/serviceref-suite/20-source-controller.yaml diff --git a/cmd/main.go b/cmd/main.go index 92fd665d24..bfec59e977 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -325,49 +325,60 @@ func main() { ruleBindingCache.AddNotifier(&ruleBindingNotify) cpc := containerprofilecache.NewContainerProfileCache(cfg, storageClient, k8sObjectCache, metricsProvider) - // Resolve serviceRef/serviceSelector/entity network neighbors against - // live cluster state (Service ClusterIPs + endpoints, Node IPs + CNI - // gateway) at projection time. Services and EndpointSlices are watched - // cluster-wide (a profile may reference any namespace's Service); the - // Node informer is field-selected to this agent's own node — the "host" - // entity is local, and a cluster-wide Node watch on every DaemonSet pod - // is O(nodes^2) traffic for no benefit. - svcInformers := informers.NewSharedInformerFactory(k8sClient.GetKubernetesClient(), 0) - nodeInformers := informers.NewSharedInformerFactoryWithOptions( - k8sClient.GetKubernetesClient(), 0, - informers.WithTweakListOptions(func(o *metav1.ListOptions) { - o.FieldSelector = "metadata.name=" + cfg.NodeName - }), - ) - serviceLister := networkpeer.NewInformerLister( - svcInformers.Core().V1().Services().Lister(), - svcInformers.Discovery().V1().EndpointSlices().Lister(), - nodeInformers.Core().V1().Nodes().Lister(), - cfg.NodeName, - ) - // Advance the lister generation on any Service/EndpointSlice/Node change, - // so the reconciler re-projects serviceRef/entity profiles when the - // cluster view moves (endpoint churn, or caches that fill after startup). - // Per-event cost is a single atomic increment; the re-projection itself - // is coalesced onto the reconcile tick, and only serviceRef-using - // profiles are eligible. - bump := toolscache.ResourceEventHandlerFuncs{ - AddFunc: func(interface{}) { serviceLister.Bump() }, - UpdateFunc: func(_, _ interface{}) { serviceLister.Bump() }, - DeleteFunc: func(interface{}) { serviceLister.Bump() }, + // Resolve serviceRef/serviceSelector/entity network neighbors against live + // cluster state (Service ClusterIPs + endpoints, Node IPs + CNI gateway) at + // projection time. Gated behind networkServiceResolutionEnabled: the + // cluster-wide Service+EndpointSlice list+watch (one per DaemonSet node) is + // only paid where profiles actually use the feature. Services and + // EndpointSlices are watched cluster-wide (a profile may reference any + // namespace's Service); the Node informer is field-selected to this agent's + // own node — the "host" entity is local, and a cluster-wide Node watch on + // every DaemonSet pod is O(nodes^2) traffic for no benefit. A TransformFunc + // strips managedFields/annotations (and per-endpoint fields beyond + // Addresses) before objects enter the cache to keep its footprint small. + if cfg.EnableNetworkServiceResolution { + svcInformers := informers.NewSharedInformerFactory(k8sClient.GetKubernetesClient(), 0) + nodeInformers := informers.NewSharedInformerFactoryWithOptions( + k8sClient.GetKubernetesClient(), 0, + informers.WithTweakListOptions(func(o *metav1.ListOptions) { + o.FieldSelector = "metadata.name=" + cfg.NodeName + }), + ) + svcInformer := svcInformers.Core().V1().Services().Informer() + sliceInformer := svcInformers.Discovery().V1().EndpointSlices().Informer() + _ = svcInformer.SetTransform(networkpeer.TrimService) + _ = sliceInformer.SetTransform(networkpeer.TrimEndpointSlice) + serviceLister := networkpeer.NewInformerLister( + svcInformers.Core().V1().Services().Lister(), + svcInformers.Discovery().V1().EndpointSlices().Lister(), + nodeInformers.Core().V1().Nodes().Lister(), + cfg.NodeName, + ) + // Advance the lister generation on any Service/EndpointSlice/Node + // change, so the reconciler re-projects serviceRef/entity profiles when + // the cluster view moves (endpoint churn, or caches that fill after + // startup). Per-event cost is a single atomic increment; the + // re-projection itself is coalesced onto the reconcile tick, and only + // serviceRef-using profiles are eligible. + bump := toolscache.ResourceEventHandlerFuncs{ + AddFunc: func(interface{}) { serviceLister.Bump() }, + UpdateFunc: func(_, _ interface{}) { serviceLister.Bump() }, + DeleteFunc: func(interface{}) { serviceLister.Bump() }, + } + _, _ = svcInformer.AddEventHandler(bump) + _, _ = sliceInformer.AddEventHandler(bump) + _, _ = nodeInformers.Core().V1().Nodes().Informer().AddEventHandler(bump) + // Start the informers and hand the lister over WITHOUT blocking on + // cache sync: node-agent's core startup (container watcher, profiling) + // must not wait on these, and a bounded wait here previously delayed + // learning enough to trip the tight completion budget of Test_22. The + // caches fill in the background; serviceRef/entity neighbors resolve on + // the next reconcile once populated. nil-until-set is a no-op in + // projection. + svcInformers.Start(ctx.Done()) + nodeInformers.Start(ctx.Done()) + cpc.SetServiceLister(serviceLister) } - _, _ = svcInformers.Core().V1().Services().Informer().AddEventHandler(bump) - _, _ = svcInformers.Discovery().V1().EndpointSlices().Informer().AddEventHandler(bump) - _, _ = nodeInformers.Core().V1().Nodes().Informer().AddEventHandler(bump) - // Start the informers and hand the lister over WITHOUT blocking on cache - // sync: node-agent's core startup (container watcher, profiling) must not - // wait on these, and a bounded wait here previously delayed learning - // enough to trip the tight completion budget of Test_22. The caches fill - // in the background; serviceRef/entity neighbors resolve on the next - // reconcile once populated. nil-until-set is a no-op in projection. - svcInformers.Start(ctx.Done()) - nodeInformers.Start(ctx.Done()) - cpc.SetServiceLister(serviceLister) cpc.Start(ctx) if cpm, ok := containerProfileManager.(*containerprofilemanagerv1.ContainerProfileManager); ok { cpm.SetCompletionNotifier(cpc) diff --git a/pkg/config/config.go b/pkg/config/config.go index cec9f41ab6..e2d6e5d4a9 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -83,6 +83,7 @@ type Config struct { EnableMalwareDetection bool `mapstructure:"malwareDetectionEnabled"` EnableNetworkStreaming bool `mapstructure:"networkStreamingEnabled"` EnableNetworkTracing bool `mapstructure:"networkServiceEnabled"` + EnableNetworkServiceResolution bool `mapstructure:"networkServiceResolutionEnabled"` EnableNodeProfile bool `mapstructure:"nodeProfileServiceEnabled"` EnablePartialProfileGeneration bool `mapstructure:"partialProfileGenerationEnabled"` EnableMetricsExporter bool `mapstructure:"prometheusExporterEnabled"` diff --git a/pkg/networkpeer/expand.go b/pkg/networkpeer/expand.go index 2f76c97897..747c7b2d62 100644 --- a/pkg/networkpeer/expand.go +++ b/pkg/networkpeer/expand.go @@ -11,10 +11,12 @@ import ( // synthesized ipAddresses neighbors (one per source neighbor, carrying the // resolved IPs and the source neighbor's own ports). // -// The synthesized neighbors are ordinary selector-free ipAddresses entries, so -// the existing port-sensitive address matcher handles them with no further +// The synthesized neighbors are ordinary selector-free ipAddresses entries +// (plus, for Service-backed specs, the Service's cluster FQDN as a dnsName so a +// client dialling it by name is allowlisted too), so the existing +// port-sensitive address matcher and DNS matcher handle them with no further // change — a serviceRef/host neighbor becomes exactly the narrow, resolved -// ipAddresses entry it stands for. Neighbors that resolve to nothing (unknown +// entry it stands for. Neighbors that resolve to nothing (unknown // Service, selector matching nothing, unknown entity) contribute nothing — // never a match-all. Callers append the result to the same direction (egress // or ingress) before projecting the profile. @@ -30,13 +32,15 @@ func ExpandServiceNeighbors(neighbors []v1beta1.NetworkNeighbor, l Lister) []v1b continue } ips := ResolveIPs(spec, l) - if len(ips) == 0 { + dnsNames := ResolveDNSNames(spec, l) + if len(ips) == 0 && len(dnsNames) == 0 { continue } out = append(out, v1beta1.NetworkNeighbor{ Identifier: n.Identifier + "-resolved", Type: n.Type, IPAddresses: ips, + DNSNames: dnsNames, Ports: n.Ports, }) } @@ -96,6 +100,12 @@ func hasServiceFields(n *v1beta1.NetworkNeighbor) bool { // if the neighbor declares none of the service/entity selectors (a plain // ipAddresses / dnsNames / podSelector neighbor is left untouched). func specFromNeighbor(n *v1beta1.NetworkNeighbor) (PeerSpec, bool) { + // Cheap-reject a plain ipAddresses/dnsNames neighbor before allocating a + // []PortProto it would only discard (hot on every projection's non-service + // neighbors). + if n.Entity == "" && n.ServiceRefName == "" && n.ServiceSelector == nil { + return PeerSpec{}, false + } spec := PeerSpec{Ports: portsFromNeighbor(n.Ports)} switch { case n.Entity != "": diff --git a/pkg/networkpeer/expand_test.go b/pkg/networkpeer/expand_test.go index dfb397b879..715a76e607 100644 --- a/pkg/networkpeer/expand_test.go +++ b/pkg/networkpeer/expand_test.go @@ -48,6 +48,10 @@ func TestExpandServiceNeighbors_Egress(t *testing.T) { if len(got.Ports) != 1 || got.Ports[0].Port == nil || *got.Ports[0].Port != 9093 { t.Errorf("ports not carried over: %+v", got.Ports) } + // serviceRef implies the Service cluster FQDN as a dnsName (R0005 suppression). + if len(got.DNSNames) != 1 || got.DNSNames[0] != "alertmanager.honey.svc.cluster.local" { + t.Errorf("serviceRef FQDN not emitted: %v", got.DNSNames) + } } // TestExpandServiceNeighbors_HostEntity: fromEntity host resolves to node + @@ -61,6 +65,10 @@ func TestExpandServiceNeighbors_HostEntity(t *testing.T) { if len(out) != 1 { t.Fatalf("expected 1 synthesized host neighbor, got %d", len(out)) } + // host entity is not a Service: no FQDN. + if len(out[0].DNSNames) != 0 { + t.Errorf("host entity must not emit a dnsName: %v", out[0].DNSNames) + } // Feed the synthesized entry through the address matcher the same way the // projection would, to prove end-to-end intent. tuples := Resolve(PeerSpec{Entity: "host", Ports: []PortProto{{Port: 9440, Protocol: "TCP"}}}, l) diff --git a/pkg/networkpeer/lister.go b/pkg/networkpeer/lister.go index eff6d349db..5c88c7d7ba 100644 --- a/pkg/networkpeer/lister.go +++ b/pkg/networkpeer/lister.go @@ -126,6 +126,31 @@ func (l *InformerLister) endpointIPs(namespace, service string) []string { return dedupe(ips) } +// TrimService and TrimEndpointSlice are informer TransformFuncs that drop the +// bulk the resolver never reads — managedFields and annotations (1–4 KiB per +// real object), and for EndpointSlices every per-endpoint field but Addresses — +// before objects enter the cluster-wide cache. Wire via Informer().SetTransform +// so a DaemonSet's per-node Service/EndpointSlice cache stays small. Identity +// and resourceVersion are preserved so listing/indexing is unaffected. +func TrimService(obj interface{}) (interface{}, error) { + if svc, ok := obj.(*corev1.Service); ok { + svc.ManagedFields = nil + svc.Annotations = nil + } + return obj, nil +} + +func TrimEndpointSlice(obj interface{}) (interface{}, error) { + if es, ok := obj.(*discoveryv1.EndpointSlice); ok { + es.ManagedFields = nil + es.Annotations = nil + for i := range es.Endpoints { + es.Endpoints[i] = discoveryv1.Endpoint{Addresses: es.Endpoints[i].Addresses} + } + } + return obj, nil +} + func podCIDRs(n *corev1.Node) []string { if len(n.Spec.PodCIDRs) > 0 { return n.Spec.PodCIDRs diff --git a/pkg/networkpeer/perf_bench_test.go b/pkg/networkpeer/perf_bench_test.go new file mode 100644 index 0000000000..e3343214b8 --- /dev/null +++ b/pkg/networkpeer/perf_bench_test.go @@ -0,0 +1,283 @@ +package networkpeer + +import ( + "fmt" + "runtime" + "testing" + + "github.com/kubescape/storage/pkg/apis/softwarecomposition/v1beta1" + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + corelisters "k8s.io/client-go/listers/core/v1" + discoverylisters "k8s.io/client-go/listers/discovery/v1" + "k8s.io/client-go/tools/cache" +) + +func benchService(i int) *corev1.Service { + return &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: fmt.Sprintf("ns-%d", i%50), + Name: fmt.Sprintf("svc-%d", i), + Labels: map[string]string{"app": fmt.Sprintf("app-%d", i), "team": fmt.Sprintf("team-%d", i%20)}, + }, + Spec: corev1.ServiceSpec{ + ClusterIP: fmt.Sprintf("10.43.%d.%d", i/256, i%256), + ClusterIPs: []string{fmt.Sprintf("10.43.%d.%d", i/256, i%256)}, + Ports: []corev1.ServicePort{{Name: "http", Port: 8080, Protocol: corev1.ProtocolTCP}}, + }, + } +} + +func benchSlice(svcIdx, sliceIdx, endpoints int) *discoveryv1.EndpointSlice { + eps := make([]discoveryv1.Endpoint, 0, endpoints) + ready := true + for e := 0; e < endpoints; e++ { + eps = append(eps, discoveryv1.Endpoint{ + Addresses: []string{fmt.Sprintf("10.42.%d.%d", (svcIdx*7+e)%256, (sliceIdx*31+e)%256)}, + Conditions: discoveryv1.EndpointConditions{Ready: &ready}, + TargetRef: &corev1.ObjectReference{Kind: "Pod", Namespace: fmt.Sprintf("ns-%d", svcIdx%50), Name: fmt.Sprintf("pod-%d-%d-%d", svcIdx, sliceIdx, e)}, + NodeName: ptrTo(fmt.Sprintf("node-%d", e%10)), + }) + } + return &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: fmt.Sprintf("ns-%d", svcIdx%50), + Name: fmt.Sprintf("svc-%d-%d", svcIdx, sliceIdx), + Labels: map[string]string{discoveryv1.LabelServiceName: fmt.Sprintf("svc-%d", svcIdx)}, + }, + AddressType: discoveryv1.AddressTypeIPv4, + Endpoints: eps, + Ports: []discoveryv1.EndpointPort{{Name: ptrTo("http"), Port: ptrTo(int32(8080)), Protocol: &[]corev1.Protocol{corev1.ProtocolTCP}[0]}}, + } +} + +func ptrTo[T any](v T) *T { return &v } + +// buildBenchLister backs an InformerLister with plain cache indexers (the same +// store type a SharedInformer uses) so benchmarks measure lister/resolution +// cost without fake-clientset watch machinery. +func buildBenchLister(tb testing.TB, nServices, slicesPerSvc, endpointsPerSlice int) *InformerLister { + tb.Helper() + svcIdx := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + sliceIdx := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + nodeIdx := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + for i := 0; i < nServices; i++ { + if err := svcIdx.Add(benchService(i)); err != nil { + tb.Fatal(err) + } + for s := 0; s < slicesPerSvc; s++ { + if err := sliceIdx.Add(benchSlice(i, s, endpointsPerSlice)); err != nil { + tb.Fatal(err) + } + } + } + if err := nodeIdx.Add(&corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: "bench-node"}, + Spec: corev1.NodeSpec{PodCIDR: "10.42.0.0/24", PodCIDRs: []string{"10.42.0.0/24"}}, + Status: corev1.NodeStatus{Addresses: []corev1.NodeAddress{{Type: corev1.NodeInternalIP, Address: "192.168.0.191"}}}, + }); err != nil { + tb.Fatal(err) + } + return NewInformerLister( + corelisters.NewServiceLister(svcIdx), + discoverylisters.NewEndpointSliceLister(sliceIdx), + corelisters.NewNodeLister(nodeIdx), + "bench-node", + ) +} + +// benchProfile builds a ContainerProfile with nPlain ordinary ipAddresses +// egress neighbors, nRefs serviceRef neighbors, and some opens/execs bulk so +// DeepCopy cost is realistic. +func benchProfile(nPlain, nRefs, nOpens int) *v1beta1.ContainerProfile { + cp := &v1beta1.ContainerProfile{} + cp.Name = "bench-cp" + port := int32(8080) + for i := 0; i < nPlain; i++ { + cp.Spec.Egress = append(cp.Spec.Egress, v1beta1.NetworkNeighbor{ + Identifier: fmt.Sprintf("plain-%d", i), + Type: "external", + IPAddresses: []string{fmt.Sprintf("52.216.%d.%d", i/256, i%256)}, + Ports: []v1beta1.NetworkPort{{Name: "TCP-8080", Protocol: "TCP", Port: &port}}, + }) + } + for i := 0; i < nRefs; i++ { + cp.Spec.Egress = append(cp.Spec.Egress, v1beta1.NetworkNeighbor{ + Identifier: fmt.Sprintf("ref-%d", i), + Type: "internal", + ServiceRefNamespace: fmt.Sprintf("ns-%d", i%50), + ServiceRefName: fmt.Sprintf("svc-%d", i), + Ports: []v1beta1.NetworkPort{{Name: "TCP-8080", Protocol: "TCP", Port: &port}}, + }) + } + for i := 0; i < nOpens; i++ { + cp.Spec.Opens = append(cp.Spec.Opens, v1beta1.OpenCalls{ + Path: fmt.Sprintf("/usr/lib/x86_64-linux-gnu/lib-%d.so.%d", i, i%9), + Flags: []string{"O_RDONLY", "O_CLOEXEC"}, + }) + cp.Spec.Execs = append(cp.Spec.Execs, v1beta1.ExecCalls{ + Path: fmt.Sprintf("/usr/bin/tool-%d", i), + Args: []string{fmt.Sprintf("--flag-%d", i)}, + }) + } + return cp +} + +func BenchmarkServiceByName_1kSvc_5kSlices(b *testing.B) { + l := buildBenchLister(b, 1000, 5, 10) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, ok := l.ServiceByName("ns-7", "svc-7"); !ok { + b.Fatal("service must resolve") + } + } +} + +func BenchmarkServicesByLabels_1kSvc(b *testing.B) { + l := buildBenchLister(b, 1000, 5, 10) + sel := map[string]string{"app": "app-7"} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if out := l.ServicesByLabels(sel, nil); len(out) != 1 { + b.Fatalf("want 1 service, got %d", len(out)) + } + } +} + +func BenchmarkResolveIPs_ServiceRef(b *testing.B) { + l := buildBenchLister(b, 1000, 5, 10) + spec := PeerSpec{ServiceRef: &ServiceRef{Namespace: "ns-7", Name: "svc-7"}} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if ips := ResolveIPs(spec, l); len(ips) == 0 { + b.Fatal("must resolve") + } + } +} + +func BenchmarkResolveIPs_Entity_Host(b *testing.B) { + l := buildBenchLister(b, 10, 1, 2) + spec := PeerSpec{Entity: "host"} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if ips := ResolveIPs(spec, l); len(ips) == 0 { + b.Fatal("must resolve") + } + } +} + +// BenchmarkWithResolvedServiceNeighbors_NoServiceFields is the 99% case: a +// profile with only plain ipAddresses neighbors. The function documents itself +// as a no-op then — this measures whether the no-op is actually free. +func BenchmarkWithResolvedServiceNeighbors_NoServiceFields(b *testing.B) { + l := buildBenchLister(b, 1000, 5, 10) + for _, n := range []int{100, 1000} { + cp := benchProfile(n, 0, 0) + b.Run(fmt.Sprintf("plainNeighbors=%d", n), func(b *testing.B) { + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + out := WithResolvedServiceNeighbors(cp, l) + if out != cp { + b.Fatal("no-op path must return the same pointer") + } + } + }) + } +} + +// BenchmarkWithResolvedServiceNeighbors_Resolving measures the full expansion: +// resolution + DeepCopy of the whole profile (opens/execs bulk included). +func BenchmarkWithResolvedServiceNeighbors_Resolving(b *testing.B) { + l := buildBenchLister(b, 1000, 5, 10) + for _, tc := range []struct{ plain, refs, opens int }{ + {plain: 20, refs: 1, opens: 200}, + {plain: 20, refs: 8, opens: 200}, + {plain: 20, refs: 8, opens: 2000}, + } { + cp := benchProfile(tc.plain, tc.refs, tc.opens) + b.Run(fmt.Sprintf("plain=%d/refs=%d/opens=%d", tc.plain, tc.refs, tc.opens), func(b *testing.B) { + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + out := WithResolvedServiceNeighbors(cp, l) + if out == cp { + b.Fatal("resolving path must copy") + } + } + }) + } +} + +func BenchmarkHasServiceNeighbors_1kPlain(b *testing.B) { + cp := benchProfile(1000, 0, 0) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if HasServiceNeighbors(cp) { + b.Fatal("plain profile must not report service neighbors") + } + } +} + +// TestInformerCacheMemoryEstimate approximates the heap retained by a +// cluster-wide Service + EndpointSlice informer cache at 1k Services / 5k +// EndpointSlices (10 endpoints each), vs Services alone. Run with -run +// InformerCacheMemoryEstimate -v. +func TestInformerCacheMemoryEstimate(t *testing.T) { + measure := func(build func() []interface{}) uint64 { + runtime.GC() + var before, after runtime.MemStats + runtime.ReadMemStats(&before) + objs := build() + runtime.GC() + runtime.ReadMemStats(&after) + runtime.KeepAlive(objs) + if after.HeapAlloc < before.HeapAlloc { + return 0 + } + return after.HeapAlloc - before.HeapAlloc + } + + svcBytes := measure(func() []interface{} { + idx := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + for i := 0; i < 1000; i++ { + _ = idx.Add(benchService(i)) + } + return []interface{}{idx} + }) + sliceBytes := measure(func() []interface{} { + idx := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + for i := 0; i < 1000; i++ { + for s := 0; s < 5; s++ { + _ = idx.Add(benchSlice(i, s, 10)) + } + } + return []interface{}{idx} + }) + strippedBytes := measure(func() []interface{} { + idx := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}) + for i := 0; i < 1000; i++ { + for s := 0; s < 5; s++ { + full := benchSlice(i, s, 10) + stripped := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{Namespace: full.Namespace, Name: full.Name, Labels: full.Labels}, + AddressType: full.AddressType, + } + for _, ep := range full.Endpoints { + stripped.Endpoints = append(stripped.Endpoints, discoveryv1.Endpoint{Addresses: ep.Addresses}) + } + _ = idx.Add(stripped) + } + } + return []interface{}{idx} + }) + t.Logf("1000 Services in indexer: ~%d KiB total, ~%d B/object", svcBytes/1024, svcBytes/1000) + t.Logf("5000 EndpointSlices (10 endpoints each) in indexer: ~%d KiB total, ~%d B/object", sliceBytes/1024, sliceBytes/5000) + t.Logf("5000 STRIPPED EndpointSlices (addresses+labels only, SetTransform mitigation): ~%d KiB total, ~%d B/object", strippedBytes/1024, strippedBytes/5000) +} diff --git a/pkg/networkpeer/resolve.go b/pkg/networkpeer/resolve.go index 0467c14b50..de8c1375d7 100644 --- a/pkg/networkpeer/resolve.go +++ b/pkg/networkpeer/resolve.go @@ -24,6 +24,12 @@ import "strings" // represent (kubelet health probes, node-sourced / masqueraded traffic). const EntityHost = "host" +// clusterDNSSuffix is the in-cluster Service DNS zone. A serviceRef / +// serviceSelector implies the cluster FQDN ..svc. so a +// client dialling the Service by DNS is allowlisted without a parallel +// dnsNames entry. Kubernetes' default zone; overridden per-cluster is rare. +const clusterDNSSuffix = "svc.cluster.local" + // PortProto is a single allowed destination port/protocol. Protocol is // upper-case ("TCP"/"UDP"); an empty Protocol matches any protocol. type PortProto struct { @@ -105,33 +111,58 @@ func ResolveIPs(spec PeerSpec, l Lister) []string { } func resolveIPs(spec PeerSpec, l Lister) []string { - switch { - case spec.Entity != "": + if spec.Entity != "" { if strings.EqualFold(spec.Entity, EntityHost) { return dedupe(l.HostIPs()) } return nil + } + var ips []string + for _, svc := range resolveServices(spec, l) { + ips = append(ips, serviceIPs(svc)...) + } + return dedupe(ips) +} + +// resolveServices returns the Services a serviceRef / serviceSelector spec +// matches. An entity spec matches no Service and returns nil; an empty +// serviceSelector fails closed (never every Service). +func resolveServices(spec PeerSpec, l Lister) []*ServiceInfo { + switch { case spec.ServiceRef != nil: svc, ok := l.ServiceByName(spec.ServiceRef.Namespace, spec.ServiceRef.Name) if !ok || svc == nil { return nil } - return serviceIPs(svc) + return []*ServiceInfo{svc} case spec.ServiceSelector != nil: - // An empty selector is NOT a cluster-wide match-all: fail closed. if len(spec.ServiceSelector) == 0 { return nil } - var ips []string - for _, svc := range l.ServicesByLabels(spec.ServiceSelector, spec.NamespaceLabels) { - ips = append(ips, serviceIPs(svc)...) - } - return dedupe(ips) + return l.ServicesByLabels(spec.ServiceSelector, spec.NamespaceLabels) default: return nil } } +// ResolveDNSNames returns the cluster FQDN(s) — ..svc. — +// of the Services a serviceRef / serviceSelector spec matches, so a client +// dialling the Service by DNS is allowlisted alongside its IPs. Entity specs +// and unresolvable selectors yield nothing. +func ResolveDNSNames(spec PeerSpec, l Lister) []string { + if l == nil { + return nil + } + var out []string + for _, svc := range resolveServices(spec, l) { + if svc == nil || svc.Namespace == "" || svc.Name == "" { + continue + } + out = append(out, svc.Name+"."+svc.Namespace+"."+clusterDNSSuffix) + } + return dedupe(out) +} + func serviceIPs(svc *ServiceInfo) []string { out := make([]string, 0, len(svc.ClusterIPs)+len(svc.EndpointIPs)) out = append(out, svc.ClusterIPs...) diff --git a/pkg/networkpeer/resolve_test.go b/pkg/networkpeer/resolve_test.go index f131f774fe..0ddcc0ccbe 100644 --- a/pkg/networkpeer/resolve_test.go +++ b/pkg/networkpeer/resolve_test.go @@ -191,6 +191,30 @@ func TestResolve_NoMatchAll(t *testing.T) { } } +// Test A7 — serviceRef/serviceSelector imply the Service cluster FQDN(s); +// entity and unresolvable specs imply none. +func TestResolveDNSNames(t *testing.T) { + l := realFluxTopology() + one := ResolveDNSNames(PeerSpec{ServiceRef: &ServiceRef{"honey", "alertmanager"}}, l) + if len(one) != 1 || one[0] != "alertmanager.honey.svc.cluster.local" { + t.Errorf("serviceRef FQDN: got %v", one) + } + fan := ResolveDNSNames(PeerSpec{ + ServiceSelector: map[string]string{"app": "guestbook"}, + NamespaceLabels: map[string]string{"kubernetes.io/metadata.name": "gitops-demo"}, + }, l) + want := map[string]bool{"guestbook-ui.gitops-demo.svc.cluster.local": true, "helm-guestbook.gitops-demo.svc.cluster.local": true} + if len(fan) != 2 || !want[fan[0]] || !want[fan[1]] { + t.Errorf("selector FQDN fanout: got %v", fan) + } + if got := ResolveDNSNames(PeerSpec{Entity: EntityHost}, l); got != nil { + t.Errorf("host entity implies no FQDN, got %v", got) + } + if got := ResolveDNSNames(PeerSpec{ServiceRef: &ServiceRef{"honey", "nope"}}, l); got != nil { + t.Errorf("unresolvable serviceRef implies no FQDN, got %v", got) + } +} + // Test A6 — a nil Lister and any-port (no Ports) behave safely. func TestResolve_Edges(t *testing.T) { if got := Resolve(PeerSpec{Entity: EntityHost}, nil); got != nil { diff --git a/tests/chart/templates/node-agent/clusterrole.yaml b/tests/chart/templates/node-agent/clusterrole.yaml index 03d5137555..a9feeed81a 100644 --- a/tests/chart/templates/node-agent/clusterrole.yaml +++ b/tests/chart/templates/node-agent/clusterrole.yaml @@ -11,6 +11,9 @@ rules: - apiGroups: [""] resources: ["events"] verbs: ["list", "watch", "create"] +- apiGroups: ["discovery.k8s.io"] + resources: ["endpointslices"] + verbs: ["get", "watch", "list"] - apiGroups: ["batch"] resources: ["jobs", "cronjobs"] verbs: ["get", "watch", "list"] diff --git a/tests/chart/templates/node-agent/configmap.yaml b/tests/chart/templates/node-agent/configmap.yaml index 523b5bbac6..6108427e90 100644 --- a/tests/chart/templates/node-agent/configmap.yaml +++ b/tests/chart/templates/node-agent/configmap.yaml @@ -14,6 +14,7 @@ data: "prometheusExporterEnabled": {{ eq .Values.nodeAgent.config.prometheusExporter "enable" }}, "runtimeDetectionEnabled": {{ eq .Values.capabilities.runtimeDetection "enable" }}, "networkServiceEnabled": {{ eq .Values.capabilities.networkPolicyService "enable" }}, + "networkServiceResolutionEnabled": true, "malwareDetectionEnabled": {{ eq .Values.capabilities.malwareDetection "enable" }}, "httpDetectionEnabled": {{ eq .Values.capabilities.httpDetection "enable" }}, "initialDelay": "{{ .Values.nodeAgent.config.learningPeriod }}", diff --git a/tests/chart/templates/node-agent/default-rules.yaml b/tests/chart/templates/node-agent/default-rules.yaml index 512b4d9ec8..913fb1f927 100644 --- a/tests/chart/templates/node-agent/default-rules.yaml +++ b/tests/chart/templates/node-agent/default-rules.yaml @@ -313,7 +313,7 @@ spec: uniqueId: "event.dstAddr + '_' + string(event.dstPort) + '_' + event.proto" ruleExpression: - eventType: "network" - expression: "event.pktType == 'OUTGOING' && !net.is_private_ip(event.dstAddr) && !cp.was_address_in_egress(event.containerId, event.dstAddr)" + expression: "event.pktType == 'OUTGOING' && !cp.was_address_in_egress(event.containerId, event.dstAddr)" profileDependency: 0 profileDataRequired: egressAddresses: all diff --git a/tests/component_test.go b/tests/component_test.go index fa37964a44..fbb10b6211 100644 --- a/tests/component_test.go +++ b/tests/component_test.go @@ -3907,91 +3907,165 @@ func Test_49_EphemeralContainerFullTreatment(t *testing.T) { }, 2*time.Minute, 10*time.Second, "id was not in the ephemeral container's learned profile — it must fire R0001 (detected + alerted like any other container)") } -// Test_50_ServiceRefNetworkNeighbor validates the serviceRef selector end to -// end (k8sstormcenter/node-agent#92): a workload whose ContainerProfile -// allowlists egress by Service NAME (default/kubernetes — the apiserver, a -// service every workload legitimately reaches) must NOT fire R0011 for that -// egress, while egress to a real but UNLISTED in-cluster Service (kube-dns) -// MUST still fire R0011. That contrast is the whole point of serviceRef over a -// broad serviceCIDR ipAddresses entry: a narrow, portable allowlist that does -// not blind R0011 to lateral movement. No toy target manifests — the peers are -// the cluster's own infrastructure Services. +// Test_50_ServiceRefNetworkNeighbor validates serviceRef/serviceSelector end to +// end (k8sstormcenter/node-agent#92) against REAL GitOps traffic: a Flux +// source-controller reconciling HelmRepository CRs. Its egress is authored +// purely as Kubernetes-native selectors — serviceRef default/kubernetes for the +// apiserver, serviceRef kube-system/kube-dns for name resolution, and a +// serviceSelector role=helm-repo fanning across the two repo Services it is +// allowed to fetch. Nothing is exec'd and no address is hardcoded; the +// controller's own reconcile loop generates every connection. +// +// The negative is the lateral move a broad serviceCIDR entry would hide: the +// HelmRepository URL is repointed at decoy-repo — a sibling Service on the same +// port, backed by its own pod, carrying none of the selector's labels — and the +// controller itself fetches it. R0011 MUST fire for that and MUST NOT fire for +// the allowlisted Services. func Test_50_ServiceRefNetworkNeighbor(t *testing.T) { start := time.Now() defer tearDownTest(t, start) - getClusterIP := func(t *testing.T, ns, name string) string { + const ( + cpName = "serviceref-flux-cp" + containerName = "manager" + ) + port80, port443, port53 := int32(80), int32(443), int32(53) + + ns := testutils.NewRandomNamespace() + k8sClient := k8sinterface.NewKubernetesApi() + storageClient := spdxv1beta1client.NewForConfigOrDie(k8sClient.K8SConfig) + + // Authored profile: NETWORK ONLY. No syscalls/execs — they are irrelevant to + // a network test and only add false-positive surface. Every peer is named, + // never addressed. + cp := &v1beta1.ContainerProfile{ + ObjectMeta: metav1.ObjectMeta{Name: cpName, Namespace: ns.Name}, + Spec: v1beta1.ContainerProfileSpec{ + LabelSelector: metav1.LabelSelector{MatchLabels: map[string]string{"app": "source-controller"}}, + Egress: []v1beta1.NetworkNeighbor{ + { + Identifier: "apiserver", + Type: v1beta1.CommunicationTypeEgress, + ServiceRefNamespace: "default", + ServiceRefName: "kubernetes", + Ports: []v1beta1.NetworkPort{{Name: "TCP-443", Protocol: v1beta1.ProtocolTCP, Port: &port443}}, + }, + { + Identifier: "cluster-dns", + Type: v1beta1.CommunicationTypeEgress, + ServiceRefNamespace: "kube-system", + ServiceRefName: "kube-dns", + Ports: []v1beta1.NetworkPort{ + {Name: "UDP-53", Protocol: v1beta1.ProtocolUDP, Port: &port53}, + {Name: "TCP-53", Protocol: v1beta1.ProtocolTCP, Port: &port53}, + }, + }, + { + Identifier: "helm-repos", + Type: v1beta1.CommunicationTypeEgress, + ServiceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"role": "helm-repo"}}, + NamespaceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"kubernetes.io/metadata.name": ns.Name}}, + Ports: []v1beta1.NetworkPort{{Name: "TCP-80", Protocol: v1beta1.ProtocolTCP, Port: &port80}}, + }, + }, + }, + } + _, err := storageClient.ContainerProfiles(ns.Name).Create(context.Background(), cp, metav1.CreateOptions{}) + require.NoError(t, err, "create authored ContainerProfile") + require.Eventually(t, func() bool { + _, e := storageClient.ContainerProfiles(ns.Name).Get(context.Background(), cpName, v1.GetOptions{}) + return e == nil + }, 30*time.Second, time.Second, "authored CP must be in storage before pod deploy") + + require.NoError(t, testutils.ApplyMultiDocDir(ns.Name, path.Join(utils.CurrentDir(), "resources/serviceref-suite")), + "apply flux source-controller + helm repo suite") + + waitDeploy := func(name string) { t.Helper() - k := k8sinterface.NewKubernetesApi() - svc, err := k.KubernetesClient.CoreV1().Services(ns).Get(context.TODO(), name, metav1.GetOptions{}) - require.NoError(t, err, "must read %s/%s ClusterIP", ns, name) - require.NotEmpty(t, svc.Spec.ClusterIP) - return svc.Spec.ClusterIP + require.Eventually(t, func() bool { + d, e := k8sClient.KubernetesClient.AppsV1().Deployments(ns.Name).Get(context.TODO(), name, metav1.GetOptions{}) + return e == nil && d.Status.ReadyReplicas > 0 + }, 3*time.Minute, 5*time.Second, "%s must become ready", name) } - countR0011 := func(alerts []testutils.Alert) int { + waitDeploy("helm-repo") + waitDeploy("decoy-repo") + waitDeploy("source-controller") + + countRule := func(ruleID string) int { + alerts, _ := testutils.GetAlerts(ns.Name) n := 0 for _, a := range alerts { - if a.Labels["rule_id"] == "R0011" { + if a.Labels["rule_id"] == ruleID && a.Labels["container_name"] == containerName { n++ } } return n } - waitAlerts := func(t *testing.T, ns string) []testutils.Alert { - t.Helper() - var alerts []testutils.Alert - require.Eventually(t, func() bool { - var err error - alerts, err = testutils.GetAlerts(ns) - return err == nil - }, 60*time.Second, 5*time.Second, "must be able to fetch alerts") - time.Sleep(10 * time.Second) - alerts, _ = testutils.GetAlerts(ns) - return alerts - } - ns := testutils.NewRandomNamespace() - _ = applyUserDefinedContainerProfile(t, ns.Name, "resources/containerprofile-serviceref-network.yaml") + helmRepoGVR := schema.GroupVersionResource{Group: "source.toolkit.fluxcd.io", Version: "v1", Resource: "helmrepositories"} + repoClient := k8sClient.DynamicClient.Resource(helmRepoGVR).Namespace(ns.Name) + newRepo := func(name, svc string) *unstructured.Unstructured { + return &unstructured.Unstructured{Object: map[string]interface{}{ + "apiVersion": "source.toolkit.fluxcd.io/v1", + "kind": "HelmRepository", + "metadata": map[string]interface{}{"name": name, "namespace": ns.Name}, + // Trailing-dot FQDN: absolute, so the resolver does not walk the + // search list and emit extra lookups. + "spec": map[string]interface{}{ + "interval": "30s", + "url": fmt.Sprintf("http://%s.%s.svc.cluster.local./", svc, ns.Name), + }, + }} + } + repoReady := func(name string) bool { + obj, e := repoClient.Get(context.TODO(), name, metav1.GetOptions{}) + if e != nil { + return false + } + conds, _, _ := unstructured.NestedSlice(obj.Object, "status", "conditions") + for _, c := range conds { + m, ok := c.(map[string]interface{}) + if ok && m["type"] == "Ready" && m["status"] == "True" { + return true + } + } + return false + } - wl, err := testutils.NewTestWorkload(ns.Name, - path.Join(utils.CurrentDir(), "resources/serviceref-client-deployment.yaml")) - require.NoError(t, err) - require.NoError(t, wl.WaitForReady(80)) - // Let node-agent load the bound profile AND sync its Service informer - // (serviceRef resolves against live cluster state) before generating traffic. + // Let node-agent bind the profile and fill its Service/EndpointSlice caches + // before any reconcile traffic is judged. time.Sleep(40 * time.Second) - apiserverIP := getClusterIP(t, "default", "kubernetes") - t.Logf("apiserver ClusterIP=%s (serviceRef-allowed); unlisted egress target=1.1.1.1:80", apiserverIP) - - // Phase 1 — egress to the apiserver, allowlisted by serviceRef - // default/kubernetes. The TCP connect is what R0011 evaluates; -k so curl - // attempts it despite the self-signed cert. - t.Run("serviceref_allowed_no_r0011", func(t *testing.T) { - for i := 0; i < 3; i++ { - so, se, e := wl.ExecIntoPod([]string{"curl", "-skm", "5", fmt.Sprintf("https://%s:443/healthz", apiserverIP)}, "curl") - t.Logf("curl apiserver → err=%v out=%q stderr=%q", e, so, se) + // Phase 1 — the controller reconciles both allowlisted repo Services while + // continuously talking to the apiserver and cluster DNS. Every one of those + // peers is named by the profile, so no egress alert may fire. + t.Run("selector_allowed_no_alert", func(t *testing.T) { + for _, r := range []struct{ name, svc string }{{"primary", "helm-primary"}, {"mirror", "helm-mirror"}} { + _, e := repoClient.Create(context.TODO(), newRepo(r.name, r.svc), metav1.CreateOptions{}) + require.NoError(t, e, "create HelmRepository %s", r.name) } - alerts := waitAlerts(t, wl.Namespace) - assert.Equal(t, 0, countR0011(alerts), - "apiserver egress is allowlisted by serviceRef default/kubernetes — R0011 must NOT fire") + for _, n := range []string{"primary", "mirror"} { + require.Eventually(t, func() bool { return repoReady(n) }, 3*time.Minute, 10*time.Second, + "HelmRepository %s must reconcile (real fetch through an allowlisted Service)", n) + } + // Two further reconcile intervals of steady-state traffic. + time.Sleep(90 * time.Second) + assert.Equal(t, 0, countRule("R0011"), + "apiserver/DNS/helm-repo egress is fully named by serviceRef+serviceSelector — R0011 must NOT fire") }) - // Phase 2 — egress NOT covered by the serviceRef must still fire R0011, - // proving serviceRef is a NARROW allowlist (only default/kubernetes), not a - // blanket that suppresses everything. Raw-IP egress to 1.1.1.1:80 is the - // proven R0011 trigger in this suite (mirrors Test_28c) and is the faithful - // analog of the flux RCA, where R0011 fired for the external github egress - // the named-service allowlist did not cover. - t.Run("uncovered_egress_fires_r0011", func(t *testing.T) { - before := countR0011(waitAlerts(t, wl.Namespace)) - for i := 0; i < 3; i++ { - so, se, e := wl.ExecIntoPod([]string{"curl", "-sm", "5", "http://1.1.1.1:80"}, "curl") - t.Logf("curl 1.1.1.1 (uncovered) → err=%v out=%q stderr=%q", e, so, se) - } + // Phase 2 — the GitOps source of truth is tampered with: primary is + // repointed at decoy-repo, a sibling Service on the same port that the + // role=helm-repo selector does not cover. source-controller fetches it on + // its own next reconcile. This is the lateral move a serviceCIDR entry hides. + t.Run("sibling_service_pivot_fires_r0011", func(t *testing.T) { + before := countRule("R0011") + patch := []byte(fmt.Sprintf(`{"spec":{"url":"http://decoy-repo.%s.svc.cluster.local./"}}`, ns.Name)) + _, e := repoClient.Patch(context.TODO(), "primary", types.MergePatchType, patch, metav1.PatchOptions{}) + require.NoError(t, e, "repoint HelmRepository at the decoy Service") require.Eventually(t, func() bool { - return countR0011(waitAlerts(t, wl.Namespace)) > before - }, 3*time.Minute, 15*time.Second, - "egress uncovered by serviceRef MUST fire R0011 — serviceRef is narrow, not a blanket allow") + return countRule("R0011") > before + }, 4*time.Minute, 15*time.Second, + "egress to an unlisted sibling Service MUST fire R0011 — the selector is narrow, not a blanket") }) } diff --git a/tests/resources/containerprofile-serviceref-network.yaml b/tests/resources/containerprofile-serviceref-network.yaml deleted file mode 100644 index 4db29ba9b2..0000000000 --- a/tests/resources/containerprofile-serviceref-network.yaml +++ /dev/null @@ -1,48 +0,0 @@ -apiVersion: spdx.softwarecomposition.kubescape.io/v1beta1 -kind: ContainerProfile -metadata: - name: serviceref-overlay -spec: - execs: - - path: /bin/sleep - - path: /usr/bin/curl - syscalls: - - socket - - connect - - sendto - - recvfrom - - read - - write - - close - - openat - - mmap - - mprotect - - munmap - - fcntl - - ioctl - - poll - - epoll_create1 - - epoll_ctl - - epoll_wait - - bind - - listen - - accept4 - - getsockopt - - setsockopt - - getsockname - - getpid - - fstat - - rt_sigaction - - rt_sigprocmask - - writev - matchLabels: - app: serviceref-client - egress: - - identifier: apiserver-serviceref - type: internal - serviceRefNamespace: default - serviceRefName: kubernetes - ports: - - name: TCP-443 - protocol: TCP - port: 443 diff --git a/tests/resources/serviceref-client-deployment.yaml b/tests/resources/serviceref-client-deployment.yaml deleted file mode 100644 index 7459155909..0000000000 --- a/tests/resources/serviceref-client-deployment.yaml +++ /dev/null @@ -1,21 +0,0 @@ -apiVersion: apps/v1 -kind: Deployment -metadata: - labels: - app: serviceref-client - name: serviceref-client -spec: - replicas: 1 - selector: - matchLabels: - app: serviceref-client - template: - metadata: - labels: - app: serviceref-client - kubescape.io/user-defined-profile: serviceref-overlay - spec: - containers: - - name: curl - image: docker.io/curlimages/curl@sha256:08e466006f0860e54fc299378de998935333e0e130a15f6f98482e9f8dab3058 - command: ["sleep", "infinity"] diff --git a/tests/resources/serviceref-suite/00-flux-source-crds.yaml b/tests/resources/serviceref-suite/00-flux-source-crds.yaml new file mode 100644 index 0000000000..bfc14c9c8d --- /dev/null +++ b/tests/resources/serviceref-suite/00-flux-source-crds.yaml @@ -0,0 +1,132 @@ +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: helmrepositories.source.toolkit.fluxcd.io +spec: + group: source.toolkit.fluxcd.io + names: + kind: HelmRepository + listKind: HelmRepositoryList + plural: helmrepositories + singular: helmrepository + scope: Namespaced + versions: + - name: v1 + served: true + storage: true + subresources: + status: {} + schema: + openAPIV3Schema: + type: object + x-kubernetes-preserve-unknown-fields: true +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: helmcharts.source.toolkit.fluxcd.io +spec: + group: source.toolkit.fluxcd.io + names: + kind: HelmChart + listKind: HelmChartList + plural: helmcharts + singular: helmchart + scope: Namespaced + versions: + - name: v1 + served: true + storage: true + subresources: + status: {} + schema: + openAPIV3Schema: + type: object + x-kubernetes-preserve-unknown-fields: true +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: gitrepositories.source.toolkit.fluxcd.io +spec: + group: source.toolkit.fluxcd.io + names: + kind: GitRepository + listKind: GitRepositoryList + plural: gitrepositories + singular: gitrepository + scope: Namespaced + versions: + - name: v1 + served: true + storage: true + subresources: + status: {} + schema: + openAPIV3Schema: + type: object + x-kubernetes-preserve-unknown-fields: true +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: buckets.source.toolkit.fluxcd.io +spec: + group: source.toolkit.fluxcd.io + names: + kind: Bucket + listKind: BucketList + plural: buckets + singular: bucket + scope: Namespaced + versions: + - name: v1 + served: true + storage: true + subresources: + status: {} + schema: + openAPIV3Schema: + type: object + x-kubernetes-preserve-unknown-fields: true + - name: v1beta2 + served: true + storage: false + subresources: + status: {} + schema: + openAPIV3Schema: + type: object + x-kubernetes-preserve-unknown-fields: true +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: ocirepositories.source.toolkit.fluxcd.io +spec: + group: source.toolkit.fluxcd.io + names: + kind: OCIRepository + listKind: OCIRepositoryList + plural: ocirepositories + singular: ocirepository + scope: Namespaced + versions: + - name: v1 + served: true + storage: true + subresources: + status: {} + schema: + openAPIV3Schema: + type: object + x-kubernetes-preserve-unknown-fields: true + - name: v1beta2 + served: true + storage: false + subresources: + status: {} + schema: + openAPIV3Schema: + type: object + x-kubernetes-preserve-unknown-fields: true diff --git a/tests/resources/serviceref-suite/10-helm-repo.yaml b/tests/resources/serviceref-suite/10-helm-repo.yaml new file mode 100644 index 0000000000..830be5de98 --- /dev/null +++ b/tests/resources/serviceref-suite/10-helm-repo.yaml @@ -0,0 +1,108 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: helm-repo-index +data: + index.yaml: | + apiVersion: v1 + entries: {} + generated: "2020-01-01T00:00:00Z" +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: helm-repo +spec: + replicas: 1 + selector: + matchLabels: + app: helm-repo + template: + metadata: + labels: + app: helm-repo + spec: + containers: + - name: nginx + image: nginx:1.27-alpine + ports: + - containerPort: 80 + volumeMounts: + - name: index + mountPath: /usr/share/nginx/html + resources: + limits: + cpu: 200m + memory: 128Mi + volumes: + - name: index + configMap: + name: helm-repo-index +--- +apiVersion: v1 +kind: Service +metadata: + name: helm-primary + labels: + role: helm-repo +spec: + selector: + app: helm-repo + ports: + - port: 80 + targetPort: 80 +--- +apiVersion: v1 +kind: Service +metadata: + name: helm-mirror + labels: + role: helm-repo +spec: + selector: + app: helm-repo + ports: + - port: 80 + targetPort: 80 +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: decoy-repo +spec: + replicas: 1 + selector: + matchLabels: + app: decoy-repo + template: + metadata: + labels: + app: decoy-repo + spec: + containers: + - name: nginx + image: nginx:1.27-alpine + ports: + - containerPort: 80 + volumeMounts: + - name: index + mountPath: /usr/share/nginx/html + resources: + limits: + cpu: 200m + memory: 128Mi + volumes: + - name: index + configMap: + name: helm-repo-index +--- +apiVersion: v1 +kind: Service +metadata: + name: decoy-repo +spec: + selector: + app: decoy-repo + ports: + - port: 80 + targetPort: 80 diff --git a/tests/resources/serviceref-suite/20-source-controller.yaml b/tests/resources/serviceref-suite/20-source-controller.yaml new file mode 100644 index 0000000000..613fcaae50 --- /dev/null +++ b/tests/resources/serviceref-suite/20-source-controller.yaml @@ -0,0 +1,117 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + name: source-controller +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: source-controller +rules: + - apiGroups: ["source.toolkit.fluxcd.io"] + resources: ["*"] + verbs: ["*"] + - apiGroups: [""] + resources: ["configmaps", "secrets", "events", "serviceaccounts"] + verbs: ["*"] + - apiGroups: ["coordination.k8s.io"] + resources: ["leases"] + verbs: ["*"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: source-controller +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: source-controller +subjects: + - kind: ServiceAccount + name: source-controller +--- +apiVersion: v1 +kind: Service +metadata: + name: source-controller +spec: + selector: + app: source-controller + ports: + - name: http + port: 80 + targetPort: 9090 +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: source-controller +spec: + replicas: 1 + selector: + matchLabels: + app: source-controller + strategy: + type: Recreate + template: + metadata: + labels: + app: source-controller + kubescape.io/user-defined-profile: serviceref-flux-cp + spec: + serviceAccountName: source-controller + terminationGracePeriodSeconds: 10 + containers: + - name: manager + image: ghcr.io/fluxcd/source-controller:v1.3.0 + imagePullPolicy: IfNotPresent + args: + - --log-level=info + - --log-encoding=json + - --storage-path=/data + - --storage-adv-addr=source-controller.$(RUNTIME_NAMESPACE).svc.cluster.local. + - --watch-all-namespaces=false + - --enable-leader-election=false + env: + - name: RUNTIME_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: TUF_ROOT + value: /tmp/.sigstore + ports: + - containerPort: 9090 + name: http + - containerPort: 9440 + name: healthz + readinessProbe: + httpGet: + path: / + port: http + resources: + limits: + cpu: 500m + memory: 512Mi + requests: + cpu: 50m + memory: 64Mi + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + readOnlyRootFilesystem: true + runAsNonRoot: true + seccompProfile: + type: RuntimeDefault + volumeMounts: + - mountPath: /data + name: data + - mountPath: /tmp + name: tmp + securityContext: + fsGroup: 1337 + volumes: + - name: data + emptyDir: {} + - name: tmp + emptyDir: {} diff --git a/tests/testutils/k8s.go b/tests/testutils/k8s.go index b2792d04b9..585af679a6 100644 --- a/tests/testutils/k8s.go +++ b/tests/testutils/k8s.go @@ -10,6 +10,7 @@ import ( "math/rand" "os" "path/filepath" + "sort" "strings" "testing" "time" @@ -28,10 +29,15 @@ import ( "github.com/stretchr/testify/assert" v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/labels" + "k8s.io/client-go/discovery" + "k8s.io/client-go/discovery/cached/memory" "k8s.io/client-go/dynamic" + "k8s.io/client-go/restmapper" "k8s.io/client-go/tools/remotecommand" "k8s.io/kubectl/pkg/scheme" ) @@ -84,6 +90,75 @@ func NewTestWorkload(namespace, resourcePath string) (*TestWorkload, error) { }, nil } +// ApplyMultiDocYAML creates every document in a multi-document YAML file, +// mapping each object's apiVersion/kind to its resource via server discovery so +// custom resources work without a compiled-in table. Namespaced objects land in +// namespace; cluster-scoped ones (CRDs) ignore it. Already-existing objects are +// not an error, so a fixture may be applied by more than one test. +func ApplyMultiDocYAML(namespace, resourcePath string) error { + k8sClient := k8sinterface.NewKubernetesApi() + raw, err := os.ReadFile(resourcePath) + if err != nil { + return err + } + dc, err := discovery.NewDiscoveryClientForConfig(k8sClient.K8SConfig) + if err != nil { + return err + } + mapper := restmapper.NewDeferredDiscoveryRESTMapper(memory.NewMemCacheClient(dc)) + for _, doc := range strings.Split(string(raw), "\n---") { + if strings.TrimSpace(doc) == "" { + continue + } + jsonData, err := yaml.YAMLToJSON([]byte(doc)) + if err != nil { + return fmt.Errorf("%s: %w", resourcePath, err) + } + obj := &unstructured.Unstructured{} + if err := obj.UnmarshalJSON(jsonData); err != nil { + return fmt.Errorf("%s: %w", resourcePath, err) + } + if obj.GetKind() == "" { + continue + } + gvk := obj.GroupVersionKind() + m, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version) + if err != nil { + return fmt.Errorf("%s %s: %w", resourcePath, gvk.String(), err) + } + var ri dynamic.ResourceInterface = k8sClient.DynamicClient.Resource(m.Resource) + if m.Scope.Name() == meta.RESTScopeNameNamespace { + ri = k8sClient.DynamicClient.Resource(m.Resource).Namespace(namespace) + } + if _, err := ri.Create(context.TODO(), obj, metav1.CreateOptions{}); err != nil && !apierrors.IsAlreadyExists(err) { + return fmt.Errorf("%s %s/%s: %w", resourcePath, obj.GetKind(), obj.GetName(), err) + } + } + return nil +} + +// ApplyMultiDocDir applies every YAML file in dir in lexical order (filenames +// are numbered so CRDs land before the resources that use them). +func ApplyMultiDocDir(namespace, dir string) error { + entries, err := os.ReadDir(dir) + if err != nil { + return err + } + var names []string + for _, e := range entries { + if !e.IsDir() { + names = append(names, e.Name()) + } + } + sort.Strings(names) + for _, n := range names { + if err := ApplyMultiDocYAML(namespace, filepath.Join(dir, n)); err != nil { + return err + } + } + return nil +} + func (w *TestWorkload) ExecIntoPod(command []string, container string) (string, string, error) { pods, err := w.GetPods() if err != nil { From 7d294ca3f1d3422eeea281d015b9767ae10b53f1 Mon Sep 17 00:00:00 2001 From: tanzee Date: Sun, 23 Aug 2026 23:31:53 +0200 Subject: [PATCH 3/5] test(component): scope the internal-egress rule to the serviceRef suite Relaxing the shipped R0011 to fire on private destinations made kube-dns egress alert for every workload that does not name it: Test_21 gained a spurious R0011 and Test_28 lost allowed_fusioncore_no_alert and mitm_coredns_poisoning. Restore the stock expression and express the internal-egress predicate as a test-only rule (R9911) bound by podSelector to this suite's pods, so nothing outside it changes. Verified on kind: Test_50 passes both phases against the stock ruleset, and Test_21 + all six Test_28 subtests are green again. Signed-off-by: tanzee --- .../templates/node-agent/default-rules.yaml | 2 +- tests/component_test.go | 25 +++++++++++++---- tests/resources/serviceref-rulebinding.yaml | 11 ++++++++ tests/resources/serviceref-rules.yaml | 27 +++++++++++++++++++ 4 files changed, 59 insertions(+), 6 deletions(-) create mode 100644 tests/resources/serviceref-rulebinding.yaml create mode 100644 tests/resources/serviceref-rules.yaml diff --git a/tests/chart/templates/node-agent/default-rules.yaml b/tests/chart/templates/node-agent/default-rules.yaml index 913fb1f927..512b4d9ec8 100644 --- a/tests/chart/templates/node-agent/default-rules.yaml +++ b/tests/chart/templates/node-agent/default-rules.yaml @@ -313,7 +313,7 @@ spec: uniqueId: "event.dstAddr + '_' + string(event.dstPort) + '_' + event.proto" ruleExpression: - eventType: "network" - expression: "event.pktType == 'OUTGOING' && !cp.was_address_in_egress(event.containerId, event.dstAddr)" + expression: "event.pktType == 'OUTGOING' && !net.is_private_ip(event.dstAddr) && !cp.was_address_in_egress(event.containerId, event.dstAddr)" profileDependency: 0 profileDataRequired: egressAddresses: all diff --git a/tests/component_test.go b/tests/component_test.go index fbb10b6211..a12c1008d0 100644 --- a/tests/component_test.go +++ b/tests/component_test.go @@ -3931,6 +3931,19 @@ func Test_50_ServiceRefNetworkNeighbor(t *testing.T) { ) port80, port443, port53 := int32(80), int32(443), int32(53) + // The shipped R0011 ignores private destinations, so in-cluster lateral + // movement — the very thing a named-Service allowlist narrows down — cannot + // trip it. R9911 is the same predicate restricted to internal addresses, + // applied for this test only and bound to this suite's pods, so no other + // namespace's expectations move. + rulesPath := path.Join(utils.CurrentDir(), "resources/serviceref-rules.yaml") + bindingPath := path.Join(utils.CurrentDir(), "resources/serviceref-rulebinding.yaml") + require.Equal(t, 0, testutils.RunCommand("kubectl", "apply", "--validate=false", "-f", rulesPath), "apply serviceRef test rules") + defer testutils.RunCommand("kubectl", "delete", "--ignore-not-found", "-f", rulesPath) + require.Equal(t, 0, testutils.RunCommand("kubectl", "apply", "--validate=false", "-f", bindingPath), "apply serviceRef test rule binding") + defer testutils.RunCommand("kubectl", "delete", "--ignore-not-found", "-f", bindingPath) + time.Sleep(20 * time.Second) + ns := testutils.NewRandomNamespace() k8sClient := k8sinterface.NewKubernetesApi() storageClient := spdxv1beta1client.NewForConfigOrDie(k8sClient.K8SConfig) @@ -4050,22 +4063,24 @@ func Test_50_ServiceRefNetworkNeighbor(t *testing.T) { } // Two further reconcile intervals of steady-state traffic. time.Sleep(90 * time.Second) + assert.Equal(t, 0, countRule("R9911"), + "apiserver/DNS/helm-repo egress is fully named by serviceRef+serviceSelector — no internal-egress alert may fire") assert.Equal(t, 0, countRule("R0011"), - "apiserver/DNS/helm-repo egress is fully named by serviceRef+serviceSelector — R0011 must NOT fire") + "no external egress is expected from the controller either") }) // Phase 2 — the GitOps source of truth is tampered with: primary is // repointed at decoy-repo, a sibling Service on the same port that the // role=helm-repo selector does not cover. source-controller fetches it on // its own next reconcile. This is the lateral move a serviceCIDR entry hides. - t.Run("sibling_service_pivot_fires_r0011", func(t *testing.T) { - before := countRule("R0011") + t.Run("sibling_service_pivot_fires_alert", func(t *testing.T) { + before := countRule("R9911") patch := []byte(fmt.Sprintf(`{"spec":{"url":"http://decoy-repo.%s.svc.cluster.local./"}}`, ns.Name)) _, e := repoClient.Patch(context.TODO(), "primary", types.MergePatchType, patch, metav1.PatchOptions{}) require.NoError(t, e, "repoint HelmRepository at the decoy Service") require.Eventually(t, func() bool { - return countRule("R0011") > before + return countRule("R9911") > before }, 4*time.Minute, 15*time.Second, - "egress to an unlisted sibling Service MUST fire R0011 — the selector is narrow, not a blanket") + "egress to an unlisted sibling Service MUST alert — the selector is narrow, not a blanket") }) } diff --git a/tests/resources/serviceref-rulebinding.yaml b/tests/resources/serviceref-rulebinding.yaml new file mode 100644 index 0000000000..24955a4841 --- /dev/null +++ b/tests/resources/serviceref-rulebinding.yaml @@ -0,0 +1,11 @@ +apiVersion: kubescape.io/v1 +kind: RuntimeRuleAlertBinding +metadata: + name: serviceref-test-binding +spec: + namespaceSelector: + podSelector: + matchLabels: + app: source-controller + rules: + - ruleName: "TEST internal egress not in profile" diff --git a/tests/resources/serviceref-rules.yaml b/tests/resources/serviceref-rules.yaml new file mode 100644 index 0000000000..b989ae570b --- /dev/null +++ b/tests/resources/serviceref-rules.yaml @@ -0,0 +1,27 @@ +apiVersion: kubescape.io/v1 +kind: Rules +metadata: + name: serviceref-test-rules + namespace: kubescape + labels: + app: kubescape +spec: + rules: + - name: "TEST internal egress not in profile" + enabled: true + id: "R9911" + description: "Test rule: egress to a cluster-internal address the profile does not allow. The shipped R0011 excludes private destinations, so in-cluster lateral movement — exactly what serviceRef/serviceSelector narrow down — is invisible to it. Scoped to the serviceRef suite via its own binding so no other test's namespace is affected." + expressions: + message: "'Unexpected internal egress to: ' + event.dstAddr + ':' + string(event.dstPort) + ' from: ' + event.containerName" + uniqueId: "'R9911_' + event.dstAddr + '_' + string(event.dstPort)" + ruleExpression: + - eventType: "network" + expression: "event.pktType == 'OUTGOING' && net.is_private_ip(event.dstAddr) && !cp.was_address_in_egress(event.containerId, event.dstAddr)" + profileDependency: 0 + profileDataRequired: + egressAddresses: all + severity: 5 + supportPolicy: false + isTriggerAlert: false + mitreTactic: "TA0008" + mitreTechnique: "T1210" From 8d5a019e985e80d78362ed8e23f52b4a59c5e07e Mon Sep 17 00:00:00 2001 From: tanzee Date: Mon, 24 Aug 2026 06:33:26 +0200 Subject: [PATCH 4/5] test(chart): make network service resolution a value Hardcoding the flag in the ConfigMap made it impossible to measure the feature's cost against itself. Expose it as nodeAgent.config.networkServiceResolution (on in the test chart, so Test_50 still exercises it) so an A/B can toggle resolution without rebuilding the image. Signed-off-by: tanzee --- tests/chart/templates/node-agent/configmap.yaml | 2 +- tests/chart/values.yaml | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/chart/templates/node-agent/configmap.yaml b/tests/chart/templates/node-agent/configmap.yaml index 6108427e90..053acf0808 100644 --- a/tests/chart/templates/node-agent/configmap.yaml +++ b/tests/chart/templates/node-agent/configmap.yaml @@ -14,7 +14,7 @@ data: "prometheusExporterEnabled": {{ eq .Values.nodeAgent.config.prometheusExporter "enable" }}, "runtimeDetectionEnabled": {{ eq .Values.capabilities.runtimeDetection "enable" }}, "networkServiceEnabled": {{ eq .Values.capabilities.networkPolicyService "enable" }}, - "networkServiceResolutionEnabled": true, + "networkServiceResolutionEnabled": {{ .Values.nodeAgent.config.networkServiceResolution | default false }}, "malwareDetectionEnabled": {{ eq .Values.capabilities.malwareDetection "enable" }}, "httpDetectionEnabled": {{ eq .Values.capabilities.httpDetection "enable" }}, "initialDelay": "{{ .Values.nodeAgent.config.learningPeriod }}", diff --git a/tests/chart/values.yaml b/tests/chart/values.yaml index 1aea3a150f..e6c87ef73d 100644 --- a/tests/chart/values.yaml +++ b/tests/chart/values.yaml @@ -58,6 +58,7 @@ nodeAgent: maxLearningPeriod: 2m learningPeriod: 1m updatePeriod: 30s + networkServiceResolution: true maxDelaySeconds: 1 prometheusExporter: enable httpExporterConfig: {} From 369fbe6bc0be2e08ebbe94d3347f5c4e7a5b23c9 Mon Sep 17 00:00:00 2001 From: tanzee Date: Mon, 24 Aug 2026 15:04:34 +0200 Subject: [PATCH 5/5] fix(cel): invalidate cached results when a profile is re-resolved MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The CEL result cache keys on SpecHash + SyncChecksum. Re-projecting a serviceRef/serviceSelector/entity profile against a moved cluster view changes neither: SpecHash tracks the rule projection spec, and SyncChecksum comes from a learned CP annotation an authored profile does not carry at all. So a result computed before the Service/EndpointSlice informers filled — 'this ClusterIP is not in egress' — was served from the LRU indefinitely, and the re-projection the lister generation correctly triggered had no observable effect. Egress to an allowlisted Service kept alerting. Carry the resolution generation on the projected profile and include it in the key, so the cache moves whenever the resolved addresses can have moved. Signed-off-by: tanzee --- .../containerprofilecache.go | 1 + .../containerprofilecache/reconciler.go | 1 + .../containerprofilecache/resolvedgen_test.go | 48 +++++++++++++++++++ pkg/objectcache/projection_types.go | 9 +++- .../cel/libraries/cache/function_cache.go | 7 ++- 5 files changed, 63 insertions(+), 3 deletions(-) create mode 100644 pkg/objectcache/containerprofilecache/resolvedgen_test.go diff --git a/pkg/objectcache/containerprofilecache/containerprofilecache.go b/pkg/objectcache/containerprofilecache/containerprofilecache.go index 410b39ec47..0c2b1574a3 100644 --- a/pkg/objectcache/containerprofilecache/containerprofilecache.go +++ b/pkg/objectcache/containerprofilecache/containerprofilecache.go @@ -617,6 +617,7 @@ func (c *ContainerProfileCacheImpl) buildEntry( entry.UsesServiceResolution = networkpeer.HasServiceNeighbors(userMerged) entry.ListerGen = c.listerGen() projected := Apply(spec, networkpeer.WithResolvedServiceNeighbors(userMerged, c.serviceLister), tree) + projected.ResolvedGen = entry.ListerGen entry.Projected = projected entry.SpecHash = projected.SpecHash diff --git a/pkg/objectcache/containerprofilecache/reconciler.go b/pkg/objectcache/containerprofilecache/reconciler.go index 85c4943aaf..b0fe54d8fe 100644 --- a/pkg/objectcache/containerprofilecache/reconciler.go +++ b/pkg/objectcache/containerprofilecache/reconciler.go @@ -497,6 +497,7 @@ func (c *ContainerProfileCacheImpl) rebuildEntryFromSources( spec := c.snapshotSpec() applyStart := time.Now() projectedCP := Apply(spec, networkpeer.WithResolvedServiceNeighbors(projected, c.serviceLister), tree) + projectedCP.ResolvedGen = c.listerGen() if c.cfg.ProfileProjection.DetailedMetricsEnabled { c.metricsManager.ObserveProjectionApplyDuration(time.Since(applyStart)) c.observeMemoryMetrics(projected, projectedCP) diff --git a/pkg/objectcache/containerprofilecache/resolvedgen_test.go b/pkg/objectcache/containerprofilecache/resolvedgen_test.go new file mode 100644 index 0000000000..7abdc844d1 --- /dev/null +++ b/pkg/objectcache/containerprofilecache/resolvedgen_test.go @@ -0,0 +1,48 @@ +package containerprofilecache + +import ( + "strconv" + "testing" + + "github.com/kubescape/node-agent/pkg/networkpeer" + "github.com/kubescape/node-agent/pkg/objectcache" +) + +// genLister is a cluster view whose generation the test drives directly. +type genLister struct{ gen int64 } + +func (g *genLister) ServiceByName(string, string) (*networkpeer.ServiceInfo, bool) { return nil, false } +func (g *genLister) ServicesByLabels(map[string]string, map[string]string) []*networkpeer.ServiceInfo { + return nil +} +func (g *genLister) HostIPs() []string { return nil } +func (g *genLister) Generation() int64 { return g.gen } + +// TestProjectedResolvedGenFeedsCacheKey: the CEL result cache keys on the +// projected profile's SpecHash+SyncChecksum+ResolvedGen. Re-resolving against a +// moved cluster view changes neither of the first two — an authored profile +// carries no SyncChecksum at all — so without ResolvedGen a result computed +// before the informers filled (e.g. "this address is not in egress") would be +// served from cache forever, and the profile's own re-projection would never +// take effect. +func TestProjectedResolvedGenFeedsCacheKey(t *testing.T) { + l := &genLister{gen: 7} + c := &ContainerProfileCacheImpl{} + c.SetServiceLister(l) + + if got := c.listerGen(); got != 7 { + t.Fatalf("listerGen: got %d want 7", got) + } + + key := func(p *objectcache.ProjectedContainerProfile) string { + return p.SpecHash + "|" + p.SyncChecksum + "|" + strconv.FormatInt(p.ResolvedGen, 10) + } + before := &objectcache.ProjectedContainerProfile{SpecHash: "spec", ResolvedGen: c.listerGen()} + + l.gen = 8 + after := &objectcache.ProjectedContainerProfile{SpecHash: "spec", ResolvedGen: c.listerGen()} + + if key(before) == key(after) { + t.Errorf("cache key must change when the profile is re-resolved against a moved cluster view") + } +} diff --git a/pkg/objectcache/projection_types.go b/pkg/objectcache/projection_types.go index 8d452c7a36..d76fe2b325 100644 --- a/pkg/objectcache/projection_types.go +++ b/pkg/objectcache/projection_types.go @@ -67,7 +67,14 @@ type ProjectedContainerProfile struct { // constraint" (back-compat for pre-projection profiles). ExecsByPath map[string][][]string - SpecHash string + SpecHash string + // ResolvedGen is the cluster-view generation the profile's serviceRef/ + // serviceSelector/entity neighbors were resolved against. It participates in + // the CEL result-cache key: re-projecting against a moved cluster view + // changes the projected addresses without touching SpecHash or SyncChecksum, + // so a result cached before the informers filled would otherwise be served + // forever. Zero for profiles that resolve nothing. + ResolvedGen int64 SyncChecksum string PolicyByRuleId map[string]v1beta1.RulePolicy CallStackTree *callstackcache.CallStackSearchTree diff --git a/pkg/rulemanager/cel/libraries/cache/function_cache.go b/pkg/rulemanager/cel/libraries/cache/function_cache.go index ba07eafcd3..22af6beed6 100644 --- a/pkg/rulemanager/cel/libraries/cache/function_cache.go +++ b/pkg/rulemanager/cel/libraries/cache/function_cache.go @@ -2,6 +2,7 @@ package cache import ( "fmt" + "strconv" "strings" "time" @@ -101,8 +102,10 @@ func HashForContainerProfile(oc objectcache.ObjectCache) func([]ref.Val) string } // Include SyncChecksum so the key changes when profile content is updated // under the same projection spec, preventing stale cached results after - // the profile learns new paths/execs/etc. - return pcp.SpecHash + "|" + pcp.SyncChecksum + // the profile learns new paths/execs/etc. ResolvedGen covers the same + // hazard for serviceRef/entity neighbors, whose projected addresses move + // with the cluster view while both other components stay put. + return pcp.SpecHash + "|" + pcp.SyncChecksum + "|" + strconv.FormatInt(pcp.ResolvedGen, 10) } }