diff --git a/README.md b/README.md index f8c0f98..324a975 100644 --- a/README.md +++ b/README.md @@ -245,7 +245,9 @@ controls that customers can map into their own security and compliance program. The Helm chart renders default-deny egress NetworkPolicies for selected pods when `networkPolicy.enabled=true`. It can also create a gVisor `RuntimeClass` and register a mutating webhook for labeled pods. The cluster must provide an -enforcing CNI and install `runsc`. See [docs/07-security.md](docs/07-security.md). +enforcing CNI and install `runsc`. `agentctl doctor network-policy` reports +passive CNI evidence. Its `--active-probe` mode tests scratch-namespace packet +flow when given an updated operator image. See [docs/07-security.md](docs/07-security.md). --- diff --git a/cmd/agentctl/doctor_sandbox.go b/cmd/agentctl/doctor_sandbox.go index 83f4c6b..f751796 100644 --- a/cmd/agentctl/doctor_sandbox.go +++ b/cmd/agentctl/doctor_sandbox.go @@ -5,20 +5,25 @@ import ( "errors" "fmt" "io" + "time" "github.com/spf13/cobra" corev1 "k8s.io/api/core/v1" nodev1 "k8s.io/api/node/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "sigs.k8s.io/controller-runtime/pkg/client" "github.com/Clawdlinux/agentic-operator-core/internal/admission" + "github.com/Clawdlinux/agentic-operator-core/internal/netpolicy" ) func newDoctorCommand(opts *cliOptions) *cobra.Command { command := &cobra.Command{Use: "doctor", Short: "Check cluster readiness"} command.AddCommand(newSandboxDoctorCommand(opts)) + command.AddCommand(newNetworkPolicyDoctorCommand(opts)) return command } @@ -85,3 +90,64 @@ func runSandboxDoctor(ctx context.Context, reader client.Reader, runtimeClassNam } return errors.New("sandbox is not ready") } + +func newNetworkPolicyDoctorCommand(opts *cliOptions) *cobra.Command { + var activeProbe bool + var probeImage string + var timeout time.Duration + var allowKnownNonEnforcing bool + var keepNamespace bool + command := &cobra.Command{ + Use: "network-policy", + Short: "Detect NetworkPolicy CNI support or run an explicit packet-flow probe", + RunE: func(command *cobra.Command, _ []string) error { + return runNetworkPolicyDoctor(command.Context(), opts.kube, opts.discovery, networkPolicyDoctorOptions{activeProbe: activeProbe, probeImage: probeImage, timeout: timeout, keepNamespace: keepNamespace, allowKnownNonEnforcing: allowKnownNonEnforcing}, command.OutOrStdout()) + }, + } + command.Flags().BoolVar(&activeProbe, "active-probe", false, "Create short-lived probe Pods and a deny-egress NetworkPolicy") + command.Flags().StringVar(&probeImage, "probe-image", "", "Operator image that includes the netprobe command (required with --active-probe)") + command.Flags().DurationVar(&timeout, "timeout", 90*time.Second, "Maximum duration for the active probe (110s maximum)") + command.Flags().BoolVar(&allowKnownNonEnforcing, "probe-known-non-enforcing", false, "Allow active probing despite a known non-enforcing CNI fingerprint") + command.Flags().BoolVar(&keepNamespace, "keep-namespace", false, "Keep the probe namespace for inspection") + return command +} + +type networkPolicyDoctorOptions struct { + activeProbe bool + probeImage string + timeout time.Duration + keepNamespace bool + allowKnownNonEnforcing bool +} + +func runNetworkPolicyDoctor(ctx context.Context, kube kubernetes.Interface, discoveryClient netpolicy.DiscoveryClient, options networkPolicyDoctorOptions, writer io.Writer) error { + if kube == nil || discoveryClient == nil { + return errors.New("kubernetes clients not initialized; check kubeconfig") + } + daemonSets := kube.AppsV1().DaemonSets(metav1.NamespaceSystem) + detection, err := netpolicy.DetectCNIEnforcement(ctx, discoveryClient, daemonSets) + if err != nil { + return fmt.Errorf("detect NetworkPolicy enforcement: %w", err) + } + if !options.activeProbe { + _, _ = fmt.Fprintf(writer, "NetworkPolicy CNI: %s\nConfigured only. Active packet probe not run.\n", detection.Enforcement) + return nil + } + result, err := netpolicy.RunActiveProbe(ctx, kube, discoveryClient, daemonSets, netpolicy.ActiveProbeOptions{ProbeImage: options.probeImage, Timeout: options.timeout, KeepNamespace: options.keepNamespace, AllowKnownNonEnforcing: options.allowKnownNonEnforcing}) + if err != nil { + return err + } + _, _ = fmt.Fprintf(writer, "NetworkPolicy probe: %s\n%s\n", result.Verdict, result.Reason) + if result.Namespace != "" { + _, _ = fmt.Fprintf(writer, "Probe namespace: %s\n", result.Namespace) + if result.Cleaned { + _, _ = fmt.Fprintln(writer, "Cleanup: namespace deletion accepted") + } else if options.keepNamespace { + _, _ = fmt.Fprintln(writer, "Cleanup: namespace kept for inspection") + } + } + if result.Verdict != netpolicy.ProbeVerdictEnforcing { + return fmt.Errorf("NetworkPolicy probe result: %s: %s", result.Verdict, result.Reason) + } + return nil +} diff --git a/cmd/agentctl/doctor_sandbox_test.go b/cmd/agentctl/doctor_sandbox_test.go index b5a0fe7..a0ea2ff 100644 --- a/cmd/agentctl/doctor_sandbox_test.go +++ b/cmd/agentctl/doctor_sandbox_test.go @@ -11,6 +11,7 @@ import ( nodev1 "k8s.io/api/node/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + kubefake "k8s.io/client-go/kubernetes/fake" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" ) @@ -124,3 +125,57 @@ func TestNewDoctorCommandIncludesSandbox(t *testing.T) { t.Fatalf("doctor command Use = %q", command.Use) } } + +func TestNewDoctorCommandIncludesNetworkPolicy(t *testing.T) { + command := newDoctorCommand(&cliOptions{}) + child, _, err := command.Find([]string{"network-policy"}) + if err != nil || child == nil || child.Name() != "network-policy" { + t.Fatalf("network-policy command = %v, %v, want network-policy", child, err) + } + if flag := child.Flags().Lookup("active-probe"); flag == nil || flag.DefValue != "false" { + t.Fatalf("active-probe flag = %#v, want default false", flag) + } +} + +func TestRunNetworkPolicyDoctorPassiveDoesNotCreateResources(t *testing.T) { + kube := kubefake.NewClientset() + var output bytes.Buffer + err := runNetworkPolicyDoctor(context.Background(), kube, doctorNetworkPolicyDiscovery{groups: []metav1.APIGroup{{Name: "cilium.io"}}}, networkPolicyDoctorOptions{}, &output) + if err != nil { + t.Fatalf("runNetworkPolicyDoctor() error = %v", err) + } + if output.String() != "NetworkPolicy CNI: Enforcing\nConfigured only. Active packet probe not run.\n" { + t.Fatalf("output = %q", output.String()) + } + for _, action := range kube.Actions() { + if action.GetVerb() == "create" || action.GetVerb() == "delete" || action.GetVerb() == "patch" || action.GetVerb() == "update" { + t.Fatalf("passive command performed mutation: %#v", action) + } + } +} + +func TestRunNetworkPolicyDoctorRequiresImageBeforeActiveProbe(t *testing.T) { + kube := kubefake.NewClientset() + var output bytes.Buffer + err := runNetworkPolicyDoctor(context.Background(), kube, doctorNetworkPolicyDiscovery{}, networkPolicyDoctorOptions{activeProbe: true}, &output) + if err == nil { + t.Fatal("runNetworkPolicyDoctor() error = nil, want missing image error") + } + for _, action := range kube.Actions() { + if action.GetVerb() == "create" || action.GetVerb() == "delete" || action.GetVerb() == "patch" || action.GetVerb() == "update" { + t.Fatalf("active command without image performed mutation: %#v", action) + } + } +} + +type doctorNetworkPolicyDiscovery struct { + groups []metav1.APIGroup + err error +} + +func (d doctorNetworkPolicyDiscovery) ServerGroups() (*metav1.APIGroupList, error) { + if d.err != nil { + return nil, d.err + } + return &metav1.APIGroupList{Groups: d.groups}, nil +} diff --git a/cmd/main.go b/cmd/main.go index 233626f..929527e 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -56,6 +56,7 @@ import ( runtimeadmission "github.com/Clawdlinux/agentic-operator-core/internal/admission" "github.com/Clawdlinux/agentic-operator-core/internal/controller" "github.com/Clawdlinux/agentic-operator-core/internal/netpolicy" + "github.com/Clawdlinux/agentic-operator-core/internal/netpolicy/netprobe" "github.com/Clawdlinux/agentic-operator-core/pkg/evaluation" "github.com/Clawdlinux/agentic-operator-core/pkg/finops" "github.com/Clawdlinux/agentic-operator-core/pkg/governance" @@ -81,6 +82,10 @@ func init() { // nolint:gocyclo func main() { + if len(os.Args) > 1 && os.Args[1] == "netprobe" { + os.Exit(netprobe.Run(os.Args[2:], os.Stdout, os.Stderr)) + } + var metricsAddr string var metricsCertPath, metricsCertName, metricsCertKey string var webhookCertPath, webhookCertName, webhookCertKey string diff --git a/docs/07-security.md b/docs/07-security.md index 9530ac8..c6de713 100644 --- a/docs/07-security.md +++ b/docs/07-security.md @@ -73,6 +73,17 @@ external OPA (when configured). Operator-supplied additions go under Managed workload namespaces require separate policy application and matching labels. Optional Cilium FQDN policies require Cilium and explicit chart configuration. +Run `agentctl doctor network-policy` for passive CNI detection. It does not +verify packet enforcement. To run an intrusive scratch-namespace test, use: + +```bash +agentctl doctor network-policy --active-probe --probe-image +``` + +The active probe requires your kubeconfig identity to create and delete a +namespace, Pods, a Service, and a NetworkPolicy. It proves only that a +deny-egress policy blocked TCP to a test Service in the scratch namespace. + ## Sandbox Technology Agent code should be treated as untrusted because the agent's instructions diff --git a/docs/10-troubleshooting.md b/docs/10-troubleshooting.md index 662d430..99c93f8 100644 --- a/docs/10-troubleshooting.md +++ b/docs/10-troubleshooting.md @@ -93,6 +93,12 @@ Confirm: Policy-object presence alone does not prove packet enforcement. +Use `agentctl doctor network-policy` to inspect passive CNI evidence. To test +packet flow, run `agentctl doctor network-policy --active-probe --probe-image +`. The command creates and deletes a scratch namespace. +It verifies only the probe policy and test Service. It does not verify your +production policy's selectors or allow-list. + ## Audit Verification `audit-verify` currently supports JSONL. The ClickHouse source adapter is a stub. diff --git a/internal/netpolicy/netprobe/runner.go b/internal/netpolicy/netprobe/runner.go new file mode 100644 index 0000000..21062cd --- /dev/null +++ b/internal/netpolicy/netprobe/runner.go @@ -0,0 +1,106 @@ +package netprobe + +import ( + "context" + "flag" + "fmt" + "io" + "net" + "strconv" + "time" +) + +const ( + ExitSuccess = 0 + ExitDialFailed = 10 + ExitDNSFailed = 20 + ExitInvalidArgs = 30 + ExitServerFailed = 40 + defaultDialTimeout = 5 * time.Second + defaultServeTimeout = 2 * time.Minute +) + +// Run executes the minimal TCP server or client used by the active policy probe. +func Run(args []string, stdout, stderr io.Writer) int { + if len(args) == 0 { + _, _ = fmt.Fprintln(stderr, "usage: netprobe serve|dial") + return ExitInvalidArgs + } + switch args[0] { + case "serve": + return runServeCommand(args[1:], stderr) + case "dial": + return runDialCommand(args[1:], stdout, stderr) + default: + _, _ = fmt.Fprintf(stderr, "unknown netprobe command %q\n", args[0]) + return ExitInvalidArgs + } +} + +func runServeCommand(args []string, stderr io.Writer) int { + flags := flag.NewFlagSet("serve", flag.ContinueOnError) + flags.SetOutput(stderr) + address := flags.String("address", ":18443", "TCP listen address") + if err := flags.Parse(args); err != nil || flags.NArg() != 0 { + return ExitInvalidArgs + } + ctx, cancel := context.WithTimeout(context.Background(), defaultServeTimeout) + defer cancel() + if err := serve(ctx, *address); err != nil { + _, _ = fmt.Fprintln(stderr, err) + return ExitServerFailed + } + return ExitSuccess +} + +func serve(ctx context.Context, address string) error { + listener, err := net.Listen("tcp", address) + if err != nil { + return fmt.Errorf("listen: %w", err) + } + defer listener.Close() + + go func() { + <-ctx.Done() + _ = listener.Close() + }() + for { + connection, err := listener.Accept() + if err != nil { + if ctx.Err() != nil { + return nil + } + return fmt.Errorf("accept: %w", err) + } + _ = connection.Close() + } +} + +func runDialCommand(args []string, stdout, stderr io.Writer) int { + if len(args) != 2 { + _, _ = fmt.Fprintln(stderr, "usage: netprobe dial ") + return ExitInvalidArgs + } + if _, err := strconv.ParseUint(args[1], 10, 16); err != nil { + _, _ = fmt.Fprintf(stderr, "invalid TCP port %q\n", args[1]) + return ExitInvalidArgs + } + ctx, cancel := context.WithTimeout(context.Background(), defaultDialTimeout) + defer cancel() + return dial(ctx, args[0], args[1], stdout, stderr) +} + +func dial(ctx context.Context, host, port string, stdout, stderr io.Writer) int { + if _, err := net.DefaultResolver.LookupHost(ctx, host); err != nil { + _, _ = fmt.Fprintf(stderr, "resolve %s: %v\n", host, err) + return ExitDNSFailed + } + connection, err := (&net.Dialer{}).DialContext(ctx, "tcp", net.JoinHostPort(host, port)) + if err != nil { + _, _ = fmt.Fprintf(stderr, "dial %s:%s: %v\n", host, port, err) + return ExitDialFailed + } + defer connection.Close() + _, _ = fmt.Fprintln(stdout, "connected") + return ExitSuccess +} diff --git a/internal/netpolicy/netprobe/runner_test.go b/internal/netpolicy/netprobe/runner_test.go new file mode 100644 index 0000000..bbb7f08 --- /dev/null +++ b/internal/netpolicy/netprobe/runner_test.go @@ -0,0 +1,50 @@ +package netprobe + +import ( + "bytes" + "context" + "net" + "strconv" + "testing" + "time" +) + +func TestRunDial(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + go func() { + connection, err := listener.Accept() + if err == nil { + _ = connection.Close() + } + }() + + port := strconv.Itoa(listener.Addr().(*net.TCPAddr).Port) + var stdout, stderr bytes.Buffer + if exitCode := Run([]string{"dial", "127.0.0.1", port}, &stdout, &stderr); exitCode != ExitSuccess { + t.Fatalf("Run(dial) = %d, stderr=%q", exitCode, stderr.String()) + } + if stdout.String() != "connected\n" { + t.Fatalf("stdout = %q, want connected", stdout.String()) + } +} + +func TestDialReturnsDNSFailure(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if exitCode := dial(ctx, "does-not-exist.invalid", "18443", &bytes.Buffer{}, &bytes.Buffer{}); exitCode != ExitDNSFailed { + t.Fatalf("dial() = %d, want %d", exitCode, ExitDNSFailed) + } +} + +func TestRunRejectsInvalidArguments(t *testing.T) { + if exitCode := Run(nil, &bytes.Buffer{}, &bytes.Buffer{}); exitCode != ExitInvalidArgs { + t.Fatalf("Run() = %d, want %d", exitCode, ExitInvalidArgs) + } + if exitCode := Run([]string{"dial", "localhost"}, &bytes.Buffer{}, &bytes.Buffer{}); exitCode != ExitInvalidArgs { + t.Fatalf("Run(dial) = %d, want %d", exitCode, ExitInvalidArgs) + } +} diff --git a/internal/netpolicy/probe.go b/internal/netpolicy/probe.go new file mode 100644 index 0000000..036d0f1 --- /dev/null +++ b/internal/netpolicy/probe.go @@ -0,0 +1,306 @@ +package netpolicy + +import ( + "context" + "fmt" + "strconv" + "time" + + corev1 "k8s.io/api/core/v1" + networkingv1 "k8s.io/api/networking/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/client-go/kubernetes" + typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" + + "github.com/Clawdlinux/agentic-operator-core/internal/netpolicy/netprobe" +) + +const ( + probeLabelKey = "agentic.clawdlinux.org/probe" + probeLabelValue = "network-policy" + probeRoleLabelKey = "agentic.clawdlinux.org/probe-role" + probeServerRole = "server" + probeClientRole = "client" + probeServiceName = "probe-target" + probePort = 18443 + defaultProbeTimeout = 90 * time.Second + defaultCleanupTimeout = 30 * time.Second + maxProbeTimeout = 110 * time.Second +) + +// ProbeVerdict is the result of a single active NetworkPolicy verification. +type ProbeVerdict string + +const ( + ProbeVerdictEnforcing ProbeVerdict = "Enforcing" + ProbeVerdictNotEnforcing ProbeVerdict = "NotEnforcing" + ProbeVerdictInconclusive ProbeVerdict = "Inconclusive" +) + +// ActiveProbeOptions configures an intrusive, CLI-only NetworkPolicy probe. +type ActiveProbeOptions struct { + ProbeImage string + Timeout time.Duration + KeepNamespace bool + AllowKnownNonEnforcing bool + waitForPodExit podExitWaiter + waitForPodRunning podRunningWaiter + waitForServiceEndpoints serviceEndpointsWaiter +} + +// ActiveProbeResult describes the evidence and cleanup state of the probe. +type ActiveProbeResult struct { + Verdict ProbeVerdict + Reason string + Namespace string + Detection Detection + ControlExitCode int32 + TestExitCode int32 + Cleaned bool +} + +type podExitWaiter func(context.Context, typedcorev1.PodInterface, string) (int32, error) +type podRunningWaiter func(context.Context, typedcorev1.PodInterface, string) error +type serviceEndpointsWaiter func(context.Context, typedcorev1.EndpointsInterface, string) error + +// RunActiveProbe creates a scratch namespace and proves whether a deny-egress +// NetworkPolicy blocks a TCP connection after an unpoliced control succeeds. +func RunActiveProbe(ctx context.Context, kube kubernetes.Interface, discoveryClient DiscoveryClient, daemonSets DaemonSetClient, options ActiveProbeOptions) (result ActiveProbeResult, err error) { + if kube == nil { + return result, fmt.Errorf("active NetworkPolicy probe requires a Kubernetes client") + } + detection, err := DetectCNIEnforcement(ctx, discoveryClient, daemonSets) + if err != nil { + return result, fmt.Errorf("detect CNI enforcement: %w", err) + } + result.Detection = detection + if detection.Enforcement == EnforcementKnownNonEnforcing && !options.AllowKnownNonEnforcing { + result.Verdict = ProbeVerdictInconclusive + result.Reason = detection.Reason + "; refusing active probe without explicit override" + return result, nil + } + if options.ProbeImage == "" { + return result, fmt.Errorf("active NetworkPolicy probe requires --probe-image") + } + if options.Timeout <= 0 { + options.Timeout = defaultProbeTimeout + } + if options.Timeout > maxProbeTimeout { + return result, fmt.Errorf("active NetworkPolicy probe timeout must not exceed %s", maxProbeTimeout) + } + if options.waitForPodExit == nil { + options.waitForPodExit = waitForPodExit + } + if options.waitForPodRunning == nil { + options.waitForPodRunning = waitForPodRunning + } + if options.waitForServiceEndpoints == nil { + options.waitForServiceEndpoints = waitForServiceEndpoints + } + probeContext, cancel := context.WithTimeout(ctx, options.Timeout) + defer cancel() + + namespace, err := kube.CoreV1().Namespaces().Create(probeContext, buildProbeNamespace(), metav1.CreateOptions{}) + if err != nil { + return inconclusive(result, "create probe namespace", err) + } + result.Namespace = namespace.Name + if !options.KeepNamespace { + defer func() { + cleanupContext, cleanupCancel := context.WithTimeout(context.Background(), defaultCleanupTimeout) + defer cleanupCancel() + if cleanupErr := kube.CoreV1().Namespaces().Delete(cleanupContext, namespace.Name, metav1.DeleteOptions{}); cleanupErr != nil && !apierrors.IsNotFound(cleanupErr) { + if err == nil { + err = fmt.Errorf("cleanup probe namespace: %w", cleanupErr) + } + return + } + result.Cleaned = true + }() + } + + if _, err := kube.CoreV1().Pods(namespace.Name).Create(probeContext, buildProbeServerPod(options.ProbeImage), metav1.CreateOptions{}); err != nil { + return inconclusive(result, "create probe server", err) + } + if _, err := kube.CoreV1().Services(namespace.Name).Create(probeContext, buildProbeService(), metav1.CreateOptions{}); err != nil { + return inconclusive(result, "create probe service", err) + } + if err := options.waitForPodRunning(probeContext, kube.CoreV1().Pods(namespace.Name), "server"); err != nil { + return inconclusive(result, "wait for probe server", err) + } + if err := options.waitForServiceEndpoints(probeContext, kube.CoreV1().Endpoints(namespace.Name), probeServiceName); err != nil { + return inconclusive(result, "wait for probe service endpoints", err) + } + if _, err := kube.CoreV1().Pods(namespace.Name).Create(probeContext, buildProbeClientPod("control", options.ProbeImage), metav1.CreateOptions{}); err != nil { + return inconclusive(result, "create control client", err) + } + result.ControlExitCode, err = options.waitForPodExit(probeContext, kube.CoreV1().Pods(namespace.Name), "control") + if err != nil { + return inconclusive(result, "wait for control client", err) + } + if result.ControlExitCode != 0 { + result.Verdict = ProbeVerdictInconclusive + result.Reason = fmt.Sprintf("control connection failed with exit code %d", result.ControlExitCode) + return result, nil + } + if _, err := kube.NetworkingV1().NetworkPolicies(namespace.Name).Create(probeContext, buildProbeDenyEgressPolicy(), metav1.CreateOptions{}); err != nil { + return inconclusive(result, "create deny-egress policy", err) + } + if _, err := kube.CoreV1().Pods(namespace.Name).Create(probeContext, buildProbeClientPod("test", options.ProbeImage), metav1.CreateOptions{}); err != nil { + return inconclusive(result, "create test client", err) + } + result.TestExitCode, err = options.waitForPodExit(probeContext, kube.CoreV1().Pods(namespace.Name), "test") + if err != nil { + return inconclusive(result, "wait for test client", err) + } + if err := options.waitForPodRunning(probeContext, kube.CoreV1().Pods(namespace.Name), "server"); err != nil { + return inconclusive(result, "verify probe server", err) + } + return classifyProbeResult(result), nil +} + +func inconclusive(result ActiveProbeResult, action string, err error) (ActiveProbeResult, error) { + result.Verdict = ProbeVerdictInconclusive + result.Reason = action + ": " + err.Error() + return result, nil +} + +func classifyProbeResult(result ActiveProbeResult) ActiveProbeResult { + switch result.TestExitCode { + case netprobe.ExitSuccess: + result.Verdict = ProbeVerdictNotEnforcing + result.Reason = "test connection succeeded despite deny-egress policy" + case netprobe.ExitDialFailed: + result.Verdict = ProbeVerdictEnforcing + result.Reason = "test connection was blocked by deny-egress policy" + case netprobe.ExitDNSFailed: + result.Verdict = ProbeVerdictInconclusive + result.Reason = "test DNS resolution failed" + default: + result.Verdict = ProbeVerdictInconclusive + result.Reason = fmt.Sprintf("test client exited with code %d", result.TestExitCode) + } + return result +} + +func buildProbeNamespace() *corev1.Namespace { + return &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{GenerateName: "agentctl-netpolicy-probe-", Labels: map[string]string{probeLabelKey: probeLabelValue, "app.kubernetes.io/managed-by": "agentctl", "pod-security.kubernetes.io/enforce": "restricted"}}} +} + +func buildProbeServerPod(image string) *corev1.Pod { + return buildProbePod("server", probeServerRole, image, []string{"/manager", "netprobe", "serve"}) +} + +func buildProbeClientPod(name, image string) *corev1.Pod { + return buildProbePod(name, probeClientRole, image, []string{"/manager", "netprobe", "dial", probeServiceName, strconv.Itoa(probePort)}) +} + +func buildProbePod(name, role, image string, command []string) *corev1.Pod { + runAsUser := int64(65532) + readOnlyRootFilesystem := true + allowPrivilegeEscalation := false + activeDeadlineSeconds := int64(120) + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name, Labels: map[string]string{probeLabelKey: probeLabelValue, probeRoleLabelKey: role}}, + Spec: corev1.PodSpec{ + RestartPolicy: corev1.RestartPolicyNever, + ActiveDeadlineSeconds: &activeDeadlineSeconds, + AutomountServiceAccountToken: boolPointer(false), + SecurityContext: &corev1.PodSecurityContext{RunAsNonRoot: boolPointer(true), RunAsUser: &runAsUser, SeccompProfile: &corev1.SeccompProfile{Type: corev1.SeccompProfileTypeRuntimeDefault}}, + Containers: []corev1.Container{{ + Name: "netprobe", + Image: image, + Command: command, + SecurityContext: &corev1.SecurityContext{AllowPrivilegeEscalation: &allowPrivilegeEscalation, ReadOnlyRootFilesystem: &readOnlyRootFilesystem, Capabilities: &corev1.Capabilities{Drop: []corev1.Capability{"ALL"}}}, + Resources: corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("10m"), corev1.ResourceMemory: resource.MustParse("32Mi")}, Limits: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("100m"), corev1.ResourceMemory: resource.MustParse("64Mi")}}, + }}, + }, + } +} + +func buildProbeService() *corev1.Service { + return &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: probeServiceName}, Spec: corev1.ServiceSpec{Selector: map[string]string{probeLabelKey: probeLabelValue, probeRoleLabelKey: probeServerRole}, Ports: []corev1.ServicePort{{Port: probePort, TargetPort: intstr.FromInt(probePort)}}}} +} + +func buildProbeDenyEgressPolicy() *networkingv1.NetworkPolicy { + return &networkingv1.NetworkPolicy{ObjectMeta: metav1.ObjectMeta{Name: "agentctl-probe-deny-egress"}, Spec: networkingv1.NetworkPolicySpec{PodSelector: metav1.LabelSelector{MatchLabels: map[string]string{probeLabelKey: probeLabelValue, probeRoleLabelKey: probeClientRole}}, PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeEgress}, Egress: []networkingv1.NetworkPolicyEgressRule{{To: []networkingv1.NetworkPolicyPeer{{NamespaceSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"kubernetes.io/metadata.name": kubeSystemNamespace}}, PodSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"k8s-app": "kube-dns"}}}}, Ports: []networkingv1.NetworkPolicyPort{{Port: intstrPtr(intstr.FromInt(53)), Protocol: protocolPointer(corev1.ProtocolUDP)}, {Port: intstrPtr(intstr.FromInt(53)), Protocol: protocolPointer(corev1.ProtocolTCP)}}}}}} +} + +func boolPointer(value bool) *bool { return &value } + +func intstrPtr(value intstr.IntOrString) *intstr.IntOrString { return &value } + +func protocolPointer(value corev1.Protocol) *corev1.Protocol { return &value } + +func waitForPodExit(ctx context.Context, pods typedcorev1.PodInterface, name string) (int32, error) { + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + for { + pod, err := pods.Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return 0, err + } + for _, status := range pod.Status.ContainerStatuses { + if status.State.Terminated != nil { + return status.State.Terminated.ExitCode, nil + } + } + if pod.Status.Phase == corev1.PodSucceeded { + return 0, nil + } + if pod.Status.Phase == corev1.PodFailed { + return 1, nil + } + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-ticker.C: + } + } +} + +func waitForPodRunning(ctx context.Context, pods typedcorev1.PodInterface, name string) error { + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + for { + pod, err := pods.Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return err + } + if pod.Status.Phase == corev1.PodRunning { + return nil + } + if pod.Status.Phase == corev1.PodFailed || pod.Status.Phase == corev1.PodSucceeded { + return fmt.Errorf("probe server reached terminal phase %s", pod.Status.Phase) + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} + +func waitForServiceEndpoints(ctx context.Context, endpoints typedcorev1.EndpointsInterface, name string) error { + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + for { + current, err := endpoints.Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return err + } + for _, subset := range current.Subsets { + if len(subset.Addresses) > 0 && len(subset.Ports) > 0 { + return nil + } + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} diff --git a/internal/netpolicy/probe_test.go b/internal/netpolicy/probe_test.go new file mode 100644 index 0000000..7529161 --- /dev/null +++ b/internal/netpolicy/probe_test.go @@ -0,0 +1,203 @@ +package netpolicy + +import ( + "context" + "errors" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + k8sruntime "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/fake" + typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1" + k8stesting "k8s.io/client-go/testing" + + "github.com/Clawdlinux/agentic-operator-core/internal/netpolicy/netprobe" +) + +func TestRunActiveProbeRefusesKnownNonEnforcingCNIWithoutCreatingResources(t *testing.T) { + kube := fake.NewClientset() + result, err := RunActiveProbe(context.Background(), kube, fakeDiscovery{}, fakeDaemonSets("kindnet"), ActiveProbeOptions{}) + if err != nil { + t.Fatalf("RunActiveProbe() error = %v", err) + } + if result.Verdict != ProbeVerdictInconclusive || result.Detection.Enforcement != EnforcementKnownNonEnforcing { + t.Fatalf("result = %#v, want known-non-enforcing refusal", result) + } + if actions := kube.Actions(); len(actions) != 0 { + t.Fatalf("Kubernetes actions = %#v, want no resource operations", actions) + } +} + +func TestRunActiveProbeRequiresImageBeforeCreatingResources(t *testing.T) { + kube := fake.NewClientset() + _, err := RunActiveProbe(context.Background(), kube, fakeDiscovery{}, daemonSetListClient{}, ActiveProbeOptions{}) + if err == nil { + t.Fatal("RunActiveProbe() error = nil, want missing image error") + } + if actions := kube.Actions(); len(actions) != 0 { + t.Fatalf("Kubernetes actions = %#v, want no resource operations", actions) + } +} + +func TestRunActiveProbeRejectsTimeoutLongerThanServerLifetime(t *testing.T) { + kube := fake.NewClientset() + _, err := RunActiveProbe(context.Background(), kube, fakeDiscovery{}, daemonSetListClient{}, ActiveProbeOptions{ProbeImage: "example/operator:latest", Timeout: maxProbeTimeout + time.Second}) + if err == nil { + t.Fatal("RunActiveProbe() error = nil, want timeout validation error") + } + if actions := kube.Actions(); len(actions) != 0 { + t.Fatalf("Kubernetes actions = %#v, want no resource operations", actions) + } +} + +func TestRunActiveProbeClassifiesBlockedTestAndCleansNamespace(t *testing.T) { + kube := fake.NewClientset() + kube.PrependReactor("create", "namespaces", func(action k8stesting.Action) (bool, k8sruntime.Object, error) { + namespace := action.(k8stesting.CreateAction).GetObject().(*corev1.Namespace) + namespace.Name = "probe-ns" + return false, nil, nil + }) + result, err := RunActiveProbe(context.Background(), kube, fakeDiscovery{}, daemonSetListClient{}, ActiveProbeOptions{ + ProbeImage: "example/operator:latest", + waitForPodRunning: func(context.Context, typedcorev1.PodInterface, string) error { + return nil + }, + waitForServiceEndpoints: func(context.Context, typedcorev1.EndpointsInterface, string) error { + return nil + }, + waitForPodExit: func(_ context.Context, _ typedcorev1.PodInterface, name string) (int32, error) { + if name == "control" { + return 0, nil + } + return 10, nil + }, + }) + if err != nil { + t.Fatalf("RunActiveProbe() error = %v", err) + } + if result.Verdict != ProbeVerdictEnforcing || !result.Cleaned || result.Namespace != "probe-ns" { + t.Fatalf("result = %#v, want cleaned enforcing probe-ns", result) + } + var createdPolicy, deletedNamespace bool + for _, action := range kube.Actions() { + if action.GetVerb() == "create" && action.GetResource().Resource == "networkpolicies" { + createdPolicy = true + } + if action.GetVerb() == "delete" && action.GetResource().Resource == "namespaces" { + deletedNamespace = true + } + } + if !createdPolicy || !deletedNamespace { + t.Fatalf("actions = %#v, want policy creation and namespace deletion", kube.Actions()) + } +} + +func TestRunActiveProbeTreatsFailedControlAsInconclusive(t *testing.T) { + kube := fake.NewClientset() + kube.PrependReactor("create", "namespaces", func(action k8stesting.Action) (bool, k8sruntime.Object, error) { + namespace := action.(k8stesting.CreateAction).GetObject().(*corev1.Namespace) + namespace.Name = "probe-ns" + return false, nil, nil + }) + result, err := RunActiveProbe(context.Background(), kube, fakeDiscovery{}, daemonSetListClient{}, ActiveProbeOptions{ + ProbeImage: "example/operator:latest", + waitForPodRunning: func(context.Context, typedcorev1.PodInterface, string) error { + return nil + }, + waitForServiceEndpoints: func(context.Context, typedcorev1.EndpointsInterface, string) error { + return nil + }, + waitForPodExit: func(context.Context, typedcorev1.PodInterface, string) (int32, error) { + return 10, nil + }, + }) + if err != nil { + t.Fatalf("RunActiveProbe() error = %v", err) + } + if result.Verdict != ProbeVerdictInconclusive || result.ControlExitCode != 10 { + t.Fatalf("result = %#v, want inconclusive failed control", result) + } + for _, action := range kube.Actions() { + if action.GetVerb() == "create" && action.GetResource().Resource == "networkpolicies" { + t.Fatalf("actions = %#v, control failure must not create a NetworkPolicy", kube.Actions()) + } + } +} + +func TestRunActiveProbeTreatsServerLossAfterControlAsInconclusive(t *testing.T) { + kube := fake.NewClientset() + kube.PrependReactor("create", "namespaces", func(action k8stesting.Action) (bool, k8sruntime.Object, error) { + namespace := action.(k8stesting.CreateAction).GetObject().(*corev1.Namespace) + namespace.Name = "probe-ns" + return false, nil, nil + }) + serverChecks := 0 + result, err := RunActiveProbe(context.Background(), kube, fakeDiscovery{}, daemonSetListClient{}, ActiveProbeOptions{ + ProbeImage: "example/operator:latest", + waitForPodRunning: func(context.Context, typedcorev1.PodInterface, string) error { + serverChecks++ + if serverChecks == 2 { + return errors.New("server stopped") + } + return nil + }, + waitForServiceEndpoints: func(context.Context, typedcorev1.EndpointsInterface, string) error { + return nil + }, + waitForPodExit: func(_ context.Context, _ typedcorev1.PodInterface, name string) (int32, error) { + if name == "control" { + return 0, nil + } + return 10, nil + }, + }) + if err != nil { + t.Fatalf("RunActiveProbe() error = %v", err) + } + if result.Verdict != ProbeVerdictInconclusive { + t.Fatalf("result = %#v, want inconclusive when server dies", result) + } +} + +func TestClassifyProbeResult(t *testing.T) { + tests := []struct { + name string + exitCode int32 + wantVerdict ProbeVerdict + }{ + {name: "blocked connection", exitCode: netprobe.ExitDialFailed, wantVerdict: ProbeVerdictEnforcing}, + {name: "connected", exitCode: netprobe.ExitSuccess, wantVerdict: ProbeVerdictNotEnforcing}, + {name: "DNS failure", exitCode: netprobe.ExitDNSFailed, wantVerdict: ProbeVerdictInconclusive}, + {name: "unexpected failure", exitCode: 1, wantVerdict: ProbeVerdictInconclusive}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + result := classifyProbeResult(ActiveProbeResult{TestExitCode: test.exitCode}) + if result.Verdict != test.wantVerdict { + t.Fatalf("verdict = %q, want %q", result.Verdict, test.wantVerdict) + } + }) + } +} + +func TestBuildProbeResourcesAreRestricted(t *testing.T) { + pod := buildProbeClientPod("test", "example/operator:latest") + if pod.Spec.AutomountServiceAccountToken == nil || *pod.Spec.AutomountServiceAccountToken { + t.Fatal("probe pod enables service account token mounting") + } + if pod.Spec.SecurityContext == nil || pod.Spec.SecurityContext.RunAsNonRoot == nil || !*pod.Spec.SecurityContext.RunAsNonRoot { + t.Fatal("probe pod does not require non-root execution") + } + container := pod.Spec.Containers[0] + if container.SecurityContext == nil || container.SecurityContext.Capabilities == nil || len(container.SecurityContext.Capabilities.Drop) != 1 || container.SecurityContext.Capabilities.Drop[0] != "ALL" { + t.Fatalf("probe container capabilities = %#v, want drop ALL", container.SecurityContext) + } + policy := buildProbeDenyEgressPolicy() + if len(policy.Spec.Egress) != 1 || len(policy.Spec.Egress[0].Ports) != 2 { + t.Fatalf("probe egress policy = %#v, want DNS-only policy", policy.Spec.Egress) + } + if policy.Spec.PodSelector.MatchLabels[probeRoleLabelKey] != probeClientRole { + t.Fatalf("policy selector = %#v, want client probe only", policy.Spec.PodSelector) + } +}