From 8a1a3d7ce6419912a2f44270fde78c2180636688 Mon Sep 17 00:00:00 2001 From: Yi-Min Lin Date: Wed, 7 Oct 2026 16:48:52 -0700 Subject: [PATCH 1/2] Delete eBPF socket storage before closing a counted conn BpfConn.Close closed the socket and then deleted its debuglet_sk_map entry by the descriptor number it had just closed. The map is an SK_STORAGE map whose key is a descriptor resolved at call time, so every delete failed with EBADF and every guest close in eBPF counting mode failed the run; a reused descriptor number could also have dropped another socket's entry. Delete the entry through RawConn.Control while the descriptor is open, then close the socket and detach the rate limit. A missing entry is not an error, the socket is closed even if the delete fails, the close error takes precedence, and Close runs once. Add unit tests with an injected deleter that check the order and that the descriptor still names the socket at delete time, and a kernel test, required by the kernel CI lane, that closes counted dialed and accepted TCP and UDP connections. Fixes #412 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 11 + internal/executor/ratelimit/ebpf/bpf_conn.go | 76 ++++- .../executor/ratelimit/ebpf/bpf_conn_test.go | 299 ++++++++++++++++++ internal/executor/ratelimit/ebpf/bpf_count.go | 14 +- scripts/ci-kernel.sh | 3 +- 5 files changed, 382 insertions(+), 21 deletions(-) create mode 100644 internal/executor/ratelimit/ebpf/bpf_conn_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 405cbf96..9063b734 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -90,6 +90,17 @@ 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, removed its socket-storage entry after + closing the descriptor, so the guest's close failed with `close_tcp: delete: + bad file descriptor` and the run ended "debuglet failed" although its output + was delivered. The entry is now removed while the descriptor is open, before + the socket is closed; an entry already gone is not an error, and the socket + is closed even when the removal fails. A descriptor number reused by another + socket in between can no longer lose that socket's entry. + ## [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..a134fcf9 100644 --- a/internal/executor/ratelimit/ebpf/bpf_conn.go +++ b/internal/executor/ratelimit/ebpf/bpf_conn.go @@ -6,20 +6,36 @@ package ebpf import ( - "github.com/netsec-ethz/debuglet/internal/executor/debuglet/socket/netutil" + "errors" + "fmt" "net" + "sync" + "syscall" "time" + "github.com/cilium/ebpf" "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 + count *BpfCount + conn net.Conn + // raw reaches the socket's descriptor while it is still open. Its Control + // holds a reference on the descriptor, so the number cannot be closed and + // reused by another socket during the delete, and it refuses to run once + // the connection is closed. + raw syscall.RawConn + // deleteStorage removes the socket's entry from debuglet_sk_map. The map + // is an SK_STORAGE map, whose key is a descriptor the kernel resolves to a + // socket at the time of the call. + deleteStorage func(fd uint32) error + 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 +45,44 @@ 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 releases the socket's map entry, the socket and its rate limit, in +// that order, once. Later calls return the first call's result. 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 + bc.closeOnce.Do(func() { bc.closeErr = bc.close() }) + return bc.closeErr +} + +func (bc *BpfConn) close() error { + // The socket storage is addressed by descriptor, so it must be deleted + // while the descriptor still names this socket: after Close it names + // nothing (EBADF) or, once reused, another socket. + deleteErr := bc.deleteSocketStorage() + // The socket is closed even when the delete failed; the kernel frees + // its storage with the socket. Close the connection before detaching + // the rate limit, otherwise rate-limited writes would be passed through + // before the connection is actually closed. + closeErr := bc.conn.Close() + detachErr := bc.count.Detach(bc.domain, bc.id, bc.resolvedIPv6) + if closeErr != nil { + return closeErr + } + if deleteErr != nil { + return deleteErr + } + return detachErr +} + +func (bc *BpfConn) deleteSocketStorage() error { + var deleteErr error + if err := bc.raw.Control(func(fd uintptr) { + deleteErr = bc.deleteStorage(uint32(fd)) + }); err != nil { + return fmt.Errorf("delete socket storage: %w", err) } - if err2 != nil { - return err2 + // An entry already gone needs no delete. + if deleteErr != nil && !errors.Is(deleteErr, ebpf.ErrKeyNotExist) { + return fmt.Errorf("delete socket storage: %w", deleteErr) } - return err + return nil } 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..78598294 --- /dev/null +++ b/internal/executor/ratelimit/ebpf/bpf_conn_test.go @@ -0,0 +1,299 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 ETH Zurich + +//go:build linux + +package ebpf + +import ( + "errors" + "fmt" + "io" + "net" + "syscall" + "testing" + "time" + + "github.com/cilium/ebpf" + "github.com/google/uuid" + "golang.org/x/sys/unix" +) + +// closeRecorder notes when the wrapped connection is closed. +type closeRecorder struct { + net.Conn + events *[]string + err error +} + +func (c *closeRecorder) Close() error { + *c.events = append(*c.events, "close") + 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 +} + +// testBpfConn wraps conn the way Attach does, with deleter in place of the +// kernel map and a recorder on the connection's Close. +func testBpfConn(t *testing.T, conn net.Conn, events *[]string, closeErr error, deleter func(fd uint32) error) *BpfConn { + t.Helper() + raw, err := conn.(syscall.Conn).SyscallConn() + if err != nil { + t.Fatal(err) + } + return &BpfConn{ + count: new(BpfCount), conn: &closeRecorder{Conn: conn, events: events, err: closeErr}, raw: raw, + deleteStorage: func(fd uint32) error { + *events = append(*events, "delete") + return deleter(fd) + }, + domain: "127.0.0.1", id: uuid.New(), + } +} + +// Regression for #412: the SK_STORAGE entry is keyed by the descriptor, so it +// must be deleted while the descriptor is open and still names this socket. +func TestBpfConnDeletesSocketStorageBeforeClosing(t *testing.T) { + for _, network := range []string{"tcp", "udp"} { + t.Run(network, func(t *testing.T) { + conn := loopbackConn(t, network) + local := conn.LocalAddr().String() + var events []string + var deletedFDs []uint32 + bc := testBpfConn(t, conn, &events, nil, func(fd uint32) error { + deletedFDs = append(deletedFDs, fd) + if _, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0); err != nil { + return fmt.Errorf("descriptor %d not open at delete: %w", fd, err) + } + name, err := unix.Getsockname(int(fd)) + if err != nil { + return fmt.Errorf("descriptor %d: %w", fd, err) + } + in4, ok := name.(*unix.SockaddrInet4) + if !ok || fmt.Sprintf("%s:%d", net.IP(in4.Addr[:]), in4.Port) != local { + return fmt.Errorf("descriptor %d names %v, not this socket %s", fd, name, local) + } + return nil + }) + if err := bc.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if fmt.Sprint(events) != "[delete close]" { + t.Fatalf("order = %v, want [delete close]", events) + } + // A second Close neither deletes by a descriptor number that may + // already name another socket nor closes again. + if err := bc.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } + if fmt.Sprint(events) != "[delete close]" || len(deletedFDs) != 1 { + t.Fatalf("second Close acted again: events %v, deletes %v", events, deletedFDs) + } + if _, err := conn.Write([]byte{1}); !errors.Is(err, net.ErrClosed) { + t.Fatalf("socket still open after Close: %v", err) + } + }) + } +} + +func TestBpfConnCloseErrors(t *testing.T) { + deleteFailed := errors.New("map delete failed") + closeFailed := errors.New("close failed") + for _, tc := range []struct { + name string + deleteErr, connErr error + want error + }{ + {name: "entry already gone", deleteErr: fmt.Errorf("lookup: %w", ebpf.ErrKeyNotExist)}, + {name: "delete fails", deleteErr: deleteFailed, want: deleteFailed}, + {name: "close error preferred", deleteErr: deleteFailed, connErr: closeFailed, want: closeFailed}, + } { + t.Run(tc.name, func(t *testing.T) { + conn := loopbackConn(t, "tcp") + var events []string + bc := testBpfConn(t, conn, &events, tc.connErr, func(uint32) error { return tc.deleteErr }) + err := bc.Close() + if tc.want == nil && err != nil || tc.want != nil && !errors.Is(err, tc.want) { + t.Fatalf("Close = %v, want %v", err, tc.want) + } + if fmt.Sprint(events) != "[delete close]" { + t.Fatalf("order = %v, want [delete close]", events) + } + if _, werr := conn.Write([]byte{1}); !errors.Is(werr, net.ErrClosed) { + t.Fatalf("socket left open after failed delete: %v", werr) + } + if again := bc.Close(); again != err { + t.Fatalf("second Close = %v, want first result %v", again, err) + } + }) + } +} + +// 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 remove only its own socket's entry. +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) (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, 1<<30); 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 + } + 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) + } + } + closeCounted := func(t *testing.T, conn net.Conn, id uuid.UUID) { + t.Helper() + 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) + } + } + + 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) + server, serverID := attach(t, accepted) + 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) + // Closing one socket must not remove another socket's entry. + 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) + 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) + }) +} diff --git a/internal/executor/ratelimit/ebpf/bpf_count.go b/internal/executor/ratelimit/ebpf/bpf_count.go index 3e08ff84..3f288ba4 100644 --- a/internal/executor/ratelimit/ebpf/bpf_count.go +++ b/internal/executor/ratelimit/ebpf/bpf_count.go @@ -124,12 +124,9 @@ func (bc *BpfCount) Attach(conn net.Conn, id uuid.UUID, addr string) (net.Conn, } var fdErr error - var socketID uint32 - 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 +149,16 @@ 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, raw: rawConn, deleteStorage: bc.deleteSocketStorage, + domain: addr, id: id, resolvedIPv6: ipv6, + }, nil +} + +// deleteSocketStorage removes the debuglet_sk_map entry of the socket that the +// open descriptor fd refers to. +func (bc *BpfCount) deleteSocketStorage(fd uint32) error { + return bc.objs.DebugletSkMap.Delete(fd) } 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 From 07bb6b9ad6fb7c48e3507f063d63725bdec1fe18 Mon Sep 17 00:00:00 2001 From: Yi-Min Lin Date: Wed, 7 Oct 2026 17:03:20 -0700 Subject: [PATCH 2/2] Leave eBPF socket storage to the kernel on close Review of the previous commit: deleting the debuglet_sk_map entry before close(2) lets everything the socket still sends after the close (the unsent send buffer, FIN, retransmissions) leave without the run's UUID, so it is neither rate-limited nor counted. Writing a send buffer and closing, repeated over fresh sockets, bypasses the destination policy; the kernel test measured 455 KB delivered within a second of Close under a 16 KiB/s limit. The entry is SK_STORAGE, owned by the socket and freed by the kernel when the socket is destroyed, so Close does not delete it at all. The tail keeps its UUID and is dropped once Detach removes the rate entry, and no delete addresses a closed or reused descriptor. Close still closes before detaching, runs once and returns the close error first. Tests: the unit test runs Close against a counter without maps, so any map access fails it. The kernel test checks that a reused descriptor number carries no entry, and that a write-then-close tail under a low limit is dropped as attributed instead of delivered. Fixes #412 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 15 +- internal/executor/ratelimit/ebpf/bpf_conn.go | 63 ++--- .../executor/ratelimit/ebpf/bpf_conn_test.go | 253 +++++++++++------- internal/executor/ratelimit/ebpf/bpf_count.go | 13 +- 4 files changed, 192 insertions(+), 152 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9063b734..90d2f22c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -93,13 +93,14 @@ changes; the linked API and deployment documentation contains operational detail ### 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, removed its socket-storage entry after - closing the descriptor, so the guest's close failed with `close_tcp: delete: - bad file descriptor` and the run ended "debuglet failed" although its output - was delivered. The entry is now removed while the descriptor is open, before - the socket is closed; an entry already gone is not an error, and the socket - is closed even when the removal fails. A descriptor number reused by another - socket in between can no longer lose that socket's entry. + 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 diff --git a/internal/executor/ratelimit/ebpf/bpf_conn.go b/internal/executor/ratelimit/ebpf/bpf_conn.go index a134fcf9..0d4b4da7 100644 --- a/internal/executor/ratelimit/ebpf/bpf_conn.go +++ b/internal/executor/ratelimit/ebpf/bpf_conn.go @@ -6,33 +6,20 @@ package ebpf import ( - "errors" - "fmt" "net" "sync" - "syscall" "time" - "github.com/cilium/ebpf" "github.com/google/uuid" "github.com/netsec-ethz/debuglet/internal/executor/debuglet/socket/netutil" ) type BpfConn struct { - count *BpfCount - conn net.Conn - // raw reaches the socket's descriptor while it is still open. Its Control - // holds a reference on the descriptor, so the number cannot be closed and - // reused by another socket during the delete, and it refuses to run once - // the connection is closed. - raw syscall.RawConn - // deleteStorage removes the socket's entry from debuglet_sk_map. The map - // is an SK_STORAGE map, whose key is a descriptor the kernel resolves to a - // socket at the time of the call. - deleteStorage func(fd uint32) error - domain string - id uuid.UUID - resolvedIPv6 netutil.IPv6 + count *BpfCount + conn net.Conn + domain string + id uuid.UUID + resolvedIPv6 netutil.IPv6 closeOnce sync.Once closeErr error @@ -46,43 +33,29 @@ func (bc *BpfConn) SetDeadline(t time.Time) error { return bc.conn.SetDeadl 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 releases the socket's map entry, the socket and its rate limit, in -// that order, once. Later calls return the first call's result. +// 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 { - // The socket storage is addressed by descriptor, so it must be deleted - // while the descriptor still names this socket: after Close it names - // nothing (EBADF) or, once reused, another socket. - deleteErr := bc.deleteSocketStorage() - // The socket is closed even when the delete failed; the kernel frees - // its storage with the socket. Close the connection before detaching - // the rate limit, otherwise rate-limited writes would be passed through - // before the connection is actually closed. + // Close connection before detaching/removing ratelimit, otherwise ratelimited writes will + // be passed through before the connection is actually closed closeErr := bc.conn.Close() detachErr := bc.count.Detach(bc.domain, bc.id, bc.resolvedIPv6) if closeErr != nil { return closeErr } - if deleteErr != nil { - return deleteErr - } return detachErr } - -func (bc *BpfConn) deleteSocketStorage() error { - var deleteErr error - if err := bc.raw.Control(func(fd uintptr) { - deleteErr = bc.deleteStorage(uint32(fd)) - }); err != nil { - return fmt.Errorf("delete socket storage: %w", err) - } - // An entry already gone needs no delete. - if deleteErr != nil && !errors.Is(deleteErr, ebpf.ErrKeyNotExist) { - return fmt.Errorf("delete socket storage: %w", deleteErr) - } - return nil -} diff --git a/internal/executor/ratelimit/ebpf/bpf_conn_test.go b/internal/executor/ratelimit/ebpf/bpf_conn_test.go index 78598294..56d680a0 100644 --- a/internal/executor/ratelimit/ebpf/bpf_conn_test.go +++ b/internal/executor/ratelimit/ebpf/bpf_conn_test.go @@ -7,27 +7,29 @@ package ebpf import ( "errors" - "fmt" "io" "net" + "net/netip" + "sync/atomic" "syscall" "testing" "time" "github.com/cilium/ebpf" "github.com/google/uuid" - "golang.org/x/sys/unix" + "github.com/netsec-ethz/debuglet/internal/bitrate" + "github.com/netsec-ethz/debuglet/internal/executor/debuglet/socket/netutil" ) -// closeRecorder notes when the wrapped connection is closed. +// closeRecorder counts closes of the wrapped connection. type closeRecorder struct { net.Conn - events *[]string + closes int err error } func (c *closeRecorder) Close() error { - *c.events = append(*c.events, "close") + c.closes++ if err := c.Conn.Close(); err != nil { return err } @@ -67,105 +69,68 @@ func loopbackConn(t *testing.T, network string) net.Conn { return nil } -// testBpfConn wraps conn the way Attach does, with deleter in place of the -// kernel map and a recorder on the connection's Close. -func testBpfConn(t *testing.T, conn net.Conn, events *[]string, closeErr error, deleter func(fd uint32) error) *BpfConn { - t.Helper() - raw, err := conn.(syscall.Conn).SyscallConn() - if err != nil { - t.Fatal(err) - } - return &BpfConn{ - count: new(BpfCount), conn: &closeRecorder{Conn: conn, events: events, err: closeErr}, raw: raw, - deleteStorage: func(fd uint32) error { - *events = append(*events, "delete") - return deleter(fd) - }, - domain: "127.0.0.1", id: uuid.New(), - } -} - -// Regression for #412: the SK_STORAGE entry is keyed by the descriptor, so it -// must be deleted while the descriptor is open and still names this socket. -func TestBpfConnDeletesSocketStorageBeforeClosing(t *testing.T) { +// 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) - local := conn.LocalAddr().String() - var events []string - var deletedFDs []uint32 - bc := testBpfConn(t, conn, &events, nil, func(fd uint32) error { - deletedFDs = append(deletedFDs, fd) - if _, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0); err != nil { - return fmt.Errorf("descriptor %d not open at delete: %w", fd, err) - } - name, err := unix.Getsockname(int(fd)) - if err != nil { - return fmt.Errorf("descriptor %d: %w", fd, err) - } - in4, ok := name.(*unix.SockaddrInet4) - if !ok || fmt.Sprintf("%s:%d", net.IP(in4.Addr[:]), in4.Port) != local { - return fmt.Errorf("descriptor %d names %v, not this socket %s", fd, name, local) - } - return nil - }) - if err := bc.Close(); err != nil { - t.Fatalf("Close: %v", err) + 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 fmt.Sprint(events) != "[delete close]" { - t.Fatalf("order = %v, want [delete close]", events) + if err := closeOnce(); err != nil { + t.Fatalf("Close: %v", err) } - // A second Close neither deletes by a descriptor number that may - // already name another socket nor closes again. - if err := bc.Close(); err != nil { + if err := closeOnce(); err != nil { t.Fatalf("second Close: %v", err) } - if fmt.Sprint(events) != "[delete close]" || len(deletedFDs) != 1 { - t.Fatalf("second Close acted again: events %v, deletes %v", events, deletedFDs) + 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 TestBpfConnCloseErrors(t *testing.T) { - deleteFailed := errors.New("map delete failed") +func TestBpfConnClosePrefersCloseError(t *testing.T) { closeFailed := errors.New("close failed") - for _, tc := range []struct { - name string - deleteErr, connErr error - want error - }{ - {name: "entry already gone", deleteErr: fmt.Errorf("lookup: %w", ebpf.ErrKeyNotExist)}, - {name: "delete fails", deleteErr: deleteFailed, want: deleteFailed}, - {name: "close error preferred", deleteErr: deleteFailed, connErr: closeFailed, want: closeFailed}, - } { - t.Run(tc.name, func(t *testing.T) { - conn := loopbackConn(t, "tcp") - var events []string - bc := testBpfConn(t, conn, &events, tc.connErr, func(uint32) error { return tc.deleteErr }) - err := bc.Close() - if tc.want == nil && err != nil || tc.want != nil && !errors.Is(err, tc.want) { - t.Fatalf("Close = %v, want %v", err, tc.want) - } - if fmt.Sprint(events) != "[delete close]" { - t.Fatalf("order = %v, want [delete close]", events) - } - if _, werr := conn.Write([]byte{1}); !errors.Is(werr, net.ErrClosed) { - t.Fatalf("socket left open after failed delete: %v", werr) - } - if again := bc.Close(); again != err { - t.Fatalf("second Close = %v, want first result %v", again, err) - } - }) + 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 remove only its own socket's entry. +// 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 { @@ -186,7 +151,7 @@ func TestKernelCounterClosesCountedConns(t *testing.T) { } }) const addr = "127.0.0.1" - attach := func(t *testing.T, conn net.Conn) (net.Conn, uuid.UUID) { + 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) @@ -194,7 +159,7 @@ func TestKernelCounterClosesCountedConns(t *testing.T) { conn.Close() t.Fatal(err) } - if err := counter.SetLimit(addr, id, 1<<30); err != nil { + if err := counter.SetLimit(addr, id, limit); err != nil { t.Fatal(err) } if err := counter.SetExecLimit(id, 1<<30); err != nil { @@ -203,6 +168,18 @@ func TestKernelCounterClosesCountedConns(t *testing.T) { 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() @@ -232,8 +209,32 @@ func TestKernelCounterClosesCountedConns(t *testing.T) { 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) } @@ -245,6 +246,7 @@ func TestKernelCounterClosesCountedConns(t *testing.T) { 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) { @@ -262,8 +264,8 @@ func TestKernelCounterClosesCountedConns(t *testing.T) { dialed.Close() t.Fatal(err) } - client, clientID := attach(t, dialed) - server, serverID := attach(t, accepted) + 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) @@ -272,7 +274,6 @@ func TestKernelCounterClosesCountedConns(t *testing.T) { exchange(t, client, server) exchange(t, server, client) closeCounted(t, client, clientID) - // Closing one socket must not remove another socket's entry. 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) } @@ -289,11 +290,83 @@ func TestKernelCounterClosesCountedConns(t *testing.T) { if err != nil { t.Fatal(err) } - sender, id := attach(t, dialed) + 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 3f288ba4..5766bd6d 100644 --- a/internal/executor/ratelimit/ebpf/bpf_count.go +++ b/internal/executor/ratelimit/ebpf/bpf_count.go @@ -124,6 +124,8 @@ func (bc *BpfCount) Attach(conn net.Conn, id uuid.UUID, addr string) (net.Conn, } var fdErr error + // 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) @@ -149,16 +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, raw: rawConn, deleteStorage: bc.deleteSocketStorage, - domain: addr, id: id, resolvedIPv6: ipv6, - }, nil -} - -// deleteSocketStorage removes the debuglet_sk_map entry of the socket that the -// open descriptor fd refers to. -func (bc *BpfCount) deleteSocketStorage(fd uint32) error { - return bc.objs.DebugletSkMap.Delete(fd) + 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 {