diff --git a/cmd/atecontroller/internal/controllers/networkpolicy_controller.go b/cmd/atecontroller/internal/controllers/networkpolicy_controller.go new file mode 100644 index 000000000..928c053a9 --- /dev/null +++ b/cmd/atecontroller/internal/controllers/networkpolicy_controller.go @@ -0,0 +1,132 @@ +// 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" + "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" + + "github.com/agent-substrate/substrate/internal/resources" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" +) + +const ( + networkPolicyFieldOwner = "ate-networkpolicy" + ateSystemNamespace = "ate-system" + atenetRouterAppName = "atenet-router" +) + +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(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"). + 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). + WithIngress( + networkingv1ac.NetworkPolicyIngressRule(). + WithFrom( + networkingv1ac.NetworkPolicyPeer(). + WithNamespaceSelector(metav1ac.LabelSelector(). + WithMatchLabels(map[string]string{"kubernetes.io/metadata.name": ateSystemNamespace})). + WithPodSelector(metav1ac.LabelSelector(). + WithMatchLabels(map[string]string{"app": atenetRouterAppName})), + ), + )) + + // 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 (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..a0208bbec --- /dev/null +++ b/cmd/atecontroller/internal/controllers/networkpolicy_controller_test.go @@ -0,0 +1,91 @@ +// 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" + + "github.com/agent-substrate/substrate/internal/resources" +) + +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 := resources.NetworkPolicyName(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 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 + hasIngress := false + for _, pt := range np.Spec.PolicyTypes { + if pt == networkingv1.PolicyTypeIngress { + hasIngress = true + } + } + if !hasIngress { + 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"] != ateSystemNamespace { + return false, nil + } + if fromPeer.PodSelector == nil || fromPeer.PodSelector.MatchLabels["app"] != atenetRouterAppName { + return false, nil + } + + // Verify Egress Rules are unmanaged (empty) + if len(np.Spec.Egress) != 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/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) + } + } + }) + } +} 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