From c2ca7a657609a3773d96a7b93fb7dcae00dfc59e Mon Sep 17 00:00:00 2001 From: Bowei Du Date: Tue, 16 Jun 2026 14:30:26 -0700 Subject: [PATCH 1/2] Add basic network policy for Ingress to Workers Restrict Ingress to Workers and hence Actors to be only from the Router component. This makes the atecontroller responsible for K8s level network policy Note: this is a belts-and-suspenders model as this applies on the Worker level. This will be refined further with Actor-specific policies. --- .../controllers/networkpolicy_controller.go | 140 ++++++++++++++++++ .../networkpolicy_controller_test.go | 93 ++++++++++++ .../controllers/workerpool_controller_test.go | 8 + cmd/atecontroller/main.go | 8 + manifests/ate-install/generated/role.yaml | 12 ++ 5 files changed, 261 insertions(+) create mode 100644 cmd/atecontroller/internal/controllers/networkpolicy_controller.go create mode 100644 cmd/atecontroller/internal/controllers/networkpolicy_controller_test.go diff --git a/cmd/atecontroller/internal/controllers/networkpolicy_controller.go b/cmd/atecontroller/internal/controllers/networkpolicy_controller.go new file mode 100644 index 000000000..0b980fd73 --- /dev/null +++ b/cmd/atecontroller/internal/controllers/networkpolicy_controller.go @@ -0,0 +1,140 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controllers + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + + networkingv1 "k8s.io/api/networking/v1" + k8errors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime" + metav1ac "k8s.io/client-go/applyconfigurations/meta/v1" + networkingv1ac "k8s.io/client-go/applyconfigurations/networking/v1" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/log" + + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" +) + +const networkPolicyFieldOwner = "ate-networkpolicy" + +type NetworkPolicyReconciler struct { + client.Client + Scheme *runtime.Scheme +} + +//+kubebuilder:rbac:groups=ate.dev,resources=workerpools,verbs=get;list;watch +//+kubebuilder:rbac:groups=networking.k8s.io,resources=networkpolicies,verbs=get;list;watch;create;update;patch;delete + +func (r *NetworkPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + log := log.FromContext(ctx) + + wp := &atev1alpha1.WorkerPool{} + if err := r.Get(ctx, req.NamespacedName, wp); err != nil { + if k8errors.IsNotFound(err) { + return ctrl.Result{}, nil + } + return ctrl.Result{}, fmt.Errorf("failed to get worker pool %q: %w", req.NamespacedName, err) + } + + if !wp.GetDeletionTimestamp().IsZero() { + log.Info("WorkerPool is being deleted, NetworkPolicy will be GC'd via OwnerReference", + "namespace", wp.Namespace, + "name", wp.Name) + return ctrl.Result{}, nil + } + + if err := r.reconcileImpl(ctx, wp); err != nil { + log.Error(err, "Failed to reconcile NetworkPolicy") + return ctrl.Result{}, err + } + + return ctrl.Result{}, nil +} + +func (r *NetworkPolicyReconciler) reconcileImpl(ctx context.Context, wp *atev1alpha1.WorkerPool) error { + log := log.FromContext(ctx) + + npAC := buildNetworkPolicyApplyConfig(wp) + + if err := r.Apply(ctx, npAC, client.FieldOwner(networkPolicyFieldOwner), client.ForceOwnership); err != nil { + return fmt.Errorf("failed to apply NetworkPolicy %s:%s: %w", *npAC.Namespace, *npAC.Name, err) + } + log.Info("reconcileImpl done", + "namespace", *npAC.Namespace, + "name", *npAC.Name) + + return nil +} + +func buildNetworkPolicyApplyConfig(wp *atev1alpha1.WorkerPool) *networkingv1ac.NetworkPolicyApplyConfiguration { + np := networkingv1ac.NetworkPolicy(npName(wp.Name), wp.Namespace). + WithOwnerReferences(metav1ac.OwnerReference(). + WithAPIVersion(atev1alpha1.GroupVersion.String()). + WithKind("WorkerPool"). + WithName(wp.Name). + WithUID(wp.UID). + WithController(true). + WithBlockOwnerDeletion(true)) + + // Ingress policy: only accept connections from the atenet-router, all ports. + np. + WithSpec(networkingv1ac.NetworkPolicySpec(). + WithPodSelector(metav1ac.LabelSelector(). + WithMatchLabels(map[string]string{"ate.dev/worker-pool": wp.Name})). + WithPolicyTypes(networkingv1.PolicyTypeIngress, networkingv1.PolicyTypeEgress). + WithIngress( + networkingv1ac.NetworkPolicyIngressRule(). + WithFrom( + networkingv1ac.NetworkPolicyPeer(). + WithNamespaceSelector(metav1ac.LabelSelector(). + WithMatchLabels(map[string]string{"kubernetes.io/metadata.name": "ate-system"})). + WithPodSelector(metav1ac.LabelSelector(). + WithMatchLabels(map[string]string{"app": "atenet-router"})), + ), + ). + WithEgress( + networkingv1ac.NetworkPolicyEgressRule(), + )) + + // TODO: don't implement any Egress policy yet. + + return np +} + +func npName(wpName string) string { + sum := sha256.Sum256([]byte(wpName)) + hash := hex.EncodeToString(sum[:]) + if len(hash) > 5 { + hash = hash[:5] + } + truncated := wpName + if len(truncated) > 30 { + truncated = truncated[:30] + } + return fmt.Sprintf("substrate-%s-%s", truncated, hash) +} + +func (r *NetworkPolicyReconciler) SetupWithManager(mgr ctrl.Manager) error { + return ctrl.NewControllerManagedBy(mgr). + Named("networkpolicy"). + For(&atev1alpha1.WorkerPool{}). + Owns(&networkingv1.NetworkPolicy{}). + Complete(r) +} diff --git a/cmd/atecontroller/internal/controllers/networkpolicy_controller_test.go b/cmd/atecontroller/internal/controllers/networkpolicy_controller_test.go new file mode 100644 index 000000000..209ac75bb --- /dev/null +++ b/cmd/atecontroller/internal/controllers/networkpolicy_controller_test.go @@ -0,0 +1,93 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controllers + +import ( + "context" + "testing" + + networkingv1 "k8s.io/api/networking/v1" + "k8s.io/apimachinery/pkg/types" +) + +func TestWorkerPoolCreatesNetworkPolicy(t *testing.T) { + wp := makeWorkerPool("test-netpolicy-create", "default", 2, "ateom:v1") + if err := k8sClient.Create(testCtx, wp); err != nil { + t.Fatalf("create WorkerPool: %v", err) + } + t.Cleanup(func() { k8sClient.Delete(testCtx, wp) }) //nolint:errcheck + + eventually(t, func(ctx context.Context) (bool, error) { + npName := npName(wp.Name) + np := &networkingv1.NetworkPolicy{} + err := k8sClient.Get(ctx, types.NamespacedName{Name: npName, Namespace: wp.Namespace}, np) + if err != nil { + return false, nil + } + + // Verify OwnerReference + if len(np.OwnerReferences) == 0 || np.OwnerReferences[0].Name != wp.Name { + return false, nil + } + + // Verify PodSelector matches the worker pool + if np.Spec.PodSelector.MatchLabels == nil || np.Spec.PodSelector.MatchLabels["ate.dev/worker-pool"] != wp.Name { + return false, nil + } + + // Verify PolicyTypes contains Ingress and Egress + hasIngress := false + hasEgress := false + for _, pt := range np.Spec.PolicyTypes { + if pt == networkingv1.PolicyTypeIngress { + hasIngress = true + } + if pt == networkingv1.PolicyTypeEgress { + hasEgress = true + } + } + if !hasIngress || !hasEgress { + return false, nil + } + + // Verify Ingress Rules (Allow only ingress from ATE router) + if len(np.Spec.Ingress) != 1 { + return false, nil + } + ingressRule := np.Spec.Ingress[0] + if len(ingressRule.From) != 1 { + return false, nil + } + fromPeer := ingressRule.From[0] + if fromPeer.NamespaceSelector == nil || fromPeer.NamespaceSelector.MatchLabels["kubernetes.io/metadata.name"] != "ate-system" { + return false, nil + } + if fromPeer.PodSelector == nil || fromPeer.PodSelector.MatchLabels["app"] != "atenet-router" { + return false, nil + } + + // Verify Egress Rules: + // For now, we allow all egress. + if len(np.Spec.Egress) != 1 { + return false, nil + } + egressRule := np.Spec.Egress[0] + if len(egressRule.To) != 0 || len(egressRule.Ports) != 0 { + return false, nil + } + + return true, nil + }) +} diff --git a/cmd/atecontroller/internal/controllers/workerpool_controller_test.go b/cmd/atecontroller/internal/controllers/workerpool_controller_test.go index 5acad6714..8d77ed04e 100644 --- a/cmd/atecontroller/internal/controllers/workerpool_controller_test.go +++ b/cmd/atecontroller/internal/controllers/workerpool_controller_test.go @@ -96,6 +96,14 @@ func TestMain(m *testing.M) { os.Exit(1) } + if err := (&NetworkPolicyReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + }).SetupWithManager(mgr); err != nil { + fmt.Fprintf(os.Stderr, "netpolicy controller setup failed: %v\n", err) + os.Exit(1) + } + testCtx, testCancel = context.WithCancel(context.Background()) go func() { _ = mgr.Start(testCtx) diff --git a/cmd/atecontroller/main.go b/cmd/atecontroller/main.go index e96bdb242..fa468fd10 100644 --- a/cmd/atecontroller/main.go +++ b/cmd/atecontroller/main.go @@ -95,6 +95,14 @@ func main() { os.Exit(1) } + if err = (&controllers.NetworkPolicyReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + }).SetupWithManager(mgr); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "NetPolicy") + os.Exit(1) + } + if err = (&controllers.ActorTemplateReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), diff --git a/manifests/ate-install/generated/role.yaml b/manifests/ate-install/generated/role.yaml index 7341d28dd..48f80e958 100644 --- a/manifests/ate-install/generated/role.yaml +++ b/manifests/ate-install/generated/role.yaml @@ -80,3 +80,15 @@ rules: - get - patch - update +- apiGroups: + - networking.k8s.io + resources: + - networkpolicies + verbs: + - create + - delete + - get + - list + - patch + - update + - watch From 5eba34ee14918208440a0fc603e0e89201cfe203 Mon Sep 17 00:00:00 2001 From: Haiyan Meng Date: Wed, 22 Jul 2026 18:43:02 +0000 Subject: [PATCH 2/2] feat(netpol): refine NetworkPolicy controller and add E2E verification - Add ate.dev/worker-pool label to NetworkPolicy metadata and remove Egress management to allow default egress behavior - Define ateSystemNamespace and atenetRouterAppName constants for Ingress peer selectors - Centralize deterministic NetworkPolicy naming logic (resources.NetworkPolicyName) and prevent trailing hyphens - Add comprehensive E2E test suite for networkpolicy covering control plane reconciliation, positive data plane verification (authorized ingress via atenet-router), and negative data plane verification (unauthorized ingress blocking) --- .../controllers/networkpolicy_controller.go | 40 +- .../networkpolicy_controller_test.go | 30 +- .../networkpolicy/networkpolicy_test.go | 401 ++++++++++++++++++ .../e2e/suites/networkpolicy/testmain_test.go | 28 ++ internal/resources/networkpolicy.go | 39 ++ internal/resources/networkpolicy_test.go | 70 +++ 6 files changed, 568 insertions(+), 40 deletions(-) create mode 100644 internal/e2e/suites/networkpolicy/networkpolicy_test.go create mode 100644 internal/e2e/suites/networkpolicy/testmain_test.go create mode 100644 internal/resources/networkpolicy.go create mode 100644 internal/resources/networkpolicy_test.go diff --git a/cmd/atecontroller/internal/controllers/networkpolicy_controller.go b/cmd/atecontroller/internal/controllers/networkpolicy_controller.go index 0b980fd73..928c053a9 100644 --- a/cmd/atecontroller/internal/controllers/networkpolicy_controller.go +++ b/cmd/atecontroller/internal/controllers/networkpolicy_controller.go @@ -16,8 +16,6 @@ package controllers import ( "context" - "crypto/sha256" - "encoding/hex" "fmt" networkingv1 "k8s.io/api/networking/v1" @@ -29,10 +27,15 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/log" + "github.com/agent-substrate/substrate/internal/resources" atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" ) -const networkPolicyFieldOwner = "ate-networkpolicy" +const ( + networkPolicyFieldOwner = "ate-networkpolicy" + ateSystemNamespace = "ate-system" + atenetRouterAppName = "atenet-router" +) type NetworkPolicyReconciler struct { client.Client @@ -84,7 +87,10 @@ func (r *NetworkPolicyReconciler) reconcileImpl(ctx context.Context, wp *atev1al } func buildNetworkPolicyApplyConfig(wp *atev1alpha1.WorkerPool) *networkingv1ac.NetworkPolicyApplyConfiguration { - np := networkingv1ac.NetworkPolicy(npName(wp.Name), wp.Namespace). + np := networkingv1ac.NetworkPolicy(resources.NetworkPolicyName(wp.Name), wp.Namespace). + WithLabels(map[string]string{ + "ate.dev/worker-pool": wp.Name, + }). WithOwnerReferences(metav1ac.OwnerReference(). WithAPIVersion(atev1alpha1.GroupVersion.String()). WithKind("WorkerPool"). @@ -98,39 +104,25 @@ func buildNetworkPolicyApplyConfig(wp *atev1alpha1.WorkerPool) *networkingv1ac.N WithSpec(networkingv1ac.NetworkPolicySpec(). WithPodSelector(metav1ac.LabelSelector(). WithMatchLabels(map[string]string{"ate.dev/worker-pool": wp.Name})). - WithPolicyTypes(networkingv1.PolicyTypeIngress, networkingv1.PolicyTypeEgress). + WithPolicyTypes(networkingv1.PolicyTypeIngress). WithIngress( networkingv1ac.NetworkPolicyIngressRule(). WithFrom( networkingv1ac.NetworkPolicyPeer(). WithNamespaceSelector(metav1ac.LabelSelector(). - WithMatchLabels(map[string]string{"kubernetes.io/metadata.name": "ate-system"})). + WithMatchLabels(map[string]string{"kubernetes.io/metadata.name": ateSystemNamespace})). WithPodSelector(metav1ac.LabelSelector(). - WithMatchLabels(map[string]string{"app": "atenet-router"})), + WithMatchLabels(map[string]string{"app": atenetRouterAppName})), ), - ). - WithEgress( - networkingv1ac.NetworkPolicyEgressRule(), )) - // TODO: don't implement any Egress policy yet. + // Egress is left unmanaged by Kubernetes NetworkPolicy for now while waiting for + // further progress on Egress API designs and because some egress rules may not be + // expressible at the Kubernetes layer. return np } -func npName(wpName string) string { - sum := sha256.Sum256([]byte(wpName)) - hash := hex.EncodeToString(sum[:]) - if len(hash) > 5 { - hash = hash[:5] - } - truncated := wpName - if len(truncated) > 30 { - truncated = truncated[:30] - } - return fmt.Sprintf("substrate-%s-%s", truncated, hash) -} - func (r *NetworkPolicyReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). Named("networkpolicy"). diff --git a/cmd/atecontroller/internal/controllers/networkpolicy_controller_test.go b/cmd/atecontroller/internal/controllers/networkpolicy_controller_test.go index 209ac75bb..a0208bbec 100644 --- a/cmd/atecontroller/internal/controllers/networkpolicy_controller_test.go +++ b/cmd/atecontroller/internal/controllers/networkpolicy_controller_test.go @@ -20,6 +20,8 @@ import ( networkingv1 "k8s.io/api/networking/v1" "k8s.io/apimachinery/pkg/types" + + "github.com/agent-substrate/substrate/internal/resources" ) func TestWorkerPoolCreatesNetworkPolicy(t *testing.T) { @@ -30,7 +32,7 @@ func TestWorkerPoolCreatesNetworkPolicy(t *testing.T) { t.Cleanup(func() { k8sClient.Delete(testCtx, wp) }) //nolint:errcheck eventually(t, func(ctx context.Context) (bool, error) { - npName := npName(wp.Name) + npName := resources.NetworkPolicyName(wp.Name) np := &networkingv1.NetworkPolicy{} err := k8sClient.Get(ctx, types.NamespacedName{Name: npName, Namespace: wp.Namespace}, np) if err != nil { @@ -42,23 +44,24 @@ func TestWorkerPoolCreatesNetworkPolicy(t *testing.T) { return false, nil } + // Verify metadata label matches the worker pool + if np.Labels == nil || np.Labels["ate.dev/worker-pool"] != wp.Name { + return false, nil + } + // Verify PodSelector matches the worker pool if np.Spec.PodSelector.MatchLabels == nil || np.Spec.PodSelector.MatchLabels["ate.dev/worker-pool"] != wp.Name { return false, nil } - // Verify PolicyTypes contains Ingress and Egress + // Verify PolicyTypes contains Ingress hasIngress := false - hasEgress := false for _, pt := range np.Spec.PolicyTypes { if pt == networkingv1.PolicyTypeIngress { hasIngress = true } - if pt == networkingv1.PolicyTypeEgress { - hasEgress = true - } } - if !hasIngress || !hasEgress { + if !hasIngress { return false, nil } @@ -71,20 +74,15 @@ func TestWorkerPoolCreatesNetworkPolicy(t *testing.T) { return false, nil } fromPeer := ingressRule.From[0] - if fromPeer.NamespaceSelector == nil || fromPeer.NamespaceSelector.MatchLabels["kubernetes.io/metadata.name"] != "ate-system" { + if fromPeer.NamespaceSelector == nil || fromPeer.NamespaceSelector.MatchLabels["kubernetes.io/metadata.name"] != ateSystemNamespace { return false, nil } - if fromPeer.PodSelector == nil || fromPeer.PodSelector.MatchLabels["app"] != "atenet-router" { + if fromPeer.PodSelector == nil || fromPeer.PodSelector.MatchLabels["app"] != atenetRouterAppName { return false, nil } - // Verify Egress Rules: - // For now, we allow all egress. - if len(np.Spec.Egress) != 1 { - return false, nil - } - egressRule := np.Spec.Egress[0] - if len(egressRule.To) != 0 || len(egressRule.Ports) != 0 { + // Verify Egress Rules are unmanaged (empty) + if len(np.Spec.Egress) != 0 { return false, nil } diff --git a/internal/e2e/suites/networkpolicy/networkpolicy_test.go b/internal/e2e/suites/networkpolicy/networkpolicy_test.go new file mode 100644 index 000000000..d2281e83a --- /dev/null +++ b/internal/e2e/suites/networkpolicy/networkpolicy_test.go @@ -0,0 +1,401 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package networkpolicy + +import ( + "context" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "strings" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + networkingv1 "k8s.io/api/networking/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/agent-substrate/substrate/internal/e2e" + "github.com/agent-substrate/substrate/internal/resources" + "github.com/agent-substrate/substrate/pkg/api/v1alpha1" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +const ( + ateSystemNamespace = "ate-system" + atenetRouterAppName = "atenet-router" +) + +func TestNetworkPolicyLifecycleAndReconciliation(t *testing.T) { + nsObj := e2e.CreateNamespace(t) + ctx := context.Background() + clients := e2e.GetClients() + + wpName := "test-wp-netpol" + wp := &v1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Name: wpName, + Namespace: nsObj.Name, + }, + Spec: v1alpha1.WorkerPoolSpec{ + Replicas: 1, + AteomImage: "ateom:v1", + }, + } + + t.Logf("Creating WorkerPool %s/%s...", nsObj.Name, wpName) + if _, err := clients.SubstrateK8s.ApiV1alpha1().WorkerPools(nsObj.Name).Create(ctx, wp, metav1.CreateOptions{}); err != nil { + t.Fatalf("failed to create WorkerPool: %v", err) + } + + expectedNpName := resources.NetworkPolicyName(wpName) + t.Logf("Waiting for NetworkPolicy %s to be created...", expectedNpName) + + var np *networkingv1.NetworkPolicy + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + var err error + np, err = clients.K8s.NetworkingV1().NetworkPolicies(nsObj.Name).Get(ctx, expectedNpName, metav1.GetOptions{}) + if err == nil { + break + } + time.Sleep(1 * time.Second) + } + + if np == nil { + t.Fatalf("timed out waiting for NetworkPolicy %s to be created in namespace %s", expectedNpName, nsObj.Name) + } + t.Logf("Successfully retrieved NetworkPolicy %s", expectedNpName) + + // Validate OwnerReference and Labels + if len(np.OwnerReferences) == 0 || np.OwnerReferences[0].Name != wpName { + t.Errorf("expected OwnerReference to %q, got %v", wpName, np.OwnerReferences) + } + if np.Labels == nil || np.Labels["ate.dev/worker-pool"] != wpName { + t.Errorf("expected label ate.dev/worker-pool=%q, got %v", wpName, np.Labels) + } + + // Validate PodSelector + if np.Spec.PodSelector.MatchLabels == nil || np.Spec.PodSelector.MatchLabels["ate.dev/worker-pool"] != wpName { + t.Errorf("expected PodSelector match label ate.dev/worker-pool=%q, got %v", wpName, np.Spec.PodSelector.MatchLabels) + } + + // Validate Ingress Rule + if len(np.Spec.Ingress) != 1 { + t.Fatalf("expected exactly 1 ingress rule, got %d", len(np.Spec.Ingress)) + } + ingressRule := np.Spec.Ingress[0] + if len(ingressRule.From) != 1 { + t.Fatalf("expected exactly 1 ingress from peer, got %d", len(ingressRule.From)) + } + fromPeer := ingressRule.From[0] + if fromPeer.NamespaceSelector == nil || fromPeer.NamespaceSelector.MatchLabels["kubernetes.io/metadata.name"] != ateSystemNamespace { + t.Errorf("expected namespace selector for %s, got %v", ateSystemNamespace, fromPeer.NamespaceSelector) + } + if fromPeer.PodSelector == nil || fromPeer.PodSelector.MatchLabels["app"] != atenetRouterAppName { + t.Errorf("expected pod selector for %s, got %v", atenetRouterAppName, fromPeer.PodSelector) + } + + // Validate Egress Rule is unmanaged (empty) + if len(np.Spec.Egress) != 0 { + t.Errorf("expected Egress to be empty/unmanaged, got %v", np.Spec.Egress) + } + + // Verify Garbage Collection upon deletion + t.Logf("Deleting WorkerPool %s/%s and verifying NetworkPolicy GC...", nsObj.Name, wpName) + if err := clients.SubstrateK8s.ApiV1alpha1().WorkerPools(nsObj.Name).Delete(ctx, wpName, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) { + t.Fatalf("failed to delete WorkerPool: %v", err) + } + + gcDeadline := time.Now().Add(60 * time.Second) + gcSuccess := false + for time.Now().Before(gcDeadline) { + _, err := clients.K8s.NetworkingV1().NetworkPolicies(nsObj.Name).Get(ctx, expectedNpName, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + gcSuccess = true + break + } + time.Sleep(2 * time.Second) + } + + if !gcSuccess { + t.Fatalf("timed out waiting for NetworkPolicy %s to be garbage collected after WorkerPool deletion", expectedNpName) + } + t.Logf("NetworkPolicy %s was successfully garbage collected", expectedNpName) +} + +func TestNetworkPolicyDataPlaneEnforcement(t *testing.T) { + nsObj := e2e.CreateNamespace(t) + ctx := context.Background() + clients := e2e.GetClients() + + // Setup WorkerPool and ActorTemplate from the standard counter demo + wp, at := setupDemoCounterTemplate(ctx, t, clients, nsObj.Name) + + // Create Atespace + if _, err := clients.SubstrateAPI.CreateAtespace(ctx, &ateapipb.CreateAtespaceRequest{ + Atespace: &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: nsObj.Name}}, + }); err != nil { + t.Fatalf("failed to create atespace: %v", err) + } + + // Create and Resume Actor + actorName := "netpol-dataplane-" + nsObj.Name + t.Logf("Creating Actor %q in Atespace %q...", actorName, nsObj.Name) + if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: nsObj.Name, Name: actorName}, + ActorTemplateNamespace: nsObj.Name, + ActorTemplateName: at.Name, + }}); err != nil { + t.Fatalf("failed to create Actor: %v", err) + } + defer func() { + _, _ = clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: nsObj.Name, Name: actorName}, + }) + _, _ = clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: nsObj.Name, Name: actorName}, + }) + }() + + t.Logf("Resuming Actor %q...", actorName) + if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: nsObj.Name, Name: actorName}, + }); err != nil { + t.Fatalf("failed to resume Actor: %v", err) + } + waitForActorRunning(ctx, t, clients, nsObj.Name, actorName) + + // === Positive Data Plane Verification (Authorized Ingress) === + t.Log("=== Verifying authorized ingress via atenet-router ===") + rc, err := e2e.NewRouterClient(ctx) + if err != nil { + t.Fatalf("NewRouterClient: %v", err) + } + defer rc.Close() + + var resp *http.Response + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + resp, err = rc.Get(ctx, nsObj.Name, actorName, "/") + if err == nil && resp.StatusCode == http.StatusOK { + break + } + if resp != nil { + resp.Body.Close() + } + time.Sleep(1 * time.Second) + } + + if err != nil { + t.Fatalf("Positive Data Plane Verification failed: GET / via atenet-router returned error: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + t.Fatalf("Positive Data Plane Verification failed: GET / via atenet-router status %d, body %q", resp.StatusCode, body) + } + body, _ := io.ReadAll(resp.Body) + t.Logf("Positive Data Plane Verification PASSED: Successfully accessed actor %q via authorized atenet-router (HTTP %d): %s", actorName, resp.StatusCode, strings.TrimSpace(string(body))) + + // === Negative Data Plane Verification (Unauthorized Ingress Blocking) === + t.Log("=== Verifying unauthorized ingress blocking ===") + rogueNsObj := e2e.CreateNamespace(t) + + // Find the IP address of the worker pod backing our WorkerPool + pods, err := clients.K8s.CoreV1().Pods(nsObj.Name).List(ctx, metav1.ListOptions{ + LabelSelector: fmt.Sprintf("ate.dev/worker-pool=%s", wp.Name), + }) + if err != nil || len(pods.Items) == 0 { + t.Fatalf("failed to list worker pods for worker pool %q: %v", wp.Name, err) + } + var workerIP string + for i := range pods.Items { + if pods.Items[i].Status.Phase == corev1.PodRunning && pods.Items[i].Status.PodIP != "" { + workerIP = pods.Items[i].Status.PodIP + t.Logf("Target worker pod %s/%s has IP %s", nsObj.Name, pods.Items[i].Name, workerIP) + break + } + } + if workerIP == "" { + t.Fatalf("no running worker pod with IP found for worker pool %q", wp.Name) + } + + // Deploy an unauthorized probe pod in an external test namespace + probePod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "unauthorized-probe", + Namespace: rogueNsObj.Name, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + { + Name: "probe", + Image: "busybox@sha256:1487d0af5f52b4ba31c7e465126ee2123fe3f2305d638e7827681e7cf6c83d5e", + Command: []string{"/bin/sleep", "3600"}, + }, + }, + RestartPolicy: corev1.RestartPolicyNever, + }, + } + t.Logf("Creating unauthorized probe pod %s/%s...", rogueNsObj.Name, probePod.Name) + if _, err := clients.K8s.CoreV1().Pods(rogueNsObj.Name).Create(ctx, probePod, metav1.CreateOptions{}); err != nil { + t.Fatalf("failed to create probe pod: %v", err) + } + waitForPodRunning(ctx, t, clients, rogueNsObj.Name, probePod.Name) + + // Attempt direct connection to worker pod IP from unauthorized probe pod + execArgs := []string{"exec", "-n", rogueNsObj.Name, probePod.Name} + if e2e.KubeContext != "" { + execArgs = append([]string{"--context=" + e2e.KubeContext}, execArgs...) + } + if e2e.KubeConfig != "" { + execArgs = append([]string{"--kubeconfig=" + e2e.KubeConfig}, execArgs...) + } + execArgs = append(execArgs, "--", "sh", "-c", fmt.Sprintf("wget -T 3 -q -O - http://%s:8080/ || nc -z -w 3 %s 8080", workerIP, workerIP)) + + t.Logf("Executing unauthorized connection attempt from %s/%s to %s:8080...", rogueNsObj.Name, probePod.Name, workerIP) + cmd := exec.Command("kubectl", execArgs...) + output, err := cmd.CombinedOutput() + if err == nil { + t.Fatalf("Negative Data Plane Verification failed: unauthorized probe pod successfully connected to worker pod %s:8080 (output: %s). NetworkPolicy did not block the connection!", workerIP, string(output)) + } + t.Logf("Negative Data Plane Verification PASSED: Unauthorized connection attempt from %s/%s to %s:8080 was blocked as expected (err: %v)", rogueNsObj.Name, probePod.Name, workerIP, err) +} + +func setupDemoCounterTemplate(ctx context.Context, t *testing.T, clients *e2e.Clients, ns string) (*v1alpha1.WorkerPool, *v1alpha1.ActorTemplate) { + t.Helper() + srcNS := "ate-demo-counter" + if v := os.Getenv("E2E_TEMPLATE_NAMESPACE"); v != "" { + srcNS = v + } + srcName := "counter" + if v := os.Getenv("E2E_TEMPLATE_NAME"); v != "" { + srcName = v + } + + existingWp, err := clients.SubstrateK8s.ApiV1alpha1().WorkerPools(srcNS).Get(ctx, srcName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to get existing WorkerPool %s/%s: %v", srcNS, srcName, err) + } + + existingAt, err := clients.SubstrateK8s.ApiV1alpha1().ActorTemplates(srcNS).Get(ctx, srcName, metav1.GetOptions{}) + if err != nil { + t.Fatalf("failed to get existing ActorTemplate %s/%s: %v", srcNS, srcName, err) + } + + wp := &v1alpha1.WorkerPool{ + ObjectMeta: metav1.ObjectMeta{ + Name: "counter", + Namespace: ns, + Labels: map[string]string{"netpol-test": ns}, + }, + Spec: v1alpha1.WorkerPoolSpec{ + Replicas: 1, + AteomImage: existingWp.Spec.AteomImage, + SandboxClass: existingWp.Spec.SandboxClass, + SandboxConfigName: existingWp.Spec.SandboxConfigName, + }, + } + if _, err := clients.SubstrateK8s.ApiV1alpha1().WorkerPools(ns).Create(ctx, wp, metav1.CreateOptions{}); err != nil { + t.Fatalf("failed to create WorkerPool: %v", err) + } + + at := &v1alpha1.ActorTemplate{ + ObjectMeta: metav1.ObjectMeta{ + Name: "counter", + Namespace: ns, + }, + Spec: v1alpha1.ActorTemplateSpec{ + WorkerSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"netpol-test": ns}, + }, + SandboxClass: existingAt.Spec.SandboxClass, + PauseImage: existingAt.Spec.PauseImage, + Containers: existingAt.Spec.Containers, + SnapshotsConfig: existingAt.Spec.SnapshotsConfig, + Volumes: existingAt.Spec.Volumes, + }, + } + if _, err := clients.SubstrateK8s.ApiV1alpha1().ActorTemplates(ns).Create(ctx, at, metav1.CreateOptions{}); err != nil { + t.Fatalf("failed to create ActorTemplate: %v", err) + } + + t.Logf("Waiting for ActorTemplate %s/%s to be Ready...", ns, at.Name) + tmplTimeout := 90 * time.Second + if v := os.Getenv("E2E_TEMPLATE_READY_TIMEOUT"); v != "" { + d, perr := time.ParseDuration(v) + if perr != nil { + t.Fatalf("invalid E2E_TEMPLATE_READY_TIMEOUT %q: %v", v, perr) + } + tmplTimeout = d + } + tmplCtx, tmplCancel := context.WithTimeout(ctx, tmplTimeout) + defer tmplCancel() + var lastPhase v1alpha1.PhaseType + for { + curAt, err := clients.SubstrateK8s.ApiV1alpha1().ActorTemplates(ns).Get(tmplCtx, at.Name, metav1.GetOptions{}) + if err == nil { + lastPhase = curAt.Status.Phase + if lastPhase == v1alpha1.PhaseReady { + t.Logf("ActorTemplate %s/%s is Ready with golden snapshot %q", ns, at.Name, curAt.Status.GoldenSnapshot) + break + } + if lastPhase == v1alpha1.PhaseFailed { + t.Fatalf("ActorTemplate %s/%s transitioned to PhaseFailed!", ns, at.Name) + } + } + select { + case <-tmplCtx.Done(): + t.Fatalf("Timed out waiting for ActorTemplate %q to be Ready after %v (last phase: %s, err: %v)", at.Name, tmplTimeout, lastPhase, err) + case <-time.After(1 * time.Second): + } + } + + return wp, at +} + +func waitForActorRunning(ctx context.Context, t *testing.T, clients *e2e.Clients, atespace, actorName string) { + t.Helper() + deadline := time.Now().Add(60 * time.Second) + for time.Now().Before(deadline) { + resp, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: atespace, Name: actorName}, + }) + if err == nil && resp.GetStatus() == ateapipb.Actor_STATUS_RUNNING { + t.Logf("Actor %q reached RUNNING status", actorName) + return + } + time.Sleep(1 * time.Second) + } + t.Fatalf("timed out waiting for actor %q to reach RUNNING status", actorName) +} + +func waitForPodRunning(ctx context.Context, t *testing.T, clients *e2e.Clients, namespace, podName string) { + t.Helper() + deadline := time.Now().Add(60 * time.Second) + for time.Now().Before(deadline) { + pod, err := clients.K8s.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{}) + if err == nil && pod.Status.Phase == corev1.PodRunning { + t.Logf("Pod %s/%s is RUNNING", namespace, podName) + return + } + time.Sleep(1 * time.Second) + } + t.Fatalf("timed out waiting for pod %s/%s to reach RUNNING status", namespace, podName) +} diff --git a/internal/e2e/suites/networkpolicy/testmain_test.go b/internal/e2e/suites/networkpolicy/testmain_test.go new file mode 100644 index 000000000..6a686e2ed --- /dev/null +++ b/internal/e2e/suites/networkpolicy/testmain_test.go @@ -0,0 +1,28 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package networkpolicy + +import ( + "os" + "testing" + + "github.com/agent-substrate/substrate/internal/e2e" +) + +func run(m *testing.M) int { + return e2e.RunTestMain(m) +} + +func TestMain(m *testing.M) { os.Exit(run(m)) } diff --git a/internal/resources/networkpolicy.go b/internal/resources/networkpolicy.go new file mode 100644 index 000000000..3bccd6a87 --- /dev/null +++ b/internal/resources/networkpolicy.go @@ -0,0 +1,39 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package resources + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" +) + +// NetworkPolicyName returns the deterministic Kubernetes NetworkPolicy name +// generated for a given WorkerPool name. +func NetworkPolicyName(wpName string) string { + sum := sha256.Sum256([]byte(wpName)) + hash := hex.EncodeToString(sum[:]) + if len(hash) > 5 { + hash = hash[:5] + } + truncated := wpName + if len(truncated) > 30 { + truncated = truncated[:30] + } + // Trim trailing hyphens to prevent double hyphens when appending "-". + truncated = strings.TrimRight(truncated, "-") + return fmt.Sprintf("substrate-%s-%s", truncated, hash) +} diff --git a/internal/resources/networkpolicy_test.go b/internal/resources/networkpolicy_test.go new file mode 100644 index 000000000..933b712d3 --- /dev/null +++ b/internal/resources/networkpolicy_test.go @@ -0,0 +1,70 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package resources + +import ( + "strings" + "testing" +) + +func TestNetworkPolicyName(t *testing.T) { + tests := []struct { + name string + wpName string + wantPrefix string + }{ + { + name: "short workerpool name", + wpName: "my-wp", + wantPrefix: "substrate-my-wp-", + }, + { + name: "long workerpool name truncated at 30 chars", + wpName: "a-very-long-workerpool-name-that-exceeds-30-chars", + wantPrefix: "substrate-a-very-long-workerpool-name-th-", + }, + { + name: "workerpool name truncated ending with a hyphen", + wpName: "this-is-a-long-worker-pool-n-with-hyphen-at-30", + // "this-is-a-long-worker-pool-n-" is 30 chars long, ending in a hyphen. + // Trimming trailing hyphens turns it into "this-is-a-long-worker-pool-n". + wantPrefix: "substrate-this-is-a-long-worker-pool-n-", + }, + { + name: "workerpool name ending with hyphens", + wpName: "workerpool-name---", + wantPrefix: "substrate-workerpool-name-", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := NetworkPolicyName(tt.wpName) + + if !strings.HasPrefix(got, tt.wantPrefix) { + t.Errorf("NetworkPolicyName(%q) = %q, want prefix %q", tt.wpName, got, tt.wantPrefix) + } + + // Ensure no double hyphens before the hash + // The format is substrate-- + parts := strings.Split(got, "-") + for i, part := range parts { + if part == "" && i > 0 && i < len(parts)-1 { + t.Errorf("NetworkPolicyName(%q) = %q contains empty part (double hyphen)", tt.wpName, got) + } + } + }) + } +}