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..bfec59e977 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,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. 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) + } 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/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 new file mode 100644 index 0000000000..747c7b2d62 --- /dev/null +++ b/pkg/networkpeer/expand.go @@ -0,0 +1,156 @@ +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 +// (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 +// 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) + 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, + }) + } + 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) { + // 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 != "": + 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..715a76e607 --- /dev/null +++ b/pkg/networkpeer/expand_test.go @@ -0,0 +1,217 @@ +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) + } + // 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 + +// 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)) + } + // 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) + 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..5c88c7d7ba --- /dev/null +++ b/pkg/networkpeer/lister.go @@ -0,0 +1,192 @@ +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) +} + +// 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 + } + 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/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 new file mode 100644 index 0000000000..de8c1375d7 --- /dev/null +++ b/pkg/networkpeer/resolve.go @@ -0,0 +1,228 @@ +// 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" + +// 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 { + 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 { + 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 []*ServiceInfo{svc} + case spec.ServiceSelector != nil: + if len(spec.ServiceSelector) == 0 { + return nil + } + 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...) + 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..0ddcc0ccbe --- /dev/null +++ b/pkg/networkpeer/resolve_test.go @@ -0,0 +1,228 @@ +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 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 { + 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..0c2b1574a3 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,15 @@ 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) + 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 bd7517c728..b0fe54d8fe 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,30 @@ 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) + projectedCP.ResolvedGen = c.listerGen() 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/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) } } 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..053acf0808 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": {{ .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: {} diff --git a/tests/component_test.go b/tests/component_test.go index 154e441e8b..a12c1008d0 100644 --- a/tests/component_test.go +++ b/tests/component_test.go @@ -3906,3 +3906,181 @@ 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 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) + + const ( + cpName = "serviceref-flux-cp" + containerName = "manager" + ) + 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) + + // 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() + 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) + } + 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"] == ruleID && a.Labels["container_name"] == containerName { + n++ + } + } + return n + } + + 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 + } + + // Let node-agent bind the profile and fill its Service/EndpointSlice caches + // before any reconcile traffic is judged. + time.Sleep(40 * time.Second) + + // 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) + } + 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("R9911"), + "apiserver/DNS/helm-repo egress is fully named by serviceRef+serviceSelector — no internal-egress alert may fire") + assert.Equal(t, 0, countRule("R0011"), + "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_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("R9911") > before + }, 4*time.Minute, 15*time.Second, + "egress to an unlisted sibling Service MUST alert — the selector is narrow, not a blanket") + }) +} 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-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" 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 {