From 62a92780cfab18cfeb15994e10ced0cdeba47644 Mon Sep 17 00:00:00 2001 From: "Xiaokang Wang (Shelikhoo)" Date: Sat, 14 Mar 2026 16:16:00 +0000 Subject: [PATCH] Add TCP Listener support for wireguard outbound (#3651) --- .../net/abstactOutbount/errors.generated.go | 9 ++ common/net/abstactOutbount/outbound.go | 133 ++++++++++++++++++ common/packetswitch/gvisorstack/config.pb.go | 112 ++++++++++++--- common/packetswitch/gvisorstack/config.proto | 7 + common/packetswitch/gvisorstack/stack.go | 11 ++ .../packetswitch/gvisorstack/tcp_listener.go | 84 +++++++++++ 6 files changed, 338 insertions(+), 18 deletions(-) create mode 100644 common/net/abstactOutbount/errors.generated.go create mode 100644 common/net/abstactOutbount/outbound.go create mode 100644 common/packetswitch/gvisorstack/tcp_listener.go diff --git a/common/net/abstactOutbount/errors.generated.go b/common/net/abstactOutbount/errors.generated.go new file mode 100644 index 000000000..9e2a73d35 --- /dev/null +++ b/common/net/abstactOutbount/errors.generated.go @@ -0,0 +1,9 @@ +package abstactOutbount + +import "github.com/v2fly/v2ray-core/v5/common/errors" + +type errPathObjHolder struct{} + +func newError(values ...interface{}) *errors.Error { + return errors.New(values...).WithPathObj(errPathObjHolder{}) +} diff --git a/common/net/abstactOutbount/outbound.go b/common/net/abstactOutbount/outbound.go new file mode 100644 index 000000000..e284587cd --- /dev/null +++ b/common/net/abstactOutbount/outbound.go @@ -0,0 +1,133 @@ +package abstactOutbount + +//go:generate go run github.com/v2fly/v2ray-core/v5/common/errors/errorgen + +import ( + "context" + "sync" + + "github.com/v2fly/v2ray-core/v5/common" + "github.com/v2fly/v2ray-core/v5/common/net" + "github.com/v2fly/v2ray-core/v5/common/signal/done" + "github.com/v2fly/v2ray-core/v5/transport" +) + +func NewOutboundListener() *OutboundListener { + return &OutboundListener{ + buffer: make(chan *connWithContext, 4), + done: done.New(), + } +} + +type connWithContext struct { + net.Conn + ctx context.Context +} + +func (c *connWithContext) GetConnectionContext() context.Context { + return c.ctx +} + +// OutboundListener is a net.Listener for listening gRPC connections. +type OutboundListener struct { + buffer chan *connWithContext + done *done.Instance +} + +func (l *OutboundListener) addWithContext(ctx context.Context, conn net.Conn) { + select { + case l.buffer <- &connWithContext{Conn: conn, ctx: ctx}: + case <-l.done.Wait(): + conn.Close() + default: + conn.Close() + } +} + +// Accept implements net.Listener. +func (l *OutboundListener) Accept() (net.Conn, error) { + select { + case <-l.done.Wait(): + return nil, newError("listen closed") + case c := <-l.buffer: + return c, nil + } +} + +// Close implement net.Listener. +func (l *OutboundListener) Close() error { + common.Must(l.done.Close()) +L: + for { + select { + case c := <-l.buffer: + c.Close() + default: + break L + } + } + return nil +} + +// Addr implements net.Listener. +func (l *OutboundListener) Addr() net.Addr { + return &net.TCPAddr{ + IP: net.IP{0, 0, 0, 0}, + Port: 0, + } +} + +func NewOutbound(tag string, listener *OutboundListener) *Outbound { + return &Outbound{ + tag: tag, + listener: listener, + } +} + +// Outbound is a outbound.Handler that handles gRPC connections. +type Outbound struct { + tag string + listener *OutboundListener + access sync.RWMutex + closed bool +} + +// Dispatch implements outbound.Handler. +func (co *Outbound) Dispatch(ctx context.Context, link *transport.Link) { + co.access.RLock() + + if co.closed { + common.Interrupt(link.Reader) + common.Interrupt(link.Writer) + co.access.RUnlock() + return + } + + closeSignal := done.New() + c := net.NewConnection(net.ConnectionInputMulti(link.Writer), net.ConnectionOutputMulti(link.Reader), net.ConnectionOnClose(closeSignal)) + co.listener.addWithContext(ctx, c) + co.access.RUnlock() + <-closeSignal.Wait() +} + +// Tag implements outbound.Handler. +func (co *Outbound) Tag() string { + return co.tag +} + +// Start implements common.Runnable. +func (co *Outbound) Start() error { + co.access.Lock() + co.closed = false + co.access.Unlock() + return nil +} + +// Close implements common.Closable. +func (co *Outbound) Close() error { + co.access.Lock() + defer co.access.Unlock() + + co.closed = true + return co.listener.Close() +} diff --git a/common/packetswitch/gvisorstack/config.pb.go b/common/packetswitch/gvisorstack/config.pb.go index 70b8176aa..85a07b6a7 100644 --- a/common/packetswitch/gvisorstack/config.pb.go +++ b/common/packetswitch/gvisorstack/config.pb.go @@ -18,6 +18,66 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) +type TCPListener struct { + state protoimpl.MessageState `protogen:"open.v1"` + Port uint32 `protobuf:"varint,1,opt,name=port,proto3" json:"port,omitempty"` + Address *routercommon.CIDR `protobuf:"bytes,2,opt,name=address,proto3" json:"address,omitempty"` + Tag string `protobuf:"bytes,3,opt,name=tag,proto3" json:"tag,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TCPListener) Reset() { + *x = TCPListener{} + mi := &file_common_packetswitch_gvisorstack_config_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TCPListener) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TCPListener) ProtoMessage() {} + +func (x *TCPListener) ProtoReflect() protoreflect.Message { + mi := &file_common_packetswitch_gvisorstack_config_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use TCPListener.ProtoReflect.Descriptor instead. +func (*TCPListener) Descriptor() ([]byte, []int) { + return file_common_packetswitch_gvisorstack_config_proto_rawDescGZIP(), []int{0} +} + +func (x *TCPListener) GetPort() uint32 { + if x != nil { + return x.Port + } + return 0 +} + +func (x *TCPListener) GetAddress() *routercommon.CIDR { + if x != nil { + return x.Address + } + return nil +} + +func (x *TCPListener) GetTag() string { + if x != nil { + return x.Tag + } + return "" +} + type Config struct { state protoimpl.MessageState `protogen:"open.v1"` Mtu uint32 `protobuf:"varint,2,opt,name=mtu,proto3" json:"mtu,omitempty"` @@ -29,13 +89,14 @@ type Config struct { SocketSettings *internet.SocketConfig `protobuf:"bytes,10,opt,name=socket_settings,json=socketSettings,proto3" json:"socket_settings,omitempty"` PreferIpv6ForUdp bool `protobuf:"varint,11,opt,name=prefer_ipv6_for_udp,json=preferIpv6ForUdp,proto3" json:"prefer_ipv6_for_udp,omitempty"` DualStackUdp bool `protobuf:"varint,12,opt,name=dual_stack_udp,json=dualStackUdp,proto3" json:"dual_stack_udp,omitempty"` + TcpListener []*TCPListener `protobuf:"bytes,13,rep,name=tcp_listener,json=tcpListener,proto3" json:"tcp_listener,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } func (x *Config) Reset() { *x = Config{} - mi := &file_common_packetswitch_gvisorstack_config_proto_msgTypes[0] + mi := &file_common_packetswitch_gvisorstack_config_proto_msgTypes[1] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -47,7 +108,7 @@ func (x *Config) String() string { func (*Config) ProtoMessage() {} func (x *Config) ProtoReflect() protoreflect.Message { - mi := &file_common_packetswitch_gvisorstack_config_proto_msgTypes[0] + mi := &file_common_packetswitch_gvisorstack_config_proto_msgTypes[1] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -60,7 +121,7 @@ func (x *Config) ProtoReflect() protoreflect.Message { // Deprecated: Use Config.ProtoReflect.Descriptor instead. func (*Config) Descriptor() ([]byte, []int) { - return file_common_packetswitch_gvisorstack_config_proto_rawDescGZIP(), []int{0} + return file_common_packetswitch_gvisorstack_config_proto_rawDescGZIP(), []int{1} } func (x *Config) GetMtu() uint32 { @@ -126,11 +187,22 @@ func (x *Config) GetDualStackUdp() bool { return false } +func (x *Config) GetTcpListener() []*TCPListener { + if x != nil { + return x.TcpListener + } + return nil +} + var File_common_packetswitch_gvisorstack_config_proto protoreflect.FileDescriptor const file_common_packetswitch_gvisorstack_config_proto_rawDesc = "" + "\n" + - ",common/packetswitch/gvisorstack/config.proto\x12*v2ray.core.common.packetswitch.gvisorstack\x1a$app/router/routercommon/common.proto\x1a\x1ftransport/internet/config.proto\x1a common/protoext/extensions.proto\"\xc3\x03\n" + + ",common/packetswitch/gvisorstack/config.proto\x12*v2ray.core.common.packetswitch.gvisorstack\x1a$app/router/routercommon/common.proto\x1a\x1ftransport/internet/config.proto\x1a common/protoext/extensions.proto\"w\n" + + "\vTCPListener\x12\x12\n" + + "\x04port\x18\x01 \x01(\rR\x04port\x12B\n" + + "\aaddress\x18\x02 \x01(\v2(.v2ray.core.app.router.routercommon.CIDRR\aaddress\x12\x10\n" + + "\x03tag\x18\x03 \x01(\tR\x03tag\"\x9f\x04\n" + "\x06Config\x12\x10\n" + "\x03mtu\x18\x02 \x01(\rR\x03mtu\x12\x1d\n" + "\n" + @@ -142,7 +214,8 @@ const file_common_packetswitch_gvisorstack_config_proto_rawDesc = "" + "\x0fsocket_settings\x18\n" + " \x01(\v2+.v2ray.core.transport.internet.SocketConfigR\x0esocketSettings\x12-\n" + "\x13prefer_ipv6_for_udp\x18\v \x01(\bR\x10preferIpv6ForUdp\x12$\n" + - "\x0edual_stack_udp\x18\f \x01(\bR\fdualStackUdpB\x9f\x01\n" + + "\x0edual_stack_udp\x18\f \x01(\bR\fdualStackUdp\x12Z\n" + + "\ftcp_listener\x18\r \x03(\v27.v2ray.core.common.packetswitch.gvisorstack.TCPListenerR\vtcpListenerB\x9f\x01\n" + ".com.v2ray.core.common.packetswitch.gvisorstackP\x01Z>github.com/v2fly/v2ray-core/v5/common/packetswitch/gvisorstack\xaa\x02*V2Ray.Core.Common.Packetswitch.Gvisorstackb\x06proto3" var ( @@ -157,21 +230,24 @@ func file_common_packetswitch_gvisorstack_config_proto_rawDescGZIP() []byte { return file_common_packetswitch_gvisorstack_config_proto_rawDescData } -var file_common_packetswitch_gvisorstack_config_proto_msgTypes = make([]protoimpl.MessageInfo, 1) +var file_common_packetswitch_gvisorstack_config_proto_msgTypes = make([]protoimpl.MessageInfo, 2) var file_common_packetswitch_gvisorstack_config_proto_goTypes = []any{ - (*Config)(nil), // 0: v2ray.core.common.packetswitch.gvisorstack.Config - (*routercommon.CIDR)(nil), // 1: v2ray.core.app.router.routercommon.CIDR - (*internet.SocketConfig)(nil), // 2: v2ray.core.transport.internet.SocketConfig + (*TCPListener)(nil), // 0: v2ray.core.common.packetswitch.gvisorstack.TCPListener + (*Config)(nil), // 1: v2ray.core.common.packetswitch.gvisorstack.Config + (*routercommon.CIDR)(nil), // 2: v2ray.core.app.router.routercommon.CIDR + (*internet.SocketConfig)(nil), // 3: v2ray.core.transport.internet.SocketConfig } var file_common_packetswitch_gvisorstack_config_proto_depIdxs = []int32{ - 1, // 0: v2ray.core.common.packetswitch.gvisorstack.Config.ips:type_name -> v2ray.core.app.router.routercommon.CIDR - 1, // 1: v2ray.core.common.packetswitch.gvisorstack.Config.routes:type_name -> v2ray.core.app.router.routercommon.CIDR - 2, // 2: v2ray.core.common.packetswitch.gvisorstack.Config.socket_settings:type_name -> v2ray.core.transport.internet.SocketConfig - 3, // [3:3] is the sub-list for method output_type - 3, // [3:3] is the sub-list for method input_type - 3, // [3:3] is the sub-list for extension type_name - 3, // [3:3] is the sub-list for extension extendee - 0, // [0:3] is the sub-list for field type_name + 2, // 0: v2ray.core.common.packetswitch.gvisorstack.TCPListener.address:type_name -> v2ray.core.app.router.routercommon.CIDR + 2, // 1: v2ray.core.common.packetswitch.gvisorstack.Config.ips:type_name -> v2ray.core.app.router.routercommon.CIDR + 2, // 2: v2ray.core.common.packetswitch.gvisorstack.Config.routes:type_name -> v2ray.core.app.router.routercommon.CIDR + 3, // 3: v2ray.core.common.packetswitch.gvisorstack.Config.socket_settings:type_name -> v2ray.core.transport.internet.SocketConfig + 0, // 4: v2ray.core.common.packetswitch.gvisorstack.Config.tcp_listener:type_name -> v2ray.core.common.packetswitch.gvisorstack.TCPListener + 5, // [5:5] is the sub-list for method output_type + 5, // [5:5] is the sub-list for method input_type + 5, // [5:5] is the sub-list for extension type_name + 5, // [5:5] is the sub-list for extension extendee + 0, // [0:5] is the sub-list for field type_name } func init() { file_common_packetswitch_gvisorstack_config_proto_init() } @@ -185,7 +261,7 @@ func file_common_packetswitch_gvisorstack_config_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_common_packetswitch_gvisorstack_config_proto_rawDesc), len(file_common_packetswitch_gvisorstack_config_proto_rawDesc)), NumEnums: 0, - NumMessages: 1, + NumMessages: 2, NumExtensions: 0, NumServices: 0, }, diff --git a/common/packetswitch/gvisorstack/config.proto b/common/packetswitch/gvisorstack/config.proto index c53659faa..6e9aa7034 100644 --- a/common/packetswitch/gvisorstack/config.proto +++ b/common/packetswitch/gvisorstack/config.proto @@ -10,6 +10,12 @@ import "app/router/routercommon/common.proto"; import "transport/internet/config.proto"; import "common/protoext/extensions.proto"; +message TCPListener { + uint32 port = 1; + v2ray.core.app.router.routercommon.CIDR address = 2; + string tag = 3; +} + message Config { uint32 mtu = 2; uint32 user_level = 3; @@ -20,4 +26,5 @@ message Config { v2ray.core.transport.internet.SocketConfig socket_settings = 10; bool prefer_ipv6_for_udp = 11; bool dual_stack_udp = 12; + repeated TCPListener tcp_listener = 13; } \ No newline at end of file diff --git a/common/packetswitch/gvisorstack/stack.go b/common/packetswitch/gvisorstack/stack.go index aea854fdc..3b6def2d5 100644 --- a/common/packetswitch/gvisorstack/stack.go +++ b/common/packetswitch/gvisorstack/stack.go @@ -5,6 +5,7 @@ import ( "fmt" "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/adapters/gonet" "gvisor.dev/gvisor/pkg/tcpip/network/ipv4" "gvisor.dev/gvisor/pkg/tcpip/network/ipv6" "gvisor.dev/gvisor/pkg/tcpip/stack" @@ -21,6 +22,8 @@ type WrappedStack struct { config *Config ctx context.Context stack *stack.Stack + + stackTCPListeners []*gonet.TCPListener } func NewStack(ctx context.Context, config *Config) (*WrappedStack, error) { @@ -58,6 +61,10 @@ func (w *WrappedStack) CreateStackFromNetworkLayerDevice(packetSwitchDevice pack }) w.stack = s + + if err := w.ApplyListeners(); err != nil { + return fmt.Errorf("failed to apply listeners: %v", err) + } return nil } @@ -159,6 +166,10 @@ func (w *WrappedStack) Close() error { if w == nil || w.stack == nil { return nil } + for _, l := range w.stackTCPListeners { + l.Close() + } + w.stackTCPListeners = nil w.stack.Close() w.stack = nil return nil diff --git a/common/packetswitch/gvisorstack/tcp_listener.go b/common/packetswitch/gvisorstack/tcp_listener.go new file mode 100644 index 000000000..fd92ddb0e --- /dev/null +++ b/common/packetswitch/gvisorstack/tcp_listener.go @@ -0,0 +1,84 @@ +package gvisorstack + +import ( + "fmt" + "io" + + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/adapters/gonet" + "gvisor.dev/gvisor/pkg/tcpip/network/ipv4" + "gvisor.dev/gvisor/pkg/tcpip/network/ipv6" + + "github.com/v2fly/v2ray-core/v5/common/net" + "github.com/v2fly/v2ray-core/v5/transport/internet/tagged" +) + +func (w *WrappedStack) ApplyListeners() error { + for _, listener := range w.config.TcpListener { + listenerIP := listener.Address.Ip + listenerPort := listener.GetPort() + listenerTag := listener.Tag + + addr := tcpip.FullAddress{ + Addr: tcpip.AddrFromSlice(listenerIP), + Port: uint16(listenerPort), + } + + netProto := ipv4.ProtocolNumber + if len(listenerIP) == net.IPv6len { + netProto = ipv6.ProtocolNumber + } + + tcpListener, err := w.CreateStackListener(addr, netProto) + if err != nil { + return fmt.Errorf("failed to create TCP listener on %v:%d: %w", listenerIP, listenerPort, err) + } + + w.stackTCPListeners = append(w.stackTCPListeners, tcpListener) + + go w.acceptLoop(tcpListener, listenerTag) + } + return nil +} + +func (w *WrappedStack) acceptLoop(listener *gonet.TCPListener, outboundTag string) { + for { + conn, err := listener.Accept() + if err != nil { + return + } + go w.handleTCPConn(conn, outboundTag) + } +} + +func (w *WrappedStack) handleTCPConn(conn net.Conn, outboundTag string) { + defer conn.Close() + + // Use the connection's local address as the destination, + // representing the address the client was connecting to on the stack. + tcpAddr, ok := conn.LocalAddr().(*net.TCPAddr) + if !ok { + return + } + + dest := net.TCPDestination(net.IPAddress(tcpAddr.IP), net.Port(tcpAddr.Port)) + + outboundConn, err := tagged.Dialer(w.ctx, dest, outboundTag) + if err != nil { + return + } + defer outboundConn.Close() + + // Bidirectional relay + done := make(chan struct{}) + go func() { + io.Copy(outboundConn, conn) + close(done) + }() + io.Copy(conn, outboundConn) + <-done +} + +func (w *WrappedStack) CreateStackListener(addr tcpip.FullAddress, netProto tcpip.NetworkProtocolNumber) (*gonet.TCPListener, error) { + return gonet.ListenTCP(w.stack, addr, netProto) +}