diff --git a/packages/envd/internal/port/forward.go b/packages/envd/internal/port/forward.go index 8c882717f4..ffbed9cfc7 100644 --- a/packages/envd/internal/port/forward.go +++ b/packages/envd/internal/port/forward.go @@ -11,6 +11,7 @@ import ( "fmt" "net" "os/exec" + "strings" "sync" "syscall" @@ -75,8 +76,11 @@ func NewForwarder( scannerSub := scanner.AddSubscriber( "port-forwarder", // We only want to forward ports that are actively listening on localhost. + // "::" is included for IPv6 wildcard sockets (:::PORT): on a dual-stack + // kernel many frameworks bind "::" rather than "0.0.0.0", so gopsutil + // reports Laddr.IP = "::" for those connections. &ScannerFilter{ - IPs: []string{"127.0.0.1", "localhost", "::1"}, + IPs: []string{"127.0.0.1", "localhost", "::1", "::"}, State: "LISTEN", }, ) @@ -124,7 +128,10 @@ func (f *Forwarder) StartForwarding(ctx context.Context) { // Let's refresh our map of currently forwarded ports and mark the currently opened ones with the "FORWARD" state. // This will make sure we won't delete them later. for _, p := range procs { - key := fmt.Sprintf("%d-%d", p.Pid, p.Laddr.Port) + // Include IP in the key so that a service listening on both + // 127.0.0.1:PORT and ::1:PORT gets two independent socats instead + // of the second entry silently overwriting the first. + key := fmt.Sprintf("%d-%d-%s", p.Pid, p.Laddr.Port, p.Laddr.IP) // We check if the opened port is in our map of forwarded ports. val, portOk := f.ports[key] @@ -133,10 +140,18 @@ func (f *Forwarder) StartForwarding(ctx context.Context) { // The actual socat process that handles forwarding should be running from the last iteration. val.state = PortStateForward } else { + // A "::" wildcard socket accepts both IPv4 and IPv6; connect + // via IPv4 so socat resolves "127.0.0.1" without relying on + // /etc/hosts containing "::1 localhost" (missing on minimal images). + family := familyToIPVersion(p.Family) + if p.Laddr.IP == "::" { + family = 4 + } + f.logger.Debug(). Str("ip", p.Laddr.IP). Uint32("port", p.Laddr.Port). - Uint32("family", familyToIPVersion(p.Family)). + Uint32("family", family). Str("state", p.Status). Msg("Detected new opened port on localhost that is not forwarded") @@ -145,7 +160,7 @@ func (f *Forwarder) StartForwarding(ctx context.Context) { pid: p.Pid, port: p.Laddr.Port, state: PortStateForward, - family: familyToIPVersion(p.Family), + family: family, } f.ports[key] = ptf f.startPortForwarding(ctx, ptf) @@ -167,12 +182,20 @@ func (f *Forwarder) StartForwarding(ctx context.Context) { func (f *Forwarder) startPortForwarding(ctx context.Context, p *PortToForward) { // https://unix.stackexchange.com/questions/311492/redirect-application-listening-on-localhost-to-listening-on-external-interface - // socat -d -d TCP4-LISTEN:4000,bind=169.254.0.21,fork TCP4:localhost:4000 + // socat -d -d TCP4-LISTEN:4000,bind=169.254.0.21,fork TCP4:127.0.0.1:4000 // reuseaddr is used to fix the "Address already in use" error when restarting socat quickly. + // + // Use literal addresses rather than "localhost" so that socat's name + // resolution does not depend on /etc/hosts containing "::1 localhost", + // which is absent on Alpine and many minimal base images. + backendAddr := "127.0.0.1" + if p.family == 6 { + backendAddr = "[::1]" + } cmd := exec.CommandContext(ctx, "socat", "-d", "-d", "-d", fmt.Sprintf("TCP4-LISTEN:%v,bind=%s,reuseaddr,fork", p.port, f.sourceIP.To4()), - fmt.Sprintf("TCP%d:localhost:%v", p.family, p.port), + fmt.Sprintf("TCP%d:%s:%v", p.family, backendAddr, p.port), ) cgroupFD, ok := f.cgroupManager.GetFileDescriptor(cgroups.ProcessTypeSocat) @@ -311,7 +334,7 @@ func (f *Forwarder) ImportForwards(forwards []*upgrade.ForwardedPort) (readopted if syscall.Kill(pid, 0) != nil { continue } - f.ports[fp.GetKey()] = &PortToForward{ + f.ports[normalizeForwardKey(fp)] = &PortToForward{ pid: fp.GetListenerPid(), port: fp.GetPort(), family: fp.GetFamily(), @@ -325,6 +348,22 @@ func (f *Forwarder) ImportForwards(forwards []*upgrade.ForwardedPort) (readopted return readopted } +// normalizeForwardKey returns the canonical map key for a re-adopted port. +// Old envd binaries exported Key as "-" (one dash); new binaries use +// "--" (two dashes). When the old format is detected the IP is +// inferred from Family so the seeded key matches what the scan loop will produce. +func normalizeForwardKey(fp *upgrade.ForwardedPort) string { + key := fp.GetKey() + if strings.Count(key, "-") == 1 { + ip := "127.0.0.1" + if fp.GetFamily() == 6 { + ip = "::1" + } + key += "-" + ip + } + return key +} + func familyToIPVersion(family uint32) uint32 { switch family { case syscall.AF_INET: diff --git a/packages/envd/internal/port/forward_handover_test.go b/packages/envd/internal/port/forward_handover_test.go index 05d0fcf12a..e933e7bcd1 100644 --- a/packages/envd/internal/port/forward_handover_test.go +++ b/packages/envd/internal/port/forward_handover_test.go @@ -28,12 +28,12 @@ func TestForwarder_ExportForwards(t *testing.T) { t.Parallel() f := newHandoverTestForwarder() - f.ports["100-8080"] = &PortToForward{pid: 100, port: 8080, family: 4, socatPid: 555, state: PortStateForward} - f.ports["101-9090"] = &PortToForward{pid: 101, port: 9090, family: 4, state: PortStateForward} // no socat + f.ports["100-8080-127.0.0.1"] = &PortToForward{pid: 100, port: 8080, family: 4, socatPid: 555, state: PortStateForward} + f.ports["101-9090-127.0.0.1"] = &PortToForward{pid: 101, port: 9090, family: 4, state: PortStateForward} // no socat out := f.ExportForwards() require.Len(t, out, 1) - assert.Equal(t, "100-8080", out[0].GetKey()) + assert.Equal(t, "100-8080-127.0.0.1", out[0].GetKey()) assert.Equal(t, uint32(8080), out[0].GetPort()) assert.Equal(t, int32(100), out[0].GetListenerPid()) assert.Equal(t, uint32(4), out[0].GetFamily()) @@ -49,7 +49,7 @@ func TestForwarder_ImportForwards_SkipsDeadSocat(t *testing.T) { f := newHandoverTestForwarder() // A very high pid that is overwhelmingly unlikely to exist: kill -0 fails. n := f.ImportForwards([]*upgrade.ForwardedPort{ - {Key: "1-1", Port: 1, ListenerPid: 1, Family: 4, SocatPid: 2147483646}, + {Key: "1-1-127.0.0.1", Port: 1, ListenerPid: 1, Family: 4, SocatPid: 2147483646}, }) assert.Zero(t, n) assert.Empty(t, f.ports) @@ -72,12 +72,39 @@ func TestForwarder_ImportForwards_ReadoptsLiveSocat(t *testing.T) { f := newHandoverTestForwarder() n := f.ImportForwards([]*upgrade.ForwardedPort{ - {Key: "100-8080", Port: 8080, ListenerPid: 100, Family: 4, SocatPid: int32(pid)}, + {Key: "100-8080-127.0.0.1", Port: 8080, ListenerPid: 100, Family: 4, SocatPid: int32(pid)}, }) require.Equal(t, 1, n) - p, ok := f.ports["100-8080"] + p, ok := f.ports["100-8080-127.0.0.1"] require.True(t, ok, "a live socat must be re-adopted into the ports map") assert.Equal(t, pid, p.socatPID()) assert.Nil(t, p.socat, "a re-adopted socat has no *exec.Cmd") } + +// TestForwarder_ImportForwards_NormalizesOldKey re-adopts a socat exported by an +// old envd binary whose key was "-" (no IP suffix), translating it to +// "--" so the next scan doesn't spawn a duplicate socat. +func TestForwarder_ImportForwards_NormalizesOldKey(t *testing.T) { + t.Parallel() + + cmd := exec.CommandContext(t.Context(), "sleep", "30") + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + t.Cleanup(func() { _ = syscall.Kill(-pid, syscall.SIGKILL) }) + + f := newHandoverTestForwarder() + // Old-format key (one dash): family 4 → must normalise to "100-8080-127.0.0.1" + n := f.ImportForwards([]*upgrade.ForwardedPort{ + {Key: "100-8080", Port: 8080, ListenerPid: 100, Family: 4, SocatPid: int32(pid)}, + }) + require.Equal(t, 1, n) + + _, oldOk := f.ports["100-8080"] + assert.False(t, oldOk, "old-format key must not remain in the map") + + p, newOk := f.ports["100-8080-127.0.0.1"] + require.True(t, newOk, "old-format key must be normalised to new format") + assert.Equal(t, pid, p.socatPID()) +} diff --git a/packages/envd/internal/port/forward_test.go b/packages/envd/internal/port/forward_test.go index 9f0f12c845..88c392afa0 100644 --- a/packages/envd/internal/port/forward_test.go +++ b/packages/envd/internal/port/forward_test.go @@ -2,12 +2,33 @@ package port import ( "context" + "syscall" "testing" "time" + gopsnet "github.com/shirou/gopsutil/v4/net" + "github.com/rs/zerolog" + + "github.com/e2b-dev/infra/packages/envd/internal/services/cgroups" ) +// newTestForwarder creates a Forwarder suitable for unit tests. It uses a +// NoopManager so startPortForwarding can be called without panicking (socat +// itself may not be installed and will simply fail to start, which is fine — +// the ports map entry is inserted before the exec attempt). +func newTestForwarder(scanner *Scanner) *Forwarder { + l := zerolog.Nop() + + return &Forwarder{ + logger: &l, + ports: make(map[string]*PortToForward), + sourceIP: defaultGatewayIP, + scannerSubscriber: scanner.AddSubscriber("test", nil), + cgroupManager: cgroups.NewNoopManager(), + } +} + // TestStartForwarding_StopsOnClosedMessages pins the defensive guard on the // scan-result receive. Nothing closes Messages today, but a one-value receive // would degrade badly if that ever changed: a closed channel is permanently @@ -16,14 +37,8 @@ import ( func TestStartForwarding_StopsOnClosedMessages(t *testing.T) { t.Parallel() - l := zerolog.Nop() scanner := NewScanner(time.Hour) - f := &Forwarder{ - logger: &l, - ports: make(map[string]*PortToForward), - sourceIP: defaultGatewayIP, - scannerSubscriber: scanner.AddSubscriber("test", nil), - } + f := newTestForwarder(scanner) returned := make(chan struct{}) go func() { @@ -41,3 +56,92 @@ func TestStartForwarding_StopsOnClosedMessages(t *testing.T) { t.Fatal("StartForwarding did not stop after Messages was closed") } } + +// TestStartForwarding_WildcardIPv6_NormalizedToFamilyFour verifies that a "::" +// wildcard listener is assigned family=4 so socat connects via 127.0.0.1 (Fix A +// normalization + Fix C). On a dual-stack kernel, frameworks like gRPC bind "::" +// by default; routing them through IPv4 avoids /etc/hosts resolution of "::1 +// localhost" on minimal images. +func TestStartForwarding_WildcardIPv6_NormalizedToFamilyFour(t *testing.T) { + t.Parallel() + + scanner := NewScanner(time.Hour) + f := newTestForwarder(scanner) + + returned := make(chan struct{}) + go func() { + defer close(returned) + f.StartForwarding(context.Background()) + }() + + f.scannerSubscriber.Messages <- []gopsnet.ConnectionStat{ + {Pid: 42, Family: syscall.AF_INET6, Status: "LISTEN", + Laddr: gopsnet.Addr{IP: "::", Port: 8080}}, + } + close(f.scannerSubscriber.Messages) + + select { + case <-returned: + case <-time.After(time.Second): + t.Fatal("StartForwarding did not stop") + } + + // The goroutine has exited so f.ports is safe to read without the lock. + key := "42-8080-::" + ptf, ok := f.ports[key] + if !ok { + t.Fatalf("expected ports[%q] but got keys %v", key, portKeys(f.ports)) + } + if ptf.family != 4 { + t.Errorf("family = %d, want 4 (wildcard :: must be normalized to IPv4)", ptf.family) + } +} + +// TestStartForwarding_DualStackKey_TwoEntries verifies that a service listening +// on both 127.0.0.1:PORT and ::1:PORT receives two independent port-forward +// entries (Fix B). Before the fix the key omitted the IP, so the second entry +// overwrote the first and only one socat was started. +func TestStartForwarding_DualStackKey_TwoEntries(t *testing.T) { + t.Parallel() + + scanner := NewScanner(time.Hour) + f := newTestForwarder(scanner) + + returned := make(chan struct{}) + go func() { + defer close(returned) + f.StartForwarding(context.Background()) + }() + + f.scannerSubscriber.Messages <- []gopsnet.ConnectionStat{ + {Pid: 100, Family: syscall.AF_INET, Status: "LISTEN", + Laddr: gopsnet.Addr{IP: "127.0.0.1", Port: 9090}}, + {Pid: 100, Family: syscall.AF_INET6, Status: "LISTEN", + Laddr: gopsnet.Addr{IP: "::1", Port: 9090}}, + } + close(f.scannerSubscriber.Messages) + + select { + case <-returned: + case <-time.After(time.Second): + t.Fatal("StartForwarding did not stop") + } + + wantKeys := []string{"100-9090-127.0.0.1", "100-9090-::1"} + for _, k := range wantKeys { + if _, ok := f.ports[k]; !ok { + t.Errorf("expected ports[%q] but got keys %v", k, portKeys(f.ports)) + } + } + if len(f.ports) != 2 { + t.Errorf("len(ports) = %d, want 2; keys: %v", len(f.ports), portKeys(f.ports)) + } +} + +func portKeys(m map[string]*PortToForward) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + return keys +} diff --git a/packages/envd/internal/port/scanfilter_test.go b/packages/envd/internal/port/scanfilter_test.go new file mode 100644 index 0000000000..3c375f0045 --- /dev/null +++ b/packages/envd/internal/port/scanfilter_test.go @@ -0,0 +1,61 @@ +package port + +import ( + "testing" + + "github.com/shirou/gopsutil/v4/net" + "github.com/stretchr/testify/assert" +) + +func TestScannerFilter_Match(t *testing.T) { + t.Parallel() + + filter := &ScannerFilter{ + IPs: []string{"127.0.0.1", "localhost", "::1", "::"}, + State: "LISTEN", + } + + tests := []struct { + name string + conn net.ConnectionStat + want bool + }{ + { + name: "IPv4 loopback matches", + conn: net.ConnectionStat{Laddr: net.Addr{IP: "127.0.0.1"}, Status: "LISTEN"}, + want: true, + }, + { + name: "IPv6 loopback matches", + conn: net.ConnectionStat{Laddr: net.Addr{IP: "::1"}, Status: "LISTEN"}, + want: true, + }, + { + name: "IPv6 wildcard matches", + conn: net.ConnectionStat{Laddr: net.Addr{IP: "::"}, Status: "LISTEN"}, + want: true, + }, + { + name: "external IP does not match", + conn: net.ConnectionStat{Laddr: net.Addr{IP: "10.0.0.1"}, Status: "LISTEN"}, + want: false, + }, + { + name: "wrong state does not match", + conn: net.ConnectionStat{Laddr: net.Addr{IP: "127.0.0.1"}, Status: "ESTABLISHED"}, + want: false, + }, + { + name: "IPv4 wildcard (0.0.0.0) does not match", + conn: net.ConnectionStat{Laddr: net.Addr{IP: "0.0.0.0"}, Status: "LISTEN"}, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tt.want, filter.Match(&tt.conn)) + }) + } +} diff --git a/packages/envd/internal/services/spec/upgrade/handover.pb.go b/packages/envd/internal/services/spec/upgrade/handover.pb.go index 15ddb02381..2c5de68f8a 100644 --- a/packages/envd/internal/services/spec/upgrade/handover.pb.go +++ b/packages/envd/internal/services/spec/upgrade/handover.pb.go @@ -203,7 +203,7 @@ type ForwardedPort struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Key string `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"` // the forwarder map key: "-" + Key string `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"` // the forwarder map key: "--" (old binaries emit "-"; ImportForwards normalises on read) Port uint32 `protobuf:"varint,2,opt,name=port,proto3" json:"port,omitempty"` ListenerPid int32 `protobuf:"varint,3,opt,name=listener_pid,json=listenerPid,proto3" json:"listener_pid,omitempty"` // pid of the guest process listening on the port Family uint32 `protobuf:"varint,4,opt,name=family,proto3" json:"family,omitempty"` // IP version (4 or 6) diff --git a/packages/envd/pkg/version.go b/packages/envd/pkg/version.go index 5e5d110c55..bc489ba2bd 100644 --- a/packages/envd/pkg/version.go +++ b/packages/envd/pkg/version.go @@ -1,3 +1,3 @@ package pkg -const Version = "0.6.13" // x-release-please-version +const Version = "0.6.14" // x-release-please-version diff --git a/packages/envd/spec/upgrade/handover.proto b/packages/envd/spec/upgrade/handover.proto index 08ae543ac8..5844071ae0 100644 --- a/packages/envd/spec/upgrade/handover.proto +++ b/packages/envd/spec/upgrade/handover.proto @@ -47,7 +47,7 @@ message MountEntry { // carrying its pid lets the new forwarder re-adopt it (suppressing a duplicate // socat and reaping it when the port closes) instead of respawning. message ForwardedPort { - string key = 1; // the forwarder map key: "-" + string key = 1; // the forwarder map key: "--" (old binaries emit "-"; ImportForwards normalises on read) uint32 port = 2; int32 listener_pid = 3; // pid of the guest process listening on the port uint32 family = 4; // IP version (4 or 6)