Fix discovery packet validation and signaling edge cases

.gitignore

@@ -68,4 +68,6 @@ fabric.properties
 # Editor-based Rest Client
 .idea/httpRequests
 
-.vscode
+.vscode
+
+.clawpatch

conn.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"log/slog"
 	"math"
 	"math/rand/v2"
@@ -69,6 +68,9 @@ type Conn struct {
 	// channelsMu guards channels from concurrent read-write access during startup and closure.
 	channelsMu sync.RWMutex
 
+	readMu  sync.Mutex
+	readBuf []byte
+
 	// once ensures that the Conn is closed only once.
 	once sync.Once
 
@@ -90,13 +92,20 @@ type Conn struct {
 // Read receives a message from the 'ReliableDataChannel'. The bytes of the message data are copied to
 // the given data. An error may be returned if the Conn has been closed by [Conn.Close].
 func (conn *Conn) Read(b []byte) (n int, err error) {
-	pk, err := conn.Receive(MessageReliabilityReliable)
-	if err != nil {
-		return n, err
+	conn.readMu.Lock()
+	defer conn.readMu.Unlock()
+
+	if len(conn.readBuf) == 0 {
+		pk, err := conn.Receive(MessageReliabilityReliable)
+		if err != nil {
+			return n, err
+		}
+		conn.readBuf = pk
 	}
-	n = copy(b, pk)
-	if n < len(pk) {
-		return n, io.ErrShortBuffer
+	n = copy(b, conn.readBuf)
+	conn.readBuf = conn.readBuf[n:]
+	if len(conn.readBuf) == 0 {
+		conn.readBuf = nil
 	}
 	return n, nil
 }

conn_test.go

@@ -1,6 +1,7 @@
 package nethernet
 
 import (
+	"context"
 	"errors"
 	"net"
 	"testing"
@@ -25,3 +26,38 @@ func TestClosedWriteError(t *testing.T) {
 		}
 	})
 }
+
+func TestConnReadKeepsRemainderWhenBufferIsShort(t *testing.T) {
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(nil)
+
+	conn := &Conn{ctx: ctx}
+	packets := make(chan []byte, 1)
+	conn.storeChannel(MessageReliabilityReliable, &dataChannel{packets: packets})
+	packets <- []byte("hello")
+
+	b := make([]byte, 2)
+	n, err := conn.Read(b)
+	if err != nil {
+		t.Fatalf("first Read() error = %v, want nil", err)
+	}
+	if got := string(b[:n]); got != "he" {
+		t.Fatalf("first Read() = %q, want %q", got, "he")
+	}
+
+	n, err = conn.Read(b)
+	if err != nil {
+		t.Fatalf("second Read() error = %v, want nil", err)
+	}
+	if got := string(b[:n]); got != "ll" {
+		t.Fatalf("second Read() = %q, want %q", got, "ll")
+	}
+
+	n, err = conn.Read(b)
+	if err != nil {
+		t.Fatalf("third Read() error = %v, want nil", err)
+	}
+	if got := string(b[:n]); got != "o" {
+		t.Fatalf("third Read() = %q, want %q", got, "o")
+	}
+}

dial.go

@@ -9,6 +9,7 @@ import (
 	"net"
 	"strconv"
 	"sync"
+	"time"
 
 	"github.com/pion/sdp/v3"
 	"github.com/pion/webrtc/v4"
@@ -220,14 +221,20 @@ func (d dialerConn) log() *slog.Logger {
 // signalError sends a SignalTypeError to the remote connection using the
 // provided [Signaling] implementation, remote network ID, and error code.
 func (d Dialer) signalError(signaling Signaling, networkID string, code int) {
-	_ = signaling.Signal(context.Background(), &Signal{
-		Type:         SignalTypeError,
-		Data:         strconv.Itoa(code),
-		ConnectionID: d.ConnectionID,
-		NetworkID:    networkID,
-	})
+	go func() {
+		ctx, cancel := context.WithTimeout(signaling.Context(), signalErrorTimeout)
+		defer cancel()
+		_ = signaling.Signal(ctx, &Signal{
+			Type:         SignalTypeError,
+			Data:         strconv.Itoa(code),
+			ConnectionID: d.ConnectionID,
+			NetworkID:    networkID,
+		})
+	}()
 }
 
+const signalErrorTimeout = time.Second * 2
+
 // startTransports starts the ICE transport as [webrtc.ICERoleControlling],
 // then starts DTLS and SCTP using the parameters from the remote description.
 // After SCTP is established, it creates the 'ReliableDataChannel' and

dial_test.go

@@ -0,0 +1,86 @@
+package nethernet
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestDialContextDoesNotWaitIndefinitelyForErrorSignal(t *testing.T) {
+	signaling := newBlockingErrorSignaling("client")
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*20)
+	defer cancel()
+
+	done := make(chan error, 1)
+	go func() {
+		_, err := (Dialer{}).DialContext(ctx, "server", signaling)
+		done <- err
+	}()
+
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.DeadlineExceeded) {
+			t.Fatalf("DialContext() error = %v, want context deadline exceeded", err)
+		}
+	case <-time.After(time.Millisecond * 250):
+		t.Fatal("DialContext() did not return promptly after its context deadline")
+	}
+
+	select {
+	case <-signaling.errorSignalStarted:
+	case <-time.After(time.Second):
+		t.Fatal("DialContext() did not attempt to signal the timeout error")
+	}
+}
+
+type blockingErrorSignaling struct {
+	id string
+
+	ctx    context.Context
+	cancel context.CancelCauseFunc
+
+	once               sync.Once
+	errorSignalStarted chan struct{}
+}
+
+func newBlockingErrorSignaling(id string) *blockingErrorSignaling {
+	ctx, cancel := context.WithCancelCause(context.Background())
+	return &blockingErrorSignaling{
+		id:                 id,
+		ctx:                ctx,
+		cancel:             cancel,
+		errorSignalStarted: make(chan struct{}),
+	}
+}
+
+func (s *blockingErrorSignaling) Signal(ctx context.Context, signal *Signal) error {
+	if signal.Type != SignalTypeError {
+		return nil
+	}
+	s.once.Do(func() {
+		close(s.errorSignalStarted)
+	})
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+func (*blockingErrorSignaling) Notify(chan<- *Signal) func() {
+	return func() {}
+}
+
+func (s *blockingErrorSignaling) Context() context.Context {
+	return s.ctx
+}
+
+func (*blockingErrorSignaling) Credentials(context.Context) (*Credentials, error) {
+	return nil, nil
+}
+
+func (s *blockingErrorSignaling) NetworkID() string {
+	return s.id
+}
+
+func (*blockingErrorSignaling) PongData([]byte) {}

discovery/example_listener_test.go

@@ -0,0 +1,70 @@
+//go:build manual
+
+package discovery
+
+import (
+	"errors"
+	"log/slog"
+	"math/rand"
+	"net"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/df-mc/go-nethernet"
+)
+
+func TestListen(t *testing.T) {
+	cfg := ListenConfig{
+		NetworkID: rand.Uint64(),
+	}
+	d, err := cfg.Listen("0.0.0.0:7551")
+	if err != nil {
+		t.Fatalf("error listening on discovery: %s", err)
+	}
+	t.Cleanup(func() {
+		if err := d.Close(); err != nil {
+			t.Errorf("error closing discovery: %s", err)
+		}
+	})
+	d.ServerData(&ServerData{
+		ServerName:     "df-mc/go-nethernet",
+		LevelName:      "Bedrock World",
+		GameType:       2,
+		PlayerCount:    1,
+		MaxPlayerCount: 8,
+		TransportLayer: 2,
+		ConnectionType: 4,
+	})
+
+	slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
+		Level: slog.LevelDebug,
+	})))
+
+	var c nethernet.ListenConfig
+	l, err := c.Listen(d)
+	if err != nil {
+		t.Fatalf("error listening: %s", err)
+	}
+	t.Cleanup(func() {
+		if err := l.Close(); err != nil {
+			t.Fatalf("error closing: %s", err)
+		}
+	})
+
+	for {
+		conn, err := l.Accept()
+		if err != nil {
+			if !errors.Is(err, net.ErrClosed) {
+				t.Fatalf("error accepting connection: %s", err)
+			}
+			return
+		}
+		t.Logf("accepted: %s", conn.RemoteAddr())
+		time.AfterFunc(time.Second*5, func() {
+			if err := conn.Close(); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}

discovery/listener.go

@@ -8,7 +8,6 @@ import (
 	"maps"
 	"math/rand"
 	"net"
-	"net/netip"
 	"strconv"
 	"strings"
 	"sync"
@@ -42,17 +41,22 @@ func (conf ListenConfig) Listen(addr string) (*Listener, error) {
 	if conf.NetworkID == 0 {
 		conf.NetworkID = rand.Uint64()
 	}
-	addrPort, err := netip.ParseAddrPort(addr)
-	if err != nil {
-		return nil, fmt.Errorf("parse address: %w", err)
+	if addr == "" {
+		addr = ":0"
 	}
 	// We hardcode network protocol for "udp" as it always expects UDP packets to be received.
 	conn, err := net.ListenPacket("udp", addr)
 	if err != nil {
 		return nil, err
 	}
 
-	if conf.BroadcastAddress == nil && addrPort.Port() != DefaultPort {
+	localAddr, ok := conn.LocalAddr().(*net.UDPAddr)
+	if !ok {
+		_ = conn.Close()
+		return nil, fmt.Errorf("unexpected local address type %T", conn.LocalAddr())
+	}
+
+	if conf.BroadcastAddress == nil && localAddr.Port != DefaultPort {
 		// If the port for the address is 7551, it means no applications are listening on this network
 		// and server discovery using limited broadcast on net.IPv4bcast is not meaningful.
 		conf.BroadcastAddress = &net.UDPAddr{
@@ -289,10 +293,10 @@ func (l *Listener) handlePacket(data []byte, addr net.Addr) error {
 	if !ok {
 		a = address{
 			networkID: senderID,
-			addr:      addr,
 		}
 	}
-	// Update or set the timestamp for expiring them in deleteInactiveAddresses.
+	// Update or set the address and timestamp for expiring them in deleteInactiveAddresses.
+	a.addr = addr
 	a.t = time.Now()
 	l.addresses[senderID] = a
 	l.addressesMu.Unlock()
@@ -369,7 +373,13 @@ func (l *Listener) handleMessage(pk *MessagePacket, senderID uint64) error {
 // ServerData stores the ServerData for responding to the clients broadcasting
 // RequestPacket with a ResponsePacket containing the binary representation.
 func (l *Listener) ServerData(d *ServerData) {
-	b, _ := d.MarshalBinary()
+	b, err := d.MarshalBinary()
+	if err != nil {
+		if l.conf.Log != nil {
+			l.conf.Log.Error("error marshaling server data", slog.Any("error", err))
+		}
+		return
+	}
 	l.pongData.Store(&b)
 }