personal_interest_mirrors/nebula
1
0
Fork 0
mirror of https://github.com/slackhq/nebula.git synced 2025-11-23 00:44:25 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

Use an interface for udp conns (#901)

Browse Source
This commit is contained in:
Nate Brown
2023-06-14 10:48:52 -05:00
committed by GitHub
parent 928731acfe
commit 3bbf5f4e67
17 changed files with 95 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
udp/conn.go
View File

@@ -1,6 +1,7 @@
package udp
import (
"github.com/slackhq/nebula/config"
"github.com/slackhq/nebula/firewall"
"github.com/slackhq/nebula/header"
)
@@ -18,3 +19,29 @@ type EncReader func(
q int,
localCache firewall.ConntrackCache,
)
type Conn interface {
Rebind() error
LocalAddr() (*Addr, error)
ListenOut(r EncReader, lhf LightHouseHandlerFunc, cache *firewall.ConntrackCacheTicker, q int)
WriteTo(b []byte, addr *Addr) error
ReloadConfig(c *config.C)
}
type NoopConn struct{}
func (NoopConn) Rebind() error {
return nil
}
func (NoopConn) LocalAddr() (*Addr, error) {
return nil, nil
}
func (NoopConn) ListenOut(_ EncReader, _ LightHouseHandlerFunc, _ *firewall.ConntrackCacheTicker, _ int) {
return
}
func (NoopConn) WriteTo(_ []byte, _ *Addr) error {
return nil
}
func (NoopConn) ReloadConfig(_ *config.C) {
return
}

2
udp/udp_android.go
View File

@@ -34,6 +34,6 @@ func NewListenConfig(multi bool) net.ListenConfig {
}
}
func (u *Conn) Rebind() error {
func (u *GenericConn) Rebind() error {
return nil
}

2
udp/udp_darwin.go
View File

@@ -37,7 +37,7 @@ func NewListenConfig(multi bool) net.ListenConfig {
}
}
func (u *Conn) Rebind() error {
func (u *GenericConn) Rebind() error {
file, err := u.File()
if err != nil {
return err

2
udp/udp_freebsd.go
View File

@@ -36,6 +36,6 @@ func NewListenConfig(multi bool) net.ListenConfig {
}
}
func (u *Conn) Rebind() error {
func (u *GenericConn) Rebind() error {
return nil
}

20
udp/udp_generic.go
View File

@@ -18,30 +18,30 @@ import (
"github.com/slackhq/nebula/header"
)
type Conn struct {
type GenericConn struct {
*net.UDPConn
l *logrus.Logger
}
func NewListener(l *logrus.Logger, ip net.IP, port int, multi bool, batch int) (*Conn, error) {
func NewListener(l *logrus.Logger, ip net.IP, 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 {
return nil, err
}
if uc, ok := pc.(*net.UDPConn); ok {
return &Conn{UDPConn: uc, l: l}, nil
return &GenericConn{UDPConn: uc, l: l}, nil
}
return nil, fmt.Errorf("Unexpected PacketConn: %T %#v", pc, pc)
}
func (uc *Conn) WriteTo(b []byte, addr *Addr) error {
_, err := uc.UDPConn.WriteToUDP(b, &net.UDPAddr{IP: addr.IP, Port: int(addr.Port)})
func (u *GenericConn) WriteTo(b []byte, addr *Addr) error {
_, err := u.UDPConn.WriteToUDP(b, &net.UDPAddr{IP: addr.IP, Port: int(addr.Port)})
return err
}
func (uc *Conn) LocalAddr() (*Addr, error) {
a := uc.UDPConn.LocalAddr()
func (u *GenericConn) LocalAddr() (*Addr, error) {
a := u.UDPConn.LocalAddr()
switch v := a.(type) {
case *net.UDPAddr:
@@ -55,11 +55,11 @@ func (uc *Conn) LocalAddr() (*Addr, error) {
}
}
func (u *Conn) ReloadConfig(c *config.C) {
func (u *GenericConn) ReloadConfig(c *config.C) {
// TODO
}
func NewUDPStatsEmitter(udpConns []*Conn) func() {
func NewUDPStatsEmitter(udpConns []Conn) func() {
// No UDP stats for non-linux
return func() {}
}
@@ -68,7 +68,7 @@ type rawMessage struct {
Len uint32
}
func (u *Conn) ListenOut(r EncReader, lhf LightHouseHandlerFunc, cache *firewall.ConntrackCacheTicker, q int) {
func (u *GenericConn) ListenOut(r EncReader, lhf LightHouseHandlerFunc, cache *firewall.ConntrackCacheTicker, q int) {
plaintext := make([]byte, MTU)
buffer := make([]byte, MTU)
h := &header.H{}

36
udp/udp_linux.go
View File

@@ -20,7 +20,7 @@ import (
//TODO: make it support reload as best you can!
type Conn struct {
type StdConn struct {
sysFd int
l *logrus.Logger
batch int
@@ -45,7 +45,7 @@ const (
type _SK_MEMINFO [_SK_MEMINFO_VARS]uint32
func NewListener(l *logrus.Logger, ip net.IP, port int, multi bool, batch int) (*Conn, error) {
func NewListener(l *logrus.Logger, ip net.IP, port int, multi bool, batch int) (Conn, error) {
syscall.ForkLock.RLock()
fd, err := unix.Socket(unix.AF_INET6, unix.SOCK_DGRAM, unix.IPPROTO_UDP)
if err == nil {
@@ -77,30 +77,30 @@ func NewListener(l *logrus.Logger, ip net.IP, port int, multi bool, batch int) (
//v, err := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_INCOMING_CPU)
//l.Println(v, err)
return &Conn{sysFd: fd, l: l, batch: batch}, err
return &StdConn{sysFd: fd, l: l, batch: batch}, err
}
func (u *Conn) Rebind() error {
func (u *StdConn) Rebind() error {
return nil
}
func (u *Conn) SetRecvBuffer(n int) error {
func (u *StdConn) SetRecvBuffer(n int) error {
return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_RCVBUFFORCE, n)
}
func (u *Conn) SetSendBuffer(n int) error {
func (u *StdConn) SetSendBuffer(n int) error {
return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_SNDBUFFORCE, n)
}
func (u *Conn) GetRecvBuffer() (int, error) {
func (u *StdConn) GetRecvBuffer() (int, error) {
return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_RCVBUF)
}
func (u *Conn) GetSendBuffer() (int, error) {
func (u *StdConn) GetSendBuffer() (int, error) {
return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_SNDBUF)
}
func (u *Conn) LocalAddr() (*Addr, error) {
func (u *StdConn) LocalAddr() (*Addr, error) {
sa, err := unix.Getsockname(u.sysFd)
if err != nil {
return nil, err
@@ -119,7 +119,7 @@ func (u *Conn) LocalAddr() (*Addr, error) {
return addr, nil
}
func (u *Conn) ListenOut(r EncReader, lhf LightHouseHandlerFunc, cache *firewall.ConntrackCacheTicker, q int) {
func (u *StdConn) ListenOut(r EncReader, lhf LightHouseHandlerFunc, cache *firewall.ConntrackCacheTicker, q int) {
plaintext := make([]byte, MTU)
h := &header.H{}
fwPacket := &firewall.Packet{}
@@ -150,7 +150,7 @@ func (u *Conn) ListenOut(r EncReader, lhf LightHouseHandlerFunc, cache *firewall
}
}
func (u *Conn) ReadSingle(msgs []rawMessage) (int, error) {
func (u *StdConn) ReadSingle(msgs []rawMessage) (int, error) {
for {
n, _, err := unix.Syscall6(
unix.SYS_RECVMSG,
@@ -171,7 +171,7 @@ func (u *Conn) ReadSingle(msgs []rawMessage) (int, error) {
}
}
func (u *Conn) ReadMulti(msgs []rawMessage) (int, error) {
func (u *StdConn) ReadMulti(msgs []rawMessage) (int, error) {
for {
n, _, err := unix.Syscall6(
unix.SYS_RECVMMSG,
@@ -191,7 +191,7 @@ func (u *Conn) ReadMulti(msgs []rawMessage) (int, error) {
}
}
func (u *Conn) WriteTo(b []byte, addr *Addr) error {
func (u *StdConn) WriteTo(b []byte, addr *Addr) error {
var rsa unix.RawSockaddrInet6
rsa.Family = unix.AF_INET6
@@ -221,7 +221,7 @@ func (u *Conn) WriteTo(b []byte, addr *Addr) error {
}
}
func (u *Conn) ReloadConfig(c *config.C) {
func (u *StdConn) ReloadConfig(c *config.C) {
b := c.GetInt("listen.read_buffer", 0)
if b > 0 {
err := u.SetRecvBuffer(b)
@@ -253,7 +253,7 @@ func (u *Conn) ReloadConfig(c *config.C) {
}
}
func (u *Conn) getMemInfo(meminfo *_SK_MEMINFO) error {
func (u *StdConn) getMemInfo(meminfo *_SK_MEMINFO) error {
var vallen uint32 = 4 * _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 {
@@ -262,11 +262,11 @@ func (u *Conn) getMemInfo(meminfo *_SK_MEMINFO) error {
return nil
}
func NewUDPStatsEmitter(udpConns []*Conn) func() {
func NewUDPStatsEmitter(udpConns []Conn) func() {
// Check if our kernel supports SO_MEMINFO before registering the gauges
var udpGauges [][_SK_MEMINFO_VARS]metrics.Gauge
var meminfo _SK_MEMINFO
if err := udpConns[0].getMemInfo(&meminfo); err == nil {
if err := udpConns[0].(*StdConn).getMemInfo(&meminfo); err == nil {
udpGauges = make([][_SK_MEMINFO_VARS]metrics.Gauge, len(udpConns))
for i := range udpConns {
udpGauges[i] = [_SK_MEMINFO_VARS]metrics.Gauge{
@@ -285,7 +285,7 @@ func NewUDPStatsEmitter(udpConns []*Conn) func() {
return func() {
for i, gauges := range udpGauges {
if err := udpConns[i].getMemInfo(&meminfo); err == nil {
if err := udpConns[i].(*StdConn).getMemInfo(&meminfo); err == nil {
for j := 0; j < _SK_MEMINFO_VARS; j++ {
gauges[j].Update(int64(meminfo[j]))
}

2
udp/udp_linux_32.go
View File

@@ -30,7 +30,7 @@ type rawMessage struct {
Len uint32
}
func (u *Conn) PrepareRawMessages(n int) ([]rawMessage, [][]byte, [][]byte) {
func (u *StdConn) PrepareRawMessages(n int) ([]rawMessage, [][]byte, [][]byte) {
msgs := make([]rawMessage, n)
buffers := make([][]byte, n)
names := make([][]byte, n)

2
udp/udp_linux_64.go
View File

@@ -33,7 +33,7 @@ type rawMessage struct {
Pad0 [4]byte
}
func (u *Conn) PrepareRawMessages(n int) ([]rawMessage, [][]byte, [][]byte) {
func (u *StdConn) PrepareRawMessages(n int) ([]rawMessage, [][]byte, [][]byte) {
msgs := make([]rawMessage, n)
buffers := make([][]byte, n)
names := make([][]byte, n)

22
udp/udp_tester.go
View File

@@ -36,7 +36,7 @@ func (u *Packet) Copy() *Packet {
return n
}
type Conn struct {
type TesterConn struct {
Addr *Addr
RxPackets chan *Packet // Packets to receive into nebula
@@ -45,8 +45,8 @@ type Conn struct {
l *logrus.Logger
}
func NewListener(l *logrus.Logger, ip net.IP, port int, _ bool, _ int) (*Conn, error) {
return &Conn{
func NewListener(l *logrus.Logger, ip net.IP, port int, _ bool, _ int) (Conn, error) {
return &TesterConn{
Addr: &Addr{ip, uint16(port)},
RxPackets: make(chan *Packet, 10),
TxPackets: make(chan *Packet, 10),
@@ -57,7 +57,7 @@ func NewListener(l *logrus.Logger, ip net.IP, port int, _ bool, _ int) (*Conn, e
// Send will place a UdpPacket onto the receive queue for nebula to consume
// 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 *Conn) Send(packet *Packet) {
func (u *TesterConn) Send(packet *Packet) {
h := &header.H{}
if err := h.Parse(packet.Data); err != nil {
panic(err)
@@ -74,7 +74,7 @@ func (u *Conn) Send(packet *Packet) {
// Get will pull a UdpPacket from the transmit queue
// nebula meant to send this message on the network, it will be encrypted
// packets were ingested from the tun side (in most cases), you can send them with Tun.Send
func (u *Conn) Get(block bool) *Packet {
func (u *TesterConn) Get(block bool) *Packet {
if block {
return <-u.TxPackets
}
@@ -91,7 +91,7 @@ func (u *Conn) Get(block bool) *Packet {
// Below this is boilerplate implementation to make nebula actually work
//********************************************************************************************************************//
func (u *Conn) WriteTo(b []byte, addr *Addr) error {
func (u *TesterConn) WriteTo(b []byte, addr *Addr) error {
p := &Packet{
Data: make([]byte, len(b), len(b)),
FromIp: make([]byte, 16),
@@ -108,7 +108,7 @@ func (u *Conn) WriteTo(b []byte, addr *Addr) error {
return nil
}
func (u *Conn) ListenOut(r EncReader, lhf LightHouseHandlerFunc, cache *firewall.ConntrackCacheTicker, q int) {
func (u *TesterConn) ListenOut(r EncReader, lhf LightHouseHandlerFunc, cache *firewall.ConntrackCacheTicker, q int) {
plaintext := make([]byte, MTU)
h := &header.H{}
fwPacket := &firewall.Packet{}
@@ -126,17 +126,17 @@ func (u *Conn) ListenOut(r EncReader, lhf LightHouseHandlerFunc, cache *firewall
}
}
func (u *Conn) ReloadConfig(*config.C) {}
func (u *TesterConn) ReloadConfig(*config.C) {}
func NewUDPStatsEmitter(_ []*Conn) func() {
func NewUDPStatsEmitter(_ []Conn) func() {
// No UDP stats for non-linux
return func() {}
}
func (u *Conn) LocalAddr() (*Addr, error) {
func (u *TesterConn) LocalAddr() (*Addr, error) {
return u.Addr, nil
}
func (u *Conn) Rebind() error {
func (u *TesterConn) Rebind() error {
return nil
}

2
udp/udp_windows.go
View File

@@ -24,6 +24,6 @@ func NewListenConfig(multi bool) net.ListenConfig {
}
}
func (u *Conn) Rebind() error {
func (u *GenericConn) Rebind() error {
return nil
}