From 8ac7eaf8337d02be1818654d6a5c65bd9f9cbd86 Mon Sep 17 00:00:00 2001 From: braginini Date: Mon, 27 Mar 2023 14:57:47 +0200 Subject: [PATCH] Bind init --- iface/bind/bind.go | 174 ++++++++++++++++++ iface/bind/udp_mux.go | 311 ++++++++++++++++++++++++++++++++ iface/bind/udp_mux_universal.go | 244 +++++++++++++++++++++++++ iface/bind/udp_muxed_conn.go | 243 +++++++++++++++++++++++++ iface/tun_unix.go | 2 + 5 files changed, 974 insertions(+) create mode 100644 iface/bind/bind.go create mode 100644 iface/bind/udp_mux.go create mode 100644 iface/bind/udp_mux_universal.go create mode 100644 iface/bind/udp_muxed_conn.go diff --git a/iface/bind/bind.go b/iface/bind/bind.go new file mode 100644 index 000000000..458f70f22 --- /dev/null +++ b/iface/bind/bind.go @@ -0,0 +1,174 @@ +package bind + +import ( + "errors" + "fmt" + "github.com/pion/stun" + log "github.com/sirupsen/logrus" + "golang.zx2c4.com/wireguard/conn" + "net" + "net/netip" + "sync" + "syscall" +) + +type ICEBind struct { + sharedConn net.PacketConn + udpMux *UniversalUDPMuxDefault + + mu sync.Mutex // protects following fields +} + +func (b *ICEBind) GetICEMux() (*UniversalUDPMuxDefault, error) { + b.mu.Lock() + defer b.mu.Unlock() + if b.udpMux == nil { + return nil, fmt.Errorf("ICEBind has not been initialized yet") + } + + return b.udpMux, nil +} + +func (b *ICEBind) Open(uport uint16) ([]conn.ReceiveFunc, uint16, error) { + b.mu.Lock() + defer b.mu.Unlock() + + if b.sharedConn != nil { + return nil, 0, conn.ErrBindAlreadyOpen + } + + port := int(uport) + ipv4Conn, port, err := listenNet("udp4", port) + if err != nil && !errors.Is(err, syscall.EAFNOSUPPORT) { + return nil, 0, err + } + b.sharedConn = ipv4Conn + b.udpMux = NewUniversalUDPMuxDefault(UniversalUDPMuxParams{UDPConn: b.sharedConn}) + + portAddr1, err := netip.ParseAddrPort(ipv4Conn.LocalAddr().String()) + if err != nil { + return nil, 0, err + } + + log.Infof("opened ICEBind on %s", ipv4Conn.LocalAddr().String()) + + return []conn.ReceiveFunc{ + b.makeReceiveIPv4(b.sharedConn), + }, + portAddr1.Port(), nil +} + +func listenNet(network string, port int) (*net.UDPConn, int, error) { + conn, err := net.ListenUDP(network, &net.UDPAddr{Port: port}) + if err != nil { + return nil, 0, err + } + + // Retrieve port. + laddr := conn.LocalAddr() + uaddr, err := net.ResolveUDPAddr( + laddr.Network(), + laddr.String(), + ) + if err != nil { + return nil, 0, err + } + return conn, uaddr.Port, nil +} + +func parseSTUNMessage(raw []byte) (*stun.Message, error) { + msg := &stun.Message{ + Raw: append([]byte{}, raw...), + } + if err := msg.Decode(); err != nil { + return nil, err + } + + return msg, nil +} + +func (b *ICEBind) makeReceiveIPv4(c net.PacketConn) conn.ReceiveFunc { + return func(buff []byte) (int, conn.Endpoint, error) { + n, endpoint, err := c.ReadFrom(buff) + if err != nil { + return 0, nil, err + } + e, err := netip.ParseAddrPort(endpoint.String()) + if err != nil { + return 0, nil, err + } + if !stun.IsMessage(buff[:20]) { + // WireGuard traffic + return n, (*conn.StdNetEndpoint)(&net.UDPAddr{ + IP: e.Addr().AsSlice(), + Port: int(e.Port()), + Zone: e.Addr().Zone(), + }), nil + } + + msg, err := parseSTUNMessage(buff[:n]) + if err != nil { + return 0, nil, err + } + + err = b.udpMux.HandleSTUNMessage(msg, endpoint) + if err != nil { + return 0, nil, err + } + if err != nil { + log.Warnf("failed to handle packet") + } + + // discard packets because they are STUN related + return 0, nil, nil //todo proper return + } +} + +func (b *ICEBind) Close() error { + b.mu.Lock() + defer b.mu.Unlock() + + var err1, err2 error + if b.sharedConn != nil { + c := b.sharedConn + b.sharedConn = nil + err1 = c.Close() + } + + if b.udpMux != nil { + m := b.udpMux + b.udpMux = nil + err2 = m.Close() + } + + if err1 != nil { + return err1 + } + + return err2 +} + +// SetMark sets the mark for each packet sent through this Bind. +// This mark is passed to the kernel as the socket option SO_MARK. +func (b *ICEBind) SetMark(mark uint32) error { + return nil +} + +func (b *ICEBind) Send(buff []byte, endpoint conn.Endpoint) error { + nend, ok := endpoint.(*conn.StdNetEndpoint) + if !ok { + return conn.ErrWrongEndpointType + } + _, err := b.sharedConn.WriteTo(buff, (*net.UDPAddr)(nend)) + return err +} + +// ParseEndpoint creates a new endpoint from a string. +func (b *ICEBind) ParseEndpoint(s string) (ep conn.Endpoint, err error) { + e, err := netip.ParseAddrPort(s) + return (*conn.StdNetEndpoint)(&net.UDPAddr{ + IP: e.Addr().AsSlice(), + Port: int(e.Port()), + Zone: e.Addr().Zone(), + }), err +} diff --git a/iface/bind/udp_mux.go b/iface/bind/udp_mux.go new file mode 100644 index 000000000..5b46c637f --- /dev/null +++ b/iface/bind/udp_mux.go @@ -0,0 +1,311 @@ +package bind + +import ( + "fmt" + "github.com/pion/stun" + log "github.com/sirupsen/logrus" + "io" + "net" + "strings" + "sync" + + "github.com/pion/logging" + "github.com/pion/transport/v2" +) + +const receiveMTU = 8192 + +// UDPMuxDefault is an implementation of the interface +type UDPMuxDefault struct { + params UDPMuxParams + + closedChan chan struct{} + closeOnce sync.Once + + // connsIPv4 and connsIPv6 are maps of all udpMuxedConn indexed by ufrag|network|candidateType + connsIPv4, connsIPv6 map[string]*udpMuxedConn + + addressMapMu sync.RWMutex + addressMap map[string][]*udpMuxedConn + + // buffer pool to recycle buffers for net.UDPAddr encodes/decodes + pool *sync.Pool + + mu sync.Mutex + + // for UDP connection listen at unspecified address + localAddrsForUnspecified []net.Addr +} + +const maxAddrSize = 512 + +// UDPMuxParams are parameters for UDPMux. +type UDPMuxParams struct { + Logger logging.LeveledLogger + UDPConn net.PacketConn + + // Required for gathering local addresses + // in case a un UDPConn is passed which does not + // bind to a specific local address. + Net transport.Net +} + +// NewUDPMuxDefault creates an implementation of UDPMux +func NewUDPMuxDefault(params UDPMuxParams) *UDPMuxDefault { + if params.Logger == nil { + params.Logger = logging.NewDefaultLoggerFactory().NewLogger("ice") + } + + return &UDPMuxDefault{ + addressMap: map[string][]*udpMuxedConn{}, + params: params, + connsIPv4: make(map[string]*udpMuxedConn), + connsIPv6: make(map[string]*udpMuxedConn), + closedChan: make(chan struct{}, 1), + pool: &sync.Pool{ + New: func() interface{} { + // big enough buffer to fit both packet and address + return newBufferHolder(receiveMTU + maxAddrSize) + }, + }, + localAddrsForUnspecified: []net.Addr{}, + } +} + +// LocalAddr returns the listening address of this UDPMuxDefault +func (m *UDPMuxDefault) LocalAddr() net.Addr { + return m.params.UDPConn.LocalAddr() +} + +// GetListenAddresses returns the list of addresses that this mux is listening on +func (m *UDPMuxDefault) GetListenAddresses() []net.Addr { + if len(m.localAddrsForUnspecified) > 0 { + return m.localAddrsForUnspecified + } + + return []net.Addr{m.LocalAddr()} +} + +// GetConn returns a PacketConn given the connection's ufrag and network address +// creates the connection if an existing one can't be found +func (m *UDPMuxDefault) GetConn(ufrag string, addr net.Addr) (net.PacketConn, error) { + + var isIPv6 bool + if udpAddr, _ := addr.(*net.UDPAddr); udpAddr != nil && udpAddr.IP.To4() == nil { + isIPv6 = true + } + m.mu.Lock() + defer m.mu.Unlock() + + if m.IsClosed() { + return nil, io.ErrClosedPipe + } + + if conn, ok := m.getConn(ufrag, isIPv6); ok { + return conn, nil + } + + c := m.createMuxedConn(ufrag) + go func() { + <-c.CloseChannel() + m.RemoveConnByUfrag(ufrag) + }() + + if isIPv6 { + m.connsIPv6[ufrag] = c + } else { + m.connsIPv4[ufrag] = c + } + + return c, nil +} + +// RemoveConnByUfrag stops and removes the muxed packet connection +func (m *UDPMuxDefault) RemoveConnByUfrag(ufrag string) { + removedConns := make([]*udpMuxedConn, 0, 2) + + // Keep lock section small to avoid deadlock with conn lock + m.mu.Lock() + if c, ok := m.connsIPv4[ufrag]; ok { + delete(m.connsIPv4, ufrag) + removedConns = append(removedConns, c) + } + if c, ok := m.connsIPv6[ufrag]; ok { + delete(m.connsIPv6, ufrag) + removedConns = append(removedConns, c) + } + m.mu.Unlock() + + if len(removedConns) == 0 { + // No need to lock if no connection was found + return + } + + m.addressMapMu.Lock() + defer m.addressMapMu.Unlock() + + for _, c := range removedConns { + addresses := c.getAddresses() + for _, addr := range addresses { + if connList, ok := m.addressMap[addr]; ok { + var newList []*udpMuxedConn + for _, conn := range connList { + if conn.params.Key != ufrag { + newList = append(newList, conn) + } + } + m.addressMap[addr] = newList + } + } + } +} + +// IsClosed returns true if the mux had been closed +func (m *UDPMuxDefault) IsClosed() bool { + select { + case <-m.closedChan: + return true + default: + return false + } +} + +// Close the mux, no further connections could be created +func (m *UDPMuxDefault) Close() error { + var err error + m.closeOnce.Do(func() { + m.mu.Lock() + defer m.mu.Unlock() + + for _, c := range m.connsIPv4 { + _ = c.Close() + } + for _, c := range m.connsIPv6 { + _ = c.Close() + } + + m.connsIPv4 = make(map[string]*udpMuxedConn) + m.connsIPv6 = make(map[string]*udpMuxedConn) + + close(m.closedChan) + + _ = m.params.UDPConn.Close() + }) + return err +} + +func (m *UDPMuxDefault) writeTo(buf []byte, rAddr net.Addr) (n int, err error) { + return m.params.UDPConn.WriteTo(buf, rAddr) +} + +func (m *UDPMuxDefault) registerConnForAddress(conn *udpMuxedConn, addr string) { + if m.IsClosed() { + return + } + + m.addressMapMu.Lock() + defer m.addressMapMu.Unlock() + + existing, ok := m.addressMap[addr] + if !ok { + existing = []*udpMuxedConn{} + } + existing = append(existing, conn) + m.addressMap[addr] = existing + + log.Debugf("ICE: registered %s for %s", addr, conn.params.Key) +} + +func (m *UDPMuxDefault) createMuxedConn(key string) *udpMuxedConn { + c := newUDPMuxedConn(&udpMuxedConnParams{ + Mux: m, + Key: key, + AddrPool: m.pool, + LocalAddr: m.LocalAddr(), + Logger: m.params.Logger, + }) + return c +} + +func (m *UDPMuxDefault) HandleSTUNMessage(msg *stun.Message, addr net.Addr) error { + + remoteAddr, ok := addr.(*net.UDPAddr) + if !ok { + return fmt.Errorf("underlying PacketConn did not return a UDPAddr") + } + + // If we have already seen this address dispatch to the appropriate destination + // If you are using the same socket for the Host and SRFLX candidates, it might be that there are more than one + // muxed connection - one for the SRFLX candidate and the other one for the HOST one. + // We will then forward STUN packets to each of these connections. + m.addressMapMu.Lock() + var destinationConnList []*udpMuxedConn + if storedConns, ok := m.addressMap[addr.String()]; ok { + for _, conn := range storedConns { + destinationConnList = append(destinationConnList, conn) + } + } + m.addressMapMu.Unlock() + + var isIPv6 bool + if udpAddr, _ := addr.(*net.UDPAddr); udpAddr != nil && udpAddr.IP.To4() == nil { + isIPv6 = true + } + + // This block is needed to discover Peer Reflexive Candidates for which we don't know the Endpoint upfront. + // However, we can take a username attribute from the STUN message which contains ufrag. + // We can use ufrag to identify the destination conn to route packet to. + attr, stunAttrErr := msg.Get(stun.AttrUsername) + if stunAttrErr == nil { + ufrag := strings.Split(string(attr), ":")[0] + + m.mu.Lock() + destinationConn := m.connsIPv4[ufrag] + if isIPv6 { + destinationConn = m.connsIPv6[ufrag] + } + + if destinationConn != nil { + exists := false + for _, conn := range destinationConnList { + if conn.params.Key == destinationConn.params.Key { + exists = true + break + } + } + if !exists { + destinationConnList = append(destinationConnList, destinationConn) + } + } + m.mu.Unlock() + } + + // Forward STUN packets to each destination connections even thought the STUN packet might not belong there. + // It will be discarded by the further ICE candidate logic if so. + for _, conn := range destinationConnList { + if err := conn.writePacket(msg.Raw, remoteAddr); err != nil { + log.Errorf("could not write packet: %v", err) + } + } + + return nil +} + +func (m *UDPMuxDefault) getConn(ufrag string, isIPv6 bool) (val *udpMuxedConn, ok bool) { + if isIPv6 { + val, ok = m.connsIPv6[ufrag] + } else { + val, ok = m.connsIPv4[ufrag] + } + return +} + +type bufferHolder struct { + buf []byte +} + +func newBufferHolder(size int) *bufferHolder { + return &bufferHolder{ + buf: make([]byte, size), + } +} diff --git a/iface/bind/udp_mux_universal.go b/iface/bind/udp_mux_universal.go new file mode 100644 index 000000000..f29210819 --- /dev/null +++ b/iface/bind/udp_mux_universal.go @@ -0,0 +1,244 @@ +package bind + +import ( + "fmt" + log "github.com/sirupsen/logrus" + "net" + "time" + + "github.com/pion/logging" + "github.com/pion/stun" + "github.com/pion/transport/v2" +) + +// UniversalUDPMuxDefault handles STUN and TURN servers packets by wrapping the original UDPConn +// It then passes packets to the UDPMux that does the actual connection muxing. +type UniversalUDPMuxDefault struct { + *UDPMuxDefault + params UniversalUDPMuxParams + + // since we have a shared socket, for srflx candidates it makes sense to have a shared mapped address across all the agents + // stun.XORMappedAddress indexed by the STUN server addr + xorMappedMap map[string]*xorMapped +} + +// UniversalUDPMuxParams are parameters for UniversalUDPMux server reflexive. +type UniversalUDPMuxParams struct { + Logger logging.LeveledLogger + UDPConn net.PacketConn + XORMappedAddrCacheTTL time.Duration + Net transport.Net +} + +// NewUniversalUDPMuxDefault creates an implementation of UniversalUDPMux embedding UDPMux +func NewUniversalUDPMuxDefault(params UniversalUDPMuxParams) *UniversalUDPMuxDefault { + if params.Logger == nil { + params.Logger = logging.NewDefaultLoggerFactory().NewLogger("ice") + } + if params.XORMappedAddrCacheTTL == 0 { + params.XORMappedAddrCacheTTL = time.Second * 25 + } + + m := &UniversalUDPMuxDefault{ + params: params, + xorMappedMap: make(map[string]*xorMapped), + } + + // wrap UDP connection, process server reflexive messages + // before they are passed to the UDPMux connection handler (connWorker) + m.params.UDPConn = &udpConn{ + PacketConn: params.UDPConn, + mux: m, + logger: params.Logger, + } + + // embed UDPMux + udpMuxParams := UDPMuxParams{ + Logger: params.Logger, + UDPConn: m.params.UDPConn, + Net: m.params.Net, + } + m.UDPMuxDefault = NewUDPMuxDefault(udpMuxParams) + + return m +} + +// udpConn is a wrapper around UDPMux conn that overrides ReadFrom and handles STUN/TURN packets +type udpConn struct { + net.PacketConn + mux *UniversalUDPMuxDefault + logger logging.LeveledLogger +} + +// GetRelayedAddr creates relayed connection to the given TURN service and returns the relayed addr. +// Not implemented yet. +func (m *UniversalUDPMuxDefault) GetRelayedAddr(turnAddr net.Addr, deadline time.Duration) (*net.Addr, error) { + return nil, fmt.Errorf("not implemented yet") +} + +// GetConnForURL add uniques to the muxed connection by concatenating ufrag and URL (e.g. STUN URL) to be able to support multiple STUN/TURN servers +// and return a unique connection per server. +func (m *UniversalUDPMuxDefault) GetConnForURL(ufrag string, url string, addr net.Addr) (net.PacketConn, error) { + return m.UDPMuxDefault.GetConn(fmt.Sprintf("%s%s", ufrag, url), addr) +} + +// HandleSTUNMessage discovers STUN packets that carry a XOR mapped address from a STUN server. +// All other STUN packets will be forwarded to the UDPMux +func (m *UniversalUDPMuxDefault) HandleSTUNMessage(msg *stun.Message, addr net.Addr) error { + + udpAddr, ok := addr.(*net.UDPAddr) + if !ok { + // message about this err will be logged in the UDPMux + return nil + } + + if m.isXORMappedResponse(msg, udpAddr.String()) { + err := m.handleXORMappedResponse(udpAddr, msg) + if err != nil { + log.Debugf("%s: %v", fmt.Errorf("failed to get XOR-MAPPED-ADDRESS response"), err) + return nil + } + return nil + } + return m.UDPMuxDefault.HandleSTUNMessage(msg, addr) +} + +// isXORMappedResponse indicates whether the message is a XORMappedAddress and is coming from the known STUN server. +func (m *UniversalUDPMuxDefault) isXORMappedResponse(msg *stun.Message, stunAddr string) bool { + m.mu.Lock() + defer m.mu.Unlock() + // check first if it is a STUN server address because remote peer can also send similar messages but as a BindingSuccess + _, ok := m.xorMappedMap[stunAddr] + _, err := msg.Get(stun.AttrXORMappedAddress) + return err == nil && ok +} + +// handleXORMappedResponse parses response from the STUN server, extracts XORMappedAddress attribute +// and set the mapped address for the server +func (m *UniversalUDPMuxDefault) handleXORMappedResponse(stunAddr *net.UDPAddr, msg *stun.Message) error { + m.mu.Lock() + defer m.mu.Unlock() + + mappedAddr, ok := m.xorMappedMap[stunAddr.String()] + if !ok { + return fmt.Errorf("no XOR address mapping") + } + + var addr stun.XORMappedAddress + if err := addr.GetFrom(msg); err != nil { + return err + } + + m.xorMappedMap[stunAddr.String()] = mappedAddr + mappedAddr.SetAddr(&addr) + + return nil +} + +// GetXORMappedAddr returns *stun.XORMappedAddress if already present for a given STUN server. +// Makes a STUN binding request to discover mapped address otherwise. +// Blocks until the stun.XORMappedAddress has been discovered or deadline. +// Method is safe for concurrent use. +func (m *UniversalUDPMuxDefault) GetXORMappedAddr(serverAddr net.Addr, deadline time.Duration) (*stun.XORMappedAddress, error) { + m.mu.Lock() + mappedAddr, ok := m.xorMappedMap[serverAddr.String()] + // if we already have a mapping for this STUN server (address already received) + // and if it is not too old we return it without making a new request to STUN server + if ok { + if mappedAddr.expired() { + mappedAddr.closeWaiters() + delete(m.xorMappedMap, serverAddr.String()) + ok = false + } else if mappedAddr.pending() { + ok = false + } + } + m.mu.Unlock() + if ok { + return mappedAddr.addr, nil + } + + // otherwise, make a STUN request to discover the address + // or wait for already sent request to complete + waitAddrReceived, err := m.sendStun(serverAddr) + if err != nil { + return nil, fmt.Errorf("%s: %s", "failed to send STUN packet", err) + } + + // block until response was handled by the connWorker routine and XORMappedAddress was updated + select { + case <-waitAddrReceived: + // when channel closed, addr was obtained + m.mu.Lock() + mappedAddr := *m.xorMappedMap[serverAddr.String()] + m.mu.Unlock() + if mappedAddr.addr == nil { + return nil, fmt.Errorf("no XOR address mapping") + } + return mappedAddr.addr, nil + case <-time.After(deadline): + return nil, fmt.Errorf("timeout while waiting for XORMappedAddr") + } +} + +// sendStun sends a STUN request via UDP conn. +// +// The returned channel is closed when the STUN response has been received. +// Method is safe for concurrent use. +func (m *UniversalUDPMuxDefault) sendStun(serverAddr net.Addr) (chan struct{}, error) { + m.mu.Lock() + defer m.mu.Unlock() + + // if record present in the map, we already sent a STUN request, + // just wait when waitAddrReceived will be closed + addrMap, ok := m.xorMappedMap[serverAddr.String()] + if !ok { + addrMap = &xorMapped{ + expiresAt: time.Now().Add(m.params.XORMappedAddrCacheTTL), + waitAddrReceived: make(chan struct{}), + } + m.xorMappedMap[serverAddr.String()] = addrMap + } + + req, err := stun.Build(stun.BindingRequest, stun.TransactionID) + if err != nil { + return nil, err + } + + if _, err = m.params.UDPConn.WriteTo(req.Raw, serverAddr); err != nil { + return nil, err + } + + return addrMap.waitAddrReceived, nil +} + +type xorMapped struct { + addr *stun.XORMappedAddress + waitAddrReceived chan struct{} + expiresAt time.Time +} + +func (a *xorMapped) closeWaiters() { + select { + case <-a.waitAddrReceived: + // notify was close, ok, that means we received duplicate response + // just exit + break + default: + // notify tha twe have a new addr + close(a.waitAddrReceived) + } +} + +func (a *xorMapped) pending() bool { + return a.addr == nil +} + +func (a *xorMapped) expired() bool { + return a.expiresAt.Before(time.Now()) +} + +func (a *xorMapped) SetAddr(addr *stun.XORMappedAddress) { + a.addr = addr + a.closeWaiters() +} diff --git a/iface/bind/udp_muxed_conn.go b/iface/bind/udp_muxed_conn.go new file mode 100644 index 000000000..6706ff07d --- /dev/null +++ b/iface/bind/udp_muxed_conn.go @@ -0,0 +1,243 @@ +package bind + +import ( + "encoding/binary" + "io" + "net" + "sync" + "time" + + "github.com/pion/logging" + "github.com/pion/transport/v2/packetio" +) + +type udpMuxedConnParams struct { + Mux *UDPMuxDefault + AddrPool *sync.Pool + Key string + LocalAddr net.Addr + Logger logging.LeveledLogger +} + +// udpMuxedConn represents a logical packet conn for a single remote as identified by ufrag +type udpMuxedConn struct { + params *udpMuxedConnParams + // remote addresses that we have sent to on this conn + addresses []string + + // channel holding incoming packets + buf *packetio.Buffer + closedChan chan struct{} + closeOnce sync.Once + mu sync.Mutex +} + +func newUDPMuxedConn(params *udpMuxedConnParams) *udpMuxedConn { + p := &udpMuxedConn{ + params: params, + buf: packetio.NewBuffer(), + closedChan: make(chan struct{}), + } + + return p +} + +func (c *udpMuxedConn) ReadFrom(b []byte) (n int, rAddr net.Addr, err error) { + buf := c.params.AddrPool.Get().(*bufferHolder) //nolint:forcetypeassert + defer c.params.AddrPool.Put(buf) + + // read address + total, err := c.buf.Read(buf.buf) + if err != nil { + return 0, nil, err + } + + dataLen := int(binary.LittleEndian.Uint16(buf.buf[:2])) + if dataLen > total || dataLen > len(b) { + return 0, nil, io.ErrShortBuffer + } + + // read data and then address + offset := 2 + copy(b, buf.buf[offset:offset+dataLen]) + offset += dataLen + + // read address len & decode address + addrLen := int(binary.LittleEndian.Uint16(buf.buf[offset : offset+2])) + offset += 2 + + if rAddr, err = decodeUDPAddr(buf.buf[offset : offset+addrLen]); err != nil { + return 0, nil, err + } + + return dataLen, rAddr, nil +} + +func (c *udpMuxedConn) WriteTo(buf []byte, rAddr net.Addr) (n int, err error) { + if c.isClosed() { + return 0, io.ErrClosedPipe + } + // each time we write to a new address, we'll register it with the mux + addr := rAddr.String() + if !c.containsAddress(addr) { + c.addAddress(addr) + } + + return c.params.Mux.writeTo(buf, rAddr) +} + +func (c *udpMuxedConn) LocalAddr() net.Addr { + return c.params.LocalAddr +} + +func (c *udpMuxedConn) SetDeadline(tm time.Time) error { + return nil +} + +func (c *udpMuxedConn) SetReadDeadline(tm time.Time) error { + return nil +} + +func (c *udpMuxedConn) SetWriteDeadline(tm time.Time) error { + return nil +} + +func (c *udpMuxedConn) CloseChannel() <-chan struct{} { + return c.closedChan +} + +func (c *udpMuxedConn) Close() error { + var err error + c.closeOnce.Do(func() { + err = c.buf.Close() + close(c.closedChan) + }) + return err +} + +func (c *udpMuxedConn) isClosed() bool { + select { + case <-c.closedChan: + return true + default: + return false + } +} + +func (c *udpMuxedConn) getAddresses() []string { + c.mu.Lock() + defer c.mu.Unlock() + addresses := make([]string, len(c.addresses)) + copy(addresses, c.addresses) + return addresses +} + +func (c *udpMuxedConn) addAddress(addr string) { + c.mu.Lock() + c.addresses = append(c.addresses, addr) + c.mu.Unlock() + + // map it on mux + c.params.Mux.registerConnForAddress(c, addr) +} + +func (c *udpMuxedConn) removeAddress(addr string) { + c.mu.Lock() + defer c.mu.Unlock() + + newAddresses := make([]string, 0, len(c.addresses)) + for _, a := range c.addresses { + if a != addr { + newAddresses = append(newAddresses, a) + } + } + + c.addresses = newAddresses +} + +func (c *udpMuxedConn) containsAddress(addr string) bool { + c.mu.Lock() + defer c.mu.Unlock() + for _, a := range c.addresses { + if addr == a { + return true + } + } + return false +} + +func (c *udpMuxedConn) writePacket(data []byte, addr *net.UDPAddr) error { + // write two packets, address and data + buf := c.params.AddrPool.Get().(*bufferHolder) //nolint:forcetypeassert + defer c.params.AddrPool.Put(buf) + + // format of buffer | data len | data bytes | addr len | addr bytes | + if len(buf.buf) < len(data)+maxAddrSize { + return io.ErrShortBuffer + } + // data len + binary.LittleEndian.PutUint16(buf.buf, uint16(len(data))) + offset := 2 + + // data + copy(buf.buf[offset:], data) + offset += len(data) + + // write address first, leaving room for its length + n, err := encodeUDPAddr(addr, buf.buf[offset+2:]) + if err != nil { + return err + } + total := offset + n + 2 + + // address len + binary.LittleEndian.PutUint16(buf.buf[offset:], uint16(n)) + + if _, err := c.buf.Write(buf.buf[:total]); err != nil { + return err + } + return nil +} + +func encodeUDPAddr(addr *net.UDPAddr, buf []byte) (int, error) { + ipData, err := addr.IP.MarshalText() + if err != nil { + return 0, err + } + total := 2 + len(ipData) + 2 + len(addr.Zone) + if total > len(buf) { + return 0, io.ErrShortBuffer + } + + binary.LittleEndian.PutUint16(buf, uint16(len(ipData))) + offset := 2 + n := copy(buf[offset:], ipData) + offset += n + binary.LittleEndian.PutUint16(buf[offset:], uint16(addr.Port)) + offset += 2 + copy(buf[offset:], addr.Zone) + return total, nil +} + +func decodeUDPAddr(buf []byte) (*net.UDPAddr, error) { + addr := net.UDPAddr{} + + offset := 0 + ipLen := int(binary.LittleEndian.Uint16(buf[:2])) + offset += 2 + // basic bounds checking + if ipLen+offset > len(buf) { + return nil, io.ErrShortBuffer + } + if err := addr.IP.UnmarshalText(buf[offset : offset+ipLen]); err != nil { + return nil, err + } + offset += ipLen + addr.Port = int(binary.LittleEndian.Uint16(buf[offset : offset+2])) + offset += 2 + zone := make([]byte, len(buf[offset:])) + copy(zone, buf[offset:]) + addr.Zone = string(zone) + + return &addr, nil +} diff --git a/iface/tun_unix.go b/iface/tun_unix.go index 2f38b5523..991e7c29b 100644 --- a/iface/tun_unix.go +++ b/iface/tun_unix.go @@ -3,6 +3,7 @@ package iface import ( + "github.com/netbirdio/netbird/iface/bind" "net" "os" @@ -18,6 +19,7 @@ type tunDevice struct { address WGAddress mtu int netInterface NetInterface + iceBind *bind.ICEBind } func newTunDevice(name string, address WGAddress, mtu int) *tunDevice {