diff --git a/packages/orchestrator/pkg/factories/run.go b/packages/orchestrator/pkg/factories/run.go index 8bb9723228..bd062114de 100644 --- a/packages/orchestrator/pkg/factories/run.go +++ b/packages/orchestrator/pkg/factories/run.go @@ -48,6 +48,7 @@ import ( "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/cgroup" "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/nbd" "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" + networkv2 "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network/v2" "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/template" "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/template/peerclient" "github.com/e2b-dev/infra/packages/orchestrator/pkg/server" @@ -718,12 +719,50 @@ func run(config cfg.Config, opts Options) (success bool) { if err != nil { logger.L().Fatal(ctx, "failed to create network pool", zap.Error(err)) } - networkPool := network.NewPool(network.NewSlotsPoolSize, network.ReusedSlotsPoolSize, slotStorage, config.NetworkConfig) - startService("network pool", func() error { - networkPool.Populate(ctx) + var networkPool network.PoolInterface + if config.NetworkConfig.NetworkVersion == 2 { + logger.L().Info(ctx, "using v2 network pool (nftables)") - return nil - }) + if err := networkv2.ValidateV2Prerequisites(); err != nil { + logger.L().Fatal(ctx, "v2 network prerequisites not met", zap.Error(err)) + } + + hostFw, hfErr := networkv2.NewHostFirewall(network.DefaultGateway(), config.NetworkConfig) + if hfErr != nil { + logger.L().Fatal(ctx, "failed to create v2 host firewall", zap.Error(hfErr)) + } + + // Startup reclaim tore down every slot that survived the last run, so + // the surviving set is empty and this drops the elements they left + // behind. Skipped when reclaim is disabled: the live slots are then + // unknown, and flushing would cut them off. + if !config.DisableStartupReclaim { + if err := hostFw.ReconcileSlots(ctx, nil); err != nil { + logger.L().Fatal(ctx, "failed to reconcile v2 host firewall slots", zap.Error(err)) + } + } + + observer, obsErr := networkv2.NewVethObserver() + if obsErr != nil { + logger.L().Fatal(ctx, "failed to create v2 veth observer", zap.Error(obsErr)) + } + + v2Pool := networkv2.NewV2Pool(slotStorage, config.NetworkConfig, hostFw, observer) + startService("network pool", func() error { + v2Pool.Populate(ctx) + + return nil + }) + networkPool = v2Pool + } else { + v1Pool := network.NewPool(network.NewSlotsPoolSize, network.ReusedSlotsPoolSize, slotStorage, config.NetworkConfig) + startService("network pool", func() error { + v1Pool.Populate(ctx) + + return nil + }) + networkPool = v1Pool + } closers = append(closers, closer{"network pool", networkPool.Close}) // sandbox factory diff --git a/packages/orchestrator/pkg/sandbox/network/host.go b/packages/orchestrator/pkg/sandbox/network/host.go index 0083b25238..e85d2e6e05 100644 --- a/packages/orchestrator/pkg/sandbox/network/host.go +++ b/packages/orchestrator/pkg/sandbox/network/host.go @@ -20,6 +20,9 @@ const loopbackInterface = "lo" // Host default gateway name var defaultGateway = utils.Must(getDefaultGateway(context.Background())) +// DefaultGateway returns the detected host default gateway interface name. +func DefaultGateway() string { return defaultGateway } + // func getDefaultGateway() (string, error) { // route, err := exec.Command( // "sh", diff --git a/packages/orchestrator/pkg/sandbox/network/network_version_test.go b/packages/orchestrator/pkg/sandbox/network/network_version_test.go new file mode 100644 index 0000000000..553c44064d --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/network_version_test.go @@ -0,0 +1,41 @@ +//go:build linux + +package network + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// An absent NETWORK_VERSION must select v1: the v2 datapath is opt-in. +func TestParseConfig_NetworkVersion(t *testing.T) { //nolint:paralleltest // t.Setenv + t.Run("absent selects v1", func(t *testing.T) { //nolint:paralleltest // t.Setenv + config, err := ParseConfig() + require.NoError(t, err) + require.Equal(t, 1, config.NetworkVersion) + }) + + t.Run("set but empty selects v1", func(t *testing.T) { //nolint:paralleltest // t.Setenv + t.Setenv("NETWORK_VERSION", "") + + config, err := ParseConfig() + require.NoError(t, err) + require.Equal(t, 1, config.NetworkVersion) + }) + + t.Run("explicit 2 selects v2", func(t *testing.T) { //nolint:paralleltest // t.Setenv + t.Setenv("NETWORK_VERSION", "2") + + config, err := ParseConfig() + require.NoError(t, err) + require.Equal(t, 2, config.NetworkVersion) + }) + + t.Run("non-numeric fails loudly", func(t *testing.T) { //nolint:paralleltest // t.Setenv + t.Setenv("NETWORK_VERSION", "two") + + _, err := ParseConfig() + require.Error(t, err) + }) +} diff --git a/packages/orchestrator/pkg/sandbox/network/pool.go b/packages/orchestrator/pkg/sandbox/network/pool.go index 6283d6f049..2e472e2bf9 100644 --- a/packages/orchestrator/pkg/sandbox/network/pool.go +++ b/packages/orchestrator/pkg/sandbox/network/pool.go @@ -87,6 +87,9 @@ type Config struct { // an explicit 0 disables marking for builds only. Range 0..63. // Set-but-empty behaves as unset (inherits): the env parser skips it. BuildSandboxEgressDSCP *uint8 `env:"BUILD_SANDBOX_EGRESS_DSCP"` + + // NetworkVersion selects v1 (iptables per-sandbox) or v2 (nftables, host sets). + NetworkVersion int `env:"NETWORK_VERSION" envDefault:"1"` } // EgressClass selects which configured egress DSCP applies to a sandbox. It diff --git a/packages/orchestrator/pkg/sandbox/network/pool_iface.go b/packages/orchestrator/pkg/sandbox/network/pool_iface.go new file mode 100644 index 0000000000..bb416082e6 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/pool_iface.go @@ -0,0 +1,18 @@ +//go:build linux + +package network + +import ( + "context" + "time" + + "github.com/e2b-dev/infra/packages/shared/pkg/grpc/orchestrator" +) + +// PoolInterface abstracts the network slot pool so v1 (*Pool) and v2 pools can +// be swapped. The existing *Pool already satisfies this interface. +type PoolInterface interface { + Get(ctx context.Context, network *orchestrator.SandboxNetworkConfig, class EgressClass) (*Slot, error) + ReturnAsync(ctx context.Context, slot *Slot, releasedFn ReleaseNotify, returnDelay time.Duration) error + Close(ctx context.Context) error +} diff --git a/packages/orchestrator/pkg/sandbox/network/slot.go b/packages/orchestrator/pkg/sandbox/network/slot.go index 87d9ff69ec..60345231c1 100644 --- a/packages/orchestrator/pkg/sandbox/network/slot.go +++ b/packages/orchestrator/pkg/sandbox/network/slot.go @@ -8,6 +8,7 @@ import ( "log" "net" "path/filepath" + "slices" "strconv" "sync/atomic" @@ -496,6 +497,11 @@ func getHostNetworkCIDR() *net.IPNet { return subnet } +// TapHostHardwareAddr returns the fixed host-side tap MAC. +func TapHostHardwareAddr() net.HardwareAddr { + return slices.Clone(tapHostHardwareAddr) +} + // getTapHostHardwareAddr parses the fixed tapHostMAC constant once at package // init, so CreateNetwork doesn't reparse a value that can never change or // fail at runtime. diff --git a/packages/orchestrator/pkg/sandbox/network/storage_local.go b/packages/orchestrator/pkg/sandbox/network/storage_local.go index f8a4e3bc47..7c53c38152 100644 --- a/packages/orchestrator/pkg/sandbox/network/storage_local.go +++ b/packages/orchestrator/pkg/sandbox/network/storage_local.go @@ -175,6 +175,11 @@ func (s *StorageLocal) Release(ips *Slot) error { return nil } +// IsNamespaceAvailable reports whether the named network namespace is free. +func IsNamespaceAvailable(dir, name string) (bool, error) { + return isNamespaceAvailable(dir, name) +} + func isNamespaceAvailable(dir, name string) (bool, error) { nsPath := filepath.Join(dir, name) _, err := os.Stat(nsPath) diff --git a/packages/orchestrator/pkg/sandbox/network/v2/absent.go b/packages/orchestrator/pkg/sandbox/network/v2/absent.go new file mode 100644 index 0000000000..b82f2d212d --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/absent.go @@ -0,0 +1,36 @@ +//go:build linux + +package v2 + +import ( + "errors" + "fmt" + "os" + + "github.com/vishvananda/netlink" + "golang.org/x/sys/unix" +) + +// appendUnlessAbsent records err unless it only reports state that is already +// gone, which makes teardown idempotent. +func appendUnlessAbsent(errs *[]error, err error, isAbsent func(error) bool, format string) { + if err == nil || isAbsent(err) { + return + } + + *errs = append(*errs, fmt.Errorf(format, err)) +} + +func isRouteNotExist(err error) bool { + return errors.Is(err, unix.ESRCH) || errors.Is(err, unix.ENOENT) +} + +func isLinkNotExist(err error) bool { + var linkNotFound netlink.LinkNotFoundError + + return errors.As(err, &linkNotFound) || errors.Is(err, unix.ENODEV) || errors.Is(err, unix.ENOENT) +} + +func isNamespaceNotExist(err error) bool { + return os.IsNotExist(err) || errors.Is(err, unix.ENOENT) +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/dscp.go b/packages/orchestrator/pkg/sandbox/network/v2/dscp.go new file mode 100644 index 0000000000..f93bc703be --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/dscp.go @@ -0,0 +1,224 @@ +//go:build linux + +package v2 + +import ( + "bytes" + "fmt" + "path/filepath" + + "github.com/containernetworking/plugins/pkg/ns" + "github.com/google/nftables" + "github.com/google/nftables/expr" + "golang.org/x/sys/unix" + + "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" +) + +const ( + maxDSCP = 63 // DSCP is the top 6 bits of the IPv4 TOS byte. + + // nsMangleChainName is the postrouting mangle chain in the in-namespace + // "slot-firewall" table, holding the egress DSCP rule — the same hook + // v1's iptables mangle/POSTROUTING rule uses, so the TOS byte on the + // veth link is identical between versions. + nsMangleChainName = "postroute_mangle" +) + +// SetupEgressDSCP creates the in-namespace postrouting mangle chain and, for +// a non-zero value, the rule stamping DSCP on everything leaving the sandbox +// netns through the vpeer uplink. Must run inside the namespace, on the conn +// SetupNamespaceNAT uses. +func SetupEgressDSCP(conn *nftables.Conn, table *nftables.Table, vpeerIface string, dscp uint8) error { + if dscp > maxDSCP { + return fmt.Errorf("egress DSCP %d out of range (0..%d)", dscp, maxDSCP) + } + + chain := conn.AddChain(&nftables.Chain{ + Name: nsMangleChainName, + Table: table, + Type: nftables.ChainTypeFilter, + Hooknum: nftables.ChainHookPostrouting, + Priority: nftables.ChainPriorityMangle, + }) + + if dscp > 0 { + conn.AddRule(&nftables.Rule{ + Table: table, + Chain: chain, + Exprs: dscpSetExprs(vpeerIface, dscp), + }) + } + + if err := conn.Flush(); err != nil { + return fmt.Errorf("flush egress DSCP setup: %w", err) + } + + return nil +} + +// ApplyEgressDSCP re-stamps the slot's egress rule so a build can be marked +// differently from a regular sandbox, mirroring v1's semantics: 0 removes the +// rule, a matching installed value is a no-op, and the add and delete land in +// one atomic batch so egress is never unmarked during a restamp and a failed +// apply keeps the previous class installed. The kernel ruleset is the source +// of truth for the currently stamped value. +func ApplyEgressDSCP(slot *network.Slot, dscp uint8) error { + if dscp > maxDSCP { + return fmt.Errorf("egress DSCP %d out of range (0..%d)", dscp, maxDSCP) + } + + nsHandle, err := ns.GetNS(filepath.Join(network.NetNamespacesDir, slot.NamespaceID())) + if err != nil { + return fmt.Errorf("get slot network namespace %q: %w", slot.NamespaceID(), err) + } + defer nsHandle.Close() + + return nsHandle.Do(func(_ ns.NetNS) error { + conn, err := nftables.New(nftables.AsLasting()) + if err != nil { + return fmt.Errorf("nftables conn in namespace: %w", err) + } + defer conn.CloseLasting() + + table := &nftables.Table{Name: "slot-firewall", Family: nftables.TableFamilyINet} + chain := &nftables.Chain{Name: nsMangleChainName, Table: table} + + current, currentDSCP, err := findDSCPRule(conn, table, chain, slot.VpeerName()) + if err != nil { + return fmt.Errorf("find DSCP rule for %s: %w", slot.NamespaceID(), err) + } + if current == nil && dscp == 0 { + return nil + } + if current != nil && currentDSCP == dscp { + return nil + } + + if dscp > 0 { + conn.AddRule(&nftables.Rule{ + Table: table, + Chain: chain, + Exprs: dscpSetExprs(slot.VpeerName(), dscp), + }) + } + if current != nil { + conn.DelRule(&nftables.Rule{ + Table: table, + Chain: chain, + Handle: current.Handle, + }) + } + + if err := conn.Flush(); err != nil { + return fmt.Errorf("flush DSCP %d rule for %s: %w", dscp, slot.NamespaceID(), err) + } + + return nil + }) +} + +// findDSCPRule returns the vpeer's DSCP rule and its stamped value, or nil +// when none is installed. +func findDSCPRule(conn *nftables.Conn, table *nftables.Table, chain *nftables.Chain, vpeerIface string) (*nftables.Rule, uint8, error) { + rules, err := conn.GetRules(table, chain) + if err != nil { + return nil, 0, fmt.Errorf("get mangle rules: %w", err) + } + + for _, r := range rules { + if dscp, ok := dscpRuleValue(r, vpeerIface); ok { + return r, dscp, nil + } + } + + return nil, 0, nil +} + +// dscpRuleValue reports whether r is the DSCP rule for the given uplink — an +// oifname match plus a payload write — and decodes the DSCP it stamps. +func dscpRuleValue(r *nftables.Rule, vpeerIface string) (uint8, bool) { + var tos uint8 + writes := false + + for _, e := range r.Exprs { + switch v := e.(type) { + case *expr.Payload: + if v.OperationType == expr.PayloadWrite { + writes = true + } + case *expr.Bitwise: + // The xor carries the new TOS byte in the second header byte. + if len(v.Xor) == 2 { + tos = v.Xor[1] + } + } + } + + if !writes || !hasOifnameCmp(r, vpeerIface) { + return 0, false + } + + return tos >> 2, true +} + +// hasOifnameCmp reports whether the rule compares oifname against the given +// interface name. +func hasOifnameCmp(r *nftables.Rule, name string) bool { + expected := ifnameBytes(name) + for i, e := range r.Exprs { + if i == 0 { + continue + } + cmp, ok := e.(*expr.Cmp) + if !ok { + continue + } + if meta, ok := r.Exprs[i-1].(*expr.Meta); ok && meta.Key == expr.MetaKeyOIFNAME && bytes.Equal(cmp.Data, expected) { + return true + } + } + + return false +} + +// dscpSetExprs builds: meta nfproto ipv4 oifname ip dscp set — +// a write of the TOS byte that preserves the ECN bits and fixes the IPv4 +// header checksum, equivalent to v1's `-o -j DSCP --set-dscp `. +// +// The read-modify-write spans the two leading header bytes rather than the TOS +// byte alone: the kernel's inet checksum fixup works on 16-bit words, so a +// 1-byte write at odd offset 1 folds the delta into the wrong half and every +// stamped packet leaves with a corrupt header checksum. Masking version/IHL +// through keeps them unchanged. +func dscpSetExprs(vpeerIface string, dscp uint8) []expr.Any { + return []expr.Any{ + &expr.Meta{Key: expr.MetaKeyNFPROTO, Register: 1}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: []byte{unix.NFPROTO_IPV4}}, + &expr.Meta{Key: expr.MetaKeyOIFNAME, Register: 1}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: ifnameBytes(vpeerIface)}, + &expr.Payload{ + OperationType: expr.PayloadLoad, + DestRegister: 1, + Base: expr.PayloadBaseNetworkHeader, + Offset: 0, // version/IHL + TOS + Len: 2, + }, + &expr.Bitwise{ + SourceRegister: 1, + DestRegister: 1, + Len: 2, + Mask: []byte{0xff, 0x03}, // keep version/IHL and the ECN bits + Xor: []byte{0x00, dscp << 2}, + }, + &expr.Payload{ + OperationType: expr.PayloadWrite, + SourceRegister: 1, + Base: expr.PayloadBaseNetworkHeader, + Offset: 0, + Len: 2, + CsumType: expr.CsumTypeInet, + CsumOffset: 10, // IPv4 header checksum + }, + } +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/dscp_egress_test.go b/packages/orchestrator/pkg/sandbox/network/v2/dscp_egress_test.go new file mode 100644 index 0000000000..ef5d19d6b9 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/dscp_egress_test.go @@ -0,0 +1,342 @@ +//go:build linux + +package v2 + +import ( + "context" + "net" + "path/filepath" + "runtime" + "testing" + + "github.com/containernetworking/plugins/pkg/ns" + "github.com/google/nftables" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/vishvananda/netlink" + "github.com/vishvananda/netns" + "golang.org/x/sys/unix" + + "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" +) + +// dscpConfig builds the host-firewall test config with the DSCP fields parsed +// from the real env path, so the nil-inherits semantics of +// BUILD_SANDBOX_EGRESS_DSCP are exercised end to end. A set-but-empty build +// value behaves as unset (the env parser skips it), so "" drives the +// inherits-sandbox path. +func dscpConfig(t *testing.T, sandboxDSCP, buildDSCP string) network.Config { + t.Helper() + + t.Setenv("SANDBOX_EGRESS_DSCP", sandboxDSCP) + t.Setenv("BUILD_SANDBOX_EGRESS_DSCP", buildDSCP) + + parsed, err := network.ParseConfig() + require.NoError(t, err) + + config := testConfig() + config.SandboxEgressDSCP = parsed.SandboxEgressDSCP + config.BuildSandboxEgressDSCP = parsed.BuildSandboxEgressDSCP + + return config +} + +// makeDSCPTestNS creates the slot's named namespace with the slot-firewall +// table and the seeded DSCP chain, as CreateNetworkV2 does, without the rest +// of the datapath. +func makeDSCPTestNS(t *testing.T, slot *network.Slot, config network.Config) { + t.Helper() + + runtime.LockOSThread() + defer runtime.UnlockOSThread() + + host, err := netns.Get() + require.NoError(t, err) + defer func() { + require.NoError(t, netns.Set(host)) + _ = host.Close() + }() + + nsHandle, err := netns.NewNamed(slot.NamespaceID()) + require.NoError(t, err) + defer nsHandle.Close() + t.Cleanup(func() { _ = netns.DeleteNamed(slot.NamespaceID()) }) + + conn, err := nftables.New(nftables.AsLasting()) + require.NoError(t, err) + defer conn.CloseLasting() + + table := conn.AddTable(&nftables.Table{Name: "slot-firewall", Family: nftables.TableFamilyINet}) + require.NoError(t, SetupEgressDSCP(conn, table, slot.VpeerName(), config.EgressDSCP(network.EgressClassSandbox))) +} + +// requireDSCPByte asserts the slot's in-namespace mangle rule stamps wantTOS — +// the TOS byte v1's EgressTOS resolution produces. wantTOS 0 asserts no rule. +func requireDSCPByte(t *testing.T, slot *network.Slot, wantTOS int) { + t.Helper() + + nsHandle, err := ns.GetNS(filepath.Join(network.NetNamespacesDir, slot.NamespaceID())) + require.NoError(t, err) + defer nsHandle.Close() + + var got []uint8 + require.NoError(t, nsHandle.Do(func(_ ns.NetNS) error { + conn, err := nftables.New(nftables.AsLasting()) + if err != nil { + return err + } + defer conn.CloseLasting() + + table := &nftables.Table{Name: "slot-firewall", Family: nftables.TableFamilyINet} + chain := &nftables.Chain{Name: nsMangleChainName, Table: table} + rules, err := conn.GetRules(table, chain) + if err != nil { + return err + } + for _, r := range rules { + if dscp, ok := dscpRuleValue(r, slot.VpeerName()); ok { + got = append(got, dscp) + } + } + + return nil + })) + + if wantTOS == 0 { + require.Emptyf(t, got, "DSCP 0 must leave no mangle rule in %s", slot.NamespaceID()) + + return + } + + require.Lenf(t, got, 1, "want exactly one DSCP rule in %s, got %v", slot.NamespaceID(), got) + require.Equal(t, wantTOS, int(got[0])<<2) +} + +// The stamped byte must equal what v1's EgressDSCP/EgressTOS resolution +// produces for the same env, class by class: seed, build override, +// nil-inherits, and 0-disables all included. +func TestEgressDSCP_V1Parity(t *testing.T) { //nolint:paralleltest // mutates host netns state (named netns per case); t.Setenv + skipIfNotLinuxRoot(t) + + tests := []struct { + name string + sandboxEnv string + buildEnv string + }{ + {"unset build value falls back to the sandbox value", "8", ""}, + {"build value overrides for builds only", "8", "16"}, + {"explicit zero disables marking for builds only", "8", "0"}, + {"builds can be marked while sandboxes are not", "0", "16"}, + {"both disabled", "0", ""}, + {"max DSCP maps to the top of the TOS byte", "8", "63"}, + } + + for _, tt := range tests { //nolint:paralleltest // named netns + t.Setenv per case + t.Run(tt.name, func(t *testing.T) { + config := dscpConfig(t, tt.sandboxEnv, tt.buildEnv) + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + makeDSCPTestNS(t, slot, config) + + // Creation seeds the untenanted class, as v1's CreateNetwork does. + requireDSCPByte(t, slot, config.EgressTOS().For(network.EgressClassSandbox)) + + for _, class := range []network.EgressClass{network.EgressClassBuild, network.EgressClassSandbox} { + require.NoError(t, ApplyEgressDSCP(slot, config.EgressDSCP(class))) + requireDSCPByte(t, slot, config.EgressTOS().For(class)) + } + }) + } +} + +func TestEgressDSCP_RestampSemantics(t *testing.T) { //nolint:paralleltest // mutates host netns state (named netns); t.Setenv + skipIfNotLinuxRoot(t) + + config := dscpConfig(t, "8", "16") + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + makeDSCPTestNS(t, slot, config) + requireDSCPByte(t, slot, 8<<2) + + // Re-applying the same class must not duplicate the rule. + require.NoError(t, ApplyEgressDSCP(slot, 8)) + requireDSCPByte(t, slot, 8<<2) + + // Out of the 6-bit range is rejected before any chain mutation, so the + // previous class stays installed. + require.Error(t, ApplyEgressDSCP(slot, 200)) + requireDSCPByte(t, slot, 8<<2) + + // 0 removes the rule outright; a second 0 is a clean no-op. + require.NoError(t, ApplyEgressDSCP(slot, 0)) + requireDSCPByte(t, slot, 0) + require.NoError(t, ApplyEgressDSCP(slot, 0)) + requireDSCPByte(t, slot, 0) +} + +type fakeStorage struct{} + +func (f *fakeStorage) Acquire(context.Context) (*network.Slot, error) { + return nil, context.Canceled +} + +func (f *fakeStorage) Release(*network.Slot) error { return nil } + +// The pool seam that protects tenants, through the same path production +// takes: Get stamps the tenant's class, returnSlot restores the untenanted +// one before the slot becomes reusable. +func TestV2Pool_GetStampsAndRecycleRestoresDSCP(t *testing.T) { //nolint:paralleltest // mutates host netns state: singleton nftables table "v2-host-firewall", named netns, veth links; t.Setenv + skipIfNotLinuxRoot(t) + + ctx := context.Background() + config := dscpConfig(t, "8", "16") + + hf := newTestHostFirewall(t, config) + + observer, err := NewVethObserver() + require.NoError(t, err) + + pool := NewV2Pool(&fakeStorage{}, config, hf, observer) + t.Cleanup(func() { _ = pool.Close(context.Background()) }) + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + pool.registry.Store(sv2) + require.NoError(t, CreateNetworkV2(ctx, slot, sv2, hf, observer)) + t.Cleanup(func() { _ = RemoveNetworkV2(ctx, slot, sv2, hf, observer) }) + + // Queue and close before any assertion can fail the test: Close ranges + // over newSlots, so leaving it open would deadlock the cleanup. + pool.newSlots <- slot + close(pool.newSlots) // Populate never runs here. + + requireDSCPByte(t, slot, 8<<2) + + got, err := pool.Get(ctx, nil, network.EgressClassBuild) + require.NoError(t, err) + require.Equal(t, slot.Idx, got.Idx) + requireDSCPByte(t, got, 16<<2) + + require.NoError(t, pool.returnSlot(ctx, got, func(context.Context, string) {}, 0)) + requireDSCPByte(t, got, 8<<2) + + select { + case reused := <-pool.reusedSlots: + require.Equal(t, slot.Idx, reused.Idx) + default: + t.Fatal("recycled slot must go back to the reuse queue, not be torn down") + } +} + +// The wire-level claim: the TOS byte on the veth link equals v1's EgressTOS +// resolution, and 0 leaves the byte untouched. +func TestEgressDSCP_WireTOS(t *testing.T) { //nolint:paralleltest // mutates host netns state: singleton nftables table "v2-host-firewall", named netns, veth links; t.Setenv + skipIfNotLinuxRoot(t) + + ctx := context.Background() + config := dscpConfig(t, "8", "16") + + hf := newTestHostFirewall(t, config) + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + require.NoError(t, CreateNetworkV2(ctx, slot, sv2, hf, nil)) + t.Cleanup(func() { _ = RemoveNetworkV2(ctx, slot, sv2, hf, nil) }) + + assert.Equal(t, config.EgressTOS().For(network.EgressClassSandbox), observeEgressTOS(t, slot)) + + require.NoError(t, ApplyEgressDSCP(slot, config.EgressDSCP(network.EgressClassBuild))) + assert.Equal(t, config.EgressTOS().For(network.EgressClassBuild), observeEgressTOS(t, slot)) + + require.NoError(t, ApplyEgressDSCP(slot, 0)) + assert.Equal(t, 0, observeEgressTOS(t, slot)) +} + +const wireProbePort = 48765 + +// observeEgressTOS sends a UDP packet from inside the slot's namespace toward +// the host and returns the IPv4 TOS byte it carries on the veth link — the +// same wire bytes v1's in-namespace mangle rule produces. The AF_PACKET +// capture sees the frame as it leaves the namespace, before any host +// netfilter hook — so it also verifies the header checksum the stamp has to +// keep valid: an L2 capture accepts a packet the next hop would drop. +func observeEgressTOS(t *testing.T, slot *network.Slot) int { + t.Helper() + + link, err := netlink.LinkByName(slot.VethName()) + require.NoError(t, err) + + fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_DGRAM, int(htons(unix.ETH_P_IP))) + require.NoError(t, err) + defer unix.Close(fd) + + require.NoError(t, unix.Bind(fd, &unix.SockaddrLinklayer{ + Protocol: htons(unix.ETH_P_IP), + Ifindex: link.Attrs().Index, + })) + require.NoError(t, unix.SetsockoptTimeval(fd, unix.SOL_SOCKET, unix.SO_RCVTIMEO, &unix.Timeval{Sec: 10})) + + sendFromNamespace(t, slot, wireProbePort) + + buf := make([]byte, 256) + for { + n, _, err := unix.Recvfrom(fd, buf, 0) + require.NoError(t, err, "no probe packet captured on %s", slot.VethName()) + if n < 20 || buf[0]>>4 != 4 { + continue + } + ihl := int(buf[0]&0x0f) * 4 + if buf[9] != unix.IPPROTO_UDP || n < ihl+8 { + continue + } + if int(buf[ihl+2])<<8|int(buf[ihl+3]) != wireProbePort { + continue + } + + require.Zerof(t, ipHeaderChecksum(buf[:ihl]), "stamped packet carries a corrupt IPv4 header checksum: % x", buf[:ihl]) + + return int(buf[1]) + } +} + +func htons(v uint16) uint16 { return v<<8 | v>>8 } + +// ipHeaderChecksum folds the one's-complement sum over a whole IPv4 header, +// including its checksum field: a valid header sums to zero. +func ipHeaderChecksum(header []byte) uint16 { + var sum uint32 + for i := 0; i+1 < len(header); i += 2 { + sum += uint32(header[i])<<8 | uint32(header[i+1]) + } + for sum>>16 != 0 { + sum = sum&0xffff + sum>>16 + } + + return ^uint16(sum) +} + +func sendFromNamespace(t *testing.T, slot *network.Slot, port int) { + t.Helper() + + runtime.LockOSThread() + defer runtime.UnlockOSThread() + + host, err := netns.Get() + require.NoError(t, err) + defer func() { + require.NoError(t, netns.Set(host)) + _ = host.Close() + }() + + target, err := netns.GetFromName(slot.NamespaceID()) + require.NoError(t, err) + defer target.Close() + require.NoError(t, netns.Set(target)) + + conn, err := net.DialUDP("udp4", nil, &net.UDPAddr{IP: slot.VethIP(), Port: port}) + require.NoError(t, err) + defer conn.Close() + + _, err = conn.Write([]byte("tos-probe")) + require.NoError(t, err) +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/host_firewall.go b/packages/orchestrator/pkg/sandbox/network/v2/host_firewall.go new file mode 100644 index 0000000000..4c5ea77dc0 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/host_firewall.go @@ -0,0 +1,575 @@ +//go:build linux + +package v2 + +import ( + "bytes" + "context" + "encoding/binary" + "errors" + "fmt" + "net" + "slices" + "sync" + + "github.com/google/nftables" + "github.com/google/nftables/expr" + "go.uber.org/zap" + "golang.org/x/sys/unix" + + "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" + "github.com/e2b-dev/infra/packages/shared/pkg/logger" +) + +// HostFirewall is a singleton that manages the host-level nftables table +// "v2-host-firewall". It uses sets so rule count stays constant regardless +// of sandbox count — lookups are O(1) hash operations. +// +// Table layout: +// +// table inet v2-host-firewall { +// set v2_veths { type ifname; } +// set v2_host_cidrs { type ipv4_addr; flags interval; } +// +// chain forward { type filter hook forward priority 0; policy accept; +// iifname @v2_veths oifname accept +// iifname oifname @v2_veths accept +// } +// chain prerouting { type nat hook prerouting priority -100; +// # service redirects (all slots share same ports) +// iifname @v2_veths tcp dport 80 ip daddr redirect to :hyperloopPort +// iifname @v2_veths tcp dport 111 ip daddr redirect to :portmapperPort +// iifname @v2_veths tcp dport 2049 ip daddr redirect to :nfsPort +// # TCP firewall proxy redirects +// iifname @v2_veths tcp dport 80 redirect to :tcpHTTPPort +// iifname @v2_veths tcp dport 443 redirect to :tcpTLSPort +// iifname @v2_veths tcp redirect to :tcpOtherPort # every remaining TCP +// } +// chain postrouting { type nat hook postrouting priority 100; +// ip saddr @v2_host_cidrs oifname masquerade +// } +// } +// +// All slots share the same redirect ports, so only per-slot data (veth names +// and host CIDRs) goes into sets and the rule count stays constant. +type HostFirewall struct { + conn *nftables.Conn + table *nftables.Table + + vethSet *nftables.Set // type ifname; elements = veth interface names + cidrSet *nftables.Set // type ipv4_addr; flags interval; elements = host CIDRs + + defaultGw string + config network.Config + mu sync.Mutex +} + +const ( + hostFwTableName = "v2-host-firewall" +) + +// NewHostFirewall creates or opens the singleton host firewall table with all +// required sets and chains. Call once per orchestrator process. +// +// Restart-safe: if the table already exists from a previous run (e.g., orchestrator +// restart with live sandboxes), existing set elements are preserved. Only chain rules +// are refreshed with current config. This prevents connectivity loss for sandboxes +// that survived the restart. +func NewHostFirewall(defaultGw string, config network.Config) (_ *HostFirewall, err error) { + conn, err := nftables.New(nftables.AsLasting()) + if err != nil { + return nil, fmt.Errorf("new nftables conn: %w", err) + } + + defer func() { + if err != nil { + err = errors.Join(err, conn.CloseLasting()) + } + }() + + // Ensure table exists — idempotent: creates if new, opens if existing. + // We do NOT delete the table on startup. Existing set elements represent + // live sandboxes whose connectivity must be preserved across restarts. + table := conn.AddTable(&nftables.Table{ + Name: hostFwTableName, + Family: nftables.TableFamilyINet, + }) + + hf := &HostFirewall{ + conn: conn, + table: table, + defaultGw: defaultGw, + config: config, + } + + if err := hf.ensureSets(); err != nil { + return nil, fmt.Errorf("ensure sets: %w", err) + } + + if err := hf.ensureChains(); err != nil { + return nil, fmt.Errorf("ensure chains: %w", err) + } + + return hf, nil +} + +// ensureSets creates sets if they don't exist, or opens handles to existing ones. +// Existing set elements (active sandbox veth names and host CIDRs) are preserved. +func (hf *HostFirewall) ensureSets() error { + // Set of veth interface names — AddSet is idempotent for existing sets + hf.vethSet = &nftables.Set{ + Table: hf.table, + Name: "v2_veths", + KeyType: nftables.TypeIFName, + } + if err := hf.conn.AddSet(hf.vethSet, nil); err != nil { + return fmt.Errorf("add veth set: %w", err) + } + + // Set of host CIDRs — interval set for /32 entries + hf.cidrSet = &nftables.Set{ + Table: hf.table, + Name: "v2_host_cidrs", + KeyType: nftables.TypeIPAddr, + Interval: true, + } + if err := hf.conn.AddSet(hf.cidrSet, nil); err != nil { + return fmt.Errorf("add cidr set: %w", err) + } + + if err := hf.conn.Flush(); err != nil { + return fmt.Errorf("flush sets: %w", err) + } + + return nil +} + +// ensureChains creates or refreshes all chains with current config. +// Chains are flushed and re-populated — this is safe because chain rules are +// config-dependent (ports, gateway interface), not state-dependent. +// Set elements (which represent live sandboxes) are NOT touched. +// The entire operation is atomic via nftables batching. +func (hf *HostFirewall) ensureChains() error { + gwBytes := ifnameBytes(hf.defaultGw) + orchIP := net.ParseIP(hf.config.OrchestratorInSandboxIPAddress).To4() + + // --- FORWARD chain --- + // Accept policy, matching v1: the slot rules below only add ACCEPTs, so + // this chain stays transparent to every other forwarding path on the host. + fwdPolicy := nftables.ChainPolicyAccept + fwdChain := hf.conn.AddChain(&nftables.Chain{ + Name: "forward", + Table: hf.table, + Type: nftables.ChainTypeFilter, + Hooknum: nftables.ChainHookForward, + Priority: nftables.ChainPriorityFilter, + Policy: &fwdPolicy, + }) + hf.conn.FlushChain(fwdChain) // clear stale rules from previous config + + // iifname @v2_veths oifname accept + hf.conn.AddRule(&nftables.Rule{ + Table: hf.table, + Chain: fwdChain, + Exprs: []expr.Any{ + &expr.Meta{Key: expr.MetaKeyIIFNAME, Register: 1}, + &expr.Lookup{SourceRegister: 1, SetName: hf.vethSet.Name, SetID: hf.vethSet.ID}, + &expr.Meta{Key: expr.MetaKeyOIFNAME, Register: 1}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: gwBytes}, + &expr.Verdict{Kind: expr.VerdictAccept}, + }, + }) + + // iifname oifname @v2_veths accept — unconditional, as v1's FORWARD + // rule is; a conntrack-state match here would break new inbound flows v1 + // allows. + hf.conn.AddRule(&nftables.Rule{ + Table: hf.table, + Chain: fwdChain, + Exprs: []expr.Any{ + &expr.Meta{Key: expr.MetaKeyIIFNAME, Register: 1}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: gwBytes}, + &expr.Meta{Key: expr.MetaKeyOIFNAME, Register: 1}, + &expr.Lookup{SourceRegister: 1, SetName: hf.vethSet.Name, SetID: hf.vethSet.ID}, + &expr.Verdict{Kind: expr.VerdictAccept}, + }, + }) + + // --- PREROUTING chain (NAT) --- + preChain := hf.conn.AddChain(&nftables.Chain{ + Name: "prerouting", + Table: hf.table, + Type: nftables.ChainTypeNAT, + Hooknum: nftables.ChainHookPrerouting, + Priority: nftables.ChainPriorityRef(-100), + }) + hf.conn.FlushChain(preChain) + + // Service redirects: iifname @v2_veths + ip daddr + tcp dport X → redirect to :port + svcRedirects := []struct { + dport uint16 + rport uint16 + }{ + {80, hf.config.HyperloopProxyPort}, + {111, hf.config.PortmapperPort}, + {2049, hf.config.NFSProxyPort}, + } + for _, svc := range svcRedirects { + hf.conn.AddRule(&nftables.Rule{ + Table: hf.table, + Chain: preChain, + Exprs: svcRedirectExprs(hf.vethSet, orchIP, svc.dport, svc.rport), + }) + } + + // TCP firewall proxy redirects: + // Port 80 (non-service, i.e., daddr != orchIP) → tcpHTTPPort + // Port 443 → tcpTLSPort + // All other TCP → tcpOtherPort + tcpRedirects := []struct { + dport uint16 + rport uint16 + }{ + {80, hf.config.SandboxTCPFirewallHTTPPort}, + {443, hf.config.SandboxTCPFirewallTLSPort}, + } + for _, tcp := range tcpRedirects { + hf.conn.AddRule(&nftables.Rule{ + Table: hf.table, + Chain: preChain, + Exprs: tcpRedirectExprs(hf.vethSet, tcp.dport, tcp.rport), + }) + } + + // Catch-all TCP redirect: any remaining TCP from v2 veths → tcpOtherPort + hf.conn.AddRule(&nftables.Rule{ + Table: hf.table, + Chain: preChain, + Exprs: tcpCatchAllExprs(hf.vethSet, hf.config.SandboxTCPFirewallOtherPort), + }) + + // --- POSTROUTING chain (NAT) --- + postChain := hf.conn.AddChain(&nftables.Chain{ + Name: "postrouting", + Table: hf.table, + Type: nftables.ChainTypeNAT, + Hooknum: nftables.ChainHookPostrouting, + Priority: nftables.ChainPriorityNATSource, + }) + hf.conn.FlushChain(postChain) + + // ip saddr @v2_host_cidrs oifname masquerade + hf.conn.AddRule(&nftables.Rule{ + Table: hf.table, + Chain: postChain, + Exprs: []expr.Any{ + &expr.Payload{DestRegister: 1, Base: expr.PayloadBaseNetworkHeader, Offset: 12, Len: 4}, + &expr.Lookup{SourceRegister: 1, SetName: hf.cidrSet.Name, SetID: hf.cidrSet.ID}, + &expr.Meta{Key: expr.MetaKeyOIFNAME, Register: 1}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: gwBytes}, + &expr.Masq{}, + }, + }) + + if err := hf.conn.Flush(); err != nil { + return fmt.Errorf("flush chains: %w", err) + } + + return nil +} + +// classifyVeths splits the set's current elements into those no live slot +// owns and the live veths the set is missing. Pure, so the padded-key +// comparison it turns on can be pinned by a unit test. +func classifyVeths(current []nftables.SetElement, desired map[string]bool) (stale []nftables.SetElement, missing []string) { + present := make(map[string]bool, len(current)) + for _, elem := range current { + name := ifnameString(elem.Key) + present[name] = true + if !desired[name] { + stale = append(stale, elem) + } + } + + for name := range desired { + if !present[name] { + missing = append(missing, name) + } + } + + slices.Sort(missing) + + return stale, missing +} + +// ReconcileSlots reconciles host firewall set membership with actual active slots. +// Call on startup after rebuilding the in-memory slot registry from surviving sandboxes. +// - Slots in the registry but missing from sets are added. +// - Set entries not in the registry are removed (stale leftovers from crashed sandboxes). +func (hf *HostFirewall) ReconcileSlots(ctx context.Context, activeSlots []*SlotV2) (err error) { + hf.mu.Lock() + defer hf.mu.Unlock() + + defer hf.resetConnOnError(ctx, &err) + + // Build desired state from active slots + desiredVeths := make(map[string]bool) + desiredCIDRs := make(map[string]bool) // key = hostIP string + for _, sv2 := range activeSlots { + desiredVeths[sv2.Slot.VethName()] = true + desiredCIDRs[sv2.Slot.HostIP.To4().String()] = true + } + + // Read current veth set elements + currentVeths, err := hf.conn.GetSetElements(hf.vethSet) + if err != nil { + return fmt.Errorf("get veth set elements: %w", err) + } + + stale, missing := classifyVeths(currentVeths, desiredVeths) + if len(stale) > 0 { + hf.conn.SetDeleteElements(hf.vethSet, stale) + } + for _, veth := range missing { + if err := hf.conn.SetAddElements(hf.vethSet, []nftables.SetElement{ + {Key: ifnameBytes(veth)}, + }); err != nil { + return fmt.Errorf("add missing veth %s: %w", veth, err) + } + } + + // Reconcile CIDR set: flush and rebuild (interval sets don't support + // element-level comparison easily, so rebuild is simpler and correct) + hf.conn.FlushSet(hf.cidrSet) + for _, sv2 := range activeSlots { + hostIP := sv2.Slot.HostIP.To4() + nextIP := incrementIP(hostIP) + if err := hf.conn.SetAddElements(hf.cidrSet, []nftables.SetElement{ + {Key: hostIP}, + {Key: nextIP, IntervalEnd: true}, + }); err != nil { + return fmt.Errorf("add host cidr for slot %d: %w", sv2.Slot.Idx, err) + } + } + + if err := hf.conn.Flush(); err != nil { + return fmt.Errorf("flush reconcile: %w", err) + } + + return nil +} + +// AddSlot adds the veth name and host CIDR for a v2 slot. +func (hf *HostFirewall) AddSlot(ctx context.Context, slotV2 *SlotV2) (err error) { + hf.mu.Lock() + defer hf.mu.Unlock() + + defer hf.resetConnOnError(ctx, &err) + + slot := slotV2.Slot + + // Add veth name to v2_veths set + if err := hf.conn.SetAddElements(hf.vethSet, []nftables.SetElement{ + {Key: ifnameBytes(slot.VethName())}, + }); err != nil { + return fmt.Errorf("add veth to set: %w", err) + } + + // Add host CIDR to v2_host_cidrs interval set (/32) + hostIP := slot.HostIP.To4() + nextIP := incrementIP(hostIP) + if err := hf.conn.SetAddElements(hf.cidrSet, []nftables.SetElement{ + {Key: hostIP}, + {Key: nextIP, IntervalEnd: true}, + }); err != nil { + return fmt.Errorf("add host cidr to set: %w", err) + } + + if err := hf.conn.Flush(); err != nil { + return fmt.Errorf("flush add slot: %w", err) + } + + return nil +} + +// RemoveSlot removes the veth name and host CIDR for a v2 slot. +func (hf *HostFirewall) RemoveSlot(ctx context.Context, slotV2 *SlotV2) (err error) { + hf.mu.Lock() + defer hf.mu.Unlock() + + defer hf.resetConnOnError(ctx, &err) + + slot := slotV2.Slot + hostIP := slot.HostIP.To4() + + // One flush per set, not one for both: an nftables batch is atomic, so an + // already-absent element aborts the whole thing and would leave the other + // set's element installed while teardown reported success. + vethErr := hf.deleteElements(hf.vethSet, []nftables.SetElement{ + {Key: ifnameBytes(slot.VethName())}, + }) + cidrErr := hf.deleteElements(hf.cidrSet, []nftables.SetElement{ + {Key: hostIP}, + {Key: incrementIP(hostIP), IntervalEnd: true}, + }) + + return errors.Join(vethErr, cidrErr) +} + +// deleteElements drops elements from a set, treating already-absent ones as +// success: teardown has to be retryable after a partial create. +func (hf *HostFirewall) deleteElements(set *nftables.Set, elements []nftables.SetElement) error { + hf.conn.SetDeleteElements(set, elements) + + if err := hf.conn.Flush(); err != nil && !errors.Is(err, unix.ENOENT) { + return fmt.Errorf("flush delete from %s: %w", set.Name, err) + } + + return nil +} + +// Close releases the netlink connection, and deletes the table only when no +// slot is left in it. Sandboxes outlive the orchestrator process: deleting the +// table while their veths are still registered would cut their connectivity +// until the next start rebuilt it. +func (hf *HostFirewall) Close() error { + hf.mu.Lock() + defer hf.mu.Unlock() + + elements, err := hf.conn.GetSetElements(hf.vethSet) + if err != nil { + return errors.Join(fmt.Errorf("get veth set elements: %w", err), hf.conn.CloseLasting()) + } + + if len(elements) == 0 { + hf.conn.DelTable(hf.table) + if err := hf.conn.Flush(); err != nil { + return errors.Join(fmt.Errorf("delete host firewall table: %w", err), hf.conn.CloseLasting()) + } + } + + return hf.conn.CloseLasting() +} + +// --- nftables expression builders --- + +// svcRedirectExprs builds: iifname @set tcp dport X ip daddr redirect to :port +func svcRedirectExprs(vethSet *nftables.Set, orchIP net.IP, dport, rport uint16) []expr.Any { + return []expr.Any{ + // Match TCP + &expr.Meta{Key: expr.MetaKeyL4PROTO, Register: 1}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: []byte{unix.IPPROTO_TCP}}, + // iifname @v2_veths + &expr.Meta{Key: expr.MetaKeyIIFNAME, Register: 1}, + &expr.Lookup{SourceRegister: 1, SetName: vethSet.Name, SetID: vethSet.ID}, + // ip daddr == orchIP + &expr.Payload{DestRegister: 1, Base: expr.PayloadBaseNetworkHeader, Offset: 16, Len: 4}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: orchIP}, + // tcp dport == dport + &expr.Payload{DestRegister: 1, Base: expr.PayloadBaseTransportHeader, Offset: 2, Len: 2}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: portBytes(dport)}, + // redirect to :rport + &expr.Immediate{Register: 1, Data: portBytes(rport)}, + &expr.Redir{RegisterProtoMin: 1}, + } +} + +// tcpRedirectExprs builds: iifname @set tcp dport X redirect to :port +func tcpRedirectExprs(vethSet *nftables.Set, dport, rport uint16) []expr.Any { + return []expr.Any{ + &expr.Meta{Key: expr.MetaKeyL4PROTO, Register: 1}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: []byte{unix.IPPROTO_TCP}}, + &expr.Meta{Key: expr.MetaKeyIIFNAME, Register: 1}, + &expr.Lookup{SourceRegister: 1, SetName: vethSet.Name, SetID: vethSet.ID}, + &expr.Payload{DestRegister: 1, Base: expr.PayloadBaseTransportHeader, Offset: 2, Len: 2}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: portBytes(dport)}, + &expr.Immediate{Register: 1, Data: portBytes(rport)}, + &expr.Redir{RegisterProtoMin: 1}, + } +} + +// tcpCatchAllExprs builds: iifname @set tcp protocol redirect to :port +func tcpCatchAllExprs(vethSet *nftables.Set, rport uint16) []expr.Any { + return []expr.Any{ + &expr.Meta{Key: expr.MetaKeyL4PROTO, Register: 1}, + &expr.Cmp{Register: 1, Op: expr.CmpOpEq, Data: []byte{unix.IPPROTO_TCP}}, + &expr.Meta{Key: expr.MetaKeyIIFNAME, Register: 1}, + &expr.Lookup{SourceRegister: 1, SetName: vethSet.Name, SetID: vethSet.ID}, + &expr.Immediate{Register: 1, Data: portBytes(rport)}, + &expr.Redir{RegisterProtoMin: 1}, + } +} + +// --- helpers --- + +// ifnameString reads a set key back as an interface name, dropping the NUL +// padding nftables stores it with. +func ifnameString(key []byte) string { + return string(bytes.TrimRight(key, "\x00")) +} + +// resetConnOnError replaces the shared conn whenever a mutation fails. A +// serialization error sticks to the conn for its lifetime, and a mutation that +// returns before reaching Flush leaves its messages queued for whoever flushes +// next; nftables.Conn exposes no way to clear either. Every slot on the host +// shares this conn, so one failure would otherwise follow every later slot. +func (hf *HostFirewall) resetConnOnError(ctx context.Context, err *error) { + if *err == nil { + return + } + + *err = errors.Join(*err, hf.resetConn(ctx)) +} + +func (hf *HostFirewall) resetConn(ctx context.Context) error { + closeErr := hf.conn.CloseLasting() + + conn, err := nftables.New(nftables.AsLasting()) + if err != nil { + err = fmt.Errorf("open new lasting nftables conn: %w", err) + + var transientErr error + conn, transientErr = nftables.New() + if transientErr != nil { + err = errors.Join(err, fmt.Errorf("open transient nftables conn: %w", transientErr)) + } + } + + resetErr := errors.Join(closeErr, err) + if conn == nil { + // Keep the old (closed, poisoned) conn rather than storing nil and + // panicking on next use; the firewall is left degraded. + logger.L().Error(ctx, "host firewall nftables conn reset failed; reusing the old conn", zap.Error(resetErr)) + + return resetErr + } + + hf.conn = conn + if resetErr != nil { + logger.L().Error(ctx, "host firewall nftables conn reset encountered errors", zap.Error(resetErr)) + } else { + logger.L().Warn(ctx, "host firewall nftables conn reset after a failed mutation") + } + + return resetErr +} + +func portBytes(port uint16) []byte { + b := make([]byte, 2) + binary.BigEndian.PutUint16(b, port) + + return b +} + +func incrementIP(ip net.IP) net.IP { + result := make(net.IP, len(ip)) + copy(result, ip) + for i := range slices.Backward(result) { + result[i]++ + if result[i] != 0 { + break + } + } + + return result +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/host_firewall_test.go b/packages/orchestrator/pkg/sandbox/network/v2/host_firewall_test.go new file mode 100644 index 0000000000..089ac32300 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/host_firewall_test.go @@ -0,0 +1,129 @@ +//go:build linux + +package v2 + +import ( + "net" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" +) + +func testConfig() network.Config { + return network.Config{ + OrchestratorInSandboxIPAddress: "192.0.2.1", + HyperloopProxyPort: 5010, + NFSProxyPort: 5011, + PortmapperPort: 5012, + SandboxTCPFirewallHTTPPort: 5016, + SandboxTCPFirewallTLSPort: 5017, + SandboxTCPFirewallOtherPort: 5018, + } +} + +func TestHostFirewall_Init(t *testing.T) { //nolint:paralleltest // creates/deletes the singleton nftables table "v2-host-firewall" shared by all HostFirewall tests + skipIfNotLinuxRoot(t) + + hf := newTestHostFirewall(t, testConfig()) + + assert.NotNil(t, hf.conn) + assert.NotNil(t, hf.table) + assert.NotNil(t, hf.vethSet) + assert.NotNil(t, hf.cidrSet) +} + +func TestHostFirewall_AddRemoveSlot(t *testing.T) { //nolint:paralleltest // creates/deletes the singleton nftables table "v2-host-firewall" shared by all HostFirewall tests + skipIfNotLinuxRoot(t) + + ctx := t.Context() + + hf := newTestHostFirewall(t, testConfig()) + + slot := makeTestSlot(t, 1) + sv2 := NewSlotV2(slot) + + // Add slot + require.NoError(t, hf.AddSlot(ctx, sv2)) + + // Verify veth set has the element + elements, err := hf.conn.GetSetElements(hf.vethSet) + require.NoError(t, err) + assert.NotEmpty(t, elements, "veth set should have elements") + + // Verify cidr set has elements + cidrElements, err := hf.conn.GetSetElements(hf.cidrSet) + require.NoError(t, err) + assert.NotEmpty(t, cidrElements, "cidr set should have elements") + + // Remove slot + require.NoError(t, hf.RemoveSlot(ctx, sv2)) + + // Verify veth set is empty + elements, err = hf.conn.GetSetElements(hf.vethSet) + require.NoError(t, err) + assert.Empty(t, elements, "veth set should be empty after removal") +} + +func TestHostFirewall_ConcurrentAccess(t *testing.T) { //nolint:paralleltest // creates/deletes the singleton nftables table "v2-host-firewall" shared by all HostFirewall tests + skipIfNotLinuxRoot(t) + + ctx := t.Context() + + hf := newTestHostFirewall(t, testConfig()) + + // Add multiple slots concurrently + done := make(chan error, 10) + for i := 1; i <= 10; i++ { + sv2 := NewSlotV2(makeTestSlot(t, i)) + t.Cleanup(func() { _ = hf.RemoveSlot(ctx, sv2) }) + go func() { + done <- hf.AddSlot(ctx, sv2) + }() + } + + for range 10 { + err := <-done + require.NoError(t, err, "concurrent AddSlot should not fail") + } + + // Verify all 10 veths added + elements, err := hf.conn.GetSetElements(hf.vethSet) + require.NoError(t, err) + assert.Len(t, elements, 10) +} + +func TestIncrementIP(t *testing.T) { + t.Parallel() + + tests := []struct { + input string + expected string + }{ + {"10.11.0.1", "10.11.0.2"}, + {"10.11.0.255", "10.11.1.0"}, + {"10.11.255.255", "10.12.0.0"}, + {"192.168.1.1", "192.168.1.2"}, + } + + for _, tt := range tests { + ip := net.ParseIP(tt.input).To4() + result := incrementIP(ip) + assert.Equal(t, tt.expected, result.String(), "incrementIP(%s)", tt.input) + } +} + +func TestPortBytes(t *testing.T) { + t.Parallel() + + b := portBytes(80) + assert.Equal(t, []byte{0, 80}, b) + + b = portBytes(443) + assert.Equal(t, []byte{1, 187}, b) + + b = portBytes(5010) + assert.Equal(t, []byte{0x13, 0x92}, b) +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/main_test.go b/packages/orchestrator/pkg/sandbox/network/v2/main_test.go new file mode 100644 index 0000000000..5bc99052f5 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/main_test.go @@ -0,0 +1,36 @@ +//go:build linux + +package v2 + +import ( + "context" + "fmt" + "os" + "testing" +) + +// The host firewall table is a host-wide singleton, so elements left by a +// crashed run would fail the emptiness assertions of a later one. Reconcile it +// to empty once per process — the same call the orchestrator makes at startup. +func TestMain(m *testing.M) { + if os.Geteuid() == 0 { + if err := reconcileHostFirewall(); err != nil { + panic(fmt.Sprintf("cannot reset the host firewall table: %v", err)) + } + } + + m.Run() +} + +func reconcileHostFirewall() error { + hf, err := NewHostFirewall("lo", testConfig()) + if err != nil { + return err + } + + if err := hf.ReconcileSlots(context.Background(), nil); err != nil { + return err + } + + return hf.Close() +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/network.go b/packages/orchestrator/pkg/sandbox/network/v2/network.go new file mode 100644 index 0000000000..f006f29a3b --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/network.go @@ -0,0 +1,321 @@ +//go:build linux + +package v2 + +import ( + "context" + "errors" + "fmt" + "net" + "runtime" + + "github.com/google/nftables" + "github.com/vishvananda/netlink" + "github.com/vishvananda/netns" + "go.uber.org/zap" + + "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" + "github.com/e2b-dev/infra/packages/shared/pkg/logger" +) + +// CreateNetworkV2 sets up the full network stack for a v2 slot. +// It reuses the same netlink setup pattern as v1 (namespace, veth, tap, routes) +// but replaces all iptables calls with nftables. +// +// Compared to v1: +// - In-namespace NAT: nftables SetupNamespaceNAT() replaces 2 iptables rules +// - Host firewall: hf.AddSlot() replaces 8 iptables rules with 2 set element adds +// - eBPF: observer.Attach() adds optional per-veth counters (best-effort) +func CreateNetworkV2(ctx context.Context, slot *network.Slot, slotV2 *SlotV2, + hf *HostFirewall, observer *VethObserver, +) error { + // Run on a goroutine of its own so the thread it pins is this call's + // alone. createNetworkV2 keeps that thread locked if it cannot get back to + // the host namespace, and the runtime destroys a still-locked thread when + // its goroutine exits — a caller's long-lived goroutine would instead keep + // building every later slot inside the namespace it got stuck in. + done := make(chan error, 1) + go func() { + done <- createNetworkV2(ctx, slot, slotV2, hf, observer) + }() + + return <-done +} + +func createNetworkV2(ctx context.Context, slot *network.Slot, slotV2 *SlotV2, + hf *HostFirewall, observer *VethObserver, +) (retErr error) { + // Prevent thread changes so we can safely manipulate namespaces + runtime.LockOSThread() + + // Save the original (host) namespace + hostNS, err := netns.Get() + if err != nil { + runtime.UnlockOSThread() + + return fmt.Errorf("cannot get current (host) namespace: %w", err) + } + + cleanupNeeded := false + defer func() { + restoreErr := netns.Set(hostNS) + if restoreErr != nil { + // Deliberately no UnlockOSThread: the thread is still inside the + // sandbox namespace, so it must die with this goroutine rather + // than be reused. The slot fails with it — its host-side state + // cannot be trusted, and cleanup would run in the wrong namespace. + retErr = errors.Join(retErr, fmt.Errorf("error resetting network namespace back to host: %w", restoreErr)) + logger.L().Error(ctx, "error resetting network namespace back to host", zap.Error(restoreErr)) + } else { + runtime.UnlockOSThread() + + if retErr != nil && cleanupNeeded { + if cleanupErr := RemoveNetworkV2(ctx, slot, slotV2, hf, observer); cleanupErr != nil { + retErr = errors.Join(retErr, fmt.Errorf("error cleaning up partially created v2 network: %w", cleanupErr)) + } + } + } + + hostNS.Close() + }() + + // An existing namespace for this index is a stale reclaim anchor from a + // failed teardown whose rules, routes and veth may still exist. Run a full + // (idempotent) teardown to reclaim them; deleting only the namespace would + // orphan that state. + available, err := network.IsNamespaceAvailable(network.NetNamespacesDir, slot.NamespaceID()) + if err != nil { + return fmt.Errorf("cannot check for stale namespace: %w", err) + } + if !available { + if err := RemoveNetworkV2(ctx, slot, slotV2, hf, observer); err != nil { + return fmt.Errorf("cannot reclaim stale v2 network slot: %w", err) + } + } + + // --- Create namespace --- + ns, err := netns.NewNamed(slot.NamespaceID()) + if err != nil { + return fmt.Errorf("cannot create new namespace: %w", err) + } + cleanupNeeded = true + + defer ns.Close() + + // --- Create veth pair (inside new namespace, then move veth to host) --- + vethAttrs := netlink.NewLinkAttrs() + vethAttrs.Name = slot.VethName() + veth := &netlink.Veth{ + LinkAttrs: vethAttrs, + PeerName: slot.VpeerName(), + } + + if err := netlink.LinkAdd(veth); err != nil { + return fmt.Errorf("error creating veth device: %w", err) + } + + vpeer, err := netlink.LinkByName(slot.VpeerName()) + if err != nil { + return fmt.Errorf("error finding vpeer: %w", err) + } + + if err := netlink.LinkSetUp(vpeer); err != nil { + return fmt.Errorf("error setting vpeer device up: %w", err) + } + + if err := netlink.AddrAdd(vpeer, &netlink.Addr{ + IPNet: &net.IPNet{IP: slot.VpeerIP(), Mask: slot.VrtMask()}, + }); err != nil { + return fmt.Errorf("error adding vpeer device address: %w", err) + } + + // Move veth to host namespace + if err := netlink.LinkSetNsFd(veth, int(hostNS)); err != nil { + return fmt.Errorf("error moving veth device to host namespace: %w", err) + } + + if err := netns.Set(hostNS); err != nil { + return fmt.Errorf("error setting network namespace: %w", err) + } + + vethInHost, err := netlink.LinkByName(slot.VethName()) + if err != nil { + return fmt.Errorf("error finding veth: %w", err) + } + + if err := netlink.LinkSetUp(vethInHost); err != nil { + return fmt.Errorf("error setting veth device up: %w", err) + } + + if err := netlink.AddrAdd(vethInHost, &netlink.Addr{ + IPNet: &net.IPNet{IP: slot.VethIP(), Mask: slot.VrtMask()}, + }); err != nil { + return fmt.Errorf("error adding veth device address: %w", err) + } + + // Switch back to sandbox namespace for tap setup + if err := netns.Set(ns); err != nil { + return fmt.Errorf("error setting network namespace to %s: %w", ns.String(), err) + } + + // --- Create TAP device for Firecracker --- + tapAttrs := netlink.NewLinkAttrs() + tapAttrs.Name = slot.TapName() + tapAttrs.Namespace = ns + tap := &netlink.Tuntap{ + Mode: netlink.TUNTAP_MODE_TAP, + LinkAttrs: tapAttrs, + } + + if err := netlink.LinkAdd(tap); err != nil { + return fmt.Errorf("error creating tap device: %w", err) + } + + // Keep resumed guests' cached gateway ARP entries valid. Tap devices are + // isolated in per-sandbox network namespaces, so sharing this MAC is safe. + if err := netlink.LinkSetHardwareAddr(tap, network.TapHostHardwareAddr()); err != nil { + return fmt.Errorf("error setting tap device hardware address: %w", err) + } + + if err := netlink.LinkSetUp(tap); err != nil { + return fmt.Errorf("error setting tap device up: %w", err) + } + + if err := netlink.AddrAdd(tap, &netlink.Addr{ + IPNet: &net.IPNet{IP: slot.TapIP(), Mask: slot.TapCIDR()}, + }); err != nil { + return fmt.Errorf("error setting address of the tap device: %w", err) + } + + // --- Loopback --- + lo, err := netlink.LinkByName("lo") + if err != nil { + return fmt.Errorf("error finding lo: %w", err) + } + if err := netlink.LinkSetUp(lo); err != nil { + return fmt.Errorf("error setting lo device up: %w", err) + } + + // --- Default route inside namespace --- + if err := netlink.RouteAdd(&netlink.Route{ + Scope: netlink.SCOPE_UNIVERSE, + Gw: slot.VethIP(), + }); err != nil { + return fmt.Errorf("error adding default NS route: %w", err) + } + + // --- In-namespace NAT (replaces 2 iptables rules) --- + // We need a fresh nftables conn inside this namespace context + nsConn, err := nftables.New(nftables.AsLasting()) + if err != nil { + return fmt.Errorf("nftables conn in namespace: %w", err) + } + + // The v1 Firewall creates a table named "slot-firewall" — reuse it + nsTable := nsConn.AddTable(&nftables.Table{ + Name: "slot-firewall", + Family: nftables.TableFamilyINet, + }) + + if err := SetupNamespaceNAT(nsConn, nsTable, slot.VpeerName(), slot.HostIPString(), slot.NamespaceIP()); err != nil { + nsConn.CloseLasting() + + return fmt.Errorf("setup namespace NAT: %w", err) + } + + // Slots are pooled: seed the untenanted (regular-sandbox) class; the pool + // re-stamps per tenant on Get and restores it on recycle. + if err := SetupEgressDSCP(nsConn, nsTable, slot.VpeerName(), hf.config.EgressDSCP(network.EgressClassSandbox)); err != nil { + nsConn.CloseLasting() + + return fmt.Errorf("setup egress DSCP: %w", err) + } + nsConn.CloseLasting() + + // --- Initialize in-namespace firewall (reuse v1 Firewall) --- + if err := slot.InitializeFirewall(); err != nil { + return fmt.Errorf("error initializing slot firewall: %w", err) + } + + // --- Switch back to host namespace --- + if err := netns.Set(hostNS); err != nil { + return fmt.Errorf("error setting network namespace to %s: %w", hostNS.String(), err) + } + + // --- Host route to namespace --- + if err := netlink.RouteAdd(&netlink.Route{ + Gw: slot.VpeerIP(), + Dst: slot.HostNet(), + }); err != nil { + return fmt.Errorf("error adding route from host to FC: %w", err) + } + + // --- Host firewall: add slot to sets (replaces 8 iptables rules) --- + if err := hf.AddSlot(ctx, slotV2); err != nil { + return fmt.Errorf("error adding slot to host firewall: %w", err) + } + + // --- eBPF observability (best-effort) --- + if observer != nil { + if err := observer.Attach(slot.VethName()); err != nil { + logger.L().Warn(ctx, "failed to attach eBPF observer (non-fatal)", zap.Error(err)) + } + } + + return nil +} + +// RemoveNetworkV2 tears down the network for a v2 slot. It is idempotent: +// state that is already gone is not an error, so it can reclaim a partially +// created or previously half-torn-down slot. +func RemoveNetworkV2(ctx context.Context, slot *network.Slot, slotV2 *SlotV2, + hf *HostFirewall, observer *VethObserver, +) error { + var errs []error + + // Close in-namespace firewall + if err := slot.CloseFirewall(); err != nil { + errs = append(errs, fmt.Errorf("error closing firewall: %w", err)) + } + + // Remove from host firewall sets + if err := hf.RemoveSlot(ctx, slotV2); err != nil { + errs = append(errs, fmt.Errorf("error removing slot from host firewall: %w", err)) + } + + // Detach eBPF observer + if observer != nil { + if err := observer.Detach(slot.VethName()); err != nil { + errs = append(errs, fmt.Errorf("error detaching eBPF observer: %w", err)) + } + } + + // Delete host route + err := netlink.RouteDel(&netlink.Route{ + Gw: slot.VpeerIP(), + Dst: slot.HostNet(), + }) + appendUnlessAbsent(&errs, err, isRouteNotExist, "error deleting route from host to FC: %w") + + // Delete veth device explicitly (prevents race on reuse) + vethLink, err := netlink.LinkByName(slot.VethName()) + if err != nil { + appendUnlessAbsent(&errs, err, isLinkNotExist, "error finding veth: %w") + } else { + appendUnlessAbsent(&errs, netlink.LinkDel(vethLink), isLinkNotExist, "error deleting veth device: %w") + } + + // The /run/netns entry is the anchor startup reclaim uses to rediscover a + // leaked slot, and every host-side item above is keyed by the slot index: + // deleting it after a failed teardown would orphan that state with no way + // to retry it. CreateNetworkV2 removes a stale anchor before reusing the + // slot. + if len(errs) > 0 { + return errors.Join(errs...) + } + + err = netns.DeleteNamed(slot.NamespaceID()) + appendUnlessAbsent(&errs, err, isNamespaceNotExist, "error deleting namespace: %w") + + return errors.Join(errs...) +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/network_test.go b/packages/orchestrator/pkg/sandbox/network/v2/network_test.go new file mode 100644 index 0000000000..02325e48ae --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/network_test.go @@ -0,0 +1,114 @@ +//go:build linux + +package v2 + +import ( + "context" + "os/exec" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCreateNetworkV2_FullLifecycle(t *testing.T) { //nolint:paralleltest // mutates host netns state: singleton nftables table "v2-host-firewall", named netns, veth links + skipIfNotLinuxRoot(t) + + ctx := context.Background() + + // Create host firewall + hf := newTestHostFirewall(t, testConfig()) + + observer, err := NewVethObserver() + require.NoError(t, err) + defer observer.Close() + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + + // Create network + require.NoError(t, CreateNetworkV2(ctx, slot, sv2, hf, observer)) + + // Verify namespace exists + out, err := exec.CommandContext(ctx, "ip", "netns", "list").Output() + require.NoError(t, err) + assert.Contains(t, string(out), slot.NamespaceID()) + + // Verify veth exists in host + out, err = exec.CommandContext(ctx, "ip", "link", "show", slot.VethName()).Output() + require.NoError(t, err) + assert.Contains(t, string(out), slot.VethName()) + + // Teardown + err = RemoveNetworkV2(ctx, slot, sv2, hf, observer) + require.NoError(t, err) + + // Verify namespace is gone — check each line for exact match + out, _ = exec.CommandContext(ctx, "ip", "netns", "list").Output() + for line := range strings.SplitSeq(string(out), "\n") { + assert.NotContains(t, line, slot.NamespaceID(), + "namespace should be removed after teardown") + } +} + +func TestCreateNetworkV2_NoIptablesRules(t *testing.T) { //nolint:paralleltest // mutates host netns state: singleton nftables table "v2-host-firewall", named netns, veth links + skipIfNotLinuxRoot(t) + + ctx := context.Background() + + hf := newTestHostFirewall(t, testConfig()) + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + + require.NoError(t, CreateNetworkV2(ctx, slot, sv2, hf, nil)) + t.Cleanup(func() { _ = RemoveNetworkV2(ctx, slot, sv2, hf, nil) }) + + // A conditional assertion would pass vacuously wherever iptables is + // missing, which is exactly the host v2 targets. + if _, err := exec.LookPath("iptables"); err != nil { + t.Skip("iptables binary not on PATH; cannot prove the absence of iptables rules") + } + + for _, chain := range []struct{ table, name string }{ + {"nat", "PREROUTING"}, + {"nat", "POSTROUTING"}, + {"filter", "FORWARD"}, + } { + out, err := exec.CommandContext(ctx, "iptables", "-t", chain.table, "-L", chain.name, "-n").Output() + require.NoError(t, err) + assert.NotContainsf(t, string(out), slot.VethName(), + "a v2 sandbox must not appear in iptables %s/%s", chain.table, chain.name) + } +} + +func TestCreateNetworkV2_CleanTeardown(t *testing.T) { //nolint:paralleltest // mutates host netns state: singleton nftables table "v2-host-firewall", named netns, veth links + skipIfNotLinuxRoot(t) + + ctx := context.Background() + + hf := newTestHostFirewall(t, testConfig()) + + observer, _ := NewVethObserver() + defer observer.Close() + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + + require.NoError(t, CreateNetworkV2(ctx, slot, sv2, hf, observer)) + require.NoError(t, RemoveNetworkV2(ctx, slot, sv2, hf, observer)) + + // Verify veth set is empty + elements, err := hf.conn.GetSetElements(hf.vethSet) + require.NoError(t, err) + assert.Empty(t, elements, "veth set should be empty after teardown") + + // Verify no routes to the slot's host IP + out, err := exec.CommandContext(ctx, "ip", "route", "show").Output() + require.NoError(t, err) + for line := range strings.SplitSeq(string(out), "\n") { + assert.NotContains(t, line, slot.HostIPString(), + "should have no route to slot host IP after teardown") + } +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/ns_nat.go b/packages/orchestrator/pkg/sandbox/network/v2/ns_nat.go new file mode 100644 index 0000000000..da279bb9e2 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/ns_nat.go @@ -0,0 +1,143 @@ +//go:build linux + +package v2 + +import ( + "fmt" + "net" + + "github.com/google/nftables" + "github.com/google/nftables/expr" + "golang.org/x/sys/unix" +) + +// SetupNamespaceNAT adds SNAT/DNAT chains to the existing "slot-firewall" nftables +// table inside a network namespace. This replaces the two iptables rules that v1 creates: +// - iptables -t nat POSTROUTING -o eth0 -s -j SNAT --to +// - iptables -t nat PREROUTING -i eth0 -d -j DNAT --to +// +// The nftables equivalent uses the same kernel conntrack, so SO_ORIGINAL_DST +// in the TCP firewall proxy continues to work unchanged. +func SetupNamespaceNAT(conn *nftables.Conn, table *nftables.Table, vpeerIface, hostIP, namespaceIP string) error { + hostIPParsed := net.ParseIP(hostIP).To4() + if hostIPParsed == nil { + return fmt.Errorf("invalid host IP: %s", hostIP) + } + + nsIPParsed := net.ParseIP(namespaceIP).To4() + if nsIPParsed == nil { + return fmt.Errorf("invalid namespace IP: %s", namespaceIP) + } + + // POSTROUTING NAT chain: SNAT namespace IP → host IP + postChain := conn.AddChain(&nftables.Chain{ + Name: "postroute_nat", + Table: table, + Type: nftables.ChainTypeNAT, + Hooknum: nftables.ChainHookPostrouting, + Priority: nftables.ChainPriorityNATSource, + }) + + // Match: oifname == vpeerIface && ip saddr == namespaceIP → snat to hostIP + conn.AddRule(&nftables.Rule{ + Table: table, + Chain: postChain, + Exprs: []expr.Any{ + // Match output interface + &expr.Meta{Key: expr.MetaKeyOIFNAME, Register: 1}, + &expr.Cmp{ + Register: 1, + Op: expr.CmpOpEq, + Data: ifnameBytes(vpeerIface), + }, + // Match source IP == namespaceIP + &expr.Payload{ + DestRegister: 1, + Base: expr.PayloadBaseNetworkHeader, + Offset: 12, // src IP offset in IPv4 + Len: 4, + }, + &expr.Cmp{ + Register: 1, + Op: expr.CmpOpEq, + Data: nsIPParsed, + }, + // SNAT to hostIP + &expr.Immediate{ + Register: 1, + Data: hostIPParsed, + }, + &expr.NAT{ + Type: expr.NATTypeSourceNAT, + Family: unix.NFPROTO_IPV4, + RegAddrMin: 1, + RegAddrMax: 1, + RegProtoMin: 0, + RegProtoMax: 0, + }, + }, + }) + + // PREROUTING NAT chain: DNAT hostIP → namespaceIP + preChain := conn.AddChain(&nftables.Chain{ + Name: "preroute_nat", + Table: table, + Type: nftables.ChainTypeNAT, + Hooknum: nftables.ChainHookPrerouting, + Priority: nftables.ChainPriorityNATDest, + }) + + // Match: iifname == vpeerIface && ip daddr == hostIP → dnat to namespaceIP + conn.AddRule(&nftables.Rule{ + Table: table, + Chain: preChain, + Exprs: []expr.Any{ + // Match input interface + &expr.Meta{Key: expr.MetaKeyIIFNAME, Register: 1}, + &expr.Cmp{ + Register: 1, + Op: expr.CmpOpEq, + Data: ifnameBytes(vpeerIface), + }, + // Match destination IP == hostIP + &expr.Payload{ + DestRegister: 1, + Base: expr.PayloadBaseNetworkHeader, + Offset: 16, // dst IP offset in IPv4 + Len: 4, + }, + &expr.Cmp{ + Register: 1, + Op: expr.CmpOpEq, + Data: hostIPParsed, + }, + // DNAT to namespaceIP + &expr.Immediate{ + Register: 1, + Data: nsIPParsed, + }, + &expr.NAT{ + Type: expr.NATTypeDestNAT, + Family: unix.NFPROTO_IPV4, + RegAddrMin: 1, + RegAddrMax: 1, + RegProtoMin: 0, + RegProtoMax: 0, + }, + }, + }) + + if err := conn.Flush(); err != nil { + return fmt.Errorf("flush namespace NAT rules: %w", err) + } + + return nil +} + +// ifnameBytes pads an interface name to IFNAMSIZ (16 bytes) with null termination. +func ifnameBytes(name string) []byte { + b := make([]byte, 16) // IFNAMSIZ + copy(b, name) + + return b +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/ns_nat_test.go b/packages/orchestrator/pkg/sandbox/network/v2/ns_nat_test.go new file mode 100644 index 0000000000..93ac6894a4 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/ns_nat_test.go @@ -0,0 +1,91 @@ +//go:build linux + +package v2 + +import ( + "os" + "runtime" + "testing" + + "github.com/google/nftables" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func skipIfNotLinuxRoot(t *testing.T) { + t.Helper() + if runtime.GOOS != "linux" { + t.Skip("skipping: requires linux") + } + if os.Getuid() != 0 { + t.Skip("skipping: requires root") + } +} + +func TestNamespaceNAT_Setup(t *testing.T) { //nolint:paralleltest // creates the nftables table "test-nat" in the host netns and lists chains family-wide + skipIfNotLinuxRoot(t) + + conn, err := nftables.New(nftables.AsLasting()) + require.NoError(t, err) + defer conn.CloseLasting() + + table := conn.AddTable(&nftables.Table{ + Name: "test-nat", + Family: nftables.TableFamilyINet, + }) + require.NoError(t, conn.Flush()) + defer func() { + conn.DelTable(table) + conn.Flush() + }() + + err = SetupNamespaceNAT(conn, table, "eth0", "10.11.0.1", "169.254.0.21") + require.NoError(t, err) + + // Verify chains were created + chains, err := conn.ListChainsOfTableFamily(nftables.TableFamilyINet) + require.NoError(t, err) + + chainNames := make(map[string]bool) + for _, c := range chains { + if c.Table.Name == "test-nat" { + chainNames[c.Name] = true + } + } + assert.True(t, chainNames["postroute_nat"], "postroute_nat chain should exist") + assert.True(t, chainNames["preroute_nat"], "preroute_nat chain should exist") +} + +func TestNamespaceNAT_InvalidIP(t *testing.T) { //nolint:paralleltest // creates the nftables table "test-nat-invalid" in the host netns + skipIfNotLinuxRoot(t) + + conn, err := nftables.New(nftables.AsLasting()) + require.NoError(t, err) + defer conn.CloseLasting() + + table := conn.AddTable(&nftables.Table{ + Name: "test-nat-invalid", + Family: nftables.TableFamilyINet, + }) + require.NoError(t, conn.Flush()) + defer func() { + conn.DelTable(table) + conn.Flush() + }() + + err = SetupNamespaceNAT(conn, table, "eth0", "not-an-ip", "169.254.0.21") + require.Error(t, err) + assert.Contains(t, err.Error(), "invalid host IP") +} + +func TestIfnameBytes(t *testing.T) { + t.Parallel() + + b := ifnameBytes("eth0") + assert.Len(t, b, 16) + assert.Equal(t, byte('e'), b[0]) + assert.Equal(t, byte('t'), b[1]) + assert.Equal(t, byte('h'), b[2]) + assert.Equal(t, byte('0'), b[3]) + assert.Equal(t, byte(0), b[4]) // null terminated +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/observability.go b/packages/orchestrator/pkg/sandbox/network/v2/observability.go new file mode 100644 index 0000000000..cd7ace2e44 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/observability.go @@ -0,0 +1,135 @@ +//go:build linux + +package v2 + +import ( + "fmt" + "sync" + + "github.com/vishvananda/netlink" +) + +// VethObserver provides per-veth packet/byte counters via netlink interface statistics. +// It is nil-safe: all methods on a nil *VethObserver are no-ops. +// +// The abstraction is kept so tc/eBPF can replace netlink stats later for +// higher resolution or per-flow counters. +type VethObserver struct { + mu sync.Mutex + attached map[string]bool // veth name → attached +} + +// NewVethObserver creates a new observer. +func NewVethObserver() (*VethObserver, error) { + return &VethObserver{ + attached: make(map[string]bool), + }, nil +} + +// Attach registers a veth interface for counter collection. +func (o *VethObserver) Attach(vethName string) error { + if o == nil { + return nil + } + o.mu.Lock() + defer o.mu.Unlock() + + if o.attached[vethName] { + return fmt.Errorf("already attached to %s", vethName) + } + // Verify the interface exists + if _, err := netlink.LinkByName(vethName); err != nil { + return fmt.Errorf("interface %s not found: %w", vethName, err) + } + o.attached[vethName] = true + + return nil +} + +// Detach unregisters a veth interface from counter collection. +func (o *VethObserver) Detach(vethName string) error { + if o == nil { + return nil + } + o.mu.Lock() + defer o.mu.Unlock() + + delete(o.attached, vethName) + + return nil +} + +// ReadCounters reads the packet and byte counters for a veth interface +// from kernel netlink interface statistics. Returns combined rx+tx values +// as seen from the host side of the veth pair. +func (o *VethObserver) ReadCounters(vethName string) (packets, bytes uint64, err error) { + if o == nil { + return 0, 0, nil + } + o.mu.Lock() + defer o.mu.Unlock() + + if !o.attached[vethName] { + return 0, 0, fmt.Errorf("not attached to %s", vethName) + } + + link, err := netlink.LinkByName(vethName) + if err != nil { + return 0, 0, fmt.Errorf("get link %s: %w", vethName, err) + } + + stats := link.Attrs().Statistics + if stats == nil { + return 0, 0, nil + } + + // Report rx+tx from the host veth perspective. + // Host veth rx = sandbox tx (sandbox → internet). + // Host veth tx = sandbox rx (internet → sandbox). + packets = stats.RxPackets + stats.TxPackets + bytes = stats.RxBytes + stats.TxBytes + + return packets, bytes, nil +} + +// ReadCountersDirectional reads rx and tx counters separately. +// rx = traffic from sandbox to internet (host veth rx). +// tx = traffic from internet to sandbox (host veth tx). +func (o *VethObserver) ReadCountersDirectional(vethName string) (rxPackets, rxBytes, txPackets, txBytes uint64, err error) { + if o == nil { + return 0, 0, 0, 0, nil + } + o.mu.Lock() + defer o.mu.Unlock() + + if !o.attached[vethName] { + return 0, 0, 0, 0, fmt.Errorf("not attached to %s", vethName) + } + + link, err := netlink.LinkByName(vethName) + if err != nil { + return 0, 0, 0, 0, fmt.Errorf("get link %s: %w", vethName, err) + } + + stats := link.Attrs().Statistics + if stats == nil { + return 0, 0, 0, 0, nil + } + + return stats.RxPackets, stats.RxBytes, stats.TxPackets, stats.TxBytes, nil +} + +// Close detaches all observers. +func (o *VethObserver) Close() error { + if o == nil { + return nil + } + o.mu.Lock() + defer o.mu.Unlock() + + for name := range o.attached { + delete(o.attached, name) + } + + return nil +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/observability_test.go b/packages/orchestrator/pkg/sandbox/network/v2/observability_test.go new file mode 100644 index 0000000000..5581689c30 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/observability_test.go @@ -0,0 +1,157 @@ +//go:build linux + +package v2 + +import ( + "fmt" + "net" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/vishvananda/netlink" +) + +// createTestVeth brings up a real veth pair and returns its host-side name; +// Attach and ReadCounters resolve the interface through netlink, so a name +// with no link behind it fails. The name is unique per call so a concurrent +// package, a repeat run, or a live orchestrator cannot collide with it. +func createTestVeth(t *testing.T) string { + t.Helper() + + name := fmt.Sprintf("vt-%d", reserveNSTestIdx(t)) + link := &netlink.Veth{ + LinkAttrs: netlink.LinkAttrs{Name: name}, + PeerName: name + "-peer", + } + + require.NoError(t, netlink.LinkAdd(link)) + t.Cleanup(func() { + _ = netlink.LinkDel(link) + }) + + return name +} + +// sendTestVethTraffic pushes packets through the veth pair so its link +// counters advance. +func sendTestVethTraffic(t *testing.T, name string) { + t.Helper() + + peer, err := netlink.LinkByName(name + "-peer") + require.NoError(t, err) + require.NoError(t, netlink.LinkSetUp(peer)) + + link, err := netlink.LinkByName(name) + require.NoError(t, err) + require.NoError(t, netlink.LinkSetUp(link)) + + // An unanswered ARP for an address on the peer's subnet is enough to move + // the counters, and needs no addressing on either end. + addr, err := netlink.ParseAddr("169.254.222.1/30") + require.NoError(t, err) + require.NoError(t, netlink.AddrAdd(link, addr)) + + dialer := net.Dialer{Timeout: time.Second} + conn, err := dialer.DialContext(t.Context(), "udp4", "169.254.222.2:9") + require.NoError(t, err) + defer conn.Close() + + for range 4 { + _, _ = conn.Write([]byte("counter-probe")) + } +} + +func TestVethObserver_AttachDetach(t *testing.T) { + t.Parallel() + + veth := createTestVeth(t) + + obs, err := NewVethObserver() + require.NoError(t, err) + defer obs.Close() + + // Attach + err = obs.Attach(veth) + require.NoError(t, err) + + // Double attach should fail + err = obs.Attach(veth) + require.Error(t, err) + + // Detach + err = obs.Detach(veth) + require.NoError(t, err) + + // Double detach is idempotent + err = obs.Detach(veth) + assert.NoError(t, err) +} + +func TestVethObserver_ReadCounters(t *testing.T) { + t.Parallel() + + veth := createTestVeth(t) + + obs, err := NewVethObserver() + require.NoError(t, err) + defer obs.Close() + + // Not attached → error + _, _, err = obs.ReadCounters(veth) + require.Error(t, err) + + // Attach and read + require.NoError(t, obs.Attach(veth)) + packets, bytes, err := obs.ReadCounters(veth) + require.NoError(t, err) + assert.Equal(t, uint64(0), packets) + assert.Equal(t, uint64(0), bytes) + + // Counters that never move would satisfy the assertions above, so make + // the link carry something and read again. + sendTestVethTraffic(t, veth) + + packets, bytes, err = obs.ReadCounters(veth) + require.NoError(t, err) + assert.Positive(t, packets, "counters must track real traffic") + assert.Positive(t, bytes, "counters must track real traffic") +} + +func TestVethObserver_NilSafe(t *testing.T) { + t.Parallel() + + var obs *VethObserver + + // A nil observer returns before touching netlink, so the name need not exist. + assert.NoError(t, obs.Attach("vt-absent")) + assert.NoError(t, obs.Detach("vt-absent")) + assert.NoError(t, obs.Close()) + + p, b, err := obs.ReadCounters("vt-absent") + require.NoError(t, err) + assert.Equal(t, uint64(0), p) + assert.Equal(t, uint64(0), b) +} + +func TestVethObserver_Close(t *testing.T) { + t.Parallel() + + veth1 := createTestVeth(t) + veth2 := createTestVeth(t) + + obs, err := NewVethObserver() + require.NoError(t, err) + + require.NoError(t, obs.Attach(veth1)) + require.NoError(t, obs.Attach(veth2)) + + err = obs.Close() + require.NoError(t, err) + + // After close, nothing should be attached + obs.mu.Lock() + assert.Empty(t, obs.attached) + obs.mu.Unlock() +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/parity_test.go b/packages/orchestrator/pkg/sandbox/network/v2/parity_test.go new file mode 100644 index 0000000000..2c8da0cf4b --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/parity_test.go @@ -0,0 +1,261 @@ +//go:build linux + +package v2 + +import ( + "bytes" + "context" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/containernetworking/plugins/pkg/ns" + "github.com/google/nftables" + "github.com/google/nftables/expr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/vishvananda/netlink" + + "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" +) + +// Resumed guests keep a cached ARP entry for the gateway, so the tap MAC must +// be the same fixed address v1 stamps. +func TestCreateNetworkV2_TapUsesFixedHostMAC(t *testing.T) { //nolint:paralleltest // mutates host netns state: singleton nftables table "v2-host-firewall", named netns, veth links + skipIfNotLinuxRoot(t) + + ctx := context.Background() + + hf := newTestHostFirewall(t, testConfig()) + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + require.NoError(t, CreateNetworkV2(ctx, slot, sv2, hf, nil)) + t.Cleanup(func() { _ = RemoveNetworkV2(ctx, slot, sv2, hf, nil) }) + + nsHandle, err := ns.GetNS(filepath.Join(network.NetNamespacesDir, slot.NamespaceID())) + require.NoError(t, err) + defer nsHandle.Close() + + var got string + require.NoError(t, nsHandle.Do(func(_ ns.NetNS) error { + tap, err := netlink.LinkByName(slot.TapName()) + if err != nil { + return err + } + got = tap.Attrs().HardwareAddr.String() + + return nil + })) + + assert.Equal(t, network.TapHostHardwareAddr().String(), got) +} + +// v1's host FORWARD rules only ever append ACCEPTs, so the v2 table must not +// drop anything: no default-drop policy, and the gw→veth ACCEPT is +// unconditional (a ct-state match would reject new inbound flows v1 allows). +func TestHostFirewall_ForwardChainMatchesV1(t *testing.T) { //nolint:paralleltest // creates/deletes the singleton nftables table "v2-host-firewall" shared by all HostFirewall tests + skipIfNotLinuxRoot(t) + + hf := newTestHostFirewall(t, testConfig()) + + chains, err := hf.conn.ListChainsOfTableFamily(nftables.TableFamilyINet) + require.NoError(t, err) + + var fwd *nftables.Chain + for _, c := range chains { + if c.Table.Name == hostFwTableName && c.Name == "forward" { + fwd = c + + break + } + } + require.NotNil(t, fwd, "forward chain must exist") + require.NotNil(t, fwd.Policy) + assert.Equal(t, nftables.ChainPolicyAccept, *fwd.Policy, "v2 must not install a default-drop forward hook") + + rules, err := hf.conn.GetRules(hf.table, fwd) + require.NoError(t, err) + require.Len(t, rules, 2, "forward chain holds exactly v1's two ACCEPTs") + + for _, r := range rules { + accepts := false + for _, e := range r.Exprs { + if v, ok := e.(*expr.Verdict); ok && v.Kind == expr.VerdictAccept { + accepts = true + } + _, isCt := e.(*expr.Ct) + assert.False(t, isCt, "forward ACCEPTs must be unconditional, as v1's are") + } + assert.True(t, accepts, "every forward rule must be an ACCEPT") + } +} + +// A slot left in the sets by a crashed run must not block its index: reclaim +// tears the slot down, then reconcile drops what it left behind. +func TestHostFirewall_ReconcileClearsStaleSlots(t *testing.T) { //nolint:paralleltest // creates/deletes the singleton nftables table "v2-host-firewall" shared by all HostFirewall tests + skipIfNotLinuxRoot(t) + + ctx := t.Context() + + hf := newTestHostFirewall(t, testConfig()) + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + require.NoError(t, hf.AddSlot(ctx, sv2)) + + require.NoError(t, hf.ReconcileSlots(ctx, nil)) + + elements, err := hf.conn.GetSetElements(hf.vethSet) + require.NoError(t, err) + assert.Empty(t, elements, "reconcile must drop set elements no live slot owns") + + // A live slot is left alone: reconcile must recognise it as already + // present, not delete and re-add it. Comparing set membership by a + // mis-trimmed key classifies every live veth as both stale and missing, + // which nets out to the same element count — so count the delete instead. + require.NoError(t, hf.AddSlot(ctx, sv2)) + require.NoError(t, hf.ReconcileSlots(ctx, []*SlotV2{sv2})) + require.NoError(t, vethSetHas(hf, slot.VethName())) + + require.NoError(t, hf.RemoveSlot(ctx, sv2)) +} + +// Sandboxes outlive the orchestrator process: closing the pool while a slot is +// still registered must leave its rules in place. +func TestHostFirewall_ClosePreservesLiveSlots(t *testing.T) { //nolint:paralleltest // creates/deletes the singleton nftables table "v2-host-firewall" shared by all HostFirewall tests + skipIfNotLinuxRoot(t) + + ctx := t.Context() + + hf, err := NewHostFirewall("lo", testConfig()) + require.NoError(t, err) + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + require.NoError(t, hf.AddSlot(ctx, sv2)) + require.NoError(t, hf.Close()) + + reopened := newTestHostFirewall(t, testConfig()) + require.NoError(t, vethSetHas(reopened, slot.VethName()), "a live slot's rules must survive Close") + + require.NoError(t, reopened.RemoveSlot(ctx, sv2)) + + // With no slot left, Close does tear the table down. + require.NoError(t, reopened.Close()) + after := newTestHostFirewall(t, testConfig()) + elements, err := after.conn.GetSetElements(after.vethSet) + require.NoError(t, err) + assert.Empty(t, elements, "Close must delete the table once no slot is left") +} + +// A live slot's padded set key must match its name: the classification is +// what decides whether reconcile deletes a live sandbox's veth, and comparing +// the raw 16-byte key against the name silently makes every live slot look +// both stale and missing. +func TestClassifyVeths(t *testing.T) { + t.Parallel() + + current := []nftables.SetElement{ + {Key: ifnameBytes("veth-1")}, + {Key: ifnameBytes("veth-99")}, + } + desired := map[string]bool{"veth-1": true, "veth-2": true} + + stale, missing := classifyVeths(current, desired) + + require.Len(t, stale, 1, "only the veth no live slot owns is stale") + require.Equal(t, ifnameBytes("veth-99"), stale[0].Key) + require.Equal(t, []string{"veth-2"}, missing) +} + +// vethSetHas reports whether the veth set holds the named interface. +func vethSetHas(hf *HostFirewall, veth string) error { + elements, err := hf.conn.GetSetElements(hf.vethSet) + if err != nil { + return err + } + + want := ifnameBytes(veth) + for _, e := range elements { + if bytes.Equal(e.Key, want) { + return nil + } + } + + return fmt.Errorf("veth %s not in set (%d elements)", veth, len(elements)) +} + +// Teardown is idempotent: repeated calls, and calls against a slot that was +// never created, must not error — that is what makes create-failure cleanup +// and startup reclaim safe. +func TestRemoveNetworkV2_Idempotent(t *testing.T) { //nolint:paralleltest // mutates host netns state: singleton nftables table "v2-host-firewall", named netns, veth links + skipIfNotLinuxRoot(t) + + ctx := context.Background() + + hf := newTestHostFirewall(t, testConfig()) + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + + // Never created. + require.NoError(t, RemoveNetworkV2(ctx, slot, sv2, hf, nil)) + + require.NoError(t, CreateNetworkV2(ctx, slot, sv2, hf, nil)) + require.NoError(t, RemoveNetworkV2(ctx, slot, sv2, hf, nil)) + require.NoError(t, RemoveNetworkV2(ctx, slot, sv2, hf, nil)) +} + +// A stale namespace from a failed teardown is the reclaim anchor: create must +// reclaim it rather than fail, so the index stays usable. +func TestCreateNetworkV2_ReclaimsStaleNamespace(t *testing.T) { //nolint:paralleltest // mutates host netns state: singleton nftables table "v2-host-firewall", named netns, veth links + skipIfNotLinuxRoot(t) + + ctx := context.Background() + + hf := newTestHostFirewall(t, testConfig()) + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + + require.NoError(t, CreateNetworkV2(ctx, slot, sv2, hf, nil)) + + // Leave the namespace and the set elements behind, as a crashed run would. + require.NoError(t, slot.CloseFirewall()) + + require.NoError(t, CreateNetworkV2(ctx, slot, sv2, hf, nil), "a stale anchor must be reclaimed, not fatal") + t.Cleanup(func() { _ = RemoveNetworkV2(ctx, slot, sv2, hf, nil) }) + + _, err := os.Stat(filepath.Join(network.NetNamespacesDir, slot.NamespaceID())) + require.NoError(t, err, "the rebuilt slot must have its namespace") +} + +// fwmark policy routing is not part of this datapath, so its sysctl must not +// gate startup. +func TestValidateV2Prerequisites_IgnoresSrcValidMark(t *testing.T) { + t.Parallel() + + // Both reads gate on the host's actual sysctls: the assertion only means + // something where src_valid_mark is off, and ip_forward would fail the + // check for an unrelated reason. + if sysctlValue(t, "/proc/sys/net/ipv4/conf/all/src_valid_mark") != "0" { + t.Skip("src_valid_mark is already 1; this box cannot show that the check is gone") + } + if sysctlValue(t, "/proc/sys/net/ipv4/ip_forward") != "1" { + t.Skip("ip_forward is off; the prerequisite check fails here for an unrelated reason") + } + + require.NoError(t, ValidateV2Prerequisites()) +} + +func sysctlValue(t *testing.T, path string) string { + t.Helper() + + raw, err := os.ReadFile(path) + require.NoError(t, err) + + return strings.TrimSpace(string(raw)) +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/pool.go b/packages/orchestrator/pkg/sandbox/network/v2/pool.go new file mode 100644 index 0000000000..03b54d73d4 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/pool.go @@ -0,0 +1,415 @@ +//go:build linux + +package v2 + +import ( + "context" + "errors" + "fmt" + "os" + "strings" + "sync" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + "go.uber.org/zap" + + "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" + "github.com/e2b-dev/infra/packages/shared/pkg/grpc/orchestrator" + "github.com/e2b-dev/infra/packages/shared/pkg/logger" + "github.com/e2b-dev/infra/packages/shared/pkg/telemetry" + "github.com/e2b-dev/infra/packages/shared/pkg/utils" +) + +var ( + meter = otel.Meter("github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network/v2") + + newSlotsAvailableCounter = utils.Must(meter.Int64UpDownCounter("orchestrator.network.v2.slots_pool.new", + metric.WithDescription("Number of new v2 network slots ready to be used."), + metric.WithUnit("{slot}"), + )) + reusableSlotsAvailableCounter = utils.Must(meter.Int64UpDownCounter("orchestrator.network.v2.slots_pool.reused", + metric.WithDescription("Number of reused v2 network slots ready to be used."), + metric.WithUnit("{slot}"), + )) + acquiredSlots = utils.Must(meter.Int64Counter("orchestrator.network.v2.slots_pool.acquired", + metric.WithDescription("Number of v2 network slots acquired."), + metric.WithUnit("{slot}"), + )) + returnedSlotCounter = utils.Must(meter.Int64Counter("orchestrator.network.v2.slots_pool.returned", + metric.WithDescription("Number of v2 network slots returned."), + metric.WithUnit("{slot}"), + )) + releasedSlotCounter = utils.Must(meter.Int64Counter("orchestrator.network.v2.slots_pool.released", + metric.WithDescription("Number of v2 network slots released."), + metric.WithUnit("{slot}"), + )) +) + +// V2Pool implements network.PoolInterface using v2 networking (nftables only, zero iptables). +type V2Pool struct { + config network.Config + storage network.Storage + hostFw *HostFirewall + observer *VethObserver + registry *SlotV2Registry + newSlots chan *network.Slot + reusedSlots chan *network.Slot + done chan struct{} + doneOnce sync.Once + + closeMu sync.RWMutex + closed bool + + // returnsWG tracks in-flight asynchronous slot returns; Close waits for + // it before draining the pool. + returnsWG sync.WaitGroup +} + +// Compile-time check that V2Pool satisfies network.PoolInterface. +var _ network.PoolInterface = (*V2Pool)(nil) + +// ValidateV2Prerequisites checks that required kernel parameters are set for v2 networking. +// Returns an error if any prerequisite is missing. Call before NewV2Pool. +func ValidateV2Prerequisites() error { + checks := []struct { + path string + expected string + desc string + }{ + {"/proc/sys/net/ipv4/ip_forward", "1", "IPv4 forwarding"}, + } + + var errs []error + for _, c := range checks { + val, err := os.ReadFile(c.path) + if err != nil { + errs = append(errs, fmt.Errorf("cannot read %s: %w", c.path, err)) + + continue + } + if strings.TrimSpace(string(val)) != c.expected { + errs = append(errs, fmt.Errorf("%s: %s must be %s, got %q", c.desc, c.path, c.expected, strings.TrimSpace(string(val)))) + } + } + if len(errs) > 0 { + return fmt.Errorf("v2 networking prerequisites not met: %w", errors.Join(errs...)) + } + + return nil +} + +func NewV2Pool(storage network.Storage, config network.Config, hostFw *HostFirewall, observer *VethObserver) *V2Pool { + return &V2Pool{ + config: config, + storage: storage, + hostFw: hostFw, + observer: observer, + registry: NewSlotV2Registry(), + newSlots: make(chan *network.Slot, network.NewSlotsPoolSize-1), + reusedSlots: make(chan *network.Slot, network.ReusedSlotsPoolSize), + done: make(chan struct{}), + } +} + +func (p *V2Pool) createNetworkSlot(ctx context.Context) (*network.Slot, error) { + slot, err := p.storage.Acquire(ctx) + if err != nil { + return nil, fmt.Errorf("failed to acquire network slot: %w", err) + } + + slotV2 := NewSlotV2(slot) + p.registry.Store(slotV2) + + if err := CreateNetworkV2(ctx, slot, slotV2, p.hostFw, p.observer); err != nil { + p.registry.Delete(slot.Idx) + releaseErr := p.storage.Release(slot) + + return nil, fmt.Errorf("failed to create v2 network: %w", errors.Join(err, releaseErr)) + } + + return slot, nil +} + +func (p *V2Pool) Populate(ctx context.Context) { + defer close(p.newSlots) + + for { + select { + case <-p.done: + return + case <-ctx.Done(): + return + default: + slot, err := p.createNetworkSlot(ctx) + if err != nil { + logger.L().Error(ctx, "[v2 network slot pool]: failed to create network", zap.Error(err)) + + continue + } + + select { + case p.newSlots <- slot: + newSlotsAvailableCounter.Add(ctx, 1) + + continue + case <-p.done: + case <-ctx.Done(): + } + + if err := p.cleanup(context.WithoutCancel(ctx), slot); err != nil { + logger.L().Error(ctx, "[v2 network slot pool]: failed to cleanup created slot while closing", zap.Error(err), zap.Int("slot_index", slot.Idx)) + } + + return + } + } +} + +func (p *V2Pool) Get(ctx context.Context, netConfig *orchestrator.SandboxNetworkConfig, class network.EgressClass) (*network.Slot, error) { + var slot *network.Slot + + select { + case <-p.done: + return nil, network.ErrClosed + case s := <-p.reusedSlots: + reusableSlotsAvailableCounter.Add(ctx, -1) + acquiredSlots.Add(ctx, 1, metric.WithAttributes(attribute.String("pool", "reused"))) + telemetry.ReportEvent(ctx, "reused v2 network slot") + slot = s + default: + select { + case <-p.done: + return nil, network.ErrClosed + case <-ctx.Done(): + return nil, ctx.Err() + case s, ok := <-p.newSlots: + // Populate closes the channel on its way out, and a closed channel + // wins this select as readily as done does. + if !ok { + return nil, network.ErrClosed + } + + newSlotsAvailableCounter.Add(ctx, -1) + acquiredSlots.Add(ctx, 1, metric.WithAttributes(attribute.String("pool", "new"))) + telemetry.ReportEvent(ctx, "new v2 network slot") + slot = s + } + } + + if err := p.configureSlot(ctx, slot, netConfig, class); err != nil { + // Never handed out, so nobody listens for the release notification. + if rerr := p.ReturnAsync(context.WithoutCancel(ctx), slot, func(context.Context, string) {}, 0); rerr != nil { + logger.L().Error(ctx, "failed to return v2 slot to pool", zap.Error(rerr), zap.Int("slot_index", slot.Idx)) + } + + return nil, err + } + + return slot, nil +} + +func (p *V2Pool) configureSlot(ctx context.Context, slot *network.Slot, netConfig *orchestrator.SandboxNetworkConfig, class network.EgressClass) error { + // Slots are created before their tenant is known, so a build re-stamps the + // rule CreateNetworkV2 seeded. No-op when both classes resolve alike. + if err := ApplyEgressDSCP(slot, p.config.EgressDSCP(class)); err != nil { + return fmt.Errorf("error setting v2 slot egress DSCP for %s: %w", class, err) + } + + if err := slot.ConfigureInternet(ctx, netConfig); err != nil { + return fmt.Errorf("error setting v2 slot internet access: %w", err) + } + + return nil +} + +func (p *V2Pool) tryTrackReturn() bool { + p.closeMu.RLock() + defer p.closeMu.RUnlock() + + if p.closed { + return false + } + + p.returnsWG.Add(1) + + return true +} + +// ReturnAsync recycles a slot in the background, logging errors instead of +// returning them. Close waits for all in-flight returns before draining the +// pool. If the pool is already closed the slot is cleaned up synchronously. +func (p *V2Pool) ReturnAsync(ctx context.Context, slot *network.Slot, releasedFn network.ReleaseNotify, returnDelay time.Duration) error { + if !p.tryTrackReturn() { + return p.returnSlot(ctx, slot, releasedFn, returnDelay) + } + + go func() { + defer p.returnsWG.Done() + + err := p.returnSlot(ctx, slot, releasedFn, returnDelay) + switch { + case err == nil: + case errors.Is(err, network.ErrClosed), errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded): + logger.L().Warn(ctx, "v2 network slot returned during pool shutdown", zap.Error(err), zap.Int("slot_index", slot.Idx)) + default: + logger.L().Error(ctx, "failed to return v2 network slot to pool", zap.Error(err), zap.Int("slot_index", slot.Idx)) + } + }() + + return nil +} + +// returnSlot recycles a slot that was used by a sandbox. It waits returnDelay +// before making the slot reusable to let inflight requests on the previous +// sandbox drain. +func (p *V2Pool) returnSlot(ctx context.Context, slot *network.Slot, releasedFn network.ReleaseNotify, returnDelay time.Duration) error { + notifyNetworkRelease := sync.OnceFunc(func() { + releasedFn(ctx, slot.HostIPString()) + }) + defer notifyNetworkRelease() + + // If the pool is closed or the context is cancelled during the delay we + // still fall through and clean up the slot to avoid leaking it. + select { + case <-ctx.Done(): + return p.cleanupWith(ctx, slot, ctx.Err()) + case <-p.done: + return p.cleanupWith(ctx, slot, network.ErrClosed) + case <-time.After(returnDelay): + } + + notifyNetworkRelease() + + return p.recycle(ctx, slot) +} + +func (p *V2Pool) recycle(ctx context.Context, slot *network.Slot) error { + // Undo any build-specific DSCP before the next tenant can inherit it. + if err := ApplyEgressDSCP(slot, p.config.EgressDSCP(network.EgressClassSandbox)); err != nil { + return p.cleanupWith(ctx, slot, fmt.Errorf("error resetting v2 slot egress DSCP: %w", err)) + } + + if err := slot.ResetInternet(ctx); err != nil { + return p.cleanupWith(ctx, slot, fmt.Errorf("error resetting v2 slot internet access: %w", err)) + } + + reused, cause := p.tryReuse(ctx, slot) + if reused { + return nil + } + + // cause is nil when the pool is simply full. + return p.cleanupWith(ctx, slot, cause) +} + +// tryReuse attempts to push the slot into the reused pool. The RLock pairs +// with Close's Lock so a send can never race the drain. +func (p *V2Pool) tryReuse(ctx context.Context, slot *network.Slot) (reused bool, cause error) { + p.closeMu.RLock() + defer p.closeMu.RUnlock() + + if p.closed { + return false, network.ErrClosed + } + + select { + case <-ctx.Done(): + return false, ctx.Err() + case <-p.done: + return false, network.ErrClosed + case p.reusedSlots <- slot: + returnedSlotCounter.Add(ctx, 1) + reusableSlotsAvailableCounter.Add(ctx, 1) + + return true, nil + default: + return false, nil + } +} + +// cleanupWith tears the slot down and attaches any cleanup error to cause. +func (p *V2Pool) cleanupWith(ctx context.Context, slot *network.Slot, cause error) error { + if cerr := p.cleanup(ctx, slot); cerr != nil { + return errors.Join(cause, fmt.Errorf("cleanup v2 slot '%d': %w", slot.Idx, cerr)) + } + + return cause +} + +func (p *V2Pool) cleanup(ctx context.Context, slot *network.Slot) error { + var errs []error + + slotV2, ok := p.registry.Load(slot.Idx) + if ok { + if err := RemoveNetworkV2(ctx, slot, slotV2, p.hostFw, p.observer); err != nil { + errs = append(errs, fmt.Errorf("cannot remove v2 network for slot '%d': %w", slot.Idx, err)) + } + p.registry.Delete(slot.Idx) + } else { + // Fallback: try basic cleanup even without v2 metadata + if err := slot.RemoveNetwork(); err != nil { + errs = append(errs, fmt.Errorf("cannot remove network for slot '%d': %w", slot.Idx, err)) + } + } + + if err := p.storage.Release(slot); err != nil { + errs = append(errs, fmt.Errorf("failed to release slot '%d': %w", slot.Idx, err)) + } + + releasedSlotCounter.Add(ctx, 1) + + return errors.Join(errs...) +} + +func (p *V2Pool) Close(ctx context.Context) error { + logger.L().Info(ctx, "Closing v2 network pool") + + p.doneOnce.Do(func() { + close(p.done) + }) + + p.closeMu.Lock() + p.closed = true + p.closeMu.Unlock() + + // Wait for in-flight asynchronous returns: each either cleans its slot + // up itself or has already pushed it into reusedSlots, drained below. + p.returnsWG.Wait() + + var errs []error + + for slot := range p.newSlots { + newSlotsAvailableCounter.Add(ctx, -1) + + if err := p.cleanup(ctx, slot); err != nil { + errs = append(errs, fmt.Errorf("failed to cleanup v2 slot '%d': %w", slot.Idx, err)) + } + } + +drain: + for { + select { + case slot := <-p.reusedSlots: + reusableSlotsAvailableCounter.Add(ctx, -1) + + if err := p.cleanup(ctx, slot); err != nil { + errs = append(errs, fmt.Errorf("failed to cleanup v2 slot '%d': %w", slot.Idx, err)) + } + default: + break drain + } + } + + if err := p.hostFw.Close(); err != nil { + errs = append(errs, fmt.Errorf("failed to close host firewall: %w", err)) + } + + if err := p.observer.Close(); err != nil { + errs = append(errs, fmt.Errorf("failed to close observer: %w", err)) + } + + return errors.Join(errs...) +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/rules_test.go b/packages/orchestrator/pkg/sandbox/network/v2/rules_test.go new file mode 100644 index 0000000000..5d7d53eb1b --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/rules_test.go @@ -0,0 +1,197 @@ +//go:build linux + +package v2 + +import ( + "bytes" + "encoding/binary" + "net" + "testing" + + "github.com/google/nftables" + "github.com/google/nftables/expr" + "github.com/stretchr/testify/require" +) + +// Helpers for asserting on installed rules through the same netlink view the +// kernel exposes, so a test fails when a rule's target, port or address is +// wrong — not merely when the chain is missing. + +func chainRules(t *testing.T, conn *nftables.Conn, table *nftables.Table, chain string) []*nftables.Rule { + t.Helper() + + rules, err := conn.GetRules(table, &nftables.Chain{Name: chain, Table: table}) + require.NoError(t, err) + + return rules +} + +// natRule reports whether the rule NATs to wantIP with the given NAT type and +// matches wantMatch in the IPv4 header. +func natRule(r *nftables.Rule, natType expr.NATType, wantIP, wantMatch net.IP) bool { + var toIP, matched bool + for _, e := range r.Exprs { + switch v := e.(type) { + case *expr.NAT: + if v.Type != natType { + return false + } + case *expr.Immediate: + toIP = toIP || bytes.Equal(v.Data, wantIP.To4()) + case *expr.Cmp: + matched = matched || bytes.Equal(v.Data, wantMatch.To4()) + } + } + + return toIP && matched +} + +// redirectPorts returns the matched destination port and the port the rule +// redirects to, for a rule that does both. +func redirectPorts(r *nftables.Rule) (dport, rport uint16, ok bool) { + var redirects bool + var ports []uint16 + + for _, e := range r.Exprs { + switch v := e.(type) { + case *expr.Redir: + redirects = true + case *expr.Cmp: + if len(v.Data) == 2 { + ports = append(ports, binary.BigEndian.Uint16(v.Data)) + } + case *expr.Immediate: + if len(v.Data) == 2 { + rport = binary.BigEndian.Uint16(v.Data) + } + } + } + + if !redirects || len(ports) == 0 { + return 0, 0, false + } + + return ports[len(ports)-1], rport, true +} + +// immediatePort returns the port a rule loads into a register, i.e. the port +// it redirects to. +func immediatePort(r *nftables.Rule) uint16 { + for _, e := range r.Exprs { + if v, ok := e.(*expr.Immediate); ok && len(v.Data) == 2 { + return binary.BigEndian.Uint16(v.Data) + } + } + + return 0 +} + +func hasExpr[T expr.Any](r *nftables.Rule) bool { + for _, e := range r.Exprs { + if _, ok := e.(T); ok { + return true + } + } + + return false +} + +// The in-namespace NAT is the whole point of SetupNamespaceNAT: pin the +// addresses and the direction, not just the chains. +func TestNamespaceNAT_RulesSNATAndDNAT(t *testing.T) { //nolint:paralleltest // creates the nftables table "test-nat-rules" in the host netns + skipIfNotLinuxRoot(t) + + const ( + hostIP = "10.11.0.1" + nsIP = "169.254.0.21" + ) + + conn, err := nftables.New(nftables.AsLasting()) + require.NoError(t, err) + defer conn.CloseLasting() + + table := conn.AddTable(&nftables.Table{Name: "test-nat-rules", Family: nftables.TableFamilyINet}) + require.NoError(t, conn.Flush()) + t.Cleanup(func() { + conn.DelTable(table) + _ = conn.Flush() + }) + + require.NoError(t, SetupNamespaceNAT(conn, table, "eth0", hostIP, nsIP)) + + post := chainRules(t, conn, table, "postroute_nat") + require.Len(t, post, 1) + require.True(t, natRule(post[0], expr.NATTypeSourceNAT, net.ParseIP(hostIP), net.ParseIP(nsIP)), + "postrouting must SNAT the namespace IP to the host IP") + + pre := chainRules(t, conn, table, "preroute_nat") + require.Len(t, pre, 1) + require.True(t, natRule(pre[0], expr.NATTypeDestNAT, net.ParseIP(nsIP), net.ParseIP(hostIP)), + "prerouting must DNAT the host IP to the namespace IP") +} + +// The host chains are v2's replacement for v1's per-slot iptables rules: pin +// every redirect's port pair and the masquerade. +func TestHostFirewall_ChainRulesMatchConfig(t *testing.T) { //nolint:paralleltest // creates/deletes the singleton nftables table "v2-host-firewall" shared by all HostFirewall tests + skipIfNotLinuxRoot(t) + + config := testConfig() + hf := newTestHostFirewall(t, config) + + got := make(map[uint16][]uint16) + var catchAll []uint16 + + for _, r := range chainRules(t, hf.conn, hf.table, "prerouting") { + dport, rport, ok := redirectPorts(r) + if !ok { + // The catch-all matches no port: every remaining TCP is redirected. + require.True(t, hasExpr[*expr.Redir](r), "every prerouting rule redirects") + catchAll = append(catchAll, immediatePort(r)) + + continue + } + got[dport] = append(got[dport], rport) + } + + require.Equal(t, []uint16{config.SandboxTCPFirewallOtherPort}, catchAll, + "exactly one catch-all TCP redirect, to the other-port") + + // Port 80 carries two rules: the hyperloop service redirect (matched on the + // orchestrator IP) first, then the TCP firewall. + require.Equal(t, []uint16{config.HyperloopProxyPort, config.SandboxTCPFirewallHTTPPort}, got[80]) + require.Equal(t, []uint16{config.PortmapperPort}, got[111]) + require.Equal(t, []uint16{config.NFSProxyPort}, got[2049]) + require.Equal(t, []uint16{config.SandboxTCPFirewallTLSPort}, got[443]) + + post := chainRules(t, hf.conn, hf.table, "postrouting") + require.Len(t, post, 1) + require.True(t, hasExpr[*expr.Masq](post[0]), "sandbox egress must be masqueraded to the gateway") +} + +// One absent element must not stop the other from being removed: an nftables +// batch is atomic, so deleting both in one flush would abort and silently +// leave a stale veth matching the forward and redirect rules. +func TestHostFirewall_RemoveSlotWithOneElementAlreadyGone(t *testing.T) { //nolint:paralleltest // creates/deletes the singleton nftables table "v2-host-firewall" shared by all HostFirewall tests + skipIfNotLinuxRoot(t) + + ctx := t.Context() + hf := newTestHostFirewall(t, testConfig()) + + slot := makeTestSlot(t, reserveNSTestIdx(t)) + sv2 := NewSlotV2(slot) + require.NoError(t, hf.AddSlot(ctx, sv2)) + + // Drop the CIDR entry behind RemoveSlot's back, as a partially reconciled + // or half-torn-down host would have. + hostIP := slot.HostIP.To4() + hf.conn.SetDeleteElements(hf.cidrSet, []nftables.SetElement{ + {Key: hostIP}, + {Key: incrementIP(hostIP), IntervalEnd: true}, + }) + require.NoError(t, hf.conn.Flush()) + + require.NoError(t, hf.RemoveSlot(ctx, sv2)) + + require.Error(t, vethSetHas(hf, slot.VethName()), + "the veth element must be gone even though the CIDR element already was") +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/slot.go b/packages/orchestrator/pkg/sandbox/network/v2/slot.go new file mode 100644 index 0000000000..a31f4bbec7 --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/slot.go @@ -0,0 +1,76 @@ +//go:build linux + +package v2 + +import ( + "fmt" + "sync" + + "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" +) + +// SlotV2 extends a base network.Slot with v2-specific metadata. +// It wraps (composition) rather than embeds so the base *Slot flows +// through the existing sandbox pipeline unchanged. +type SlotV2 struct { + Slot *network.Slot + + NetworkVersion int + SandboxID string + ExecutionID string +} + +// NewSlotV2 wraps a base Slot with v2 metadata. +func NewSlotV2(slot *network.Slot) *SlotV2 { + return &SlotV2{ + Slot: slot, + NetworkVersion: 2, + } +} + +// SlotV2Registry tracks v2 metadata for slots by their index. +type SlotV2Registry struct { + mu sync.RWMutex + slots map[int]*SlotV2 +} + +func NewSlotV2Registry() *SlotV2Registry { + return &SlotV2Registry{ + slots: make(map[int]*SlotV2), + } +} + +func (r *SlotV2Registry) Store(slotV2 *SlotV2) { + r.mu.Lock() + defer r.mu.Unlock() + r.slots[slotV2.Slot.Idx] = slotV2 +} + +func (r *SlotV2Registry) Load(idx int) (*SlotV2, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + s, ok := r.slots[idx] + + return s, ok +} + +func (r *SlotV2Registry) Delete(idx int) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.slots, idx) +} + +func (r *SlotV2Registry) Range(fn func(idx int, s *SlotV2) bool) { + r.mu.RLock() + defer r.mu.RUnlock() + for idx, s := range r.slots { + if !fn(idx, s) { + return + } + } +} + +// String returns a debug-friendly representation. +func (s *SlotV2) String() string { + return fmt.Sprintf("SlotV2{idx=%d}", s.Slot.Idx) +} diff --git a/packages/orchestrator/pkg/sandbox/network/v2/slot_test.go b/packages/orchestrator/pkg/sandbox/network/v2/slot_test.go new file mode 100644 index 0000000000..4b41876fec --- /dev/null +++ b/packages/orchestrator/pkg/sandbox/network/v2/slot_test.go @@ -0,0 +1,127 @@ +//go:build linux + +package v2 + +import ( + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/network" +) + +// Slot indices for tests that create real netns/veth state. The v1 package +// reserves 30000+ for the same purpose and `go test ./...` runs the two +// packages concurrently, so the ranges must not overlap. +var nsTestIdx atomic.Int32 + +func reserveNSTestIdx(t *testing.T) int { + t.Helper() + + return 31000 + int(nsTestIdx.Add(1)) +} + +// newTestHostFirewall opens the singleton host firewall and registers its +// close first, so it runs after every later-registered slot teardown (t.Cleanup +// is LIFO). Closing it while a slot is still registered deliberately preserves +// the table, which would leak that slot's elements into the next test. +func newTestHostFirewall(t *testing.T, config network.Config) *HostFirewall { + t.Helper() + + hf, err := NewHostFirewall("lo", config) + require.NoError(t, err) + t.Cleanup(func() { _ = hf.Close() }) + + return hf +} + +func makeTestSlot(t *testing.T, idx int) *network.Slot { + t.Helper() + cfg := network.Config{ + OrchestratorInSandboxIPAddress: "192.0.2.1", + HyperloopProxyPort: 5010, + NFSProxyPort: 5011, + PortmapperPort: 5012, + SandboxTCPFirewallHTTPPort: 5016, + SandboxTCPFirewallTLSPort: 5017, + SandboxTCPFirewallOtherPort: 5018, + } + slot, err := network.NewSlot("test-key", idx, cfg, network.NewNoopEgressProxy()) + require.NoError(t, err) + + return slot +} + +func TestSlotV2_Creation(t *testing.T) { + t.Parallel() + + slot := makeTestSlot(t, 1) + sv2 := NewSlotV2(slot) + + assert.Equal(t, 2, sv2.NetworkVersion) + assert.Equal(t, slot, sv2.Slot) + assert.Empty(t, sv2.SandboxID) +} + +func TestSlotV2_String(t *testing.T) { + t.Parallel() + + slot := makeTestSlot(t, 5) + sv2 := NewSlotV2(slot) + + assert.Contains(t, sv2.String(), "idx=5") +} + +func TestSlotV2Registry(t *testing.T) { + t.Parallel() + + reg := NewSlotV2Registry() + + slot1 := makeTestSlot(t, 1) + sv2_1 := NewSlotV2(slot1) + sv2_1.SandboxID = "sb-1" + + slot2 := makeTestSlot(t, 2) + sv2_2 := NewSlotV2(slot2) + sv2_2.SandboxID = "sb-2" + + // Store + reg.Store(sv2_1) + reg.Store(sv2_2) + + // Load + loaded, ok := reg.Load(1) + assert.True(t, ok) + assert.Equal(t, "sb-1", loaded.SandboxID) + + _, ok = reg.Load(99) + assert.False(t, ok) + + // Delete + reg.Delete(1) + _, ok = reg.Load(1) + assert.False(t, ok) + + // Range + count := 0 + reg.Range(func(_ int, _ *SlotV2) bool { + count++ + + return true + }) + assert.Equal(t, 1, count) +} + +func TestSlotV2_EmbeddedSlotMethods(t *testing.T) { + t.Parallel() + + slot := makeTestSlot(t, 3) + sv2 := NewSlotV2(slot) + + // Access base slot methods through composition + assert.Equal(t, "veth-3", sv2.Slot.VethName()) + assert.Equal(t, "ns-3", sv2.Slot.NamespaceID()) + assert.NotNil(t, sv2.Slot.HostIP) +} diff --git a/packages/orchestrator/pkg/sandbox/sandbox.go b/packages/orchestrator/pkg/sandbox/sandbox.go index e45333e293..f15eb34c21 100644 --- a/packages/orchestrator/pkg/sandbox/sandbox.go +++ b/packages/orchestrator/pkg/sandbox/sandbox.go @@ -455,7 +455,7 @@ func (m *Metadata) ExecutionDuration() (time.Duration, bool) { type Factory struct { Sandboxes *Map config cfg.BuilderConfig - networkPool *network.Pool + networkPool network.PoolInterface devicePool *nbd.DevicePool featureFlags *featureflags.Client hostStatsDelivery hoststats.Delivery @@ -466,7 +466,7 @@ type Factory struct { func NewFactory( config cfg.BuilderConfig, - networkPool *network.Pool, + networkPool network.PoolInterface, devicePool *nbd.DevicePool, featureFlags *featureflags.Client, hostStatsDelivery hoststats.Delivery, @@ -2318,7 +2318,7 @@ func createCgroup(ctx context.Context, cgroupManager cgroup.Manager, cgroupName func getNetworkSlot( ctx context.Context, - networkPool *network.Pool, + networkPool network.PoolInterface, cleanup *Cleanup, networkConfig *orchestrator.SandboxNetworkConfig, networkReleased network.ReleaseNotify, diff --git a/packages/orchestrator/pkg/server/main.go b/packages/orchestrator/pkg/server/main.go index c8f4f37584..8b71a4d158 100644 --- a/packages/orchestrator/pkg/server/main.go +++ b/packages/orchestrator/pkg/server/main.go @@ -68,7 +68,7 @@ type Server struct { sandboxFactory *sandbox.Factory info *service.ServiceInfo proxy *proxy.SandboxProxy - networkPool *network.Pool + networkPool network.PoolInterface templateCache *template.Cache devicePool *nbd.DevicePool persistence storage.StorageProvider @@ -101,7 +101,7 @@ type Server struct { type ServiceConfig struct { Config cfg.Config Tel *telemetry.Client - NetworkPool *network.Pool + NetworkPool network.PoolInterface DevicePool *nbd.DevicePool TemplateCache *template.Cache Info *service.ServiceInfo