diff --git a/.gitignore b/.gitignore index d957fc9..37b8144 100644 --- a/.gitignore +++ b/.gitignore @@ -68,4 +68,6 @@ fabric.properties # Editor-based Rest Client .idea/httpRequests -.vscode \ No newline at end of file +.vscode + +.clawpatch \ No newline at end of file diff --git a/conn.go b/conn.go index 27f3026..7cd233f 100644 --- a/conn.go +++ b/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 } diff --git a/conn_test.go b/conn_test.go index 93bd81a..8ea8613 100644 --- a/conn_test.go +++ b/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") + } +} diff --git a/dial.go b/dial.go index 5923dac..6075f36 100644 --- a/dial.go +++ b/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 diff --git a/dial_test.go b/dial_test.go new file mode 100644 index 0000000..e42972e --- /dev/null +++ b/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) {} diff --git a/discovery/example_listener_test.go b/discovery/example_listener_test.go new file mode 100644 index 0000000..78652ee --- /dev/null +++ b/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) + } + }) + } +} diff --git a/discovery/listener.go b/discovery/listener.go index 60843be..70f0e51 100644 --- a/discovery/listener.go +++ b/discovery/listener.go @@ -8,7 +8,6 @@ import ( "maps" "math/rand" "net" - "net/netip" "strconv" "strings" "sync" @@ -42,9 +41,8 @@ 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) @@ -52,7 +50,13 @@ func (conf ListenConfig) Listen(addr string) (*Listener, error) { 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) } diff --git a/discovery/listener_test.go b/discovery/listener_test.go index 78652ee..cd1ba33 100644 --- a/discovery/listener_test.go +++ b/discovery/listener_test.go @@ -1,70 +1,71 @@ -//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) +func TestListenAcceptsClientAddressForms(t *testing.T) { + for _, tt := range []struct { + name string + addr string + }{ + {name: "empty", addr: ""}, + {name: "portZero", addr: ":0"}, + } { + t.Run(tt.name, func(t *testing.T) { + l, err := (ListenConfig{}).Listen(tt.addr) + if err != nil { + t.Fatalf("Listen(%q): %v", tt.addr, err) + } + t.Cleanup(func() { + if err := l.Close(); err != nil { + t.Errorf("Close: %v", err) + } + }) + + localAddr, ok := l.conn.LocalAddr().(*net.UDPAddr) + if !ok { + t.Fatalf("local address = %T, want *net.UDPAddr", l.conn.LocalAddr()) + } + if localAddr.Port == DefaultPort { + t.Skipf("port-zero bind selected default port %d", DefaultPort) + } + if l.conf.BroadcastAddress == nil { + t.Fatal("BroadcastAddress is nil") + } + if !l.conf.BroadcastAddress.IP.Equal(net.IPv4bcast) { + t.Fatalf("BroadcastAddress.IP = %v, want %v", l.conf.BroadcastAddress.IP, net.IPv4bcast) + } + if l.conf.BroadcastAddress.Port != DefaultPort { + t.Fatalf("BroadcastAddress.Port = %d, want %d", l.conf.BroadcastAddress.Port, DefaultPort) + } + }) } - 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, - }))) +func TestHandlePacketUpdatesAddressForKnownSender(t *testing.T) { + const ( + localID uint64 = 1 + senderID uint64 = 2 + ) + l := &Listener{ + conf: ListenConfig{NetworkID: localID}, + addresses: make(map[uint64]address), + } + packet := Marshal(&MessagePacket{RecipientID: localID, Data: "Ping"}, senderID) + firstAddr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 19132} + secondAddr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 19133} - var c nethernet.ListenConfig - l, err := c.Listen(d) - if err != nil { - t.Fatalf("error listening: %s", err) + if err := l.handlePacket(packet, firstAddr); err != nil { + t.Fatalf("handlePacket(firstAddr): %v", err) + } + if err := l.handlePacket(packet, secondAddr); err != nil { + t.Fatalf("handlePacket(secondAddr): %v", 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) - } - }) + got := l.addresses[senderID] + if got.addr != secondAddr { + t.Fatalf("cached addr = %v, want %v", got.addr, secondAddr) } } diff --git a/discovery/packet.go b/discovery/packet.go index aa58ec0..ac3a581 100644 --- a/discovery/packet.go +++ b/discovery/packet.go @@ -67,6 +67,9 @@ func Unmarshal(b []byte) (Packet, uint64, error) { if err := binary.Read(buf, binary.LittleEndian, &length); err != nil { return nil, 0, fmt.Errorf("read length: %w", err) } + if remaining := buf.Len(); int(length) != remaining { + return nil, 0, fmt.Errorf("invalid packet length: %d, remaining %d", length, remaining) + } h := &Header{} if err := h.Read(buf); err != nil { return nil, 0, fmt.Errorf("read header: %w", err) @@ -98,11 +101,16 @@ func readBytes[L ~uint32 | ~uint8](r io.Reader) ([]byte, error) { if err := binary.Read(r, binary.LittleEndian, &length); err != nil { return nil, fmt.Errorf("read length: %w", err) } - b := make([]byte, length) - if n, err := r.Read(b); err != nil { + n := uint32(length) + if n > maxPacketPayloadLength { + return nil, fmt.Errorf("invalid length: %d, max %d", n, maxPacketPayloadLength) + } + if l, ok := r.(interface{ Len() int }); ok && n > uint32(l.Len()) { + return nil, fmt.Errorf("invalid length: %d, remaining %d", n, l.Len()) + } + b := make([]byte, n) + if _, err := io.ReadFull(r, b); err != nil { return nil, err - } else if n != int(length) { - return nil, fmt.Errorf("invalid length: %d, expected %d", n, length) } return b, nil } @@ -114,6 +122,9 @@ func writeBytes[L ~uint32 | ~uint8](w io.Writer, b []byte) { } const ( + // maxPacketPayloadLength is 65,535 bytes, matching the uint16 length prefix. + maxPacketPayloadLength = 1<<16 - 1 + IDRequestPacket uint16 = iota IDResponsePacket IDMessagePacket diff --git a/discovery/packet_test.go b/discovery/packet_test.go new file mode 100644 index 0000000..5b78ba2 --- /dev/null +++ b/discovery/packet_test.go @@ -0,0 +1,74 @@ +package discovery + +import ( + "bytes" + "crypto/hmac" + "crypto/sha256" + "encoding/binary" + "strings" + "testing" +) + +func TestUnmarshalRejectsInvalidPacketLength(t *testing.T) { + body := &bytes.Buffer{} + (&Header{PacketID: IDRequestPacket, SenderID: 1}).Write(body) + + _, _, err := Unmarshal(sealPayload(append( + binary.LittleEndian.AppendUint16(nil, uint16(body.Len()+1)), + body.Bytes()..., + ))) + if err == nil || !strings.Contains(err.Error(), "invalid packet length") { + t.Fatalf("Unmarshal() error = %v, want invalid packet length", err) + } +} + +func TestUnmarshalRejectsOversizedNestedLengths(t *testing.T) { + const oversizedLength = uint32(maxPacketPayloadLength) + + tests := []struct { + name string + packetID uint16 + write func(*bytes.Buffer) + }{ + { + name: "message data", + packetID: IDMessagePacket, + write: func(body *bytes.Buffer) { + _ = binary.Write(body, binary.LittleEndian, uint64(2)) + _ = binary.Write(body, binary.LittleEndian, oversizedLength) + }, + }, + { + name: "response application data", + packetID: IDResponsePacket, + write: func(body *bytes.Buffer) { + _ = binary.Write(body, binary.LittleEndian, oversizedLength) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, _, err := Unmarshal(rawPacket(tt.packetID, tt.write)) + if err == nil || !strings.Contains(err.Error(), "invalid length") { + t.Fatalf("Unmarshal() error = %v, want invalid length", err) + } + }) + } +} + +func rawPacket(packetID uint16, write func(*bytes.Buffer)) []byte { + body := &bytes.Buffer{} + (&Header{PacketID: packetID, SenderID: 1}).Write(body) + write(body) + return sealPayload(append( + binary.LittleEndian.AppendUint16(nil, uint16(body.Len())), + body.Bytes()..., + )) +} + +func sealPayload(payload []byte) []byte { + hash := hmac.New(sha256.New, key[:]) + hash.Write(payload) + return append(hash.Sum(nil), encrypt(payload)...) +} diff --git a/discovery/server_data.go b/discovery/server_data.go index 8ede236..5106728 100644 --- a/discovery/server_data.go +++ b/discovery/server_data.go @@ -44,10 +44,18 @@ type ServerData struct { // MarshalBinary ... func (d *ServerData) MarshalBinary() ([]byte, error) { buf := &bytes.Buffer{} + serverName := []byte(d.ServerName) + if len(serverName) > maxServerDataNameLength { + return nil, fmt.Errorf("server name length %d exceeds %d byte limit", len(serverName), maxServerDataNameLength) + } + levelName := []byte(d.LevelName) + if len(levelName) > maxServerDataNameLength { + return nil, fmt.Errorf("level name length %d exceeds %d byte limit", len(levelName), maxServerDataNameLength) + } _ = binary.Write(buf, binary.LittleEndian, version) - writeBytes[uint8](buf, []byte(d.ServerName)) - writeBytes[uint8](buf, []byte(d.LevelName)) + writeBytes[uint8](buf, serverName) + writeBytes[uint8](buf, levelName) _ = binary.Write(buf, binary.LittleEndian, d.GameType<<1) _ = binary.Write(buf, binary.LittleEndian, d.PlayerCount) _ = binary.Write(buf, binary.LittleEndian, d.MaxPlayerCount) @@ -116,3 +124,6 @@ func (d *ServerData) UnmarshalBinary(data []byte) error { // version is the current version of ServerData as supported by the `discovery` package. const version uint8 = 4 + +// maxServerDataNameLength is 255 bytes, matching the uint8 string length prefix. +const maxServerDataNameLength = 1<<8 - 1 diff --git a/discovery/server_data_test.go b/discovery/server_data_test.go new file mode 100644 index 0000000..9f609b5 --- /dev/null +++ b/discovery/server_data_test.go @@ -0,0 +1,42 @@ +package discovery + +import ( + "strings" + "testing" +) + +func TestServerDataMarshalBinaryRejectsOverlongNames(t *testing.T) { + tests := []struct { + name string + data *ServerData + }{ + { + name: "server name", + data: testServerData(strings.Repeat("s", maxServerDataNameLength+1), "world"), + }, + { + name: "level name", + data: testServerData("server", strings.Repeat("l", maxServerDataNameLength+1)), + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if _, err := tt.data.MarshalBinary(); err == nil { + t.Fatal("MarshalBinary() error = nil, want overlong name error") + } + }) + } +} + +func testServerData(serverName, levelName string) *ServerData { + return &ServerData{ + ServerName: serverName, + LevelName: levelName, + GameType: 2, + PlayerCount: 1, + MaxPlayerCount: 8, + TransportLayer: 2, + ConnectionType: 4, + } +}