From 0d77139c24e4222740d982b0267281c3f0e349c8 Mon Sep 17 00:00:00 2001 From: Darien Raymond Date: Mon, 1 Feb 2016 15:36:33 +0000 Subject: [PATCH] fix udp issue in socks proxy --- proxy/freedom/freedom.go | 2 +- proxy/socks/socks.go | 1 + proxy/socks/udp.go | 64 ++++++++++++++--------------- testing/scenarios/socks_end_test.go | 22 +++++----- transport/hub/udp_server.go | 64 +++++++++++++++++++++++++++++ 5 files changed, 110 insertions(+), 43 deletions(-) create mode 100644 transport/hub/udp_server.go diff --git a/proxy/freedom/freedom.go b/proxy/freedom/freedom.go index 5902a427..93b44a73 100644 --- a/proxy/freedom/freedom.go +++ b/proxy/freedom/freedom.go @@ -1,7 +1,7 @@ package freedom import ( - "io" + "io" "net" "sync" diff --git a/proxy/socks/socks.go b/proxy/socks/socks.go index 7f570036..876d4ed2 100644 --- a/proxy/socks/socks.go +++ b/proxy/socks/socks.go @@ -32,6 +32,7 @@ type SocksServer struct { tcpListener *hub.TCPHub udpConn *net.UDPConn udpAddress v2net.Destination + udpServer *hub.UDPServer listeningPort v2net.Port } diff --git a/proxy/socks/udp.go b/proxy/socks/udp.go index 4eb69ce1..2dd496cc 100644 --- a/proxy/socks/udp.go +++ b/proxy/socks/udp.go @@ -7,6 +7,7 @@ import ( "github.com/v2ray/v2ray-core/common/log" v2net "github.com/v2ray/v2ray-core/common/net" "github.com/v2ray/v2ray-core/proxy/socks/protocol" + "github.com/v2ray/v2ray-core/transport/hub" ) func (this *SocksServer) ListenUDP(port v2net.Port) error { @@ -23,6 +24,7 @@ func (this *SocksServer) ListenUDP(port v2net.Port) error { this.udpMutex.Lock() this.udpAddress = v2net.UDPDestination(this.config.Address, port) this.udpConn = conn + this.udpServer = hub.NewUDPServer(this.packetDispatcher) this.udpMutex.Unlock() go this.AcceptPackets() @@ -63,38 +65,36 @@ func (this *SocksServer) AcceptPackets() error { udpPacket := v2net.NewPacket(request.Destination(), request.Data, false) log.Info("Socks: Send packet to ", udpPacket.Destination(), " with ", request.Data.Len(), " bytes") - go this.handlePacket(udpPacket, addr, request.Address, request.Port) + this.udpServer.Dispatch( + v2net.UDPDestination(v2net.IPAddress(addr.IP), v2net.Port(addr.Port)), udpPacket, + func(packet v2net.Packet) { + response := &protocol.Socks5UDPRequest{ + Fragment: 0, + Address: udpPacket.Destination().Address(), + Port: udpPacket.Destination().Port(), + Data: packet.Chunk(), + } + log.Info("Socks: Writing back UDP response with ", response.Data.Len(), " bytes to ", packet.Destination()) + + udpMessage := alloc.NewSmallBuffer().Clear() + response.Write(udpMessage) + + this.udpMutex.RLock() + if !this.accepting { + this.udpMutex.RUnlock() + return + } + nBytes, err := this.udpConn.WriteToUDP(udpMessage.Value, &net.UDPAddr{ + IP: packet.Destination().Address().IP(), + Port: int(packet.Destination().Port()), + }) + this.udpMutex.RUnlock() + udpMessage.Release() + response.Data.Release() + if err != nil { + log.Error("Socks: failed to write UDP message (", nBytes, " bytes) to ", packet.Destination(), ": ", err) + } + }) } return nil } - -func (this *SocksServer) handlePacket(packet v2net.Packet, clientAddr *net.UDPAddr, targetAddr v2net.Address, port v2net.Port) { - ray := this.packetDispatcher.DispatchToOutbound(packet) - close(ray.InboundInput()) - - for data := range ray.InboundOutput() { - response := &protocol.Socks5UDPRequest{ - Fragment: 0, - Address: targetAddr, - Port: port, - Data: data, - } - log.Info("Socks: Writing back UDP response with ", data.Len(), " bytes from ", targetAddr, " to ", clientAddr) - - udpMessage := alloc.NewSmallBuffer().Clear() - response.Write(udpMessage) - - this.udpMutex.RLock() - if !this.accepting { - this.udpMutex.RUnlock() - return - } - nBytes, err := this.udpConn.WriteToUDP(udpMessage.Value, clientAddr) - this.udpMutex.RUnlock() - udpMessage.Release() - response.Data.Release() - if err != nil { - log.Error("Socks: failed to write UDP message (", nBytes, " bytes) to ", clientAddr, ": ", err) - } - } -} diff --git a/testing/scenarios/socks_end_test.go b/testing/scenarios/socks_end_test.go index 0e60fcc7..d82f43c0 100644 --- a/testing/scenarios/socks_end_test.go +++ b/testing/scenarios/socks_end_test.go @@ -189,18 +189,20 @@ func TestUDPAssociate(t *testing.T) { }) assert.Error(err).IsNil() - udpPayload := "UDP request to udp server." - udpRequest := socks5UDPRequest(v2net.UDPDestination(v2net.IPAddress([]byte{127, 0, 0, 1}), targetPort), []byte(udpPayload)) + for i := 0; i < 100; i++ { + udpPayload := "UDP request to udp server." + udpRequest := socks5UDPRequest(v2net.UDPDestination(v2net.IPAddress([]byte{127, 0, 0, 1}), targetPort), []byte(udpPayload)) - nBytes, err = udpConn.Write(udpRequest) - assert.Int(nBytes).Equals(len(udpRequest)) - assert.Error(err).IsNil() + nBytes, err = udpConn.Write(udpRequest) + assert.Int(nBytes).Equals(len(udpRequest)) + assert.Error(err).IsNil() - udpResponse := make([]byte, 1024) - nBytes, err = udpConn.Read(udpResponse) - assert.Error(err).IsNil() - assert.Bytes(udpResponse[:nBytes]).Equals( - socks5UDPRequest(v2net.UDPDestination(v2net.IPAddress([]byte{127, 0, 0, 1}), targetPort), []byte("Processed: UDP request to udp server."))) + udpResponse := make([]byte, 1024) + nBytes, err = udpConn.Read(udpResponse) + assert.Error(err).IsNil() + assert.Bytes(udpResponse[:nBytes]).Equals( + socks5UDPRequest(v2net.UDPDestination(v2net.IPAddress([]byte{127, 0, 0, 1}), targetPort), []byte("Processed: UDP request to udp server."))) + } udpConn.Close() conn.Close() diff --git a/transport/hub/udp_server.go b/transport/hub/udp_server.go new file mode 100644 index 00000000..4845d938 --- /dev/null +++ b/transport/hub/udp_server.go @@ -0,0 +1,64 @@ +package hub + +import ( + "sync" + + "github.com/v2ray/v2ray-core/app/dispatcher" + v2net "github.com/v2ray/v2ray-core/common/net" + "github.com/v2ray/v2ray-core/transport/ray" +) + +type UDPResponseCallback func(packet v2net.Packet) + +type connEntry struct { + inboundRay ray.InboundRay + callback UDPResponseCallback +} + +type UDPServer struct { + sync.RWMutex + conns map[string]*connEntry + packetDispatcher dispatcher.PacketDispatcher +} + +func NewUDPServer(packetDispatcher dispatcher.PacketDispatcher) *UDPServer { + return &UDPServer{ + conns: make(map[string]*connEntry), + packetDispatcher: packetDispatcher, + } +} + +func (this *UDPServer) locateExistingAndDispatch(dest string, packet v2net.Packet) bool { + this.RLock() + defer this.RUnlock() + if entry, found := this.conns[dest]; found { + entry.inboundRay.InboundInput() <- packet.Chunk() + return true + } + return false +} + +func (this *UDPServer) Dispatch(source v2net.Destination, packet v2net.Packet, callback UDPResponseCallback) { + destString := source.String() + "-" + packet.Destination().String() + if this.locateExistingAndDispatch(destString, packet) { + return + } + + this.Lock() + inboundRay := this.packetDispatcher.DispatchToOutbound(v2net.NewPacket(packet.Destination(), packet.Chunk(), true)) + this.conns[destString] = &connEntry{ + inboundRay: inboundRay, + callback: callback, + } + this.Unlock() + go this.handleConnection(destString, inboundRay, source, callback) +} + +func (this *UDPServer) handleConnection(destString string, inboundRay ray.InboundRay, source v2net.Destination, callback UDPResponseCallback) { + for buffer := range inboundRay.InboundOutput() { + callback(v2net.NewPacket(source, buffer, false)) + } + this.Lock() + delete(this.conns, destString) + this.Unlock() +}