From 5efa016c98e9218041b87a7e1360a4ebe9af7c16 Mon Sep 17 00:00:00 2001 From: Free-Guy-IR Date: Wed, 23 Sep 2026 04:55:24 +0330 Subject: [PATCH 1/2] fix(xray): leave unchanged users alone during user sync SyncUser and UpdateUsers removed and re-added every user on every inbound they stay on, even when nothing about the account changed. On a core that closes a user's connections when the user is removed, every such sync would disconnect the user. A user is now removed and added again only when their account on that inbound changed. Users who leave an inbound are removed as before. An account whose add fails is dropped from the local state, so the next sync retries it. --- backend/xray/config.go | 23 +++ backend/xray/user.go | 13 +- backend/xray/user_sync_test.go | 287 +++++++++++++++++++++++++++++++++ 3 files changed, 320 insertions(+), 3 deletions(-) create mode 100644 backend/xray/user_sync_test.go diff --git a/backend/xray/config.go b/backend/xray/config.go index 8199a75d..722ad744 100644 --- a/backend/xray/config.go +++ b/backend/xray/config.go @@ -6,6 +6,7 @@ import ( "log" "net" "net/netip" + "reflect" "slices" "sort" "strings" @@ -298,6 +299,28 @@ func (i *Inbound) updateUsers(accounts []api.Account, removeEmails []string) { } } +func (i *Inbound) hasAccount(account api.Account) bool { + i.mu.RLock() + defer i.mu.RUnlock() + + current, ok := i.clients[account.GetEmail()] + return ok && reflect.DeepEqual(current, account) +} + +func (i *Inbound) changedAccounts(accounts []api.Account) []api.Account { + i.mu.RLock() + defer i.mu.RUnlock() + + changed := make([]api.Account, 0, len(accounts)) + for _, account := range accounts { + current, ok := i.clients[account.GetEmail()] + if !ok || !reflect.DeepEqual(current, account) { + changed = append(changed, account) + } + } + return changed +} + func (i *Inbound) removeUser(email string) { i.mu.Lock() defer i.mu.Unlock() diff --git a/backend/xray/user.go b/backend/xray/user.go index 671622b4..0391bdd5 100644 --- a/backend/xray/user.go +++ b/backend/xray/user.go @@ -140,12 +140,17 @@ func (x *Xray) SyncUser(ctx context.Context, user *common.User) error { continue } - _ = handler.RemoveInboundUser(ctx, inbound.Tag, user.Email) account, isActive := isActiveInbound(inbound, userInbounds, proxySetting) + if isActive && inbound.hasAccount(account) { + continue + } + + _ = handler.RemoveInboundUser(ctx, inbound.Tag, user.Email) if isActive { inbound.updateUser(account) err = handler.AddInboundUser(ctx, inbound.Tag, accountForAPI(inbound, account)) if err != nil { + inbound.removeUser(user.GetEmail()) log.Println(err) errMessage.WriteString("\n" + err.Error()) } @@ -229,15 +234,17 @@ func (x *Xray) UpdateUsers(ctx context.Context, users []*common.User) error { } inbound := inboundByTag[tag] - inbound.updateUsers(update.accounts, removeEmails) + accounts := inbound.changedAccounts(update.accounts) + inbound.updateUsers(accounts, removeEmails) for _, email := range removeEmails { handler.RemoveInboundUser(ctx, tag, email) } - for _, account := range update.accounts { + for _, account := range accounts { _ = handler.RemoveInboundUser(ctx, tag, account.GetEmail()) if err := handler.AddInboundUser(ctx, tag, accountForAPI(inbound, account)); err != nil { + inbound.removeUser(account.GetEmail()) log.Println(err) errMessage.WriteString("\n" + err.Error()) } diff --git a/backend/xray/user_sync_test.go b/backend/xray/user_sync_test.go new file mode 100644 index 00000000..6586f6e4 --- /dev/null +++ b/backend/xray/user_sync_test.go @@ -0,0 +1,287 @@ +package xray + +import ( + "context" + "errors" + "net" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/xtls/xray-core/app/proxyman/command" + "google.golang.org/grpc" + + "github.com/pasarguard/node/backend/xray/api" + "github.com/pasarguard/node/common" +) + +const recordingInbounds = `{"inbounds":[ + {"tag":"vless","protocol":"vless","settings":{"clients":[],"decryption":"none"}}, + {"tag":"vmess","protocol":"vmess","settings":{"clients":[]}}, + {"tag":"trojan","protocol":"trojan","settings":{"clients":[]}}, + {"tag":"ss","protocol":"shadowsocks","settings":{"clients":[],"method":"aes-128-gcm","network":"tcp,udp"}}, + {"tag":"ss2022","protocol":"shadowsocks","settings":{"clients":[],"method":"2022-blake3-aes-128-gcm","password":"MDEyMzQ1Njc4OWFiY2RlZg==","network":"tcp,udp"}}, + {"tag":"hysteria","protocol":"hysteria","settings":{"clients":[]}} +]}` + +var recordingTags = []string{"vless", "vmess", "trojan", "ss", "ss2022", "hysteria"} + +type recordingInboundService struct { + command.UnimplementedHandlerServiceServer + mu sync.Mutex + operations []string + failAdds int +} + +func (s *recordingInboundService) AlterInbound(_ context.Context, request *command.AlterInboundRequest) (*command.AlterInboundResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + + operation := request.GetOperation().GetType() + switch { + case strings.HasSuffix(operation, ".AddUserOperation"): + operation = "add" + case strings.HasSuffix(operation, ".RemoveUserOperation"): + operation = "remove" + } + s.operations = append(s.operations, request.GetTag()+":"+operation) + + if operation == "add" && s.failAdds > 0 { + s.failAdds-- + return nil, errors.New("add rejected") + } + return &command.AlterInboundResponse{}, nil +} + +func (s *recordingInboundService) take() []string { + s.mu.Lock() + defer s.mu.Unlock() + + operations := s.operations + s.operations = nil + slices.Sort(operations) + return operations +} + +func (s *recordingInboundService) rejectNextAdd() { + s.mu.Lock() + defer s.mu.Unlock() + s.failAdds++ +} + +func newRecordingXray(t *testing.T) (*Xray, *recordingInboundService) { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + service := &recordingInboundService{} + server := grpc.NewServer() + command.RegisterHandlerServiceServer(server, service) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + + handler, err := api.NewXrayAPI(listener.Addr().(*net.TCPAddr).Port) + if err != nil { + t.Fatal(err) + } + t.Cleanup(handler.Close) + + cfg, err := NewConfig(recordingInbounds, nil) + if err != nil { + t.Fatal(err) + } + + return &Xray{config: cfg, handler: handler}, service +} + +func recordingUser(email string, inbounds ...string) *common.User { + return &common.User{ + Email: email, + Inbounds: inbounds, + Proxies: &common.Proxy{ + Vmess: &common.Vmess{Id: "6f1c6c5e-3f59-4a0e-9d8a-2b4f1c9d7e10"}, + Vless: &common.Vless{Id: "0b6f2a4e-6a54-4a51-9a3a-6f1e2b1f7c11", Flow: "xtls-rprx-vision"}, + Trojan: &common.Trojan{Password: "example-trojan-key"}, + Shadowsocks: &common.Shadowsocks{Password: "example-shadowsocks-key", Method: "aes-128-gcm"}, + Hysteria: &common.Hysteria{Auth: "hysteria secret"}, + }, + } +} + +func operations(pairs ...string) []string { + slices.Sort(pairs) + return pairs +} + +func replaced(tags ...string) []string { + var pairs []string + for _, tag := range tags { + pairs = append(pairs, tag+":remove", tag+":add") + } + return pairs +} + +func removed(tags ...string) []string { + var pairs []string + for _, tag := range tags { + pairs = append(pairs, tag+":remove") + } + return pairs +} + +func assertOperations(t *testing.T, got []string, want []string) { + t.Helper() + + if !slices.Equal(got, want) { + t.Fatalf("operations sent to xray:\n got %v\n want %v", got, want) + } +} + +func syncContext(t *testing.T) context.Context { + t.Helper() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + t.Cleanup(cancel) + return ctx +} + +func TestSyncUserLeavesAnUnchangedUserAlone(t *testing.T) { + x, service := newRecordingXray(t) + ctx := syncContext(t) + user := recordingUser("1.alice", recordingTags...) + + if err := x.SyncUser(ctx, user); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), operations(replaced(recordingTags...)...)) + + if err := x.SyncUser(ctx, user); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), nil) +} + +func TestSyncUserLeavesAUserLoadedAtStartupAlone(t *testing.T) { + x, service := newRecordingXray(t) + ctx := syncContext(t) + user := recordingUser("1.alice", recordingTags...) + + x.config.syncUsers([]*common.User{user}) + + if err := x.SyncUser(ctx, user); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), nil) +} + +func TestSyncUserReplacesOnlyTheAccountThatChanged(t *testing.T) { + x, service := newRecordingXray(t) + ctx := syncContext(t) + user := recordingUser("1.alice", recordingTags...) + + if err := x.SyncUser(ctx, user); err != nil { + t.Fatal(err) + } + service.take() + + user.Proxies.Trojan.Password = "rotated trojan secret" + if err := x.SyncUser(ctx, user); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), operations(replaced("trojan")...)) +} + +func TestSyncUserRemovesAUserFromTheInboundsTheyLeave(t *testing.T) { + x, service := newRecordingXray(t) + ctx := syncContext(t) + + if err := x.SyncUser(ctx, recordingUser("1.alice", recordingTags...)); err != nil { + t.Fatal(err) + } + service.take() + + if err := x.SyncUser(ctx, recordingUser("1.alice", "vless", "trojan")); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), operations(removed("vmess", "ss", "ss2022", "hysteria")...)) + + if err := x.SyncUser(ctx, recordingUser("1.alice")); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), operations(removed(recordingTags...)...)) +} + +func TestSyncUserRetriesAnAddThatFailed(t *testing.T) { + x, service := newRecordingXray(t) + ctx := syncContext(t) + user := recordingUser("1.alice", "vless") + + service.rejectNextAdd() + if err := x.SyncUser(ctx, user); err == nil { + t.Fatal("a rejected add was not reported") + } + service.take() + + if err := x.SyncUser(ctx, user); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), operations(append(replaced("vless"), removed("vmess", "trojan", "ss", "ss2022", "hysteria")...)...)) + + if err := x.SyncUser(ctx, user); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), operations(removed("vmess", "trojan", "ss", "ss2022", "hysteria")...)) +} + +func TestUpdateUsersLeavesUnchangedUsersAlone(t *testing.T) { + x, service := newRecordingXray(t) + ctx := syncContext(t) + alice := recordingUser("1.alice", "vless", "trojan") + bob := recordingUser("2.bob", "vless", "trojan") + + if err := x.UpdateUsers(ctx, []*common.User{alice, bob}); err != nil { + t.Fatal(err) + } + service.take() + + bob.Proxies.Vless.Id = "9c3e8f5a-1b2d-4c6e-8f0a-3d5b7c9e1f21" + carol := recordingUser("3.carol") + if err := x.UpdateUsers(ctx, []*common.User{alice, bob, carol}); err != nil { + t.Fatal(err) + } + + assertOperations(t, service.take(), operations( + "vless:remove", "vless:remove", "vless:add", + "trojan:remove", + "vmess:remove", "vmess:remove", "vmess:remove", + "ss:remove", "ss:remove", "ss:remove", + "ss2022:remove", "ss2022:remove", "ss2022:remove", + "hysteria:remove", "hysteria:remove", "hysteria:remove", + )) +} + +func TestUpdateUsersRetriesAnAddThatFailed(t *testing.T) { + x, service := newRecordingXray(t) + ctx := syncContext(t) + users := []*common.User{recordingUser("1.alice", "trojan")} + + service.rejectNextAdd() + if err := x.UpdateUsers(ctx, users); err == nil { + t.Fatal("a rejected add was not reported") + } + service.take() + + if err := x.UpdateUsers(ctx, users); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), operations(append(replaced("trojan"), removed("vless", "vmess", "ss", "ss2022", "hysteria")...)...)) + + if err := x.UpdateUsers(ctx, users); err != nil { + t.Fatal(err) + } + assertOperations(t, service.take(), operations(removed("vless", "vmess", "ss", "ss2022", "hysteria")...)) +} From 6e7fe4d9f82600e24fa45dc050600497765c5d29 Mon Sep 17 00:00:00 2001 From: Free-Guy-IR Date: Wed, 23 Sep 2026 04:55:24 +0330 Subject: [PATCH 2/2] test(xray): keep an unchanged user's open connection across syncs Runs the real core with one VLESS inbound: an open connection must keep flowing after SyncUser and UpdateUsers resend the same user, and a removed user must not open a new connection. The test also logs how much the removed user's already open connection still received. --- backend/xray/user_session_test.go | 176 ++++++++++++++++++++++++++++++ 1 file changed, 176 insertions(+) create mode 100644 backend/xray/user_session_test.go diff --git a/backend/xray/user_session_test.go b/backend/xray/user_session_test.go new file mode 100644 index 00000000..7222687a --- /dev/null +++ b/backend/xray/user_session_test.go @@ -0,0 +1,176 @@ +package xray + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "os" + "strconv" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/pasarguard/node/common" + "github.com/pasarguard/node/config" + "github.com/pasarguard/node/pkg/netutil" +) + +const sessionTestConfig = `{ + "log": {"loglevel": "warning"}, + "inbounds": [{ + "tag": "VLESS TCP SESSION", + "listen": "127.0.0.1", + "port": %d, + "protocol": "vless", + "settings": {"clients": [], "decryption": "none"} + }], + "outbounds": [{ + "tag": "direct", + "protocol": "freedom", + "settings": {"finalRules": [{"action": "allow", "ip": ["127.0.0.1"]}]} + }] +}` + +const sessionTestInbound = "VLESS TCP SESSION" + +func startByteSource(t *testing.T) *net.TCPAddr { + t.Helper() + + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + + go func() { + chunk := make([]byte, 32*1024) + for { + conn, err := listener.Accept() + if err != nil { + return + } + go func() { + defer conn.Close() + for { + if _, err := conn.Write(chunk); err != nil { + return + } + } + }() + } + }() + + return listener.Addr().(*net.TCPAddr) +} + +func openVlessStream(t *testing.T, inboundPort int, id uuid.UUID, target *net.TCPAddr) net.Conn { + t.Helper() + + conn, err := net.DialTimeout("tcp4", net.JoinHostPort("127.0.0.1", strconv.Itoa(inboundPort)), 3*time.Second) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + + request := []byte{0} + request = append(request, id[:]...) + request = append(request, 0, 1) + request = binary.BigEndian.AppendUint16(request, uint16(target.Port)) + request = append(request, 1) + request = append(request, target.IP.To4()...) + if _, err := conn.Write(request); err != nil { + t.Fatal(err) + } + return conn +} + +func receive(conn net.Conn, window time.Duration) (int64, error) { + _ = conn.SetReadDeadline(time.Now().Add(window)) + received, err := io.Copy(io.Discard, conn) + switch { + case errors.Is(err, os.ErrDeadlineExceeded): + return received, nil + case err == nil: + return received, io.EOF + } + return received, err +} + +func requireFlowing(t *testing.T, conn net.Conn, moment string) { + t.Helper() + + received, err := receive(conn, 500*time.Millisecond) + if err != nil || received == 0 { + t.Fatalf("open connection stopped %s: received %d bytes, err %v", moment, received, err) + } +} + +func sessionUser(email string, id uuid.UUID, inbounds ...string) *common.User { + return &common.User{ + Email: email, + Inbounds: inbounds, + Proxies: &common.Proxy{Vless: &common.Vless{Id: id.String()}}, + } +} + +func TestUserSyncKeepsAnUnchangedUsersOpenConnection(t *testing.T) { + source := startByteSource(t) + inboundPort := netutil.FindFreePort() + + xrayConfig, err := NewConfig(fmt.Sprintf(sessionTestConfig, inboundPort), nil) + if err != nil { + t.Fatal(err) + } + + aliceID := uuid.New() + alice := sessionUser("1.alice", aliceID, sessionTestInbound) + + back, err := New( + context.Background(), + xrayConfig, + []*common.User{alice}, + netutil.FindFreePort(), + netutil.FindFreePort(), + &config.Config{ + XrayExecutablePath: executablePath, + XrayAssetsPath: assetsPath, + GeneratedConfigPath: configPath, + LogBufferSize: 1000, + }, + ) + if err != nil { + t.Fatal(err) + } + defer back.Shutdown() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + stream := openVlessStream(t, inboundPort, aliceID, source) + requireFlowing(t, stream, "before any sync") + + if err := back.SyncUser(ctx, alice); err != nil { + t.Fatal(err) + } + requireFlowing(t, stream, "after SyncUser sent the same user again") + + if err := back.UpdateUsers(ctx, []*common.User{alice}); err != nil { + t.Fatal(err) + } + requireFlowing(t, stream, "after UpdateUsers sent the same user again") + + if err := back.SyncUser(ctx, sessionUser("1.alice", aliceID)); err != nil { + t.Fatal(err) + } + + if received, _ := receive(openVlessStream(t, inboundPort, aliceID, source), time.Second); received != 0 { + t.Fatalf("a removed user opened a new connection and received %d bytes", received) + } + + received, err := receive(stream, time.Second) + t.Logf("the connection the removed user had already opened received %d bytes in the next second (err %v)", received, err) +}