From f542e75e711cec0dc26cf94566079fa760bdd9e8 Mon Sep 17 00:00:00 2001 From: nerpa Date: Sat, 10 Oct 2026 16:26:33 +0300 Subject: [PATCH] feat(amnezia): add AmneziaWG outbound/inbound Adds the "amnezia" inbound/outbound based on amneziawg-go v0.2.19. --- go.mod | 5 +- go.sum | 8 +- infra/conf/amnezia.go | 162 ++++++ infra/conf/amnezia_params.go | 47 ++ proxy/amnezia/amnezia.go | 18 + proxy/amnezia/bind.go | 161 ++++++ proxy/amnezia/client.go | 513 ++++++++++++++++++ proxy/amnezia/config.go | 60 +++ proxy/amnezia/config.pb.go | 433 +++++++++++++++ proxy/amnezia/config.proto | 48 ++ proxy/amnezia/icmp.go | 91 ++++ proxy/amnezia/icmp_test.go | 176 +++++++ proxy/amnezia/netstack.go | 838 ++++++++++++++++++++++++++++++ proxy/amnezia/params.go | 37 ++ proxy/amnezia/queue_bench_test.go | 80 +++ proxy/amnezia/server.go | 401 ++++++++++++++ proxy/amnezia/tun.go | 381 ++++++++++++++ proxy/amnezia/tun_default.go | 18 + proxy/amnezia/tun_linux.go | 278 ++++++++++ proxy/amnezia/tun_test.go | 91 ++++ 20 files changed, 3842 insertions(+), 4 deletions(-) create mode 100644 infra/conf/amnezia.go create mode 100644 infra/conf/amnezia_params.go create mode 100644 proxy/amnezia/amnezia.go create mode 100644 proxy/amnezia/bind.go create mode 100644 proxy/amnezia/client.go create mode 100644 proxy/amnezia/config.go create mode 100644 proxy/amnezia/config.pb.go create mode 100644 proxy/amnezia/config.proto create mode 100644 proxy/amnezia/icmp.go create mode 100644 proxy/amnezia/icmp_test.go create mode 100644 proxy/amnezia/netstack.go create mode 100644 proxy/amnezia/params.go create mode 100644 proxy/amnezia/queue_bench_test.go create mode 100644 proxy/amnezia/server.go create mode 100644 proxy/amnezia/tun.go create mode 100644 proxy/amnezia/tun_default.go create mode 100644 proxy/amnezia/tun_linux.go create mode 100644 proxy/amnezia/tun_test.go diff --git a/go.mod b/go.mod index b9abcee3..971a1e4a 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module github.com/xtls/xray-core go 1.27 require ( + github.com/amnezia-vpn/amneziawg-go v0.2.19 github.com/apernet/quic-go v0.61.1-0.20260806010916-184d081eef3e github.com/cloudflare/circl v1.6.5 github.com/ghodss/yaml v1.0.1-0.20220118164431-d8423dcdf344 @@ -24,6 +25,7 @@ require ( github.com/yuin/gopher-lua v1.1.2 go4.org/netipx v0.0.0-20231129151722-fdeea329fbba golang.org/x/crypto v0.57.0 + golang.org/x/crypto/x509roots/fallback v0.0.0-20261005185213-c3db4df58582 golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 golang.org/x/net v0.59.0 golang.org/x/sync v0.23.0 @@ -42,7 +44,7 @@ require ( require ( github.com/andybalholm/brotli v1.0.6 // indirect - github.com/google/btree v1.1.2 // indirect + github.com/google/btree v1.1.3 // indirect github.com/google/gopacket v1.1.19 // indirect github.com/huin/goupnp v1.2.0 // indirect github.com/jackpal/go-nat-pmp v1.0.2 // indirect @@ -57,7 +59,6 @@ require ( github.com/vishvananda/netns v0.0.5 // indirect github.com/wlynxg/anet v0.0.5 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect - golang.org/x/crypto/x509roots/fallback v0.0.0-20261005185213-c3db4df58582 // indirect golang.org/x/text v0.42.0 // indirect golang.org/x/time v0.14.0 // indirect golang.org/x/tools v0.49.0 // indirect diff --git a/go.sum b/go.sum index c7f69432..b8723dfc 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/amnezia-vpn/amneziawg-go v0.2.19 h1:l3rOmrA4o5z38kpgnA5iSk1yOm7Cv3AafIi4vxpSEV0= +github.com/amnezia-vpn/amneziawg-go v0.2.19/go.mod h1:aMgOk9MuX0xI7b5TKAYp8pLM54RlXcOPzDvYw3YEO5A= github.com/andybalholm/brotli v1.0.6 h1:Yf9fFpf49Zrxb9NlQaluyE92/+X7UVHlhMNJN2sxfOI= github.com/andybalholm/brotli v1.0.6/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig= github.com/apernet/quic-go v0.61.1-0.20260806010916-184d081eef3e h1:5mgtR5gwIgBKMiGI1QdXldZZ+SNor06Nbu1wCBulQBg= @@ -15,8 +17,8 @@ github.com/golang/mock v1.7.0-rc.1 h1:YojYx61/OLFsiv6Rw1Z96LpldJIy31o+UHmwAUMJ6/ github.com/golang/mock v1.7.0-rc.1/go.mod h1:s42URUywIqd+OcERslBJvOjepvNymP31m3q8d/GkuRs= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= -github.com/google/btree v1.1.2 h1:xf4v41cLI2Z6FxbKm+8Bu+m8ifhj15JuZ9sa0jZCMUU= -github.com/google/btree v1.1.2/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4= +github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg= +github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8= @@ -87,6 +89,8 @@ github.com/yuin/goldmark v1.4.1/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1 github.com/yuin/gopher-lua v0.0.0-20190206043414-8bfc7677f583/go.mod h1:gqRgreBUhTSL0GeU64rtZ3Uq3wtjOa/TB2YfrtkCbVQ= github.com/yuin/gopher-lua v1.1.2 h1:yF/FjE3hD65tBbt0VXLE13HWS9h34fdzJmrWRXwobGA= github.com/yuin/gopher-lua v1.1.2/go.mod h1:7aRmXIWl37SqRf0koeyylBEzJ+aPt8A+mmkQ4f1ntR8= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko= go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o= go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= diff --git a/infra/conf/amnezia.go b/infra/conf/amnezia.go new file mode 100644 index 00000000..b5d89f0b --- /dev/null +++ b/infra/conf/amnezia.go @@ -0,0 +1,162 @@ +package conf + +import ( + "encoding/base64" + "encoding/hex" + "strconv" + "strings" + + "github.com/xtls/xray-core/common/errors" + "github.com/xtls/xray-core/common/protocol" + "github.com/xtls/xray-core/common/serial" + "github.com/xtls/xray-core/common/task" + "github.com/xtls/xray-core/proxy/amnezia" + "google.golang.org/protobuf/proto" +) + +type AmneziaPeerConfig struct { + PublicKey string `json:"publicKey"` + PreSharedKey string `json:"preSharedKey"` + Endpoint string `json:"endpoint"` + KeepAlive uint32 `json:"keepAlive"` + AllowedIPs []string `json:"allowedIPs,omitempty"` + + Level uint32 `json:"level"` + Email string `json:"email"` +} + +func (c *AmneziaPeerConfig) Build() (*amnezia.PeerConfig, error) { + var err error + config := new(amnezia.PeerConfig) + + if c.PublicKey != "" { + config.PublicKey, err = ParseAmneziaKey(c.PublicKey) + if err != nil { + return nil, err + } + } + + if c.PreSharedKey != "" { + config.PreSharedKey, err = ParseAmneziaKey(c.PreSharedKey) + if err != nil { + return nil, err + } + } + + config.Endpoint = c.Endpoint + if c.KeepAlive != 0 { + config.KeepAlive = strconv.FormatUint(uint64(c.KeepAlive), 10) + } + if c.AllowedIPs == nil { + config.AllowedIps = []string{"0.0.0.0/0", "::0/0"} + } else { + config.AllowedIps = c.AllowedIPs + } + + return config, nil +} + +type AmneziaConfig struct { + AmneziaParams + IsClient bool `json:""` + + NoKernelTun bool `json:"noKernelTun"` + SecretKey string `json:"secretKey"` + Address []string `json:"address"` + Peers []*AmneziaPeerConfig `json:"peers"` + MTU int32 `json:"mtu"` + Reserved []byte `json:"reserved"` + DNS []string `json:"remoteDNS"` +} + +func (c *AmneziaConfig) Build() (proto.Message, error) { + config := new(amnezia.DeviceConfig) + + var err error + config.SecretKey, err = ParseAmneziaKey(c.SecretKey) + if err != nil { + return nil, errors.New("invalid Amnezia secret key: %w", err) + } + + if c.Address == nil { + // bogon ips + config.Endpoint = []string{"10.0.0.1", "fd59:7153:2388:b5fd:0000:0000:0000:0001"} + } else { + config.Endpoint = c.Address + } + + if c.IsClient { + config.Peers = make([]*amnezia.PeerConfig, len(c.Peers)) + for i, p := range c.Peers { + msg, err := p.Build() + if err != nil { + return nil, err + } + config.Peers[i] = msg + } + } else { + config.Users = make([]*protocol.User, len(c.Peers)) + processUser := func(idx int) error { + p := c.Peers[idx] + m, err := p.Build() + if err != nil { + return err + } + config.Users[idx] = &protocol.User{ + Email: p.Email, + Level: p.Level, + Account: serial.ToTypedMessage(m), + } + return nil + } + if err := task.ParallelForN(len(c.Peers), processUser); err != nil { + return nil, err + } + } + + if c.MTU == 0 { + config.Mtu = 1420 + } else { + config.Mtu = c.MTU + } + + if len(c.Reserved) != 0 && len(c.Reserved) != 3 { + return nil, errors.New(`"reserved" should be empty or 3 bytes`) + } + config.Reserved = c.Reserved + + config.IsClient = c.IsClient + config.NoKernelTun = c.NoKernelTun + config.DNS = c.DNS + c.AmneziaParams.apply(config) + + return config, nil +} + +func ParseAmneziaKey(str string) (string, error) { + var err error + + if str == "" { + return "", errors.New("key must not be empty") + } + + if len(str) == 64 { + _, err = hex.DecodeString(str) + if err == nil { + return str, nil + } + } + + var dat []byte + str = strings.TrimSuffix(str, "=") + if strings.ContainsRune(str, '+') || strings.ContainsRune(str, '/') { + dat, err = base64.RawStdEncoding.DecodeString(str) + } else { + dat, err = base64.RawURLEncoding.DecodeString(str) + } + if err == nil { + return hex.EncodeToString(dat), nil + } + + return "", errors.New("failed to deserialize key").Base(err) +} diff --git a/infra/conf/amnezia_params.go b/infra/conf/amnezia_params.go new file mode 100644 index 00000000..3888be24 --- /dev/null +++ b/infra/conf/amnezia_params.go @@ -0,0 +1,47 @@ +package conf + +import "github.com/xtls/xray-core/proxy/amnezia" + +type AmneziaParams struct { + JunkCount int32 `json:"jc"` + JunkMin int32 `json:"jmin"` + JunkMax int32 `json:"jmax"` + + InitPadding int32 `json:"s1"` + ResponsePadding int32 `json:"s2"` + CookiePadding int32 `json:"s3"` + TransportPadding int32 `json:"s4"` + + InitHeader string `json:"h1"` + ResponseHeader string `json:"h2"` + CookieHeader string `json:"h3"` + TransportHeader string `json:"h4"` + + Signature1 string `json:"i1"` + Signature2 string `json:"i2"` + Signature3 string `json:"i3"` + Signature4 string `json:"i4"` + Signature5 string `json:"i5"` +} + +func (p *AmneziaParams) apply(config *amnezia.DeviceConfig) { + config.Jc = p.JunkCount + config.JMin = p.JunkMin + config.JMax = p.JunkMax + + config.S1 = p.InitPadding + config.S2 = p.ResponsePadding + config.S3 = p.CookiePadding + config.S4 = p.TransportPadding + + config.H1 = p.InitHeader + config.H2 = p.ResponseHeader + config.H3 = p.CookieHeader + config.H4 = p.TransportHeader + + config.I1 = p.Signature1 + config.I2 = p.Signature2 + config.I3 = p.Signature3 + config.I4 = p.Signature4 + config.I5 = p.Signature5 +} diff --git a/proxy/amnezia/amnezia.go b/proxy/amnezia/amnezia.go new file mode 100644 index 00000000..8a522292 --- /dev/null +++ b/proxy/amnezia/amnezia.go @@ -0,0 +1,18 @@ +package amnezia + +import ( + "context" + + "github.com/xtls/xray-core/common" +) + +func init() { + common.Must(common.RegisterConfig((*DeviceConfig)(nil), func(ctx context.Context, config interface{}) (interface{}, error) { + deviceConfig := config.(*DeviceConfig) + if deviceConfig.IsClient { + return NewClient(ctx, deviceConfig) + } else { + return NewServer(ctx, deviceConfig) + } + })) +} diff --git a/proxy/amnezia/bind.go b/proxy/amnezia/bind.go new file mode 100644 index 00000000..773a0e47 --- /dev/null +++ b/proxy/amnezia/bind.go @@ -0,0 +1,161 @@ +package amnezia + +import ( + "context" + goerrors "errors" + "io" + "net" + "net/netip" + "strconv" + "sync" + "syscall" + + "github.com/amnezia-vpn/amneziawg-go/conn" + "github.com/xtls/xray-core/common" + "github.com/xtls/xray-core/common/errors" +) + +type bind struct { + resolveFunc func(host string) (net.IP, error) + listenFunc func() (net.PacketConn, error) + downFunc func() error + reserved []byte + + net.PacketConn + closeCh chan struct{} + mu sync.Mutex +} + +func (b *bind) Open(port uint16) (fns []conn.ReceiveFunc, actualPort uint16, err error) { + b.mu.Lock() + defer b.mu.Unlock() + + if b.PacketConn != nil { + return nil, 0, conn.ErrBindAlreadyOpen + } + + c, err := b.listenFunc() + if err != nil { + return nil, 0, err + } + b.PacketConn = c + ch := make(chan struct{}) + b.closeCh = ch + + return []conn.ReceiveFunc{ + func(bufs [][]byte, sizes []int, eps []conn.Endpoint) (n int, err error) { + for { + n, addr, err := c.ReadFrom(bufs[0]) + if err != nil { + if goerrors.Is(err, io.EOF) || goerrors.Is(err, io.ErrClosedPipe) || goerrors.Is(err, net.ErrClosed) { + select { + case <-ch: + default: + errors.LogErrorInner(context.Background(), err, "unexpected closed") + b.mu.Lock() + downFunc := b.downFunc + b.mu.Unlock() + if downFunc != nil { + go func() { + common.Must(downFunc()) + }() + } + } + return 0, net.ErrClosed + } + errors.LogErrorInner(context.Background(), err, "bind recv err") + continue + } + if n > 3 { + bufs[0][1] = 0 + bufs[0][2] = 0 + bufs[0][3] = 0 + } + sizes[0] = n + eps[0] = &conn.StdNetEndpoint{AddrPort: addr.(*net.UDPAddr).AddrPort()} + return 1, nil + } + }, + }, uint16(c.LocalAddr().(*net.UDPAddr).Port), nil +} + +// setDownFunc sets downFunc after the device is created, since the device may already be using the bind. +func (b *bind) setDownFunc(f func() error) { + b.mu.Lock() + defer b.mu.Unlock() + b.downFunc = f +} + +func (b *bind) Close() error { + b.mu.Lock() + defer b.mu.Unlock() + if b.PacketConn != nil { + close(b.closeCh) + _ = b.PacketConn.Close() + b.PacketConn = nil + } + return nil +} + +func (b *bind) SetMark(mark uint32) error { + return nil +} + +func (b *bind) Send(bufs [][]byte, ep conn.Endpoint) (err error) { + b.mu.Lock() + c := b.PacketConn + b.mu.Unlock() + + if c == nil { + return syscall.EAFNOSUPPORT + } + + for i := range bufs { + if len(bufs[i]) > 3 && len(b.reserved) == 3 { + bufs[i][1] = b.reserved[0] + bufs[i][2] = b.reserved[1] + bufs[i][3] = b.reserved[2] + } + _, err = c.WriteTo(bufs[i], net.UDPAddrFromAddrPort(ep.(*conn.StdNetEndpoint).AddrPort)) + if err != nil { + errors.LogErrorInner(context.Background(), err, "bind send err") + break + } + } + return err +} + +func (b *bind) ParseEndpoint(s string) (conn.Endpoint, error) { + if b.resolveFunc == nil { + e, err := netip.ParseAddrPort(s) + if err != nil { + return nil, err + } + return &conn.StdNetEndpoint{ + AddrPort: e, + }, nil + } + host, sport, err := net.SplitHostPort(s) + if err != nil { + return nil, err + } + port, err := strconv.Atoi(sport) + if err != nil { + return nil, err + } + if port < 0 || port > 65535 { + return nil, errors.New("invalid port " + sport) + } + ip, err := b.resolveFunc(host) + if err != nil { + return nil, err + } + addr, _ := netip.AddrFromSlice(ip) + return &conn.StdNetEndpoint{ + AddrPort: netip.AddrPortFrom(addr, uint16(port)), + }, nil +} + +func (b *bind) BatchSize() int { + return 1 +} diff --git a/proxy/amnezia/client.go b/proxy/amnezia/client.go new file mode 100644 index 00000000..a51ed6fc --- /dev/null +++ b/proxy/amnezia/client.go @@ -0,0 +1,513 @@ +package amnezia + +import ( + "context" + "fmt" + "net/netip" + "reflect" + "strings" + "sync" + "time" + + "github.com/amnezia-vpn/amneziawg-go/tun" + + "github.com/amnezia-vpn/amneziawg-go/device" + "github.com/xtls/xray-core/common" + "github.com/xtls/xray-core/common/buf" + "github.com/xtls/xray-core/common/dice" + "github.com/xtls/xray-core/common/errors" + "github.com/xtls/xray-core/common/log" + "github.com/xtls/xray-core/common/net" + "github.com/xtls/xray-core/common/net/cnc" + "github.com/xtls/xray-core/common/session" + "github.com/xtls/xray-core/common/signal" + "github.com/xtls/xray-core/common/task" + "github.com/xtls/xray-core/core" + "github.com/xtls/xray-core/features/dns" + "github.com/xtls/xray-core/features/policy" + "github.com/xtls/xray-core/features/stats" + "github.com/xtls/xray-core/transport" + "github.com/xtls/xray-core/transport/internet" +) + +type Handler struct { + conf *DeviceConfig + policyManager policy.Manager + dns dns.Client + + streamSettings *internet.MemoryStreamConfig + uplinkCounter stats.Counter + downlinkCounter stats.Counter + + tun tun.Device + tnet *Net + dev *device.Device + mu sync.Mutex +} + +func NewClient(ctx context.Context, conf *DeviceConfig) (*Handler, error) { + v := core.MustFromContext(ctx) + p := v.GetFeature(policy.ManagerType()).(policy.Manager) + d := v.GetFeature(dns.ClientType()).(dns.Client) + + streamSettings := session.StreamSettingsFromContext(ctx).(*internet.MemoryStreamConfig) + tag := session.FullHandlerFromContext(ctx).Tag() + var uplinkCounter stats.Counter + var downlinkCounter stats.Counter + if len(tag) > 0 && p.ForSystem().Stats.OutboundUplink { + statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager) + name := "outbound>>>" + tag + ">>>traffic>>>uplink" + c, _ := statsManager.GetOrRegisterCounter(name) + if c != nil { + uplinkCounter = c + } + } + if len(tag) > 0 && p.ForSystem().Stats.OutboundDownlink { + statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager) + name := "outbound>>>" + tag + ">>>traffic>>>downlink" + c, _ := statsManager.GetOrRegisterCounter(name) + if c != nil { + downlinkCounter = c + } + } + + if len(conf.Peers) == 0 { + return nil, errors.New("empty peers") + } + for _, peer := range conf.Peers { + if peer.PublicKey == "" { + return nil, errors.New("peer without publickey") + } + if peer.Endpoint == "" { + return nil, errors.New("peer without endpoint") + } + } + + localAddresses := make([]netip.Addr, 0, len(conf.Endpoint)) + for _, localaddress := range conf.Endpoint { + addr, err := netip.ParseAddr(localaddress) + if err == nil { + localAddresses = append(localAddresses, addr) + continue + } + prefix, err := netip.ParsePrefix(localaddress) + if err == nil { + localAddresses = append(localAddresses, prefix.Addr()) + continue + } + return nil, err + } + + dns := conf.DNS + if len(dns) == 0 { + dns = []string{"1.1.1.1", "1.0.0.1", "2606:4700:4700::1111", "2606:4700:4700::1001"} + } + dnses := make([]netip.Addr, 0, len(dns)) + for _, dns := range dns { + dnses = append(dnses, netip.MustParseAddr(dns)) + } + + kernelTunSupported, err := KernelTunSupported() + if err != nil { + errors.LogWarningInner(context.Background(), err, "Failed to check kernel TUN support") + } + var tun tun.Device + var tnet *Net + if !conf.NoKernelTun && kernelTunSupported { + errors.LogWarning(context.Background(), "Using kernel TUN") + tun, tnet, err = createKernelTun(localAddresses, dnses, int(conf.Mtu)) + } else { + errors.LogWarning(context.Background(), "Using gVisor TUN") + tun, tnet, _, err = CreateNetTUN(localAddresses, dnses, int(conf.Mtu), true) + } + if err != nil { + return nil, err + } + + return &Handler{ + conf: conf, + policyManager: p, + dns: d, + + streamSettings: streamSettings, + uplinkCounter: uplinkCounter, + downlinkCounter: downlinkCounter, + + tun: tun, + tnet: tnet, + }, nil +} + +// Process implements proxy.Outbound.Process. +func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer internet.Dialer) error { + outbounds := session.OutboundsFromContext(ctx) + ob := outbounds[len(outbounds)-1] + if !ob.Target.IsValid() { + return errors.New("target not specified") + } + ob.Name = "amnezia" + ob.CanSpliceCopy = 3 + dialer.SetOutboundGateway(ctx, ob) + + if err := h.init(ctx); err != nil { + return err + } + + var newCtx context.Context + var newCancel context.CancelFunc + if session.TimeoutOnlyFromContext(ctx) { + newCtx, newCancel = context.WithCancel(context.Background()) + } + + sessionPolicy := h.policyManager.ForLevel(0) + ctx, cancel := context.WithCancel(ctx) + timer := signal.CancelAfterInactivity(ctx, func() { + cancel() + if newCancel != nil { + newCancel() + } + }, sessionPolicy.Timeouts.ConnectionIdle) + + if newCtx != nil { + ctx = newCtx + } + + var reader buf.Reader + var writer buf.Writer + + switch ob.Target.Network { + case net.Network_TCP: + var conn net.Conn + var err error + if sessionPolicy.Timeouts.Handshake != 0 { + timeoutCtx, timeoutCancel := context.WithTimeout(ctx, sessionPolicy.Timeouts.Handshake) + conn, err = h.tnet.DialContext(timeoutCtx, "tcp", ob.Target.NetAddr()) + timeoutCancel() + } else { + conn, err = h.tnet.Dial("tcp", ob.Target.NetAddr()) + } + if err != nil { + return errors.New("failed to create TCP connection").Base(err) + } + defer conn.Close() + reader = buf.NewReader(conn) + writer = buf.NewWriter(conn) + case net.Network_UDP: + conn, err := h.tnet.Dial("udp", ob.Target.NetAddr()) + if err != nil { + return errors.New("failed to create UDP connection").Base(err) + } + defer conn.Close() + c := &UDPConnClient{ + PacketConn: conn.(*net.PacketConnWrapper).PacketConn, + Dest: conn.RemoteAddr().(*net.UDPAddr), + } + reader = c + writer = c + default: + panic(ob.Target.Network) + } + + requestFunc := func() error { + defer timer.SetTimeout(sessionPolicy.Timeouts.DownlinkOnly) + return buf.Copy(link.Reader, writer, buf.UpdateActivity(timer)) + } + + responseFunc := func() error { + defer timer.SetTimeout(sessionPolicy.Timeouts.UplinkOnly) + return buf.Copy(reader, link.Writer, buf.UpdateActivity(timer)) + } + + responseDonePost := task.OnSuccess(responseFunc, task.Close(link.Writer)) + if err := task.Run(ctx, requestFunc, responseDonePost); err != nil { + common.Interrupt(link.Reader) + common.Interrupt(link.Writer) + return errors.New("connection ends").Base(err) + } + + return nil +} + +func (h *Handler) Close() (err error) { + h.mu.Lock() + defer h.mu.Unlock() + if h.dev != nil { + h.dev.Close() + h.dev = nil + h.tun = nil + } else if h.tun != nil { + h.tun.Close() + h.tun = nil + } + return nil +} + +func (h *Handler) init(ctx context.Context) error { + h.mu.Lock() + defer h.mu.Unlock() + if h.tun == nil { + return errors.New("closed") + } + if h.dev != nil { + return h.dev.Up() + } + resolveFunc := h.resolveLocal + listenFunc := func() (net.PacketConn, error) { + dest, err := net.ParseDestination("udp:" + h.conf.Peers[0].Endpoint) + if err != nil { + return nil, err + } + var pktConn net.PacketConn + if h.streamSettings.FinalMask != nil { + conn, err := h.streamSettings.FinalMask.DialUDP(ctx, dest) + if err != nil { + return nil, errors.New("failed to dial to dest").Base(err) + } + pktConn = conn.(*net.PacketConnWrapper).PacketConn + } else { + conn, err := internet.DialSystem(ctx, dest, h.streamSettings.SocketSettings) + if err != nil { + return nil, errors.New("failed to dial to dest").Base(err) + } + switch c := conn.(type) { + case *net.PacketConnWrapper: + pktConn = c.PacketConn + case *cnc.Connection: + pktConn = &internet.FakePacketConn{Conn: c} + default: + panic(reflect.TypeOf(c)) + } + } + if h.uplinkCounter != nil || h.downlinkCounter != nil { + pktConn = &PacketCounterConnection{ + PacketConn: pktConn, + ReadCounter: h.downlinkCounter, + WriteCounter: h.uplinkCounter, + } + } + return pktConn, nil + } + // device.NewDevice may use the bind right away (Up -> BindUpdate -> Open), + // so everything it reads must be set before creating the device. + bind := &bind{ + resolveFunc: resolveFunc, + listenFunc: listenFunc, + reserved: h.conf.Reserved, + } + logger := &device.Logger{ + Verbosef: func(format string, args ...any) { + log.Record(&log.GeneralMessage{ + Severity: log.Severity_Debug, + Content: fmt.Sprintf(format, args...), + }) + }, + Errorf: func(format string, args ...any) { + log.Record(&log.GeneralMessage{ + Severity: log.Severity_Error, + Content: fmt.Sprintf(format, args...), + }) + }, + } + dev := device.NewDevice(h.tun, bind, logger) + bind.setDownFunc(dev.Down) + var cfg strings.Builder + cfg.WriteString("private_key=" + h.conf.SecretKey + "\n") + cfg.WriteString(h.conf.awgUAPI()) + for _, peer := range h.conf.Peers { + cfg.WriteString("public_key=" + peer.PublicKey + "\n") + if peer.PreSharedKey != "" { + cfg.WriteString("preshared_key=" + peer.PreSharedKey + "\n") + } + cfg.WriteString("endpoint=" + peer.Endpoint + "\n") + for _, ip := range peer.AllowedIps { + cfg.WriteString("allowed_ip=" + ip + "\n") + } + if peer.KeepAlive != "" { + cfg.WriteString("persistent_keepalive_interval=" + peer.KeepAlive + "\n") + } + } + err := dev.IpcSet(cfg.String()) + if err != nil { + return err + } + err = dev.Up() + if err != nil { + return err + } + h.dev = dev + return nil +} + +func (h *Handler) resolveLocal(host string) (net.IP, error) { + if ip := net.ParseIP(host); ip != nil { + return ip, nil + } + ips, _, err := h.dns.LookupIP(host, dns.IPOption{IPv4Enable: true, IPv6Enable: true}) + if err != nil { + return nil, err + } + got := ips + if h.streamSettings.SocketSettings != nil { + var got4, got6 []net.IP + for _, ip := range ips { + if ip.To4() != nil { + got4 = append(got4, ip) + } else { + got6 = append(got6, ip) + } + } + switch h.streamSettings.SocketSettings.DomainStrategy { + case internet.DomainStrategy_AS_IS, internet.DomainStrategy_USE_IP, internet.DomainStrategy_FORCE_IP: + got = ips + case internet.DomainStrategy_USE_IP4, internet.DomainStrategy_FORCE_IP4: + got = got4 + case internet.DomainStrategy_USE_IP6, internet.DomainStrategy_FORCE_IP6: + got = got6 + case internet.DomainStrategy_USE_IP46, internet.DomainStrategy_FORCE_IP46: + got = got4 + if len(got) == 0 { + got = got6 + } + case internet.DomainStrategy_USE_IP64, internet.DomainStrategy_FORCE_IP64: + got = got6 + if len(got) == 0 { + got = got4 + } + } + if len(got) == 0 { + return nil, dns.ErrEmptyResponse + } + } + return got[dice.Roll(len(got))], nil +} + +type UDPConnClient struct { + net.PacketConn + Dest *net.UDPAddr +} + +func (c *UDPConnClient) ReadMultiBuffer() (buf.MultiBuffer, error) { + b := buf.New() + b.Resize(0, buf.Size) + n, addr, err := c.PacketConn.ReadFrom(b.Bytes()) + if err != nil { + b.Release() + return nil, err + } + b.Resize(0, int32(n)) + + b.UDP = &net.Destination{ + Address: net.IPAddress(addr.(*net.UDPAddr).IP), + Port: net.Port(addr.(*net.UDPAddr).Port), + Network: net.Network_UDP, + } + + return buf.MultiBuffer{b}, nil +} + +func (c *UDPConnClient) WriteMultiBuffer(mb buf.MultiBuffer) error { + for i, b := range mb { + dst := c.Dest + if b.UDP != nil { + if b.UDP.Address.Family().IsDomain() { + if b.UDP.Port != net.Port(dst.Port) { + dst = &net.UDPAddr{IP: dst.IP, Port: int(b.UDP.Port)} + } + } else { + dst = b.UDP.RawNetAddr().(*net.UDPAddr) + } + } + _, err := c.PacketConn.WriteTo(b.Bytes(), dst) + if err != nil { + buf.ReleaseMulti(mb[i:]) + return err + } + b.Release() + } + return nil +} + +type PacketCounterConnection struct { + net.PacketConn + ReadCounter stats.Counter + WriteCounter stats.Counter +} + +func (c *PacketCounterConnection) ReadFrom(p []byte) (n int, addr net.Addr, err error) { + n, addr, err = c.PacketConn.ReadFrom(p) + if err == nil && c.ReadCounter != nil { + c.ReadCounter.Add(int64(n)) + } + return +} + +func (c *PacketCounterConnection) WriteTo(p []byte, addr net.Addr) (n int, err error) { + n, err = c.PacketConn.WriteTo(p, addr) + if err == nil && c.WriteCounter != nil { + c.WriteCounter.Add(int64(n)) + } + return +} + +type entry struct { + saddr []string + deadline time.Time +} + +type cache struct { + running bool + m map[string]entry + mu sync.Mutex +} + +func (c *cache) run() { + if c.running { + return + } + c.running = true + if c.m == nil { + c.m = make(map[string]entry) + } + go c.gc() +} + +func (c *cache) gc() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for now := range ticker.C { + c.mu.Lock() + for key, entry := range c.m { + if now.After(entry.deadline) { + delete(c.m, key) + } + } + if len(c.m) == 0 { + c.running = false + c.mu.Unlock() + return + } + c.mu.Unlock() + } +} + +func (c *cache) LookupHost(host string) []string { + c.mu.Lock() + defer c.mu.Unlock() + c.run() + if entry, ok := c.m[host]; ok { + if time.Now().Before(entry.deadline) { + return entry.saddr + } + delete(c.m, host) + } + return nil +} + +func (c *cache) Cache(host string, saddr []string, ttl uint32) { + c.mu.Lock() + defer c.mu.Unlock() + c.m[host] = entry{ + saddr: saddr, + deadline: time.Now().Add(time.Second * time.Duration(ttl)), + } +} diff --git a/proxy/amnezia/config.go b/proxy/amnezia/config.go new file mode 100644 index 00000000..9faa576e --- /dev/null +++ b/proxy/amnezia/config.go @@ -0,0 +1,60 @@ +package amnezia + +import ( + "encoding/hex" + "net/netip" + + "github.com/xtls/xray-core/common/protocol" + "google.golang.org/protobuf/proto" +) + +func (p *PeerConfig) AsAccount() (protocol.Account, error) { + pub, err := ParseKey(p.PublicKey) + if err != nil { + return nil, err + } + + allowedIPs := make([]netip.Prefix, 0, len(p.AllowedIps)) + for i := range p.AllowedIps { + p, err := netip.ParsePrefix(p.AllowedIps[i]) + if err != nil { + return nil, err + } + allowedIPs = append(allowedIPs, p) + } + + return &MemoryAccount{ + Pub: *pub, + AllowedIPs: allowedIPs, + PreSharedKey: p.PreSharedKey, + KeepAlive: p.KeepAlive, + }, nil +} + +type MemoryAccount struct { + Pub [32]byte + AllowedIPs []netip.Prefix + PreSharedKey string + KeepAlive string +} + +func (a *MemoryAccount) Equals(other protocol.Account) bool { + if b, ok := other.(*MemoryAccount); ok { + return a.Pub == b.Pub + } + return false +} + +func (a *MemoryAccount) ToProto() proto.Message { + allowedIPs := make([]string, 0, len(a.AllowedIPs)) + for i := range a.AllowedIPs { + allowedIPs = append(allowedIPs, a.AllowedIPs[i].String()) + } + + return &PeerConfig{ + PublicKey: hex.EncodeToString(a.Pub[:]), + AllowedIps: allowedIPs, + PreSharedKey: a.PreSharedKey, + KeepAlive: a.KeepAlive, + } +} diff --git a/proxy/amnezia/config.pb.go b/proxy/amnezia/config.pb.go new file mode 100644 index 00000000..f1e399cf --- /dev/null +++ b/proxy/amnezia/config.pb.go @@ -0,0 +1,433 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v6.33.5 +// source: proxy/amnezia/config.proto + +package amnezia + +import ( + protocol "github.com/xtls/xray-core/common/protocol" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type PeerConfig struct { + state protoimpl.MessageState `protogen:"open.v1"` + PublicKey string `protobuf:"bytes,1,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"` + PreSharedKey string `protobuf:"bytes,2,opt,name=pre_shared_key,json=preSharedKey,proto3" json:"pre_shared_key,omitempty"` + Endpoint string `protobuf:"bytes,3,opt,name=endpoint,proto3" json:"endpoint,omitempty"` + KeepAlive string `protobuf:"bytes,4,opt,name=keep_alive,json=keepAlive,proto3" json:"keep_alive,omitempty"` + AllowedIps []string `protobuf:"bytes,5,rep,name=allowed_ips,json=allowedIps,proto3" json:"allowed_ips,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *PeerConfig) Reset() { + *x = PeerConfig{} + mi := &file_proxy_amnezia_config_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *PeerConfig) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PeerConfig) ProtoMessage() {} + +func (x *PeerConfig) ProtoReflect() protoreflect.Message { + mi := &file_proxy_amnezia_config_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PeerConfig.ProtoReflect.Descriptor instead. +func (*PeerConfig) Descriptor() ([]byte, []int) { + return file_proxy_amnezia_config_proto_rawDescGZIP(), []int{0} +} + +func (x *PeerConfig) GetPublicKey() string { + if x != nil { + return x.PublicKey + } + return "" +} + +func (x *PeerConfig) GetPreSharedKey() string { + if x != nil { + return x.PreSharedKey + } + return "" +} + +func (x *PeerConfig) GetEndpoint() string { + if x != nil { + return x.Endpoint + } + return "" +} + +func (x *PeerConfig) GetKeepAlive() string { + if x != nil { + return x.KeepAlive + } + return "" +} + +func (x *PeerConfig) GetAllowedIps() []string { + if x != nil { + return x.AllowedIps + } + return nil +} + +type DeviceConfig struct { + state protoimpl.MessageState `protogen:"open.v1"` + SecretKey string `protobuf:"bytes,1,opt,name=secret_key,json=secretKey,proto3" json:"secret_key,omitempty"` + Endpoint []string `protobuf:"bytes,2,rep,name=endpoint,proto3" json:"endpoint,omitempty"` + Peers []*PeerConfig `protobuf:"bytes,3,rep,name=peers,proto3" json:"peers,omitempty"` + Users []*protocol.User `protobuf:"bytes,5,rep,name=users,proto3" json:"users,omitempty"` + Mtu int32 `protobuf:"varint,4,opt,name=mtu,proto3" json:"mtu,omitempty"` + Reserved []byte `protobuf:"bytes,6,opt,name=reserved,proto3" json:"reserved,omitempty"` + IsClient bool `protobuf:"varint,8,opt,name=is_client,json=isClient,proto3" json:"is_client,omitempty"` + NoKernelTun bool `protobuf:"varint,9,opt,name=no_kernel_tun,json=noKernelTun,proto3" json:"no_kernel_tun,omitempty"` + DNS []string `protobuf:"bytes,10,rep,name=DNS,proto3" json:"DNS,omitempty"` + // AmneziaWG + Jc int32 `protobuf:"varint,11,opt,name=jc,proto3" json:"jc,omitempty"` + JMin int32 `protobuf:"varint,12,opt,name=j_min,json=jMin,proto3" json:"j_min,omitempty"` + JMax int32 `protobuf:"varint,13,opt,name=j_max,json=jMax,proto3" json:"j_max,omitempty"` + S1 int32 `protobuf:"varint,14,opt,name=s1,proto3" json:"s1,omitempty"` // InitPadding + S2 int32 `protobuf:"varint,15,opt,name=s2,proto3" json:"s2,omitempty"` // ResponsePadding + S3 int32 `protobuf:"varint,16,opt,name=s3,proto3" json:"s3,omitempty"` // CookiePadding + S4 int32 `protobuf:"varint,17,opt,name=s4,proto3" json:"s4,omitempty"` // TransportPadding + H1 string `protobuf:"bytes,18,opt,name=h1,proto3" json:"h1,omitempty"` // InitHeader + H2 string `protobuf:"bytes,19,opt,name=h2,proto3" json:"h2,omitempty"` // ResponseHeader + H3 string `protobuf:"bytes,20,opt,name=h3,proto3" json:"h3,omitempty"` // CookieHeader + H4 string `protobuf:"bytes,21,opt,name=h4,proto3" json:"h4,omitempty"` // TransportHeader + I1 string `protobuf:"bytes,22,opt,name=i1,proto3" json:"i1,omitempty"` // Signature (1-5) + I2 string `protobuf:"bytes,23,opt,name=i2,proto3" json:"i2,omitempty"` + I3 string `protobuf:"bytes,24,opt,name=i3,proto3" json:"i3,omitempty"` + I4 string `protobuf:"bytes,25,opt,name=i4,proto3" json:"i4,omitempty"` + I5 string `protobuf:"bytes,26,opt,name=i5,proto3" json:"i5,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *DeviceConfig) Reset() { + *x = DeviceConfig{} + mi := &file_proxy_amnezia_config_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *DeviceConfig) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*DeviceConfig) ProtoMessage() {} + +func (x *DeviceConfig) ProtoReflect() protoreflect.Message { + mi := &file_proxy_amnezia_config_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use DeviceConfig.ProtoReflect.Descriptor instead. +func (*DeviceConfig) Descriptor() ([]byte, []int) { + return file_proxy_amnezia_config_proto_rawDescGZIP(), []int{1} +} + +func (x *DeviceConfig) GetSecretKey() string { + if x != nil { + return x.SecretKey + } + return "" +} + +func (x *DeviceConfig) GetEndpoint() []string { + if x != nil { + return x.Endpoint + } + return nil +} + +func (x *DeviceConfig) GetPeers() []*PeerConfig { + if x != nil { + return x.Peers + } + return nil +} + +func (x *DeviceConfig) GetUsers() []*protocol.User { + if x != nil { + return x.Users + } + return nil +} + +func (x *DeviceConfig) GetMtu() int32 { + if x != nil { + return x.Mtu + } + return 0 +} + +func (x *DeviceConfig) GetReserved() []byte { + if x != nil { + return x.Reserved + } + return nil +} + +func (x *DeviceConfig) GetIsClient() bool { + if x != nil { + return x.IsClient + } + return false +} + +func (x *DeviceConfig) GetNoKernelTun() bool { + if x != nil { + return x.NoKernelTun + } + return false +} + +func (x *DeviceConfig) GetDNS() []string { + if x != nil { + return x.DNS + } + return nil +} + +func (x *DeviceConfig) GetJc() int32 { + if x != nil { + return x.Jc + } + return 0 +} + +func (x *DeviceConfig) GetJMin() int32 { + if x != nil { + return x.JMin + } + return 0 +} + +func (x *DeviceConfig) GetJMax() int32 { + if x != nil { + return x.JMax + } + return 0 +} + +func (x *DeviceConfig) GetS1() int32 { + if x != nil { + return x.S1 + } + return 0 +} + +func (x *DeviceConfig) GetS2() int32 { + if x != nil { + return x.S2 + } + return 0 +} + +func (x *DeviceConfig) GetS3() int32 { + if x != nil { + return x.S3 + } + return 0 +} + +func (x *DeviceConfig) GetS4() int32 { + if x != nil { + return x.S4 + } + return 0 +} + +func (x *DeviceConfig) GetH1() string { + if x != nil { + return x.H1 + } + return "" +} + +func (x *DeviceConfig) GetH2() string { + if x != nil { + return x.H2 + } + return "" +} + +func (x *DeviceConfig) GetH3() string { + if x != nil { + return x.H3 + } + return "" +} + +func (x *DeviceConfig) GetH4() string { + if x != nil { + return x.H4 + } + return "" +} + +func (x *DeviceConfig) GetI1() string { + if x != nil { + return x.I1 + } + return "" +} + +func (x *DeviceConfig) GetI2() string { + if x != nil { + return x.I2 + } + return "" +} + +func (x *DeviceConfig) GetI3() string { + if x != nil { + return x.I3 + } + return "" +} + +func (x *DeviceConfig) GetI4() string { + if x != nil { + return x.I4 + } + return "" +} + +func (x *DeviceConfig) GetI5() string { + if x != nil { + return x.I5 + } + return "" +} + +var File_proxy_amnezia_config_proto protoreflect.FileDescriptor + +const file_proxy_amnezia_config_proto_rawDesc = "" + + "\n" + + "\x1aproxy/amnezia/config.proto\x12\x12xray.proxy.amnezia\x1a\x1acommon/protocol/user.proto\"\xad\x01\n" + + "\n" + + "PeerConfig\x12\x1d\n" + + "\n" + + "public_key\x18\x01 \x01(\tR\tpublicKey\x12$\n" + + "\x0epre_shared_key\x18\x02 \x01(\tR\fpreSharedKey\x12\x1a\n" + + "\bendpoint\x18\x03 \x01(\tR\bendpoint\x12\x1d\n" + + "\n" + + "keep_alive\x18\x04 \x01(\tR\tkeepAlive\x12\x1f\n" + + "\vallowed_ips\x18\x05 \x03(\tR\n" + + "allowedIps\"\xbc\x04\n" + + "\fDeviceConfig\x12\x1d\n" + + "\n" + + "secret_key\x18\x01 \x01(\tR\tsecretKey\x12\x1a\n" + + "\bendpoint\x18\x02 \x03(\tR\bendpoint\x124\n" + + "\x05peers\x18\x03 \x03(\v2\x1e.xray.proxy.amnezia.PeerConfigR\x05peers\x120\n" + + "\x05users\x18\x05 \x03(\v2\x1a.xray.common.protocol.UserR\x05users\x12\x10\n" + + "\x03mtu\x18\x04 \x01(\x05R\x03mtu\x12\x1a\n" + + "\breserved\x18\x06 \x01(\fR\breserved\x12\x1b\n" + + "\tis_client\x18\b \x01(\bR\bisClient\x12\"\n" + + "\rno_kernel_tun\x18\t \x01(\bR\vnoKernelTun\x12\x10\n" + + "\x03DNS\x18\n" + + " \x03(\tR\x03DNS\x12\x0e\n" + + "\x02jc\x18\v \x01(\x05R\x02jc\x12\x13\n" + + "\x05j_min\x18\f \x01(\x05R\x04jMin\x12\x13\n" + + "\x05j_max\x18\r \x01(\x05R\x04jMax\x12\x0e\n" + + "\x02s1\x18\x0e \x01(\x05R\x02s1\x12\x0e\n" + + "\x02s2\x18\x0f \x01(\x05R\x02s2\x12\x0e\n" + + "\x02s3\x18\x10 \x01(\x05R\x02s3\x12\x0e\n" + + "\x02s4\x18\x11 \x01(\x05R\x02s4\x12\x0e\n" + + "\x02h1\x18\x12 \x01(\tR\x02h1\x12\x0e\n" + + "\x02h2\x18\x13 \x01(\tR\x02h2\x12\x0e\n" + + "\x02h3\x18\x14 \x01(\tR\x02h3\x12\x0e\n" + + "\x02h4\x18\x15 \x01(\tR\x02h4\x12\x0e\n" + + "\x02i1\x18\x16 \x01(\tR\x02i1\x12\x0e\n" + + "\x02i2\x18\x17 \x01(\tR\x02i2\x12\x0e\n" + + "\x02i3\x18\x18 \x01(\tR\x02i3\x12\x0e\n" + + "\x02i4\x18\x19 \x01(\tR\x02i4\x12\x0e\n" + + "\x02i5\x18\x1a \x01(\tR\x02i5BX\n" + + "\x16com.xray.proxy.amneziaP\x01Z'github.com/xtls/xray-core/proxy/amnezia\xaa\x02\x12Xray.Proxy.Amneziab\x06proto3" + +var ( + file_proxy_amnezia_config_proto_rawDescOnce sync.Once + file_proxy_amnezia_config_proto_rawDescData []byte +) + +func file_proxy_amnezia_config_proto_rawDescGZIP() []byte { + file_proxy_amnezia_config_proto_rawDescOnce.Do(func() { + file_proxy_amnezia_config_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_proxy_amnezia_config_proto_rawDesc), len(file_proxy_amnezia_config_proto_rawDesc))) + }) + return file_proxy_amnezia_config_proto_rawDescData +} + +var file_proxy_amnezia_config_proto_msgTypes = make([]protoimpl.MessageInfo, 2) +var file_proxy_amnezia_config_proto_goTypes = []any{ + (*PeerConfig)(nil), // 0: xray.proxy.amnezia.PeerConfig + (*DeviceConfig)(nil), // 1: xray.proxy.amnezia.DeviceConfig + (*protocol.User)(nil), // 2: xray.common.protocol.User +} +var file_proxy_amnezia_config_proto_depIdxs = []int32{ + 0, // 0: xray.proxy.amnezia.DeviceConfig.peers:type_name -> xray.proxy.amnezia.PeerConfig + 2, // 1: xray.proxy.amnezia.DeviceConfig.users:type_name -> xray.common.protocol.User + 2, // [2:2] is the sub-list for method output_type + 2, // [2:2] is the sub-list for method input_type + 2, // [2:2] is the sub-list for extension type_name + 2, // [2:2] is the sub-list for extension extendee + 0, // [0:2] is the sub-list for field type_name +} + +func init() { file_proxy_amnezia_config_proto_init() } +func file_proxy_amnezia_config_proto_init() { + if File_proxy_amnezia_config_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_proxy_amnezia_config_proto_rawDesc), len(file_proxy_amnezia_config_proto_rawDesc)), + NumEnums: 0, + NumMessages: 2, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_proxy_amnezia_config_proto_goTypes, + DependencyIndexes: file_proxy_amnezia_config_proto_depIdxs, + MessageInfos: file_proxy_amnezia_config_proto_msgTypes, + }.Build() + File_proxy_amnezia_config_proto = out.File + file_proxy_amnezia_config_proto_goTypes = nil + file_proxy_amnezia_config_proto_depIdxs = nil +} diff --git a/proxy/amnezia/config.proto b/proxy/amnezia/config.proto new file mode 100644 index 00000000..7f5eca75 --- /dev/null +++ b/proxy/amnezia/config.proto @@ -0,0 +1,48 @@ +syntax = "proto3"; + +package xray.proxy.amnezia; +option csharp_namespace = "Xray.Proxy.Amnezia"; +option go_package = "github.com/xtls/xray-core/proxy/amnezia"; +option java_package = "com.xray.proxy.amnezia"; +option java_multiple_files = true; + +import "common/protocol/user.proto"; + +message PeerConfig { + string public_key = 1; + string pre_shared_key = 2; + string endpoint = 3; + string keep_alive = 4; + repeated string allowed_ips = 5; +} + +message DeviceConfig { + string secret_key = 1; + repeated string endpoint = 2; + repeated PeerConfig peers = 3; + repeated xray.common.protocol.User users = 5; + int32 mtu = 4; + + bytes reserved = 6; + bool is_client = 8; + bool no_kernel_tun = 9; + repeated string DNS = 10; + + // AmneziaWG + int32 jc = 11; + int32 j_min = 12; + int32 j_max = 13; + int32 s1 = 14; // InitPadding + int32 s2 = 15; // ResponsePadding + int32 s3 = 16; // CookiePadding + int32 s4 = 17; // TransportPadding + string h1 = 18; // InitHeader + string h2 = 19; // ResponseHeader + string h3 = 20; // CookieHeader + string h4 = 21; // TransportHeader + string i1 = 22; // Signature (1-5) + string i2 = 23; + string i3 = 24; + string i4 = 25; + string i5 = 26; +} diff --git a/proxy/amnezia/icmp.go b/proxy/amnezia/icmp.go new file mode 100644 index 00000000..92d9805a --- /dev/null +++ b/proxy/amnezia/icmp.go @@ -0,0 +1,91 @@ +package amnezia + +import ( + "context" + + "github.com/xtls/xray-core/common/errors" + tunicmp "github.com/xtls/xray-core/proxy/tun/icmp" + "gvisor.dev/gvisor/pkg/buffer" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/tcpip/transport/icmp" +) + +// CreateICMPEchoResponder answers ICMP echo requests from peers locally, the way +// the TUN inbound does: ICMP is not proxied, but ping and connectivity checks +// through the tunnel get a reply instead of timing out. +// +// In promiscuous mode gVisor skips its own IPv4 echo reply for addresses that are +// not assigned to the NIC and leaves it to a custom handler; IPv6 is registered +// too so both families behave the same. +func CreateICMPEchoResponder(gstack *stack.Stack) { + gstack.SetTransportProtocolHandler(icmp.ProtocolNumber4, func(id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool { + return handleICMPEcho(gstack, header.IPv4ProtocolNumber, id, pkt) + }) + gstack.SetTransportProtocolHandler(icmp.ProtocolNumber6, func(id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool { + return handleICMPEcho(gstack, header.IPv6ProtocolNumber, id, pkt) + }) +} + +func handleICMPEcho(gstack *stack.Stack, netProto tcpip.NetworkProtocolNumber, id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool { + srcIP := id.RemoteAddress + dstIP := id.LocalAddress + if srcIP.Len() == 0 || dstIP.Len() == 0 { + return true + } + + headerBytes := pkt.TransportHeader().Slice() + payloadBytes := pkt.Data().AsRange().ToSlice() + message := make([]byte, len(headerBytes)+len(payloadBytes)) + copy(message, headerBytes) + copy(message[len(headerBytes):], payloadBytes) + + if _, _, ok := tunicmp.ParseEchoRequest(netProto, message); !ok { + return true + } + + reply, err := tunicmp.BuildLocalEchoReply(netProto, message, dstIP, srcIP) + if err != nil { + errors.LogInfoInner(context.Background(), err, "failed to build local icmp echo reply") + return true + } + if err := writeRawICMPPacket(gstack, netProto, reply, dstIP, srcIP); err != nil { + errors.LogInfoInner(context.Background(), err, "failed to write local icmp echo reply") + } + return true +} + +func writeRawICMPPacket(gstack *stack.Stack, netProto tcpip.NetworkProtocolNumber, message []byte, srcIP, dstIP tcpip.Address) error { + pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ + ReserveHeaderBytes: header.IPv6MinimumSize, + Payload: buffer.MakeWithData(message), + }) + defer pkt.DecRef() + + if netProto == header.IPv4ProtocolNumber { + ipHdr := header.IPv4(pkt.NetworkHeader().Push(header.IPv4MinimumSize)) + ipHdr.Encode(&header.IPv4Fields{ + TotalLength: uint16(header.IPv4MinimumSize + len(message)), + TTL: 64, + Protocol: uint8(header.ICMPv4ProtocolNumber), + SrcAddr: srcIP, + DstAddr: dstIP, + }) + ipHdr.SetChecksum(^ipHdr.CalculateChecksum()) + } else { + ipHdr := header.IPv6(pkt.NetworkHeader().Push(header.IPv6MinimumSize)) + ipHdr.Encode(&header.IPv6Fields{ + PayloadLength: uint16(len(message)), + TransportProtocol: header.ICMPv6ProtocolNumber, + HopLimit: 64, + SrcAddr: srcIP, + DstAddr: dstIP, + }) + } + + if err := gstack.WriteRawPacket(1, netProto, buffer.MakeWithView(pkt.ToView())); err != nil { + return errors.New("failed to write raw icmp packet back to stack ", err) + } + return nil +} diff --git a/proxy/amnezia/icmp_test.go b/proxy/amnezia/icmp_test.go new file mode 100644 index 00000000..6730b043 --- /dev/null +++ b/proxy/amnezia/icmp_test.go @@ -0,0 +1,176 @@ +package amnezia + +import ( + "bytes" + "net/netip" + "testing" + "time" + + "github.com/xtls/xray-core/common/net" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/checksum" + "gvisor.dev/gvisor/pkg/tcpip/header" +) + +func newICMPTestStack(t *testing.T) *netTun { + t.Helper() + dev, _, gstack, err := CreateNetTUN([]netip.Addr{ + netip.MustParseAddr("10.66.0.1"), + netip.MustParseAddr("fd00::1"), + }, nil, 1420, false) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { dev.Close() }) + CreateForwarder(gstack, func(conn net.Conn, dest net.Destination) { conn.Close() }) + CreateICMPEchoResponder(gstack) + return dev.(*netTun) +} + +// startReader must run before the request is written: the stack may answer +// synchronously inside Write, and netTun hands packets over an unbuffered channel. +func startReader(dev *netTun) <-chan []byte { + got := make(chan []byte, 1) + go func() { + buf := make([]byte, 2048) + sizes := make([]int, 1) + if _, err := dev.Read([][]byte{buf}, sizes, 0); err == nil { + got <- buf[:sizes[0]] + } + }() + return got +} + +func awaitPacket(t *testing.T, got <-chan []byte) []byte { + t.Helper() + select { + case p := <-got: + return p + case <-time.After(2 * time.Second): + t.Fatal("no echo reply from the stack") + return nil + } +} + +func TestICMPv4EchoReply(t *testing.T) { + dev := newICMPTestStack(t) + src := tcpip.AddrFrom4([4]byte{10, 66, 0, 2}) + dst := tcpip.AddrFrom4([4]byte{1, 1, 1, 1}) + payload := []byte("xray amnezia ping") + + icmpMsg := make([]byte, header.ICMPv4MinimumSize+len(payload)) + req := header.ICMPv4(icmpMsg) + req.SetType(header.ICMPv4Echo) + req.SetIdent(0x1234) + req.SetSequence(7) + copy(req.Payload(), payload) + req.SetChecksum(header.ICMPv4Checksum(req[:header.ICMPv4MinimumSize], checksum.Checksum(payload, 0))) + + pkt := make([]byte, header.IPv4MinimumSize+len(icmpMsg)) + ip := header.IPv4(pkt) + ip.Encode(&header.IPv4Fields{ + TotalLength: uint16(len(pkt)), + TTL: 64, + Protocol: uint8(header.ICMPv4ProtocolNumber), + SrcAddr: src, + DstAddr: dst, + }) + ip.SetChecksum(^ip.CalculateChecksum()) + copy(pkt[header.IPv4MinimumSize:], icmpMsg) + + got := startReader(dev) + if _, err := dev.Write([][]byte{pkt}, 0); err != nil { + t.Fatal(err) + } + + reply := header.IPv4(awaitPacket(t, got)) + if !reply.IsValid(len(reply)) { + t.Fatal("invalid ipv4 reply") + } + if reply.SourceAddress() != dst || reply.DestinationAddress() != src { + t.Fatalf("reply addresses %v -> %v, want %v -> %v", reply.SourceAddress(), reply.DestinationAddress(), dst, src) + } + if reply.TransportProtocol() != header.ICMPv4ProtocolNumber { + t.Fatalf("reply protocol %v, want icmpv4", reply.TransportProtocol()) + } + echo := header.ICMPv4(reply.Payload()) + if echo.Type() != header.ICMPv4EchoReply { + t.Fatalf("reply type %v, want echo reply", echo.Type()) + } + if echo.Ident() != 0x1234 || echo.Sequence() != 7 { + t.Fatalf("reply ident/seq %#x/%d, want 0x1234/7", echo.Ident(), echo.Sequence()) + } + if !bytes.Equal(echo.Payload(), payload) { + t.Fatalf("reply payload %q, want %q", echo.Payload(), payload) + } + if checksum.Checksum(echo, 0) != 0xffff { + t.Fatal("bad icmpv4 checksum") + } +} + +func TestICMPv6EchoReply(t *testing.T) { + dev := newICMPTestStack(t) + src := tcpip.AddrFrom16([16]byte{0xfd, 15: 2}) + dst := tcpip.AddrFrom16([16]byte{0x26, 0x06, 0x47, 0x00, 0x47, 0x00, 15: 0x11}) + payload := []byte("xray amnezia ping6") + + icmpMsg := make([]byte, header.ICMPv6MinimumSize+len(payload)) + req := header.ICMPv6(icmpMsg) + req.SetType(header.ICMPv6EchoRequest) + req.SetIdent(0x4321) + req.SetSequence(9) + copy(req.Payload(), payload) + req.SetChecksum(header.ICMPv6Checksum(header.ICMPv6ChecksumParams{ + Header: req[:header.ICMPv6MinimumSize], + Src: src, + Dst: dst, + PayloadCsum: checksum.Checksum(payload, 0), + PayloadLen: len(payload), + })) + + pkt := make([]byte, header.IPv6MinimumSize+len(icmpMsg)) + ip := header.IPv6(pkt) + ip.Encode(&header.IPv6Fields{ + PayloadLength: uint16(len(icmpMsg)), + TransportProtocol: header.ICMPv6ProtocolNumber, + HopLimit: 64, + SrcAddr: src, + DstAddr: dst, + }) + copy(pkt[header.IPv6MinimumSize:], icmpMsg) + + got := startReader(dev) + if _, err := dev.Write([][]byte{pkt}, 0); err != nil { + t.Fatal(err) + } + + reply := header.IPv6(awaitPacket(t, got)) + if !reply.IsValid(len(reply)) { + t.Fatal("invalid ipv6 reply") + } + if reply.SourceAddress() != dst || reply.DestinationAddress() != src { + t.Fatalf("reply addresses %v -> %v, want %v -> %v", reply.SourceAddress(), reply.DestinationAddress(), dst, src) + } + echo := header.ICMPv6(reply.Payload()) + if echo.Type() != header.ICMPv6EchoReply { + t.Fatalf("reply type %v, want echo reply", echo.Type()) + } + if echo.Ident() != 0x4321 || echo.Sequence() != 9 { + t.Fatalf("reply ident/seq %#x/%d, want 0x4321/9", echo.Ident(), echo.Sequence()) + } + if !bytes.Equal(echo.Payload(), payload) { + t.Fatalf("reply payload %q, want %q", echo.Payload(), payload) + } + zeroed := header.ICMPv6(append([]byte(nil), echo[:header.ICMPv6MinimumSize]...)) + zeroed.SetChecksum(0) + want := header.ICMPv6Checksum(header.ICMPv6ChecksumParams{ + Header: zeroed, + Src: dst, + Dst: src, + PayloadCsum: checksum.Checksum(echo.Payload(), 0), + PayloadLen: len(echo.Payload()), + }) + if echo.Checksum() != want { + t.Fatalf("icmpv6 checksum %#x, want %#x", echo.Checksum(), want) + } +} diff --git a/proxy/amnezia/netstack.go b/proxy/amnezia/netstack.go new file mode 100644 index 00000000..8937af53 --- /dev/null +++ b/proxy/amnezia/netstack.go @@ -0,0 +1,838 @@ +/* SPDX-License-Identifier: MIT + * + * Copyright (C) 2017-2025 Amnezia LLC. All Rights Reserved. + */ + +package amnezia + +import ( + "context" + "crypto/rand" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "net/netip" + "os" + "regexp" + "strconv" + "strings" + "syscall" + "time" + + "github.com/amnezia-vpn/amneziawg-go/tun" + xnet "github.com/xtls/xray-core/common/net" + + "golang.org/x/net/dns/dnsmessage" + "gvisor.dev/gvisor/pkg/buffer" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/adapters/gonet" + "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/link/channel" + "gvisor.dev/gvisor/pkg/tcpip/network/ipv4" + "gvisor.dev/gvisor/pkg/tcpip/network/ipv6" + "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/tcpip/transport/icmp" + "gvisor.dev/gvisor/pkg/tcpip/transport/tcp" + "gvisor.dev/gvisor/pkg/tcpip/transport/udp" +) + +type netTun struct { + ep *channel.Endpoint + stack *stack.Stack + events chan tun.Event + notifyHandle *channel.NotificationHandle + incomingPacket chan *buffer.View + closed chan struct{} + mtu int + dnsServers []netip.Addr + hasV4, hasV6 bool +} + +func CreateNetTUN(localAddresses, dnsServers []netip.Addr, mtu int, handleLocal bool) (tun.Device, *Net, *stack.Stack, error) { + opts := stack.Options{ + NetworkProtocols: []stack.NetworkProtocolFactory{ipv4.NewProtocol, ipv6.NewProtocol}, + TransportProtocols: []stack.TransportProtocolFactory{tcp.NewProtocol, udp.NewProtocol, icmp.NewProtocol6, icmp.NewProtocol4}, + HandleLocal: handleLocal, + } + dev := &netTun{ + ep: channel.New(1024, uint32(mtu), ""), + stack: stack.New(opts), + events: make(chan tun.Event, 10), + incomingPacket: make(chan *buffer.View), + closed: make(chan struct{}), + dnsServers: dnsServers, + mtu: mtu, + } + sackEnabledOpt := tcpip.TCPSACKEnabled(true) // TCP SACK is disabled by default + tcpipErr := dev.stack.SetTransportProtocolOption(tcp.ProtocolNumber, &sackEnabledOpt) + if tcpipErr != nil { + return nil, nil, nil, fmt.Errorf("could not enable TCP SACK: %v", tcpipErr) + } + dev.notifyHandle = dev.ep.AddNotify(dev) + tcpipErr = dev.stack.CreateNIC(1, dev.ep) + if tcpipErr != nil { + return nil, nil, nil, fmt.Errorf("CreateNIC: %v", tcpipErr) + } + for _, ip := range localAddresses { + var protoNumber tcpip.NetworkProtocolNumber + if ip.Is4() { + protoNumber = ipv4.ProtocolNumber + } else if ip.Is6() { + protoNumber = ipv6.ProtocolNumber + } + protoAddr := tcpip.ProtocolAddress{ + Protocol: protoNumber, + AddressWithPrefix: tcpip.AddrFromSlice(ip.AsSlice()).WithPrefix(), + } + tcpipErr := dev.stack.AddProtocolAddress(1, protoAddr, stack.AddressProperties{}) + if tcpipErr != nil { + return nil, nil, nil, fmt.Errorf("AddProtocolAddress(%v): %v", ip, tcpipErr) + } + if ip.Is4() { + dev.hasV4 = true + } else if ip.Is6() { + dev.hasV6 = true + } + } + if dev.hasV4 { + dev.stack.AddRoute(tcpip.Route{Destination: header.IPv4EmptySubnet, NIC: 1}) + } + if dev.hasV6 { + dev.stack.AddRoute(tcpip.Route{Destination: header.IPv6EmptySubnet, NIC: 1}) + } + + tnet := &Net{ + DialContextTCPAddrPort: dev.DialContextTCPAddrPort, + DialUDPAddrPort: dev.DialUDPAddrPort, + dnsServers: dev.dnsServers, + hasV4: dev.hasV4, + hasV6: dev.hasV6, + } + + dev.events <- tun.EventUp + return dev, tnet, dev.stack, nil +} + +func (tun *netTun) Name() (string, error) { + return "go", nil +} + +func (tun *netTun) File() *os.File { + return nil +} + +func (tun *netTun) Events() <-chan tun.Event { + return tun.events +} + +func (tun *netTun) Read(buf [][]byte, sizes []int, offset int) (int, error) { + var view *buffer.View + select { + case view = <-tun.incomingPacket: + case <-tun.closed: + return 0, os.ErrClosed + } + + n, err := view.Read(buf[0][offset:]) + view.Release() + if err != nil { + return 0, err + } + sizes[0] = n + return 1, nil +} + +func (tun *netTun) Write(buf [][]byte, offset int) (int, error) { + for _, buf := range buf { + packet := buf[offset:] + if len(packet) == 0 { + continue + } + + pkb := stack.NewPacketBuffer(stack.PacketBufferOptions{Payload: buffer.MakeWithData(packet)}) + switch packet[0] >> 4 { + case 4: + tun.ep.InjectInbound(header.IPv4ProtocolNumber, pkb) + case 6: + tun.ep.InjectInbound(header.IPv6ProtocolNumber, pkb) + default: + return 0, syscall.EAFNOSUPPORT + } + } + return len(buf), nil +} + +func (tun *netTun) WriteNotify() { + pkt := tun.ep.Read() + if pkt == nil { + return + } + + view := pkt.ToView() + pkt.DecRef() + + select { + case tun.incomingPacket <- view: + case <-tun.closed: + view.Release() + } +} + +func (tun *netTun) Close() error { + tun.stack.RemoveNIC(1) + tun.stack.Close() + tun.ep.RemoveNotify(tun.notifyHandle) + tun.ep.Close() + + if tun.events != nil { + close(tun.events) + } + + // we don't close incomingPacket, because WriteNotify may be mid-send on it (DNS lookup) and would panic. + if tun.closed != nil { + close(tun.closed) + } + + return nil +} + +func (tun *netTun) MTU() (int, error) { + return tun.mtu, nil +} + +func (tun *netTun) BatchSize() int { + return 1 +} + +func (tun *netTun) DialContextTCPAddrPort(ctx context.Context, addr netip.AddrPort) (net.Conn, error) { + fa, pn := convertToFullAddr(addr) + return gonet.DialContextTCP(ctx, tun.stack, fa, pn) +} + +func (tun *netTun) DialUDPAddrPort(laddr, raddr netip.AddrPort) (net.Conn, error) { + var pn tcpip.NetworkProtocolNumber = ipv6.ProtocolNumber + if raddr.IsValid() || raddr.Port() > 0 { + _, pn = convertToFullAddr(raddr) + } + conn, err := gonet.DialUDP(tun.stack, nil, nil, pn) + if err != nil { + return nil, err + } + return &xnet.PacketConnWrapper{ + PacketConn: conn, + Dest: net.UDPAddrFromAddrPort(raddr), + }, nil +} + +type Net struct { + DialContextTCPAddrPort func(ctx context.Context, addr netip.AddrPort) (net.Conn, error) + DialUDPAddrPort func(laddr, raddr netip.AddrPort) (net.Conn, error) + dnsServers []netip.Addr + hasV4, hasV6 bool + cache cache +} + +func convertToFullAddr(endpoint netip.AddrPort) (tcpip.FullAddress, tcpip.NetworkProtocolNumber) { + var protoNumber tcpip.NetworkProtocolNumber + if endpoint.Addr().Is4() { + protoNumber = ipv4.ProtocolNumber + } else { + protoNumber = ipv6.ProtocolNumber + } + return tcpip.FullAddress{ + NIC: 1, + Addr: tcpip.AddrFromSlice(endpoint.Addr().AsSlice()), + Port: endpoint.Port(), + }, protoNumber +} + +var ( + errNoSuchHost = errors.New("no such host") + errLameReferral = errors.New("lame referral") + errCannotUnmarshalDNSMessage = errors.New("cannot unmarshal DNS message") + errCannotMarshalDNSMessage = errors.New("cannot marshal DNS message") + errServerMisbehaving = errors.New("server misbehaving") + errInvalidDNSResponse = errors.New("invalid DNS response") + errNoAnswerFromDNSServer = errors.New("no answer from DNS server") + errServerTemporarilyMisbehaving = errors.New("server misbehaving") + errCanceled = errors.New("operation was canceled") + errTimeout = errors.New("i/o timeout") + errNumericPort = errors.New("port must be numeric") + errNoSuitableAddress = errors.New("no suitable address found") + errMissingAddress = errors.New("missing address") +) + +func (net *Net) LookupHost(host string) (addrs []string, err error) { + return net.LookupContextHost(context.Background(), host) +} + +func isDomainName(s string) bool { + l := len(s) + if l == 0 || l > 254 || l == 254 && s[l-1] != '.' { + return false + } + last := byte('.') + nonNumeric := false + partlen := 0 + for i := 0; i < len(s); i++ { + c := s[i] + switch { + default: + return false + case 'a' <= c && c <= 'z' || 'A' <= c && c <= 'Z' || c == '_': + nonNumeric = true + partlen++ + case '0' <= c && c <= '9': + partlen++ + case c == '-': + if last == '.' { + return false + } + partlen++ + nonNumeric = true + case c == '.': + if last == '.' || last == '-' { + return false + } + if partlen > 63 || partlen == 0 { + return false + } + partlen = 0 + } + last = c + } + if last == '-' || partlen > 63 { + return false + } + return nonNumeric +} + +func randU16() uint16 { + var b [2]byte + _, err := rand.Read(b[:]) + if err != nil { + panic(err) + } + return binary.LittleEndian.Uint16(b[:]) +} + +func newRequest(q dnsmessage.Question) (id uint16, udpReq, tcpReq []byte, err error) { + id = randU16() + b := dnsmessage.NewBuilder(make([]byte, 2, 514), dnsmessage.Header{ID: id, RecursionDesired: true}) + b.EnableCompression() + if err := b.StartQuestions(); err != nil { + return 0, nil, nil, err + } + if err := b.Question(q); err != nil { + return 0, nil, nil, err + } + tcpReq, err = b.Finish() + udpReq = tcpReq[2:] + l := len(tcpReq) - 2 + tcpReq[0] = byte(l >> 8) + tcpReq[1] = byte(l) + return id, udpReq, tcpReq, err +} + +func equalASCIIName(x, y dnsmessage.Name) bool { + if x.Length != y.Length { + return false + } + for i := 0; i < int(x.Length); i++ { + a := x.Data[i] + b := y.Data[i] + if 'A' <= a && a <= 'Z' { + a += 0x20 + } + if 'A' <= b && b <= 'Z' { + b += 0x20 + } + if a != b { + return false + } + } + return true +} + +func checkResponse(reqID uint16, reqQues dnsmessage.Question, respHdr dnsmessage.Header, respQues dnsmessage.Question) bool { + if !respHdr.Response { + return false + } + if reqID != respHdr.ID { + return false + } + if reqQues.Type != respQues.Type || reqQues.Class != respQues.Class || !equalASCIIName(reqQues.Name, respQues.Name) { + return false + } + return true +} + +func dnsPacketRoundTrip(c net.Conn, id uint16, query dnsmessage.Question, b []byte) (dnsmessage.Parser, dnsmessage.Header, error) { + if _, err := c.Write(b); err != nil { + return dnsmessage.Parser{}, dnsmessage.Header{}, err + } + b = make([]byte, 512) + for { + n, err := c.Read(b) + if err != nil { + return dnsmessage.Parser{}, dnsmessage.Header{}, err + } + var p dnsmessage.Parser + h, err := p.Start(b[:n]) + if err != nil { + continue + } + q, err := p.Question() + if err != nil || !checkResponse(id, query, h, q) { + continue + } + return p, h, nil + } +} + +func dnsStreamRoundTrip(c net.Conn, id uint16, query dnsmessage.Question, b []byte) (dnsmessage.Parser, dnsmessage.Header, error) { + if _, err := c.Write(b); err != nil { + return dnsmessage.Parser{}, dnsmessage.Header{}, err + } + b = make([]byte, 1280) + if _, err := io.ReadFull(c, b[:2]); err != nil { + return dnsmessage.Parser{}, dnsmessage.Header{}, err + } + l := int(b[0])<<8 | int(b[1]) + if l > len(b) { + b = make([]byte, l) + } + n, err := io.ReadFull(c, b[:l]) + if err != nil { + return dnsmessage.Parser{}, dnsmessage.Header{}, err + } + var p dnsmessage.Parser + h, err := p.Start(b[:n]) + if err != nil { + return dnsmessage.Parser{}, dnsmessage.Header{}, errCannotUnmarshalDNSMessage + } + q, err := p.Question() + if err != nil { + return dnsmessage.Parser{}, dnsmessage.Header{}, errCannotUnmarshalDNSMessage + } + if !checkResponse(id, query, h, q) { + return dnsmessage.Parser{}, dnsmessage.Header{}, errInvalidDNSResponse + } + return p, h, nil +} + +func (tnet *Net) exchange(ctx context.Context, server netip.Addr, q dnsmessage.Question, timeout time.Duration) (dnsmessage.Parser, dnsmessage.Header, error) { + q.Class = dnsmessage.ClassINET + id, udpReq, tcpReq, err := newRequest(q) + if err != nil { + return dnsmessage.Parser{}, dnsmessage.Header{}, errCannotMarshalDNSMessage + } + + for _, useUDP := range []bool{true, false} { + ctx, cancel := context.WithDeadline(ctx, time.Now().Add(timeout)) + defer cancel() + + var c net.Conn + var err error + if useUDP { + c, err = tnet.DialUDPAddrPort(netip.AddrPort{}, netip.AddrPortFrom(server, 53)) + } else { + c, err = tnet.DialContextTCPAddrPort(ctx, netip.AddrPortFrom(server, 53)) + } + + if err != nil { + return dnsmessage.Parser{}, dnsmessage.Header{}, err + } + if d, ok := ctx.Deadline(); ok && !d.IsZero() { + err := c.SetDeadline(d) + if err != nil { + return dnsmessage.Parser{}, dnsmessage.Header{}, err + } + } + var p dnsmessage.Parser + var h dnsmessage.Header + if useUDP { + p, h, err = dnsPacketRoundTrip(c, id, q, udpReq) + } else { + p, h, err = dnsStreamRoundTrip(c, id, q, tcpReq) + } + c.Close() + if err != nil { + if err == context.Canceled { + err = errCanceled + } else if err == context.DeadlineExceeded { + err = errTimeout + } + return dnsmessage.Parser{}, dnsmessage.Header{}, err + } + if err := p.SkipQuestion(); err != dnsmessage.ErrSectionDone { + return dnsmessage.Parser{}, dnsmessage.Header{}, errInvalidDNSResponse + } + if h.Truncated { + continue + } + return p, h, nil + } + return dnsmessage.Parser{}, dnsmessage.Header{}, errNoAnswerFromDNSServer +} + +func checkHeader(p *dnsmessage.Parser, h dnsmessage.Header) error { + if h.RCode == dnsmessage.RCodeNameError { + return errNoSuchHost + } + _, err := p.AnswerHeader() + if err != nil && err != dnsmessage.ErrSectionDone { + return errCannotUnmarshalDNSMessage + } + if h.RCode == dnsmessage.RCodeSuccess && !h.Authoritative && !h.RecursionAvailable && err == dnsmessage.ErrSectionDone { + return errLameReferral + } + if h.RCode != dnsmessage.RCodeSuccess && h.RCode != dnsmessage.RCodeNameError { + if h.RCode == dnsmessage.RCodeServerFailure { + return errServerTemporarilyMisbehaving + } + return errServerMisbehaving + } + return nil +} + +func skipToAnswer(p *dnsmessage.Parser, qtype dnsmessage.Type) error { + for { + h, err := p.AnswerHeader() + if err == dnsmessage.ErrSectionDone { + return errNoSuchHost + } + if err != nil { + return errCannotUnmarshalDNSMessage + } + if h.Type == qtype { + return nil + } + if err := p.SkipAnswer(); err != nil { + return errCannotUnmarshalDNSMessage + } + } +} + +func (tnet *Net) tryOneName(ctx context.Context, name string, qtype dnsmessage.Type) (dnsmessage.Parser, string, error) { + var lastErr error + + n, err := dnsmessage.NewName(name) + if err != nil { + return dnsmessage.Parser{}, "", errCannotMarshalDNSMessage + } + q := dnsmessage.Question{ + Name: n, + Type: qtype, + Class: dnsmessage.ClassINET, + } + + for i := 0; i < 2; i++ { + for _, server := range tnet.dnsServers { + p, h, err := tnet.exchange(ctx, server, q, time.Second*5) + if err != nil { + dnsErr := &net.DNSError{ + Err: err.Error(), + Name: name, + Server: server.String(), + } + if nerr, ok := err.(net.Error); ok && nerr.Timeout() { + dnsErr.IsTimeout = true + } + if _, ok := err.(*net.OpError); ok { + dnsErr.IsTemporary = true + } + lastErr = dnsErr + continue + } + + if err := checkHeader(&p, h); err != nil { + dnsErr := &net.DNSError{ + Err: err.Error(), + Name: name, + Server: server.String(), + } + if err == errServerTemporarilyMisbehaving { + dnsErr.IsTemporary = true + } + if err == errNoSuchHost { + dnsErr.IsNotFound = true + return p, server.String(), dnsErr + } + lastErr = dnsErr + continue + } + + err = skipToAnswer(&p, qtype) + if err == nil { + return p, server.String(), nil + } + lastErr = &net.DNSError{ + Err: err.Error(), + Name: name, + Server: server.String(), + } + if err == errNoSuchHost { + lastErr.(*net.DNSError).IsNotFound = true + return p, server.String(), lastErr + } + } + } + return dnsmessage.Parser{}, "", lastErr +} + +func (tnet *Net) LookupContextHost(ctx context.Context, host string) ([]string, error) { + if saddr := tnet.cache.LookupHost(host); saddr != nil { + return saddr, nil + } + if host == "" || (!tnet.hasV6 && !tnet.hasV4) { + return nil, &net.DNSError{Err: errNoSuchHost.Error(), Name: host, IsNotFound: true} + } + zlen := len(host) + if strings.IndexByte(host, ':') != -1 { + if zidx := strings.LastIndexByte(host, '%'); zidx != -1 { + zlen = zidx + } + } + if ip, err := netip.ParseAddr(host[:zlen]); err == nil { + return []string{ip.String()}, nil + } + + if !isDomainName(host) { + return nil, &net.DNSError{Err: errNoSuchHost.Error(), Name: host, IsNotFound: true} + } + type result struct { + p dnsmessage.Parser + server string + error + } + var addrsV4, addrsV6 []netip.Addr + lanes := 0 + if tnet.hasV4 { + lanes++ + } + if tnet.hasV6 { + lanes++ + } + lane := make(chan result, lanes) + var lastErr error + if tnet.hasV4 { + go func() { + p, server, err := tnet.tryOneName(ctx, host+".", dnsmessage.TypeA) + lane <- result{p, server, err} + }() + } + if tnet.hasV6 { + go func() { + p, server, err := tnet.tryOneName(ctx, host+".", dnsmessage.TypeAAAA) + lane <- result{p, server, err} + }() + } + ttl := uint32(300) + for l := 0; l < lanes; l++ { + result := <-lane + if result.error != nil { + if lastErr == nil { + lastErr = result.error + } + continue + } + + loop: + for { + h, err := result.p.AnswerHeader() + if err != nil && err != dnsmessage.ErrSectionDone { + lastErr = &net.DNSError{ + Err: errCannotMarshalDNSMessage.Error(), + Name: host, + Server: result.server, + } + } + if err != nil { + break + } + switch h.Type { + case dnsmessage.TypeA: + a, err := result.p.AResource() + if err != nil { + lastErr = &net.DNSError{ + Err: errCannotMarshalDNSMessage.Error(), + Name: host, + Server: result.server, + } + break loop + } + ttl = min(ttl, h.TTL) + addrsV4 = append(addrsV4, netip.AddrFrom4(a.A)) + + case dnsmessage.TypeAAAA: + aaaa, err := result.p.AAAAResource() + if err != nil { + lastErr = &net.DNSError{ + Err: errCannotMarshalDNSMessage.Error(), + Name: host, + Server: result.server, + } + break loop + } + ttl = min(ttl, h.TTL) + addrsV6 = append(addrsV6, netip.AddrFrom16(aaaa.AAAA)) + + default: + if err := result.p.SkipAnswer(); err != nil { + lastErr = &net.DNSError{ + Err: errCannotMarshalDNSMessage.Error(), + Name: host, + Server: result.server, + } + break loop + } + continue + } + } + } + // We don't do RFC6724. Instead just put V6 addresses first if an IPv6 address is enabled + var addrs []netip.Addr + if tnet.hasV6 { + addrs = append(addrsV6, addrsV4...) + } else { + addrs = append(addrsV4, addrsV6...) + } + + if len(addrs) == 0 && lastErr != nil { + return nil, lastErr + } + saddrs := make([]string, 0, len(addrs)) + for _, ip := range addrs { + saddrs = append(saddrs, ip.String()) + } + tnet.cache.Cache(host, saddrs, ttl) + return saddrs, nil +} + +func partialDeadline(now, deadline time.Time, addrsRemaining int) (time.Time, error) { + if deadline.IsZero() { + return deadline, nil + } + timeRemaining := deadline.Sub(now) + if timeRemaining <= 0 { + return time.Time{}, errTimeout + } + timeout := timeRemaining / time.Duration(addrsRemaining) + const saneMinimum = 2 * time.Second + if timeout < saneMinimum { + if timeRemaining < saneMinimum { + timeout = timeRemaining + } else { + timeout = saneMinimum + } + } + return now.Add(timeout), nil +} + +var protoSplitter = regexp.MustCompile(`^(tcp|udp|ping)(4|6)?$`) + +func (tnet *Net) DialContext(ctx context.Context, network, address string) (net.Conn, error) { + if ctx == nil { + panic("nil context") + } + var acceptV4, acceptV6 bool + matches := protoSplitter.FindStringSubmatch(network) + if matches == nil { + return nil, &net.OpError{Op: "dial", Err: net.UnknownNetworkError(network)} + } else if len(matches[2]) == 0 { + acceptV4 = true + acceptV6 = true + } else { + acceptV4 = matches[2][0] == '4' + acceptV6 = !acceptV4 + } + var host string + var port int + if matches[1] == "ping" { + host = address + } else { + var sport string + var err error + host, sport, err = net.SplitHostPort(address) + if err != nil { + return nil, &net.OpError{Op: "dial", Err: err} + } + port, err = strconv.Atoi(sport) + if err != nil || port < 0 || port > 65535 { + return nil, &net.OpError{Op: "dial", Err: errNumericPort} + } + } + allAddr, err := tnet.LookupContextHost(ctx, host) + if err != nil { + return nil, &net.OpError{Op: "dial", Err: err} + } + var addrs []netip.AddrPort + for _, addr := range allAddr { + ip, err := netip.ParseAddr(addr) + if err == nil && ((ip.Is4() && acceptV4) || (ip.Is6() && acceptV6)) { + addrs = append(addrs, netip.AddrPortFrom(ip, uint16(port))) + } + } + if len(addrs) == 0 && len(allAddr) != 0 { + return nil, &net.OpError{Op: "dial", Err: errNoSuitableAddress} + } + + var firstErr error + for i, addr := range addrs { + select { + case <-ctx.Done(): + err := ctx.Err() + if err == context.Canceled { + err = errCanceled + } else if err == context.DeadlineExceeded { + err = errTimeout + } + return nil, &net.OpError{Op: "dial", Err: err} + default: + } + + dialCtx := ctx + if deadline, hasDeadline := ctx.Deadline(); hasDeadline { + partialDeadline, err := partialDeadline(time.Now(), deadline, len(addrs)-i) + if err != nil { + if firstErr == nil { + firstErr = &net.OpError{Op: "dial", Err: err} + } + break + } + if partialDeadline.Before(deadline) { + var cancel context.CancelFunc + dialCtx, cancel = context.WithDeadline(ctx, partialDeadline) + defer cancel() + } + } + + var c net.Conn + switch matches[1] { + case "tcp": + c, err = tnet.DialContextTCPAddrPort(dialCtx, addr) + case "udp": + c, err = tnet.DialUDPAddrPort(netip.AddrPort{}, addr) + case "ping": + err = errors.New("not support") + // c, err = tnet.DialPingAddr(netip.Addr{}, addr.Addr()) + } + if err == nil { + return c, nil + } + if firstErr == nil { + firstErr = err + } + } + if firstErr == nil { + firstErr = &net.OpError{Op: "dial", Err: errMissingAddress} + } + return nil, firstErr +} + +func (tnet *Net) Dial(network, address string) (net.Conn, error) { + return tnet.DialContext(context.Background(), network, address) +} diff --git a/proxy/amnezia/params.go b/proxy/amnezia/params.go new file mode 100644 index 00000000..23ab27ff --- /dev/null +++ b/proxy/amnezia/params.go @@ -0,0 +1,37 @@ +package amnezia + +import ( + "strconv" + "strings" +) + +func (c *DeviceConfig) awgUAPI() string { + var sb strings.Builder + addInt := func(key string, v int32) { + sb.WriteString(key + "=" + strconv.FormatInt(int64(v), 10) + "\n") + } + addStr := func(key, v string) { + sb.WriteString(key + "=" + v + "\n") + } + + addInt("jc", c.Jc) + addInt("jmin", c.JMin) + addInt("jmax", c.JMax) + + addInt("s1", c.S1) + addInt("s2", c.S2) + addInt("s3", c.S3) + addInt("s4", c.S4) + + addStr("h1", c.H1) + addStr("h2", c.H2) + addStr("h3", c.H3) + addStr("h4", c.H4) + + addStr("i1", c.I1) + addStr("i2", c.I2) + addStr("i3", c.I3) + addStr("i4", c.I4) + addStr("i5", c.I5) + return sb.String() +} diff --git a/proxy/amnezia/queue_bench_test.go b/proxy/amnezia/queue_bench_test.go new file mode 100644 index 00000000..3df2c43e --- /dev/null +++ b/proxy/amnezia/queue_bench_test.go @@ -0,0 +1,80 @@ +package amnezia + +import ( + "runtime" + "testing" +) + +const benchBatch = 64 + +// Raw cost of queueing and draining a small burst, as one flow's reader does. +func BenchmarkQueueBurstChan(b *testing.B) { + ch := make(chan *packet, udpQueueLimit) + p := &packet{} + b.ReportAllocs() + for i := 0; i < b.N; i++ { + for j := 0; j < benchBatch; j++ { + ch <- p + } + for j := 0; j < benchBatch; j++ { + <-ch + } + } +} + +func BenchmarkQueueBurstPacketQueue(b *testing.B) { + q := newPacketQueue(udpQueueLimit) + p := &packet{} + b.ReportAllocs() + for i := 0; i < b.N; i++ { + for j := 0; j < benchBatch; j++ { + q.push(p) + } + for j := 0; j < benchBatch; j++ { + q.pop() + } + } +} + +// Producer and consumer on different goroutines; the producer yields when the +// queue is full instead of spinning, like a blocking channel send would. +func BenchmarkQueueStreamChan(b *testing.B) { + ch := make(chan *packet, udpQueueLimit) + p := &packet{} + done := make(chan struct{}) + go func() { + for range ch { + } + close(done) + }() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + ch <- p + } + close(ch) + <-done +} + +func BenchmarkQueueStreamPacketQueue(b *testing.B) { + q := newPacketQueue(udpQueueLimit) + p := &packet{} + done := make(chan struct{}) + go func() { + for { + if _, ok := q.pop(); !ok { + break + } + } + close(done) + }() + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + for !q.push(p) { + runtime.Gosched() + } + } + q.close() + <-done +} diff --git a/proxy/amnezia/server.go b/proxy/amnezia/server.go new file mode 100644 index 00000000..2146692d --- /dev/null +++ b/proxy/amnezia/server.go @@ -0,0 +1,401 @@ +package amnezia + +import ( + "context" + "encoding/hex" + "fmt" + "net/netip" + "reflect" + "strings" + "sync" + + "github.com/amnezia-vpn/amneziawg-go/device" + "github.com/amnezia-vpn/amneziawg-go/tun" + "github.com/xtls/xray-core/common" + "github.com/xtls/xray-core/common/buf" + c "github.com/xtls/xray-core/common/ctx" + "github.com/xtls/xray-core/common/errors" + "github.com/xtls/xray-core/common/log" + "github.com/xtls/xray-core/common/net" + "github.com/xtls/xray-core/common/protocol" + "github.com/xtls/xray-core/common/session" + "github.com/xtls/xray-core/core" + "github.com/xtls/xray-core/features/policy" + "github.com/xtls/xray-core/features/routing" + "github.com/xtls/xray-core/features/stats" + "github.com/xtls/xray-core/transport" + "github.com/xtls/xray-core/transport/internet" + "github.com/xtls/xray-core/transport/internet/stat" + "golang.org/x/crypto/curve25519" + "gvisor.dev/gvisor/pkg/tcpip/stack" +) + +type Server struct { + conf *DeviceConfig + ctx context.Context + policyManager policy.Manager + dispatcher routing.Dispatcher + + tag string + src net.Destination + sniffingRequest session.SniffingRequest + streamSettings *internet.MemoryStreamConfig + uplinkCounter stats.Counter + downlinkCounter stats.Counter + + tun tun.Device + stack *stack.Stack + dev *device.Device + mu sync.Mutex + + pub [32]byte + users *sync.Map +} + +func NewServer(ctx context.Context, conf *DeviceConfig) (*Server, error) { + v := core.MustFromContext(ctx) + p := v.GetFeature(policy.ManagerType()).(policy.Manager) + d := v.GetFeature(routing.DispatcherType()).(routing.Dispatcher) + + inbound := session.InboundFromContext(ctx) + content := session.ContentFromContext(ctx) + streamSettings := session.StreamSettingsFromContext(ctx).(*internet.MemoryStreamConfig) + tag := inbound.Tag + var uplinkCounter stats.Counter + var downlinkCounter stats.Counter + if len(tag) > 0 && p.ForSystem().Stats.InboundUplink { + statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager) + name := "inbound>>>" + tag + ">>>traffic>>>uplink" + c, _ := statsManager.GetOrRegisterCounter(name) + if c != nil { + uplinkCounter = c + } + } + if len(tag) > 0 && p.ForSystem().Stats.InboundDownlink { + statsManager := v.GetFeature(stats.ManagerType()).(stats.Manager) + name := "inbound>>>" + tag + ">>>traffic>>>downlink" + c, _ := statsManager.GetOrRegisterCounter(name) + if c != nil { + downlinkCounter = c + } + } + + localAddresses := make([]netip.Addr, 0, len(conf.Endpoint)) + for _, localaddress := range conf.Endpoint { + addr, err := netip.ParseAddr(localaddress) + if err == nil { + localAddresses = append(localAddresses, addr) + continue + } + prefix, err := netip.ParsePrefix(localaddress) + if err == nil { + localAddresses = append(localAddresses, prefix.Addr()) + continue + } + return nil, err + } + + tun, _, stack, err := CreateNetTUN(localAddresses, nil, int(conf.Mtu), false) + if err != nil { + return nil, err + } + + pri := common.Must2(ParseKey(conf.SecretKey)) + var pub [32]byte + curve25519.ScalarBaseMult(&pub, pri) + + users := &sync.Map{} + for _, u := range conf.Users { + user, err := u.ToMemoryUser() + if err != nil { + return nil, err + } + users.Store(user.Account.(*MemoryAccount).Pub, user) + } + + s := &Server{ + conf: conf, + ctx: core.ToBackgroundDetachedContext(ctx), + policyManager: p, + dispatcher: d, + + tag: inbound.Tag, + src: inbound.Source, + sniffingRequest: content.SniffingRequest, + streamSettings: streamSettings, + uplinkCounter: uplinkCounter, + downlinkCounter: downlinkCounter, + + tun: tun, + stack: stack, + + pub: pub, + users: users, + } + // Install the stack's protocol handlers before the device can deliver packets to it (Start -> dev.Up). + CreateForwarder(stack, s.HandleConnection) + CreateICMPEchoResponder(stack) + return s, nil +} + +func (s *Server) AddUser(ctx context.Context, user *protocol.MemoryUser) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.dev == nil { + return errors.New("too early") + } + peer := user.Account.(*MemoryAccount) + if peer.Pub == s.pub { + return errors.New("invalid public key") + } + var sb strings.Builder + sb.WriteString("public_key=" + hex.EncodeToString(peer.Pub[:]) + "\n") + sb.WriteString("replace_allowed_ips=true\n") + for i := range peer.AllowedIPs { + sb.WriteString("allowed_ip=" + peer.AllowedIPs[i].String() + "\n") + } + if peer.PreSharedKey != "" { + sb.WriteString("preshared_key=" + peer.PreSharedKey + "\n") + } + if peer.KeepAlive != "" { + sb.WriteString("persistent_keepalive_interval=" + peer.KeepAlive + "\n") + } + err := s.dev.IpcSet(sb.String()) + if err != nil { + return err + } + s.users.Store(peer.Pub, user) + return nil +} + +func (s *Server) RemoveUser(ctx context.Context, email string) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.dev == nil { + return errors.New("too early") + } + if user := s.GetUser(ctx, email); user != nil { + peer := user.Account.(*MemoryAccount) + err := s.dev.IpcSet("public_key=" + hex.EncodeToString(peer.Pub[:]) + "\nremove=true\n") + if err != nil { + return err + } + s.users.Delete(peer.Pub) + } + return nil +} + +func (s *Server) GetUser(ctx context.Context, email string) (user *protocol.MemoryUser) { + s.users.Range(func(key, value any) bool { + if value.(*protocol.MemoryUser).Email == email { + user = value.(*protocol.MemoryUser) + return false + } + return true + }) + return +} + +func (s *Server) GetUserByAddr(ctx context.Context, addr netip.Addr) (user *protocol.MemoryUser) { + s.users.Range(func(key, value any) bool { + peer := value.(*protocol.MemoryUser).Account.(*MemoryAccount) + for i := range peer.AllowedIPs { + if peer.AllowedIPs[i].Contains(addr) { + user = value.(*protocol.MemoryUser) + return false + } + } + return true + }) + return +} + +func (s *Server) GetUsers(ctx context.Context) (users []*protocol.MemoryUser) { + s.users.Range(func(key, value interface{}) bool { + users = append(users, value.(*protocol.MemoryUser)) + return true + }) + return +} + +func (s *Server) GetUsersCount(context.Context) (count int64) { + s.users.Range(func(key, value interface{}) bool { + count++ + return true + }) + return +} + +// Network implements proxy.Inbound.Network. +func (*Server) Network() []net.Network { + return []net.Network{} +} + +// Process implements proxy.Inbound.Process. +func (s *Server) Process(ctx context.Context, network net.Network, conn stat.Connection, dispatcher routing.Dispatcher) error { + return nil +} + +// Close implements common.Closable.Close. +func (s *Server) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.dev != nil { + s.dev.Close() + s.dev = nil + s.tun = nil + } else if s.tun != nil { + s.tun.Close() + s.tun = nil + } + return nil +} + +// Start implements common.Runnable.Start. +func (s *Server) Start() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.dev != nil { + return nil + } + if s.src.Address.Family().IsDomain() { + return errors.New("address is domain") + } + listenFunc := func() (net.PacketConn, error) { + var pktConn net.PacketConn + var err error + if s.streamSettings.FinalMask != nil { + pktConn, err = s.streamSettings.FinalMask.ListenPacket(context.Background(), &net.UDPAddr{IP: s.src.Address.IP(), Port: int(s.src.Port)}) + } else { + pktConn, err = internet.ListenSystemPacket(context.Background(), &net.UDPAddr{IP: s.src.Address.IP(), Port: int(s.src.Port)}, s.streamSettings.SocketSettings) + } + if err != nil { + return nil, err + } + if s.uplinkCounter != nil || s.downlinkCounter != nil { + pktConn = &PacketCounterConnection{ + PacketConn: pktConn, + ReadCounter: s.uplinkCounter, + WriteCounter: s.downlinkCounter, + } + } + return pktConn, nil + } + bind := &bind{ + listenFunc: listenFunc, + } + logger := &device.Logger{ + Verbosef: func(format string, args ...any) { + log.Record(&log.GeneralMessage{ + Severity: log.Severity_Debug, + Content: fmt.Sprintf(format, args...), + }) + }, + Errorf: func(format string, args ...any) { + log.Record(&log.GeneralMessage{ + Severity: log.Severity_Error, + Content: fmt.Sprintf(format, args...), + }) + }, + } + dev := device.NewDevice(s.tun, bind, logger) + var cfg strings.Builder + cfg.WriteString("private_key=" + s.conf.SecretKey + "\n") + cfg.WriteString(s.conf.awgUAPI()) + s.users.Range(func(key, value any) bool { + peer := value.(*protocol.MemoryUser).Account.(*MemoryAccount) + cfg.WriteString("public_key=" + hex.EncodeToString(peer.Pub[:]) + "\n") + for i := range peer.AllowedIPs { + cfg.WriteString("allowed_ip=" + peer.AllowedIPs[i].String() + "\n") + } + if peer.PreSharedKey != "" { + cfg.WriteString("preshared_key=" + peer.PreSharedKey + "\n") + } + if peer.KeepAlive != "" { + cfg.WriteString("persistent_keepalive_interval=" + peer.KeepAlive + "\n") + } + return true + }) + err := dev.IpcSet(cfg.String()) + if err != nil { + return err + } + err = dev.Up() + if err != nil { + return err + } + s.dev = dev + return nil +} + +func (s *Server) HandleConnection(conn net.Conn, dest net.Destination) { + defer conn.Close() + ctx, cancel := context.WithCancel(s.ctx) + defer cancel() + ctx = c.ContextWithID(ctx, session.NewID()) + + remote := conn.RemoteAddr() + if remote == nil { + errors.LogError(context.Background(), "nil remote") + return + } + + var addr netip.Addr + switch v := remote.(type) { + case *net.TCPAddr: + addr, _ = netip.AddrFromSlice(v.IP) + case *net.UDPAddr: + addr, _ = netip.AddrFromSlice(v.IP) + default: + errors.LogError(context.Background(), "invalid addr type ", reflect.TypeOf(v)) + return + } + + user := s.GetUserByAddr(context.TODO(), addr) + if user == nil { + errors.LogError(context.Background(), "nil user form ", remote, " to ", dest) + return + } + + source := net.DestinationFromAddr(remote) + inbound := session.Inbound{ + Name: "amnezia", + Tag: s.tag, + CanSpliceCopy: 3, + Source: source, + User: user, + } + + ctx = session.ContextWithInbound(ctx, &inbound) + ctx = session.ContextWithContent(ctx, &session.Content{ + SniffingRequest: s.sniffingRequest, + }) + ctx = session.SubContextFromMuxInbound(ctx) + + ctx = log.ContextWithAccessMessage(ctx, &log.AccessMessage{ + From: inbound.Source, + To: dest, + Status: log.AccessAccepted, + Reason: "", + }) + errors.LogInfo(ctx, "processing from ", source, " to ", dest) + + link := &transport.Link{ + Reader: &buf.TimeoutWrapperReader{Reader: buf.NewReader(conn)}, + Writer: buf.NewWriter(conn), + } + if err := s.dispatcher.DispatchLink(ctx, dest, link); err != nil { + errors.LogError(ctx, errors.New("connection closed").Base(err)) + } +} + +func ParseKey(str string) (*[32]byte, error) { + slice, err := hex.DecodeString(str) + if err != nil { + return nil, err + } + if len(slice) != 32 { + return nil, errors.New("len(slice) != 32") + } + return (*[32]byte)(slice), nil +} diff --git a/proxy/amnezia/tun.go b/proxy/amnezia/tun.go new file mode 100644 index 00000000..35686c71 --- /dev/null +++ b/proxy/amnezia/tun.go @@ -0,0 +1,381 @@ +package amnezia + +import ( + "context" + "fmt" + "io" + "runtime" + "strconv" + "strings" + "sync" + "time" + + "github.com/xtls/xray-core/common/buf" + "github.com/xtls/xray-core/common/errors" + "github.com/xtls/xray-core/common/net" + "gvisor.dev/gvisor/pkg/buffer" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/adapters/gonet" + "gvisor.dev/gvisor/pkg/tcpip/checksum" + "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/tcpip/transport/tcp" + "gvisor.dev/gvisor/pkg/tcpip/transport/udp" + "gvisor.dev/gvisor/pkg/waiter" +) + +func CalculateInterfaceName(name string) (tunName string) { + if runtime.GOOS == "darwin" { + tunName = "utun" + } else if name != "" { + tunName = name + } else { + tunName = "tun" + } + interfaces, err := net.Interfaces() + if err != nil { + return + } + var tunIndex int + for _, netInterface := range interfaces { + if strings.HasPrefix(netInterface.Name, tunName) { + index, parseErr := strconv.ParseInt(netInterface.Name[len(tunName):], 10, 16) + if parseErr == nil { + tunIndex = int(index) + 1 + } + } + } + tunName = fmt.Sprintf("%s%d", tunName, tunIndex) + return +} + +func CreateForwarder(gstack *stack.Stack, handler func(conn net.Conn, dest net.Destination)) { + gstack.SetPromiscuousMode(1, true) + gstack.SetSpoofing(1, true) + + tcpForwarder := tcp.NewForwarder(gstack, 0, 65535, func(r *tcp.ForwarderRequest) { + go func(r *tcp.ForwarderRequest) { + var wq waiter.Queue + id := r.ID() + + ep, err := r.CreateEndpoint(&wq) + if err != nil { + errors.LogError(context.Background(), err.String()) + r.Complete(true) + return + } + + options := ep.SocketOptions() + options.SetKeepAlive(false) + options.SetReuseAddress(true) + options.SetReusePort(true) + + handler(gonet.NewTCPConn(&wq, ep), net.TCPDestination(net.IPAddress(id.LocalAddress.AsSlice()), net.Port(id.LocalPort))) + + ep.Close() + r.Complete(false) + }(r) + }) + gstack.SetTransportProtocolHandler(tcp.ProtocolNumber, tcpForwarder.HandlePacket) + + manager := &udpManager{ + stack: gstack, + handler: handler, + m: make(map[string]*udpConn), + } + + gstack.SetTransportProtocolHandler(udp.ProtocolNumber, func(id stack.TransportEndpointID, pkt *stack.PacketBuffer) bool { + data := pkt.Data().AsRange().ToSlice() + // if len(data) == 0 { + // return false + // } + srcIP := net.IPAddress(id.RemoteAddress.AsSlice()) + dstIP := net.IPAddress(id.LocalAddress.AsSlice()) + if srcIP == nil || dstIP == nil { + panic(id) + } + src := net.UDPDestination(srcIP, net.Port(id.RemotePort)) + dst := net.UDPDestination(dstIP, net.Port(id.LocalPort)) + manager.feed(src, dst, data) + return true + }) +} + +type udpManager struct { + stack *stack.Stack + handler func(conn net.Conn, dest net.Destination) + m map[string]*udpConn + mutex sync.RWMutex +} + +func (m *udpManager) feed(src net.Destination, dst net.Destination, data []byte) { + m.mutex.RLock() + uc, ok := m.m[src.NetAddr()] + if ok { + if !uc.queue.push(&packet{p: data, dest: &dst}) { + errors.LogDebug(context.Background(), "drop udp with size ", len(data), " to ", dst.NetAddr(), " original ", uc.dst.NetAddr(), " > queue full") + } + m.mutex.RUnlock() + return + } + m.mutex.RUnlock() + + m.mutex.Lock() + defer m.mutex.Unlock() + + uc, ok = m.m[src.NetAddr()] + if !ok { + uc = &udpConn{ + queue: newPacketQueue(udpQueueLimit), + src: src, + dst: dst, + } + uc.writeFunc = m.writeRawUDPPacket + uc.closeFunc = func() { + m.mutex.Lock() + m.close(uc) + m.mutex.Unlock() + } + m.m[src.NetAddr()] = uc + go m.handler(uc, dst) + } + + if !uc.queue.push(&packet{p: data, dest: &dst}) { + errors.LogDebug(context.Background(), "drop udp with size ", len(data), " to ", dst.NetAddr(), " original ", uc.dst.NetAddr(), " > queue full 2") + } +} + +func (m *udpManager) close(uc *udpConn) { + if !uc.closed { + uc.closed = true + uc.queue.close() + delete(m.m, uc.src.NetAddr()) + } +} + +func (m *udpManager) writeRawUDPPacket(payload []byte, src net.Destination, dst net.Destination) error { + udpLen := header.UDPMinimumSize + len(payload) + srcIP := tcpip.AddrFromSlice(src.Address.IP()) + dstIP := tcpip.AddrFromSlice(dst.Address.IP()) + + // build packet with appropriate IP header size + isIPv4 := dst.Address.Family().IsIPv4() + ipHdrSize := header.IPv6MinimumSize + ipProtocol := header.IPv6ProtocolNumber + if isIPv4 { + ipHdrSize = header.IPv4MinimumSize + ipProtocol = header.IPv4ProtocolNumber + } + + pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ + ReserveHeaderBytes: ipHdrSize + header.UDPMinimumSize, + Payload: buffer.MakeWithData(payload), + }) + defer pkt.DecRef() + + // Build UDP header + udpHdr := header.UDP(pkt.TransportHeader().Push(header.UDPMinimumSize)) + udpHdr.Encode(&header.UDPFields{ + SrcPort: uint16(src.Port), + DstPort: uint16(dst.Port), + Length: uint16(udpLen), + }) + + // Calculate and set UDP checksum + xsum := header.PseudoHeaderChecksum(header.UDPProtocolNumber, srcIP, dstIP, uint16(udpLen)) + udpHdr.SetChecksum(^udpHdr.CalculateChecksum(checksum.Checksum(payload, xsum))) + + // Build IP header + if isIPv4 { + ipHdr := header.IPv4(pkt.NetworkHeader().Push(header.IPv4MinimumSize)) + ipHdr.Encode(&header.IPv4Fields{ + TotalLength: uint16(header.IPv4MinimumSize + udpLen), + TTL: 64, + Protocol: uint8(header.UDPProtocolNumber), + SrcAddr: srcIP, + DstAddr: dstIP, + }) + ipHdr.SetChecksum(^ipHdr.CalculateChecksum()) + } else { + ipHdr := header.IPv6(pkt.NetworkHeader().Push(header.IPv6MinimumSize)) + ipHdr.Encode(&header.IPv6Fields{ + PayloadLength: uint16(udpLen), + TransportProtocol: header.UDPProtocolNumber, + HopLimit: 64, + SrcAddr: srcIP, + DstAddr: dstIP, + }) + } + + // dispatch the packet + err := m.stack.WriteRawPacket(1, ipProtocol, buffer.MakeWithView(pkt.ToView())) + if err != nil { + return errors.New("failed to write raw udp packet back to stack err ", err) + } + + return nil +} + +type packet struct { + p []byte + dest *net.Destination +} + +type udpConn struct { + queue *packetQueue + src net.Destination + dst net.Destination + writeFunc func(payload []byte, src net.Destination, dst net.Destination) error + closeFunc func() + closed bool +} + +func (c *udpConn) ReadMultiBuffer() (buf.MultiBuffer, error) { + for { + q, ok := c.queue.pop() + if !ok { + return nil, io.EOF + } + + b := buf.New() + if _, err := b.Write(q.p); err != nil { + errors.LogErrorInner(context.Background(), err, "drop packet to ", q.dest, " with size ", len(q.p)) + b.Release() + continue + } + + b.UDP = q.dest + + return buf.MultiBuffer{b}, nil + } +} + +func (c *udpConn) Read(p []byte) (int, error) { + q, ok := c.queue.pop() + if !ok { + return 0, io.EOF + } + n := copy(p, q.p) + if n != len(q.p) { + return 0, io.ErrShortBuffer + } + return n, nil +} + +func (c *udpConn) WriteMultiBuffer(mb buf.MultiBuffer) error { + for i, b := range mb { + dst := c.dst + if b.UDP != nil { + if b.UDP.Address.Family().IsDomain() { + errors.LogError(context.Background(), "impossible domain packet ", b.UDP, " reply via original target ", dst) + } else { + dst = *b.UDP + } + } + err := c.writeFunc(b.Bytes(), dst, c.src) + if err != nil { + buf.ReleaseMulti(mb[i:]) + return err + } + b.Release() + } + return nil +} + +func (c *udpConn) Write(p []byte) (int, error) { + err := c.writeFunc(p, c.dst, c.src) + if err != nil { + return 0, err + } + return len(p), nil +} + +func (c *udpConn) Close() error { + c.closeFunc() + return nil +} + +func (c *udpConn) LocalAddr() net.Addr { + return c.dst.RawNetAddr() +} + +func (c *udpConn) RemoteAddr() net.Addr { + return c.src.RawNetAddr() +} + +func (c *udpConn) SetDeadline(t time.Time) error { + return nil +} + +func (c *udpConn) SetReadDeadline(t time.Time) error { + return nil +} + +func (c *udpConn) SetWriteDeadline(t time.Time) error { + return nil +} + +// udpQueueLimit bounds the packets waiting for one UDP flow; more are dropped. +const udpQueueLimit = 1024 + +// packetQueue holds the packets waiting for one UDP flow. Unlike a buffered +// channel of the same bound it only allocates for packets actually queued, so +// the many idle flows kept until the idle timeout cost next to nothing. +type packetQueue struct { + mu sync.Mutex + items []*packet + limit int + notify chan struct{} + closed bool +} + +func newPacketQueue(limit int) *packetQueue { + return &packetQueue{limit: limit, notify: make(chan struct{}, 1)} +} + +// push queues p and reports whether it was accepted. +func (q *packetQueue) push(p *packet) bool { + q.mu.Lock() + defer q.mu.Unlock() + if q.closed || len(q.items) >= q.limit { + return false + } + q.items = append(q.items, p) + select { + case q.notify <- struct{}{}: + default: + } + return true +} + +// pop blocks until a packet is queued or the queue is closed and drained. +func (q *packetQueue) pop() (*packet, bool) { + for { + q.mu.Lock() + if len(q.items) > 0 { + p := q.items[0] + q.items[0] = nil + q.items = q.items[1:] + if len(q.items) == 0 { + q.items = nil + } + q.mu.Unlock() + return p, true + } + if q.closed { + q.mu.Unlock() + return nil, false + } + q.mu.Unlock() + <-q.notify + } +} + +func (q *packetQueue) close() { + q.mu.Lock() + defer q.mu.Unlock() + if !q.closed { + q.closed = true + close(q.notify) + } +} diff --git a/proxy/amnezia/tun_default.go b/proxy/amnezia/tun_default.go new file mode 100644 index 00000000..96e63000 --- /dev/null +++ b/proxy/amnezia/tun_default.go @@ -0,0 +1,18 @@ +//go:build !linux + +package amnezia + +import ( + "errors" + "net/netip" + + "github.com/amnezia-vpn/amneziawg-go/tun" +) + +func createKernelTun([]netip.Addr, []netip.Addr, int) (tdev tun.Device, tnet *Net, err error) { + return nil, nil, errors.New("not implemented") +} + +func KernelTunSupported() (bool, error) { + return false, nil +} diff --git a/proxy/amnezia/tun_linux.go b/proxy/amnezia/tun_linux.go new file mode 100644 index 00000000..06cdbb15 --- /dev/null +++ b/proxy/amnezia/tun_linux.go @@ -0,0 +1,278 @@ +//go:build linux + +package amnezia + +import ( + "context" + goerrors "errors" + "fmt" + "net" + "net/netip" + "os" + "sync/atomic" + "syscall" + + "golang.org/x/sys/unix" + + "github.com/amnezia-vpn/amneziawg-go/tun" + "github.com/vishvananda/netlink" + "github.com/xtls/xray-core/common/errors" + xnet "github.com/xtls/xray-core/common/net" +) + +var tableIndex atomic.Uint32 + +func init() { + tableIndex.Store(10230) +} + +type kernelTun struct { + tun.Device + + dialer *net.Dialer + lc *net.ListenConfig + handle *netlink.Handle + linkAddrs []netlink.Addr + routes []*netlink.Route + rules []*netlink.Rule +} + +func createKernelTun(localAddresses, dnsServers []netip.Addr, mtu int) (tdev tun.Device, tnet *Net, err error) { + var v4, v6 *netip.Addr + for _, prefixes := range localAddresses { + if v4 == nil && prefixes.Is4() { + x := prefixes + v4 = &x + } + if v6 == nil && prefixes.Is6() { + x := prefixes + v6 = &x + } + } + + writeSysctlZero := func(path string) error { + _, err := os.Stat(path) + if os.IsNotExist(err) { + return nil + } + if err != nil { + return err + } + return os.WriteFile(path, []byte("0"), 0o644) + } + + // system configs. + if v4 != nil { + if err = writeSysctlZero("/proc/sys/net/ipv4/conf/all/rp_filter"); err != nil { + return nil, nil, fmt.Errorf("failed to disable ipv4 rp_filter for all: %w", err) + } + } + if v6 != nil { + if err = writeSysctlZero("/proc/sys/net/ipv6/conf/all/disable_ipv6"); err != nil { + return nil, nil, fmt.Errorf("failed to enable ipv6: %w", err) + } + if err = writeSysctlZero("/proc/sys/net/ipv6/conf/all/rp_filter"); err != nil { + return nil, nil, fmt.Errorf("failed to disable ipv6 rp_filter for all: %w", err) + } + } + + n := CalculateInterfaceName("wg") + wgt, err := tun.CreateTUN(n, mtu) + if err != nil { + return nil, nil, err + } + defer func() { + if err != nil { + _ = wgt.Close() + } + }() + + // disable linux rp_filter for tunnel device to avoid packet drop. + // the operation require root privilege on container require '--privileged' flag. + if v4 != nil { + if err = writeSysctlZero("/proc/sys/net/ipv4/conf/" + n + "/rp_filter"); err != nil { + return nil, nil, fmt.Errorf("failed to disable ipv4 rp_filter for tunnel: %w", err) + } + } + if v6 != nil { + if err = writeSysctlZero("/proc/sys/net/ipv6/conf/" + n + "/rp_filter"); err != nil { + return nil, nil, fmt.Errorf("failed to disable ipv6 rp_filter for tunnel: %w", err) + } + } + + var ipv6TableIndex int + if v6 != nil { + r := &netlink.Route{} + for { + ipv6TableIndex = int(tableIndex.Add(1)) - 1 + r.Table = ipv6TableIndex + routeList, fErr := netlink.RouteListFiltered(netlink.FAMILY_V6, r, netlink.RT_FILTER_TABLE) + if fErr != nil { + return nil, nil, errors.New("failed to pre check routes for table: ", ipv6TableIndex).Base(fErr) + } + if len(routeList) == 0 { + errors.LogInfo(context.Background(), "allocate new ipv6 table index: ", ipv6TableIndex) + break + } + // to prevent infinite loop + if ipv6TableIndex > 65535 { + return nil, nil, errors.New("failed to find available ipv6 table index") + } + } + } + + t := &kernelTun{ + Device: wgt, + } + + t.handle, err = netlink.NewHandle() + if err != nil { + return nil, nil, err + } + defer func() { + if err != nil { + t.Close() + } + }() + + l, err := netlink.LinkByName(n) + if err != nil { + return nil, nil, err + } + + if v4 != nil { + addr := netlink.Addr{ + IPNet: &net.IPNet{ + IP: v4.AsSlice(), + Mask: net.CIDRMask(v4.BitLen(), v4.BitLen()), + }, + } + t.linkAddrs = append(t.linkAddrs, addr) + } + if v6 != nil { + addr := netlink.Addr{ + IPNet: &net.IPNet{ + IP: v6.AsSlice(), + Mask: net.CIDRMask(v6.BitLen(), v6.BitLen()), + }, + } + t.linkAddrs = append(t.linkAddrs, addr) + + rt := &netlink.Route{ + LinkIndex: l.Attrs().Index, + Dst: &net.IPNet{ + IP: net.IPv6zero, + Mask: net.CIDRMask(0, 128), + }, + Table: ipv6TableIndex, + } + t.routes = append(t.routes, rt) + + r := netlink.NewRule() + r.Table, r.Family, r.Src = ipv6TableIndex, unix.AF_INET6, addr.IPNet + t.rules = append(t.rules, r) + r = netlink.NewRule() + r.Table, r.Family, r.OifName = ipv6TableIndex, unix.AF_INET6, n + t.rules = append(t.rules, r) + } + + for _, addr := range t.linkAddrs { + if err = t.handle.AddrAdd(l, &addr); err != nil { + return nil, nil, fmt.Errorf("failed to add address %s to %s: %w", addr, n, err) + } + } + if err = t.handle.LinkSetMTU(l, mtu); err != nil { + return nil, nil, err + } + if err = t.handle.LinkSetUp(l); err != nil { + return nil, nil, err + } + + for _, route := range t.routes { + if err = t.handle.RouteAdd(route); err != nil { + return nil, nil, fmt.Errorf("failed to add route %s: %w", route, err) + } + } + for _, rule := range t.rules { + if err = t.handle.RuleAdd(rule); err != nil { + return nil, nil, fmt.Errorf("failed to add rule %s: %w", rule, err) + } + } + + dialer := &net.Dialer{} + dialer.Control = func(network, address string, c syscall.RawConn) error { + return c.Control(func(fd uintptr) { + if err := syscall.BindToDevice(int(fd), n); err != nil { + errors.LogInfoInner(context.Background(), err, "failed to bind to device") + } + }) + } + lc := &net.ListenConfig{} + lc.Control = func(network, address string, c syscall.RawConn) error { + return c.Control(func(fd uintptr) { + if err := syscall.BindToDevice(int(fd), n); err != nil { + errors.LogInfoInner(context.Background(), err, "failed to bind to device") + } + }) + } + t.dialer = dialer + t.lc = lc + + tnet = &Net{ + DialContextTCPAddrPort: t.DialContextTCPAddrPort, + DialUDPAddrPort: t.DialUDPAddrPort, + dnsServers: dnsServers, + hasV4: v4 != nil, + hasV6: v6 != nil, + } + + return t, tnet, nil +} + +func (tun *kernelTun) Close() (err error) { + var errs []error + for _, rule := range tun.rules { + if err = tun.handle.RuleDel(rule); err != nil { + errs = append(errs, fmt.Errorf("failed to delete rule: %w", err)) + } + } + for _, route := range tun.routes { + if err = tun.handle.RouteDel(route); err != nil { + errs = append(errs, fmt.Errorf("failed to delete route: %w", err)) + } + } + if err = tun.Device.Close(); err != nil { + errs = append(errs, fmt.Errorf("failed to close device: %w", err)) + } + tun.handle.Close() + errs = append(errs, tun.Device.Close()) + return goerrors.Join(errs...) +} + +func (tun *kernelTun) DialContextTCPAddrPort(ctx context.Context, addr netip.AddrPort) (net.Conn, error) { + return tun.dialer.DialContext(ctx, "tcp", addr.String()) +} + +func (tun *kernelTun) DialUDPAddrPort(laddr, raddr netip.AddrPort) (net.Conn, error) { + conn, err := tun.lc.ListenPacket(context.Background(), "udp", ":0") + if err != nil { + return nil, err + } + return &xnet.PacketConnWrapper{ + PacketConn: conn, + Dest: net.UDPAddrFromAddrPort(raddr), + }, nil +} + +func KernelTunSupported() (bool, error) { + var hdr unix.CapUserHeader + hdr.Version = unix.LINUX_CAPABILITY_VERSION_3 + hdr.Pid = 0 // 0 means current process + + var data unix.CapUserData + if err := unix.Capget(&hdr, &data); err != nil { + return false, fmt.Errorf("failed to get capabilities: %v", err) + } + + return (data.Effective & (1 << unix.CAP_NET_ADMIN)) != 0, nil +} diff --git a/proxy/amnezia/tun_test.go b/proxy/amnezia/tun_test.go new file mode 100644 index 00000000..716da006 --- /dev/null +++ b/proxy/amnezia/tun_test.go @@ -0,0 +1,91 @@ +package amnezia + +import ( + "testing" + "time" + + "github.com/xtls/xray-core/common/net" +) + +// BenchmarkUDPManagerNewSession measures what one new UDP flow costs the +// inbound while it stays open: QUIC and DNS open many short flows, and each +// one lives until the connection idle timeout. +func BenchmarkUDPManagerNewSession(b *testing.B) { + m := &udpManager{ + handler: func(conn net.Conn, dest net.Destination) {}, + m: make(map[string]*udpConn), + } + dst := net.UDPDestination(net.ParseAddress("1.1.1.1"), 443) + payload := make([]byte, 1200) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + src := net.UDPDestination(net.IPAddress([]byte{10, byte(i >> 16), byte(i >> 8), byte(i)}), net.Port(1024+i%60000)) + m.feed(src, dst, payload) + } +} + +func TestPacketQueueOrderAndClose(t *testing.T) { + q := newPacketQueue(udpQueueLimit) + for i := 0; i < 3; i++ { + if !q.push(&packet{p: []byte{byte(i)}}) { + t.Fatalf("push %d rejected", i) + } + } + for i := 0; i < 3; i++ { + p, ok := q.pop() + if !ok || p.p[0] != byte(i) { + t.Fatalf("pop %d: got %v, %v", i, p, ok) + } + } + q.close() + if _, ok := q.pop(); ok { + t.Fatal("pop after close returned a packet") + } + if q.push(&packet{}) { + t.Fatal("push after close accepted") + } +} + +func TestPacketQueueLimit(t *testing.T) { + q := newPacketQueue(udpQueueLimit) + for i := 0; i < udpQueueLimit; i++ { + if !q.push(&packet{}) { + t.Fatalf("push %d rejected below the limit", i) + } + } + if q.push(&packet{}) { + t.Fatal("push above the limit accepted") + } +} + +func TestPacketQueueCloseUnblocksReader(t *testing.T) { + q := newPacketQueue(udpQueueLimit) + done := make(chan bool) + go func() { + _, ok := q.pop() + done <- ok + }() + q.close() + select { + case ok := <-done: + if ok { + t.Fatal("blocked pop returned a packet after close") + } + case <-time.After(time.Second): + t.Fatal("close did not wake the reader") + } +} + +func TestPacketQueueDropsDrainedStorage(t *testing.T) { + q := newPacketQueue(udpQueueLimit) + for i := 0; i < 100; i++ { + q.push(&packet{}) + } + for i := 0; i < 100; i++ { + q.pop() + } + if q.items != nil { + t.Fatalf("drained queue still holds %d slots", cap(q.items)) + } +}