split listening settings from inbound proxies and apply context

proxy/dokodemo/dokodemo.go

@@ -2,7 +2,6 @@ package dokodemo
 
 import (
 	"context"
-	"sync"
 
 	"v2ray.com/core/app"
 	"v2ray.com/core/app/dispatcher"
@@ -14,21 +13,13 @@ import (
 	"v2ray.com/core/common/signal"
 	"v2ray.com/core/proxy"
 	"v2ray.com/core/transport/internet"
-	"v2ray.com/core/transport/internet/udp"
 )
 
 type DokodemoDoor struct {
-	tcpMutex         sync.RWMutex
-	udpMutex         sync.RWMutex
 	config           *Config
-	accepting        bool
 	address          net.Address
 	port             net.Port
 	packetDispatcher dispatcher.Interface
-	tcpListener      *internet.TCPHub
-	udpHub           *udp.Hub
-	udpServer        *udp.Server
-	meta             *proxy.InboundHandlerMeta
 }
 
 func New(ctx context.Context, config *Config) (*DokodemoDoor, error) {
@@ -36,10 +27,6 @@ func New(ctx context.Context, config *Config) (*DokodemoDoor, error) {
 	if space == nil {
 		return nil, errors.New("Dokodemo: No space in context.")
 	}
-	meta := proxy.InboundMetaFromContext(ctx)
-	if meta == nil {
-		return nil, errors.New("Dokodemo: No outbound meta in context.")
-	}
 	if config.NetworkList == nil || config.NetworkList.Size() == 0 {
 		return nil, errors.New("DokodemoDoor: No network specified.")
 	}
@@ -47,7 +34,6 @@ func New(ctx context.Context, config *Config) (*DokodemoDoor, error) {
 		config:  config,
 		address: config.GetPredefinedAddress(),
 		port:    net.Port(config.Port),
-		meta:    meta,
 	}
 	space.OnInitialize(func() error {
 		d.packetDispatcher = dispatcher.FromSpace(space)
@@ -59,140 +45,28 @@ func New(ctx context.Context, config *Config) (*DokodemoDoor, error) {
 	return d, nil
 }
 
-func (v *DokodemoDoor) Port() net.Port {
-	return v.meta.Port
+func (d *DokodemoDoor) Network() net.NetworkList {
+	return *(d.config.NetworkList)
 }
 
-func (v *DokodemoDoor) Close() {
-	v.accepting = false
-	if v.tcpListener != nil {
-		v.tcpMutex.Lock()
-		v.tcpListener.Close()
-		v.tcpListener = nil
-		v.tcpMutex.Unlock()
-	}
-	if v.udpHub != nil {
-		v.udpMutex.Lock()
-		v.udpHub.Close()
-		v.udpHub = nil
-		v.udpMutex.Unlock()
-	}
-}
-
-func (v *DokodemoDoor) Network() net.NetworkList {
-	return *(v.config.NetworkList)
-}
-
-func (v *DokodemoDoor) Start() error {
-	if v.accepting {
-		return nil
-	}
-	v.accepting = true
-
-	if v.config.NetworkList.HasNetwork(net.Network_TCP) {
-		err := v.ListenTCP()
-		if err != nil {
-			return err
-		}
-	}
-	if v.config.NetworkList.HasNetwork(net.Network_UDP) {
-		err := v.ListenUDP()
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (v *DokodemoDoor) ListenUDP() error {
-	v.udpServer = udp.NewServer(v.packetDispatcher)
-	udpHub, err := udp.ListenUDP(
-		v.meta.Address, v.meta.Port, udp.ListenOption{
-			Callback:            v.handleUDPPackets,
-			ReceiveOriginalDest: v.config.FollowRedirect,
-			Concurrency:         2,
-		})
-	if err != nil {
-		log.Error("Dokodemo failed to listen on ", v.meta.Address, ":", v.meta.Port, ": ", err)
-		return err
-	}
-	v.udpMutex.Lock()
-	v.udpHub = udpHub
-	v.udpMutex.Unlock()
-	return nil
-}
-
-func (v *DokodemoDoor) handleUDPPackets(payload *buf.Buffer, session *proxy.SessionInfo) {
-	if session.Destination.Network == net.Network_Unknown && v.address != nil && v.port > 0 {
-		session.Destination = net.UDPDestination(v.address, v.port)
-	}
-	if session.Destination.Network == net.Network_Unknown {
-		log.Info("Dokodemo: Unknown destination, stop forwarding...")
-		return
-	}
-	session.Inbound = v.meta
-	v.udpServer.Dispatch(session, payload, v.handleUDPResponse)
-}
-
-func (v *DokodemoDoor) handleUDPResponse(dest net.Destination, payload *buf.Buffer) {
-	defer payload.Release()
-	v.udpMutex.RLock()
-	defer v.udpMutex.RUnlock()
-	if !v.accepting {
-		return
-	}
-	v.udpHub.WriteTo(payload.Bytes(), dest)
-}
-
-func (v *DokodemoDoor) ListenTCP() error {
-	tcpListener, err := internet.ListenTCP(v.meta.Address, v.meta.Port, v.HandleTCPConnection, v.meta.StreamSettings)
-	if err != nil {
-		log.Error("Dokodemo: Failed to listen on ", v.meta.Address, ":", v.meta.Port, ": ", err)
-		return err
-	}
-	v.tcpMutex.Lock()
-	v.tcpListener = tcpListener
-	v.tcpMutex.Unlock()
-	return nil
-}
-
-func (v *DokodemoDoor) HandleTCPConnection(conn internet.Connection) {
-	defer conn.Close()
+func (d *DokodemoDoor) Process(ctx context.Context, network net.Network, conn internet.Connection) error {
+	log.Debug("Dokodemo: processing connection from: ", conn.RemoteAddr())
 	conn.SetReusable(false)
-
-	var dest net.Destination
-	if v.config.FollowRedirect {
-		originalDest := GetOriginalDestination(conn)
-		if originalDest.Network != net.Network_Unknown {
-			log.Info("Dokodemo: Following redirect to: ", originalDest)
-			dest = originalDest
-		}
-	}
-	if dest.Network == net.Network_Unknown && v.address != nil && v.port > net.Port(0) {
-		dest = net.TCPDestination(v.address, v.port)
-	}
-
-	if dest.Network == net.Network_Unknown {
-		log.Info("Dokodemo: Unknown destination, stop forwarding...")
-		return
-	}
-	log.Info("Dokodemo: Handling request to ", dest)
-
-	ray := v.packetDispatcher.DispatchToOutbound(&proxy.SessionInfo{
-		Source:      net.DestinationFromAddr(conn.RemoteAddr()),
-		Destination: dest,
-		Inbound:     v.meta,
+	ctx = proxy.ContextWithDestination(ctx, net.Destination{
+		Network: network,
+		Address: d.address,
+		Port:    d.port,
 	})
-
-	reader := net.NewTimeOutReader(v.config.Timeout, conn)
+	inboundRay := d.packetDispatcher.DispatchToOutbound(ctx)
 
 	requestDone := signal.ExecuteAsync(func() error {
-		defer ray.InboundInput().Close()
+		defer inboundRay.InboundInput().Close()
 
-		v2reader := buf.NewReader(reader)
+		timedReader := net.NewTimeOutReader(d.config.Timeout, conn)
+		chunkReader := buf.NewReader(timedReader)
 
-		if err := buf.PipeUntilEOF(v2reader, ray.InboundInput()); err != nil {
-			log.Info("Dokodemo: Failed to transport all TCP request: ", err)
+		if err := buf.PipeUntilEOF(chunkReader, inboundRay.InboundInput()); err != nil {
+			log.Info("Dokodemo: Failed to transport request: ", err)
 			return err
 		}
 
@@ -202,18 +76,21 @@ func (v *DokodemoDoor) HandleTCPConnection(conn internet.Connection) {
 	responseDone := signal.ExecuteAsync(func() error {
 		v2writer := buf.NewWriter(conn)
 
-		if err := buf.PipeUntilEOF(ray.InboundOutput(), v2writer); err != nil {
-			log.Info("Dokodemo: Failed to transport all TCP response: ", err)
+		if err := buf.PipeUntilEOF(inboundRay.InboundOutput(), v2writer); err != nil {
+			log.Info("Dokodemo: Failed to transport response: ", err)
 			return err
 		}
 		return nil
 	})
 
 	if err := signal.ErrorOrFinish2(requestDone, responseDone); err != nil {
-		ray.InboundInput().CloseError()
-		ray.InboundOutput().CloseError()
+		inboundRay.InboundInput().CloseError()
+		inboundRay.InboundOutput().CloseError()
 		log.Info("Dokodemo: Connection ends with ", err)
+		return err
 	}
+
+	return nil
 }
 
 func init() {

proxy/dokodemo/dokodemo_test.go

@@ -1,173 +0,0 @@
-package dokodemo_test
-
-import (
-	"net"
-	"testing"
-
-	"context"
-
-	"v2ray.com/core/app"
-	"v2ray.com/core/app/dispatcher"
-	_ "v2ray.com/core/app/dispatcher/impl"
-	"v2ray.com/core/app/proxyman"
-	_ "v2ray.com/core/app/proxyman/outbound"
-	"v2ray.com/core/common/dice"
-	v2net "v2ray.com/core/common/net"
-	"v2ray.com/core/proxy"
-	. "v2ray.com/core/proxy/dokodemo"
-	"v2ray.com/core/proxy/freedom"
-	"v2ray.com/core/testing/assert"
-	"v2ray.com/core/testing/servers/tcp"
-	"v2ray.com/core/testing/servers/udp"
-	"v2ray.com/core/transport/internet"
-	_ "v2ray.com/core/transport/internet/tcp"
-)
-
-func TestDokodemoTCP(t *testing.T) {
-	assert := assert.On(t)
-
-	tcpServer := &tcp.Server{
-		MsgProcessor: func(data []byte) []byte {
-			buffer := make([]byte, 0, 2048)
-			buffer = append(buffer, []byte("Processed: ")...)
-			buffer = append(buffer, data...)
-			return buffer
-		},
-	}
-	_, err := tcpServer.Start()
-	assert.Error(err).IsNil()
-
-	defer tcpServer.Close()
-
-	space := app.NewSpace()
-	ctx := app.ContextWithSpace(context.Background(), space)
-	app.AddApplicationToSpace(ctx, new(dispatcher.Config))
-	app.AddApplicationToSpace(ctx, new(proxyman.OutboundConfig))
-
-	ohm := proxyman.OutboundHandlerManagerFromSpace(space)
-	freedom, err := freedom.New(proxy.ContextWithOutboundMeta(ctx, &proxy.OutboundHandlerMeta{
-		Address: v2net.LocalHostIP,
-		StreamSettings: &internet.StreamConfig{
-			Protocol: internet.TransportProtocol_TCP,
-		},
-	}), &freedom.Config{})
-	assert.Error(err).IsNil()
-	ohm.SetDefaultHandler(freedom)
-
-	data2Send := "Data to be sent to remote."
-
-	port := v2net.Port(dice.Roll(20000) + 10000)
-
-	ctx = proxy.ContextWithInboundMeta(ctx, &proxy.InboundHandlerMeta{
-		Address: v2net.LocalHostIP,
-		Port:    port,
-		StreamSettings: &internet.StreamConfig{
-			Protocol: internet.TransportProtocol_TCP,
-		}})
-
-	dokodemo, err := New(ctx, &Config{
-		Address:     v2net.NewIPOrDomain(v2net.LocalHostIP),
-		Port:        uint32(tcpServer.Port),
-		NetworkList: v2net.Network_TCP.AsList(),
-		Timeout:     600,
-	})
-	assert.Error(err).IsNil()
-	defer dokodemo.Close()
-
-	assert.Error(space.Initialize()).IsNil()
-
-	err = dokodemo.Start()
-	assert.Error(err).IsNil()
-	assert.Port(port).Equals(dokodemo.Port())
-
-	tcpClient, err := net.DialTCP("tcp", nil, &net.TCPAddr{
-		IP:   []byte{127, 0, 0, 1},
-		Port: int(port),
-		Zone: "",
-	})
-	assert.Error(err).IsNil()
-
-	tcpClient.Write([]byte(data2Send))
-	tcpClient.CloseWrite()
-
-	response := make([]byte, 1024)
-	nBytes, err := tcpClient.Read(response)
-	assert.Error(err).IsNil()
-	tcpClient.Close()
-
-	assert.String("Processed: " + data2Send).Equals(string(response[:nBytes]))
-}
-
-func TestDokodemoUDP(t *testing.T) {
-	assert := assert.On(t)
-
-	udpServer := &udp.Server{
-		MsgProcessor: func(data []byte) []byte {
-			buffer := make([]byte, 0, 2048)
-			buffer = append(buffer, []byte("Processed: ")...)
-			buffer = append(buffer, data...)
-			return buffer
-		},
-	}
-	_, err := udpServer.Start()
-	assert.Error(err).IsNil()
-
-	defer udpServer.Close()
-
-	space := app.NewSpace()
-	ctx := app.ContextWithSpace(context.Background(), space)
-	app.AddApplicationToSpace(ctx, new(dispatcher.Config))
-	app.AddApplicationToSpace(ctx, new(proxyman.OutboundConfig))
-
-	ohm := proxyman.OutboundHandlerManagerFromSpace(space)
-	freedom, err := freedom.New(proxy.ContextWithOutboundMeta(ctx, &proxy.OutboundHandlerMeta{
-		Address: v2net.AnyIP,
-		StreamSettings: &internet.StreamConfig{
-			Protocol: internet.TransportProtocol_TCP,
-		},
-	}), &freedom.Config{})
-	assert.Error(err).IsNil()
-	ohm.SetDefaultHandler(freedom)
-
-	data2Send := "Data to be sent to remote."
-
-	port := v2net.Port(dice.Roll(20000) + 10000)
-
-	ctx = proxy.ContextWithInboundMeta(ctx, &proxy.InboundHandlerMeta{
-		Address: v2net.LocalHostIP,
-		Port:    port,
-		StreamSettings: &internet.StreamConfig{
-			Protocol: internet.TransportProtocol_TCP,
-		}})
-
-	dokodemo, err := New(ctx, &Config{
-		Address:     v2net.NewIPOrDomain(v2net.LocalHostIP),
-		Port:        uint32(udpServer.Port),
-		NetworkList: v2net.Network_UDP.AsList(),
-		Timeout:     600,
-	})
-	assert.Error(err).IsNil()
-	defer dokodemo.Close()
-
-	assert.Error(space.Initialize()).IsNil()
-
-	err = dokodemo.Start()
-	assert.Error(err).IsNil()
-	assert.Port(port).Equals(dokodemo.Port())
-
-	udpClient, err := net.DialUDP("udp", nil, &net.UDPAddr{
-		IP:   []byte{127, 0, 0, 1},
-		Port: int(port),
-		Zone: "",
-	})
-	assert.Error(err).IsNil()
-	defer udpClient.Close()
-
-	udpClient.Write([]byte(data2Send))
-
-	response := make([]byte, 1024)
-	nBytes, addr, err := udpClient.ReadFromUDP(response)
-	assert.Error(err).IsNil()
-	assert.IP(addr.IP).Equals(v2net.LocalHostIP.IP())
-	assert.Bytes(response[:nBytes]).Equals([]byte("Processed: " + data2Send))
-}

proxy/dokodemo/sockopt_linux.go

@@ -1,35 +0,0 @@
-// +build linux
-
-package dokodemo
-
-import (
-	"syscall"
-
-	"v2ray.com/core/common/log"
-	"v2ray.com/core/common/net"
-	"v2ray.com/core/transport/internet"
-)
-
-const SO_ORIGINAL_DST = 80
-
-func GetOriginalDestination(conn internet.Connection) net.Destination {
-	tcpConn, ok := conn.(internet.SysFd)
-	if !ok {
-		log.Info("Dokodemo: Failed to get sys fd.")
-		return net.Destination{}
-	}
-	fd, err := tcpConn.SysFd()
-	if err != nil {
-		log.Info("Dokodemo: Failed to get original destination: ", err)
-		return net.Destination{}
-	}
-
-	addr, err := syscall.GetsockoptIPv6Mreq(fd, syscall.IPPROTO_IP, SO_ORIGINAL_DST)
-	if err != nil {
-		log.Info("Dokodemo: Failed to call getsockopt: ", err)
-		return net.Destination{}
-	}
-	ip := net.IPAddress(addr.Multiaddr[4:8])
-	port := uint16(addr.Multiaddr[2])<<8 + uint16(addr.Multiaddr[3])
-	return net.TCPDestination(ip, net.Port(port))
-}

proxy/dokodemo/sockopt_other.go

@@ -1,12 +0,0 @@
-// +build !linux
-
-package dokodemo
-
-import (
-	"v2ray.com/core/common/net"
-	"v2ray.com/core/transport/internet"
-)
-
-func GetOriginalDestination(conn internet.Connection) net.Destination {
-	return net.Destination{}
-}