diff --git a/CHANGELOG.md b/CHANGELOG.md index 405cbf96..90d2f22c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -90,6 +90,18 @@ changes; the linked API and deployment documentation contains operational detail databases require the explicit upgrade to schema 26. See `docs/operations/configuration.md#destination-limits-and-opt-outs`. +### Fixed +- An executor counting packets with eBPF (`enforcement_mode: ebpf`) no longer + fails every run that closes a socket (#412). Closing a counted TCP or UDP + connection, dialed or accepted, deleted its socket-storage entry by the + descriptor it had just closed, so the guest's close failed with `close_tcp: + delete: bad file descriptor` and the run ended "debuglet failed" although + its output was delivered. Close no longer deletes the entry: it belongs to + the socket, and the kernel frees it when the socket is destroyed. Deleting + it before the close instead would let the data a socket still holds after + close leave unattributed and past its rate limit; that data keeps the run's + UUID and is dropped once the run's rate entry is removed. + ## [0.3.0-rc.1] - 2026-10-06 ### Security diff --git a/internal/executor/ratelimit/ebpf/bpf_conn.go b/internal/executor/ratelimit/ebpf/bpf_conn.go index e2a50640..0d4b4da7 100644 --- a/internal/executor/ratelimit/ebpf/bpf_conn.go +++ b/internal/executor/ratelimit/ebpf/bpf_conn.go @@ -6,20 +6,23 @@ package ebpf import ( - "github.com/netsec-ethz/debuglet/internal/executor/debuglet/socket/netutil" "net" + "sync" "time" "github.com/google/uuid" + "github.com/netsec-ethz/debuglet/internal/executor/debuglet/socket/netutil" ) type BpfConn struct { count *BpfCount conn net.Conn - socketID uint32 domain string id uuid.UUID resolvedIPv6 netutil.IPv6 + + closeOnce sync.Once + closeErr error } func (bc *BpfConn) Read(b []byte) (n int, err error) { return bc.conn.Read(b) } @@ -29,16 +32,30 @@ func (bc *BpfConn) RemoteAddr() net.Addr { return bc.conn.RemoteAd func (bc *BpfConn) SetDeadline(t time.Time) error { return bc.conn.SetDeadline(t) } func (bc *BpfConn) SetReadDeadline(t time.Time) error { return bc.conn.SetReadDeadline(t) } func (bc *BpfConn) SetWriteDeadline(t time.Time) error { return bc.conn.SetWriteDeadline(t) } + +// Close closes the socket and then detaches its rate limit, once. Later calls +// return the first call's result. +// +// Close does not delete the socket's debuglet_sk_map entry. That map is an +// SK_STORAGE map: the entry belongs to the socket and the kernel frees it when +// the socket is destroyed. Until then, data the kernel still sends after +// close(2) (the unsent send buffer, the FIN, retransmissions) keeps this run's +// UUID and is dropped once Detach has removed the rate entry. Deleting the +// entry earlier would let that tail leave unattributed and unthrottled, and +// deleting it after close(2) addresses a descriptor that no longer names the +// socket (#412). func (bc *BpfConn) Close() error { + bc.closeOnce.Do(func() { bc.closeErr = bc.close() }) + return bc.closeErr +} + +func (bc *BpfConn) close() error { // Close connection before detaching/removing ratelimit, otherwise ratelimited writes will // be passed through before the connection is actually closed - err := bc.conn.Close() - err2 := bc.count.Detach(bc.domain, bc.id, bc.resolvedIPv6) - if err3 := bc.count.objs.DebugletSkMap.Delete(bc.socketID); err3 != nil { - return err3 - } - if err2 != nil { - return err2 + closeErr := bc.conn.Close() + detachErr := bc.count.Detach(bc.domain, bc.id, bc.resolvedIPv6) + if closeErr != nil { + return closeErr } - return err + return detachErr } diff --git a/internal/executor/ratelimit/ebpf/bpf_conn_test.go b/internal/executor/ratelimit/ebpf/bpf_conn_test.go new file mode 100644 index 00000000..56d680a0 --- /dev/null +++ b/internal/executor/ratelimit/ebpf/bpf_conn_test.go @@ -0,0 +1,372 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 ETH Zurich + +//go:build linux + +package ebpf + +import ( + "errors" + "io" + "net" + "net/netip" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/cilium/ebpf" + "github.com/google/uuid" + "github.com/netsec-ethz/debuglet/internal/bitrate" + "github.com/netsec-ethz/debuglet/internal/executor/debuglet/socket/netutil" +) + +// closeRecorder counts closes of the wrapped connection. +type closeRecorder struct { + net.Conn + closes int + err error +} + +func (c *closeRecorder) Close() error { + c.closes++ + if err := c.Conn.Close(); err != nil { + return err + } + return c.err +} + +// loopbackConn returns one end of an unprivileged loopback TCP or UDP pair. +func loopbackConn(t *testing.T, network string) net.Conn { + t.Helper() + switch network { + case "tcp": + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { listener.Close() }) + conn, err := net.Dial("tcp4", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + return conn + case "udp": + receiver, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { receiver.Close() }) + conn, err := net.DialUDP("udp4", nil, receiver.LocalAddr().(*net.UDPAddr)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + return conn + } + t.Fatalf("unknown network %q", network) + return nil +} + +// Regression for #412. Close must not touch debuglet_sk_map: deleting by the +// closed descriptor fails (EBADF) and deleting before close lets the socket's +// remaining data leave unattributed. The counter here has no maps, so any map +// access panics; the destination is registered twice so that a single Detach +// leaves no rate-map delete either. +func TestBpfConnCloseLeavesSocketStorageToTheKernel(t *testing.T) { + ip := netutil.ToIPv6(netip.MustParseAddr("127.0.0.1")) + for _, network := range []string{"tcp", "udp"} { + t.Run(network, func(t *testing.T) { + conn := loopbackConn(t, network) + recorder := &closeRecorder{Conn: conn} + counter := new(BpfCount) + id := uuid.New() + counter.destinations.Add("127.0.0.1", id, ip) + counter.destinations.Add("127.0.0.1", id, ip) + bc := &BpfConn{count: counter, conn: recorder, domain: "127.0.0.1", id: id, resolvedIPv6: ip} + closeOnce := func() (err error) { + defer func() { + if r := recover(); r != nil { + t.Fatalf("Close touched a counter map: %v", r) + } + }() + return bc.Close() + } + if err := closeOnce(); err != nil { + t.Fatalf("Close: %v", err) + } + if err := closeOnce(); err != nil { + t.Fatalf("second Close: %v", err) + } + if recorder.closes != 1 { + t.Fatalf("socket closed %d times, want 1", recorder.closes) + } + if _, err := conn.Write([]byte{1}); !errors.Is(err, net.ErrClosed) { + t.Fatalf("socket still open after Close: %v", err) + } + // Exactly one of the two attachments was detached: one is left. + if remaining, ok := counter.destinations.Remove("127.0.0.1", id, ip); !ok || remaining != 0 { + t.Fatalf("attachments left after Close = %d (ok %v), want one detached", remaining, ok) + } + }) + } +} + +func TestBpfConnClosePrefersCloseError(t *testing.T) { + closeFailed := errors.New("close failed") + conn := loopbackConn(t, "tcp") + recorder := &closeRecorder{Conn: conn, err: closeFailed} + bc := &BpfConn{count: new(BpfCount), conn: recorder, domain: "127.0.0.1", id: uuid.New()} + if err := bc.Close(); !errors.Is(err, closeFailed) { + t.Fatalf("Close = %v, want %v", err, closeFailed) + } + if err := bc.Close(); !errors.Is(err, closeFailed) || recorder.closes != 1 { + t.Fatalf("second Close = %v after %d closes, want the first result once", err, recorder.closes) + } +} + +// TestKernelCounterClosesCountedConns attaches the counter to loopback and +// closes counted TCP (both the dialed and the accepted end) and UDP sockets. +// Each Close must succeed and detach its rate limit, its descriptor number +// must carry no entry once reused, and data a TCP socket still holds when it +// is closed must stay attributed to its run instead of leaving unthrottled. +func TestKernelCounterClosesCountedConns(t *testing.T) { + iface, err := net.InterfaceByName("lo") + if err != nil { + t.Fatal(err) + } + counter, err := NewBPFCount(iface) + if err != nil { + for _, errno := range []syscall.Errno{syscall.EPERM, syscall.EACCES, syscall.EINVAL} { + if errors.Is(err, errno) { + t.Skipf("counter load requires kernel capabilities: %v", err) + } + } + t.Fatal(err) + } + t.Cleanup(func() { + if err := counter.Close(); err != nil { + t.Error(err) + } + }) + const addr = "127.0.0.1" + attach := func(t *testing.T, conn net.Conn, limit bitrate.Bitrate) (net.Conn, uuid.UUID) { + t.Helper() + id := uuid.New() + attached, err := counter.Attach(conn, id, addr) + if err != nil { + conn.Close() + t.Fatal(err) + } + if err := counter.SetLimit(addr, id, limit); err != nil { + t.Fatal(err) + } + if err := counter.SetExecLimit(id, 1<<30); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = counter.DeleteExecLimit(id) }) + return attached, id + } + fdOf := func(t *testing.T, conn net.Conn) uint32 { + t.Helper() + raw, err := conn.(*BpfConn).conn.(syscall.Conn).SyscallConn() + if err != nil { + t.Fatal(err) + } + var socketFD uint32 + if err := raw.Control(func(fd uintptr) { socketFD = uint32(fd) }); err != nil { + t.Fatal(err) + } + return socketFD + } + stored := func(t *testing.T, conn net.Conn) (uuid.UUID, error) { + t.Helper() + raw, err := conn.(*BpfConn).conn.(syscall.Conn).SyscallConn() + if err != nil { + t.Fatal(err) + } + var value countDebugletUuid + var lookupErr error + if err := raw.Control(func(fd uintptr) { + lookupErr = counter.objs.DebugletSkMap.Lookup(uint32(fd), &value) + }); err != nil { + t.Fatal(err) + } + return uuid.UUID(value.Uuid), lookupErr + } + exchange := func(t *testing.T, from, to net.Conn) { + t.Helper() + payload := []byte("debuglet #412") + if _, err := from.Write(payload); err != nil { + t.Fatal(err) + } + if err := to.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil { + t.Fatal(err) + } + got := make([]byte, len(payload)) + if _, err := io.ReadFull(to, got); err != nil { + t.Fatalf("counted traffic not delivered: %v", err) + } + } + // reusedFDHasNoEntry opens sockets until one gets the closed descriptor's + // number and checks that the number does not carry the old entry. + reusedFDHasNoEntry := func(t *testing.T, fd uint32) { + t.Helper() + for range 64 { + fresh, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_DGRAM|syscall.SOCK_CLOEXEC, 0) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { syscall.Close(fresh) }) + if uint32(fresh) > fd { + break + } + if uint32(fresh) == fd { + var value countDebugletUuid + if err := counter.objs.DebugletSkMap.Lookup(fd, &value); !errors.Is(err, ebpf.ErrKeyNotExist) { + t.Fatalf("socket reusing descriptor %d has entry %v: %v", fd, uuid.UUID(value.Uuid), err) + } + return + } + } + t.Fatalf("descriptor %d was not reused", fd) + } + closeCounted := func(t *testing.T, conn net.Conn, id uuid.UUID) { + t.Helper() + fd := fdOf(t, conn) + if err := conn.Close(); err != nil { + t.Fatalf("Close of counted %s conn: %v", conn.LocalAddr().Network(), err) + } + if err := conn.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } + var rate uint64 + key := countDebugletKey{Uuid: [16]byte(id), Ipv6: [16]byte{10: 0xff, 11: 0xff, 12: 127, 15: 1}} + if err := counter.objs.RatesMap.Lookup(&key, &rate); !errors.Is(err, ebpf.ErrKeyNotExist) { + t.Fatalf("rate limit left after Close: %v", err) + } + reusedFDHasNoEntry(t, fd) + } + + t.Run("tcp", func(t *testing.T) { + listener, err := net.Listen("tcp4", addr+":0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + dialed, err := net.Dial("tcp4", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + accepted, err := listener.Accept() + if err != nil { + dialed.Close() + t.Fatal(err) + } + client, clientID := attach(t, dialed, 1<<30) + server, serverID := attach(t, accepted, 1<<30) + for conn, id := range map[net.Conn]uuid.UUID{client: clientID, server: serverID} { + if got, err := stored(t, conn); err != nil || got != id { + t.Fatalf("socket storage = %v, %v; want %v", got, err, id) + } + } + exchange(t, client, server) + exchange(t, server, client) + closeCounted(t, client, clientID) + if got, err := stored(t, server); err != nil || got != serverID { + t.Fatalf("other socket's storage after Close = %v, %v; want %v", got, err, serverID) + } + closeCounted(t, server, serverID) + }) + + t.Run("udp", func(t *testing.T) { + receiver, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP(addr)}) + if err != nil { + t.Fatal(err) + } + defer receiver.Close() + dialed, err := net.DialUDP("udp4", nil, receiver.LocalAddr().(*net.UDPAddr)) + if err != nil { + t.Fatal(err) + } + sender, id := attach(t, dialed, 1<<30) + if got, err := stored(t, sender); err != nil || got != id { + t.Fatalf("socket storage = %v, %v; want %v", got, err, id) + } + exchange(t, sender, receiver) + closeCounted(t, sender, id) + }) + + // A guest that fills its send buffer under a low limit and closes must not + // get the buffered remainder delivered: it keeps the run's UUID and is + // dropped once the rate entry is gone, instead of leaving unattributed. + t.Run("write-then-close tail", func(t *testing.T) { + listener, err := net.Listen("tcp4", addr+":0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + dialed, err := net.Dial("tcp4", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + accepted, err := listener.Accept() + if err != nil { + dialed.Close() + t.Fatal(err) + } + defer accepted.Close() + if err := dialed.(*net.TCPConn).SetWriteBuffer(4 << 20); err != nil { + t.Fatal(err) + } + const limit = 16 * 1024 * 8 * bitrate.Bit // 16 KiB/s + client, _ := attach(t, dialed, limit) + + var received atomic.Int64 + done := make(chan struct{}) + go func() { + defer close(done) + buffer := make([]byte, 64<<10) + for { + n, err := accepted.Read(buffer) + received.Add(int64(n)) + if err != nil { + return + } + } + }() + if err := client.SetWriteDeadline(time.Now().Add(500 * time.Millisecond)); err != nil { + t.Fatal(err) + } + written, _ := client.Write(make([]byte, 8<<20)) + atClose := received.Load() + dropsBefore, _, err := counter.DropTotals() + if err != nil { + t.Fatal(err) + } + if err := client.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + time.Sleep(time.Second) + tail := received.Load() - atClose + dropsAfter, _, err := counter.DropTotals() + if err != nil { + t.Fatal(err) + } + _ = accepted.SetReadDeadline(time.Now()) + <-done + t.Logf("written %d, delivered %d before Close, %d after; egress drops %d -> %d", + written, atClose, tail, dropsBefore[1], dropsAfter[1]) + if unsent := int64(written) - atClose; unsent < 128<<10 { + t.Fatalf("fixture buffered only %d unsent bytes at Close; the check needs at least 128 KiB", unsent) + } + // At most one burst may follow Close; the buffered remainder may not. + if tail > 64<<10 { + t.Fatalf("%d bytes left the closed socket after Close, past its rate limit", tail) + } + if dropsAfter[1] <= dropsBefore[1] { + t.Fatalf("no egress drop after Close (%d -> %d): the tail left unattributed", dropsBefore[1], dropsAfter[1]) + } + }) +} diff --git a/internal/executor/ratelimit/ebpf/bpf_count.go b/internal/executor/ratelimit/ebpf/bpf_count.go index 3e08ff84..5766bd6d 100644 --- a/internal/executor/ratelimit/ebpf/bpf_count.go +++ b/internal/executor/ratelimit/ebpf/bpf_count.go @@ -124,12 +124,11 @@ func (bc *BpfCount) Attach(conn net.Conn, id uuid.UUID, addr string) (net.Conn, } var fdErr error - var socketID uint32 - + // The entry lives as long as the socket; the kernel frees it when the + // socket is destroyed, so BpfConn.Close does not delete it. err = rawConn.Control(func(fd uintptr) { info := countDebugletUuid{Uuid: [16]byte(id)} fdErr = bc.objs.DebugletSkMap.Update(uint32(fd), &info, ebpf.UpdateAny) - socketID = uint32(fd) }) if err != nil { return nil, fmt.Errorf("failed call control: %w", err) @@ -152,7 +151,7 @@ func (bc *BpfCount) Attach(conn net.Conn, id uuid.UUID, addr string) (net.Conn, defer bc.mu.Unlock() bc.destinations.Add(addr, id, ipv6) - return &BpfConn{count: bc, conn: conn, socketID: socketID, domain: addr, id: id, resolvedIPv6: ipv6}, nil + return &BpfConn{count: bc, conn: conn, domain: addr, id: id, resolvedIPv6: ipv6}, nil } func (bc *BpfCount) SetLimit(addr string, id uuid.UUID, limit bitrate.Bitrate) error { diff --git a/scripts/ci-kernel.sh b/scripts/ci-kernel.sh index 05d2d42e..be6e5362 100644 --- a/scripts/ci-kernel.sh +++ b/scripts/ci-kernel.sh @@ -71,6 +71,7 @@ required = { ("github.com/netsec-ethz/debuglet/internal/executor/tagger/ebpf", "TestDisclosureWaitsForEveryTagger"), ("github.com/netsec-ethz/debuglet/internal/executor/tagger", "TestTaggedDatagramsReachTheWire"), ("github.com/netsec-ethz/debuglet/internal/executor/ratelimit/ebpf", "TestBPFCounterLinuxLoad"), + ("github.com/netsec-ethz/debuglet/internal/executor/ratelimit/ebpf", "TestKernelCounterClosesCountedConns"), ("github.com/netsec-ethz/debuglet/internal/executor", "TestHeterogeneousExecutorCapabilitiesSelection"), ("github.com/netsec-ethz/debuglet/internal/executor", "TestRestartRetiresAStaleTaggerFilterBeforeTheTail"), ("github.com/netsec-ethz/debuglet/internal/executor", "TestRestartWithChangedNetworkWithholdsRetiredTail"), @@ -82,7 +83,7 @@ for package, test in sorted(missing): print(f"Missing passing {package}/{test}; kernel checks did not run.", file=sys.stderr) if skips or missing: sys.exit(1) -print("Tagger load, kernel/Go tag parity, debuglet-tag-v1 kernel vectors, legacy tc, receiver-verified TCP tagging, kernel-slot TESLA disclosure, user-space datagram tagging, packet-counter, heterogeneous capability-selection and IPv4-only tagged-run listener checks passed with zero skipped tests.") +print("Tagger load, kernel/Go tag parity, debuglet-tag-v1 kernel vectors, legacy tc, receiver-verified TCP tagging, kernel-slot TESLA disclosure, user-space datagram tagging, packet-counter, counted-socket close, heterogeneous capability-selection and IPv4-only tagged-run listener checks passed with zero skipped tests.") PY # Regenerate only after loading the committed objects. The pinned tools image