Add iOS Client Support Bundle: Packet Bridge Receiver, Subscription Improvment, DNS and other udp improvement (#3809)

* Improve data url subscription link processing

* consider disk persisted source as added by API and removeable

* feat: add Packet Bridge transport for iOS

* fix(tun): preserve inbound tag for packet encoding

* feat(tun): bypass packet encoding for selected UDP ports

* fix(tun): clean stale stream packetaddr entries

Close packetaddr dispatchers when their read workers terminate so the TUN
sorter removes stale source entries. Make dispatcher close idempotent and
interrupt packet links before taking the reader lock to avoid cleanup
deadlocks. Add regression coverage for stream termination, repeated close,
and recreation of the same source tuple.

* fix lint issue for tun, sub

* fix packet connection adaptor close processing
This commit is contained in:
Xiaokang Wang (Shelikhoo)
2026-08-23 16:02:37 +01:00
committed by GitHub
parent deb36f9009
commit cbd4d234fe
22 changed files with 1333 additions and 78 deletions
+17 -2
View File
@@ -1,3 +1,9 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.11
// protoc v5.28.1
// source: app/subscription/config.proto
package subscription
import (
@@ -99,6 +105,7 @@ type Config struct {
NonnativeConverterOverlay []byte `protobuf:"bytes,2,opt,name=nonnative_converter_overlay,json=nonnativeConverterOverlay,proto3" json:"nonnative_converter_overlay,omitempty"`
NonnativeConverterOverlayFile string `protobuf:"bytes,96002,opt,name=nonnative_converter_overlay_file,json=nonnativeConverterOverlayFile,proto3" json:"nonnative_converter_overlay_file,omitempty"`
Persistence bool `protobuf:"varint,3,opt,name=persistence,proto3" json:"persistence,omitempty"`
DefaultDialerTag string `protobuf:"bytes,4,opt,name=default_dialer_tag,json=defaultDialerTag,proto3" json:"default_dialer_tag,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -161,6 +168,13 @@ func (x *Config) GetPersistence() bool {
return false
}
func (x *Config) GetDefaultDialerTag() string {
if x != nil {
return x.DefaultDialerTag
}
return ""
}
var File_app_subscription_config_proto protoreflect.FileDescriptor
const file_app_subscription_config_proto_rawDesc = "" +
@@ -172,12 +186,13 @@ const file_app_subscription_config_proto_rawDesc = "" +
"\n" +
"tag_prefix\x18\x03 \x01(\tR\ttagPrefix\x12(\n" +
"\x10import_using_tag\x18\x04 \x01(\tR\x0eimportUsingTag\x124\n" +
"\x16default_expire_seconds\x18\x05 \x01(\x04R\x14defaultExpireSeconds\"\xba\x02\n" +
"\x16default_expire_seconds\x18\x05 \x01(\x04R\x14defaultExpireSeconds\"\xe8\x02\n" +
"\x06Config\x12C\n" +
"\aimports\x18\x01 \x03(\v2).v2ray.core.app.subscription.ImportSourceR\aimports\x12>\n" +
"\x1bnonnative_converter_overlay\x18\x02 \x01(\fR\x19nonnativeConverterOverlay\x12l\n" +
" nonnative_converter_overlay_file\x18\x82\xee\x05 \x01(\tB!\x82\xb5\x18\x1d\"\x1bnonnative_converter_overlayR\x1dnonnativeConverterOverlayFile\x12 \n" +
"\vpersistence\x18\x03 \x01(\bR\vpersistence:\x1b\x82\xb5\x18\x17\n" +
"\vpersistence\x18\x03 \x01(\bR\vpersistence\x12,\n" +
"\x12default_dialer_tag\x18\x04 \x01(\tR\x10defaultDialerTag:\x1b\x82\xb5\x18\x17\n" +
"\aservice\x12\fsubscriptionBr\n" +
"\x1fcom.v2ray.core.app.subscriptionP\x01Z/github.com/v2fly/v2ray-core/v5/app/subscription\xaa\x02\x1bV2Ray.Core.App.Subscriptionb\x06proto3"
+2 -1
View File
@@ -30,4 +30,5 @@ message Config {
string nonnative_converter_overlay_file = 96002 [(v2ray.core.common.protoext.field_opt).convert_time_read_file_into = "nonnative_converter_overlay"];
bool persistence = 3;
}
string default_dialer_tag = 4;
}
+36 -1
View File
@@ -6,6 +6,8 @@ import (
"net/url"
"strings"
"github.com/vincent-petithory/dataurl"
"github.com/v2fly/v2ray-core/v5/app/subscription/containers"
"github.com/v2fly/v2ray-core/v5/common"
)
@@ -16,6 +18,25 @@ func newURLLineParser() containers.SubscriptionContainerDocumentParser {
type parser struct{}
func parseDataURLLine(content string) ([]byte, bool, error) {
if !strings.HasPrefix(content, "data:") {
return nil, false, nil
}
dataURL, err := dataurl.DecodeString(content)
if err != nil {
return nil, true, newError("unable to decode dataURL").Base(err)
}
if dataURL.Type != "application" {
return nil, true, newError("unsupported media type: ", dataURL.Type)
}
if !strings.HasPrefix(dataURL.Subtype, "vnd.v2ray.subscription-singular") {
return nil, true, newError("unsupported media subtype: ", dataURL.Subtype)
}
return dataURL.Data, true, nil
}
func (p parser) ParseSubscriptionContainerDocument(rawConfig []byte) (*containers.Container, error) {
result := &containers.Container{}
result.Kind = "URLLine"
@@ -33,12 +54,26 @@ func (p parser) ParseSubscriptionContainerDocument(rawConfig []byte) (*container
for scanner.Scan() {
content := scanner.Text()
content = strings.TrimSpace(content)
if content == "" {
continue
}
if strings.HasPrefix(content, "#") {
continue
}
if strings.HasPrefix(content, "//") {
continue
}
if dataURLData, ok, err := parseDataURLLine(content); err != nil {
failedLine++
continue
} else if ok {
parsedLine++
result.ServerSpecs = append(result.ServerSpecs, containers.UnparsedServerConf{
KindHint: "",
Content: dataURLData,
})
continue
}
_, err := url.Parse(content)
if err != nil {
failedLine++
@@ -48,7 +83,7 @@ func (p parser) ParseSubscriptionContainerDocument(rawConfig []byte) (*container
}
result.ServerSpecs = append(result.ServerSpecs, containers.UnparsedServerConf{
KindHint: "URL",
Content: scanner.Bytes(),
Content: []byte(content),
})
}
@@ -0,0 +1,72 @@
package subscriptionmanager
import (
"testing"
"github.com/v2fly/v2ray-core/v5/app/proxyman"
"github.com/v2fly/v2ray-core/v5/app/subscription"
"github.com/v2fly/v2ray-core/v5/app/subscription/specs"
"github.com/v2fly/v2ray-core/v5/common/serial"
)
func TestMaterializeAppliesDefaultDialerTag(t *testing.T) {
manager := &SubscriptionManagerImpl{
config: &subscription.Config{DefaultDialerTag: "packetbridge-bypass"},
}
outbound, err := manager.materialize("test", "subscription_test", &specs.SubscriptionServerConfig{
Configuration: &specs.ServerConfiguration{},
})
if err != nil {
t.Fatalf("materialize failed: %v", err)
}
senderInstance, err := serial.GetInstanceOf(outbound.SenderSettings)
if err != nil {
t.Fatalf("failed to decode sender settings: %v", err)
}
sender := senderInstance.(*proxyman.SenderConfig)
if sender.ProxySettings == nil {
t.Fatal("proxy settings were not applied")
}
if sender.ProxySettings.Tag != "packetbridge-bypass" {
t.Fatalf("unexpected proxy tag: %q", sender.ProxySettings.Tag)
}
if !sender.ProxySettings.TransportLayerProxy {
t.Fatal("transport-layer proxy was not enabled")
}
}
func TestMaterializePreservesLegacyBehaviorWithoutDefault(t *testing.T) {
manager := &SubscriptionManagerImpl{config: &subscription.Config{}}
outbound, err := manager.materialize("test", "subscription_test", &specs.SubscriptionServerConfig{
Configuration: &specs.ServerConfiguration{},
})
if err != nil {
t.Fatalf("materialize failed: %v", err)
}
senderInstance, err := serial.GetInstanceOf(outbound.SenderSettings)
if err != nil {
t.Fatalf("failed to decode sender settings: %v", err)
}
if senderInstance.(*proxyman.SenderConfig).ProxySettings != nil {
t.Fatal("unexpected proxy settings without default_dialer_tag")
}
}
func TestImportSourceDefaultDialerPrecedence(t *testing.T) {
original := &subscription.ImportSource{ImportUsingTag: ""}
effective := importSourceWithDefaultDialer(original, "packetbridge-bypass")
if effective.ImportUsingTag != "packetbridge-bypass" {
t.Fatalf("default tag was not applied: %q", effective.ImportUsingTag)
}
if original.ImportUsingTag != "" {
t.Fatal("source was mutated")
}
explicit := &subscription.ImportSource{ImportUsingTag: "explicit"}
effective = importSourceWithDefaultDialer(explicit, "packetbridge-bypass")
if effective.ImportUsingTag != "explicit" {
t.Fatalf("explicit tag was overwritten: %q", effective.ImportUsingTag)
}
}
@@ -163,7 +163,7 @@ func (s *SubscriptionManagerImpl) loadAllFromPersistentStorage() error {
if err != nil {
return newError("failed to get import source: ", err)
}
if err := s.addTrackedSubscriptionFromImportSource(&importSource, false); err != nil {
if err := s.addTrackedSubscriptionFromImportSource(&importSource, true); err != nil {
return newError("failed to add tracked subscription: ", err)
}
}
@@ -32,6 +32,13 @@ func (s *SubscriptionManagerImpl) materialize(subscriptionName, tagName string,
serverSpec.Configuration.SecuritySettings)
}
if s.config.DefaultDialerTag != "" {
senderSettings.ProxySettings = &internet.ProxyConfig{
Tag: s.config.DefaultDialerTag,
TransportLayerProxy: true,
}
}
outboundConf.SenderSettings = serial.ToTypedMessage(senderSettings)
outboundConf.ProxySettings = serverSpec.Configuration.ProtocolSettings
@@ -7,7 +7,9 @@ import (
"unicode"
"golang.org/x/crypto/sha3"
"google.golang.org/protobuf/proto"
"github.com/v2fly/v2ray-core/v5/app/subscription"
"github.com/v2fly/v2ray-core/v5/app/subscription/containers"
"github.com/v2fly/v2ray-core/v5/app/subscription/documentfetcher"
"github.com/v2fly/v2ray-core/v5/app/subscription/specs"
@@ -25,21 +27,29 @@ func (s *SubscriptionManagerImpl) updateSubscription(subscriptionName string) er
if err != nil {
return newError("failed to get fetcher: ", err)
}
if strings.HasPrefix(importSource.Url, "data:") {
isDataURL := strings.HasPrefix(importSource.Url, "data:")
if isDataURL {
docFetcher, err = documentfetcher.GetFetcher("dataurl")
if err != nil {
return newError("failed to get fetcher: ", err)
}
}
containerParser := ""
if isDataURL {
containerParser = "DataURLSingle"
}
downloadedDocument, err := docFetcher.DownloadDocument(s.ctx, importSource)
downloadedDocument, err := docFetcher.DownloadDocument(
s.ctx,
importSourceWithDefaultDialer(importSource, s.config.DefaultDialerTag),
)
if err != nil {
return newError("failed to download document: ", err)
}
trackedSub.originalDocument = downloadedDocument
container, err := containers.TryAllParsers(trackedSub.originalDocument, "")
container, err := containers.TryAllParsers(trackedSub.originalDocument, containerParser)
if err != nil {
return newError("failed to parse document: ", err)
}
@@ -72,6 +82,14 @@ func (s *SubscriptionManagerImpl) updateSubscription(subscriptionName string) er
return nil
}
func importSourceWithDefaultDialer(importSource *subscription.ImportSource, defaultDialerTag string) *subscription.ImportSource {
effectiveImportSource := proto.Clone(importSource).(*subscription.ImportSource)
if effectiveImportSource.ImportUsingTag == "" {
effectiveImportSource.ImportUsingTag = defaultDialerTag
}
return effectiveImportSource
}
func (s *SubscriptionManagerImpl) polyfillServerConfig(document *specs.SubscriptionServerConfig, hash string) {
document.Id = hash
+143 -38
View File
@@ -26,27 +26,105 @@ const (
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
type UDPBridgeConfig struct {
state protoimpl.MessageState `protogen:"open.v1"`
ListenAddress string `protobuf:"bytes,1,opt,name=listen_address,json=listenAddress,proto3" json:"listen_address,omitempty"`
ListenPort uint32 `protobuf:"varint,2,opt,name=listen_port,json=listenPort,proto3" json:"listen_port,omitempty"`
PeerAddress string `protobuf:"bytes,3,opt,name=peer_address,json=peerAddress,proto3" json:"peer_address,omitempty"`
PeerPort uint32 `protobuf:"varint,4,opt,name=peer_port,json=peerPort,proto3" json:"peer_port,omitempty"`
QueueSize uint32 `protobuf:"varint,5,opt,name=queue_size,json=queueSize,proto3" json:"queue_size,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *UDPBridgeConfig) Reset() {
*x = UDPBridgeConfig{}
mi := &file_app_tun_config_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *UDPBridgeConfig) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*UDPBridgeConfig) ProtoMessage() {}
func (x *UDPBridgeConfig) ProtoReflect() protoreflect.Message {
mi := &file_app_tun_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 UDPBridgeConfig.ProtoReflect.Descriptor instead.
func (*UDPBridgeConfig) Descriptor() ([]byte, []int) {
return file_app_tun_config_proto_rawDescGZIP(), []int{0}
}
func (x *UDPBridgeConfig) GetListenAddress() string {
if x != nil {
return x.ListenAddress
}
return ""
}
func (x *UDPBridgeConfig) GetListenPort() uint32 {
if x != nil {
return x.ListenPort
}
return 0
}
func (x *UDPBridgeConfig) GetPeerAddress() string {
if x != nil {
return x.PeerAddress
}
return ""
}
func (x *UDPBridgeConfig) GetPeerPort() uint32 {
if x != nil {
return x.PeerPort
}
return 0
}
func (x *UDPBridgeConfig) GetQueueSize() uint32 {
if x != nil {
return x.QueueSize
}
return 0
}
type Config struct {
state protoimpl.MessageState `protogen:"open.v1"`
Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
Mtu uint32 `protobuf:"varint,2,opt,name=mtu,proto3" json:"mtu,omitempty"`
UserLevel uint32 `protobuf:"varint,3,opt,name=user_level,json=userLevel,proto3" json:"user_level,omitempty"`
PacketEncoding packetaddr.PacketAddrType `protobuf:"varint,4,opt,name=packet_encoding,json=packetEncoding,proto3,enum=v2ray.core.net.packetaddr.PacketAddrType" json:"packet_encoding,omitempty"`
Tag string `protobuf:"bytes,5,opt,name=tag,proto3" json:"tag,omitempty"`
Ips []*routercommon.CIDR `protobuf:"bytes,6,rep,name=ips,proto3" json:"ips,omitempty"`
Routes []*routercommon.CIDR `protobuf:"bytes,7,rep,name=routes,proto3" json:"routes,omitempty"`
EnablePromiscuousMode bool `protobuf:"varint,8,opt,name=enable_promiscuous_mode,json=enablePromiscuousMode,proto3" json:"enable_promiscuous_mode,omitempty"`
EnableSpoofing bool `protobuf:"varint,9,opt,name=enable_spoofing,json=enableSpoofing,proto3" json:"enable_spoofing,omitempty"`
SocketSettings *internet.SocketConfig `protobuf:"bytes,10,opt,name=socket_settings,json=socketSettings,proto3" json:"socket_settings,omitempty"`
SniffingSettings *proxyman.SniffingConfig `protobuf:"bytes,11,opt,name=sniffing_settings,json=sniffingSettings,proto3" json:"sniffing_settings,omitempty"`
PreopenedFd *int32 `protobuf:"varint,12,opt,name=preopened_fd,json=preopenedFd,proto3,oneof" json:"preopened_fd,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
state protoimpl.MessageState `protogen:"open.v1"`
Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
Mtu uint32 `protobuf:"varint,2,opt,name=mtu,proto3" json:"mtu,omitempty"`
UserLevel uint32 `protobuf:"varint,3,opt,name=user_level,json=userLevel,proto3" json:"user_level,omitempty"`
PacketEncoding packetaddr.PacketAddrType `protobuf:"varint,4,opt,name=packet_encoding,json=packetEncoding,proto3,enum=v2ray.core.net.packetaddr.PacketAddrType" json:"packet_encoding,omitempty"`
Tag string `protobuf:"bytes,5,opt,name=tag,proto3" json:"tag,omitempty"`
Ips []*routercommon.CIDR `protobuf:"bytes,6,rep,name=ips,proto3" json:"ips,omitempty"`
Routes []*routercommon.CIDR `protobuf:"bytes,7,rep,name=routes,proto3" json:"routes,omitempty"`
EnablePromiscuousMode bool `protobuf:"varint,8,opt,name=enable_promiscuous_mode,json=enablePromiscuousMode,proto3" json:"enable_promiscuous_mode,omitempty"`
EnableSpoofing bool `protobuf:"varint,9,opt,name=enable_spoofing,json=enableSpoofing,proto3" json:"enable_spoofing,omitempty"`
SocketSettings *internet.SocketConfig `protobuf:"bytes,10,opt,name=socket_settings,json=socketSettings,proto3" json:"socket_settings,omitempty"`
SniffingSettings *proxyman.SniffingConfig `protobuf:"bytes,11,opt,name=sniffing_settings,json=sniffingSettings,proto3" json:"sniffing_settings,omitempty"`
PreopenedFd *int32 `protobuf:"varint,12,opt,name=preopened_fd,json=preopenedFd,proto3,oneof" json:"preopened_fd,omitempty"`
UdpBridge *UDPBridgeConfig `protobuf:"bytes,13,opt,name=udp_bridge,json=udpBridge,proto3" json:"udp_bridge,omitempty"`
PacketEncodingBypassPorts []uint32 `protobuf:"varint,14,rep,packed,name=packet_encoding_bypass_ports,json=packetEncodingBypassPorts,proto3" json:"packet_encoding_bypass_ports,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Config) Reset() {
*x = Config{}
mi := &file_app_tun_config_proto_msgTypes[0]
mi := &file_app_tun_config_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
@@ -58,7 +136,7 @@ func (x *Config) String() string {
func (*Config) ProtoMessage() {}
func (x *Config) ProtoReflect() protoreflect.Message {
mi := &file_app_tun_config_proto_msgTypes[0]
mi := &file_app_tun_config_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
@@ -71,7 +149,7 @@ func (x *Config) ProtoReflect() protoreflect.Message {
// Deprecated: Use Config.ProtoReflect.Descriptor instead.
func (*Config) Descriptor() ([]byte, []int) {
return file_app_tun_config_proto_rawDescGZIP(), []int{0}
return file_app_tun_config_proto_rawDescGZIP(), []int{1}
}
func (x *Config) GetName() string {
@@ -158,11 +236,33 @@ func (x *Config) GetPreopenedFd() int32 {
return 0
}
func (x *Config) GetUdpBridge() *UDPBridgeConfig {
if x != nil {
return x.UdpBridge
}
return nil
}
func (x *Config) GetPacketEncodingBypassPorts() []uint32 {
if x != nil {
return x.PacketEncodingBypassPorts
}
return nil
}
var File_app_tun_config_proto protoreflect.FileDescriptor
const file_app_tun_config_proto_rawDesc = "" +
"\n" +
"\x14app/tun/config.proto\x12\x12v2ray.core.app.tun\x1a\x19app/proxyman/config.proto\x1a$app/router/routercommon/common.proto\x1a common/protoext/extensions.proto\x1a\"common/net/packetaddr/config.proto\x1a\x1ftransport/internet/config.proto\"\x8b\x05\n" +
"\x14app/tun/config.proto\x12\x12v2ray.core.app.tun\x1a\x19app/proxyman/config.proto\x1a$app/router/routercommon/common.proto\x1a common/protoext/extensions.proto\x1a\"common/net/packetaddr/config.proto\x1a\x1ftransport/internet/config.proto\"\xb8\x01\n" +
"\x0fUDPBridgeConfig\x12%\n" +
"\x0elisten_address\x18\x01 \x01(\tR\rlistenAddress\x12\x1f\n" +
"\vlisten_port\x18\x02 \x01(\rR\n" +
"listenPort\x12!\n" +
"\fpeer_address\x18\x03 \x01(\tR\vpeerAddress\x12\x1b\n" +
"\tpeer_port\x18\x04 \x01(\rR\bpeerPort\x12\x1d\n" +
"\n" +
"queue_size\x18\x05 \x01(\rR\tqueueSize\"\x90\x06\n" +
"\x06Config\x12\x12\n" +
"\x04name\x18\x01 \x01(\tR\x04name\x12\x10\n" +
"\x03mtu\x18\x02 \x01(\rR\x03mtu\x12\x1d\n" +
@@ -177,7 +277,10 @@ const file_app_tun_config_proto_rawDesc = "" +
"\x0fsocket_settings\x18\n" +
" \x01(\v2+.v2ray.core.transport.internet.SocketConfigR\x0esocketSettings\x12T\n" +
"\x11sniffing_settings\x18\v \x01(\v2'.v2ray.core.app.proxyman.SniffingConfigR\x10sniffingSettings\x12&\n" +
"\fpreopened_fd\x18\f \x01(\x05H\x00R\vpreopenedFd\x88\x01\x01:\x12\x82\xb5\x18\x0e\n" +
"\fpreopened_fd\x18\f \x01(\x05H\x00R\vpreopenedFd\x88\x01\x01\x12B\n" +
"\n" +
"udp_bridge\x18\r \x01(\v2#.v2ray.core.app.tun.UDPBridgeConfigR\tudpBridge\x12?\n" +
"\x1cpacket_encoding_bypass_ports\x18\x0e \x03(\rR\x19packetEncodingBypassPorts:\x12\x82\xb5\x18\x0e\n" +
"\aservice\x12\x03tunB\x0f\n" +
"\r_preopened_fdBW\n" +
"\x16com.v2ray.core.app.tunP\x01Z&github.com/v2fly/v2ray-core/v5/app/tun\xaa\x02\x12V2Ray.Core.App.Tunb\x06proto3"
@@ -194,25 +297,27 @@ func file_app_tun_config_proto_rawDescGZIP() []byte {
return file_app_tun_config_proto_rawDescData
}
var file_app_tun_config_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
var file_app_tun_config_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
var file_app_tun_config_proto_goTypes = []any{
(*Config)(nil), // 0: v2ray.core.app.tun.Config
(packetaddr.PacketAddrType)(0), // 1: v2ray.core.net.packetaddr.PacketAddrType
(*routercommon.CIDR)(nil), // 2: v2ray.core.app.router.routercommon.CIDR
(*internet.SocketConfig)(nil), // 3: v2ray.core.transport.internet.SocketConfig
(*proxyman.SniffingConfig)(nil), // 4: v2ray.core.app.proxyman.SniffingConfig
(*UDPBridgeConfig)(nil), // 0: v2ray.core.app.tun.UDPBridgeConfig
(*Config)(nil), // 1: v2ray.core.app.tun.Config
(packetaddr.PacketAddrType)(0), // 2: v2ray.core.net.packetaddr.PacketAddrType
(*routercommon.CIDR)(nil), // 3: v2ray.core.app.router.routercommon.CIDR
(*internet.SocketConfig)(nil), // 4: v2ray.core.transport.internet.SocketConfig
(*proxyman.SniffingConfig)(nil), // 5: v2ray.core.app.proxyman.SniffingConfig
}
var file_app_tun_config_proto_depIdxs = []int32{
1, // 0: v2ray.core.app.tun.Config.packet_encoding:type_name -> v2ray.core.net.packetaddr.PacketAddrType
2, // 1: v2ray.core.app.tun.Config.ips:type_name -> v2ray.core.app.router.routercommon.CIDR
2, // 2: v2ray.core.app.tun.Config.routes:type_name -> v2ray.core.app.router.routercommon.CIDR
3, // 3: v2ray.core.app.tun.Config.socket_settings:type_name -> v2ray.core.transport.internet.SocketConfig
4, // 4: v2ray.core.app.tun.Config.sniffing_settings:type_name -> v2ray.core.app.proxyman.SniffingConfig
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
2, // 0: v2ray.core.app.tun.Config.packet_encoding:type_name -> v2ray.core.net.packetaddr.PacketAddrType
3, // 1: v2ray.core.app.tun.Config.ips:type_name -> v2ray.core.app.router.routercommon.CIDR
3, // 2: v2ray.core.app.tun.Config.routes:type_name -> v2ray.core.app.router.routercommon.CIDR
4, // 3: v2ray.core.app.tun.Config.socket_settings:type_name -> v2ray.core.transport.internet.SocketConfig
5, // 4: v2ray.core.app.tun.Config.sniffing_settings:type_name -> v2ray.core.app.proxyman.SniffingConfig
0, // 5: v2ray.core.app.tun.Config.udp_bridge:type_name -> v2ray.core.app.tun.UDPBridgeConfig
6, // [6:6] is the sub-list for method output_type
6, // [6:6] is the sub-list for method input_type
6, // [6:6] is the sub-list for extension type_name
6, // [6:6] is the sub-list for extension extendee
0, // [0:6] is the sub-list for field type_name
}
func init() { file_app_tun_config_proto_init() }
@@ -220,14 +325,14 @@ func file_app_tun_config_proto_init() {
if File_app_tun_config_proto != nil {
return
}
file_app_tun_config_proto_msgTypes[0].OneofWrappers = []any{}
file_app_tun_config_proto_msgTypes[1].OneofWrappers = []any{}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_app_tun_config_proto_rawDesc), len(file_app_tun_config_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
+10
View File
@@ -12,6 +12,14 @@ import "common/protoext/extensions.proto";
import "common/net/packetaddr/config.proto";
import "transport/internet/config.proto";
message UDPBridgeConfig {
string listen_address = 1;
uint32 listen_port = 2;
string peer_address = 3;
uint32 peer_port = 4;
uint32 queue_size = 5;
}
message Config {
option (v2ray.core.common.protoext.message_opt).type = "service";
option (v2ray.core.common.protoext.message_opt).short_name = "tun";
@@ -28,4 +36,6 @@ message Config {
v2ray.core.transport.internet.SocketConfig socket_settings = 10;
v2ray.core.app.proxyman.SniffingConfig sniffing_settings = 11;
optional int32 preopened_fd = 12;
UDPBridgeConfig udp_bridge = 13;
repeated uint32 packet_encoding_bypass_ports = 14;
}
+90
View File
@@ -0,0 +1,90 @@
package tun
import (
"context"
"fmt"
"testing"
"google.golang.org/protobuf/encoding/protojson"
"github.com/v2fly/v2ray-core/v5/common/session"
)
func TestUDPBridgeConfigJSON(t *testing.T) {
var config Config
if err := protojson.Unmarshal([]byte(`{
"name": "packetbridge",
"mtu": 1500,
"udp_bridge": {
"listen_address": "127.0.0.1",
"listen_port": 9090,
"peer_address": "127.0.0.1",
"peer_port": 9091,
"queue_size": 512
}
}`), &config); err != nil {
t.Fatalf("failed to parse config: %v", err)
}
if config.UdpBridge == nil {
t.Fatal("udp_bridge was not decoded")
}
if config.UdpBridge.ListenPort != 9090 || config.UdpBridge.PeerPort != 9091 {
t.Fatalf("unexpected bridge ports: %+v", config.UdpBridge)
}
}
func TestPacketEncodingBypassPortsJSON(t *testing.T) {
var config Config
if err := protojson.Unmarshal([]byte(`{
"packet_encoding": "Stream",
"packet_encoding_bypass_ports": [53, 123]
}`), &config); err != nil {
t.Fatalf("failed to parse config: %v", err)
}
if got := config.PacketEncodingBypassPorts; len(got) != 2 || got[0] != 53 || got[1] != 123 {
t.Fatalf("packet_encoding_bypass_ports = %v, want [53 123]", got)
}
}
func TestUDPBridgeRejectsPreopenedFD(t *testing.T) {
fd := int32(3)
tun := new(TUN)
err := tun.Init(context.Background(), &Config{
PreopenedFd: &fd,
UdpBridge: &UDPBridgeConfig{},
}, nil, nil)
if err == nil {
t.Fatal("expected preopened_fd/udp_bridge conflict")
}
}
func TestPacketEncodingContextCarriesTUNInboundTag(t *testing.T) {
tun := &TUN{
ctx: context.Background(),
config: &Config{Tag: "packet-bridge-in"},
}
inbound := session.InboundFromContext(tun.packetEncodingContext())
if inbound == nil {
t.Fatal("packet encoding context has no inbound session")
}
if inbound.Tag != tun.config.Tag {
t.Fatalf("packet encoding context inbound tag = %q, want %q", inbound.Tag, tun.config.Tag)
}
}
func TestPacketEncodingBypassPortsValidation(t *testing.T) {
for _, port := range []uint32{0, 65536} {
t.Run(fmt.Sprint(port), func(t *testing.T) {
tun := new(TUN)
err := tun.Init(context.Background(), &Config{
PacketEncodingBypassPorts: []uint32{port},
}, nil, nil)
if err == nil {
t.Fatalf("expected packet_encoding_bypass_ports value %d to be rejected", port)
}
})
}
}
@@ -0,0 +1,9 @@
package udpbridge
import "github.com/v2fly/v2ray-core/v5/common/errors"
type errPathObjHolder struct{}
func newError(values ...interface{}) *errors.Error {
return errors.New(values...).WithPathObj(errPathObjHolder{})
}
+251
View File
@@ -0,0 +1,251 @@
package udpbridge
import (
"context"
"encoding/binary"
"net"
"sync"
"gvisor.dev/gvisor/pkg/buffer"
"gvisor.dev/gvisor/pkg/tcpip"
"gvisor.dev/gvisor/pkg/tcpip/header"
"gvisor.dev/gvisor/pkg/tcpip/link/channel"
"gvisor.dev/gvisor/pkg/tcpip/stack"
"github.com/v2fly/v2ray-core/v5/app/tun/device"
)
const (
defaultMTU = 1500
minimumMTU = 576
maximumMTU = 9000
defaultQueueSize = 512
maximumQueueSize = 4096
)
// Options describes the loopback UDP transport used to exchange raw IP
// packets with an external packet-tunnel provider.
type Options struct {
ListenAddress string
ListenPort uint32
PeerAddress string
PeerPort uint32
QueueSize uint32
}
// UDPBridge is a gVisor link endpoint backed by a pair of loopback UDP ports.
// Each UDP datagram contains exactly one complete IPv4 or IPv6 packet.
type UDPBridge struct {
*channel.Endpoint
conn *net.UDPConn
peer *net.UDPAddr
mtu uint32
ctx context.Context
cancel context.CancelFunc
closeOnce sync.Once
wg sync.WaitGroup
}
var _ device.Device = (*UDPBridge)(nil)
// New creates and starts a UDP-backed link endpoint.
func New(deviceOptions device.Options, options Options) (*UDPBridge, error) {
if deviceOptions.PreopenedFDSet {
return nil, newError("preopened file descriptors are not supported").AtError()
}
mtu := deviceOptions.MTU
if mtu == 0 {
mtu = defaultMTU
}
if mtu < minimumMTU || mtu > maximumMTU {
return nil, newError("MTU must be between ", minimumMTU, " and ", maximumMTU).AtError()
}
queueSize := options.QueueSize
if queueSize == 0 {
queueSize = defaultQueueSize
}
if queueSize > maximumQueueSize {
return nil, newError("queue_size must not exceed ", maximumQueueSize).AtError()
}
if options.ListenPort == 0 || options.ListenPort > 65535 {
return nil, newError("listen_port must be between 1 and 65535").AtError()
}
if options.PeerPort == 0 || options.PeerPort > 65535 {
return nil, newError("peer_port must be between 1 and 65535").AtError()
}
if options.ListenPort == options.PeerPort {
return nil, newError("listen_port and peer_port must be different").AtError()
}
listenIP, listenIsIPv4, err := parseLoopbackIP(options.ListenAddress)
if err != nil {
return nil, newError("invalid listen_address").Base(err).AtError()
}
peerIP, peerIsIPv4, err := parseLoopbackIP(options.PeerAddress)
if err != nil {
return nil, newError("invalid peer_address").Base(err).AtError()
}
if listenIsIPv4 != peerIsIPv4 {
return nil, newError("listen_address and peer_address must use the same address family").AtError()
}
network := "udp6"
if listenIsIPv4 {
network = "udp4"
}
listenAddress := &net.UDPAddr{IP: listenIP, Port: int(options.ListenPort)}
peerAddress := &net.UDPAddr{IP: peerIP, Port: int(options.PeerPort)}
conn, err := net.ListenUDP(network, listenAddress)
if err != nil {
return nil, newError("failed to listen on ", listenAddress).Base(err).AtError()
}
ctx, cancel := context.WithCancel(context.Background())
bridge := &UDPBridge{
Endpoint: channel.New(int(queueSize), mtu, ""),
conn: conn,
peer: peerAddress,
mtu: mtu,
ctx: ctx,
cancel: cancel,
}
bridge.wg.Add(2)
go bridge.readLoop()
go bridge.writeLoop()
return bridge, nil
}
func parseLoopbackIP(value string) (net.IP, bool, error) {
ip := net.ParseIP(value)
if ip == nil {
return nil, false, newError("address must be an IP literal")
}
if !ip.IsLoopback() {
return nil, false, newError("address must be loopback")
}
if ipv4 := ip.To4(); ipv4 != nil {
return ipv4, true, nil
}
return ip.To16(), false, nil
}
func (b *UDPBridge) readLoop() {
defer b.wg.Done()
datagram := make([]byte, int(b.mtu)+1)
for {
n, source, err := b.conn.ReadFromUDP(datagram)
if err != nil {
if b.ctx.Err() != nil {
return
}
newError("failed to read UDP packet").Base(err).AtWarning().WriteToLog()
continue
}
if !sameUDPAddress(source, b.peer) {
newError("discarding UDP packet from unexpected peer ", source).AtDebug().WriteToLog()
continue
}
if n > int(b.mtu) {
newError("discarding packet larger than MTU").AtDebug().WriteToLog()
continue
}
protocol, ok := rawIPProtocol(datagram[:n])
if !ok {
newError("discarding malformed raw IP packet").AtDebug().WriteToLog()
continue
}
packet := stack.NewPacketBuffer(stack.PacketBufferOptions{
Payload: buffer.MakeWithData(datagram[:n]),
})
b.InjectInbound(protocol, packet)
packet.DecRef()
}
}
func (b *UDPBridge) writeLoop() {
defer b.wg.Done()
for {
packet := b.ReadContext(b.ctx)
if packet == nil {
return
}
slices := packet.AsSlices()
size := 0
for _, part := range slices {
size += len(part)
}
if size > int(b.mtu) {
packet.DecRef()
newError("discarding outbound packet larger than MTU").AtWarning().WriteToLog()
continue
}
datagram := make([]byte, size)
offset := 0
for _, part := range slices {
offset += copy(datagram[offset:], part)
}
packet.DecRef()
if _, err := b.conn.WriteToUDP(datagram, b.peer); err != nil {
if b.ctx.Err() != nil {
return
}
newError("failed to write UDP packet").Base(err).AtWarning().WriteToLog()
}
}
}
func sameUDPAddress(left, right *net.UDPAddr) bool {
return left != nil && right != nil && left.Port == right.Port && left.IP.Equal(right.IP)
}
func rawIPProtocol(packet []byte) (tcpip.NetworkProtocolNumber, bool) {
if len(packet) == 0 {
return 0, false
}
switch packet[0] >> 4 {
case 4:
if len(packet) < header.IPv4MinimumSize {
return 0, false
}
headerLength := int(packet[0]&0x0f) * 4
totalLength := int(binary.BigEndian.Uint16(packet[2:4]))
if headerLength < header.IPv4MinimumSize || totalLength != len(packet) || totalLength < headerLength {
return 0, false
}
return header.IPv4ProtocolNumber, true
case 6:
if len(packet) < header.IPv6MinimumSize {
return 0, false
}
totalLength := header.IPv6MinimumSize + int(binary.BigEndian.Uint16(packet[4:6]))
if totalLength != len(packet) {
return 0, false
}
return header.IPv6ProtocolNumber, true
default:
return 0, false
}
}
// Close stops both packet pumps and releases the UDP port. It is safe to call
// more than once.
func (b *UDPBridge) Close() {
b.closeOnce.Do(func() {
b.cancel()
_ = b.conn.Close()
b.Endpoint.Close()
b.wg.Wait()
})
}
+304
View File
@@ -0,0 +1,304 @@
package udpbridge
import (
"bytes"
"encoding/binary"
"net"
"testing"
"time"
"gvisor.dev/gvisor/pkg/buffer"
"gvisor.dev/gvisor/pkg/tcpip"
"gvisor.dev/gvisor/pkg/tcpip/header"
"gvisor.dev/gvisor/pkg/tcpip/stack"
"github.com/v2fly/v2ray-core/v5/app/tun/device"
)
type deliveredPacket struct {
protocol tcpip.NetworkProtocolNumber
payload []byte
}
type recordingDispatcher struct {
packets chan deliveredPacket
}
func (d *recordingDispatcher) DeliverNetworkPacket(protocol tcpip.NetworkProtocolNumber, packet *stack.PacketBuffer) {
packetBuffer := packet.ToBuffer()
d.packets <- deliveredPacket{
protocol: protocol,
payload: packetBuffer.Flatten(),
}
}
func (*recordingDispatcher) DeliverLinkPacket(tcpip.NetworkProtocolNumber, *stack.PacketBuffer) {}
func TestUDPBridgeBidirectionalIPv4(t *testing.T) {
bridge, peer := newTestBridge(t)
defer bridge.Close()
defer peer.Close()
dispatcher := &recordingDispatcher{packets: make(chan deliveredPacket, 1)}
bridge.Attach(dispatcher)
ipv4Packet := validIPv4Packet()
if _, err := peer.WriteToUDP(ipv4Packet, bridge.conn.LocalAddr().(*net.UDPAddr)); err != nil {
t.Fatalf("failed to send inbound packet: %v", err)
}
select {
case delivered := <-dispatcher.packets:
if delivered.protocol != header.IPv4ProtocolNumber {
t.Fatalf("unexpected protocol: %v", delivered.protocol)
}
if !bytes.Equal(delivered.payload, ipv4Packet) {
t.Fatalf("unexpected inbound payload: %x", delivered.payload)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for inbound packet")
}
packet := stack.NewPacketBuffer(stack.PacketBufferOptions{
Payload: buffer.MakeWithData(ipv4Packet),
})
list := stack.PacketBufferList{}
list.PushBack(packet)
written, writeErr := bridge.WritePackets(list)
packet.DecRef()
if writeErr != nil {
t.Fatalf("WritePackets failed: %v", writeErr)
}
if written != 1 {
t.Fatalf("unexpected write count: %d", written)
}
if err := peer.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
t.Fatalf("failed to set deadline: %v", err)
}
received := make([]byte, 2048)
n, source, err := peer.ReadFromUDP(received)
if err != nil {
t.Fatalf("failed to receive outbound packet: %v", err)
}
if source.Port != bridge.conn.LocalAddr().(*net.UDPAddr).Port {
t.Fatalf("outbound packet came from port %d, want %d", source.Port, bridge.conn.LocalAddr().(*net.UDPAddr).Port)
}
if !bytes.Equal(received[:n], ipv4Packet) {
t.Fatalf("unexpected outbound payload: %x", received[:n])
}
}
func TestUDPBridgeFiltersPeerAndMalformedPackets(t *testing.T) {
bridge, peer := newTestBridge(t)
defer bridge.Close()
defer peer.Close()
dispatcher := &recordingDispatcher{packets: make(chan deliveredPacket, 3)}
bridge.Attach(dispatcher)
destination := bridge.conn.LocalAddr().(*net.UDPAddr)
unexpectedPeer, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0})
if err != nil {
t.Fatalf("failed to create unexpected peer: %v", err)
}
defer unexpectedPeer.Close()
if _, err := unexpectedPeer.WriteToUDP(validIPv4Packet(), destination); err != nil {
t.Fatalf("failed to send unexpected-peer packet: %v", err)
}
if _, err := peer.WriteToUDP([]byte{0x45, 0x00}, destination); err != nil {
t.Fatalf("failed to send malformed packet: %v", err)
}
ipv6Packet := validIPv6Packet()
if _, err := peer.WriteToUDP(ipv6Packet, destination); err != nil {
t.Fatalf("failed to send IPv6 packet: %v", err)
}
select {
case delivered := <-dispatcher.packets:
if delivered.protocol != header.IPv6ProtocolNumber || !bytes.Equal(delivered.payload, ipv6Packet) {
t.Fatalf("unexpected delivered packet: protocol=%v payload=%x", delivered.protocol, delivered.payload)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for valid packet")
}
select {
case extra := <-dispatcher.packets:
t.Fatalf("unexpected extra packet: protocol=%v payload=%x", extra.protocol, extra.payload)
case <-time.After(100 * time.Millisecond):
}
}
func TestUDPBridgeValidationAndPortConflict(t *testing.T) {
deviceOptions := device.Options{MTU: 1500}
validOptions := Options{
ListenAddress: "127.0.0.1",
ListenPort: 19090,
PeerAddress: "127.0.0.1",
PeerPort: 19091,
}
tests := []struct {
name string
deviceOptions device.Options
options Options
}{
{name: "non-loopback", deviceOptions: deviceOptions, options: withListenAddress(validOptions, "192.0.2.1")},
{name: "hostname", deviceOptions: deviceOptions, options: withListenAddress(validOptions, "localhost")},
{name: "missing listen port", deviceOptions: deviceOptions, options: withListenPort(validOptions, 0)},
{name: "same ports", deviceOptions: deviceOptions, options: withPeerPort(validOptions, validOptions.ListenPort)},
{name: "mixed families", deviceOptions: deviceOptions, options: withPeerAddress(validOptions, "::1")},
{name: "queue too large", deviceOptions: deviceOptions, options: withQueueSize(validOptions, maximumQueueSize+1)},
{name: "MTU too small", deviceOptions: device.Options{MTU: minimumMTU - 1}, options: validOptions},
{name: "preopened fd", deviceOptions: device.Options{MTU: 1500, PreopenedFDSet: true}, options: validOptions},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
bridge, err := New(test.deviceOptions, test.options)
if err == nil {
bridge.Close()
t.Fatal("expected validation error")
}
})
}
occupied, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0})
if err != nil {
t.Fatalf("failed to reserve port: %v", err)
}
defer occupied.Close()
options := validOptions
options.ListenPort = uint32(occupied.LocalAddr().(*net.UDPAddr).Port)
if bridge, err := New(deviceOptions, options); err == nil {
bridge.Close()
t.Fatal("expected occupied-port error")
}
}
func TestUDPBridgeCloseIsIdempotent(t *testing.T) {
bridge, peer := newTestBridge(t)
defer peer.Close()
done := make(chan struct{})
go func() {
bridge.Close()
bridge.Close()
close(done)
}()
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("Close did not unblock packet pumps")
}
}
func TestRawIPProtocol(t *testing.T) {
tests := []struct {
name string
packet []byte
protocol tcpip.NetworkProtocolNumber
valid bool
}{
{name: "IPv4", packet: validIPv4Packet(), protocol: header.IPv4ProtocolNumber, valid: true},
{name: "IPv6", packet: validIPv6Packet(), protocol: header.IPv6ProtocolNumber, valid: true},
{name: "empty"},
{name: "unknown version", packet: []byte{0x70}},
{name: "short IPv4", packet: []byte{0x45}},
{name: "invalid IPv4 length", packet: append(validIPv4Packet(), 0)},
{name: "short IPv6", packet: []byte{0x60}},
{name: "invalid IPv6 length", packet: append(validIPv6Packet(), 0)},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
protocol, valid := rawIPProtocol(test.packet)
if valid != test.valid || protocol != test.protocol {
t.Fatalf("rawIPProtocol() = (%v, %v), want (%v, %v)", protocol, valid, test.protocol, test.valid)
}
})
}
}
func newTestBridge(t *testing.T) (*UDPBridge, *net.UDPConn) {
t.Helper()
peer, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0})
if err != nil {
t.Fatalf("failed to bind peer: %v", err)
}
listenPort := unusedUDPPort(t)
peerPort := peer.LocalAddr().(*net.UDPAddr).Port
if listenPort == peerPort {
listenPort = unusedUDPPort(t)
}
bridge, err := New(device.Options{MTU: 1500}, Options{
ListenAddress: "127.0.0.1",
ListenPort: uint32(listenPort),
PeerAddress: "127.0.0.1",
PeerPort: uint32(peerPort),
QueueSize: 8,
})
if err != nil {
peer.Close()
t.Fatalf("failed to create bridge: %v", err)
}
return bridge, peer
}
func unusedUDPPort(t *testing.T) int {
t.Helper()
conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0})
if err != nil {
t.Fatalf("failed to select UDP port: %v", err)
}
port := conn.LocalAddr().(*net.UDPAddr).Port
if err := conn.Close(); err != nil {
t.Fatalf("failed to release UDP port: %v", err)
}
return port
}
func validIPv4Packet() []byte {
packet := make([]byte, header.IPv4MinimumSize)
packet[0] = 0x45
binary.BigEndian.PutUint16(packet[2:4], uint16(len(packet)))
return packet
}
func validIPv6Packet() []byte {
packet := make([]byte, header.IPv6MinimumSize)
packet[0] = 0x60
binary.BigEndian.PutUint16(packet[4:6], uint16(len(packet)-header.IPv6MinimumSize))
return packet
}
func withListenAddress(options Options, value string) Options {
options.ListenAddress = value
return options
}
func withListenPort(options Options, value uint32) Options {
options.ListenPort = value
return options
}
func withPeerAddress(options Options, value string) Options {
options.PeerAddress = value
return options
}
func withPeerPort(options Options, value uint32) Options {
options.PeerPort = value
return options
}
func withQueueSize(options Options, value uint32) Options {
options.QueueSize = value
return options
}
+2 -2
View File
@@ -27,8 +27,8 @@ type packetAddrDevice struct {
func (p *packetAddrDevice) DeliverNetworkPacket(protocol tcpip.NetworkProtocolNumber, pkt *stack.PacketBuffer) {
buf := pkt.ToBuffer()
_, err := p.sorter.OnPacketReceived(buf.Flatten())
if err != nil {
handled, err := p.sorter.OnPacketReceived(buf.Flatten())
if err != nil || !handled {
p.dispatcherAccess.RLock()
dispatcher := p.secondaryDispatcher
p.dispatcherAccess.RUnlock()
+42 -7
View File
@@ -11,9 +11,12 @@ import (
core "github.com/v2fly/v2ray-core/v5"
"github.com/v2fly/v2ray-core/v5/app/tun/device"
"github.com/v2fly/v2ray-core/v5/app/tun/device/gvisor"
"github.com/v2fly/v2ray-core/v5/app/tun/device/udpbridge"
"github.com/v2fly/v2ray-core/v5/app/tun/tunsorter"
"github.com/v2fly/v2ray-core/v5/common"
"github.com/v2fly/v2ray-core/v5/common/net"
"github.com/v2fly/v2ray-core/v5/common/net/packetaddr"
"github.com/v2fly/v2ray-core/v5/common/session"
"github.com/v2fly/v2ray-core/v5/features/policy"
"github.com/v2fly/v2ray-core/v5/features/routing"
)
@@ -26,10 +29,11 @@ type TUN struct {
policyManager policy.Manager
config *Config
stack *stack.Stack
device device.Device
preopenedFD int
preopenedFDSet bool
stack *stack.Stack
device device.Device
preopenedFD int
preopenedFDSet bool
packetEncodingBypassPorts []net.Port
}
func (t *TUN) Type() interface{} {
@@ -37,7 +41,6 @@ func (t *TUN) Type() interface{} {
}
func (t *TUN) Start() error {
DeviceConstructor := gvisor.New
deviceOptions := device.Options{
Name: t.config.Name,
MTU: t.config.Mtu,
@@ -49,7 +52,19 @@ func (t *TUN) Start() error {
t.preopenedFDSet = false
}
tunDevice, err := DeviceConstructor(deviceOptions)
var tunDevice device.Device
var err error
if bridge := t.config.UdpBridge; bridge != nil {
tunDevice, err = udpbridge.New(deviceOptions, udpbridge.Options{
ListenAddress: bridge.ListenAddress,
ListenPort: bridge.ListenPort,
PeerAddress: bridge.PeerAddress,
PeerPort: bridge.PeerPort,
QueueSize: bridge.QueueSize,
})
} else {
tunDevice, err = gvisor.New(deviceOptions)
}
if err != nil {
return newError("failed to create device").Base(err).AtError()
}
@@ -57,7 +72,13 @@ func (t *TUN) Start() error {
if t.config.PacketEncoding != packetaddr.PacketAddrType_None {
writer := device.NewLinkWriterToWriter(tunDevice)
sorter := tunsorter.NewTunSorter(writer, t.dispatcher, t.config.PacketEncoding, t.ctx)
sorter := tunsorter.NewTunSorter(
writer,
t.dispatcher,
t.config.PacketEncoding,
t.packetEncodingContext(),
t.packetEncodingBypassPorts,
)
tunDeviceLayered := NewDeviceWithSorter(tunDevice, sorter)
tunDevice = tunDeviceLayered
}
@@ -75,6 +96,10 @@ func (t *TUN) Start() error {
return nil
}
func (t *TUN) packetEncodingContext() context.Context {
return session.ContextWithInbound(t.ctx, &session.Inbound{Tag: t.config.Tag})
}
func (t *TUN) Close() error {
if t.stack != nil {
t.stack.Close()
@@ -100,7 +125,17 @@ func (t *TUN) Init(ctx context.Context, config *Config, dispatcher routing.Dispa
t.dispatcher = dispatcher
t.policyManager = policyManager
t.preopenedFD = -1
t.packetEncodingBypassPorts = make([]net.Port, 0, len(config.PacketEncodingBypassPorts))
for _, port := range config.PacketEncodingBypassPorts {
if port == 0 || port > 65535 {
return newError("invalid packet_encoding_bypass_ports value: ", port).AtError()
}
t.packetEncodingBypassPorts = append(t.packetEncodingBypassPorts, net.Port(port))
}
if config.PreopenedFd != nil {
if config.UdpBridge != nil {
return newError("preopened_fd and udp_bridge cannot be used together").AtError()
}
if *config.PreopenedFd < 0 {
return newError("invalid preopened_fd: ", *config.PreopenedFd).AtError()
}
+29 -11
View File
@@ -16,28 +16,46 @@ import (
//go:generate go run github.com/v2fly/v2ray-core/v5/common/errors/errorgen
func NewTunSorter(tunWriter io.Writer, dispatcher routing.Dispatcher, packetAddrType packetaddr.PacketAddrType, ctx context.Context) *TunSorter {
func NewTunSorter(
tunWriter io.Writer,
dispatcher routing.Dispatcher,
packetAddrType packetaddr.PacketAddrType,
ctx context.Context,
packetEncodingBypassPorts []net.Port,
) *TunSorter {
bypassPorts := make(map[net.Port]struct{}, len(packetEncodingBypassPorts))
for _, port := range packetEncodingBypassPorts {
bypassPorts[port] = struct{}{}
}
return &TunSorter{
tunWriter: tunWriter,
dispatcher: dispatcher,
packetAddrType: packetAddrType,
ctx: ctx,
tunWriter: tunWriter,
dispatcher: dispatcher,
packetAddrType: packetAddrType,
packetEncodingBypassPorts: bypassPorts,
ctx: ctx,
}
}
type TunSorter struct {
tunWriter io.Writer
dispatcher routing.Dispatcher
packetAddrType packetaddr.PacketAddrType
tunWriter io.Writer
dispatcher routing.Dispatcher
packetAddrType packetaddr.PacketAddrType
packetEncodingBypassPorts map[net.Port]struct{}
trackedConnections sync.Map
ctx context.Context
}
func (t *TunSorter) OnPacketReceived(b []byte) (n int, err error) {
// OnPacketReceived handles UDP packets that should use packetaddr encoding.
// It returns handled=false when the packet is not UDP or its destination port
// is configured to use the regular TUN UDP path.
func (t *TunSorter) OnPacketReceived(b []byte) (handled bool, err error) {
src, dst, data, err := packetparse.TryParseAsUDPPacket(b)
if err != nil {
return 0, err
return false, err
}
if _, bypass := t.packetEncodingBypassPorts[dst.Port]; bypass {
return false, nil
}
conn := newTrackedUDPConnection(src, t)
trackedConnection, loaded := t.trackedConnections.LoadOrStore(src.String(), conn)
@@ -46,7 +64,7 @@ func (t *TunSorter) OnPacketReceived(b []byte) (n int, err error) {
t.onNewConnection(conn)
}
conn.onNewPacket(dst, data)
return len(b), nil
return true, nil
}
func (t *TunSorter) onNewConnection(connection *trackedUDPConnection) {
+150
View File
@@ -0,0 +1,150 @@
package tunsorter
import (
"context"
"io"
"testing"
"time"
"github.com/v2fly/v2ray-core/v5/app/tun/packetparse"
"github.com/v2fly/v2ray-core/v5/common/net"
"github.com/v2fly/v2ray-core/v5/common/net/packetaddr"
"github.com/v2fly/v2ray-core/v5/features/routing"
"github.com/v2fly/v2ray-core/v5/transport"
"github.com/v2fly/v2ray-core/v5/transport/pipe"
)
type recordingDispatcher struct {
destination net.Destination
downlinkWriter *pipe.Writer
dispatchCount int
}
func (d *recordingDispatcher) Dispatch(_ context.Context, destination net.Destination) (*transport.Link, error) {
d.destination = destination
d.dispatchCount++
_, uplinkWriter := pipe.New(pipe.WithSizeLimit(1024))
downlinkReader, downlinkWriter := pipe.New(pipe.WithSizeLimit(1024))
d.downlinkWriter = downlinkWriter
return &transport.Link{Reader: downlinkReader, Writer: uplinkWriter}, nil
}
func (*recordingDispatcher) Start() error { return nil }
func (*recordingDispatcher) Close() error { return nil }
func (*recordingDispatcher) Type() interface{} { return routing.DispatcherType() }
func TestPacketEncodingBypassPort(t *testing.T) {
sorter := NewTunSorter(
io.Discard,
nil,
packetaddr.PacketAddrType_Stream,
context.Background(),
[]net.Port{53},
)
src := net.UDPDestination(net.ParseAddress("198.18.0.2"), 49152)
dst := net.UDPDestination(net.ParseAddress("1.1.1.1"), 53)
packet, err := packetparse.TryConstructUDPPacket(src, dst, []byte("dns query"))
if err != nil {
t.Fatalf("failed to construct UDP packet: %v", err)
}
handled, err := sorter.OnPacketReceived(packet)
if err != nil {
t.Fatalf("OnPacketReceived returned an error: %v", err)
}
if handled {
t.Fatal("DNS packet was handled by packetaddr instead of being bypassed")
}
}
func TestUnlistedPortStillUsesStreamPacketEncoding(t *testing.T) {
dispatcher := new(recordingDispatcher)
sorter := NewTunSorter(io.Discard, dispatcher, packetaddr.PacketAddrType_Stream, context.Background(), []net.Port{53})
src := net.UDPDestination(net.ParseAddress("198.18.0.2"), 49152)
dst := net.UDPDestination(net.ParseAddress("1.1.1.1"), 443)
packet, err := packetparse.TryConstructUDPPacket(src, dst, []byte("quic packet"))
if err != nil {
t.Fatalf("failed to construct UDP packet: %v", err)
}
handled, err := sorter.OnPacketReceived(packet)
if err != nil {
t.Fatalf("OnPacketReceived returned an error: %v", err)
}
if !handled {
t.Fatal("unlisted UDP port bypassed packet encoding")
}
if dispatcher.destination.Network != net.Network_TCP || dispatcher.destination.Address.Domain() != "st.packet-addr.v2fly.arpa" || dispatcher.destination.Port != 0 {
t.Fatalf("packetaddr destination = %v, want tcp:st.packet-addr.v2fly.arpa:0", dispatcher.destination)
}
tracked, found := sorter.trackedConnections.Load(src.String())
if !found {
t.Fatal("stream packet encoding did not track the UDP source")
}
if err := tracked.(*trackedUDPConnection).packetDispatcher.Close(); err != nil {
t.Fatalf("failed to close packet dispatcher: %v", err)
}
if _, found := sorter.trackedConnections.Load(src.String()); found {
t.Fatal("explicit dispatcher close did not remove the tracked UDP source")
}
if err := tracked.(*trackedUDPConnection).packetDispatcher.Close(); err != nil {
t.Fatalf("second dispatcher close returned an error: %v", err)
}
}
func TestStreamPacketEncodingRemovesTrackedConnectionWhenStreamEnds(t *testing.T) {
dispatcher := new(recordingDispatcher)
sorter := NewTunSorter(io.Discard, dispatcher, packetaddr.PacketAddrType_Stream, context.Background(), nil)
src := net.UDPDestination(net.ParseAddress("198.18.0.2"), 49153)
dst := net.UDPDestination(net.ParseAddress("9.9.9.9"), 443)
packet, err := packetparse.TryConstructUDPPacket(src, dst, []byte("quic packet"))
if err != nil {
t.Fatalf("failed to construct UDP packet: %v", err)
}
handled, err := sorter.OnPacketReceived(packet)
if err != nil {
t.Fatalf("OnPacketReceived returned an error: %v", err)
}
if !handled {
t.Fatal("UDP packet was not handled by stream packet encoding")
}
if _, found := sorter.trackedConnections.Load(src.String()); !found {
t.Fatal("stream packet encoding did not track the UDP source")
}
if dispatcher.downlinkWriter == nil {
t.Fatal("dispatcher did not create a downlink stream")
}
if err := dispatcher.downlinkWriter.Close(); err != nil {
t.Fatalf("failed to end downlink stream: %v", err)
}
deadline := time.Now().Add(time.Second)
for time.Now().Before(deadline) {
if _, found := sorter.trackedConnections.Load(src.String()); !found {
break
}
time.Sleep(time.Millisecond)
}
if _, found := sorter.trackedConnections.Load(src.String()); found {
t.Fatal("tracked UDP source was not removed after the stream ended")
}
handled, err = sorter.OnPacketReceived(packet)
if err != nil {
t.Fatalf("OnPacketReceived after cleanup returned an error: %v", err)
}
if !handled {
t.Fatal("UDP packet was not handled after tracked connection cleanup")
}
if dispatcher.dispatchCount != 2 {
t.Fatalf("dispatch count = %d, want 2 after recreating the stream", dispatcher.dispatchCount)
}
if tracked, found := sorter.trackedConnections.Load(src.String()); found {
_ = tracked.(*trackedUDPConnection).packetDispatcher.Close()
}
}
+4 -2
View File
@@ -140,10 +140,12 @@ func (c *packetConnectionAdaptor) WriteTo(p []byte, addr gonet.Addr) (n int, err
}
func (c *packetConnectionAdaptor) Close() error {
readerErr := common.Interrupt(c.link.Reader)
writerErr := common.Interrupt(c.link.Writer)
c.readerAccess.Lock()
defer c.readerAccess.Unlock()
c.readerBuffer = buf.ReleaseMulti(c.readerBuffer)
return common.Interrupt(c.link)
c.readerAccess.Unlock()
return errors.Combine(readerErr, writerErr)
}
func (c packetConnectionAdaptor) LocalAddr() gonet.Addr {
@@ -0,0 +1,116 @@
package packetaddr
import (
"io"
"sync"
"testing"
"time"
"github.com/v2fly/v2ray-core/v5/common/buf"
"github.com/v2fly/v2ray-core/v5/common/net"
"github.com/v2fly/v2ray-core/v5/transport"
)
type blockingInterruptibleReader struct {
started chan struct{}
interrupted chan struct{}
startOnce sync.Once
interruptOnce sync.Once
}
func newBlockingInterruptibleReader() *blockingInterruptibleReader {
return &blockingInterruptibleReader{
started: make(chan struct{}),
interrupted: make(chan struct{}),
}
}
func (r *blockingInterruptibleReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
r.startOnce.Do(func() {
close(r.started)
})
<-r.interrupted
return nil, io.ErrClosedPipe
}
func (r *blockingInterruptibleReader) Interrupt() {
r.interruptOnce.Do(func() {
close(r.interrupted)
})
}
type recordingInterruptibleWriter struct {
interrupted chan struct{}
interruptOnce sync.Once
}
func newRecordingInterruptibleWriter() *recordingInterruptibleWriter {
return &recordingInterruptibleWriter{interrupted: make(chan struct{})}
}
func (w *recordingInterruptibleWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
buf.ReleaseMulti(mb)
return nil
}
func (w *recordingInterruptibleWriter) Interrupt() {
w.interruptOnce.Do(func() {
close(w.interrupted)
})
}
func TestPacketConnectionAdaptorCloseInterruptsBlockedRead(t *testing.T) {
reader := newBlockingInterruptibleReader()
writer := newRecordingInterruptibleWriter()
conn, err := ToPacketAddrConn(&transport.Link{
Reader: reader,
Writer: writer,
}, net.Destination{
Network: net.Network_TCP,
Address: net.DomainAddress(streamPacketMagicAddress),
})
if err != nil {
t.Fatalf("failed to create packet connection adaptor: %v", err)
}
readDone := make(chan error, 1)
go func() {
_, _, err := conn.ReadFrom(make([]byte, buf.Size))
readDone <- err
}()
select {
case <-reader.started:
case <-time.After(time.Second):
t.Fatal("ReadFrom did not block on the link reader")
}
closeDone := make(chan error, 1)
go func() {
closeDone <- conn.Close()
}()
select {
case err := <-closeDone:
if err != nil {
t.Fatalf("Close returned an error: %v", err)
}
case <-time.After(time.Second):
t.Fatal("Close blocked while ReadFrom was waiting on the link reader")
}
select {
case <-writer.interrupted:
default:
t.Fatal("Close did not interrupt the link writer")
}
select {
case err := <-readDone:
if err == nil {
t.Fatal("ReadFrom returned without an error after interruption")
}
case <-time.After(time.Second):
t.Fatal("ReadFrom did not return after Close interrupted the link reader")
}
}
+3
View File
@@ -100,6 +100,7 @@ require (
github.com/stretchr/objx v0.5.2 // indirect
github.com/wlynxg/anet v0.0.5 // indirect
golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 // indirect
golang.org/x/mobile v0.0.0-20260709172247-6129f5bee9d5 // indirect
golang.org/x/mod v0.38.0 // indirect
golang.org/x/text v0.41.0 // indirect
golang.org/x/time v0.14.0 // indirect
@@ -108,3 +109,5 @@ require (
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
nhooyr.io/websocket v1.8.6 // indirect
)
tool golang.org/x/mobile/cmd/gobind
+6
View File
@@ -644,6 +644,8 @@ golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHl
golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE=
golang.org/x/mobile v0.0.0-20190719004257-d2bd2a29d028/go.mod h1:E/iHnbuqvinMTCcRqshq8CkpyQDoeVncDDYHnLhea+o=
golang.org/x/mobile v0.0.0-20260709172247-6129f5bee9d5 h1:Mn1OzFmF0ZKX/ZayHz/UdnWHufPp1wlD9lZ5U8LRDFY=
golang.org/x/mobile v0.0.0-20260709172247-6129f5bee9d5/go.mod h1:YX+n47s+53POxN3dx9cIGxG3hGUm/lD64hvrRJFbcSA=
golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY=
golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
@@ -805,6 +807,10 @@ golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc
golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE=
golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk=
golang.org/x/tools/go/expect v0.1.1-deprecated h1:jpBZDwmgPhXsKZC6WhL20P4b/wmnpsEAGHaNy0n/rJM=
golang.org/x/tools/go/expect v0.1.1-deprecated/go.mod h1:eihoPOH+FgIqa3FpoTwguz/bVUSGBlGQU67vpBeOrBY=
golang.org/x/tools/go/packages/packagestest v0.1.1-deprecated h1:1h2MnaIAIXISqTFKdENegdpAgUXz6NrPEsbIeWaBRvM=
golang.org/x/tools/go/packages/packagestest v0.1.1-deprecated/go.mod h1:RVAQXBGNv1ib0J382/DPCRS/BPnsGebyM1Gj5VSDpG8=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+18 -10
View File
@@ -2,6 +2,7 @@ package udp
import (
"context"
"sync"
"github.com/v2fly/v2ray-core/v5/common/buf"
"github.com/v2fly/v2ray-core/v5/common/net"
@@ -11,19 +12,24 @@ import (
)
type PacketAddrDispatcher struct {
conn net.PacketConn
callback ResponseCallback
ctx context.Context
conn net.PacketConn
callback ResponseCallback
ctx context.Context
closeOnce sync.Once
closeErr error
}
func (p PacketAddrDispatcher) Close() error {
if p.ctx.Value(DispatcherConnectionTerminationSignalReceiverMark) != nil {
p.ctx.Value(DispatcherConnectionTerminationSignalReceiverMark).(DispatcherConnectionTerminationSignalReceiver).Close()
}
return p.conn.Close()
func (p *PacketAddrDispatcher) Close() error {
p.closeOnce.Do(func() {
if receiver := p.ctx.Value(DispatcherConnectionTerminationSignalReceiverMark); receiver != nil {
_ = receiver.(DispatcherConnectionTerminationSignalReceiver).Close()
}
p.closeErr = p.conn.Close()
})
return p.closeErr
}
func (p PacketAddrDispatcher) Dispatch(ctx context.Context, destination net.Destination, payload *buf.Buffer) {
func (p *PacketAddrDispatcher) Dispatch(ctx context.Context, destination net.Destination, payload *buf.Buffer) {
if destination.Network != net.Network_UDP {
return
}
@@ -36,11 +42,13 @@ func (p PacketAddrDispatcher) Dispatch(ctx context.Context, destination net.Dest
p.conn.WriteTo(payload.Bytes(), &net.UDPAddr{IP: destination.Address.IP(), Port: int(destination.Port.Value())})
}
func (p PacketAddrDispatcher) readWorker() {
func (p *PacketAddrDispatcher) readWorker() {
defer p.Close()
for {
readBuf := buf.New()
n, addr, err := p.conn.ReadFrom(readBuf.Extend(2048))
if err != nil {
readBuf.Release()
return
}
readBuf.Resize(0, int32(n))