Files
3x-ui/internal/amneziawgnet/egress.go
T
MHSanaei d103c0c012 fix(amneziawg): keep the fallback egress port below the panel's reserved ports
When 64900 was taken, the egress asked the OS for any free port. Where the
ephemeral range reaches above 62789 (Windows), that could be the Xray API port
or a node-egress, TUIC or AmneziaWG relay port, which then failed to bind. The
fallback now picks a random free loopback port below 62789.
2026-10-24 12:23:24 +02:00

721 lines
20 KiB
Go

package amneziawgnet
import (
"context"
"crypto/hmac"
"encoding/binary"
"errors"
"fmt"
"io"
mrand "math/rand/v2"
"net"
"net/netip"
"sync"
"time"
"gvisor.dev/gvisor/pkg/tcpip"
"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
"github.com/mhsanaei/3x-ui/v3/internal/amneziawg"
"github.com/mhsanaei/3x-ui/v3/internal/logger"
)
// EgressBasePort is the loopback port the panel's SOCKS5 egress server tries
// first; generated amneziawg socks bridges dial whichever port it holds.
const EgressBasePort = 64900
// reservedLoopbackFloor is where the panel's own loopback listeners start (Xray API 62789, node
// egress, TUIC and AmneziaWG relays); a fallback egress port must stay below it.
const reservedLoopbackFloor = 62789
// listenTCP opens a loopback listener; a test stands in for the OS through it.
var listenTCP = func(ctx context.Context, addr string) (net.Listener, error) {
return (&net.ListenConfig{}).Listen(ctx, "tcp", addr)
}
// socks5EgressServer is a minimal loopback SOCKS5 server routing Xray's
// bridged amneziawg outbounds into their embedded devices' netstacks.
type socks5EgressServer struct {
mu sync.Mutex
stacks map[string]*Device // outbound tag -> its device
dns map[string]string // outbound tag -> its DNS server
tracked map[net.Conn]struct{}
// dnsServer resolves domain targets through the outbound netstack.
dnsServer string
listener net.Listener // nil when stopped; acceptLoop takes it as an arg
closing chan struct{} // per-listener lifetime signal, rearmed by Listen
wg sync.WaitGroup
}
var (
egressOnce sync.Once
egressServer *socks5EgressServer
)
// GetEgressServer returns the process-wide SOCKS5 egress server singleton.
func GetEgressServer() *socks5EgressServer {
egressOnce.Do(func() {
egressServer = &socks5EgressServer{
stacks: map[string]*Device{},
dns: map[string]string{},
tracked: map[net.Conn]struct{}{},
}
})
return egressServer
}
// currentDNSServer reads dnsServer under lock -- custom per-tag DNS preferred.
func (s *socks5EgressServer) currentDNSServer(tag ...string) string {
s.mu.Lock()
defer s.mu.Unlock()
if len(tag) > 0 && tag[0] != "" {
if custom, ok := s.dns[tag[0]]; ok && custom != "" {
return custom
}
}
return s.dnsServer
}
// SetDNSServer overrides the domain-target resolver (tests).
func (s *socks5EgressServer) SetDNSServer(addr string) {
s.mu.Lock()
defer s.mu.Unlock()
s.dnsServer = addr
}
// SetStack registers or replaces the device backing an outbound tag.
func (s *socks5EgressServer) SetStack(tag string, dev *Device, dnsServer ...string) {
norm := ""
if len(dnsServer) > 0 && dnsServer[0] != "" {
norm = amneziawg.NormalizeDNSServer(dnsServer[0])
}
s.mu.Lock()
prevDev := s.stacks[tag]
prevDNS := s.dns[tag]
s.stacks[tag] = dev
if norm != "" {
s.dns[tag] = norm
} else {
delete(s.dns, tag)
}
changed := prevDev != dev || prevDNS != norm
s.mu.Unlock()
if changed {
flushTunnelDNSCacheForTag(tag)
}
}
// DeleteStack drops an outbound tag's registration (outbound removed).
func (s *socks5EgressServer) DeleteStack(tag string) {
s.mu.Lock()
delete(s.stacks, tag)
delete(s.dns, tag)
s.mu.Unlock()
flushTunnelDNSCacheForTag(tag)
}
// Listen starts accepting on EgressBasePort, or on any free loopback port when
// the OS refuses it. Idempotent; the caller's reconcile tick retries a failure.
func (s *socks5EgressServer) Listen() error {
s.mu.Lock()
defer s.mu.Unlock()
if s.listener != nil {
return nil
}
ln, err := listenTCP(context.Background(), fmt.Sprintf("127.0.0.1:%d", EgressBasePort))
if err != nil {
var fallbackErr error
if ln, fallbackErr = listenBelowReserved(context.Background()); fallbackErr != nil {
return fmt.Errorf("amneziawgnet: egress listen: %w", errors.Join(err, fallbackErr))
}
logger.Warningf("amneziawgnet: egress port %d unavailable, using a free port: %v", EgressBasePort, err)
}
s.listener = ln
s.closing = make(chan struct{})
logger.Infof("amneziawgnet: egress socks listening on %s", ln.Addr())
s.wg.Add(1)
go s.acceptLoop(ln, s.closing)
return nil
}
// fallbackPortFloor keeps the fallback clear of the well-known and registered low ports.
const fallbackPortFloor = 40000
// listenBelowReserved binds a random loopback port under reservedLoopbackFloor, retrying taken or
// OS-reserved ones; an OS-chosen port could land inside the panel's own ranges.
func listenBelowReserved(ctx context.Context) (net.Listener, error) {
var errs []error
for range 64 {
port := fallbackPortFloor + mrand.IntN(reservedLoopbackFloor-fallbackPortFloor)
ln, err := listenTCP(ctx, fmt.Sprintf("127.0.0.1:%d", port))
if err == nil {
return ln, nil
}
errs = append(errs, err)
}
return nil, errors.Join(errs...)
}
// Port is the port generated socks bridges must dial: the bound one, or
// EgressBasePort while nothing is bound, since Listen tries it first.
func (s *socks5EgressServer) Port() int {
if port, ok := s.boundPort(); ok {
return port
}
return EgressBasePort
}
func (s *socks5EgressServer) boundPort() (int, bool) {
s.mu.Lock()
defer s.mu.Unlock()
if s.listener == nil {
return 0, false
}
addr, ok := s.listener.Addr().(*net.TCPAddr)
if !ok {
return 0, false
}
return addr.Port, true
}
// EgressPort is the process-wide egress server's Port.
func EgressPort() int {
return GetEgressServer().Port()
}
// Close stops the listener and in-flight handlers; signal first so an accept
// error always observes closing.
func (s *socks5EgressServer) Close() {
s.mu.Lock()
ln := s.listener
s.listener = nil
if ln == nil {
s.mu.Unlock()
return
}
close(s.closing)
tracked := s.tracked
s.tracked = map[net.Conn]struct{}{}
s.mu.Unlock()
ln.Close()
for conn := range tracked {
conn.Close()
}
s.wg.Wait()
}
func (s *socks5EgressServer) acceptLoop(ln net.Listener, closing chan struct{}) {
defer s.wg.Done()
for {
conn, err := ln.Accept()
if err != nil {
select {
case <-closing:
return
default:
}
logger.Warningf("amneziawgnet: egress accept: %v", err)
continue
}
select {
case <-closing:
conn.Close()
return
default:
}
s.mu.Lock()
if s.listener == nil {
s.mu.Unlock()
conn.Close()
return
}
s.tracked[conn] = struct{}{}
s.wg.Add(1)
s.mu.Unlock()
go func(c net.Conn) {
defer s.wg.Done()
defer func() {
s.mu.Lock()
delete(s.tracked, c)
s.mu.Unlock()
}()
s.handleConn(c)
}(conn)
}
}
// stackFor resolves a tag to its live device at use time, so rebuilds take
// effect for new connections without touching the listener.
func (s *socks5EgressServer) stackFor(tag string) (*Device, bool) {
s.mu.Lock()
defer s.mu.Unlock()
dev, ok := s.stacks[tag]
return dev, ok
}
func (s *socks5EgressServer) handleConn(conn net.Conn) {
defer conn.Close()
// Bound the pre-auth handshake so a silent client never pins a handler
// indefinitely across Close() and wg.Wait().
_ = conn.SetDeadline(time.Now().Add(portForwardDialTimeout))
method, err := socks5Greeting(conn)
if err != nil || method == 0xFF {
return
}
user := ""
if method == 0x02 {
// RFC 1929 sub-negotiation: VER(1) | ULEN(1) | UNAME | PLEN(1) |
// PASSWD -- the leading 0x01 version byte must be consumed first.
var ver [1]byte
if _, err := io.ReadFull(conn, ver[:]); err != nil {
return
}
var ulen [1]byte
if _, err := io.ReadFull(conn, ulen[:]); err != nil {
return
}
uname := make([]byte, ulen[0])
if _, err := io.ReadFull(conn, uname); err != nil {
return
}
user = string(uname)
var plen [1]byte
if _, err := io.ReadFull(conn, plen[:]); err != nil {
return
}
pass := make([]byte, plen[0])
if _, err := io.ReadFull(conn, pass); err != nil {
return
}
if !hmac.Equal(pass, []byte(SocksPassword())) {
_, _ = conn.Write([]byte{0x01, 0x01})
return
}
if _, err := conn.Write([]byte{0x01, 0x00}); err != nil {
return
}
}
var req [4]byte
if _, err := io.ReadFull(conn, req[:]); err != nil {
return
}
// Handshake complete: clear deadline for the relay phase.
_ = conn.SetDeadline(time.Time{})
target, err := readSocksRequestTarget(conn, req[3])
if err != nil {
writeSocksReply(conn, 0x01, netip.AddrPort{})
return
}
switch req[1] {
case 0x01: // CONNECT
dev, ok := s.stackFor(user)
if !ok {
writeSocksReply(conn, 0x05, netip.AddrPort{})
return
}
dest, err := target.resolveTunnelVia(s.currentDNSServer(user), user, dev)
if err != nil {
logger.Warningf("amneziawgnet: egress %q: resolve %s: %v", user, target, err)
writeSocksReply(conn, 0x04, netip.AddrPort{})
return
}
s.relayTCP(dev, user, conn, dest)
case 0x03: // UDP ASSOCIATE
dev, ok := s.stackFor(user)
if !ok {
writeSocksReply(conn, 0x05, netip.AddrPort{})
return
}
s.relayUDP(dev, user, udpControl{conn: conn}, target)
default:
writeSocksReply(conn, 0x07, netip.AddrPort{})
}
}
// socks5Greeting requires RFC 1929 username/password auth (0x02).
// Returns 0xFF when unauthenticated or unsupported.
func socks5Greeting(conn net.Conn) (byte, error) {
var hdr [2]byte
if _, err := io.ReadFull(conn, hdr[:]); err != nil {
return 0xFF, err
}
methods := make([]byte, hdr[1])
if _, err := io.ReadFull(conn, methods); err != nil {
return 0xFF, err
}
hasUserPass := false
for _, m := range methods {
if m == 0x02 {
hasUserPass = true
break
}
}
if !hasUserPass {
_, _ = conn.Write([]byte{0x05, 0xFF})
return 0xFF, nil
}
if _, err := conn.Write([]byte{0x05, 0x02}); err != nil {
return 0xFF, err
}
return 0x02, nil
}
// socksTarget is a parsed SOCKS5 request address: an IP, or the raw hostname
// for ATYP 0x03 (resolved through the outbound's tunnel, never host-side).
type socksTarget struct {
host string
ip netip.Addr
port uint16
}
func readSocksRequestTarget(r io.Reader, atyp byte) (socksTarget, error) {
var t socksTarget
switch atyp {
case 0x01:
var b [4]byte
if _, err := io.ReadFull(r, b[:]); err != nil {
return t, err
}
t.ip = netip.AddrFrom4(b)
case 0x04:
var b [16]byte
if _, err := io.ReadFull(r, b[:]); err != nil {
return t, err
}
t.ip = netip.AddrFrom16(b)
case 0x03:
var l [1]byte
if _, err := io.ReadFull(r, l[:]); err != nil {
return t, err
}
name := make([]byte, l[0])
if _, err := io.ReadFull(r, name); err != nil {
return t, err
}
t.host = string(name)
default:
return t, fmt.Errorf("unsupported SOCKS5 request address type %d", atyp)
}
var portBytes [2]byte
if _, err := io.ReadFull(r, portBytes[:]); err != nil {
return t, err
}
t.port = binary.BigEndian.Uint16(portBytes[:])
return t, nil
}
func (t socksTarget) String() string {
if t.ip.IsValid() {
return netip.AddrPortFrom(t.ip, t.port).String()
}
return fmt.Sprintf("%s:%d", t.host, t.port)
}
// Domain targets resolve via the tunnel; reply-side helper must not be used here.
func (t socksTarget) resolveTunnelVia(dnsServer, tag string, dev *Device) (netip.AddrPort, error) {
if t.ip.IsValid() {
if !tunnelSupportsAddr(dev.LocalAddresses(), t.ip) {
return netip.AddrPort{}, fmt.Errorf("tunnel has no route for %s (device addresses %v)", t.ip, dev.LocalAddresses())
}
return netip.AddrPortFrom(t.ip, t.port), nil
}
ctx, cancel := context.WithTimeout(context.Background(), tunnelResolveTimeout)
defer cancel()
addr, err := resolveTunnelVia(ctx, dev, tag, dnsServer, t.host)
if err != nil {
return netip.AddrPort{}, err
}
return netip.AddrPortFrom(addr, t.port), nil
}
// writeSocksReply emits a reply with an empty v4 bind address; Xray only
// reads the code byte.
func writeSocksReply(w io.Writer, code byte, _ netip.AddrPort) {
out := []byte{0x05, code, 0x00, 0x01, 0, 0, 0, 0, 0, 0}
_, _ = w.Write(out)
}
// relayTCP dials dest inside the tagged outbound's netstack and pipes both
// directions until either side closes.
func (s *socks5EgressServer) relayTCP(dev *Device, tag string, upstream net.Conn, dest netip.AddrPort) {
fa := tcpip.FullAddress{
NIC: 1,
Addr: tcpip.AddrFromSlice(dest.Addr().AsSlice()),
Port: dest.Port(),
}
// Bound dial with portForwardDialTimeout so unreachable peers do not
// pin goroutines and netstack endpoints in s.tracked.
dctx, dcancel := context.WithTimeout(context.Background(), portForwardDialTimeout)
defer dcancel()
tunnelConn, err := gonet.DialContextTCP(dctx, dev.Stack, fa, tunnelNetwork(dest.Addr()))
if err != nil {
logger.Warningf("amneziawgnet: egress %q: dial tunnel %s: %v", tag, dest, err)
writeSocksReply(upstream, 0x01, netip.AddrPort{})
return
}
defer tunnelConn.Close()
if _, err := upstream.Write([]byte{0x05, 0x00, 0x00, 0x01, 0, 0, 0, 0, 0, 0}); err != nil {
return
}
done := make(chan struct{}, 2)
go func() { _, _ = io.Copy(tunnelConn, upstream); done <- struct{}{} }()
go func() { _, _ = io.Copy(upstream, tunnelConn); done <- struct{}{} }()
<-done
}
// udpControl is the control half of one UDP ASSOCIATE: the TCP connection
// whose lifetime bounds the association (RFC 1928).
type udpControl struct{ conn net.Conn }
// egressUDPSession is one UDP ASSOCIATE flow: host-facing socket plus a
// connected tunnel endpoint whose source port makes replies answerable.
type egressUDPSession struct {
dst netip.AddrPort
conn *gonet.UDPConn
}
// relayUDP answers the associate request and relays datagrams to
// per-destination tunnel endpoints until the control connection closes.
func (s *socks5EgressServer) relayUDP(dev *Device, tag string, ctl udpControl, _ socksTarget) {
udpConn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
if err != nil {
logger.Warningf("amneziawgnet: egress %q: udp bind: %v", tag, err)
writeSocksReply(ctl.conn, 0x01, netip.AddrPort{})
return
}
defer udpConn.Close()
local := udpConn.LocalAddr().(*net.UDPAddr)
ip4 := local.IP.To4()
reply := []byte{
0x05, 0x00, 0x00, 0x01, ip4[0], ip4[1], ip4[2], ip4[3],
byte(local.Port >> 8), byte(local.Port),
}
if _, err := ctl.conn.Write(reply); err != nil {
return
}
sessions := &udpEgressSessions{m: map[netip.AddrPort]*egressUDPSession{}}
// Reader: strip per-datagram SOCKS5 headers and forward into the tunnel;
// only the associated client's address is accepted.
go func() {
var client netip.AddrPort
buf := make([]byte, 65536)
for {
n, from, err := udpConn.ReadFrom(buf)
if err != nil {
return
}
if src, ok := udpAddrPort(from); ok {
if client.IsValid() && src != client {
continue // RFC 1928: only the associated client may send
}
client = src
}
data := buf[:n]
if len(data) < 4 {
continue
}
atyp := data[3]
var dst netip.AddrPort
var payloadOff int
if atyp == 0x03 {
name, port, hdrLen, perr := parseDatagramDomainHeader(data)
if perr != nil {
continue
}
// Resolve off reader loop so slow tunnel DNS lookups do not
// stall other destinations on this association.
go func(client netip.AddrPort, name string, port uint16, hdrLen int, datagram []byte) {
dnsSrv := s.currentDNSServer(tag)
rctx, rcancel := context.WithTimeout(context.Background(), tunnelResolveTimeout)
daddr, rerr := resolveTunnelVia(rctx, dev, tag, dnsSrv, name)
rcancel()
if rerr != nil {
logger.Warningf("amneziawgnet: egress %q: resolve udp %q (dns=%s): %v", tag, name, dnsSrv, rerr)
return
}
s.deliverUDPDatagram(dev, tag, udpConn, client, sessions, netip.AddrPortFrom(daddr, port), datagram[hdrLen:])
}(client, name, port, hdrLen, append([]byte(nil), data...))
continue
} else {
hdrLen := 4 + addrLen(atyp) + 2
if hdrLen <= 6 || len(data) < hdrLen {
continue
}
d, derr := parseDatagramHeader(data[:hdrLen])
if derr != nil {
logger.Warningf("amneziawgnet: egress %q: udp header: %v", tag, derr)
continue
}
dst = d
payloadOff = hdrLen
}
s.deliverUDPDatagram(dev, tag, udpConn, client, sessions, dst, data[payloadOff:])
}
}()
// Control-conn close tears down relaying -- relay.go UDPRelay contract.
buf := make([]byte, 512)
for {
if _, err := ctl.conn.Read(buf); err != nil {
sessions.closeAll()
return
}
}
}
// udpEgressSessions guards the association's session map: the reader
// goroutine inserts while the control-conn teardown iterates.
type udpEgressSessions struct {
mu sync.Mutex
m map[netip.AddrPort]*egressUDPSession
}
func (s *udpEgressSessions) getOrDial(dev *Device, tag string, udpConn *net.UDPConn, client netip.AddrPort, dst netip.AddrPort) *egressUDPSession {
s.mu.Lock()
defer s.mu.Unlock()
if sess, ok := s.m[dst]; ok {
return sess
}
if !tunnelSupportsAddr(dev.LocalAddresses(), dst.Addr()) {
logger.Warningf("amneziawgnet: egress %q: dial udp %s: tunnel has no route for address family (device addresses %v)", tag, dst, dev.LocalAddresses())
return nil
}
raddr := tcpip.FullAddress{
NIC: 1,
Addr: tcpip.AddrFromSlice(dst.Addr().AsSlice()),
Port: dst.Port(),
}
conn, err := gonet.DialUDP(dev.Stack, nil, &raddr, tunnelNetwork(dst.Addr()))
if err != nil {
logger.Warningf("amneziawgnet: egress %q: dial udp %s: %v", tag, dst, err)
return nil
}
sess := &egressUDPSession{dst: dst, conn: conn}
s.m[dst] = sess
go pumpUDPEgress(udpConn, client, sess, s)
return sess
}
func (s *udpEgressSessions) closeAll() {
s.mu.Lock()
defer s.mu.Unlock()
for _, sess := range s.m {
sess.conn.Close()
}
}
// deliverUDPDatagram forwards one payload to dst through the tunnel endpoint.
// Safe for concurrent use across resolver and direct-path goroutines.
func (s *socks5EgressServer) deliverUDPDatagram(dev *Device, tag string, udpConn *net.UDPConn, client netip.AddrPort, sessions *udpEgressSessions, dst netip.AddrPort, payload []byte) {
if !client.IsValid() {
return // nothing to reply to yet
}
sess := sessions.getOrDial(dev, tag, udpConn, client, dst)
if sess == nil {
return
}
if _, werr := sess.conn.Write(payload); werr != nil {
logger.Warningf("amneziawgnet: egress %q: send udp to %s: %v", tag, dst, werr)
}
}
// pumpUDPEgress reads replies from one connected tunnel endpoint and writes
// them back to the associated client as SOCKS5 UDP datagrams.
func pumpUDPEgress(udpConn *net.UDPConn, client netip.AddrPort, sess *egressUDPSession, sessions *udpEgressSessions) {
defer func() {
sessions.mu.Lock()
delete(sessions.m, sess.dst)
sessions.mu.Unlock()
sess.conn.Close()
}()
buf := make([]byte, 65536)
for {
// Reap idle egress sessions to avoid holding them indefinitely.
_ = sess.conn.SetReadDeadline(time.Now().Add(portForwardUDPIdleTimeout))
n, err := sess.conn.Read(buf)
if err != nil {
return
}
hdr := make([]byte, 0, 3+1+16+2+n)
hdr = append(hdr, 0x00, 0x00, 0x00) // RSV RSV FRAG(=0)
if sess.dst.Addr().Is4() {
b := sess.dst.Addr().As4()
hdr = append(hdr, 0x01)
hdr = append(hdr, b[:]...)
} else {
b := sess.dst.Addr().As16()
hdr = append(hdr, 0x04)
hdr = append(hdr, b[:]...)
}
var portBytes [2]byte
binary.BigEndian.PutUint16(portBytes[:], sess.dst.Port())
hdr = append(hdr, portBytes[:]...)
hdr = append(hdr, buf[:n]...)
if _, err := udpConn.WriteTo(hdr, net.UDPAddrFromAddrPort(client)); err != nil {
return
}
}
}
// parseDatagramDomainHeader decodes a domain SOCKS5 UDP header (RSV RSV FRAG
// 0x03 LEN NAME PORT) into name, port, and header length.
func parseDatagramDomainHeader(data []byte) (name string, port uint16, hdrLen int, err error) {
if len(data) < 5 {
return "", 0, 0, fmt.Errorf("short domain header")
}
l := int(data[4])
hdrLen = 4 + 1 + l + 2
if l == 0 || len(data) < hdrLen {
return "", 0, 0, fmt.Errorf("short domain payload")
}
name = string(data[5 : 5+l])
port = binary.BigEndian.Uint16(data[5+l : 7+l])
return name, port, hdrLen, nil
}
// addrLen returns the wire length of a SOCKS5 address of the given ATYP.
func addrLen(atyp byte) int {
switch atyp {
case 0x01:
return 4
case 0x04:
return 16
default:
return -1
}
}
// parseDatagramHeader decodes the destination from the front of a SOCKS5 UDP
// datagram header block (RSV RSV FRAG ATYP ADDR PORT).
func parseDatagramHeader(hdr []byte) (netip.AddrPort, error) {
if len(hdr) < 4 {
return netip.AddrPort{}, fmt.Errorf("short header")
}
atyp := hdr[3]
body := hdr[4:]
switch atyp {
case 0x01:
if len(body) < 6 {
return netip.AddrPort{}, fmt.Errorf("short v4")
}
return netip.AddrPortFrom(netip.AddrFrom4([4]byte(body[:4])), binary.BigEndian.Uint16(body[4:6])), nil
case 0x04:
if len(body) < 18 {
return netip.AddrPort{}, fmt.Errorf("short v6")
}
return netip.AddrPortFrom(netip.AddrFrom16([16]byte(body[:16])), binary.BigEndian.Uint16(body[16:18])), nil
default:
return netip.AddrPort{}, fmt.Errorf("unsupported atyp %d", atyp)
}
}