fix(envd/port): detect :::PORT wildcard listeners and fix dual-stack key collision

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 "<pid>-<port>" (one dash); new binaries use
+// "<pid>-<port>-<ip>" (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:

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 "<pid>-<port>" (no IP suffix), translating it to
+// "<pid>-<port>-<ip>" 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())
+}

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
+}

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))
+		})
+	}
+}