mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-21 21:09:07 +02:00
[client] pqkem: run the data-path transport over a net.PacketConn
Preparation for running the ML-KEM exchange on the js/wasm client. There, and in other netstack-mode clients, the overlay IP exists only inside the userspace stack, so the KEM socket has to be opened on the netstack. The netstack returns a *gonet.UDPConn rather than a *net.UDPConn, so the transport can no longer hold the host socket's concrete type. Hold the socket as a net.PacketConn, which both implement, and convert endpoints at the edges. Sources are unmapped explicitly, since a udp4 socket's net.UDPAddr may carry the 16-byte form. No behaviour change: the transport still binds on the host, and the next commit opens it on the netstack. Adds a host round-trip test that also covers the fallback to an ephemeral port.
This commit is contained in:
@@ -19,7 +19,7 @@ const DefaultPort = 51833
|
||||
// overlay IP. It implements pqkem.Transport — the manager owns the remoteID<->endpoint
|
||||
// routing and drives this socket's lifecycle (Run / Close).
|
||||
type pqTransport struct {
|
||||
conn *net.UDPConn
|
||||
conn net.PacketConn
|
||||
port int
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
@@ -36,21 +36,32 @@ func newPQTransport(overlayIP netip.Addr) (*pqTransport, error) {
|
||||
// so the transport binds over IPv4. Unmap first so AsSlice() yields 4 bytes for an
|
||||
// IPv4-mapped IPv6 address (a hardcoded "udp4" would otherwise fail on its 16 bytes).
|
||||
overlayIP = overlayIP.Unmap()
|
||||
ip := net.IP(overlayIP.AsSlice())
|
||||
conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: ip, Port: DefaultPort})
|
||||
listen := func(port uint16) (net.PacketConn, error) {
|
||||
conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IP(overlayIP.AsSlice()), Port: int(port)})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return conn, nil
|
||||
}
|
||||
conn, err := listen(DefaultPort)
|
||||
if err != nil {
|
||||
log.Debugf("pqkem: default port %d unavailable on %s (%v), using an ephemeral port", DefaultPort, overlayIP, err)
|
||||
conn, err = net.ListenUDP("udp4", &net.UDPAddr{IP: ip, Port: 0})
|
||||
conn, err = listen(0)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("bind pqkem udp on overlay %s: %w", overlayIP, err)
|
||||
}
|
||||
}
|
||||
return &pqTransport{conn: conn, port: conn.LocalAddr().(*net.UDPAddr).Port}, nil
|
||||
local, ok := conn.LocalAddr().(*net.UDPAddr)
|
||||
if !ok {
|
||||
_ = conn.Close()
|
||||
return nil, fmt.Errorf("pqkem transport bound to a non-UDP address %s", conn.LocalAddr())
|
||||
}
|
||||
return &pqTransport{conn: conn, port: local.Port}, nil
|
||||
}
|
||||
|
||||
// Send implements pqkem.Transport.
|
||||
func (t *pqTransport) Send(endpoint netip.AddrPort, msg []byte) error {
|
||||
_, err := t.conn.WriteToUDPAddrPort(msg, endpoint)
|
||||
_, err := t.conn.WriteTo(msg, net.UDPAddrFromAddrPort(endpoint))
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -65,13 +76,18 @@ func (t *pqTransport) Run(onInbound func(src netip.AddrPort, msg []byte)) {
|
||||
defer t.wg.Done()
|
||||
buf := make([]byte, 2048)
|
||||
for {
|
||||
n, src, err := t.conn.ReadFromUDPAddrPort(buf)
|
||||
n, addr, err := t.conn.ReadFrom(buf)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
src, ok := addr.(*net.UDPAddr)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
ap := src.AddrPort()
|
||||
msg := make([]byte, n)
|
||||
copy(msg, buf[:n])
|
||||
onInbound(src, msg)
|
||||
onInbound(netip.AddrPortFrom(ap.Addr().Unmap(), ap.Port()), msg)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
package internal
|
||||
|
||||
import (
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type pqInbound struct {
|
||||
src netip.AddrPort
|
||||
msg []byte
|
||||
}
|
||||
|
||||
func runPQTransport(t *testing.T, tr *pqTransport) <-chan pqInbound {
|
||||
t.Helper()
|
||||
got := make(chan pqInbound, 1)
|
||||
tr.Run(func(src netip.AddrPort, msg []byte) {
|
||||
select {
|
||||
case got <- pqInbound{src: src, msg: msg}:
|
||||
default:
|
||||
}
|
||||
})
|
||||
t.Cleanup(func() { _ = tr.Close() })
|
||||
return got
|
||||
}
|
||||
|
||||
func receivePQ(t *testing.T, got <-chan pqInbound) pqInbound {
|
||||
t.Helper()
|
||||
select {
|
||||
case in := <-got:
|
||||
return in
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("no datagram received")
|
||||
return pqInbound{}
|
||||
}
|
||||
}
|
||||
|
||||
// TestPQTransport_HostRoundTrip sends a datagram between two host-bound transports. The
|
||||
// second bind collides with the first on DefaultPort, so it also covers the fallback to
|
||||
// an ephemeral port. The source must come back as a plain (unmapped) IPv4 endpoint,
|
||||
// since the manager matches it against registered peer endpoints.
|
||||
func TestPQTransport_HostRoundTrip(t *testing.T) {
|
||||
loopback := netip.MustParseAddr("127.0.0.1")
|
||||
|
||||
a, err := newPQTransport(loopback)
|
||||
require.NoError(t, err)
|
||||
b, err := newPQTransport(loopback)
|
||||
require.NoError(t, err)
|
||||
require.NotEqual(t, a.LocalPort(), b.LocalPort(), "two transports on one IP must get distinct ports")
|
||||
|
||||
runPQTransport(t, a)
|
||||
gotB := runPQTransport(t, b)
|
||||
|
||||
require.NoError(t, a.Send(netip.AddrPortFrom(loopback, uint16(b.LocalPort())), []byte("offer")))
|
||||
|
||||
in := receivePQ(t, gotB)
|
||||
assert.Equal(t, []byte("offer"), in.msg, "payload must arrive intact")
|
||||
assert.Equal(t, netip.AddrPortFrom(loopback, uint16(a.LocalPort())), in.src, "source must be the sender's unmapped IPv4 endpoint")
|
||||
}
|
||||
Reference in New Issue
Block a user