xinyin025/3x-ui
1
0
Fork 0
mirror of https://github.com/MHSanaei/3x-ui.git synced 2026-09-05 01:47:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(amneziawg): auto-wire the SOCKS5 relay into every embedded Device

Browse Source 
Manager.ensureLocked now attaches AttachTCPForwarder/AttachUDPHandler to
every Device it builds, relaying into that instance's own loopback SOCKS5
inbound automatically -- no caller needs to know relay.go exists at all.
Peer identity is re-looked-up via Manager.Lookup on every connection rather
than captured once at attach time, so a reconfigure-in-place (peers added/
removed without a full rebuild) doesn't leave the forwarder working off a
stale peer index.

Added TestManagerEnsureAutomaticallyWiresRelay: drives this through the
real Manager.Ensure entry point (not manual wiring like the existing
relay_e2e_test.go) against a real xray-core process, confirming the
automatic attachment and the port/password Manager derives internally
actually agree with what a real SOCKS5 inbound expects.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Kuzz007
2026-08-02 14:42:00 +03:00
parent 3450d872d9
commit 124665c5ef
2 changed files with 257 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/amneziawgnet/manager.go
+70 -13
View File
@@ -2,9 +2,12 @@ package amneziawgnet
import (
"fmt"
"net/netip"
"strings"
"sync"
"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
"github.com/mhsanaei/3x-ui/v3/internal/amneziawg"
"github.com/mhsanaei/3x-ui/v3/internal/logger"
)
@@ -18,12 +21,13 @@ type Desired struct {
Options DeviceOptions
}
// managed is one running embedded interface: the live Device, the peer
// lookup index built from its current peer list, and enough of its own
// configuration to decide whether a later Ensure call can reconfigure it in
// place or needs to rebuild it from scratch.
// managed is one running embedded interface: the live Device, its UDP relay
// sessions, the peer lookup index built from its current peer list, and
// enough of its own configuration to decide whether a later Ensure call can
// reconfigure it in place or needs to rebuild it from scratch.
type managed struct {
dev *Device
udpRelay *UDPRelay
peers *PeerIndex
inst amneziawg.Instance
structFP string
@@ -33,12 +37,11 @@ type managed struct {
// inbound id -- the same shape as internal/amneziawg.Manager (GetManager()
// + sync.Once, mu-guarded map, Ensure/Reconcile/StopAll/HasRunning), so a
// caller already familiar with that Manager needs to learn nothing new here.
// Unlike that Manager, this one doesn't attach any traffic handling by
// itself: Ensure/Reconcile only bring each Instance's Device up to date.
// Attaching a forwarder/UDP handler (see forwarder.go / udp.go) using the
// Device and PeerIndex returned by Lookup is left to the caller -- today a
// test harness, later the Phase 2 SOCKS5 relay wiring -- since this package
// doesn't yet know what that handler should do with a recovered connection.
// Every Device this Manager builds gets its TCP forwarder and UDP handler
// attached automatically (see ensureLocked), relaying into that instance's
// own loopback SOCKS5 inbound (SOCKSPortForInbound/SocksPassword) -- a
// caller only needs to keep calling Ensure/Reconcile with fresh Instance
// data; it doesn't need to know relay.go exists at all.
type Manager struct {
mu sync.Mutex
ifaces map[int]*managed
@@ -97,6 +100,7 @@ func (m *Manager) ensureLocked(d Desired) error {
}
if exists {
cur.udpRelay.Close()
cur.dev.Close()
delete(m.ifaces, inst.Id)
}
@@ -104,8 +108,47 @@ func (m *Manager) ensureLocked(d Desired) error {
if err != nil {
return err
}
relay := socksRelayForInstance(inst)
udpRelay := NewUDPRelay(relay, dev.Stack)
inboundID := inst.Id // captured for the closures below, which outlive this call
AttachTCPForwarder(dev.Stack, func(conn *gonet.TCPConn, dest netip.AddrPort) {
srcAddrPort, err := netip.ParseAddrPort(conn.RemoteAddr().String())
if err != nil {
conn.Close()
return
}
// Re-fetched on every connection, not captured once at attach time:
// a reconfigure-in-place (peers added/removed, no rebuild) replaces
// cur.peers without ever re-attaching the forwarder, so a stale
// captured index would silently miss newly-added peers.
_, peers, ok := m.Lookup(inboundID)
if !ok {
conn.Close()
return
}
peer, ok := peers.Lookup(srcAddrPort.Addr().Unmap())
if !ok {
conn.Close()
return
}
relay.RelayTCP(conn, peer.Email, dest)
})
AttachUDPHandler(dev.Stack, func(src, dst netip.AddrPort, payload []byte) {
_, peers, ok := m.Lookup(inboundID)
if !ok {
return
}
peer, ok := peers.Lookup(src.Addr())
if !ok {
return
}
udpRelay.Handle(src, dst, peer.Email, payload)
})
m.ifaces[inst.Id] = &managed{
dev: dev,
udpRelay: udpRelay,
peers: NewPeerIndex(inst.Peers),
inst: inst,
structFP: structFP,
@@ -114,6 +157,17 @@ func (m *Manager) ensureLocked(d Desired) error {
return nil
}
// socksRelayForInstance derives the loopback SOCKS5 relay address/password
// for inst -- both fully determined by its id and the process-wide
// password (SOCKSPortForInbound/SocksPassword), so no per-instance state
// needs threading through Desired/DeviceOptions for this.
func socksRelayForInstance(inst amneziawg.Instance) SocksRelay {
return SocksRelay{
Addr: fmt.Sprintf("127.0.0.1:%d", SOCKSPortForInbound(inst.Id)),
Password: SocksPassword(),
}
}
// addressFingerprint captures the two Instance fields that can't be changed
// on a running Device via IpcSet alone (they're fixed when the gVisor
// netstack is built) -- everything else (keys, listen port, obfuscation,
@@ -137,6 +191,7 @@ func (m *Manager) Reconcile(desired []Desired) {
if _, ok := want[id]; ok {
continue
}
cur.udpRelay.Close()
cur.dev.Close()
delete(m.ifaces, id)
logger.Infof("amneziawgnet: stopped embedded interface for removed inbound %d", id)
@@ -153,6 +208,7 @@ func (m *Manager) StopAll() {
m.mu.Lock()
defer m.mu.Unlock()
for id, cur := range m.ifaces {
cur.udpRelay.Close()
cur.dev.Close()
delete(m.ifaces, id)
}
@@ -166,9 +222,10 @@ func (m *Manager) HasRunning() bool {
}
// Lookup returns the running Device and PeerIndex for inbound id, if any --
// for a caller that wants to attach its own forwarder/handler (a test
// harness today, the Phase 2 SOCKS5 relay wiring later) once the interface
// is up.
// the forwarder/UDP-handler closures ensureLocked attaches use this to
// re-fetch the current peer index on every connection (see ensureLocked's
// comment on why), and it's equally available to a test harness or any
// other caller that wants read access to a managed interface's state.
func (m *Manager) Lookup(id int) (dev *Device, peers *PeerIndex, ok bool) {
m.mu.Lock()
defer m.mu.Unlock()
internal/amneziawgnet/relay_e2e_test.go
+187
View File
@@ -284,6 +284,193 @@ func TestSocksRelayAgainstRealXray(t *testing.T) {
}
}
// TestManagerEnsureAutomaticallyWiresRelay is Phase 3's own real proof: unlike
// TestSocksRelayAgainstRealXray above (which builds a Device and attaches
// RelayTCP/UDPRelay by hand), this drives everything through the public
// Manager.Ensure entry point the real app actually calls -- confirming
// ensureLocked's own forwarder/UDP-handler attachment (added this phase)
// really does relay a fresh Device's traffic into Xray with zero manual
// wiring from the caller. Uses the exact port/password
// (SOCKSPortForInbound/SocksPassword) the Manager computes internally, so
// this only passes if that internal derivation and the externally-visible
// contract genuinely agree.
func TestManagerEnsureAutomaticallyWiresRelay(t *testing.T) {
bin := os.Getenv("XRAY_E2E_BINARY")
if bin == "" {
t.Skip("set XRAY_E2E_BINARY to an xray binary to run this test")
}
localIP, ok := firstNonLoopbackIPv4()
if !ok {
t.Skip("no non-loopback IPv4 address available on this host")
}
const wantEmail = "manager-e2e-peer@example.com"
const listenPort = 58716
const inboundID = 5
tcpEcho, tcpEchoAddr := startTCPEcho(t, localIP)
defer tcpEcho.Close()
serverPriv, serverPub, err := wireguard.GenerateWireguardKeypair()
if err != nil {
t.Fatalf("generate server keypair: %v", err)
}
clientPriv, clientPub, err := wireguard.GenerateWireguardKeypair()
if err != nil {
t.Fatalf("generate client keypair: %v", err)
}
inst := amneziawg.Instance{
Id: inboundID,
InterfaceName: "awgtest5",
ListenPort: listenPort,
PrivateKey: serverPriv,
PublicKey: serverPub,
Address: []string{"10.205.0.1/24"},
MTU: 1420,
Obfuscation: amneziawg.Obfuscation20{
Jc: 4, Jmin: 40, Jmax: 70,
S1: 20, S2: 30, S3: 20, S4: 20,
},
Peers: []amneziawg.Peer{{
Email: wantEmail,
PublicKey: clientPub,
AllowedIPs: []string{"10.205.0.2/32"},
}},
}
// A real xray-core process with a SOCKS5 inbound at exactly the port and
// password ensureLocked will derive on its own for this instance --
// SocksPassword() is cached (sync.Once), so calling it here first and
// again inside Manager.Ensure below returns the identical value.
socksPort := SOCKSPortForInbound(inboundID)
password := SocksPassword()
settingsJSON, err := SocksInboundSettings([]string{wantEmail}, password)
if err != nil {
t.Fatalf("SocksInboundSettings: %v", err)
}
var rawSettings any
if err := json.Unmarshal(settingsJSON, &rawSettings); err != nil {
t.Fatalf("unmarshal generated SOCKS5 settings: %v", err)
}
xrayCfg := map[string]any{
"log": map[string]any{"loglevel": "debug"},
"inbounds": []any{
map[string]any{
"listen": "127.0.0.1",
"port": socksPort,
"protocol": "socks",
"settings": rawSettings,
"tag": "awg-e2e-manager",
},
},
"outbounds": []any{
map[string]any{"protocol": "freedom", "settings": map[string]any{}, "tag": "direct"},
},
"policy": map[string]any{
"levels": map[string]any{
"0": map[string]any{"statsUserUplink": true, "statsUserDownlink": true},
},
},
"stats": map[string]any{},
}
cfgBytes, err := json.MarshalIndent(xrayCfg, "", " ")
if err != nil {
t.Fatalf("marshal xray config: %v", err)
}
cfgPath := filepath.Join(t.TempDir(), "config.json")
if err := os.WriteFile(cfgPath, cfgBytes, 0o644); err != nil {
t.Fatalf("write xray config: %v", err)
}
var xrayLog syncBuffer
cmd := exec.Command(bin, "-c", cfgPath)
cmd.Stdout = &xrayLog
cmd.Stderr = &xrayLog
if err := cmd.Start(); err != nil {
t.Fatalf("start xray: %v", err)
}
defer func() {
_ = cmd.Process.Kill()
_, _ = cmd.Process.Wait()
}()
waitForPort(t, socksPort)
// A throwaway Manager, not the process-wide singleton, so this test
// doesn't interact with any other test's state.
m := &Manager{ifaces: map[int]*managed{}}
defer m.StopAll()
if err := m.Ensure(Desired{Instance: inst}); err != nil {
t.Fatalf("Manager.Ensure: %v", err)
}
dev, _, ok := m.Lookup(inboundID)
if !ok {
t.Fatal("Lookup after Ensure: not found")
}
defer dev.Close() // StopAll would also do this; explicit for clarity
clientTun, clientNet, err := netstack.CreateNetTUN(
[]netip.Addr{netip.MustParseAddr("10.205.0.2")},
[]netip.Addr{netip.MustParseAddr("1.1.1.1")}, 1420)
if err != nil {
t.Fatalf("client CreateNetTUN: %v", err)
}
clientDev := device.NewDevice(clientTun, awgconn.NewDefaultBind(), device.NewLogger(device.LogLevelSilent, ""))
defer clientDev.Close()
clientPrivHex, err := wireguard.KeyToHex(clientPriv)
if err != nil {
t.Fatalf("client key to hex: %v", err)
}
serverPubHex, err := wireguard.KeyToHex(serverPub)
if err != nil {
t.Fatalf("server key to hex: %v", err)
}
clientConf := fmt.Sprintf(
"private_key=%s\njc=4\njmin=40\njmax=70\ns1=20\ns2=30\ns3=20\ns4=20\npublic_key=%s\nendpoint=127.0.0.1:%d\nallowed_ip=0.0.0.0/0\n",
clientPrivHex, serverPubHex, listenPort)
if err := clientDev.IpcSet(clientConf); err != nil {
t.Fatalf("client IpcSet: %v", err)
}
if err := clientDev.Up(); err != nil {
t.Fatalf("client Up: %v", err)
}
const tcpMsg = "hello via Manager.Ensure's automatic relay wiring"
dialDeadline := time.Now().Add(10 * time.Second)
var conn net.Conn
for {
c, dialErr := clientNet.DialContext(t.Context(), "tcp", tcpEchoAddr.String())
if dialErr == nil {
conn = c
break
}
if time.Now().After(dialDeadline) {
t.Fatalf("client TCP dial via tunnel never succeeded: %v", dialErr)
}
time.Sleep(150 * time.Millisecond)
}
defer conn.Close()
if _, err := conn.Write([]byte(tcpMsg)); err != nil {
t.Fatalf("client TCP write: %v", err)
}
buf := make([]byte, len(tcpMsg))
if _, err := readFull(conn, buf, 10*time.Second); err != nil {
t.Fatalf("client TCP read: %v", err)
}
if string(buf) != tcpMsg {
t.Errorf("TCP echo = %q, want %q", buf, tcpMsg)
}
_ = cmd.Process.Kill()
_, _ = cmd.Process.Wait()
log := xrayLog.String()
wantUp := fmt.Sprintf("user>>>%s>>>traffic>>>uplink", wantEmail)
if !strings.Contains(log, wantUp) {
t.Errorf("xray log missing uplink stats counter %q (Manager.Ensure's automatic relay wiring may not be attributing traffic correctly)\nfull log:\n%s", wantUp, log)
}
}
// firstNonLoopbackIPv4 finds a real, locally-bound IPv4 address suitable as
// a relay-reachable test destination.
func firstNonLoopbackIPv4() (netip.Addr, bool) {