Support dynamically updating Softnet policy

.golangci.yml

@@ -72,6 +72,7 @@ linters:
 
     # It's OK to not initialize some struct fields
     - exhaustruct
+    - exhaustruct_v5
 
     # We'll control the variable name length ourselves
     - varnamelen

go.mod

@@ -44,8 +44,10 @@ require (
 	go.uber.org/zap v1.27.1
 	golang.org/x/crypto v0.49.0
 	golang.org/x/exp v0.0.0-20250218142911-aa4b98e5adaa
+	golang.org/x/exp/jsonrpc2 v0.0.0-20260718201538-764159d718ef
 	golang.org/x/net v0.52.0
 	golang.org/x/sync v0.20.0
+	golang.org/x/sys v0.46.0
 	golang.org/x/term v0.41.0
 	golang.org/x/text v0.35.0
 	google.golang.org/grpc v1.79.3
@@ -143,6 +145,7 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/arch v0.24.0 // indirect
-	golang.org/x/sys v0.42.0 // indirect
+	golang.org/x/exp/event v0.0.0-20260611194520-c48552f49976 // indirect
+	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
 	google.golang.org/genproto v0.0.0-20230410155749-daa745c078e1 // indirect
 )

go.sum

@@ -397,6 +397,10 @@ golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtC
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20250218142911-aa4b98e5adaa h1:t2QcU6V556bFjYgu4L6C+6VrCPyJZ+eyRsABUPs1mz4=
 golang.org/x/exp v0.0.0-20250218142911-aa4b98e5adaa/go.mod h1:BHOTPb3L19zxehTsLoJXVaTktb06DFgmdW6Wb9s8jqk=
+golang.org/x/exp/event v0.0.0-20260611194520-c48552f49976 h1:aMc3mP8qjwSEdljznT5CYeyvpoFP7e3VCNqZZwUf3F0=
+golang.org/x/exp/event v0.0.0-20260611194520-c48552f49976/go.mod h1:cV3mj8TmLcC/Zki//yCMYw/nSC1BpHamZNkMSU7lgPw=
+golang.org/x/exp/jsonrpc2 v0.0.0-20260718201538-764159d718ef h1:VmBLplHGs0Tp6iOiqQMfK/Ku2m5mAMeFJAFzywb6NwI=
+golang.org/x/exp/jsonrpc2 v0.0.0-20260718201538-764159d718ef/go.mod h1:it3Sp7J2OmdyHxl5q3GMU4d2QZsFQGglbij9kqQVVaI=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
@@ -450,8 +454,8 @@ golang.org/x/sys v0.0.0-20221010170243-090e33056c14/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
-golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw=
+golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -480,6 +484,8 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da h1:noIWHXmPHxILtqtCOPIhSt0ABwskkZKjD3bXGnZGpNY=
+golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90=
 gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
 gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=

internal/controller/api_vms.go

@@ -1,6 +1,7 @@
 package controller
 
 import (
+	"context"
 	"encoding/base64"
 	"encoding/json"
 	"errors"
@@ -16,8 +17,8 @@ import (
 	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
 	"github.com/cirruslabs/orchard/pkg/client"
 	"github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/cirruslabs/orchard/rpc"
 	"github.com/gin-gonic/gin"
-	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
 	"github.com/samber/lo"
 )
@@ -163,8 +164,12 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder
 	}
 
 	name := ctx.Param("name")
+	var affectedWorker string
+
+	response := controller.storeUpdate(func(txn storepkg.Transaction) responder.Responder {
+		// Re-initialize to guard against potential transaction retry
+		affectedWorker = ""
 
-	return controller.storeUpdate(func(txn storepkg.Transaction) responder.Responder {
 		dbVM, err := txn.GetVM(name)
 		if err != nil {
 			return responder.Error(err)
@@ -216,7 +221,7 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder
 				"transition: only suspendable VMs can be suspended"))
 		}
 
-		if cmp.Equal(dbVM.VMSpec, userVM.VMSpec) {
+		if dbVM.SemanticallyEqual(userVM.VMSpec) {
 			// Nothing was changed
 			return responder.JSON(http.StatusOK, dbVM)
 		}
@@ -231,8 +236,24 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder
 			return responder.Code(http.StatusInternalServerError)
 		}
 
+		affectedWorker = dbVM.Worker
+
 		return responder.JSON(http.StatusOK, dbVM)
 	})
+
+	if affectedWorker != "" {
+		notifyContext, cancel := context.WithTimeout(ctx.Request.Context(), time.Second)
+		defer cancel()
+
+		if err := controller.workerNotifier.Notify(notifyContext, affectedWorker, &rpc.WatchInstruction{
+			Action: &rpc.WatchInstruction_SyncVmsAction{},
+		}); err != nil {
+			controller.logger.Warnf("failed to reactively sync updated VM %s on worker %s: %v",
+				name, affectedWorker, err)
+		}
+	}
+
+	return response
 }
 
 func (controller *Controller) updateVMState(ctx *gin.Context) responder.Responder {

internal/tests/spec_update_test.go

@@ -3,13 +3,17 @@ package tests
 import (
 	"context"
 	"fmt"
+	"net"
+	"os"
 	"runtime"
 	"testing"
 	"time"
 
+	commandssh "github.com/cirruslabs/orchard/internal/command/ssh"
 	"github.com/cirruslabs/orchard/internal/imageconstant"
 	"github.com/cirruslabs/orchard/internal/tests/devcontroller"
 	"github.com/cirruslabs/orchard/internal/tests/wait"
+	"github.com/cirruslabs/orchard/internal/worker"
 	"github.com/cirruslabs/orchard/internal/worker/ondiskname"
 	"github.com/cirruslabs/orchard/internal/worker/vmmanager"
 	"github.com/cirruslabs/orchard/internal/worker/vmmanager/tart"
@@ -18,14 +22,25 @@ import (
 	"github.com/shirou/gopsutil/v4/process"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
+	"golang.org/x/crypto/ssh"
 )
 
 func TestSpecUpdateSoftnet(t *testing.T) {
 	if runtime.GOOS != "darwin" {
 		t.Skip("Softnet is only supported on macOS with Tart")
 	}
 
-	devClient, _, _ := devcontroller.StartIntegrationTestEnvironment(t)
+	if os.Getenv("ORCHARD_SKIP_SOFTNET_TESTS") != "" {
+		t.Skip("softnet tests require root")
+	}
+
+	devClient, _, _ := devcontroller.StartIntegrationTestEnvironmentWithAdditionalOpts(
+		t,
+		false,
+		nil,
+		false,
+		[]worker.Option{worker.WithSoftnetPolicyUpdates(false)},
+	)
 
 	// Create a VM
 	vmName := "test"
@@ -92,7 +107,17 @@ func TestSpecUpdateSoftnetSuspendable(t *testing.T) {
 		t.Skip("Softnet is only supported on macOS with Tart")
 	}
 
-	devClient, _, _ := devcontroller.StartIntegrationTestEnvironment(t)
+	if os.Getenv("ORCHARD_SKIP_SOFTNET_TESTS") != "" {
+		t.Skip("softnet tests require root")
+	}
+
+	devClient, _, _ := devcontroller.StartIntegrationTestEnvironmentWithAdditionalOpts(
+		t,
+		false,
+		nil,
+		false,
+		[]worker.Option{worker.WithSoftnetPolicyUpdates(false)},
+	)
 
 	// Create a suspendable VM with Softnet enabled
 	vmName := "test"
@@ -159,6 +184,130 @@ func TestSpecUpdateSoftnetSuspendable(t *testing.T) {
 	require.True(t, sliceContainsAnotherSlice(tartRunCmdline, []string{"--net-softnet-block", "0.0.0.0/0"}))
 }
 
+//nolint:gosec,modernize,perfsprint,staticcheck // preserve the original integration test
+func TestSpecUpdateSoftnetPolicy(t *testing.T) {
+	if runtime.GOOS != "darwin" {
+		t.Skip("Softnet is only supported on macOS with Tart")
+	}
+
+	if os.Getenv("ORCHARD_SKIP_SOFTNET_TESTS") != "" {
+		t.Skip("softnet tests require root")
+	}
+
+	devClient, _, _ := devcontroller.StartIntegrationTestEnvironmentWithAdditionalOpts(
+		t,
+		false,
+		nil,
+		false,
+		[]worker.Option{worker.WithSoftnetPolicyUpdates(true)},
+	)
+
+	// Create a VM with Softnet enabled
+	vmName := "test"
+
+	err := devClient.VMs().Create(t.Context(), &v1.VM{
+		Meta: v1.Meta{
+			Name: vmName,
+		},
+		Image:    imageconstant.DefaultMacosImage,
+		CPU:      4,
+		Memory:   8 * 1024,
+		Headless: true,
+		VMSpec: v1.VMSpec{
+			NetSoftnet: true,
+		},
+	})
+	require.NoError(t, err)
+
+	// Wait for the VM to start
+	var vm *v1.VM
+
+	require.True(t, wait.Wait(2*time.Minute, func() bool {
+		vm, err = devClient.VMs().Get(context.Background(), vmName)
+		require.NoError(t, err)
+
+		t.Logf("Waiting for the VM to start. Current status: %s", vm.Status)
+
+		return vm.Status == v1.VMStatusRunning
+	}), "failed to start a VM")
+
+	// Ensure that the VM is using "--net-softnet"
+	tartVMName := ondiskname.New(vmName, vm.UID, vm.RestartCount).String()
+
+	tartRunCmdline, err := tartRunProcessCmdline(tartVMName)
+	require.NoError(t, err)
+	require.Contains(t, tartRunCmdline, "--net-softnet")
+
+	// Connect to the VM over SSH
+	var netConn net.Conn
+
+	require.True(t, wait.Wait(2*time.Minute, func() bool {
+		netConn, err = devClient.VMs().PortForward(t.Context(), vmName, 22, 120)
+		if err != nil {
+			t.Logf("Waiting for SSH to become available: %v", err)
+		}
+
+		return err == nil
+	}), "failed to connect to the VM over SSH")
+	defer netConn.Close()
+
+	username, password := commandssh.ChooseUsernameAndPassword(t.Context(), devClient, vmName, "", "")
+
+	sshConn, chans, reqs, err := ssh.NewClientConn(netConn, "", &ssh.ClientConfig{
+		User:            username,
+		Auth:            []ssh.AuthMethod{ssh.Password(password)},
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+	})
+	require.NoError(t, err)
+
+	sshClient := ssh.NewClient(sshConn, chans, reqs)
+	defer sshClient.Close()
+
+	curl := func(address string) error {
+		session, err := sshClient.NewSession()
+		require.NoError(t, err)
+		defer session.Close()
+
+		return session.Run(fmt.Sprintf(
+			"/usr/bin/curl -4 -k -sS -o /dev/null --connect-timeout 5 --max-time 10 https://%s",
+			address,
+		))
+	}
+
+	// Ensure that the address is reachable before blocking it
+	require.NoError(t, curl("1.1.1.1"))
+
+	// Update the Softnet policy
+	restartCountBeforePolicyUpdate := vm.RestartCount
+	vm.NetSoftnetBlock = []string{"1.1.1.1/32"}
+
+	vm, err = devClient.VMs().Update(t.Context(), *vm)
+	require.NoError(t, err)
+	require.EqualValues(t, 1, vm.Generation)
+	require.EqualValues(t, 0, vm.ObservedGeneration)
+
+	require.True(t, wait.Wait(2*time.Minute, func() bool {
+		vm, err = devClient.VMs().Get(context.Background(), vmName)
+		require.NoError(t, err)
+
+		t.Logf("Waiting for the VM's observed generation to be updated...")
+
+		return vm.ObservedGeneration == 1
+	}), "failed to wait for the VM's observed generation to be updated")
+
+	// Ensure that the policy was updated without restarting the VM
+	require.Equal(t, restartCountBeforePolicyUpdate, vm.RestartCount)
+	require.Equal(t, tartVMName, vm.TartName)
+
+	updatedTartRunCmdline, err := tartRunProcessCmdline(tartVMName)
+	require.NoError(t, err)
+	require.Equal(t, tartRunCmdline, updatedTartRunCmdline)
+
+	// Ensure that the new policy is applied without disrupting other traffic
+	require.Error(t, curl("1.1.1.1"))
+	require.NoError(t, curl("1.0.0.1"))
+}
+
 func TestSpecUpdatePowerStateSuspend(t *testing.T) {
 	if runtime.GOOS != "darwin" {
 		t.Skip("VM suspension is only supported on macOS with Tart")

internal/worker/option.go

@@ -4,6 +4,7 @@ import (
 	"github.com/cirruslabs/orchard/internal/dialer"
 	"github.com/cirruslabs/orchard/internal/worker/runtime"
 	v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
+	"github.com/samber/mo"
 	"go.uber.org/zap"
 )
 
@@ -52,6 +53,12 @@ func WithSynthetic() Option {
 	}
 }
 
+func WithSoftnetPolicyUpdates(enabled bool) Option {
+	return func(worker *Worker) {
+		worker.softnetPolicyUpdates = mo.Some(enabled)
+	}
+}
+
 func WithLogger(logger *zap.Logger) Option {
 	return func(worker *Worker) {
 		worker.logger = logger.Sugar()

internal/worker/runtime/runtime.go

@@ -19,6 +19,7 @@ type Runtime interface {
 		eventStreamer *client.EventStreamer,
 		vmPullTimeHistogram metric.Float64Histogram,
 		dialer dialer.Dialer,
+		softnetPolicyUpdates bool,
 		logger *zap.SugaredLogger,
 	) vmmanager.VM
 	ListVMs(ctx context.Context, logger *zap.SugaredLogger) ([]vmmanager.VMInfo, error)

internal/worker/runtime/synthetic.go

@@ -37,6 +37,7 @@ func (synthetic *Synthetic) NewVM(
 	eventStreamer *client.EventStreamer,
 	vmPullTimeHistogram metric.Float64Histogram,
 	_ dialer.Dialer,
+	_ bool,
 	logger *zap.SugaredLogger,
 ) vmmanager.VM {
 	return syntheticpkg.NewVM(vmResource, eventStreamer, vmPullTimeHistogram, logger)

internal/worker/runtime/tart.go

@@ -31,9 +31,10 @@ func (tart *Tart) NewVM(
 	eventStreamer *client.EventStreamer,
 	vmPullTimeHistogram metric.Float64Histogram,
 	dialer dialer.Dialer,
+	softnetPolicyUpdates bool,
 	logger *zap.SugaredLogger,
 ) vmmanager.VM {
-	return tartpkg.NewVM(vmResource, eventStreamer, vmPullTimeHistogram, dialer, logger)
+	return tartpkg.NewVM(vmResource, eventStreamer, vmPullTimeHistogram, dialer, softnetPolicyUpdates, logger)
 }
 
 func (tart *Tart) ListVMs(ctx context.Context, logger *zap.SugaredLogger) ([]vmmanager.VMInfo, error) {

internal/worker/runtime/vetu.go

@@ -31,6 +31,7 @@ func (vetu *Vetu) NewVM(
 	eventStreamer *client.EventStreamer,
 	vmPullTimeHistogram metric.Float64Histogram,
 	dialer dialer.Dialer,
+	_ bool,
 	logger *zap.SugaredLogger,
 ) vmmanager.VM {
 	return vetupkg.NewVM(vmResource, eventStreamer, vmPullTimeHistogram, dialer, logger)