diff --git a/internal/tuic/server.go b/internal/tuic/server.go index 4c0ff5e05..f1622edf1 100644 --- a/internal/tuic/server.go +++ b/internal/tuic/server.go @@ -922,6 +922,11 @@ func (s *Server) relayUDPResponses( sess *udpRelaySession, ) { defer associations.release(assocID, association) + ctx, stop := context.WithCancel(ctx) + defer stop() + if !associations.attachResponseStop(assocID, association, stop) { + return + } buf := make([]byte, s.maxUdpRelayPacketSize+263) var nextPktID uint16 for { diff --git a/internal/tuic/stream_udp_lifecycle_test.go b/internal/tuic/stream_udp_lifecycle_test.go index 5c447b00c..87b9d3496 100644 --- a/internal/tuic/stream_udp_lifecycle_test.go +++ b/internal/tuic/stream_udp_lifecycle_test.go @@ -210,3 +210,45 @@ func TestUdpAssociationTouchDoesNotRefreshReusedID(t *testing.T) { t.Fatal("replacement association timestamp changed after stale touch") } } + +// A response relay blocked on stream credit the client never grants must still end when +// its association is dissociated; otherwise each Packet/Dissociate round leaves one behind. +func TestDissociateStopsAResponseRelayBlockedOnTheClient(t *testing.T) { + socksAddr, cleanup := startMockSocks5Server(t, "udp-dissociate@x", "lifecycle-socks-password") + defer cleanup() + server, _, clientID, _ := startLifecycleTestServer(t, socksAddr, "udp-dissociate@x") + serverConn, user := authenticatedServerConnection(t, server, clientID) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + session, err := server.relay.DialUDP(ctx, user.Email) + if err != nil { + t.Fatalf("open SOCKS UDP session: %v", err) + } + const associationID uint16 = 61245 + association := &udpAssociation{ + responseTransport: packetTransportStream, + relay: &udpRelaySession{relay: session, responseTransport: packetTransportStream}, + lastActive: time.Now(), + } + registry := newUdpAssociationRegistry(maxUdpRelayPacketSize) + registry.associations[associationID] = association + responseDone := make(chan struct{}) + go func() { + defer close(responseDone) + server.relayUDPResponses(ctx, serverConn, user, associationID, association, registry, association.relay) + }() + target := &Address{Type: AddrTypeIPv4, IP: net.IPv4(8, 8, 8, 8), Port: 53} + for range 150 { + if _, err := session.Send(target, []byte("echo")); err != nil { + t.Fatalf("send SOCKS datagram: %v", err) + } + } + time.Sleep(300 * time.Millisecond) + registry.dissociate(associationID) + select { + case <-responseDone: + case <-time.After(2 * time.Second): + t.Fatal("the response relay is still running after its association was dissociated") + } +} diff --git a/internal/tuic/udp_associations.go b/internal/tuic/udp_associations.go index 9e7de4e7b..98d62e36d 100644 --- a/internal/tuic/udp_associations.go +++ b/internal/tuic/udp_associations.go @@ -20,7 +20,10 @@ type udpAssociation struct { lastActive time.Time opening bool openCancel context.CancelFunc - pending []pendingUdpPacket + // stopResponses ends the response relay, which may be blocked on stream credit the + // client never grants; closing the SOCKS side alone does not reach it. + stopResponses context.CancelFunc + pending []pendingUdpPacket } type udpAssociationRegistry struct { @@ -122,6 +125,7 @@ func (r *udpAssociationRegistry) dissociate(assocID uint16) bool { association := r.associations[assocID] if association != nil { r.clearPendingLocked(association) + association.stopResponsesLocked() } delete(r.associations, assocID) r.datagramReassembler.clearAssociation(assocID) @@ -160,6 +164,7 @@ func (r *udpAssociationRegistry) expireLocked(now time.Time) { } delete(r.associations, assocID) r.clearPendingLocked(association) + association.stopResponsesLocked() r.datagramReassembler.clearAssociation(assocID) r.streamReassembler.clearAssociation(assocID) if association.relay != nil { @@ -172,6 +177,7 @@ func (r *udpAssociationRegistry) closeAll() { r.mu.Lock() for _, association := range r.associations { r.clearPendingLocked(association) + association.stopResponsesLocked() if association.relay != nil { _ = association.relay.relay.Close() } @@ -186,6 +192,7 @@ func (r *udpAssociationRegistry) release(id uint16, expected *udpAssociation) { r.mu.Lock() if r.associations[id] == expected { r.clearPendingLocked(expected) + expected.stopResponsesLocked() delete(r.associations, id) r.datagramReassembler.clearAssociation(id) r.streamReassembler.clearAssociation(id) @@ -195,3 +202,21 @@ func (r *udpAssociationRegistry) release(id uint16, expected *udpAssociation) { _ = expected.relay.relay.Close() } } + +// attachResponseStop records the response relay's cancel on a live association; false +// means the association is already gone and the relay must not start. +func (r *udpAssociationRegistry) attachResponseStop(assocID uint16, expected *udpAssociation, stop context.CancelFunc) bool { + r.mu.Lock() + defer r.mu.Unlock() + if r.associations[assocID] != expected { + return false + } + expected.stopResponses = stop + return true +} + +func (a *udpAssociation) stopResponsesLocked() { + if a.stopResponses != nil { + a.stopResponses() + } +}