Add wireguard chain proxy support (machine assisted) (#3816)

This commit is contained in:
Xiaokang Wang (Shelikhoo)
2026-08-25 18:54:54 +01:00
committed by GitHub
parent cbd4d234fe
commit 3861a91901
20 changed files with 3795 additions and 150 deletions
@@ -0,0 +1,136 @@
package specs_test
import (
"fmt"
"strings"
"testing"
"github.com/v2fly/v2ray-core/v5/app/subscription/specs"
"github.com/v2fly/v2ray-core/v5/common/net/packetaddr"
"github.com/v2fly/v2ray-core/v5/common/serial"
wireguard "github.com/v2fly/v2ray-core/v5/proxy/wireguard/outbound"
)
const wireguardConfigFullName = "v2ray.core.proxy.wireguard.outbound.Config"
func TestWireguardRestrictedSubscriptionRequiresOptIn(t *testing.T) {
tests := []struct {
name string
settings string
}{
{name: "omitted", settings: `{}`},
{name: "false", settings: `{"restricted":false}`},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
_, err := parseWireguardSubscription(t, "wireguard", test.settings)
if err == nil {
t.Fatal("restricted subscription load unexpectedly succeeded")
}
if !strings.Contains(err.Error(), "component has not opted in for load in restricted mode") {
t.Fatalf("restricted subscription load error = %q, want restricted-mode opt-in error", err)
}
})
}
}
func TestWireguardRestrictedSubscriptionPreservesSettings(t *testing.T) {
tests := []struct {
name string
encodingJSON string
wantEncoding packetaddr.PacketAddrType
timeoutJSON string
wantTimeout uint32
wantTimeoutSet bool
}{
{
name: "default encoding and timeout",
wantEncoding: packetaddr.PacketAddrType_None,
},
{
name: "packet encoding and disabled timeout",
encodingJSON: `,"outboundPacketEncoding":"Packet"`,
wantEncoding: packetaddr.PacketAddrType_Packet,
timeoutJSON: `,"outboundNoReceiveTimeoutSec":0`,
wantTimeoutSet: true,
},
{
name: "stream encoding and custom timeout",
encodingJSON: `,"outboundPacketEncoding":"Stream"`,
wantEncoding: packetaddr.PacketAddrType_Stream,
timeoutJSON: `,"outboundNoReceiveTimeoutSec":137`,
wantTimeout: 137,
wantTimeoutSet: true,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
settings := fmt.Sprintf(
`{"restricted":true,"wgDevice":{"mtu":1412},"domainStrategy":"USE_IP4"%s%s}`,
test.encodingJSON,
test.timeoutJSON,
)
server, err := parseWireguardSubscription(t, "wireguard", settings)
if err != nil {
t.Fatalf("restricted subscription load failed: %v", err)
}
message, err := serial.GetInstanceOf(server.Configuration.ProtocolSettings)
if err != nil {
t.Fatalf("failed to unpack protocol settings: %v", err)
}
config, ok := message.(*wireguard.Config)
if !ok {
t.Fatalf("protocol settings type = %T, want *outbound.Config", message)
}
if !config.GetRestricted() {
t.Fatal("restricted flag was not preserved")
}
if got := config.GetOutboundPacketEncoding(); got != test.wantEncoding {
t.Fatalf("outbound packet encoding = %v, want %v", got, test.wantEncoding)
}
if got := config.GetDomainStrategy(); got != wireguard.Config_USE_IP4 {
t.Fatalf("domain strategy = %v, want %v", got, wireguard.Config_USE_IP4)
}
if config.GetWgDevice() == nil || config.GetWgDevice().GetMtu() != 1412 {
t.Fatalf("WireGuard device settings were not preserved: %+v", config.GetWgDevice())
}
if got := config.OutboundNoReceiveTimeoutSec != nil; got != test.wantTimeoutSet {
t.Fatalf("timeout presence = %v, want %v", got, test.wantTimeoutSet)
}
if got := config.GetOutboundNoReceiveTimeoutSec(); got != test.wantTimeout {
t.Fatalf("timeout = %d, want %d", got, test.wantTimeout)
}
})
}
}
func TestWireguardRestrictedSubscriptionFullNameCannotBypassOptIn(t *testing.T) {
protocol := "#" + wireguardConfigFullName
if _, err := parseWireguardSubscription(t, protocol, `{}`); err == nil {
t.Fatal("full protobuf name bypassed restricted loading enforcement")
} else if !strings.Contains(err.Error(), "component has not opted in for load in restricted mode") {
t.Fatalf("full protobuf name error = %q, want restricted-mode opt-in error", err)
}
server, err := parseWireguardSubscription(t, protocol, `{"restricted":true}`)
if err != nil {
t.Fatalf("full protobuf name with explicit opt-in failed: %v", err)
}
if got := serial.V2Type(server.Configuration.ProtocolSettings); got != wireguardConfigFullName {
t.Fatalf("protocol settings type = %q, want %q", got, wireguardConfigFullName)
}
}
func parseWireguardSubscription(t *testing.T, protocol, settings string) (*specs.SubscriptionServerConfig, error) {
t.Helper()
raw := []byte(fmt.Sprintf(`{"protocol":%q,"settings":%s}`, protocol, settings))
parser := specs.NewOutboundParser()
outbound, err := parser.ParseOutboundConfig(raw)
if err != nil {
return nil, err
}
return parser.ToSubscriptionServerConfig(outbound)
}
+15 -4
View File
@@ -24,8 +24,11 @@ type MessageOpt struct {
// allow_restricted_mode_load allow this config to be loaded in restricted mode
// this is typically used when a an attacker can control the content
AllowRestrictedModeLoad bool `protobuf:"varint,86002,opt,name=allow_restricted_mode_load,json=allowRestrictedModeLoad,proto3" json:"allow_restricted_mode_load,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
// allow_restricted_mode_load_if_set names a singular bool field on this
// message. Restricted loading is allowed when that field is true.
AllowRestrictedModeLoadIfSet string `protobuf:"bytes,86003,opt,name=allow_restricted_mode_load_if_set,json=allowRestrictedModeLoadIfSet,proto3" json:"allow_restricted_mode_load_if_set,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *MessageOpt) Reset() {
@@ -86,6 +89,13 @@ func (x *MessageOpt) GetAllowRestrictedModeLoad() bool {
return false
}
func (x *MessageOpt) GetAllowRestrictedModeLoadIfSet() string {
if x != nil {
return x.AllowRestrictedModeLoadIfSet
}
return ""
}
type FieldOpt struct {
state protoimpl.MessageState `protogen:"open.v1"`
AnyWants []string `protobuf:"bytes,1,rep,name=any_wants,json=anyWants,proto3" json:"any_wants,omitempty"`
@@ -217,14 +227,15 @@ var File_common_protoext_extensions_proto protoreflect.FileDescriptor
const file_common_protoext_extensions_proto_rawDesc = "" +
"\n" +
" common/protoext/extensions.proto\x12\x1av2ray.core.common.protoext\x1a google/protobuf/descriptor.proto\"\xb8\x01\n" +
" common/protoext/extensions.proto\x12\x1av2ray.core.common.protoext\x1a google/protobuf/descriptor.proto\"\x83\x02\n" +
"\n" +
"MessageOpt\x12\x12\n" +
"\x04type\x18\x01 \x03(\tR\x04type\x12\x1d\n" +
"\n" +
"short_name\x18\x02 \x03(\tR\tshortName\x128\n" +
"\x17transport_original_name\x18\xf1\x9f\x05 \x01(\tR\x15transportOriginalName\x12=\n" +
"\x1aallow_restricted_mode_load\x18\xf2\x9f\x05 \x01(\bR\x17allowRestrictedModeLoad\"\xd0\x02\n" +
"\x1aallow_restricted_mode_load\x18\xf2\x9f\x05 \x01(\bR\x17allowRestrictedModeLoad\x12I\n" +
"!allow_restricted_mode_load_if_set\x18\xf3\x9f\x05 \x01(\tR\x1callowRestrictedModeLoadIfSet\"\xd0\x02\n" +
"\bFieldOpt\x12\x1b\n" +
"\tany_wants\x18\x01 \x03(\tR\banyWants\x12%\n" +
"\x0eallowed_values\x18\x02 \x03(\tR\rallowedValues\x12.\n" +
+5 -1
View File
@@ -25,6 +25,10 @@ message MessageOpt{
// allow_restricted_mode_load allow this config to be loaded in restricted mode
// this is typically used when a an attacker can control the content
bool allow_restricted_mode_load = 86002;
// allow_restricted_mode_load_if_set names a singular bool field on this
// message. Restricted loading is allowed when that field is true.
string allow_restricted_mode_load_if_set = 86003;
}
message FieldOpt{
@@ -40,4 +44,4 @@ message FieldOpt{
string convert_time_resource_loading = 6;
// convert_time_parse_ip parse a string ip address, and put its binary representation into another field
string convert_time_parse_ip = 7;
}
}
+25 -1
View File
@@ -4,6 +4,7 @@ import (
"context"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/reflect/protoreflect"
"github.com/v2fly/v2ray-core/v5/common/protoext"
)
@@ -28,7 +29,30 @@ func enforceRestriction(config proto.Message) error {
if err != nil {
return newError("unable to find message options").Base(err)
}
if !msgOpts.AllowRestrictedModeLoad {
return enforceRestrictionWithOptions(
config,
msgOpts.GetAllowRestrictedModeLoad(),
msgOpts.GetAllowRestrictedModeLoadIfSet(),
)
}
func enforceRestrictionWithOptions(config proto.Message, allowRestrictedModeLoad bool, allowRestrictedModeLoadIfSet string) error {
if allowRestrictedModeLoad {
return nil
}
if allowRestrictedModeLoadIfSet == "" {
return newError("component has not opted in for load in restricted mode")
}
message := config.ProtoReflect()
field := message.Descriptor().Fields().ByName(protoreflect.Name(allowRestrictedModeLoadIfSet))
if field == nil {
return newError("allow_restricted_mode_load_if_set references unknown field: ", allowRestrictedModeLoadIfSet)
}
if field.Cardinality() == protoreflect.Repeated || field.Kind() != protoreflect.BoolKind {
return newError("allow_restricted_mode_load_if_set must reference a singular bool field: ", allowRestrictedModeLoadIfSet)
}
if !message.Has(field) || !message.Get(field).Bool() {
return newError("component has not opted in for load in restricted mode")
}
return nil
+106
View File
@@ -0,0 +1,106 @@
package registry
import (
"strings"
"testing"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/descriptorpb"
)
func TestEnforceRestrictionWithOptions(t *testing.T) {
t.Parallel()
tests := []struct {
name string
config proto.Message
allow bool
allowIfSet string
wantErrSubstr string
}{
{
name: "unconditional option allows load",
config: &descriptorpb.FileOptions{},
allow: true,
allowIfSet: "field_that_does_not_exist",
},
{
name: "conditional bool true allows load",
config: &descriptorpb.FileOptions{JavaMultipleFiles: proto.Bool(true)},
allowIfSet: "java_multiple_files",
},
{
name: "conditional bool false denies load",
config: &descriptorpb.FileOptions{JavaMultipleFiles: proto.Bool(false)},
allowIfSet: "java_multiple_files",
wantErrSubstr: "component has not opted in for load in restricted mode",
},
{
name: "unset conditional bool denies load",
config: &descriptorpb.FileOptions{},
allowIfSet: "java_multiple_files",
wantErrSubstr: "component has not opted in for load in restricted mode",
},
{
name: "unset conditional bool with true schema default denies load",
config: &descriptorpb.FileOptions{},
allowIfSet: "cc_enable_arenas",
wantErrSubstr: "component has not opted in for load in restricted mode",
},
{
name: "explicit conditional bool matching true schema default allows load",
config: &descriptorpb.FileOptions{CcEnableArenas: proto.Bool(true)},
allowIfSet: "cc_enable_arenas",
},
{
name: "no opt-in denies load",
config: &descriptorpb.FileOptions{},
wantErrSubstr: "component has not opted in for load in restricted mode",
},
{
name: "unknown conditional field fails closed",
config: &descriptorpb.FileOptions{},
allowIfSet: "field_that_does_not_exist",
wantErrSubstr: "allow_restricted_mode_load_if_set references unknown field: field_that_does_not_exist",
},
{
name: "JSON field name is not accepted",
config: &descriptorpb.FileOptions{JavaMultipleFiles: proto.Bool(true)},
allowIfSet: "javaMultipleFiles",
wantErrSubstr: "allow_restricted_mode_load_if_set references unknown field: javaMultipleFiles",
},
{
name: "non-bool conditional field fails closed",
config: &descriptorpb.FileOptions{JavaPackage: proto.String("example")},
allowIfSet: "java_package",
wantErrSubstr: "allow_restricted_mode_load_if_set must reference a singular bool field: java_package",
},
{
name: "repeated conditional field fails closed",
config: &descriptorpb.FileDescriptorProto{},
allowIfSet: "dependency",
wantErrSubstr: "allow_restricted_mode_load_if_set must reference a singular bool field: dependency",
},
}
for _, test := range tests {
test := test
t.Run(test.name, func(t *testing.T) {
t.Parallel()
err := enforceRestrictionWithOptions(test.config, test.allow, test.allowIfSet)
if test.wantErrSubstr == "" {
if err != nil {
t.Fatalf("enforceRestrictionWithOptions() error = %v, want nil", err)
}
return
}
if err == nil {
t.Fatalf("enforceRestrictionWithOptions() error = nil, want error containing %q", test.wantErrSubstr)
}
if !strings.Contains(err.Error(), test.wantErrSubstr) {
t.Fatalf("enforceRestrictionWithOptions() error = %q, want error containing %q", err, test.wantErrSubstr)
}
})
}
}
+61 -22
View File
@@ -1,7 +1,7 @@
package outbound
import (
_ "github.com/v2fly/v2ray-core/v5/common/net/packetaddr"
packetaddr "github.com/v2fly/v2ray-core/v5/common/net/packetaddr"
gvisorstack "github.com/v2fly/v2ray-core/v5/common/packetswitch/gvisorstack"
_ "github.com/v2fly/v2ray-core/v5/common/protoext"
wgcommon "github.com/v2fly/v2ray-core/v5/proxy/wireguard/wgcommon"
@@ -72,14 +72,22 @@ func (Config_DomainStrategy) EnumDescriptor() ([]byte, []int) {
}
type Config struct {
state protoimpl.MessageState `protogen:"open.v1"`
WgDevice *wgcommon.DeviceConfig `protobuf:"bytes,1,opt,name=wg_device,json=wgDevice,proto3" json:"wg_device,omitempty"`
Stack *gvisorstack.Config `protobuf:"bytes,2,opt,name=stack,proto3" json:"stack,omitempty"`
// v2ray.core.net.packetaddr.PacketAddrType outbound_packet_encoding = 3;
state protoimpl.MessageState `protogen:"open.v1"`
WgDevice *wgcommon.DeviceConfig `protobuf:"bytes,1,opt,name=wg_device,json=wgDevice,proto3" json:"wg_device,omitempty"`
Stack *gvisorstack.Config `protobuf:"bytes,2,opt,name=stack,proto3" json:"stack,omitempty"`
OutboundPacketEncoding packetaddr.PacketAddrType `protobuf:"varint,3,opt,name=outbound_packet_encoding,json=outboundPacketEncoding,proto3,enum=v2ray.core.net.packetaddr.PacketAddrType" json:"outbound_packet_encoding,omitempty"`
// listen_on_system_network must be false when restricted is true.
ListenOnSystemNetwork bool `protobuf:"varint,4,opt,name=listen_on_system_network,json=listenOnSystemNetwork,proto3" json:"listen_on_system_network,omitempty"`
DomainStrategy Config_DomainStrategy `protobuf:"varint,5,opt,name=domain_strategy,json=domainStrategy,proto3,enum=v2ray.core.proxy.wireguard.outbound.Config_DomainStrategy" json:"domain_strategy,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
// Number of seconds without a successful receive before the logical outbound
// connection is replaced. When omitted, the default is 60 seconds. An
// explicit value of 0 disables no-receive-based replacement.
OutboundNoReceiveTimeoutSec *uint32 `protobuf:"varint,6,opt,name=outbound_no_receive_timeout_sec,json=outboundNoReceiveTimeoutSec,proto3,oneof" json:"outbound_no_receive_timeout_sec,omitempty"`
// restricted opts this config into restricted-mode loading. Restricted
// WireGuard outbounds must use a logical-network transport.
Restricted bool `protobuf:"varint,7,opt,name=restricted,proto3" json:"restricted,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Config) Reset() {
@@ -126,6 +134,13 @@ func (x *Config) GetStack() *gvisorstack.Config {
return nil
}
func (x *Config) GetOutboundPacketEncoding() packetaddr.PacketAddrType {
if x != nil {
return x.OutboundPacketEncoding
}
return packetaddr.PacketAddrType(0)
}
func (x *Config) GetListenOnSystemNetwork() bool {
if x != nil {
return x.ListenOnSystemNetwork
@@ -140,23 +155,44 @@ func (x *Config) GetDomainStrategy() Config_DomainStrategy {
return Config_AS_IS
}
func (x *Config) GetOutboundNoReceiveTimeoutSec() uint32 {
if x != nil && x.OutboundNoReceiveTimeoutSec != nil {
return *x.OutboundNoReceiveTimeoutSec
}
return 0
}
func (x *Config) GetRestricted() bool {
if x != nil {
return x.Restricted
}
return false
}
var File_proxy_wireguard_outbound_config_proto protoreflect.FileDescriptor
const file_proxy_wireguard_outbound_config_proto_rawDesc = "" +
"\n" +
"%proxy/wireguard/outbound/config.proto\x12#v2ray.core.proxy.wireguard.outbound\x1a%proxy/wireguard/wgcommon/config.proto\x1a,common/packetswitch/gvisorstack/config.proto\x1a\"common/net/packetaddr/config.proto\x1a common/protoext/extensions.proto\"\x9e\x03\n" +
"%proxy/wireguard/outbound/config.proto\x12#v2ray.core.proxy.wireguard.outbound\x1a%proxy/wireguard/wgcommon/config.proto\x1a,common/packetswitch/gvisorstack/config.proto\x1a\"common/net/packetaddr/config.proto\x1a common/protoext/extensions.proto\"\xa0\x05\n" +
"\x06Config\x12N\n" +
"\twg_device\x18\x01 \x01(\v21.v2ray.core.proxy.wireguard.wgcommon.DeviceConfigR\bwgDevice\x12H\n" +
"\x05stack\x18\x02 \x01(\v22.v2ray.core.common.packetswitch.gvisorstack.ConfigR\x05stack\x127\n" +
"\x05stack\x18\x02 \x01(\v22.v2ray.core.common.packetswitch.gvisorstack.ConfigR\x05stack\x12c\n" +
"\x18outbound_packet_encoding\x18\x03 \x01(\x0e2).v2ray.core.net.packetaddr.PacketAddrTypeR\x16outboundPacketEncoding\x127\n" +
"\x18listen_on_system_network\x18\x04 \x01(\bR\x15listenOnSystemNetwork\x12c\n" +
"\x0fdomain_strategy\x18\x05 \x01(\x0e2:.v2ray.core.proxy.wireguard.outbound.Config.DomainStrategyR\x0edomainStrategy\"A\n" +
"\x0fdomain_strategy\x18\x05 \x01(\x0e2:.v2ray.core.proxy.wireguard.outbound.Config.DomainStrategyR\x0edomainStrategy\x12I\n" +
"\x1foutbound_no_receive_timeout_sec\x18\x06 \x01(\rH\x00R\x1boutboundNoReceiveTimeoutSec\x88\x01\x01\x12\x1e\n" +
"\n" +
"restricted\x18\a \x01(\bR\n" +
"restricted\"A\n" +
"\x0eDomainStrategy\x12\t\n" +
"\x05AS_IS\x10\x00\x12\n" +
"\n" +
"\x06USE_IP\x10\x01\x12\v\n" +
"\aUSE_IP4\x10\x02\x12\v\n" +
"\aUSE_IP6\x10\x03:\x19\x82\xb5\x18\x15\n" +
"\boutbound\x12\twireguardB\x8a\x01\n" +
"\aUSE_IP6\x10\x03:'\x82\xb5\x18#\n" +
"\boutbound\x12\twireguard\x9a\xff)\n" +
"restrictedB\"\n" +
" _outbound_no_receive_timeout_secB\x8a\x01\n" +
"'com.v2ray.core.proxy.wireguard.outboundP\x01Z7github.com/v2fly/v2ray-core/v5/proxy/wireguard/outbound\xaa\x02#V2Ray.Core.Proxy.Wireguard.Outboundb\x06proto3"
var (
@@ -174,20 +210,22 @@ func file_proxy_wireguard_outbound_config_proto_rawDescGZIP() []byte {
var file_proxy_wireguard_outbound_config_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
var file_proxy_wireguard_outbound_config_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
var file_proxy_wireguard_outbound_config_proto_goTypes = []any{
(Config_DomainStrategy)(0), // 0: v2ray.core.proxy.wireguard.outbound.Config.DomainStrategy
(*Config)(nil), // 1: v2ray.core.proxy.wireguard.outbound.Config
(*wgcommon.DeviceConfig)(nil), // 2: v2ray.core.proxy.wireguard.wgcommon.DeviceConfig
(*gvisorstack.Config)(nil), // 3: v2ray.core.common.packetswitch.gvisorstack.Config
(Config_DomainStrategy)(0), // 0: v2ray.core.proxy.wireguard.outbound.Config.DomainStrategy
(*Config)(nil), // 1: v2ray.core.proxy.wireguard.outbound.Config
(*wgcommon.DeviceConfig)(nil), // 2: v2ray.core.proxy.wireguard.wgcommon.DeviceConfig
(*gvisorstack.Config)(nil), // 3: v2ray.core.common.packetswitch.gvisorstack.Config
(packetaddr.PacketAddrType)(0), // 4: v2ray.core.net.packetaddr.PacketAddrType
}
var file_proxy_wireguard_outbound_config_proto_depIdxs = []int32{
2, // 0: v2ray.core.proxy.wireguard.outbound.Config.wg_device:type_name -> v2ray.core.proxy.wireguard.wgcommon.DeviceConfig
3, // 1: v2ray.core.proxy.wireguard.outbound.Config.stack:type_name -> v2ray.core.common.packetswitch.gvisorstack.Config
0, // 2: v2ray.core.proxy.wireguard.outbound.Config.domain_strategy:type_name -> v2ray.core.proxy.wireguard.outbound.Config.DomainStrategy
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
4, // 2: v2ray.core.proxy.wireguard.outbound.Config.outbound_packet_encoding:type_name -> v2ray.core.net.packetaddr.PacketAddrType
0, // 3: v2ray.core.proxy.wireguard.outbound.Config.domain_strategy:type_name -> v2ray.core.proxy.wireguard.outbound.Config.DomainStrategy
4, // [4:4] is the sub-list for method output_type
4, // [4:4] is the sub-list for method input_type
4, // [4:4] is the sub-list for extension type_name
4, // [4:4] is the sub-list for extension extendee
0, // [0:4] is the sub-list for field type_name
}
func init() { file_proxy_wireguard_outbound_config_proto_init() }
@@ -195,6 +233,7 @@ func file_proxy_wireguard_outbound_config_proto_init() {
if File_proxy_wireguard_outbound_config_proto != nil {
return
}
file_proxy_wireguard_outbound_config_proto_msgTypes[0].OneofWrappers = []any{}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
+13 -2
View File
@@ -14,11 +14,13 @@ import "common/protoext/extensions.proto";
message Config{
option (v2ray.core.common.protoext.message_opt).type = "outbound";
option (v2ray.core.common.protoext.message_opt).short_name = "wireguard";
option (v2ray.core.common.protoext.message_opt).allow_restricted_mode_load_if_set = "restricted";
v2ray.core.proxy.wireguard.wgcommon.DeviceConfig wg_device = 1;
v2ray.core.common.packetswitch.gvisorstack.Config stack = 2;
//v2ray.core.net.packetaddr.PacketAddrType outbound_packet_encoding = 3;
v2ray.core.net.packetaddr.PacketAddrType outbound_packet_encoding = 3;
// listen_on_system_network must be false when restricted is true.
bool listen_on_system_network = 4;
enum DomainStrategy {
@@ -28,4 +30,13 @@ message Config{
USE_IP6 = 3;
}
DomainStrategy domain_strategy = 5;
}
// Number of seconds without a successful receive before the logical outbound
// connection is replaced. When omitted, the default is 60 seconds. An
// explicit value of 0 disables no-receive-based replacement.
optional uint32 outbound_no_receive_timeout_sec = 6;
// restricted opts this config into restricted-mode loading. Restricted
// WireGuard outbounds must use a logical-network transport.
bool restricted = 7;
}
+108
View File
@@ -0,0 +1,108 @@
package outbound
import (
"testing"
"google.golang.org/protobuf/encoding/protojson"
"github.com/v2fly/v2ray-core/v5/common/net/packetaddr"
"github.com/v2fly/v2ray-core/v5/common/protoext"
)
func TestConfigProtoJSONOutboundPacketEncoding(t *testing.T) {
tests := []struct {
name string
json string
want packetaddr.PacketAddrType
}{
{name: "omitted", json: `{}`, want: packetaddr.PacketAddrType_None},
{name: "none", json: `{"outboundPacketEncoding":"None"}`, want: packetaddr.PacketAddrType_None},
{name: "packet", json: `{"outboundPacketEncoding":"Packet"}`, want: packetaddr.PacketAddrType_Packet},
{name: "stream", json: `{"outboundPacketEncoding":"Stream"}`, want: packetaddr.PacketAddrType_Stream},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
config := new(Config)
if err := protojson.Unmarshal([]byte(test.json), config); err != nil {
t.Fatalf("failed to unmarshal config: %v", err)
}
if got := config.GetOutboundPacketEncoding(); got != test.want {
t.Fatalf("GetOutboundPacketEncoding() = %v, want %v", got, test.want)
}
})
}
}
func TestConfigProtoJSONOutboundNoReceiveTimeoutPresence(t *testing.T) {
tests := []struct {
name string
json string
wantPresent bool
want uint32
}{
{name: "omitted", json: `{}`, wantPresent: false},
{name: "disabled", json: `{"outboundNoReceiveTimeoutSec":0}`, wantPresent: true, want: 0},
{name: "custom", json: `{"outboundNoReceiveTimeoutSec":120}`, wantPresent: true, want: 120},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
config := new(Config)
if err := protojson.Unmarshal([]byte(test.json), config); err != nil {
t.Fatalf("failed to unmarshal config: %v", err)
}
if got := config.OutboundNoReceiveTimeoutSec != nil; got != test.wantPresent {
t.Fatalf("timeout presence = %v, want %v", got, test.wantPresent)
}
if got := config.GetOutboundNoReceiveTimeoutSec(); got != test.want {
t.Fatalf("GetOutboundNoReceiveTimeoutSec() = %d, want %d", got, test.want)
}
})
}
}
func TestConfigProtoJSONRestricted(t *testing.T) {
tests := []struct {
name string
json string
want bool
}{
{name: "omitted", json: `{}`},
{name: "false", json: `{"restricted":false}`},
{name: "true", json: `{"restricted":true}`, want: true},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
config := new(Config)
if err := protojson.Unmarshal([]byte(test.json), config); err != nil {
t.Fatalf("failed to unmarshal config: %v", err)
}
if got := config.GetRestricted(); got != test.want {
t.Fatalf("GetRestricted() = %v, want %v", got, test.want)
}
})
}
}
func TestConfigRestrictedLoadOption(t *testing.T) {
options, err := protoext.GetMessageOptions((&Config{}).ProtoReflect().Descriptor())
if err != nil {
t.Fatalf("failed to get message options: %v", err)
}
if got, want := options.GetAllowRestrictedModeLoadIfSet(), "restricted"; got != want {
t.Fatalf("allow_restricted_mode_load_if_set = %q, want %q", got, want)
}
}
func TestConfigProtoJSONRestrictedSystemNetworkRejected(t *testing.T) {
config := new(Config)
if err := protojson.Unmarshal([]byte(`{"restricted":true,"listenOnSystemNetwork":true}`), config); err != nil {
t.Fatalf("failed to unmarshal config: %v", err)
}
if err := validateWireguardConfig(config); err == nil {
t.Fatal("validateWireguardConfig() accepted restricted system-network config decoded from JSON")
}
}
@@ -0,0 +1,199 @@
package outbound
import (
"context"
"sync"
"github.com/v2fly/v2ray-core/v5/common/buf"
cnet "github.com/v2fly/v2ray-core/v5/common/net"
"github.com/v2fly/v2ray-core/v5/features/routing"
"github.com/v2fly/v2ray-core/v5/transport"
"github.com/v2fly/v2ray-core/v5/transport/internet"
)
type wireguardUnderlayContextKey struct{}
type wireguardUnderlayMarker struct {
outbound *WireguardOutbound
parent *wireguardUnderlayMarker
}
func markWireguardUnderlay(ctx context.Context, outbound *WireguardOutbound) context.Context {
if ctx == nil {
ctx = context.Background()
}
parent, _ := ctx.Value(wireguardUnderlayContextKey{}).(*wireguardUnderlayMarker)
return context.WithValue(ctx, wireguardUnderlayContextKey{}, &wireguardUnderlayMarker{
outbound: outbound,
parent: parent,
})
}
func isWireguardUnderlayFor(ctx context.Context, outbound *WireguardOutbound) bool {
if ctx == nil || outbound == nil {
return false
}
marker, _ := ctx.Value(wireguardUnderlayContextKey{}).(*wireguardUnderlayMarker)
for marker != nil {
if marker.outbound == outbound {
return true
}
marker = marker.parent
}
return false
}
// preserveWireguardUnderlayMarkers copies only the immutable recursion marker
// chain to a long-lived context. Request cancellation and other request-scoped
// values must not become parents of the shared WireGuard session.
func preserveWireguardUnderlayMarkers(base, source context.Context) context.Context {
if base == nil {
base = context.Background()
}
if source == nil {
return base
}
marker, _ := source.Value(wireguardUnderlayContextKey{}).(*wireguardUnderlayMarker)
if marker == nil {
return base
}
return context.WithValue(base, wireguardUnderlayContextKey{}, marker)
}
// dialerDispatcher adapts the outbound handler's internet.Dialer to the
// routing.Dispatcher expected by the logical packet-connection implementations.
// The dialer remains responsible for applying sender proxy settings.
type dialerDispatcher struct {
outbound *WireguardOutbound
dialer internet.Dialer
}
func newDialerDispatcher(outbound *WireguardOutbound, dialer internet.Dialer) (routing.Dispatcher, error) {
if dialer == nil {
return nil, newError("internet dialer is not configured for wireguard outbound")
}
return &dialerDispatcher{
outbound: outbound,
dialer: dialer,
}, nil
}
func (d *dialerDispatcher) Dispatch(ctx context.Context, destination cnet.Destination) (*transport.Link, error) {
if d == nil || d.dialer == nil {
return nil, newError("internet dialer is not configured for wireguard outbound")
}
dialCtx, cancel := context.WithCancel(markWireguardUnderlay(ctx, d.outbound))
conn, err := d.dialer.Dial(dialCtx, destination)
if err != nil {
cancel()
return nil, newError("failed to dial wireguard underlay to ", destination).Base(err)
}
if conn == nil {
cancel()
return nil, newError("internet dialer returned a nil wireguard underlay connection")
}
return newDialerLink(conn, destination.Network, cancel), nil
}
func (*dialerDispatcher) Start() error { return nil }
func (*dialerDispatcher) Close() error { return nil }
func (*dialerDispatcher) Type() interface{} { return routing.DispatcherType() }
type dialedConnectionOwner struct {
connection internet.Connection
cancel context.CancelFunc
closeOnce sync.Once
closeErr error
}
func (o *dialedConnectionOwner) Close() error {
if o == nil {
return nil
}
o.closeOnce.Do(func() {
if o.cancel != nil {
o.cancel()
}
if o.connection != nil {
o.closeErr = o.connection.Close()
}
})
return o.closeErr
}
type ownedLinkReader struct {
buf.Reader
owner *dialedConnectionOwner
}
func (r *ownedLinkReader) Interrupt() {
_ = r.owner.Close()
}
func (r *ownedLinkReader) Close() error {
return r.owner.Close()
}
type ownedLinkWriter struct {
buf.Writer
owner *dialedConnectionOwner
}
func (w *ownedLinkWriter) Interrupt() {
_ = w.owner.Close()
}
func (w *ownedLinkWriter) Close() error {
return w.owner.Close()
}
// statCounterPacketReader preserves packet boundaries by reading from the
// wrapped packet connection's MultiBuffer/packet reader directly. Calling the
// generic StatCouterConnection.Read method could merge adjacent UDP buffers.
type statCounterPacketReader struct {
reader buf.Reader
connection *internet.StatCouterConnection
}
func (r *statCounterPacketReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
mb, err := r.reader.ReadMultiBuffer()
if r.connection.ReadCounter != nil {
r.connection.ReadCounter.Add(int64(mb.Len()))
}
return mb, err
}
func newPacketReader(conn internet.Connection) buf.Reader {
if statConn, ok := conn.(*internet.StatCouterConnection); ok && statConn.Connection != nil {
return &statCounterPacketReader{
reader: newPacketReader(statConn.Connection),
connection: statConn,
}
}
return buf.NewPacketReader(conn)
}
func newDialerLink(conn internet.Connection, network cnet.Network, cancel context.CancelFunc) *transport.Link {
owner := &dialedConnectionOwner{connection: conn, cancel: cancel}
var reader buf.Reader
if network == cnet.Network_UDP {
reader = newPacketReader(conn)
} else {
reader = buf.NewReader(conn)
}
return &transport.Link{
Reader: &ownedLinkReader{
Reader: reader,
owner: owner,
},
Writer: &ownedLinkWriter{
Writer: buf.NewWriter(conn),
owner: owner,
},
}
}
@@ -0,0 +1,386 @@
package outbound
import (
"context"
"io"
gonet "net"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/v2fly/v2ray-core/v5/common"
"github.com/v2fly/v2ray-core/v5/common/buf"
cnet "github.com/v2fly/v2ray-core/v5/common/net"
"github.com/v2fly/v2ray-core/v5/common/net/packetaddr"
"github.com/v2fly/v2ray-core/v5/transport/internet"
)
type dialerTestRecord struct {
ctx context.Context
destination cnet.Destination
connection *dialerTestConnection
}
type recordingDialer struct {
mu sync.Mutex
records []dialerTestRecord
factory func(context.Context, cnet.Destination) (internet.Connection, *dialerTestConnection, error)
}
func (d *recordingDialer) Dial(ctx context.Context, destination cnet.Destination) (internet.Connection, error) {
if d.factory != nil {
connection, tracked, err := d.factory(ctx, destination)
if err != nil {
return nil, err
}
d.mu.Lock()
d.records = append(d.records, dialerTestRecord{ctx: ctx, destination: destination, connection: tracked})
d.mu.Unlock()
return connection, nil
}
connection := newDialerTestConnection()
d.mu.Lock()
d.records = append(d.records, dialerTestRecord{ctx: ctx, destination: destination, connection: connection})
d.mu.Unlock()
return connection, nil
}
func (*recordingDialer) Address() cnet.Address { return nil }
func (d *recordingDialer) snapshot() []dialerTestRecord {
d.mu.Lock()
defer d.mu.Unlock()
return append([]dialerTestRecord(nil), d.records...)
}
type dialerTestConnection struct {
reads chan []byte
writes chan []byte
closed chan struct{}
closeOnce sync.Once
closeCalls atomic.Int32
}
func newDialerTestConnection() *dialerTestConnection {
return &dialerTestConnection{
reads: make(chan []byte, 8),
writes: make(chan []byte, 8),
closed: make(chan struct{}),
}
}
func (c *dialerTestConnection) Read(p []byte) (int, error) {
select {
case payload := <-c.reads:
return copy(p, payload), nil
case <-c.closed:
return 0, gonet.ErrClosed
}
}
func (c *dialerTestConnection) Write(p []byte) (int, error) {
payload := append([]byte(nil), p...)
select {
case c.writes <- payload:
return len(p), nil
case <-c.closed:
return 0, gonet.ErrClosed
}
}
func (c *dialerTestConnection) Close() error {
c.closeCalls.Add(1)
c.closeOnce.Do(func() { close(c.closed) })
return nil
}
func (*dialerTestConnection) LocalAddr() gonet.Addr {
return &gonet.UDPAddr{IP: gonet.IPv4zero}
}
func (*dialerTestConnection) RemoteAddr() gonet.Addr {
return &gonet.UDPAddr{IP: gonet.IPv4zero}
}
func (*dialerTestConnection) SetDeadline(time.Time) error { return nil }
func (*dialerTestConnection) SetReadDeadline(time.Time) error { return nil }
func (*dialerTestConnection) SetWriteDeadline(time.Time) error { return nil }
func dialerTestAwaitWrite(t *testing.T, connection *dialerTestConnection, want string) {
t.Helper()
select {
case got := <-connection.writes:
if string(got) != want {
t.Fatalf("write = %q, want %q", got, want)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for dialed connection write")
}
}
func TestDialerDispatcherMarksContextAndOwnsConnection(t *testing.T) {
type contextKey struct{}
wireguard := new(WireguardOutbound)
otherWireguard := new(WireguardOutbound)
dialer := new(recordingDialer)
dispatcher, err := newDialerDispatcher(wireguard, dialer)
if err != nil {
t.Fatal(err)
}
destination := cnet.UDPDestination(cnet.ParseAddress("192.0.2.1"), 51820)
baseCtx := context.WithValue(context.Background(), contextKey{}, "preserved")
link, err := dispatcher.Dispatch(baseCtx, destination)
if err != nil {
t.Fatal(err)
}
records := dialer.snapshot()
if len(records) != 1 {
t.Fatalf("dial count = %d, want 1", len(records))
}
if records[0].destination != destination {
t.Fatalf("destination = %v, want %v", records[0].destination, destination)
}
if got := records[0].ctx.Value(contextKey{}); got != "preserved" {
t.Fatalf("context value = %v, want preserved", got)
}
if !isWireguardUnderlayFor(records[0].ctx, wireguard) {
t.Fatal("dial context is not marked for its WireGuard outbound")
}
if isWireguardUnderlayFor(records[0].ctx, otherWireguard) {
t.Fatal("dial context was marked for an unrelated WireGuard outbound")
}
otherSessionCtx := preserveWireguardUnderlayMarkers(context.Background(), records[0].ctx)
otherDialCtx := markWireguardUnderlay(otherSessionCtx, otherWireguard)
if !isWireguardUnderlayFor(otherDialCtx, wireguard) || !isWireguardUnderlayFor(otherDialCtx, otherWireguard) {
t.Fatal("WireGuard underlay marker chain was not preserved across shared-session context creation")
}
if err := wireguard.Process(records[0].ctx, nil, nil); err == nil || !strings.Contains(err.Error(), "cannot use itself") {
t.Fatalf("recursive Process error = %v, want self-underlay rejection", err)
}
var closeGroup sync.WaitGroup
for range 8 {
closeGroup.Add(2)
go func() {
defer closeGroup.Done()
_ = common.Interrupt(link.Reader)
}()
go func() {
defer closeGroup.Done()
_ = common.Interrupt(link.Writer)
}()
}
closeGroup.Wait()
if got := records[0].connection.closeCalls.Load(); got != 1 {
t.Fatalf("underlying Close calls = %d, want 1", got)
}
select {
case <-records[0].ctx.Done():
default:
t.Fatal("closing the link did not cancel the dial context")
}
}
func TestDialerDispatcherPlainUDPReusesConnectionsByDestination(t *testing.T) {
wireguard := new(WireguardOutbound)
dialer := new(recordingDialer)
dispatcher, err := newDialerDispatcher(wireguard, dialer)
if err != nil {
t.Fatal(err)
}
packetConn, err := createLogicalPacketConn(context.Background(), dispatcher, packetaddr.PacketAddrType_None)
if err != nil {
t.Fatal(err)
}
defer func() { _ = packetConn.Close() }()
addressA := &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.1"), Port: 51820}
addressB := &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.2"), Port: 51821}
for _, write := range []struct {
payload string
address *gonet.UDPAddr
}{
{payload: "a1", address: addressA},
{payload: "a2", address: addressA},
{payload: "b1", address: addressB},
} {
if n, err := packetConn.WriteTo([]byte(write.payload), write.address); err != nil || n != len(write.payload) {
t.Fatalf("WriteTo(%q) = (%d, %v)", write.payload, n, err)
}
}
records := dialer.snapshot()
if len(records) != 2 {
t.Fatalf("dial count = %d, want one per destination (2)", len(records))
}
if got, want := records[0].destination, cnet.DestinationFromAddr(addressA); got != want {
t.Fatalf("first destination = %v, want %v", got, want)
}
if got, want := records[1].destination, cnet.DestinationFromAddr(addressB); got != want {
t.Fatalf("second destination = %v, want %v", got, want)
}
dialerTestAwaitWrite(t, records[0].connection, "a1")
dialerTestAwaitWrite(t, records[0].connection, "a2")
dialerTestAwaitWrite(t, records[1].connection, "b1")
records[1].connection.reads <- []byte("reply from b")
readBuffer := make([]byte, 64)
n, source, err := packetConn.ReadFrom(readBuffer)
if err != nil {
t.Fatal(err)
}
if got := string(readBuffer[:n]); got != "reply from b" {
t.Fatalf("reply = %q, want %q", got, "reply from b")
}
if got := source.String(); got != addressB.String() {
t.Fatalf("reply source = %q, want %q", got, addressB)
}
if err := packetConn.Close(); err != nil {
t.Fatal(err)
}
if err := packetConn.Close(); err != nil {
t.Fatal(err)
}
for i, record := range records {
if got := record.connection.closeCalls.Load(); got != 1 {
t.Fatalf("connection %d Close calls = %d, want 1", i, got)
}
}
}
func TestDialerDispatcherPacketAddrMagicDestinations(t *testing.T) {
tests := []struct {
name string
encoding packetaddr.PacketAddrType
wantNetwork cnet.Network
wantAddress string
}{
{
name: "packet",
encoding: packetaddr.PacketAddrType_Packet,
wantNetwork: cnet.Network_UDP,
wantAddress: "sp.packet-addr.v2fly.arpa",
},
{
name: "stream",
encoding: packetaddr.PacketAddrType_Stream,
wantNetwork: cnet.Network_TCP,
wantAddress: "st.packet-addr.v2fly.arpa",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
dialer := new(recordingDialer)
dispatcher, err := newDialerDispatcher(new(WireguardOutbound), dialer)
if err != nil {
t.Fatal(err)
}
packetConn, err := createLogicalPacketConn(context.Background(), dispatcher, test.encoding)
if err != nil {
t.Fatal(err)
}
records := dialer.snapshot()
if len(records) != 1 {
t.Fatalf("dial count = %d, want 1", len(records))
}
destination := records[0].destination
if destination.Network != test.wantNetwork {
t.Fatalf("network = %v, want %v", destination.Network, test.wantNetwork)
}
if got := destination.Address.String(); got != test.wantAddress {
t.Fatalf("address = %q, want %q", got, test.wantAddress)
}
if destination.Port != 0 {
t.Fatalf("port = %d, want 0", destination.Port)
}
if err := packetConn.Close(); err != nil {
t.Fatal(err)
}
if got := records[0].connection.closeCalls.Load(); got != 1 {
t.Fatalf("underlying Close calls = %d, want 1", got)
}
})
}
}
type multiBufferDialerTestConnection struct {
*dialerTestConnection
readMulti chan buf.MultiBuffer
}
func newMultiBufferDialerTestConnection() *multiBufferDialerTestConnection {
return &multiBufferDialerTestConnection{
dialerTestConnection: newDialerTestConnection(),
readMulti: make(chan buf.MultiBuffer, 1),
}
}
func (c *multiBufferDialerTestConnection) ReadMultiBuffer() (buf.MultiBuffer, error) {
select {
case mb := <-c.readMulti:
return mb, nil
case <-c.closed:
return nil, io.ErrClosedPipe
}
}
type dialerTestCounter struct {
value atomic.Int64
}
func (c *dialerTestCounter) Value() int64 { return c.value.Load() }
func (c *dialerTestCounter) Set(value int64) int64 {
return c.value.Swap(value)
}
func (c *dialerTestCounter) Add(delta int64) int64 {
return c.value.Add(delta) - delta
}
func TestDialerDispatcherPreservesUDPBoundariesThroughStatConnection(t *testing.T) {
underlying := newMultiBufferDialerTestConnection()
readCounter := new(dialerTestCounter)
statConnection := &internet.StatCouterConnection{
Connection: underlying,
ReadCounter: readCounter,
}
dialer := &recordingDialer{
factory: func(context.Context, cnet.Destination) (internet.Connection, *dialerTestConnection, error) {
return statConnection, underlying.dialerTestConnection, nil
},
}
dispatcher, err := newDialerDispatcher(new(WireguardOutbound), dialer)
if err != nil {
t.Fatal(err)
}
link, err := dispatcher.Dispatch(context.Background(), cnet.UDPDestination(cnet.ParseAddress("192.0.2.1"), 51820))
if err != nil {
t.Fatal(err)
}
defer func() { _ = common.Interrupt(link.Reader) }()
underlying.readMulti <- buf.MultiBuffer{buf.FromBytes([]byte("first")), buf.FromBytes([]byte("second"))}
mb, err := link.Reader.ReadMultiBuffer()
if err != nil {
t.Fatal(err)
}
defer buf.ReleaseMulti(mb)
if len(mb) != 2 {
t.Fatalf("packet buffer count = %d, want 2", len(mb))
}
if got := string(mb[0].Bytes()); got != "first" {
t.Fatalf("first packet = %q, want first", got)
}
if got := string(mb[1].Bytes()); got != "second" {
t.Fatalf("second packet = %q, want second", got)
}
if got := readCounter.Value(); got != int64(len("first")+len("second")) {
t.Fatalf("read counter = %d, want %d", got, len("first")+len("second"))
}
}
+275 -84
View File
@@ -20,6 +20,7 @@ import (
"github.com/v2fly/v2ray-core/v5/common/signal"
"github.com/v2fly/v2ray-core/v5/common/task"
"github.com/v2fly/v2ray-core/v5/features/dns"
"github.com/v2fly/v2ray-core/v5/features/routing"
"github.com/v2fly/v2ray-core/v5/proxy/wireguard/wgcommon"
"github.com/v2fly/v2ray-core/v5/transport"
"github.com/v2fly/v2ray-core/v5/transport/internet"
@@ -29,11 +30,14 @@ import (
//go:generate go run github.com/v2fly/v2ray-core/v5/common/errors/errorgen
func NewWireguardOutbound(ctx context.Context, config *Config) (*WireguardOutbound, error) {
if err := validateWireguardConfig(config); err != nil {
return nil, err
}
w := &WireguardOutbound{
ctx: ctx,
config: config,
}
// Acquire dns client feature if available
// Acquire the DNS client feature.
if err := core.RequireFeatures(ctx, func(d dns.Client) error {
w.dnsClient = d
return nil
@@ -52,6 +56,16 @@ func NewWireguardOutbound(ctx context.Context, config *Config) (*WireguardOutbou
return w, nil
}
func validateWireguardConfig(config *Config) error {
if config == nil {
return newError("nil config")
}
if config.GetRestricted() && config.GetListenOnSystemNetwork() {
return newError("restricted WireGuard outbound cannot listen on the system network")
}
return nil
}
type WireguardOutbound struct {
ctx context.Context
config *Config
@@ -60,8 +74,9 @@ type WireguardOutbound struct {
}
type WireguardOutboundSession struct {
ctx context.Context
config *Config
ctx context.Context
config *Config
closeOnce sync.Once
stack *gvisorstack.WrappedStack
wireguardDevice *wgcommon.WrappedWireguardDevice
@@ -111,72 +126,260 @@ func (s *WireguardOutboundSession) initFromConfig(ctx context.Context, config *C
const ConnectionState = "ConnectionState"
type ClientConnState struct {
session *WireguardOutboundSession
initOnce *sync.Once
mu sync.Mutex
mu sync.Mutex
ready *sync.Cond
creating bool
active int
closed bool
session *WireguardOutboundSession
sessionCancel context.CancelFunc
}
func (c *ClientConnState) GetOrCreateSession(create func() (*WireguardOutboundSession, error)) (*WireguardOutboundSession, error) {
var errOuter error
c.initOnce.Do(func() {
sess, err := create()
if err != nil {
errOuter = err
return
}
c.mu.Lock()
c.session = sess
c.mu.Unlock()
return c.GetOrCreateSessionWithContext(context.Background(), func(context.Context) (*WireguardOutboundSession, error) {
return create()
})
if errOuter != nil {
return nil, newError("failed to initialize UDP State").Base(errOuter)
}
func (c *ClientConnState) GetOrCreateSessionWithContext(
ctx context.Context,
create func(context.Context) (*WireguardOutboundSession, error),
) (*WireguardOutboundSession, error) {
sess, _, err := c.getOrCreateSessionWithContext(ctx, create, false)
return sess, err
}
func (c *ClientConnState) AcquireOrCreateSessionWithContext(
ctx context.Context,
create func(context.Context) (*WireguardOutboundSession, error),
) (*WireguardOutboundSession, func(), error) {
return c.getOrCreateSessionWithContext(ctx, create, true)
}
func (c *ClientConnState) getOrCreateSessionWithContext(
ctx context.Context,
create func(context.Context) (*WireguardOutboundSession, error),
acquire bool,
) (*WireguardOutboundSession, func(), error) {
if ctx == nil {
ctx = context.Background()
}
c.mu.Lock()
for c.creating {
c.ready.Wait()
}
if c.closed {
c.mu.Unlock()
return nil, nil, newError("UDP connection state is closed")
}
if c.session != nil {
sess := c.session
release := c.acquireLocked(acquire)
c.mu.Unlock()
return sess, release, nil
}
sessionCtx, sessionCancel := context.WithCancel(ctx)
c.creating = true
c.sessionCancel = sessionCancel
c.mu.Unlock()
sess, err := create(sessionCtx)
if err != nil {
sessionCancel()
c.mu.Lock()
c.sessionCancel = nil
c.creating = false
c.ready.Broadcast()
c.mu.Unlock()
return nil, nil, newError("failed to initialize UDP State").Base(err)
}
if sess == nil {
sessionCancel()
c.mu.Lock()
c.sessionCancel = nil
c.creating = false
c.ready.Broadcast()
c.mu.Unlock()
return nil, nil, newError("failed to initialize UDP State: nil session")
}
c.mu.Lock()
if c.closed {
c.mu.Unlock()
sessionCancel()
_ = sess.Close()
c.mu.Lock()
c.sessionCancel = nil
c.creating = false
c.ready.Broadcast()
c.mu.Unlock()
return nil, nil, newError("UDP connection state is closed")
}
c.session = sess
release := c.acquireLocked(acquire)
c.creating = false
c.ready.Broadcast()
c.mu.Unlock()
return sess, release, nil
}
func (c *ClientConnState) acquireLocked(acquire bool) func() {
if !acquire {
return nil
}
c.active++
var once sync.Once
return func() {
once.Do(func() {
c.mu.Lock()
c.active--
c.ready.Broadcast()
c.mu.Unlock()
})
}
return c.session, nil
}
func (c *ClientConnState) IsTransientStorageLifecycleReceiver() {}
func (c *ClientConnState) Close() error {
c.mu.Lock()
defer c.mu.Unlock()
if c.session == nil {
return nil
c.closed = true
cancel := c.sessionCancel
c.mu.Unlock()
if cancel != nil {
cancel()
}
c.mu.Lock()
for c.creating || c.active != 0 {
c.ready.Wait()
}
sess := c.session
c.session = nil
// close interconnect devices first to stop any further packet injections
if sess.interconnect != nil {
_ = sess.interconnect.GetLSideDevice().Close()
_ = sess.interconnect.GetRSideDevice().Close()
sess.interconnect = nil
c.sessionCancel = nil
c.mu.Unlock()
if sess == nil {
return nil
}
return sess.Close()
}
// close system packet conn
if sess.systemPacketConn != nil {
_ = sess.systemPacketConn.Close()
sess.systemPacketConn = nil
func (s *WireguardOutboundSession) Close() error {
if s == nil {
return nil
}
s.closeOnce.Do(func() {
// Close interconnect devices first to stop any further packet injections.
if s.interconnect != nil {
_ = s.interconnect.GetLSideDevice().Close()
_ = s.interconnect.GetRSideDevice().Close()
}
// close wireguard device
if sess.wireguardDevice != nil {
_ = sess.wireguardDevice.Close()
sess.wireguardDevice = nil
}
if s.systemPacketConn != nil {
_ = s.systemPacketConn.Close()
}
// Close stack last to quiesce any gVisor internal goroutines that may
// hold references to PacketBuffers (prevents dec-ref races).
if sess.stack != nil {
_ = sess.stack.Close()
sess.stack = nil
}
if s.wireguardDevice != nil {
_ = s.wireguardDevice.Close()
}
// Close stack last to quiesce any gVisor internal goroutines that may
// hold references to PacketBuffers (prevents dec-ref races). Keep the
// session pointers stable so concurrent shutdown paths cannot race on
// nil assignments.
if s.stack != nil {
_ = s.stack.Close()
}
})
return nil
}
const defaultOutboundNoReceiveTimeout = time.Minute
func outboundNoReceiveTimeout(config *Config) time.Duration {
if config == nil || config.OutboundNoReceiveTimeoutSec == nil {
return defaultOutboundNoReceiveTimeout
}
return time.Duration(config.GetOutboundNoReceiveTimeoutSec()) * time.Second
}
func createLogicalPacketConn(
ctx context.Context,
dispatcher routing.Dispatcher,
encoding packetaddr.PacketAddrType,
) (cnet.PacketConn, error) {
switch encoding {
case packetaddr.PacketAddrType_None:
return newWireguardPlainPacketConn(ctx, dispatcher)
case packetaddr.PacketAddrType_Packet:
return packetaddr.CreatePacketAddrConn(ctx, dispatcher, false)
case packetaddr.PacketAddrType_Stream:
return packetaddr.CreatePacketAddrConn(ctx, dispatcher, true)
default:
return nil, newError("unsupported outbound packet encoding: ", encoding)
}
}
func (w *WireguardOutbound) createSession(ctx context.Context, dialer internet.Dialer) (*WireguardOutboundSession, error) {
s := &WireguardOutboundSession{
ctx: ctx,
config: w.config,
dnsClient: w.dnsClient,
}
fail := func(err error) (*WireguardOutboundSession, error) {
_ = s.Close()
return nil, err
}
if err := s.initFromConfig(ctx, w.config); err != nil {
return fail(err)
}
if w.config.GetListenOnSystemNetwork() {
packetConn, err := internet.ListenSystemPacket(ctx, &gonet.UDPAddr{IP: cnet.AnyIP.IP(), Port: 0}, nil)
if err != nil {
return fail(newError("failed to listen on system network").Base(err))
}
s.systemPacketConn = packetConn
s.wireguardDevice.SetConn(packetConn)
} else {
encoding := w.config.GetOutboundPacketEncoding()
switch encoding {
case packetaddr.PacketAddrType_None, packetaddr.PacketAddrType_Packet, packetaddr.PacketAddrType_Stream:
default:
return fail(newError("unsupported outbound packet encoding: ", encoding))
}
dispatcher, err := newDialerDispatcher(w, dialer)
if err != nil {
return fail(err)
}
factory := func(ctx context.Context) (cnet.PacketConn, error) {
return createLogicalPacketConn(ctx, dispatcher, encoding)
}
bind := wgcommon.NewReconnectingNetPacketConnToWg(ctx, factory, outboundNoReceiveTimeout(w.config))
s.wireguardDevice.SetBind(bind)
}
if err := s.wireguardDevice.InitDevice(); err != nil {
return fail(newError("failed to init wireguard device").Base(err))
}
if err := s.wireguardDevice.SetupDeviceWithoutPeers(); err != nil {
return fail(newError("failed to setup wireguard device").Base(err))
}
if err := s.wireguardDevice.AddOrReplacePeers(s.config.GetWgDevice().GetPeers()); err != nil {
return fail(newError("failed to add peers").Base(err))
}
if err := s.wireguardDevice.Up(); err != nil {
return fail(newError("failed to bring up wireguard device").Base(err))
}
return s, nil
}
func (w *WireguardOutbound) Process(ctx context.Context, link *transport.Link, dialer internet.Dialer) error {
// keep dialer for address family preference when resolving domain
_ = dialer
if isWireguardUnderlayFor(ctx, w) {
return newError("wireguard outbound cannot use itself as its underlay")
}
storage := envctx.EnvironmentFromContext(w.ctx).(environment.ProxyEnvironment).TransientStorage()
stateIfc, err := storage.Get(ctx, ConnectionState)
if err != nil {
@@ -188,44 +391,31 @@ func (w *WireguardOutbound) Process(ctx context.Context, link *transport.Link, d
}
// create session if needed
sess, err := clientState.GetOrCreateSession(func() (*WireguardOutboundSession, error) {
s := &WireguardOutboundSession{ctx: ctx, config: w.config}
s.dnsClient = w.dnsClient
if err := s.initFromConfig(ctx, w.config); err != nil {
return nil, err
}
if !w.config.ListenOnSystemNetwork {
// SORRRRRY, I tried but it was v2ray's udp support was too difficult to work with
return nil, newError("unimplemented: listenOnSystemNetwork=false is not implemented yet")
}
packetConn, err := internet.ListenSystemPacket(w.ctx, &gonet.UDPAddr{IP: cnet.AnyIP.IP(), Port: 0}, nil)
if err != nil {
return nil, newError("failed to listen on system network").Base(err)
}
s.systemPacketConn = packetConn
s.wireguardDevice.SetConn(packetConn)
// initialize wireguard device now that conn present
if err := s.wireguardDevice.InitDevice(); err != nil {
return nil, newError("failed to init wireguard device").Base(err)
}
if err := s.wireguardDevice.SetupDeviceWithoutPeers(); err != nil {
return nil, newError("failed to setup wireguard device").Base(err)
}
if err := s.wireguardDevice.AddOrReplacePeers(s.config.WgDevice.GetPeers()); err != nil {
return nil, newError("failed to add peers").Base(err)
}
if err := s.wireguardDevice.Up(); err != nil {
return nil, newError("failed to bring up wireguard device").Base(err)
}
return s, nil
})
createSession := func(ctx context.Context) (*WireguardOutboundSession, error) {
return w.createSession(ctx, dialer)
}
sessionBaseCtx := preserveWireguardUnderlayMarkers(w.ctx, ctx)
sess, releaseSession, err := clientState.AcquireOrCreateSessionWithContext(sessionBaseCtx, createSession)
if err != nil {
return newError("failed to create or fetch session").Base(err)
}
defer releaseSession()
// Tie every user of the shared session to its lifetime. ClientConnState
// cancels this context and waits for leases before closing the gVisor stack,
// avoiding concurrent wrapper teardown and use.
sessionCtx := sess.ctx
if sessionCtx == nil {
sessionCtx = context.Background()
}
processCtx, cancelProcess := context.WithCancel(ctx)
stopSessionCancel := context.AfterFunc(sessionCtx, cancelProcess)
if sessionCtx.Err() != nil {
cancelProcess()
}
defer stopSessionCancel()
defer cancelProcess()
ctx = processCtx
{
debugData, err := sess.wireguardDevice.Debug()
@@ -377,15 +567,16 @@ func (w *WireguardOutbound) Close() error {
return nil
}
clientState, ok := stateIfc.(*ClientConnState)
if !ok || clientState.session == nil {
if !ok {
return nil
}
_ = clientState.Close()
return nil
return clientState.Close()
}
func NewClientConnState() (*ClientConnState, error) {
return &ClientConnState{initOnce: &sync.Once{}}, nil
state := &ClientConnState{}
state.ready = sync.NewCond(&state.mu)
return state, nil
}
func init() {
+349
View File
@@ -0,0 +1,349 @@
package outbound
import (
"context"
"errors"
gonet "net"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
cnet "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 {
mu sync.Mutex
destination cnet.Destination
}
func (d *recordingDispatcher) Dispatch(_ context.Context, destination cnet.Destination) (*transport.Link, error) {
d.mu.Lock()
d.destination = destination
d.mu.Unlock()
_, uplinkWriter := pipe.New(pipe.WithSizeLimit(1024))
downlinkReader, _ := pipe.New(pipe.WithSizeLimit(1024))
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 (d *recordingDispatcher) lastDestination() cnet.Destination {
d.mu.Lock()
defer d.mu.Unlock()
return d.destination
}
func TestValidateWireguardConfig(t *testing.T) {
tests := []struct {
name string
config *Config
wantErr bool
}{
{name: "nil config", wantErr: true},
{
name: "restricted system network",
config: &Config{
Restricted: true,
ListenOnSystemNetwork: true,
},
wantErr: true,
},
{
name: "restricted logical network",
config: &Config{Restricted: true},
},
{
name: "unrestricted system network",
config: &Config{ListenOnSystemNetwork: true},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
err := validateWireguardConfig(test.config)
if got := err != nil; got != test.wantErr {
t.Fatalf("validateWireguardConfig() error = %v, wantErr %v", err, test.wantErr)
}
})
}
}
func TestNewWireguardOutboundValidatesBeforeInitialization(t *testing.T) {
_, err := NewWireguardOutbound(context.Background(), &Config{
Restricted: true,
ListenOnSystemNetwork: true,
})
if err == nil {
t.Fatal("NewWireguardOutbound() accepted restricted system network config")
}
if !strings.Contains(err.Error(), "restricted WireGuard outbound cannot listen on the system network") {
t.Fatalf("NewWireguardOutbound() error = %q, want restricted system-network validation error", err)
}
}
func TestCreateLogicalPacketConn(t *testing.T) {
tests := []struct {
name string
encoding packetaddr.PacketAddrType
wantNetwork cnet.Network
wantAddress string
writeFirst bool
}{
{
name: "plain UDP",
encoding: packetaddr.PacketAddrType_None,
wantNetwork: cnet.Network_UDP,
wantAddress: "192.0.2.1",
writeFirst: true,
},
{
name: "packet packetaddr",
encoding: packetaddr.PacketAddrType_Packet,
wantNetwork: cnet.Network_UDP,
wantAddress: "sp.packet-addr.v2fly.arpa",
},
{
name: "stream packetaddr",
encoding: packetaddr.PacketAddrType_Stream,
wantNetwork: cnet.Network_TCP,
wantAddress: "st.packet-addr.v2fly.arpa",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
dispatcher := new(recordingDispatcher)
conn, err := createLogicalPacketConn(context.Background(), dispatcher, test.encoding)
if err != nil {
t.Fatalf("createLogicalPacketConn() failed: %v", err)
}
defer func() { _ = conn.Close() }()
if test.writeFirst {
if _, err := conn.WriteTo([]byte("wireguard"), &gonet.UDPAddr{IP: gonet.ParseIP(test.wantAddress), Port: 51820}); err != nil {
t.Fatalf("WriteTo() failed: %v", err)
}
}
destination := dispatcher.lastDestination()
if destination.Network != test.wantNetwork {
t.Fatalf("network = %v, want %v", destination.Network, test.wantNetwork)
}
if got := destination.Address.String(); got != test.wantAddress {
t.Fatalf("address = %q, want %q", got, test.wantAddress)
}
})
}
}
func TestCreateLogicalPacketConnRejectsUnsupportedEncoding(t *testing.T) {
if _, err := createLogicalPacketConn(context.Background(), new(recordingDispatcher), packetaddr.PacketAddrType(99)); err == nil {
t.Fatal("createLogicalPacketConn() accepted an unsupported encoding")
}
}
func TestOutboundNoReceiveTimeout(t *testing.T) {
disabled := uint32(0)
custom := uint32(7)
tests := []struct {
name string
config *Config
want time.Duration
}{
{name: "nil config", config: nil, want: time.Minute},
{name: "omitted", config: new(Config), want: time.Minute},
{name: "disabled", config: &Config{OutboundNoReceiveTimeoutSec: &disabled}, want: 0},
{name: "custom", config: &Config{OutboundNoReceiveTimeoutSec: &custom}, want: 7 * time.Second},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
if got := outboundNoReceiveTimeout(test.config); got != test.want {
t.Fatalf("outboundNoReceiveTimeout() = %v, want %v", got, test.want)
}
})
}
}
func TestClientConnStateRetriesFailedInitialization(t *testing.T) {
state, err := NewClientConnState()
if err != nil {
t.Fatal(err)
}
attempts := 0
want := new(WireguardOutboundSession)
create := func() (*WireguardOutboundSession, error) {
attempts++
if attempts == 1 {
return nil, errors.New("temporary failure")
}
return want, nil
}
if _, err := state.GetOrCreateSession(create); err == nil {
t.Fatal("first initialization unexpectedly succeeded")
}
got, err := state.GetOrCreateSession(create)
if err != nil {
t.Fatalf("second initialization failed: %v", err)
}
if got != want {
t.Fatalf("session = %p, want %p", got, want)
}
got, err = state.GetOrCreateSession(create)
if err != nil || got != want {
t.Fatalf("cached session = (%p, %v), want (%p, nil)", got, err, want)
}
if attempts != 2 {
t.Fatalf("initialization attempts = %d, want 2", attempts)
}
}
func TestClientConnStateConcurrentInitialization(t *testing.T) {
state, err := NewClientConnState()
if err != nil {
t.Fatal(err)
}
want := new(WireguardOutboundSession)
var attempts atomic.Int32
started := make(chan struct{})
release := make(chan struct{})
create := func() (*WireguardOutboundSession, error) {
if attempt := attempts.Add(1); attempt != 1 {
return nil, errors.New("duplicate concurrent initialization")
}
close(started)
<-release
return want, nil
}
const callers = 16
results := make(chan *WireguardOutboundSession, callers)
errs := make(chan error, callers)
var workers sync.WaitGroup
workers.Add(callers)
for range callers {
go func() {
defer workers.Done()
session, err := state.GetOrCreateSession(create)
results <- session
errs <- err
}()
}
<-started
close(release)
workers.Wait()
close(results)
close(errs)
for err := range errs {
if err != nil {
t.Fatalf("GetOrCreateSession() failed: %v", err)
}
}
for got := range results {
if got != want {
t.Fatalf("session = %p, want %p", got, want)
}
}
if got := attempts.Load(); got != 1 {
t.Fatalf("initialization attempts = %d, want 1", got)
}
}
func TestClientConnStateClosePreventsInitialization(t *testing.T) {
state, err := NewClientConnState()
if err != nil {
t.Fatal(err)
}
if err := state.Close(); err != nil {
t.Fatalf("Close() failed: %v", err)
}
if _, err := state.GetOrCreateSession(func() (*WireguardOutboundSession, error) {
return new(WireguardOutboundSession), nil
}); err == nil {
t.Fatal("closed state allowed initialization")
}
}
func TestClientConnStateCloseCancelsInitialization(t *testing.T) {
state, err := NewClientConnState()
if err != nil {
t.Fatal(err)
}
started := make(chan struct{})
createDone := make(chan error, 1)
go func() {
_, err := state.GetOrCreateSessionWithContext(context.Background(), func(ctx context.Context) (*WireguardOutboundSession, error) {
close(started)
<-ctx.Done()
return nil, ctx.Err()
})
createDone <- err
}()
<-started
closeDone := make(chan error, 1)
go func() { closeDone <- state.Close() }()
select {
case err := <-closeDone:
if err != nil {
t.Fatalf("Close() failed: %v", err)
}
case <-time.After(time.Second):
t.Fatal("Close() did not cancel in-flight initialization")
}
select {
case err := <-createDone:
if err == nil {
t.Fatal("canceled initialization unexpectedly succeeded")
}
case <-time.After(time.Second):
t.Fatal("initialization did not return after cancellation")
}
}
func TestClientConnStateCloseWaitsForActiveSession(t *testing.T) {
state, err := NewClientConnState()
if err != nil {
t.Fatal(err)
}
sess, release, err := state.AcquireOrCreateSessionWithContext(context.Background(), func(ctx context.Context) (*WireguardOutboundSession, error) {
return &WireguardOutboundSession{ctx: ctx}, nil
})
if err != nil {
t.Fatal(err)
}
closeDone := make(chan error, 1)
go func() { closeDone <- state.Close() }()
select {
case <-sess.ctx.Done():
case <-time.After(time.Second):
t.Fatal("Close() did not cancel the shared session context")
}
select {
case err := <-closeDone:
t.Fatalf("Close() returned before the active session was released: %v", err)
default:
}
release()
release()
select {
case err := <-closeDone:
if err != nil {
t.Fatalf("Close() failed: %v", err)
}
case <-time.After(time.Second):
t.Fatal("Close() did not finish after the active session was released")
}
}
@@ -0,0 +1,323 @@
package outbound
import (
"context"
gonet "net"
"sync"
"time"
"github.com/v2fly/v2ray-core/v5/common"
"github.com/v2fly/v2ray-core/v5/common/buf"
cnet "github.com/v2fly/v2ray-core/v5/common/net"
"github.com/v2fly/v2ray-core/v5/features/routing"
"github.com/v2fly/v2ray-core/v5/transport"
)
// wireguardPlainPacketConn presents the multi-destination PacketConn needed by
// a WireGuard bind while keeping each peer on its own dispatcher link. Keeping
// this adapter local is important: the dispatcher is backed by the outbound
// handler's dialer, so its links retain sender proxy settings and WireGuard's
// underlay-recursion markers without changing the shared UDP dispatcher.
type wireguardPlainPacketConn struct {
ctx context.Context
cancel context.CancelFunc
dispatcher routing.Dispatcher
mu sync.Mutex
closed bool
closeErr error
entries map[cnet.Destination]*wireguardPlainPacketEntry
pending map[cnet.Destination]*wireguardPlainPacketDial
received chan wireguardPlainPacket
done chan struct{}
closeOnce sync.Once
}
type wireguardPlainPacket struct {
payload []byte
source cnet.Destination
}
type wireguardPlainPacketDial struct {
done chan struct{}
entry *wireguardPlainPacketEntry
err error
}
type wireguardPlainPacketEntry struct {
destination cnet.Destination
link *transport.Link
cancel context.CancelFunc
writeMu sync.Mutex
closeOnce sync.Once
}
func newWireguardPlainPacketConn(ctx context.Context, dispatcher routing.Dispatcher) (cnet.PacketConn, error) {
if dispatcher == nil {
return nil, newError("routing dispatcher is not configured for plain WireGuard UDP")
}
if ctx == nil {
ctx = context.Background()
}
if err := ctx.Err(); err != nil {
return nil, err
}
connCtx, cancel := context.WithCancel(ctx)
conn := &wireguardPlainPacketConn{
ctx: connCtx,
cancel: cancel,
dispatcher: dispatcher,
entries: make(map[cnet.Destination]*wireguardPlainPacketEntry),
pending: make(map[cnet.Destination]*wireguardPlainPacketDial),
received: make(chan wireguardPlainPacket, 64),
done: make(chan struct{}),
}
go conn.watchContext()
return conn, nil
}
func (c *wireguardPlainPacketConn) watchContext() {
<-c.ctx.Done()
c.shutdown(c.ctx.Err())
}
func (c *wireguardPlainPacketConn) getOrCreateEntry(destination cnet.Destination) (*wireguardPlainPacketEntry, error) {
c.mu.Lock()
if c.closed {
err := c.closedErrorLocked()
c.mu.Unlock()
return nil, err
}
if entry := c.entries[destination]; entry != nil {
c.mu.Unlock()
return entry, nil
}
if dial := c.pending[destination]; dial != nil {
c.mu.Unlock()
select {
case <-dial.done:
if dial.err != nil {
return nil, dial.err
}
if dial.entry == nil {
return nil, newError("plain WireGuard UDP dial completed without a link")
}
return dial.entry, nil
case <-c.done:
return nil, c.closedError()
}
}
dial := &wireguardPlainPacketDial{done: make(chan struct{})}
c.pending[destination] = dial
c.mu.Unlock()
entry, err := c.dialEntry(destination)
c.mu.Lock()
delete(c.pending, destination)
if c.closed {
if err == nil {
err = c.closedErrorLocked()
}
} else if err == nil {
c.entries[destination] = entry
}
dial.entry = entry
dial.err = err
close(dial.done)
c.mu.Unlock()
if err != nil {
if entry != nil {
entry.close()
}
c.shutdown(err)
return nil, err
}
go c.readEntry(entry)
return entry, nil
}
func (c *wireguardPlainPacketConn) dialEntry(destination cnet.Destination) (*wireguardPlainPacketEntry, error) {
entryCtx, cancel := context.WithCancel(c.ctx)
link, err := c.dispatcher.Dispatch(entryCtx, destination)
if err != nil {
entry := &wireguardPlainPacketEntry{link: link, cancel: cancel}
entry.close()
return nil, newError("failed to dial plain WireGuard UDP to ", destination).Base(err)
}
entry := &wireguardPlainPacketEntry{
destination: destination,
link: link,
cancel: cancel,
}
if link == nil {
entry.close()
return nil, newError("dispatcher returned a nil link for plain WireGuard UDP to ", destination)
}
if link.Reader == nil || link.Writer == nil {
entry.close()
return nil, newError("dispatcher returned an incomplete link for plain WireGuard UDP to ", destination)
}
return entry, nil
}
func (c *wireguardPlainPacketConn) readEntry(entry *wireguardPlainPacketEntry) {
for {
mb, err := entry.link.Reader.ReadMultiBuffer()
if err != nil {
buf.ReleaseMulti(mb)
c.shutdown(newError("failed to read plain WireGuard UDP from ", entry.destination).Base(err))
return
}
for i, packet := range mb {
if packet == nil {
continue
}
payload := append([]byte(nil), packet.Bytes()...)
packet.Release()
mb[i] = nil
select {
case c.received <- wireguardPlainPacket{payload: payload, source: entry.destination}:
case <-c.done:
buf.ReleaseMulti(mb[i+1:])
return
}
}
buf.ReleaseMulti(mb)
}
}
func (c *wireguardPlainPacketConn) ReadFrom(p []byte) (int, gonet.Addr, error) {
select {
case packet := <-c.received:
select {
case <-c.done:
return 0, nil, c.closedError()
default:
}
n := copy(p, packet.payload)
return n, &gonet.UDPAddr{
IP: append(gonet.IP(nil), packet.source.Address.IP()...),
Port: int(packet.source.Port),
}, nil
case <-c.done:
return 0, nil, c.closedError()
}
}
func (c *wireguardPlainPacketConn) WriteTo(p []byte, addr gonet.Addr) (int, error) {
udpAddr, ok := addr.(*gonet.UDPAddr)
if !ok || udpAddr == nil {
return 0, newError("plain WireGuard UDP requires a UDP destination")
}
if udpAddr.Port < 0 || udpAddr.Port > 65535 {
return 0, newError("invalid plain WireGuard UDP destination port: ", udpAddr.Port)
}
address := cnet.IPAddress(udpAddr.IP)
if address == nil {
return 0, newError("invalid plain WireGuard UDP destination address: ", udpAddr.IP)
}
destination := cnet.UDPDestination(address, cnet.Port(udpAddr.Port))
entry, err := c.getOrCreateEntry(destination)
if err != nil {
return 0, err
}
packet := buf.NewWithSize(int32(len(p)))
if _, err := packet.Write(p); err != nil {
packet.Release()
return 0, newError("failed to buffer plain WireGuard UDP packet").Base(err)
}
entry.writeMu.Lock()
select {
case <-c.done:
entry.writeMu.Unlock()
packet.Release()
return 0, c.closedError()
default:
}
err = entry.link.Writer.WriteMultiBuffer(buf.MultiBuffer{packet})
entry.writeMu.Unlock()
if err != nil {
err = newError("failed to write plain WireGuard UDP to ", destination).Base(err)
c.shutdown(err)
return 0, err
}
return len(p), nil
}
func (c *wireguardPlainPacketConn) Close() error {
c.shutdown(gonet.ErrClosed)
return nil
}
func (c *wireguardPlainPacketConn) shutdown(err error) {
if err == nil {
err = gonet.ErrClosed
}
c.closeOnce.Do(func() {
c.mu.Lock()
c.closed = true
c.closeErr = err
entries := make([]*wireguardPlainPacketEntry, 0, len(c.entries))
for destination, entry := range c.entries {
entries = append(entries, entry)
delete(c.entries, destination)
}
close(c.done)
c.mu.Unlock()
c.cancel()
for _, entry := range entries {
entry.close()
}
})
}
func (c *wireguardPlainPacketConn) closedError() error {
c.mu.Lock()
defer c.mu.Unlock()
return c.closedErrorLocked()
}
func (c *wireguardPlainPacketConn) closedErrorLocked() error {
if c.closeErr != nil {
return c.closeErr
}
return gonet.ErrClosed
}
func (c *wireguardPlainPacketConn) LocalAddr() gonet.Addr {
return &gonet.UDPAddr{IP: gonet.IPv4zero}
}
func (*wireguardPlainPacketConn) SetDeadline(time.Time) error { return nil }
func (*wireguardPlainPacketConn) SetReadDeadline(time.Time) error { return nil }
func (*wireguardPlainPacketConn) SetWriteDeadline(time.Time) error { return nil }
func (e *wireguardPlainPacketEntry) close() {
if e == nil {
return
}
e.closeOnce.Do(func() {
if e.cancel != nil {
e.cancel()
}
if e.link != nil {
_ = common.Interrupt(e.link.Reader)
_ = common.Interrupt(e.link.Writer)
}
})
}
@@ -0,0 +1,484 @@
package outbound
import (
"context"
"errors"
"io"
gonet "net"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/v2fly/v2ray-core/v5/common/buf"
cnet "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"
)
type plainPacketConnTestDispatcher struct {
dispatch func(context.Context, cnet.Destination) (*transport.Link, error)
}
func (d *plainPacketConnTestDispatcher) Dispatch(ctx context.Context, destination cnet.Destination) (*transport.Link, error) {
return d.dispatch(ctx, destination)
}
func (*plainPacketConnTestDispatcher) Start() error { return nil }
func (*plainPacketConnTestDispatcher) Close() error { return nil }
func (*plainPacketConnTestDispatcher) Type() interface{} { return routing.DispatcherType() }
type plainPacketConnTestReadResult struct {
buffer buf.MultiBuffer
err error
}
type plainPacketConnTestReader struct {
results chan plainPacketConnTestReadResult
interrupted chan struct{}
secondRead chan struct{}
interruptOnce sync.Once
secondReadOnce sync.Once
interruptCalls atomic.Int32
readCalls atomic.Int32
}
func newPlainPacketConnTestReader() *plainPacketConnTestReader {
return &plainPacketConnTestReader{
results: make(chan plainPacketConnTestReadResult, 8),
interrupted: make(chan struct{}),
secondRead: make(chan struct{}),
}
}
func (r *plainPacketConnTestReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
if r.readCalls.Add(1) == 2 {
r.secondReadOnce.Do(func() { close(r.secondRead) })
}
select {
case result := <-r.results:
return result.buffer, result.err
case <-r.interrupted:
return nil, io.ErrClosedPipe
}
}
func (r *plainPacketConnTestReader) Interrupt() {
r.interruptCalls.Add(1)
r.interruptOnce.Do(func() { close(r.interrupted) })
}
type plainPacketConnTestWriter struct {
writes chan []byte
interrupted chan struct{}
interruptOnce sync.Once
interruptCalls atomic.Int32
err error
}
func newPlainPacketConnTestWriter() *plainPacketConnTestWriter {
return &plainPacketConnTestWriter{
writes: make(chan []byte, 64),
interrupted: make(chan struct{}),
}
}
func (w *plainPacketConnTestWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
payload := make([]byte, mb.Len())
mb.Copy(payload)
buf.ReleaseMulti(mb)
if w.err != nil {
return w.err
}
select {
case w.writes <- payload:
return nil
case <-w.interrupted:
return gonet.ErrClosed
}
}
func (w *plainPacketConnTestWriter) Interrupt() {
w.interruptCalls.Add(1)
w.interruptOnce.Do(func() { close(w.interrupted) })
}
type plainPacketConnTestRay struct {
ctx context.Context
reader *plainPacketConnTestReader
writer *plainPacketConnTestWriter
}
func newPlainPacketConnTestRay(ctx context.Context) *plainPacketConnTestRay {
return &plainPacketConnTestRay{
ctx: ctx,
reader: newPlainPacketConnTestReader(),
writer: newPlainPacketConnTestWriter(),
}
}
func (r *plainPacketConnTestRay) link() *transport.Link {
return &transport.Link{Reader: r.reader, Writer: r.writer}
}
func plainPacketConnTestWait(t *testing.T, signal <-chan struct{}, description string) {
t.Helper()
select {
case <-signal:
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for " + description)
}
}
func plainPacketConnTestWrite(t *testing.T, conn gonet.PacketConn, payload string, address gonet.Addr) {
t.Helper()
if n, err := conn.WriteTo([]byte(payload), address); err != nil || n != len(payload) {
t.Fatalf("WriteTo(%q) = (%d, %v), want (%d, nil)", payload, n, err, len(payload))
}
}
func plainPacketConnTestAwaitPayload(t *testing.T, writer *plainPacketConnTestWriter, want string) {
t.Helper()
select {
case payload := <-writer.writes:
if got := string(payload); got != want {
t.Fatalf("written payload = %q, want %q", got, want)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for payload " + want)
}
}
func plainPacketConnTestBuffer(payload string) *buf.Buffer {
buffer := buf.New()
_, _ = buffer.WriteString(payload)
return buffer
}
func TestWireguardPlainPacketConnReusesDestinationConnections(t *testing.T) {
var access sync.Mutex
rays := make(map[cnet.Destination]*plainPacketConnTestRay)
dispatcher := &plainPacketConnTestDispatcher{dispatch: func(ctx context.Context, destination cnet.Destination) (*transport.Link, error) {
ray := newPlainPacketConnTestRay(ctx)
access.Lock()
rays[destination] = ray
access.Unlock()
return ray.link(), nil
}}
conn, err := createLogicalPacketConn(context.Background(), dispatcher, packetaddr.PacketAddrType_None)
if err != nil {
t.Fatal(err)
}
defer func() { _ = conn.Close() }()
addressA := &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.1"), Port: 51820}
addressB := &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.2"), Port: 51821}
plainPacketConnTestWrite(t, conn, "a1", addressA)
plainPacketConnTestWrite(t, conn, "a2", addressA)
plainPacketConnTestWrite(t, conn, "b1", addressB)
access.Lock()
if len(rays) != 2 {
access.Unlock()
t.Fatalf("dispatch count = %d, want one per destination (2)", len(rays))
}
rayA := rays[cnet.DestinationFromAddr(addressA)]
rayB := rays[cnet.DestinationFromAddr(addressB)]
access.Unlock()
if rayA == nil || rayB == nil {
t.Fatalf("dispatched destinations do not contain both peers: %v", rays)
}
plainPacketConnTestAwaitPayload(t, rayA.writer, "a1")
plainPacketConnTestAwaitPayload(t, rayA.writer, "a2")
plainPacketConnTestAwaitPayload(t, rayB.writer, "b1")
rayB.reader.results <- plainPacketConnTestReadResult{buffer: buf.MultiBuffer{plainPacketConnTestBuffer("reply")}}
readBuffer := make([]byte, 32)
n, source, err := conn.ReadFrom(readBuffer)
if err != nil {
t.Fatal(err)
}
if got := string(readBuffer[:n]); got != "reply" {
t.Fatalf("read payload = %q, want reply", got)
}
if got, want := source.String(), addressB.String(); got != want {
t.Fatalf("source = %q, want %q", got, want)
}
}
func TestWireguardPlainPacketConnDeduplicatesConcurrentDial(t *testing.T) {
var dispatchCalls atomic.Int32
dispatchStarted := make(chan struct{})
releaseDispatch := make(chan struct{})
var startOnce sync.Once
dispatcher := &plainPacketConnTestDispatcher{dispatch: func(ctx context.Context, destination cnet.Destination) (*transport.Link, error) {
dispatchCalls.Add(1)
startOnce.Do(func() { close(dispatchStarted) })
<-releaseDispatch
return newPlainPacketConnTestRay(ctx).link(), nil
}}
conn, err := newWireguardPlainPacketConn(context.Background(), dispatcher)
if err != nil {
t.Fatal(err)
}
defer func() { _ = conn.Close() }()
const writers = 24
startWrites := make(chan struct{})
writeResults := make(chan error, writers)
address := &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.10"), Port: 51820}
for i := 0; i < writers; i++ {
go func() {
<-startWrites
_, err := conn.WriteTo([]byte("packet"), address)
writeResults <- err
}()
}
close(startWrites)
plainPacketConnTestWait(t, dispatchStarted, "first dispatch")
// Keep the first dial pending long enough for all concurrent writers to
// reach the per-destination in-flight entry.
time.Sleep(100 * time.Millisecond)
close(releaseDispatch)
for i := 0; i < writers; i++ {
select {
case err := <-writeResults:
if err != nil {
t.Fatalf("concurrent WriteTo failed: %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for concurrent WriteTo")
}
}
if got := dispatchCalls.Load(); got != 1 {
t.Fatalf("dispatch count = %d, want one concurrent dial", got)
}
}
func TestWireguardPlainPacketConnDialErrorTerminatesGeneration(t *testing.T) {
wantErr := errors.New("dial failed")
var attempts atomic.Int32
dispatcher := &plainPacketConnTestDispatcher{dispatch: func(ctx context.Context, destination cnet.Destination) (*transport.Link, error) {
attempts.Add(1)
return nil, wantErr
}}
conn, err := newWireguardPlainPacketConn(context.Background(), dispatcher)
if err != nil {
t.Fatal(err)
}
defer func() { _ = conn.Close() }()
readResult := make(chan error, 1)
go func() {
_, _, err := conn.ReadFrom(make([]byte, 32))
readResult <- err
}()
address := &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.20"), Port: 51820}
if n, err := conn.WriteTo([]byte("first"), address); n != 0 || !errors.Is(err, wantErr) {
t.Fatalf("first WriteTo = (%d, %v), want (0, %v)", n, err, wantErr)
}
select {
case err := <-readResult:
if !errors.Is(err, wantErr) {
t.Fatalf("ReadFrom error = %v, want dial error", err)
}
case <-time.After(2 * time.Second):
t.Fatal("ReadFrom remained blocked after dial failure")
}
if n, err := conn.WriteTo([]byte("second"), address); n != 0 || !errors.Is(err, wantErr) {
t.Fatalf("second WriteTo = (%d, %v), want terminal dial error", n, err)
}
if got := attempts.Load(); got != 1 {
t.Fatalf("dial attempts = %d, want terminal generation to avoid redial", got)
}
}
func TestWireguardPlainPacketConnPropagatesReadError(t *testing.T) {
wantErr := errors.New("underlay read failed")
response := plainPacketConnTestBuffer("must be released")
var ray *plainPacketConnTestRay
dispatcher := &plainPacketConnTestDispatcher{dispatch: func(ctx context.Context, destination cnet.Destination) (*transport.Link, error) {
ray = newPlainPacketConnTestRay(ctx)
ray.reader.results <- plainPacketConnTestReadResult{
buffer: buf.MultiBuffer{response},
err: wantErr,
}
return ray.link(), nil
}}
conn, err := newWireguardPlainPacketConn(context.Background(), dispatcher)
if err != nil {
t.Fatal(err)
}
defer func() { _ = conn.Close() }()
plainPacketConnTestWrite(t, conn, "request", &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.30"), Port: 51820})
plainPacketConnTestWait(t, ray.reader.interrupted, "failed ray reader interruption")
plainPacketConnTestWait(t, ray.writer.interrupted, "failed ray writer interruption")
if n, source, err := conn.ReadFrom(make([]byte, 32)); n != 0 || source != nil || !errors.Is(err, wantErr) {
t.Fatalf("ReadFrom = (%d, %v, %v), want (0, nil, %v)", n, source, err, wantErr)
}
if response.Len() != 0 {
t.Fatal("buffer returned alongside a read error was not released")
}
}
func TestWireguardPlainPacketConnCloseUnblocksRead(t *testing.T) {
dispatcher := &plainPacketConnTestDispatcher{dispatch: func(context.Context, cnet.Destination) (*transport.Link, error) {
t.Fatal("unexpected dispatch")
return nil, nil
}}
conn, err := newWireguardPlainPacketConn(context.Background(), dispatcher)
if err != nil {
t.Fatal(err)
}
readResult := make(chan error, 1)
go func() {
_, _, err := conn.ReadFrom(make([]byte, 32))
readResult <- err
}()
if err := conn.Close(); err != nil {
t.Fatal(err)
}
if err := conn.Close(); err != nil {
t.Fatal(err)
}
select {
case err := <-readResult:
if !errors.Is(err, gonet.ErrClosed) {
t.Fatalf("ReadFrom error = %v, want net.ErrClosed", err)
}
case <-time.After(2 * time.Second):
t.Fatal("ReadFrom remained blocked after Close")
}
if n, err := conn.WriteTo([]byte("rejected"), &gonet.UDPAddr{IP: gonet.IPv4zero, Port: 51820}); n != 0 || !errors.Is(err, gonet.ErrClosed) {
t.Fatalf("WriteTo after Close = (%d, %v), want (0, net.ErrClosed)", n, err)
}
}
func TestWireguardPlainPacketConnContextCancellation(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
dispatchCtx := make(chan context.Context, 1)
var ray *plainPacketConnTestRay
dispatcher := &plainPacketConnTestDispatcher{dispatch: func(ctx context.Context, destination cnet.Destination) (*transport.Link, error) {
dispatchCtx <- ctx
ray = newPlainPacketConnTestRay(ctx)
return ray.link(), nil
}}
conn, err := newWireguardPlainPacketConn(ctx, dispatcher)
if err != nil {
t.Fatal(err)
}
defer func() { _ = conn.Close() }()
plainPacketConnTestWrite(t, conn, "request", &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.40"), Port: 51820})
rayCtx := <-dispatchCtx
readResult := make(chan error, 1)
go func() {
_, _, err := conn.ReadFrom(make([]byte, 32))
readResult <- err
}()
cancel()
select {
case err := <-readResult:
if !errors.Is(err, context.Canceled) {
t.Fatalf("ReadFrom error = %v, want context.Canceled", err)
}
case <-time.After(2 * time.Second):
t.Fatal("ReadFrom remained blocked after context cancellation")
}
plainPacketConnTestWait(t, rayCtx.Done(), "ray context cancellation")
plainPacketConnTestWait(t, ray.reader.interrupted, "ray reader interruption")
plainPacketConnTestWait(t, ray.writer.interrupted, "ray writer interruption")
}
func TestWireguardPlainPacketConnOwnsBuffersAndConnections(t *testing.T) {
t.Run("read releases delivered buffer", func(t *testing.T) {
response := plainPacketConnTestBuffer("response")
var ray *plainPacketConnTestRay
dispatcher := &plainPacketConnTestDispatcher{dispatch: func(ctx context.Context, destination cnet.Destination) (*transport.Link, error) {
ray = newPlainPacketConnTestRay(ctx)
ray.reader.results <- plainPacketConnTestReadResult{buffer: buf.MultiBuffer{response}}
return ray.link(), nil
}}
conn, err := newWireguardPlainPacketConn(context.Background(), dispatcher)
if err != nil {
t.Fatal(err)
}
defer func() { _ = conn.Close() }()
plainPacketConnTestWrite(t, conn, "request", &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.50"), Port: 51820})
plainPacketConnTestWait(t, ray.reader.secondRead, "response to enter read queue")
readBuffer := make([]byte, 32)
n, _, err := conn.ReadFrom(readBuffer)
if err != nil {
t.Fatal(err)
}
if got := string(readBuffer[:n]); got != "response" {
t.Fatalf("response = %q, want response", got)
}
if response.Len() != 0 {
t.Fatal("delivered response buffer was not released")
}
})
t.Run("close releases queue and owns rays exactly once", func(t *testing.T) {
queued := plainPacketConnTestBuffer("queued response")
var access sync.Mutex
var rays []*plainPacketConnTestRay
dispatcher := &plainPacketConnTestDispatcher{dispatch: func(ctx context.Context, destination cnet.Destination) (*transport.Link, error) {
ray := newPlainPacketConnTestRay(ctx)
access.Lock()
if len(rays) == 0 {
ray.reader.results <- plainPacketConnTestReadResult{buffer: buf.MultiBuffer{queued}}
}
rays = append(rays, ray)
access.Unlock()
return ray.link(), nil
}}
conn, err := newWireguardPlainPacketConn(context.Background(), dispatcher)
if err != nil {
t.Fatal(err)
}
plainPacketConnTestWrite(t, conn, "one", &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.60"), Port: 51820})
plainPacketConnTestWrite(t, conn, "two", &gonet.UDPAddr{IP: gonet.ParseIP("192.0.2.61"), Port: 51820})
access.Lock()
firstRay := rays[0]
access.Unlock()
plainPacketConnTestWait(t, firstRay.reader.secondRead, "response to enter queue")
var closeGroup sync.WaitGroup
for i := 0; i < 12; i++ {
closeGroup.Add(1)
go func() {
defer closeGroup.Done()
if err := conn.Close(); err != nil {
t.Errorf("Close failed: %v", err)
}
}()
}
closeGroup.Wait()
if queued.Len() != 0 {
t.Fatal("queued response buffer was not released by Close")
}
access.Lock()
defer access.Unlock()
if len(rays) != 2 {
t.Fatalf("ray count = %d, want 2", len(rays))
}
for i, ray := range rays {
plainPacketConnTestWait(t, ray.ctx.Done(), "ray context cancellation")
if got := ray.reader.interruptCalls.Load(); got != 1 {
t.Errorf("ray %d reader Interrupt calls = %d, want 1", i, got)
}
if got := ray.writer.interruptCalls.Load(); got != 1 {
t.Errorf("ray %d writer Interrupt calls = %d, want 1", i, got)
}
}
})
}
+6 -2
View File
@@ -23,8 +23,12 @@ func (w *WrappedWireguardDevice) InitDevice() error {
return err
}
// Create wireguard bind adapter from provided PacketConn
bind := NewNetPacketConnToWg(w.conn)
// Use an explicit bind when provided, otherwise preserve the static
// PacketConn adapter used by existing callers.
bind := w.bind
if bind == nil {
bind = NewNetPacketConnToWg(w.conn)
}
// Create the wireguard device with our logger adapter.
dev := device.NewDevice(tunDev, bind, NewDeviceLoggerAdapter())
+28 -10
View File
@@ -85,23 +85,40 @@ func (n *netPacketConnToWg) Open(port uint16) (fns []conn.ReceiveFunc, actualPor
}
n.mu.Lock()
n.closed = false
n.mu.Unlock()
// Clear the read deadline that Close() may have set so reads can proceed.
_ = n.conn.SetReadDeadline(time.Time{})
packetConn := n.conn
// determine actualPort from LocalAddr if possible
if la := n.conn.LocalAddr(); la != nil {
if la := packetConn.LocalAddr(); la != nil {
if ua, ok := la.(*gonet.UDPAddr); ok {
n.actualPort = uint16(ua.Port)
}
}
actualPort = n.actualPort
n.mu.Unlock()
// Clear the read deadline that Close() may have set so reads can proceed.
_ = packetConn.SetReadDeadline(time.Time{})
fn := func(packets [][]byte, sizes []int, eps []conn.Endpoint) (int, error) {
var i int
for i = 0; i < len(packets); i++ {
nRead, addr, err := n.conn.ReadFrom(packets[i])
n.mu.Lock()
closed := n.closed
n.mu.Unlock()
if closed {
return 0, gonet.ErrClosed
}
nRead, addr, err := packetConn.ReadFrom(packets[i])
n.mu.Lock()
closed = n.closed
n.mu.Unlock()
if closed {
// Close uses a deadline to wake a blocked system-socket read. Map
// that implementation-specific error to the error required by
// conn.Bind so WireGuard stops its receive routine immediately.
return 0, gonet.ErrClosed
}
if err != nil {
if i == 0 {
return 0, err
@@ -131,16 +148,17 @@ func (n *netPacketConnToWg) Open(port uint16) (fns []conn.ReceiveFunc, actualPor
func (n *netPacketConnToWg) Close() error {
n.mu.Lock()
defer n.mu.Unlock()
n.closed = true
packetConn := n.conn
n.mu.Unlock()
// Do NOT close the underlying conn here. WireGuard calls Close()+Open()
// internally during BindUpdate(). The actual PacketConn lifecycle is
// managed externally by the session that created it.
//
// Set a past read deadline to unblock any pending ReadFrom calls in
// receive goroutines so that WireGuard's stopping.Wait() can complete.
if n.conn != nil {
_ = n.conn.SetReadDeadline(time.Unix(1, 0))
if packetConn != nil {
_ = packetConn.SetReadDeadline(time.Unix(1, 0))
}
return nil
}
+13 -7
View File
@@ -75,9 +75,8 @@ func (w *networkLayerWriter) Write(packet []byte) (int, error) {
p := make([]byte, len(packet))
copy(p, packet)
w.parent.mu.RLock()
closed := w.parent.closed
w.parent.mu.RUnlock()
if closed {
defer w.parent.mu.RUnlock()
if w.parent.closed {
return 0, errors.New("device closed")
}
select {
@@ -141,10 +140,18 @@ func (n *NetworkLayerDeviceToWireguardTunDeviceAdaptor) Read(bufs [][]byte, size
}
func (n *NetworkLayerDeviceToWireguardTunDeviceAdaptor) Write(bufs [][]byte, offset int) (int, error) {
written := 0
if n.networkLayerSwitch == nil {
n.mu.RLock()
closed := n.closed
device := n.networkLayerSwitch
n.mu.RUnlock()
if closed {
return 0, os.ErrClosed
}
if device == nil {
return 0, errors.New("no network layer writer attached")
}
written := 0
for i := 0; i < len(bufs); i++ {
b := bufs[i]
if b == nil || len(b) <= offset {
@@ -156,7 +163,7 @@ func (n *NetworkLayerDeviceToWireguardTunDeviceAdaptor) Write(bufs [][]byte, off
payload := b[offset:]
cp := make([]byte, len(payload))
copy(cp, payload)
_, err := n.networkLayerSwitch.Write(cp)
_, err := device.Write(cp)
if err != nil {
return written, err
}
@@ -189,7 +196,6 @@ func (n *NetworkLayerDeviceToWireguardTunDeviceAdaptor) Close() error {
// Close events channel to signal no more events.
close(n.events)
dev := n.networkLayerSwitch
n.networkLayerSwitch = nil
n.mu.Unlock()
if dev != nil {
_ = dev.Close()
@@ -0,0 +1,478 @@
package wgcommon
import (
"context"
"errors"
"net"
"net/netip"
"sync"
"time"
"golang.zx2c4.com/wireguard/conn"
)
// PacketConnFactory creates a PacketConn for one logical WireGuard underlay
// generation. The supplied context is canceled when the bind is closed.
type PacketConnFactory func(context.Context) (net.PacketConn, error)
var (
errNilPacketConnFactory = errors.New("wireguard: packet connection factory is nil")
errNilPacketConn = errors.New("wireguard: packet connection factory returned nil")
errPacketConnUnavailable = errors.New("wireguard: packet connection is unavailable")
)
// reconnectingNetPacketConnToWg owns the PacketConns returned by its factory.
// WireGuard's Bind Close/Open lifecycle is distinct from connection generation:
// a connection may be replaced multiple times while one Bind Open remains live.
type reconnectingNetPacketConnToWg struct {
ctx context.Context
factory PacketConnFactory
noReceiveTimeout time.Duration
now func() time.Time
retryWait func(context.Context, time.Duration) error
// replaceMu serializes factories and connection generation changes. Close
// deliberately does not take it: it must be able to cancel an in-flight
// factory and physically close a blocked read immediately.
replaceMu sync.Mutex
mu sync.Mutex
open bool
lifecycle uint64
generation uint64
openCtx context.Context
cancel context.CancelFunc
conn net.PacketConn
actualPort uint16
lastReceive time.Time
}
// NewReconnectingNetPacketConnToWg constructs a WireGuard bind that owns and
// recreates factory-produced PacketConns. A zero or negative timeout disables
// replacement based solely on the absence of received packets; visible I/O
// errors still replace the connection.
func NewReconnectingNetPacketConnToWg(ctx context.Context, factory PacketConnFactory, noReceiveTimeout time.Duration) conn.Bind {
return newReconnectingNetPacketConnToWg(ctx, factory, noReceiveTimeout, time.Now)
}
func newReconnectingNetPacketConnToWg(ctx context.Context, factory PacketConnFactory, noReceiveTimeout time.Duration, now func() time.Time) conn.Bind {
if ctx == nil {
ctx = context.Background()
}
if now == nil {
now = time.Now
}
return &reconnectingNetPacketConnToWg{
ctx: ctx,
factory: factory,
noReceiveTimeout: noReceiveTimeout,
now: now,
retryWait: waitForReconnectRetry,
}
}
func waitForReconnectRetry(ctx context.Context, delay time.Duration) error {
timer := time.NewTimer(delay)
defer timer.Stop()
select {
case <-timer.C:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (b *reconnectingNetPacketConnToWg) Open(port uint16) ([]conn.ReceiveFunc, uint16, error) {
b.replaceMu.Lock()
defer b.replaceMu.Unlock()
b.mu.Lock()
if b.open {
b.mu.Unlock()
return nil, 0, conn.ErrBindAlreadyOpen
}
if b.factory == nil {
b.mu.Unlock()
return nil, 0, errNilPacketConnFactory
}
b.lifecycle++
lifecycle := b.lifecycle
b.openCtx, b.cancel = context.WithCancel(b.ctx)
openCtx := b.openCtx
b.open = true
b.conn = nil
b.actualPort = 0
b.lastReceive = time.Time{}
b.mu.Unlock()
packetConn, err := b.factory(openCtx)
if err == nil && packetConn == nil {
err = errNilPacketConn
}
if err != nil && packetConn != nil {
_ = packetConn.Close()
packetConn = nil
}
if err == nil && openCtx.Err() != nil {
_ = packetConn.Close()
packetConn = nil
err = openCtx.Err()
}
if err != nil {
b.failOpen(lifecycle)
return nil, 0, err
}
b.mu.Lock()
if !b.open || b.lifecycle != lifecycle {
b.mu.Unlock()
_ = packetConn.Close()
return nil, 0, net.ErrClosed
}
b.installLocked(packetConn)
actualPort := b.actualPort
b.mu.Unlock()
receive := func(packets [][]byte, sizes []int, eps []conn.Endpoint) (int, error) {
return b.receive(lifecycle, packets, sizes, eps)
}
return []conn.ReceiveFunc{receive}, actualPort, nil
}
func (b *reconnectingNetPacketConnToWg) failOpen(lifecycle uint64) {
b.mu.Lock()
if b.open && b.lifecycle == lifecycle {
cancel := b.cancel
b.open = false
b.openCtx = nil
b.cancel = nil
b.lifecycle++
b.mu.Unlock()
if cancel != nil {
cancel()
}
return
}
b.mu.Unlock()
}
func (b *reconnectingNetPacketConnToWg) installLocked(packetConn net.PacketConn) {
b.conn = packetConn
b.generation++
b.lastReceive = b.now()
b.actualPort = packetConnPort(packetConn)
}
func packetConnPort(packetConn net.PacketConn) uint16 {
if packetConn == nil {
return 0
}
if udpAddr, ok := packetConn.LocalAddr().(*net.UDPAddr); ok && udpAddr != nil {
return uint16(udpAddr.Port)
}
return 0
}
func (b *reconnectingNetPacketConnToWg) Close() error {
b.mu.Lock()
if !b.open {
b.mu.Unlock()
return nil
}
b.open = false
b.lifecycle++
b.generation++
packetConn := b.conn
b.conn = nil
b.actualPort = 0
b.lastReceive = time.Time{}
cancel := b.cancel
b.cancel = nil
b.openCtx = nil
b.mu.Unlock()
if cancel != nil {
cancel()
}
if packetConn != nil {
return packetConn.Close()
}
return nil
}
func (b *reconnectingNetPacketConnToWg) SetMark(mark uint32) error {
return nil
}
func (b *reconnectingNetPacketConnToWg) Send(bufs [][]byte, ep conn.Endpoint) error {
packetConn, generation, lifecycle, err := b.connectionForSend()
if err != nil {
return err
}
udpAddr, err := net.ResolveUDPAddr("udp", ep.DstToString())
if err != nil {
return err
}
for _, packet := range bufs {
if _, err := packetConn.WriteTo(packet, udpAddr); err != nil {
// The failed packet is deliberately not replayed. WireGuard handles
// retransmission, while a synchronous replacement prepares the next
// send to use a fresh logical underlay.
_, _, _ = b.replace(lifecycle, generation, false)
return err
}
}
return nil
}
func (b *reconnectingNetPacketConnToWg) connectionForSend() (net.PacketConn, uint64, uint64, error) {
b.mu.Lock()
if !b.open {
b.mu.Unlock()
return nil, 0, 0, net.ErrClosed
}
lifecycle := b.lifecycle
packetConn := b.conn
generation := b.generation
expired := packetConn != nil && b.expiredLocked()
b.mu.Unlock()
if packetConn == nil {
return b.ensureConnection(lifecycle)
}
if expired {
packetConn, generation, replaceErr := b.replace(lifecycle, generation, true)
return packetConn, generation, lifecycle, replaceErr
}
return packetConn, generation, lifecycle, nil
}
func (b *reconnectingNetPacketConnToWg) expiredLocked() bool {
return b.noReceiveTimeout > 0 &&
!b.lastReceive.IsZero() &&
b.now().Sub(b.lastReceive) >= b.noReceiveTimeout
}
func (b *reconnectingNetPacketConnToWg) ensureConnection(lifecycle uint64) (net.PacketConn, uint64, uint64, error) {
b.replaceMu.Lock()
defer b.replaceMu.Unlock()
b.mu.Lock()
if !b.open || b.lifecycle != lifecycle {
b.mu.Unlock()
return nil, 0, lifecycle, net.ErrClosed
}
if b.conn != nil {
packetConn, generation := b.conn, b.generation
b.mu.Unlock()
return packetConn, generation, lifecycle, nil
}
openCtx := b.openCtx
b.mu.Unlock()
return b.createAndInstall(lifecycle, openCtx)
}
// replace invalidates expectedGeneration when it is still current. When
// onlyIfExpired is true it rechecks the receive timestamp while holding the
// replacement lock, preventing a concurrent successful receive from causing
// an unnecessary rotation.
func (b *reconnectingNetPacketConnToWg) replace(lifecycle, expectedGeneration uint64, onlyIfExpired bool) (net.PacketConn, uint64, error) {
b.replaceMu.Lock()
defer b.replaceMu.Unlock()
b.mu.Lock()
if !b.open || b.lifecycle != lifecycle {
b.mu.Unlock()
return nil, 0, net.ErrClosed
}
if b.generation != expectedGeneration {
packetConn, generation := b.conn, b.generation
b.mu.Unlock()
if packetConn == nil {
return nil, generation, errPacketConnUnavailable
}
return packetConn, generation, nil
}
if b.conn != nil && onlyIfExpired && !b.expiredLocked() {
packetConn, generation := b.conn, b.generation
b.mu.Unlock()
return packetConn, generation, nil
}
packetConn := b.conn
b.conn = nil
b.generation++
b.actualPort = 0
b.lastReceive = time.Time{}
openCtx := b.openCtx
b.mu.Unlock()
// Logical PacketConns may implement deadlines as no-ops. Physical Close is
// therefore required to wake the receive function before installing the
// next generation.
if packetConn != nil {
_ = packetConn.Close()
}
newConn, generation, _, err := b.createAndInstall(lifecycle, openCtx)
return newConn, generation, err
}
func (b *reconnectingNetPacketConnToWg) createAndInstall(lifecycle uint64, openCtx context.Context) (net.PacketConn, uint64, uint64, error) {
packetConn, err := b.factory(openCtx)
if err == nil && packetConn == nil {
err = errNilPacketConn
}
if err != nil && packetConn != nil {
_ = packetConn.Close()
packetConn = nil
}
if err == nil && openCtx.Err() != nil {
_ = packetConn.Close()
packetConn = nil
err = openCtx.Err()
}
if err != nil {
b.mu.Lock()
closed := !b.open || b.lifecycle != lifecycle
b.mu.Unlock()
if closed {
return nil, 0, lifecycle, net.ErrClosed
}
return nil, 0, lifecycle, err
}
b.mu.Lock()
if !b.open || b.lifecycle != lifecycle || openCtx != b.openCtx {
b.mu.Unlock()
_ = packetConn.Close()
return nil, 0, lifecycle, net.ErrClosed
}
if b.conn != nil {
// A defensive guard for future callers that might create outside
// replaceMu. Never overwrite a newer generation.
current, generation := b.conn, b.generation
b.mu.Unlock()
_ = packetConn.Close()
return current, generation, lifecycle, nil
}
b.installLocked(packetConn)
generation := b.generation
b.mu.Unlock()
return packetConn, generation, lifecycle, nil
}
func (b *reconnectingNetPacketConnToWg) receive(lifecycle uint64, packets [][]byte, sizes []int, eps []conn.Endpoint) (int, error) {
retryAttempt := 0
for i := 0; i < len(packets); i++ {
for {
packetConn, generation, _, err := b.ensureConnection(lifecycle)
if err != nil {
if errors.Is(err, net.ErrClosed) {
return 0, net.ErrClosed
}
if err := b.waitForReceiveRetry(lifecycle, retryAttempt); err != nil {
return 0, err
}
retryAttempt++
continue
}
retryAttempt = 0
n, addr, err := packetConn.ReadFrom(packets[i])
if err != nil {
if _, _, replaceErr := b.replace(lifecycle, generation, false); replaceErr != nil {
if errors.Is(replaceErr, net.ErrClosed) {
return 0, net.ErrClosed
}
if err := b.waitForReceiveRetry(lifecycle, retryAttempt); err != nil {
return 0, err
}
retryAttempt++
continue
}
retryAttempt = 0
continue
}
b.mu.Lock()
if !b.open || b.lifecycle != lifecycle {
b.mu.Unlock()
return 0, net.ErrClosed
}
if b.generation != generation || b.conn != packetConn {
b.mu.Unlock()
// A packet racing with replacement belongs to a physically closed
// generation. Discard it and continue on the current connection.
continue
}
b.lastReceive = b.now()
b.mu.Unlock()
sizes[i] = n
eps[i] = endpointFromAddr(addr)
break
}
}
return len(packets), nil
}
func (b *reconnectingNetPacketConnToWg) waitForReceiveRetry(lifecycle uint64, attempt int) error {
b.mu.Lock()
if !b.open || b.lifecycle != lifecycle {
b.mu.Unlock()
return net.ErrClosed
}
openCtx := b.openCtx
wait := b.retryWait
b.mu.Unlock()
// Start at 100ms and cap at one second. The cap lets a recovered logical
// route become useful promptly without allowing a failed factory to spin.
delay := 100 * time.Millisecond
for i := 0; i < attempt && delay < time.Second; i++ {
delay *= 2
if delay > time.Second {
delay = time.Second
}
}
if wait == nil {
wait = waitForReconnectRetry
}
if err := wait(openCtx, delay); err != nil {
b.mu.Lock()
closed := !b.open || b.lifecycle != lifecycle
b.mu.Unlock()
if closed {
return net.ErrClosed
}
return err
}
return nil
}
func endpointFromAddr(addr net.Addr) conn.Endpoint {
if udpAddr, ok := addr.(*net.UDPAddr); ok && udpAddr != nil {
ip, _ := netip.AddrFromSlice(udpAddr.IP)
return &wgEndpoint{ap: netip.AddrPortFrom(ip, uint16(udpAddr.Port))}
}
if addr != nil {
if addrPort, err := netip.ParseAddrPort(addr.String()); err == nil {
return &wgEndpoint{ap: addrPort}
}
}
return &wgEndpoint{}
}
func (b *reconnectingNetPacketConnToWg) ParseEndpoint(s string) (conn.Endpoint, error) {
addrPort, err := netip.ParseAddrPort(s)
if err != nil {
return nil, err
}
return &wgEndpoint{ap: addrPort}, nil
}
func (b *reconnectingNetPacketConnToWg) BatchSize() int {
return 1
}
@@ -0,0 +1,754 @@
package wgcommon
import (
"context"
"errors"
"net"
"sync"
"sync/atomic"
"testing"
"time"
"golang.zx2c4.com/wireguard/conn"
)
type reconnectReadResult struct {
packet []byte
addr net.Addr
err error
}
type reconnectTestPacketConn struct {
reads chan reconnectReadResult
readStarted chan struct{}
closed chan struct{}
closeOnce sync.Once
closes atomic.Int32
writesMu sync.Mutex
writes [][]byte
write func([]byte, net.Addr) (int, error)
deadline func(time.Time) error
local net.Addr
}
func newReconnectTestPacketConn(port int) *reconnectTestPacketConn {
return &reconnectTestPacketConn{
reads: make(chan reconnectReadResult, 8),
closed: make(chan struct{}),
local: &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: port},
}
}
func (c *reconnectTestPacketConn) ReadFrom(packet []byte) (int, net.Addr, error) {
if c.readStarted != nil {
select {
case c.readStarted <- struct{}{}:
default:
}
}
select {
case result := <-c.reads:
n := copy(packet, result.packet)
if result.addr == nil {
result.addr = &net.UDPAddr{IP: net.IPv4(192, 0, 2, 1), Port: 51820}
}
return n, result.addr, result.err
case <-c.closed:
return 0, nil, net.ErrClosed
}
}
func (c *reconnectTestPacketConn) WriteTo(packet []byte, addr net.Addr) (int, error) {
select {
case <-c.closed:
return 0, net.ErrClosed
default:
}
if c.write != nil {
n, err := c.write(packet, addr)
if err != nil {
return n, err
}
}
copyOfPacket := append([]byte(nil), packet...)
c.writesMu.Lock()
c.writes = append(c.writes, copyOfPacket)
c.writesMu.Unlock()
return len(packet), nil
}
func (c *reconnectTestPacketConn) Close() error {
c.closeOnce.Do(func() {
c.closes.Add(1)
close(c.closed)
})
return nil
}
func (c *reconnectTestPacketConn) LocalAddr() net.Addr { return c.local }
func (c *reconnectTestPacketConn) SetDeadline(time.Time) error { return nil }
func (c *reconnectTestPacketConn) SetReadDeadline(deadline time.Time) error {
if c.deadline != nil {
return c.deadline(deadline)
}
return nil
}
func (c *reconnectTestPacketConn) SetWriteDeadline(time.Time) error { return nil }
func (c *reconnectTestPacketConn) writeCount() int {
c.writesMu.Lock()
defer c.writesMu.Unlock()
return len(c.writes)
}
type reconnectFactory struct {
mu sync.Mutex
conns []net.PacketConn
errs []error
calls int
callCh chan int
}
func newReconnectFactory(conns ...net.PacketConn) *reconnectFactory {
return &reconnectFactory{conns: conns, callCh: make(chan int, 16)}
}
func (f *reconnectFactory) create(context.Context) (net.PacketConn, error) {
f.mu.Lock()
index := f.calls
f.calls++
call := f.calls
var packetConn net.PacketConn
var err error
if index < len(f.conns) {
packetConn = f.conns[index]
} else {
err = errors.New("unexpected packet connection factory call")
}
if index < len(f.errs) && f.errs[index] != nil {
err = f.errs[index]
}
f.mu.Unlock()
f.callCh <- call
return packetConn, err
}
func (f *reconnectFactory) callCount() int {
f.mu.Lock()
defer f.mu.Unlock()
return f.calls
}
func (f *reconnectFactory) awaitCall(t *testing.T, want int) {
t.Helper()
select {
case got := <-f.callCh:
if got != want {
t.Fatalf("unexpected factory call: got %d, want %d", got, want)
}
case <-time.After(time.Second):
t.Fatalf("timed out waiting for factory call %d", want)
}
}
type reconnectClock struct {
mu sync.Mutex
now time.Time
}
func (c *reconnectClock) Now() time.Time {
c.mu.Lock()
defer c.mu.Unlock()
return c.now
}
func (c *reconnectClock) Advance(elapsed time.Duration) {
c.mu.Lock()
c.now = c.now.Add(elapsed)
c.mu.Unlock()
}
func parseReconnectEndpoint(t *testing.T, bind conn.Bind) conn.Endpoint {
t.Helper()
endpoint, err := bind.ParseEndpoint("198.51.100.1:51820")
if err != nil {
t.Fatal(err)
}
return endpoint
}
func TestReconnectingBindOpenClosesConnReturnedWithError(t *testing.T) {
errFactory := errors.New("factory failed")
packetConn := newReconnectTestPacketConn(10001)
factory := newReconnectFactory(packetConn)
factory.errs = []error{errFactory}
bind := NewReconnectingNetPacketConnToWg(context.Background(), factory.create, time.Minute)
if _, _, err := bind.Open(0); !errors.Is(err, errFactory) {
t.Fatalf("Open error = %v, want %v", err, errFactory)
}
factory.awaitCall(t, 1)
if got := packetConn.closes.Load(); got != 1 {
t.Fatalf("factory connection close count = %d, want 1", got)
}
}
func TestReconnectingBindWriteErrorReplacesWithoutReplay(t *testing.T) {
errWrite := errors.New("write failed")
first := newReconnectTestPacketConn(10001)
first.write = func([]byte, net.Addr) (int, error) { return 0, errWrite }
second := newReconnectTestPacketConn(10002)
factory := newReconnectFactory(first, second)
bind := NewReconnectingNetPacketConnToWg(context.Background(), factory.create, time.Minute)
t.Cleanup(func() { _ = bind.Close() })
if _, _, err := bind.Open(0); err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 1)
endpoint := parseReconnectEndpoint(t, bind)
if err := bind.Send([][]byte{[]byte("failed")}, endpoint); !errors.Is(err, errWrite) {
t.Fatalf("Send error = %v, want %v", err, errWrite)
}
factory.awaitCall(t, 2)
if got := first.closes.Load(); got != 1 {
t.Fatalf("first connection close count = %d, want 1", got)
}
if got := second.writeCount(); got != 0 {
t.Fatalf("failed datagram was replayed on replacement: %d writes", got)
}
if err := bind.Send([][]byte{[]byte("next")}, endpoint); err != nil {
t.Fatal(err)
}
if got := second.writeCount(); got != 1 {
t.Fatalf("second connection write count = %d, want 1", got)
}
}
func TestReconnectingBindReplacementClosesConnReturnedWithError(t *testing.T) {
errWrite := errors.New("write failed")
errFactory := errors.New("factory failed")
first := newReconnectTestPacketConn(10001)
first.write = func([]byte, net.Addr) (int, error) { return 0, errWrite }
failedReplacement := newReconnectTestPacketConn(10002)
third := newReconnectTestPacketConn(10003)
factory := newReconnectFactory(first, failedReplacement, third)
factory.errs = []error{nil, errFactory}
bind := NewReconnectingNetPacketConnToWg(context.Background(), factory.create, time.Minute)
t.Cleanup(func() { _ = bind.Close() })
if _, _, err := bind.Open(0); err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 1)
endpoint := parseReconnectEndpoint(t, bind)
if err := bind.Send([][]byte{[]byte("failed")}, endpoint); !errors.Is(err, errWrite) {
t.Fatalf("Send error = %v, want %v", err, errWrite)
}
factory.awaitCall(t, 2)
if got := failedReplacement.closes.Load(); got != 1 {
t.Fatalf("failed replacement close count = %d, want 1", got)
}
if err := bind.Send([][]byte{[]byte("next")}, endpoint); err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 3)
if got := third.writeCount(); got != 1 {
t.Fatalf("third connection write count = %d, want 1", got)
}
}
func TestReconnectingBindReadErrorPhysicallyClosesAndContinues(t *testing.T) {
errRead := errors.New("read failed")
first := newReconnectTestPacketConn(10001)
second := newReconnectTestPacketConn(10002)
factory := newReconnectFactory(first, second)
bind := NewReconnectingNetPacketConnToWg(context.Background(), factory.create, time.Minute)
t.Cleanup(func() { _ = bind.Close() })
receivers, _, err := bind.Open(0)
if err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 1)
type receiveResult struct {
n int
size int
ep conn.Endpoint
err error
}
resultCh := make(chan receiveResult, 1)
go func() {
packet := make([]byte, 64)
sizes := make([]int, 1)
endpoints := make([]conn.Endpoint, 1)
n, err := receivers[0]([][]byte{packet}, sizes, endpoints)
resultCh <- receiveResult{n: n, size: sizes[0], ep: endpoints[0], err: err}
}()
first.reads <- reconnectReadResult{err: errRead}
factory.awaitCall(t, 2)
if got := first.closes.Load(); got != 1 {
t.Fatalf("first connection close count = %d, want 1", got)
}
second.reads <- reconnectReadResult{packet: []byte("reply")}
select {
case result := <-resultCh:
if result.err != nil {
t.Fatal(result.err)
}
if result.n != 1 || result.size != len("reply") || result.ep == nil {
t.Fatalf("unexpected receive result: %+v", result)
}
case <-time.After(time.Second):
t.Fatal("timed out waiting for receive after replacement")
}
}
func TestReconnectingBindReceiveRetriesFactoryFailure(t *testing.T) {
errRead := errors.New("read failed")
errFactory := errors.New("factory failed")
first := newReconnectTestPacketConn(10001)
second := newReconnectTestPacketConn(10002)
factory := newReconnectFactory(first, nil, second)
factory.errs = []error{nil, errFactory}
reconnectingBind := newReconnectingNetPacketConnToWg(context.Background(), factory.create, time.Minute, time.Now).(*reconnectingNetPacketConnToWg)
reconnectingBind.retryWait = func(context.Context, time.Duration) error { return nil }
var bind conn.Bind = reconnectingBind
t.Cleanup(func() { _ = bind.Close() })
receivers, _, err := bind.Open(0)
if err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 1)
receiveErr := make(chan error, 1)
go func() {
packet := make([]byte, 64)
sizes := make([]int, 1)
endpoints := make([]conn.Endpoint, 1)
_, err := receivers[0]([][]byte{packet}, sizes, endpoints)
receiveErr <- err
}()
first.reads <- reconnectReadResult{err: errRead}
factory.awaitCall(t, 2)
factory.awaitCall(t, 3)
second.reads <- reconnectReadResult{packet: []byte("recovered")}
select {
case err := <-receiveErr:
if err != nil {
t.Fatal(err)
}
case <-time.After(time.Second):
t.Fatal("same receive function did not recover after factory failure")
}
if got := factory.callCount(); got != 3 {
t.Fatalf("factory call count = %d, want 3", got)
}
}
func TestReconnectingBindCloseCancelsFactoryRetryBackoff(t *testing.T) {
errRead := errors.New("read failed")
errFactory := errors.New("factory failed")
first := newReconnectTestPacketConn(10001)
factory := newReconnectFactory(first, nil)
factory.errs = []error{nil, errFactory}
reconnectingBind := newReconnectingNetPacketConnToWg(context.Background(), factory.create, time.Minute, time.Now).(*reconnectingNetPacketConnToWg)
retryStarted := make(chan struct{})
reconnectingBind.retryWait = func(ctx context.Context, _ time.Duration) error {
close(retryStarted)
<-ctx.Done()
return ctx.Err()
}
var bind conn.Bind = reconnectingBind
receivers, _, err := bind.Open(0)
if err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 1)
receiveErr := make(chan error, 1)
go func() {
packet := make([]byte, 64)
sizes := make([]int, 1)
endpoints := make([]conn.Endpoint, 1)
_, err := receivers[0]([][]byte{packet}, sizes, endpoints)
receiveErr <- err
}()
first.reads <- reconnectReadResult{err: errRead}
factory.awaitCall(t, 2)
select {
case <-retryStarted:
case <-time.After(time.Second):
t.Fatal("receive function did not enter retry backoff")
}
if err := bind.Close(); err != nil {
t.Fatal(err)
}
select {
case err := <-receiveErr:
if !errors.Is(err, net.ErrClosed) {
t.Fatalf("receive error = %v, want net.ErrClosed", err)
}
case <-time.After(time.Second):
t.Fatal("Close did not cancel receive retry backoff")
}
if got := factory.callCount(); got != 2 {
t.Fatalf("closed bind retried factory: %d calls", got)
}
}
func TestReconnectingBindConcurrentFailuresCreateOneGeneration(t *testing.T) {
errWrite := errors.New("write failed")
first := newReconnectTestPacketConn(10001)
started := make(chan struct{}, 2)
release := make(chan struct{})
first.write = func([]byte, net.Addr) (int, error) {
started <- struct{}{}
<-release
return 0, errWrite
}
second := newReconnectTestPacketConn(10002)
factory := newReconnectFactory(first, second)
bind := NewReconnectingNetPacketConnToWg(context.Background(), factory.create, time.Minute)
t.Cleanup(func() { _ = bind.Close() })
if _, _, err := bind.Open(0); err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 1)
endpoint := parseReconnectEndpoint(t, bind)
errorsCh := make(chan error, 2)
for range 2 {
go func() { errorsCh <- bind.Send([][]byte{[]byte("packet")}, endpoint) }()
}
for range 2 {
select {
case <-started:
case <-time.After(time.Second):
t.Fatal("timed out waiting for concurrent writes")
}
}
close(release)
for range 2 {
if err := <-errorsCh; !errors.Is(err, errWrite) {
t.Fatalf("Send error = %v, want %v", err, errWrite)
}
}
factory.awaitCall(t, 2)
if got := factory.callCount(); got != 2 {
t.Fatalf("factory call count = %d, want 2", got)
}
if got := first.closes.Load(); got != 1 {
t.Fatalf("first connection close count = %d, want 1", got)
}
}
func TestReconnectingBindConcurrentFailuresDoNotStormAfterFactoryError(t *testing.T) {
errWrite := errors.New("write failed")
errFactory := errors.New("factory failed")
first := newReconnectTestPacketConn(10001)
started := make(chan struct{}, 2)
release := make(chan struct{})
first.write = func([]byte, net.Addr) (int, error) {
started <- struct{}{}
<-release
return 0, errWrite
}
third := newReconnectTestPacketConn(10003)
factory := newReconnectFactory(first, nil, third)
factory.errs = []error{nil, errFactory}
bind := NewReconnectingNetPacketConnToWg(context.Background(), factory.create, time.Minute)
t.Cleanup(func() { _ = bind.Close() })
if _, _, err := bind.Open(0); err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 1)
endpoint := parseReconnectEndpoint(t, bind)
errorsCh := make(chan error, 2)
for range 2 {
go func() { errorsCh <- bind.Send([][]byte{[]byte("packet")}, endpoint) }()
}
for range 2 {
select {
case <-started:
case <-time.After(time.Second):
t.Fatal("timed out waiting for concurrent writes")
}
}
close(release)
for range 2 {
if err := <-errorsCh; !errors.Is(err, errWrite) {
t.Fatalf("Send error = %v, want %v", err, errWrite)
}
}
factory.awaitCall(t, 2)
if got := factory.callCount(); got != 2 {
t.Fatalf("concurrent stale failures caused %d factory calls, want 2", got)
}
if err := bind.Send([][]byte{[]byte("recovered")}, endpoint); err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 3)
if got := third.writeCount(); got != 1 {
t.Fatalf("recovered connection write count = %d, want 1", got)
}
}
func TestReconnectingBindTimeoutIsLazyAndReceiveResetsIt(t *testing.T) {
clock := &reconnectClock{now: time.Unix(1_000, 0)}
first := newReconnectTestPacketConn(10001)
first.readStarted = make(chan struct{}, 1)
second := newReconnectTestPacketConn(10002)
third := newReconnectTestPacketConn(10003)
factory := newReconnectFactory(first, second, third)
bind := newReconnectingNetPacketConnToWg(context.Background(), factory.create, time.Minute, clock.Now)
t.Cleanup(func() { _ = bind.Close() })
receivers, _, err := bind.Open(0)
if err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 1)
endpoint := parseReconnectEndpoint(t, bind)
clock.Advance(10 * time.Minute)
if got := factory.callCount(); got != 1 {
t.Fatalf("idle bind churned connections: %d factory calls", got)
}
resultCh := make(chan error, 1)
go func() {
packet := make([]byte, 64)
sizes := make([]int, 1)
endpoints := make([]conn.Endpoint, 1)
_, err := receivers[0]([][]byte{packet}, sizes, endpoints)
resultCh <- err
}()
select {
case <-first.readStarted:
case <-time.After(time.Second):
t.Fatal("receive function did not block on the first generation")
}
if err := bind.Send([][]byte{[]byte("wake")}, endpoint); err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 2)
if got := first.closes.Load(); got != 1 {
t.Fatalf("timed-out connection close count = %d, want 1", got)
}
second.reads <- reconnectReadResult{packet: []byte("received")}
select {
case err := <-resultCh:
if err != nil {
t.Fatal(err)
}
case <-time.After(time.Second):
t.Fatal("timed out waiting for receive on replacement")
}
if got := factory.callCount(); got != 2 {
t.Fatalf("stale read error replaced the new generation: %d calls", got)
}
clock.Advance(59 * time.Second)
if err := bind.Send([][]byte{[]byte("before threshold")}, endpoint); err != nil {
t.Fatal(err)
}
if got := factory.callCount(); got != 2 {
t.Fatalf("connection replaced before threshold: %d calls", got)
}
clock.Advance(time.Second)
if err := bind.Send([][]byte{[]byte("at threshold")}, endpoint); err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 3)
if got := second.closes.Load(); got != 1 {
t.Fatalf("second connection close count = %d, want 1", got)
}
}
func TestReconnectingBindZeroTimeoutDisablesSilentReplacement(t *testing.T) {
clock := &reconnectClock{now: time.Unix(1_000, 0)}
packetConn := newReconnectTestPacketConn(10001)
factory := newReconnectFactory(packetConn)
bind := newReconnectingNetPacketConnToWg(context.Background(), factory.create, 0, clock.Now)
t.Cleanup(func() { _ = bind.Close() })
if _, _, err := bind.Open(0); err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 1)
clock.Advance(24 * time.Hour)
if err := bind.Send([][]byte{[]byte("still live")}, parseReconnectEndpoint(t, bind)); err != nil {
t.Fatal(err)
}
if got := factory.callCount(); got != 1 {
t.Fatalf("zero timeout created %d connections, want 1", got)
}
}
func TestReconnectingBindCloseStopsOldReceiversAndAllowsReopen(t *testing.T) {
first := newReconnectTestPacketConn(10001)
second := newReconnectTestPacketConn(10002)
factory := newReconnectFactory(first, second)
bind := NewReconnectingNetPacketConnToWg(context.Background(), factory.create, time.Minute)
oldReceivers, _, err := bind.Open(0)
if err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 1)
receiveErr := make(chan error, 1)
go func() {
packet := make([]byte, 64)
sizes := make([]int, 1)
endpoints := make([]conn.Endpoint, 1)
_, err := oldReceivers[0]([][]byte{packet}, sizes, endpoints)
receiveErr <- err
}()
if err := bind.Close(); err != nil {
t.Fatal(err)
}
select {
case err := <-receiveErr:
if !errors.Is(err, net.ErrClosed) {
t.Fatalf("receive error = %v, want net.ErrClosed", err)
}
case <-time.After(time.Second):
t.Fatal("Close did not unblock receive")
}
if err := bind.Send([][]byte{[]byte("closed")}, parseReconnectEndpoint(t, bind)); !errors.Is(err, net.ErrClosed) {
t.Fatalf("Send after Close error = %v, want net.ErrClosed", err)
}
if got := factory.callCount(); got != 1 {
t.Fatalf("closed bind recreated a connection: %d calls", got)
}
newReceivers, _, err := bind.Open(0)
if err != nil {
t.Fatal(err)
}
factory.awaitCall(t, 2)
defer func() { _ = bind.Close() }()
if len(newReceivers) != 1 {
t.Fatalf("new receive function count = %d, want 1", len(newReceivers))
}
packet := make([]byte, 64)
sizes := make([]int, 1)
endpoints := make([]conn.Endpoint, 1)
if _, err := oldReceivers[0]([][]byte{packet}, sizes, endpoints); !errors.Is(err, net.ErrClosed) {
t.Fatalf("old receiver after reopen error = %v, want net.ErrClosed", err)
}
}
func TestReconnectingBindCloseCancelsInFlightReplacement(t *testing.T) {
errWrite := errors.New("write failed")
first := newReconnectTestPacketConn(10001)
first.write = func([]byte, net.Addr) (int, error) { return 0, errWrite }
late := newReconnectTestPacketConn(10002)
third := newReconnectTestPacketConn(10003)
replacementStarted := make(chan struct{})
var calls atomic.Int32
factory := func(ctx context.Context) (net.PacketConn, error) {
switch calls.Add(1) {
case 1:
return first, nil
case 2:
close(replacementStarted)
<-ctx.Done()
return late, nil
case 3:
return third, nil
default:
return nil, errors.New("unexpected factory call")
}
}
bind := NewReconnectingNetPacketConnToWg(context.Background(), factory, time.Minute)
if _, _, err := bind.Open(0); err != nil {
t.Fatal(err)
}
endpoint := parseReconnectEndpoint(t, bind)
sendDone := make(chan error, 1)
go func() { sendDone <- bind.Send([][]byte{[]byte("failed")}, endpoint) }()
select {
case <-replacementStarted:
case <-time.After(time.Second):
t.Fatal("replacement factory did not start")
}
if err := bind.Close(); err != nil {
t.Fatal(err)
}
select {
case err := <-sendDone:
if !errors.Is(err, errWrite) {
t.Fatalf("Send error = %v, want %v", err, errWrite)
}
case <-time.After(time.Second):
t.Fatal("in-flight replacement did not stop after Close")
}
if got := late.closes.Load(); got != 1 {
t.Fatalf("late factory connection close count = %d, want 1", got)
}
if got := calls.Load(); got != 2 {
t.Fatalf("factory call count after Close = %d, want 2", got)
}
if _, _, err := bind.Open(0); err != nil {
t.Fatal(err)
}
defer func() { _ = bind.Close() }()
if err := bind.Send([][]byte{[]byte("reopened")}, endpoint); err != nil {
t.Fatal(err)
}
if got := third.writeCount(); got != 1 {
t.Fatalf("reopened connection write count = %d, want 1", got)
}
}
func TestStaticBindCloseMapsReadWakeupToNetErrClosed(t *testing.T) {
packetConn := newReconnectTestPacketConn(10001)
packetConn.deadline = func(deadline time.Time) error {
if !deadline.IsZero() {
packetConn.reads <- reconnectReadResult{err: errors.New("deadline wakeup")}
}
return nil
}
bind := NewNetPacketConnToWg(packetConn)
receivers, _, err := bind.Open(0)
if err != nil {
t.Fatal(err)
}
receiveErr := make(chan error, 1)
go func() {
packet := make([]byte, 64)
sizes := make([]int, 1)
endpoints := make([]conn.Endpoint, 1)
_, err := receivers[0]([][]byte{packet}, sizes, endpoints)
receiveErr <- err
}()
if err := bind.Close(); err != nil {
t.Fatal(err)
}
select {
case err := <-receiveErr:
if !errors.Is(err, net.ErrClosed) {
t.Fatalf("receive error = %v, want net.ErrClosed", err)
}
case <-time.After(time.Second):
t.Fatal("static bind Close did not unblock receive")
}
}
+31 -17
View File
@@ -2,7 +2,9 @@ package wgcommon
import (
"context"
"sync"
"golang.zx2c4.com/wireguard/conn"
"golang.zx2c4.com/wireguard/device"
"github.com/v2fly/v2ray-core/v5/common/net"
@@ -21,8 +23,10 @@ type WrappedWireguardDevice struct {
ctx context.Context
device *device.Device
tunnel packetswitch.NetworkLayerDevice
conn net.PacketConn
tunnel packetswitch.NetworkLayerDevice
conn net.PacketConn
bind conn.Bind
closeOnce sync.Once
}
func (w *WrappedWireguardDevice) Up() error {
@@ -40,27 +44,37 @@ func (w *WrappedWireguardDevice) SetTunnel(t packetswitch.NetworkLayerDevice) {
// SetConn sets the underlying packet connection used by the wrapped WireGuard device.
func (w *WrappedWireguardDevice) SetConn(c net.PacketConn) {
w.conn = c
w.bind = nil
}
// SetBind sets an explicit WireGuard bind. It is useful for binds that manage
// multiple PacketConn generations over the device lifetime.
func (w *WrappedWireguardDevice) SetBind(bind conn.Bind) {
w.bind = bind
w.conn = nil
}
func (w *WrappedWireguardDevice) Close() error {
if w == nil {
return nil
}
// Bring device down if initialized
if w.device != nil {
_ = w.device.Down()
w.device = nil
}
// Close tunnel if present
if w.tunnel != nil {
_ = w.tunnel.Close()
w.tunnel = nil
}
// Close underlying packet conn if present
if w.conn != nil {
_ = w.conn.Close()
w.conn = nil
}
w.closeOnce.Do(func() {
// Close the device rather than only bringing it down. Down stops peers
// and the bind, while Close also terminates WireGuard's long-lived worker
// pools and closes the TUN adaptor.
if w.device != nil {
w.device.Close()
}
if w.tunnel != nil {
_ = w.tunnel.Close()
}
if w.conn != nil {
_ = w.conn.Close()
}
if w.bind != nil {
_ = w.bind.Close()
}
})
return nil
}