DebTik/internal/aerotun/integration_test.go

package aerotun

import (
	"context"
	"encoding/json"
	"fmt"
	"net"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"syscall"
	"testing"
	"time"

	"proxmox-router/internal/bwtest"
)

// Opt in: AEROTUN_INTEGRATION=1 go test -run TestVethBandwidthIntegration -v ./internal/aerotun
// All links/routes and test processes live in disposable network namespaces.
func TestVethBandwidthIntegration(t *testing.T) {
	if os.Getenv("AEROTUN_INTEGRATION") != "1" {
		t.Skip("requires root, network namespaces, ip, nsenter and iperf3")
	}
	stage := os.Getenv("AEROTUN_TEST_STAGE")
	if stage == "peer" {
		runNamespacePeer(t)
		return
	}
	if stage == "" {
		ctx, cancel := context.WithTimeout(t.Context(), 100*time.Second)
		defer cancel()
		cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestVethBandwidthIntegration$", "-test.v")
		cmd.Env = append(os.Environ(), "AEROTUN_TEST_STAGE=harness")
		cmd.SysProcAttr = &syscall.SysProcAttr{Cloneflags: syscall.CLONE_NEWNET}
		out, err := cmd.CombinedOutput()
		if err != nil {
			t.Fatalf("isolated harness: %v\n%s", err, out)
		}
		t.Log(string(out))
		return
	}
	t.Run("default-route", testNamespaceDefaultRoute)
	versions := []string{"2", "3"}
	if selected := os.Getenv("AEROTUN_TEST_ONLY_VERSION"); selected != "" {
		versions = []string{selected}
	}
	for _, version := range versions {
		t.Run("v"+version, func(t *testing.T) {
			t.Setenv("AEROTUN_TEST_VERSION", version)
			for _, compress := range []bool{false, true} {
				t.Run(fmt.Sprintf("compression=%v", compress), func(t *testing.T) { runBandwidthPair(t, compress) })
			}
		})
	}
}
func waitForFile(t *testing.T, path string) {
	t.Helper()
	deadline := time.Now().Add(12 * time.Second)
	for time.Now().Before(deadline) {
		if _, err := os.Stat(path); err == nil {
			return
		}
		time.Sleep(25 * time.Millisecond)
	}
	t.Fatalf("timed out waiting for %s", path)
}
func runNamespacePeer(t *testing.T) {
	dir := os.Getenv("AEROTUN_TEST_DIR")
	role := os.Getenv("AEROTUN_TEST_ROLE")
	waitForFile(t, filepath.Join(dir, role+".start"))
	cfg := TunnelInstance{Name: "atest0", Mode: role, ListenPort: 51830, PSK: testPSK, MTU: 1420, LocalAddress: "10.99.0.1/32", PeerAddress: "10.99.0.2/32", Compression: os.Getenv("AEROTUN_TEST_COMPRESS") == "true"}
	if role == "client" {
		cfg.LocalAddress, cfg.PeerAddress = cfg.PeerAddress, cfg.LocalAddress
		cfg.RemoteEndpoint = "192.0.2.1:51830"
	}
	expectedMTU := 1292
	if data, err := os.ReadFile(filepath.Join(dir, role+".config")); err == nil {
		if err = json.Unmarshal(data, &cfg); err != nil {
			t.Fatal(err)
		}
		expectedMTU = 1320
	}
	r, err := NewTunnelRunner(cfg)
	if err != nil {
		t.Fatal(err)
	}
	// Exercise the container fallback even on hosts which provide native TUN.
	if _, ok := r.dev.(*vethTun); !ok {
		r.Stop()
		t.Fatal("integration requires veth fallback (/dev/net/tun absent)")
	}
	if err = r.Start(); err != nil {
		r.Stop()
		t.Fatal(err)
	}
	defer r.Stop()
	if role == "server" {
		benchmark := bwtest.NewServer(bwtest.ServerConfig{Enabled: true, ListenAddr: "10.99.0.1", Port: 15200})
		if err := benchmark.Start(); err != nil {
			t.Fatal(err)
		}
		defer benchmark.Stop()
	}
	_ = os.WriteFile(filepath.Join(dir, role+".ready"), nil, 0600)
	deadline := time.Now().Add(45 * time.Second)
	for time.Now().Before(deadline) {
		if r.GetTelemetry().Connected && r.GetTelemetry().MTU == expectedMTU {
			_ = os.WriteFile(filepath.Join(dir, role+".connected"), nil, 0600)
		}
		if _, err := os.Stat(filepath.Join(dir, "stop")); err == nil {
			stats := r.GetTelemetry()
			if stats.TxBytes == 0 || stats.RxBytes == 0 {
				t.Fatal("no bidirectional data delivered")
			}
			if cfg.Compression && stats.CompressedPackets == 0 {
				t.Fatal("compressed path was not exercised")
			}
			t.Logf("delivered tx=%d rx=%d compressed=%d replay drops=%d", stats.TxBytes, stats.RxBytes, stats.CompressedPackets, stats.ReplayDrops)
			return
		}
		time.Sleep(50 * time.Millisecond)
	}
	t.Fatal("peer helper timed out")
}
func runBandwidthPair(t *testing.T, compress bool) {
	dir := t.TempDir()
	type process struct {
		cmd *exec.Cmd
		log *os.File
	}
	if os.Getenv("AEROTUN_TEST_VERSION") == "3" {
		client, server := secureConfigs(t)
		client.Compression = compress
		server.Compression = compress
		client.ListenPort = 51830
		server.ListenPort = 51830
		for role, cfg := range map[string]TunnelInstance{"server": server, "client": client} {
			b, _ := json.Marshal(cfg)
			if err := os.WriteFile(filepath.Join(dir, role+".config"), b, 0600); err != nil {
				t.Fatal(err)
			}
		}
	}
	var peers []process
	defer func() {
		_ = os.WriteFile(filepath.Join(dir, "stop"), nil, 0600)
		for _, p := range peers {
			err := p.cmd.Wait()
			_ = p.log.Close()
			b, _ := os.ReadFile(p.log.Name())
			if err != nil {
				t.Errorf("peer: %v\n%s", err, b)
			}
		}
	}()
	for _, role := range []string{"server", "client"} {
		logFile, err := os.Create(filepath.Join(dir, role+".log"))
		if err != nil {
			t.Fatal(err)
		}
		cmd := exec.Command(os.Args[0], "-test.run=^TestVethBandwidthIntegration$", "-test.v")
		cmd.Env = append(os.Environ(), "AEROTUN_TEST_STAGE=peer", "AEROTUN_TEST_DIR="+dir, "AEROTUN_TEST_ROLE="+role, "AEROTUN_TEST_COMPRESS="+strconv.FormatBool(compress))
		cmd.SysProcAttr = &syscall.SysProcAttr{Cloneflags: syscall.CLONE_NEWNET}
		cmd.Stdout = logFile
		cmd.Stderr = logFile
		if err = cmd.Start(); err != nil {
			t.Fatal(err)
		}
		peers = append(peers, process{cmd, logFile})
	}
	run := func(args ...string) {
		t.Helper()
		out, err := exec.Command("ip", args...).CombinedOutput()
		if err != nil {
			t.Fatalf("ip %v: %v %s", args, err, out)
		}
	}
	run("link", "add", "atest-a", "type", "veth", "peer", "name", "atest-b")
	for i, p := range peers {
		name := []string{"atest-a", "atest-b"}[i]
		pid := strconv.Itoa(p.cmd.Process.Pid)
		run("link", "set", name, "netns", pid)
		for _, args := range [][]string{{"link", "set", "lo", "up"}, {"link", "set", name, "up"}, {"addr", "add", fmt.Sprintf("192.0.2.%d/30", i+1), "dev", name}} {
			out, err := exec.Command("nsenter", append([]string{"-t", pid, "-n", "ip"}, args...)...).CombinedOutput()
			if err != nil {
				t.Fatalf("namespace IP setup: %v %s", err, out)
			}
		}
	}
	// Model production: server cached path MTU 1380, client WireGuard link 1420.
	for i, mtu := range []string{"1380", "1420"} {
		out, err := exec.Command("nsenter", "-t", strconv.Itoa(peers[i].cmd.Process.Pid), "-n", "ip", "link", "set", []string{"atest-a", "atest-b"}[i], "mtu", mtu).CombinedOutput()
		if err != nil {
			t.Fatalf("set asymmetric MTU: %v %s", err, out)
		}
	}
	for _, role := range []string{"server", "client"} {
		_ = os.WriteFile(filepath.Join(dir, role+".start"), nil, 0600)
		waitForFile(t, filepath.Join(dir, role+".ready"))
	}
	for _, role := range []string{"server", "client"} {
		waitForFile(t, filepath.Join(dir, role+".connected"))
	}
	nsCmd := func(ctx context.Context, idx int, args ...string) *exec.Cmd {
		return exec.CommandContext(ctx, "nsenter", append([]string{"-t", strconv.Itoa(peers[idx].cmd.Process.Pid), "-n"}, args...)...)
	}
	for _, mode := range []string{"tcp-forward", "tcp-reverse", "udp"} {
		ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
		defer cancel()
		server := nsCmd(ctx, 0, "iperf3", "-s", "-1", "-B", "10.99.0.1")
		if err := server.Start(); err != nil {
			t.Fatal(err)
		}
		time.Sleep(150 * time.Millisecond)
		args := []string{"iperf3", "-c", "10.99.0.1", "-t", "2", "-J"}
		if compress {
			args = append(args, "--repeating-payload")
		}
		switch mode {
		case "tcp-forward":
			args = append(args, "-P", "2")
		case "tcp-reverse":
			args = append(args, "-R")
		case "udp":
			args = append(args, "-u", "-b", "20M", "-l", "1472")
		}
		out, err := nsCmd(ctx, 1, args...).Output()
		if err != nil {
			_ = server.Process.Kill()
			_ = server.Wait()
			t.Fatalf("%s: %v %s", mode, err, out)
		}
		if err = server.Wait(); err != nil {
			t.Fatal(err)
		}
		var result struct {
			Error string `json:"error"`
			End   struct {
				Received struct {
					Bytes uint64  `json:"bytes"`
					BPS   float64 `json:"bits_per_second"`
				} `json:"sum_received"`
			} `json:"end"`
		}
		if err = json.Unmarshal(out, &result); err != nil || result.Error != "" {
			t.Fatalf("iperf JSON: %v %s", err, out)
		}
		if result.End.Received.Bytes < 1<<20 {
			t.Fatalf("%s delivered too little data: %s", mode, strings.TrimSpace(string(out)))
		}
		t.Logf("%s: %.1f Mbit/s, %d received bytes", mode, result.End.Received.BPS/1e6, result.End.Received.Bytes)
	}
	if !compress {
		ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second)
		defer cancel()
		for i, name := range []string{"atest-a", "atest-b"} {
			out, err := nsCmd(ctx, i, "tc", "qdisc", "add", "dev", name, "root", "netem", "rate", "5mbit", "delay", "15ms", "limit", "64").CombinedOutput()
			if err != nil {
				t.Fatalf("constrain underlay: %v %s", err, out)
			}
		}
		cmd := nsCmd(ctx, 1, os.Args[0], "-test.run=^TestUDPOverAeroTunConstrainedLink$", "-test.v")
		cmd.Env = append(os.Environ(), "AEROTUN_BW_CLIENT=1")
		out, err := cmd.CombinedOutput()
		if err != nil {
			t.Fatalf("UDP feedback over constrained AeroTUN: %v\n%s", err, out)
		}
		t.Log(string(out))
	}
}

func testNamespaceDefaultRoute(t *testing.T) {
	run := func(args ...string) []byte {
		t.Helper()
		out, err := runIP(args...)
		if err != nil {
			t.Fatal(err)
		}
		return out
	}
	run("link", "add", "testwan", "type", "dummy")
	defer runIP("link", "del", "testwan")
	run("link", "set", "testwan", "up")
	run("addr", "add", "198.18.0.2/24", "dev", "testwan")
	run("route", "add", "default", "via", "198.18.0.1", "dev", "testwan")
	run("link", "add", "testoverlay", "type", "dummy")
	defer runIP("link", "del", "testoverlay")
	run("link", "set", "testoverlay", "up")
	r := routeManager{run: runIP}
	if err := r.install("testoverlay", net.ParseIP("203.0.113.9")); err != nil {
		t.Fatal(err)
	}
	defer r.cleanup()
	for _, check := range []struct{ target, dev string }{{"203.0.113.9", "testwan"}, {"8.8.8.8", "testoverlay"}, {"200.0.0.1", "testoverlay"}} {
		out := run("-j", "route", "get", check.target)
		var paths []struct {
			Dev string `json:"dev"`
		}
		if err := json.Unmarshal(out, &paths); err != nil || len(paths) != 1 || paths[0].Dev != check.dev {
			t.Fatalf("route to %s: %s %v", check.target, out, err)
		}
	}
	r.cleanup()
	out := run("route", "show", "default")
	if !strings.Contains(string(out), "via 198.18.0.1 dev testwan") {
		t.Fatalf("original default lost: %s", out)
	}
	if out = run("route", "show", "exact", "203.0.113.9/32"); len(out) != 0 {
		t.Fatalf("endpoint route leaked: %s", out)
	}
}

func TestUDPOverAeroTunConstrainedLink(t *testing.T) {
	if os.Getenv("AEROTUN_BW_CLIENT") != "1" {
		t.Skip("run by isolated integration harness")
	}
	for _, direction := range []string{"rx", "both"} {
		t.Run(direction, func(t *testing.T) {
			client := bwtest.NewClient()
			defer client.Stop()
			if err := client.Run(t.Context(), bwtest.SessionRequest{Target: "10.99.0.1", Port: 15200, Protocol: "udp", Direction: direction, Streams: 4, PacketSize: 1472, Duration: 4}); err != nil {
				t.Fatal(err)
			}
			deadline := time.Now().Add(6 * time.Second)
			for client.GetStatus().Running && time.Now().Before(deadline) {
				time.Sleep(20 * time.Millisecond)
			}
			st := client.GetStatus()
			if st.Running || st.Error != "" || st.Summary == nil {
				t.Fatalf("UDP benchmark failed: %+v", st)
			}
			expectedSize := 1264
			if os.Getenv("AEROTUN_TEST_VERSION") == "3" {
				expectedSize = 1292
			}
			if st.PacketSize != expectedSize {
				t.Fatalf("datagram not clamped to negotiated overlay MTU: %d", st.PacketSize)
			}
			if st.Summary.TotalRxBytes < 100000 || (direction == "both" && st.Summary.TotalTxBytes < 100000) {
				t.Fatalf("reverse transfer starved: %+v", st.Summary)
			}
			if st.Summary.AvgLossPct > 40 {
				t.Fatalf("automatic UDP flooded constrained link: %+v", st.Summary)
			}
			t.Logf("%s: confirmed TX %.2f / RX %.2f Mbit/s, loss %.1f%%, payload %d", direction, st.Summary.AvgTxMbps, st.Summary.AvgRxMbps, st.Summary.AvgLossPct, st.PacketSize)
		})
	}
}