diff --git a/proxy/shadowsocks_2022/relay_test.go b/proxy/shadowsocks_2022/relay_test.go index ccf870d1b..096852ca6 100644 --- a/proxy/shadowsocks_2022/relay_test.go +++ b/proxy/shadowsocks_2022/relay_test.go @@ -374,98 +374,3 @@ func TestRelayTCPHandshakeForwarding(t *testing.T) { }) } } - -// SIP022 ยง3.2.4: a relay session is routed by client session ID, not by source address. -// After the client moves to a new source address, the inbound worker of the old address -// eventually expires and its context is canceled; the session must survive that. -func TestUDPSessionSurvivesOldSourceExpiry(t *testing.T) { - method, err := GetCipherMethod(MethodAES128GCM) - common.Must(err) - key := make([]byte, method.KeySaltLength) - _, _ = rand.Read(key) - - inbound, err := NewServer(newTestContext(), &ServerConfig{ - Method: MethodAES128GCM, - Key: base64.StdEncoding.EncodeToString(key), - }) - common.Must(err) - - var dispatchCount atomic.Int32 - var received atomic.Int32 - disp := &dummyDispatcher{ - onDispatch: func(ctx context.Context, d net.Destination) (*transport.Link, error) { - dispatchCount.Add(1) - linkR, linkW := gonet.Pipe() - // Like a real outbound, end the link when the dispatch context is canceled. - go func() { - <-ctx.Done() - linkW.Close() - linkR.Close() - }() - t.Cleanup(func() { - linkW.Close() - linkR.Close() - }) - return &transport.Link{ - Reader: buf.NewReader(linkR), - Writer: &customWriter{write: func(mb buf.MultiBuffer) error { - received.Add(int32(len(mb))) - buf.ReleaseMulti(mb) - return nil - }}, - }, nil - }, - } - - clientCodec, err := NewUDPPacketCodec(method, [][]byte{key}) - common.Must(err) - session, err := clientCodec.NewClientSession() - common.Must(err) - dest := net.UDPDestination(net.LocalHostIP, 443) - send := func(conn gonet.Conn) { - pkt, err := session.EncodePacket(dest, []byte("payload")) - common.Must(err) - defer pkt.Release() - if _, err := conn.Write(pkt.Bytes()); err != nil { - t.Fatal(err) - } - } - process := func(ctx context.Context) gonet.Conn { - clientConn, serverConn := gonet.Pipe() - t.Cleanup(func() { - clientConn.Close() - serverConn.Close() - }) - go func() { - _ = inbound.Process(ctx, net.Network_UDP, &dummyStatConn{Conn: serverConn}, disp) - }() - return clientConn - } - - ctxA, cancelA := context.WithCancel(newTestContext()) - defer cancelA() - connA := process(ctxA) - send(connA) - - // The client's source address changes. - ctxB, cancelB := context.WithCancel(newTestContext()) - defer cancelB() - connB := process(ctxB) - send(connB) - time.Sleep(50 * time.Millisecond) - - // The inbound worker of the old address expires. - cancelA() - connA.Close() - time.Sleep(50 * time.Millisecond) - - send(connB) - time.Sleep(50 * time.Millisecond) - - if n := dispatchCount.Load(); n != 1 { - t.Fatalf("session was dispatched %d times, want 1", n) - } - if n := received.Load(); n != 3 { - t.Fatalf("received %d packets, want 3", n) - } -} diff --git a/proxy/shadowsocks_2022/shadowsocks_2022_test.go b/proxy/shadowsocks_2022/shadowsocks_2022_test.go index b7a600194..382696efb 100644 --- a/proxy/shadowsocks_2022/shadowsocks_2022_test.go +++ b/proxy/shadowsocks_2022/shadowsocks_2022_test.go @@ -496,60 +496,3 @@ func TestClientUDPSessionMultiDestination(t *testing.T) { }) } } - -// Packets of one client session may be decoded concurrently, e.g. by the inbound -// workers of the client's old and new source address after a NAT rebinding. -func TestUDPServerConcurrentDecodeSameSession(t *testing.T) { - for _, methodName := range []string{MethodAES128GCM, MethodAES256GCM, MethodChaCha20Poly1305} { - t.Run(methodName, func(t *testing.T) { - method, err := GetCipherMethod(methodName) - common.Must(err) - rawKey := make([]byte, method.KeySaltLength) - _, _ = rand.Read(rawKey) - - clientCodec, err := NewUDPPacketCodec(method, [][]byte{rawKey}) - common.Must(err) - serverCodec, err := NewUDPServerCodec(method, rawKey, time.Minute) - common.Must(err) - session, err := clientCodec.NewClientSession() - common.Must(err) - - dest := net.UDPDestination(net.LocalHostIP, net.Port(443)) - packets := make([][]byte, 256) - for i := range packets { - pkt, err := session.EncodePacket(dest, []byte("payload")) - common.Must(err) - packets[i] = append([]byte(nil), pkt.Bytes()...) - pkt.Release() - } - - const workers = 4 - var wg sync.WaitGroup - var mu sync.Mutex - accepted := make([]int, len(packets)) - for w := 0; w < workers; w++ { - wg.Add(1) - go func() { - defer wg.Done() - // Every worker sees every packet, so each packet must be accepted exactly once. - for i, pkt := range packets { - if _, err := serverCodec.DecodePacket(pkt); err == nil { - mu.Lock() - accepted[i]++ - mu.Unlock() - } else if err != ErrPacketIdNotUnique { - t.Errorf("packet %d: unexpected error: %v", i, err) - } - } - }() - } - wg.Wait() - - for i, n := range accepted { - if n != 1 { - t.Fatalf("packet %d accepted %d times, want 1", i, n) - } - } - }) - } -}