personal_interest_mirrors/nebula
1
0
Fork 0
mirror of https://github.com/slackhq/nebula.git synced 2026-05-16 04:47:38 +02:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into multiport

Browse Source
This commit is contained in:
Wade Simmons
2026-05-06 14:26:49 -04:00
parent 0824035906 4fb5cdb4fa
commit 2f50b3c54f
138 changed files with 10562 additions and 4541 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
udp/conn.go
View File

@@ -16,7 +16,7 @@ type EncReader func(
type Conn interface {
Rebind() error
LocalAddr() (netip.AddrPort, error)
ListenOut(r EncReader)
ListenOut(r EncReader) error
WriteTo(b []byte, addr netip.AddrPort) error
ReloadConfig(c *config.C)
SupportsMultipleReaders() bool
@@ -31,8 +31,8 @@ func (NoopConn) Rebind() error {
func (NoopConn) LocalAddr() (netip.AddrPort, error) {
return netip.AddrPort{}, nil
}
func (NoopConn) ListenOut(_ EncReader) {
return
func (NoopConn) ListenOut(_ EncReader) error {
return nil
}
func (NoopConn) SupportsMultipleReaders() bool {
return false

5
udp/udp_android.go
View File

@@ -9,11 +9,12 @@ import (
"net/netip"
"syscall"
"github.com/sirupsen/logrus"
"log/slog"
"golang.org/x/sys/unix"
)
func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
return NewGenericListener(l, ip, port, multi, batch)
}

5
udp/udp_bsd.go
View File

@@ -12,11 +12,12 @@ import (
"net/netip"
"syscall"
"github.com/sirupsen/logrus"
"log/slog"
"golang.org/x/sys/unix"
)
func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
return NewGenericListener(l, ip, port, multi, batch)
}

15
udp/udp_darwin.go
View File

@@ -8,12 +8,12 @@ import (
"encoding/binary"
"errors"
"fmt"
"log/slog"
"net"
"net/netip"
"syscall"
"unsafe"
"github.com/sirupsen/logrus"
"github.com/slackhq/nebula/config"
"golang.org/x/sys/unix"
)
@@ -22,12 +22,12 @@ type StdConn struct {
*net.UDPConn
isV4 bool
sysFd uintptr
l *logrus.Logger
l *slog.Logger
}
var _ Conn = &StdConn{}
func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
lc := NewListenConfig(multi)
pc, err := lc.ListenPacket(context.TODO(), "udp", net.JoinHostPort(ip.String(), fmt.Sprintf("%v", port)))
if err != nil {
@@ -165,7 +165,7 @@ func NewUDPStatsEmitter(udpConns []Conn) func() {
return func() {}
}
func (u *StdConn) ListenOut(r EncReader) {
func (u *StdConn) ListenOut(r EncReader) error {
buffer := make([]byte, MTU)
for {
@@ -173,11 +173,10 @@ func (u *StdConn) ListenOut(r EncReader) {
n, rua, err := u.ReadFromUDPAddrPort(buffer)
if err != nil {
if errors.Is(err, net.ErrClosed) {
u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
return
return err
}
u.l.WithError(err).Error("unexpected udp socket receive error")
u.l.Error("unexpected udp socket receive error", "error", err)
}
r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n])
@@ -197,7 +196,7 @@ func (u *StdConn) Rebind() error {
}
if err != nil {
u.l.WithError(err).Error("Failed to rebind udp socket")
u.l.Error("Failed to rebind udp socket", "error", err)
}
return nil

13
udp/udp_generic.go
View File

@@ -12,22 +12,22 @@ import (
"context"
"errors"
"fmt"
"log/slog"
"net"
"net/netip"
"time"
"github.com/sirupsen/logrus"
"github.com/slackhq/nebula/config"
)
type GenericConn struct {
*net.UDPConn
l *logrus.Logger
l *slog.Logger
}
var _ Conn = &GenericConn{}
func NewGenericListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
func NewGenericListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
lc := NewListenConfig(multi)
pc, err := lc.ListenPacket(context.TODO(), "udp", net.JoinHostPort(ip.String(), fmt.Sprintf("%v", port)))
if err != nil {
@@ -73,7 +73,7 @@ type rawMessage struct {
Len uint32
}
func (u *GenericConn) ListenOut(r EncReader) {
func (u *GenericConn) ListenOut(r EncReader) error {
buffer := make([]byte, MTU)
var lastRecvErr time.Time
@@ -83,13 +83,12 @@ func (u *GenericConn) ListenOut(r EncReader) {
n, rua, err := u.ReadFromUDPAddrPort(buffer)
if err != nil {
if errors.Is(err, net.ErrClosed) {
u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
return
return err
}
// Dampen unexpected message warns to once per minute
if lastRecvErr.IsZero() || time.Since(lastRecvErr) > time.Minute {
lastRecvErr = time.Now()
u.l.WithError(err).Warn("unexpected udp socket receive error")
u.l.Warn("unexpected udp socket receive error", "error", err)
}
continue
}

350
udp/udp_linux.go
View File

@@ -4,72 +4,73 @@
package udp
import (
"context"
"encoding/binary"
"fmt"
"log/slog"
"net"
"net/netip"
"syscall"
"unsafe"
"github.com/rcrowley/go-metrics"
"github.com/sirupsen/logrus"
"github.com/slackhq/nebula/config"
"golang.org/x/sys/unix"
)
type StdConn struct {
sysFd int
isV4 bool
l *logrus.Logger
batch int
udpConn *net.UDPConn
rawConn syscall.RawConn
isV4 bool
l *slog.Logger
batch int
}
func maybeIPV4(ip net.IP) (net.IP, bool) {
ip4 := ip.To4()
if ip4 != nil {
return ip4, true
}
return ip, false
}
func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
af := unix.AF_INET6
if ip.Is4() {
af = unix.AF_INET
}
syscall.ForkLock.RLock()
fd, err := unix.Socket(af, unix.SOCK_DGRAM, unix.IPPROTO_UDP)
if err == nil {
unix.CloseOnExec(fd)
}
syscall.ForkLock.RUnlock()
func setReusePort(network, address string, c syscall.RawConn) error {
var opErr error
err := c.Control(func(fd uintptr) {
opErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
//CloseOnExec already set by the runtime
})
if err != nil {
return err
}
return opErr
}
func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
listen := netip.AddrPortFrom(ip, uint16(port))
lc := net.ListenConfig{}
if multi {
lc.Control = setReusePort
}
//this context is only used during the bind operation, you can't cancel it to kill the socket
pc, err := lc.ListenPacket(context.Background(), "udp", listen.String())
if err != nil {
unix.Close(fd)
return nil, fmt.Errorf("unable to open socket: %s", err)
}
if multi {
if err = unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEPORT, 1); err != nil {
return nil, fmt.Errorf("unable to set SO_REUSEPORT: %s", err)
}
udpConn := pc.(*net.UDPConn)
rawConn, err := udpConn.SyscallConn()
if err != nil {
_ = udpConn.Close()
return nil, err
}
//gotta find out if we got an AF_INET6 socket or not:
out := &StdConn{
udpConn: udpConn,
rawConn: rawConn,
l: l,
batch: batch,
}
var sa unix.Sockaddr
if ip.Is4() {
sa4 := &unix.SockaddrInet4{Port: port}
sa4.Addr = ip.As4()
sa = sa4
} else {
sa6 := &unix.SockaddrInet6{Port: port}
sa6.Addr = ip.As16()
sa = sa6
}
if err = unix.Bind(fd, sa); err != nil {
return nil, fmt.Errorf("unable to bind to socket: %s", err)
af, err := out.getSockOptInt(unix.SO_DOMAIN)
if err != nil {
_ = out.Close()
return nil, err
}
out.isV4 = af == unix.AF_INET
return &StdConn{sysFd: fd, isV4: ip.Is4(), l: l, batch: batch}, err
return out, nil
}
func (u *StdConn) SupportsMultipleReaders() bool {
@@ -80,62 +81,133 @@ func (u *StdConn) Rebind() error {
return nil
}
func (u *StdConn) getSockOptInt(opt int) (int, error) {
if u.rawConn == nil {
return 0, fmt.Errorf("no UDP connection")
}
var out int
var opErr error
err := u.rawConn.Control(func(fd uintptr) {
out, opErr = unix.GetsockoptInt(int(fd), unix.SOL_SOCKET, opt)
})
if err != nil {
return 0, err
}
return out, opErr
}
func (u *StdConn) setSockOptInt(opt int, n int) error {
if u.rawConn == nil {
return fmt.Errorf("no UDP connection")
}
var opErr error
err := u.rawConn.Control(func(fd uintptr) {
opErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, opt, n)
})
if err != nil {
return err
}
return opErr
}
func (u *StdConn) SetRecvBuffer(n int) error {
return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_RCVBUFFORCE, n)
return u.setSockOptInt(unix.SO_RCVBUFFORCE, n)
}
func (u *StdConn) SetSendBuffer(n int) error {
return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_SNDBUFFORCE, n)
return u.setSockOptInt(unix.SO_SNDBUFFORCE, n)
}
func (u *StdConn) SetSoMark(mark int) error {
return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_MARK, mark)
return u.setSockOptInt(unix.SO_MARK, mark)
}
func (u *StdConn) GetRecvBuffer() (int, error) {
return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_RCVBUF)
return u.getSockOptInt(unix.SO_RCVBUF)
}
func (u *StdConn) GetSendBuffer() (int, error) {
return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_SNDBUF)
return u.getSockOptInt(unix.SO_SNDBUF)
}
func (u *StdConn) GetSoMark() (int, error) {
return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_MARK)
return u.getSockOptInt(unix.SO_MARK)
}
func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
sa, err := unix.Getsockname(u.sysFd)
if err != nil {
return netip.AddrPort{}, err
}
a := u.udpConn.LocalAddr()
switch sa := sa.(type) {
case *unix.SockaddrInet4:
return netip.AddrPortFrom(netip.AddrFrom4(sa.Addr), uint16(sa.Port)), nil
case *unix.SockaddrInet6:
return netip.AddrPortFrom(netip.AddrFrom16(sa.Addr), uint16(sa.Port)), nil
switch v := a.(type) {
case *net.UDPAddr:
addr, ok := netip.AddrFromSlice(v.IP)
if !ok {
return netip.AddrPort{}, fmt.Errorf("LocalAddr returned invalid IP address: %s", v.IP)
}
return netip.AddrPortFrom(addr, uint16(v.Port)), nil
default:
return netip.AddrPort{}, fmt.Errorf("unsupported sock type: %T", sa)
return netip.AddrPort{}, fmt.Errorf("LocalAddr returned: %#v", a)
}
}
func (u *StdConn) ListenOut(r EncReader) {
func recvmmsg(fd uintptr, msgs []rawMessage) (int, bool, error) {
var errno syscall.Errno
n, _, errno := unix.Syscall6(
unix.SYS_RECVMMSG,
fd,
uintptr(unsafe.Pointer(&msgs[0])),
uintptr(len(msgs)),
unix.MSG_WAITFORONE,
0,
0,
)
if errno == syscall.EAGAIN || errno == syscall.EWOULDBLOCK {
// No data available, block for I/O and try again.
return int(n), false, nil
}
if errno != 0 {
return int(n), true, &net.OpError{Op: "recvmmsg", Err: errno}
}
return int(n), true, nil
}
func (u *StdConn) listenOutSingle(r EncReader) error {
var err error
var n int
var from netip.AddrPort
buffer := make([]byte, MTU)
for {
n, from, err = u.udpConn.ReadFromUDPAddrPort(buffer)
if err != nil {
return err
}
from = netip.AddrPortFrom(from.Addr().Unmap(), from.Port())
r(from, buffer[:n])
}
}
func (u *StdConn) listenOutBatch(r EncReader) error {
var ip netip.Addr
var n int
var operr error
msgs, buffers, names := u.PrepareRawMessages(u.batch)
read := u.ReadMulti
if u.batch == 1 {
read = u.ReadSingle
//reader needs to capture variables from this function, since it's used as a lambda with rawConn.Read
//defining it outside the loop so it gets re-used
reader := func(fd uintptr) (done bool) {
n, done, operr = recvmmsg(fd, msgs)
return done
}
for {
n, err := read(msgs)
err := u.rawConn.Read(reader)
if err != nil {
u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
return
return err
}
if operr != nil {
return operr
}
for i := 0; i < n; i++ {
@@ -150,106 +222,17 @@ func (u *StdConn) ListenOut(r EncReader) {
}
}
func (u *StdConn) ReadSingle(msgs []rawMessage) (int, error) {
for {
n, _, err := unix.Syscall6(
unix.SYS_RECVMSG,
uintptr(u.sysFd),
uintptr(unsafe.Pointer(&(msgs[0].Hdr))),
0,
0,
0,
0,
)
if err != 0 {
return 0, &net.OpError{Op: "recvmsg", Err: err}
}
msgs[0].Len = uint32(n)
return 1, nil
}
}
func (u *StdConn) ReadMulti(msgs []rawMessage) (int, error) {
for {
n, _, err := unix.Syscall6(
unix.SYS_RECVMMSG,
uintptr(u.sysFd),
uintptr(unsafe.Pointer(&msgs[0])),
uintptr(len(msgs)),
unix.MSG_WAITFORONE,
0,
0,
)
if err != 0 {
return 0, &net.OpError{Op: "recvmmsg", Err: err}
}
return int(n), nil
func (u *StdConn) ListenOut(r EncReader) error {
if u.batch == 1 {
return u.listenOutSingle(r)
} else {
return u.listenOutBatch(r)
}
}
func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort) error {
if u.isV4 {
return u.writeTo4(b, ip)
}
return u.writeTo6(b, ip)
}
func (u *StdConn) writeTo6(b []byte, ip netip.AddrPort) error {
var rsa unix.RawSockaddrInet6
rsa.Family = unix.AF_INET6
rsa.Addr = ip.Addr().As16()
binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&rsa.Port))[:], ip.Port())
for {
_, _, err := unix.Syscall6(
unix.SYS_SENDTO,
uintptr(u.sysFd),
uintptr(unsafe.Pointer(&b[0])),
uintptr(len(b)),
uintptr(0),
uintptr(unsafe.Pointer(&rsa)),
uintptr(unix.SizeofSockaddrInet6),
)
if err != 0 {
return &net.OpError{Op: "sendto", Err: err}
}
return nil
}
}
func (u *StdConn) writeTo4(b []byte, ip netip.AddrPort) error {
if !ip.Addr().Is4() {
return ErrInvalidIPv6RemoteForSocket
}
var rsa unix.RawSockaddrInet4
rsa.Family = unix.AF_INET
rsa.Addr = ip.Addr().As4()
binary.BigEndian.PutUint16((*[2]byte)(unsafe.Pointer(&rsa.Port))[:], ip.Port())
for {
_, _, err := unix.Syscall6(
unix.SYS_SENDTO,
uintptr(u.sysFd),
uintptr(unsafe.Pointer(&b[0])),
uintptr(len(b)),
uintptr(0),
uintptr(unsafe.Pointer(&rsa)),
uintptr(unix.SizeofSockaddrInet4),
)
if err != 0 {
return &net.OpError{Op: "sendto", Err: err}
}
return nil
}
_, err := u.udpConn.WriteToUDPAddrPort(b, ip)
return err
}
func (u *StdConn) ReloadConfig(c *config.C) {
@@ -259,12 +242,12 @@ func (u *StdConn) ReloadConfig(c *config.C) {
if err == nil {
s, err := u.GetRecvBuffer()
if err == nil {
u.l.WithField("size", s).Info("listen.read_buffer was set")
u.l.Info("listen.read_buffer was set", "size", s)
} else {
u.l.WithError(err).Warn("Failed to get listen.read_buffer")
u.l.Warn("Failed to get listen.read_buffer", "error", err)
}
} else {
u.l.WithError(err).Error("Failed to set listen.read_buffer")
u.l.Error("Failed to set listen.read_buffer", "error", err)
}
}
@@ -274,12 +257,12 @@ func (u *StdConn) ReloadConfig(c *config.C) {
if err == nil {
s, err := u.GetSendBuffer()
if err == nil {
u.l.WithField("size", s).Info("listen.write_buffer was set")
u.l.Info("listen.write_buffer was set", "size", s)
} else {
u.l.WithError(err).Warn("Failed to get listen.write_buffer")
u.l.Warn("Failed to get listen.write_buffer", "error", err)
}
} else {
u.l.WithError(err).Error("Failed to set listen.write_buffer")
u.l.Error("Failed to set listen.write_buffer", "error", err)
}
}
@@ -290,27 +273,40 @@ func (u *StdConn) ReloadConfig(c *config.C) {
if err == nil {
s, err := u.GetSoMark()
if err == nil {
u.l.WithField("mark", s).Info("listen.so_mark was set")
u.l.Info("listen.so_mark was set", "mark", s)
} else {
u.l.WithError(err).Warn("Failed to get listen.so_mark")
u.l.Warn("Failed to get listen.so_mark", "error", err)
}
} else {
u.l.WithError(err).Error("Failed to set listen.so_mark")
u.l.Error("Failed to set listen.so_mark", "error", err)
}
}
}
func (u *StdConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error {
var vallen uint32 = 4 * unix.SK_MEMINFO_VARS
_, _, err := unix.Syscall6(unix.SYS_GETSOCKOPT, uintptr(u.sysFd), uintptr(unix.SOL_SOCKET), uintptr(unix.SO_MEMINFO), uintptr(unsafe.Pointer(meminfo)), uintptr(unsafe.Pointer(&vallen)), 0)
if err != 0 {
if u.rawConn == nil {
return fmt.Errorf("no UDP connection")
}
var opErr error
err := u.rawConn.Control(func(fd uintptr) {
_, _, syserr := unix.Syscall6(unix.SYS_GETSOCKOPT, fd, uintptr(unix.SOL_SOCKET), uintptr(unix.SO_MEMINFO), uintptr(unsafe.Pointer(meminfo)), uintptr(unsafe.Pointer(&vallen)), 0)
if syserr != 0 {
opErr = syserr
}
})
if err != nil {
return err
}
return nil
return opErr
}
func (u *StdConn) Close() error {
return syscall.Close(u.sysFd)
if u.udpConn != nil {
return u.udpConn.Close()
}
return nil
}
func NewUDPStatsEmitter(udpConns []Conn) func() {

5
udp/udp_netbsd.go
View File

@@ -11,11 +11,12 @@ import (
"net/netip"
"syscall"
"github.com/sirupsen/logrus"
"log/slog"
"golang.org/x/sys/unix"
)
func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
return NewGenericListener(l, ip, port, multi, batch)
}

19
udp/udp_rio_windows.go
View File

@@ -9,6 +9,7 @@ import (
"errors"
"fmt"
"io"
"log/slog"
"net"
"net/netip"
"sync"
@@ -17,7 +18,6 @@ import (
"time"
"unsafe"
"github.com/sirupsen/logrus"
"github.com/slackhq/nebula/config"
"golang.org/x/sys/windows"
"golang.zx2c4.com/wireguard/conn/winrio"
@@ -53,14 +53,14 @@ type ringBuffer struct {
type RIOConn struct {
isOpen atomic.Bool
l *logrus.Logger
l *slog.Logger
sock windows.Handle
rx, tx ringBuffer
rq winrio.Rq
results [packetsPerRing]winrio.Result
}
func NewRIOListener(l *logrus.Logger, addr netip.Addr, port int) (*RIOConn, error) {
func NewRIOListener(l *slog.Logger, addr netip.Addr, port int) (*RIOConn, error) {
if !winrio.Initialize() {
return nil, errors.New("could not initialize winrio")
}
@@ -83,7 +83,7 @@ func NewRIOListener(l *logrus.Logger, addr netip.Addr, port int) (*RIOConn, erro
return u, nil
}
func (u *RIOConn) bind(l *logrus.Logger, sa windows.Sockaddr) error {
func (u *RIOConn) bind(l *slog.Logger, sa windows.Sockaddr) error {
var err error
u.sock, err = winrio.Socket(windows.AF_INET6, windows.SOCK_DGRAM, windows.IPPROTO_UDP)
if err != nil {
@@ -103,7 +103,7 @@ func (u *RIOConn) bind(l *logrus.Logger, sa windows.Sockaddr) error {
if err != nil {
// This is a best-effort to prevent errors from being returned by the udp recv operation.
// Quietly log a failure and continue.
l.WithError(err).Debug("failed to set UDP_CONNRESET ioctl")
l.Debug("failed to set UDP_CONNRESET ioctl", "error", err)
}
ret = 0
@@ -114,7 +114,7 @@ func (u *RIOConn) bind(l *logrus.Logger, sa windows.Sockaddr) error {
if err != nil {
// This is a best-effort to prevent errors from being returned by the udp recv operation.
// Quietly log a failure and continue.
l.WithError(err).Debug("failed to set UDP_NETRESET ioctl")
l.Debug("failed to set UDP_NETRESET ioctl", "error", err)
}
err = u.rx.Open()
@@ -140,7 +140,7 @@ func (u *RIOConn) bind(l *logrus.Logger, sa windows.Sockaddr) error {
return nil
}
func (u *RIOConn) ListenOut(r EncReader) {
func (u *RIOConn) ListenOut(r EncReader) error {
buffer := make([]byte, MTU)
var lastRecvErr time.Time
@@ -151,13 +151,12 @@ func (u *RIOConn) ListenOut(r EncReader) {
if err != nil {
if errors.Is(err, net.ErrClosed) {
u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
return
return err
}
// Dampen unexpected message warns to once per minute
if lastRecvErr.IsZero() || time.Since(lastRecvErr) > time.Minute {
lastRecvErr = time.Now()
u.l.WithError(err).Warn("unexpected udp socket receive error")
u.l.Warn("unexpected udp socket receive error", "error", err)
}
continue
}

141
udp/udp_tester.go
View File

@@ -4,11 +4,13 @@
package udp
import (
"context"
"io"
"log/slog"
"net/netip"
"sync/atomic"
"os"
"sync"
"github.com/sirupsen/logrus"
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/header"
)
@@ -19,32 +21,72 @@ type Packet struct {
Data []byte
}
// Copy returns a fresh *Packet (from the freelist) with a duplicate Data buffer.
func (u *Packet) Copy() *Packet {
n := &Packet{
To: u.To,
From: u.From,
Data: make([]byte, len(u.Data)),
n := acquirePacket()
n.To = u.To
n.From = u.From
if cap(n.Data) < len(u.Data) {
n.Data = make([]byte, len(u.Data))
} else {
n.Data = n.Data[:len(u.Data)]
}
copy(n.Data, u.Data)
return n
}
// Release returns p to the harness packet freelist.
// Callers that pull a *Packet from Get / TxPackets must Release when done.
// Channel-backed instead of sync.Pool because sync.Pool's per-P caches drain badly under cross-goroutine Get/Put,
// and putting a []byte in a Pool escapes the slice header to heap.
func (p *Packet) Release() {
if p == nil {
return
}
p.Data = p.Data[:0]
select {
case packetFreelist <- p:
default:
// Freelist full; drop the *Packet for the GC.
}
}
// packetFreelist retains *Packet structs (and their backing Data arrays) so steady-state allocation drops to zero.
var packetFreelist = make(chan *Packet, 64)
func acquirePacket() *Packet {
select {
case p := <-packetFreelist:
return p
default:
return &Packet{}
}
}
type TesterConn struct {
Addr netip.AddrPort
RxPackets chan *Packet // Packets to receive into nebula
TxPackets chan *Packet // Packets transmitted outside by nebula
closed atomic.Bool
l *logrus.Logger
// done is closed exactly once by Close. Senders select on it so they
// never race with a channel close; readers exit when it fires. The
// packet channels are intentionally never closed - that was the source
// of `send on closed channel` panics when a WriteTo/Send from another
// goroutine passed the close check and reached the send just after
// Close ran.
done chan struct{}
closeOnce sync.Once
l *slog.Logger
}
func NewListener(l *logrus.Logger, ip netip.Addr, port int, _ bool, _ int) (Conn, error) {
func NewListener(l *slog.Logger, ip netip.Addr, port int, _ bool, _ int) (Conn, error) {
return &TesterConn{
Addr: netip.AddrPortFrom(ip, uint16(port)),
RxPackets: make(chan *Packet, 10),
TxPackets: make(chan *Packet, 10),
done: make(chan struct{}),
l: l,
}, nil
}
@@ -53,21 +95,23 @@ func NewListener(l *logrus.Logger, ip netip.Addr, port int, _ bool, _ int) (Conn
// this is an encrypted packet or a handshake message in most cases
// packets were transmitted from another nebula node, you can send them with Tun.Send
func (u *TesterConn) Send(packet *Packet) {
if u.closed.Load() {
return
if u.l.Enabled(context.Background(), slog.LevelDebug) {
// Parse the header only under debug logging, otherwise the
// allocation would show up in every Send call.
var h header.H
if err := h.Parse(packet.Data); err != nil {
panic(err)
}
u.l.Debug("UDP receiving injected packet",
"header", &h,
"udpAddr", packet.From,
"dataLen", len(packet.Data),
)
}
h := &header.H{}
if err := h.Parse(packet.Data); err != nil {
panic(err)
select {
case <-u.done:
case u.RxPackets <- packet:
}
if u.l.Level >= logrus.DebugLevel {
u.l.WithField("header", h).
WithField("udpAddr", packet.From).
WithField("dataLen", len(packet.Data)).
Debug("UDP receiving injected packet")
}
u.RxPackets <- packet
}
// Get will pull a UdpPacket from the transmit queue
@@ -75,7 +119,12 @@ func (u *TesterConn) Send(packet *Packet) {
// packets were ingested from the tun side (in most cases), you can send them with Tun.Send
func (u *TesterConn) Get(block bool) *Packet {
if block {
return <-u.TxPackets
select {
case <-u.done:
return nil
case p := <-u.TxPackets:
return p
}
}
select {
@@ -91,28 +140,33 @@ func (u *TesterConn) Get(block bool) *Packet {
//********************************************************************************************************************//
func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error {
if u.closed.Load() {
return io.ErrClosedPipe
p := acquirePacket()
if cap(p.Data) < len(b) {
p.Data = make([]byte, len(b))
} else {
p.Data = p.Data[:len(b)]
}
p := &Packet{
Data: make([]byte, len(b), len(b)),
From: u.Addr,
To: addr,
}
copy(p.Data, b)
u.TxPackets <- p
return nil
p.From = u.Addr
p.To = addr
select {
case <-u.done:
p.Release()
return io.ErrClosedPipe
case u.TxPackets <- p:
return nil
}
}
func (u *TesterConn) ListenOut(r EncReader) {
func (u *TesterConn) ListenOut(r EncReader) error {
for {
p, ok := <-u.RxPackets
if !ok {
return
select {
case <-u.done:
return os.ErrClosed
case p := <-u.RxPackets:
r(p.From, p.Data)
p.Release()
}
r(p.From, p.Data)
}
}
@@ -136,9 +190,8 @@ func (u *TesterConn) Rebind() error {
}
func (u *TesterConn) Close() error {
if u.closed.CompareAndSwap(false, true) {
close(u.RxPackets)
close(u.TxPackets)
}
u.closeOnce.Do(func() {
close(u.done)
})
return nil
}

7
udp/udp_windows.go
View File

@@ -5,14 +5,13 @@ package udp
import (
"fmt"
"log/slog"
"net"
"net/netip"
"syscall"
"github.com/sirupsen/logrus"
)
func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
if multi {
//NOTE: Technically we can support it with RIO but it wouldn't be at the socket level
// The udp stack would need to be reworked to hide away the implementation differences between
@@ -25,7 +24,7 @@ func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch in
return rc, nil
}
l.WithError(err).Error("Falling back to standard udp sockets")
l.Error("Falling back to standard udp sockets", "error", err)
return NewGenericListener(l, ip, port, multi, batch)
}