132 lines
4.1 KiB
Go
132 lines
4.1 KiB
Go
package e2e
|
|
|
|
import (
|
|
crand "crypto/rand"
|
|
"fmt"
|
|
"net"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/iceBear67/redapricot/client"
|
|
"github.com/iceBear67/redapricot/client/wire"
|
|
)
|
|
|
|
// echoRounds pushes payload through the tunnel `rounds` times on one player
|
|
// connection, failing the test if any round stalls.
|
|
func echoRounds(t *testing.T, hubAddr string, rounds, size int) {
|
|
t.Helper()
|
|
fast := dialPlayer(t, hubAddr, "mc.local")
|
|
defer fast.Close()
|
|
payload := make([]byte, size)
|
|
for i := range payload {
|
|
payload[i] = byte(i*13 + 5)
|
|
}
|
|
for i := 0; i < rounds; i++ {
|
|
playerEcho(t, fast, payload)
|
|
}
|
|
}
|
|
|
|
// TestSlowPlayerDoesNotStallOthers: a player that stops reading while megabytes
|
|
// are echoed back to it must not stall another stream on the same worker
|
|
// connection (maxConn=1 forces sharing). Before per-stream flow control, the
|
|
// hub paused the whole worker socket once that player's write queue filled,
|
|
// freezing every other stream's downstream data.
|
|
func TestSlowPlayerDoesNotStallOthers(t *testing.T) {
|
|
const psk = "e2e-slowplayer"
|
|
port := freePort(t)
|
|
hubAddr := fmt.Sprintf("127.0.0.1:%d", port)
|
|
startHub(t, port, psk)
|
|
dest := newMockDest(t, modeEcho)
|
|
startClient(t, hubAddr, psk, 1, []client.Mapping{
|
|
{Pattern: "mc.local", Destination: dest.addr},
|
|
})
|
|
|
|
// Slow player: 4 MiB goes out, gets echoed back, and is never read. The
|
|
// write blocks once buffers fill; the error on test-end close is expected.
|
|
slow := dialPlayer(t, hubAddr, "mc.local")
|
|
defer slow.Close()
|
|
go func() {
|
|
_, _ = slow.Write(make([]byte, 4*1024*1024))
|
|
}()
|
|
|
|
// Let the slow stream jam: its flow-control window fills and stays full.
|
|
time.Sleep(1 * time.Second)
|
|
|
|
// The fast player shares the single worker conn and must still round-trip.
|
|
echoRounds(t, hubAddr, 10, 8*1024)
|
|
}
|
|
|
|
// TestSlowDestinationDoesNotStallOthers: a destination that never reads must
|
|
// only stall its own stream. Before the async delivery queue, the client wrote
|
|
// to destinations inline in the worker readLoop, so one blocked destination
|
|
// froze every stream on the connection.
|
|
func TestSlowDestinationDoesNotStallOthers(t *testing.T) {
|
|
const psk = "e2e-slowdest"
|
|
port := freePort(t)
|
|
hubAddr := fmt.Sprintf("127.0.0.1:%d", port)
|
|
startHub(t, port, psk)
|
|
dest := newMockDest(t, modeEcho)
|
|
hole := newMockDest(t, modeBlackhole)
|
|
startClient(t, hubAddr, psk, 1, []client.Mapping{
|
|
{Pattern: "mc.local", Destination: dest.addr},
|
|
{Pattern: "hole.local", Destination: hole.addr},
|
|
})
|
|
|
|
// This player's destination never reads: the client-side write jams after
|
|
// kernel buffers fill, with the stream's queue bounded by its window.
|
|
stuck := dialPlayer(t, hubAddr, "hole.local")
|
|
defer stuck.Close()
|
|
go func() {
|
|
_, _ = stuck.Write(make([]byte, 2*1024*1024))
|
|
}()
|
|
|
|
time.Sleep(1 * time.Second)
|
|
|
|
echoRounds(t, hubAddr, 10, 8*1024)
|
|
}
|
|
|
|
// TestLegacyClientRejected: per-stream flow control is mandatory. A client that
|
|
// performs the old session establishment — a Rekey message without the trailing
|
|
// feature flags — must be closed by the hub before SessionReady.
|
|
func TestLegacyClientRejected(t *testing.T) {
|
|
const psk = "e2e-legacyreject"
|
|
port := freePort(t)
|
|
hubAddr := fmt.Sprintf("127.0.0.1:%d", port)
|
|
startHub(t, port, psk)
|
|
|
|
conn, err := net.DialTimeout("tcp", hubAddr, 5*time.Second)
|
|
if err != nil {
|
|
t.Fatalf("dial hub: %v", err)
|
|
}
|
|
defer conn.Close()
|
|
|
|
pskBytes := []byte(psk)
|
|
hs := wire.BuildHandshake(767, wire.PSKAddress(pskBytes), 25565, 17)
|
|
if _, err := conn.Write(hs); err != nil {
|
|
t.Fatalf("handshake: %v", err)
|
|
}
|
|
fc := wire.NewFramedConn(conn,
|
|
wire.CipherFor(pskBytes, wire.DirS2C),
|
|
wire.CipherFor(pskBytes, wire.DirC2S),
|
|
)
|
|
rnd := make([]byte, 16)
|
|
if _, err := crand.Read(rnd); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// Pre-flow-control Rekey: Magic|RandLen|Rand|Timestamp with no flags.
|
|
legacyRekey := wire.NewWriter().
|
|
U8(0x01). // control-session magic
|
|
VarInt(len(rnd)).
|
|
Bytes(rnd).
|
|
I64(time.Now().UnixMilli()).
|
|
Out()
|
|
if err := fc.WriteFrame(legacyRekey); err != nil {
|
|
t.Fatalf("rekey: %v", err)
|
|
}
|
|
|
|
_ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
|
|
if payload, err := fc.ReadFrame(); err == nil {
|
|
t.Fatalf("expected the hub to close a legacy session, got frame %v", payload)
|
|
}
|
|
}
|