From a69f2dee13937c184456347cd8c112baf8497f74 Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Sat, 31 Oct 2026 14:14:01 +0100 Subject: [PATCH] fix(node): keep a node's online and IP reports to its own clients A node's online tree and both client-IP reports were stored as sent: they could name any client of the master or another node, under any panel's GUID, dated into the future, and in any number. That marked master clients online, fed source addresses for clients the node does not serve into the IP limit and its fail2ban log, and planted rows the retention sweep never removed. Each report is now cut to the clients linked to the node's inbounds and to the node's own key and its sub-nodes' origins (never this panel's or another node's GUID); timestamps are capped at now, and one client's list from a node keeps at most its 256 newest entries. --- internal/web/job/node_traffic_sync_job.go | 2 +- internal/web/service/inbound_client_ips.go | 50 ++++++++++++++ internal/web/service/inbound_node.go | 42 +++++++++++- internal/web/service/inbound_node_ips.go | 65 +++++++++++++++++++ internal/web/service/node_shared_guid_test.go | 12 ++++ .../web/service/node_unsynced_online_test.go | 65 +++++++++++++++++++ 6 files changed, 233 insertions(+), 3 deletions(-) diff --git a/internal/web/job/node_traffic_sync_job.go b/internal/web/job/node_traffic_sync_job.go index b81938c88..2d194641a 100644 --- a/internal/web/job/node_traffic_sync_job.go +++ b/internal/web/job/node_traffic_sync_job.go @@ -452,7 +452,7 @@ func (j *NodeTrafficSyncJob) syncOne(mgr *runtime.Manager, n *model.Node, doIpSy nodeIps, err := rt.FetchAllClientIps(ipCtx) if err == nil && len(nodeIps) > 0 { - if err := j.inboundService.MergeInboundClientIps(nodeIps); err != nil { + if err := j.inboundService.MergeNodeClientIps(n.Id, nodeIps); err != nil { logger.Warningf("node traffic sync: merge client ips from %s failed: %v", n.Name, err) } } else if err != nil { diff --git a/internal/web/service/inbound_client_ips.go b/internal/web/service/inbound_client_ips.go index 31039af4c..82c3e80dc 100644 --- a/internal/web/service/inbound_client_ips.go +++ b/internal/web/service/inbound_client_ips.go @@ -307,3 +307,53 @@ func (s *InboundService) PruneStaleClientIps() error { } return pruneStaleNodeClientIps(cutoff) } + +// nodeAttachedEmails is every client linked to one of the node's inbounds here: what a +// node reports about any other client (online state, source IPs) is not its to report. +func nodeAttachedEmails(db *gorm.DB, nodeID int) (map[string]struct{}, error) { + var emails []string + err := db.Table("clients"). + Joins("JOIN client_inbounds ON client_inbounds.client_id = clients.id"). + Joins("JOIN inbounds ON inbounds.id = client_inbounds.inbound_id"). + Where("inbounds.node_id = ?", nodeID). + Distinct().Pluck("clients.email", &emails).Error + attached := make(map[string]struct{}, len(emails)) + for _, email := range emails { + attached[email] = struct{}{} + } + return attached, err +} + +// MergeNodeClientIps merges one node's flat client-IP report, cut to the node's clients. +func (s *InboundService) MergeNodeClientIps(nodeID int, rows []model.InboundClientIps) error { + attached, err := nodeAttachedEmails(database.GetDB(), nodeID) + if err != nil { + return err + } + now := time.Now().Unix() + kept := make([]model.InboundClientIps, 0, len(rows)) + for _, row := range rows { + if _, ok := attached[row.ClientEmail]; !ok { + continue + } + var entries []model.ClientIpEntry + if json.Unmarshal([]byte(row.Ips), &entries) != nil { + continue + } + clampClientIpEntries(entries, now) + row.Ips, _ = model.MarshalClientIpEntries(entries) + kept = append(kept, row) + } + if len(kept) == 0 { + return nil + } + return s.MergeInboundClientIps(kept) +} + +// clampClientIpEntries caps reported timestamps at now: a future-dated entry would never +// age out of the retention window and would keep counting toward the IP limit. +func clampClientIpEntries(entries []model.ClientIpEntry, now int64) { + for i := range entries { + entries[i].Timestamp = min(entries[i].Timestamp, now) + } +} diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 1b888a0e8..c47ef8672 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -1499,7 +1499,17 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } } + // The online report outlives this merge, so it is cut first to the node's own clients. + if len(snap.OnlineTree) > 0 || len(snap.OnlineEmails) > 0 { + attached, err := nodeAttachedEmails(db, nodeID) + if err != nil { + logger.Warningf("setRemoteTraffic: load the clients of node %d failed, dropping its online report: %v", nodeID, err) + } + keepAttachedOnlineEmails(snap, attached) + } + if process := currentXrayProcess(); process != nil { + hostedGuids := activeInboundGuidKeys(snap.Inbounds, tagToCentral, originGuidFor) tree := snap.OnlineTree switch { case len(tree) == 0 && len(snap.OnlineEmails) > 0: @@ -1514,7 +1524,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi tree = remapGuidTreeKey(tree, nodeRow.Guid, selfKey) } } - process.SetNodeOnlineTree(nodeID, tree) + process.SetNodeOnlineTree(nodeID, filterGuidTreeKeys(tree, hostedGuids)) activeTree := normalizeActiveInboundTreeTags(snap.ActiveInboundTree, tagToCentral) if guidShared && len(activeTree) > 0 { @@ -1523,7 +1533,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } } if len(activeTree) > 0 { - activeTree = filterGuidTreeKeys(activeTree, activeInboundGuidKeys(snap.Inbounds, tagToCentral, originGuidFor)) + activeTree = filterGuidTreeKeys(activeTree, hostedGuids) } process.SetNodeActiveInboundTree(nodeID, activeTree) } @@ -1700,6 +1710,34 @@ func remapGuidTreeKey(tree map[string][]string, from, to string) map[string][]st return remapped } +// keepAttachedOnlineEmails cuts a node's online report to the clients linked to its inbounds, +// once each per GUID, into fresh slices so the reported ones are not pinned. +func keepAttachedOnlineEmails(snap *runtime.TrafficSnapshot, attached map[string]struct{}) { + keep := func(emails []string) []string { + var kept []string + seen := make(map[string]struct{}) + for _, email := range emails { + if _, ok := attached[email]; !ok { + continue + } + if _, dup := seen[email]; dup { + continue + } + seen[email] = struct{}{} + kept = append(kept, email) + } + return kept + } + snap.OnlineEmails = keep(snap.OnlineEmails) + tree := make(map[string][]string, len(snap.OnlineTree)) + for guid, emails := range snap.OnlineTree { + if kept := keep(emails); len(kept) > 0 { + tree[guid] = kept + } + } + snap.OnlineTree = tree +} + func normalizeActiveInboundTreeTags(tree map[string][]string, tagToCentral map[string]*model.Inbound) map[string][]string { if len(tree) == 0 { return nil diff --git a/internal/web/service/inbound_node_ips.go b/internal/web/service/inbound_node_ips.go index 7ef267cab..76f75fd46 100644 --- a/internal/web/service/inbound_node_ips.go +++ b/internal/web/service/inbound_node_ips.go @@ -1,13 +1,16 @@ package service import ( + "cmp" "encoding/json" + "slices" "sort" "time" "github.com/mhsanaei/3x-ui/v3/internal/database" "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "gorm.io/gorm" "gorm.io/gorm/clause" ) @@ -138,6 +141,13 @@ func (s *InboundService) MergeClientIpsByGuid(node *model.Node, trees map[string } } } + if node != nil { + scoped, err := scopeNodeClientIps(database.GetDB(), node, trees) + if err != nil { + return err + } + trees = scoped + } for guid, perEmail := range trees { // Stored once per email: an oversized key would multiply into every row. if len(guid) > maxPanelGuidLen { @@ -334,3 +344,58 @@ func pruneStaleNodeClientIps(cutoff int64) error { } return nil } + +// maxReportedIpsPerClient bounds one (node, client) list; the newest entries are kept. +const maxReportedIpsPerClient = 256 + +// scopeNodeClientIps keeps only what a node may report: its own key and its sub-nodes' +// origins (never this panel's or another node's), its own clients, and no future dates. +func scopeNodeClientIps(db *gorm.DB, node *model.Node, trees map[string]map[string][]model.ClientIpEntry) (map[string]map[string][]model.ClientIpEntry, error) { + allowed := map[string]struct{}{effectiveNodeKey(node): {}} + var origins []string + if err := db.Model(&model.Inbound{}).Where("node_id = ? AND origin_node_guid <> ''", node.Id). + Distinct().Pluck("origin_node_guid", &origins).Error; err != nil { + return nil, err + } + for _, origin := range origins { + allowed[origin] = struct{}{} + } + var others []*model.Node + if err := db.Where("id <> ?", node.Id).Find(&others).Error; err != nil { + return nil, err + } + for _, other := range others { + delete(allowed, effectiveNodeKey(other)) + delete(allowed, other.Guid) + } + if panelGuid, err := (&SettingService{}).GetPanelGuid(); err == nil { + delete(allowed, panelGuid) + } + attached, err := nodeAttachedEmails(db, node.Id) + if err != nil { + return nil, err + } + now := time.Now().Unix() + scoped := make(map[string]map[string][]model.ClientIpEntry, len(trees)) + for guid, perEmail := range trees { + if _, ok := allowed[guid]; !ok { + continue + } + kept := make(map[string][]model.ClientIpEntry, len(perEmail)) + for email, entries := range perEmail { + if _, ok := attached[email]; !ok { + continue + } + clampClientIpEntries(entries, now) + if len(entries) > maxReportedIpsPerClient { + slices.SortFunc(entries, func(a, b model.ClientIpEntry) int { return cmp.Compare(b.Timestamp, a.Timestamp) }) + entries = entries[:maxReportedIpsPerClient] + } + kept[email] = entries + } + if len(kept) > 0 { + scoped[guid] = kept + } + } + return scoped, nil +} diff --git a/internal/web/service/node_shared_guid_test.go b/internal/web/service/node_shared_guid_test.go index bb77f1ceb..94d7d1a79 100644 --- a/internal/web/service/node_shared_guid_test.go +++ b/internal/web/service/node_shared_guid_test.go @@ -142,6 +142,18 @@ func TestMergeClientIpsByGuid_RemapsClonedNodeSubtree(t *testing.T) { t.Fatalf("create node: %v", err) } } + // A node reports only clients linked to its own inbounds. + nodeIb := &model.Inbound{UserId: 1, Tag: "a-in", Port: 41501, Protocol: model.VLESS, NodeID: &n1.Id, Settings: `{"clients":[]}`} + client := &model.ClientRecord{Email: "u@x", Enable: true, SubID: "s-u"} + if err := db.Create(nodeIb).Error; err != nil { + t.Fatalf("create inbound: %v", err) + } + if err := db.Create(client).Error; err != nil { + t.Fatalf("create client: %v", err) + } + if err := db.Create(&model.ClientInbound{ClientId: client.Id, InboundId: nodeIb.Id}).Error; err != nil { + t.Fatalf("link client: %v", err) + } if err := svc.MergeClientIpsByGuid(n1, map[string]map[string][]model.ClientIpEntry{ "dup": {"u@x": {{IP: "1.1.1.1", Timestamp: now}}}, diff --git a/internal/web/service/node_unsynced_online_test.go b/internal/web/service/node_unsynced_online_test.go index 7a272aa1b..486c64123 100644 --- a/internal/web/service/node_unsynced_online_test.go +++ b/internal/web/service/node_unsynced_online_test.go @@ -3,7 +3,10 @@ package service import ( "fmt" "reflect" + "strconv" + "strings" "testing" + "time" "github.com/mhsanaei/3x-ui/v3/internal/database" "github.com/mhsanaei/3x-ui/v3/internal/database/model" @@ -67,3 +70,65 @@ func TestSetRemoteTrafficFailureClearsNodeOnlineClients(t *testing.T) { t.Errorf("online by guid after a failed merge = %v, want none", got) } } + +// A node may report only the clients linked to its own inbounds, under its own GUID: an +// unscoped report marked any master client online and grew without bound. +func TestSetRemoteTrafficKeepsOnlyTheNodesOwnClientsOnline(t *testing.T) { + db := initTrafficTestDB(t) + useOnlineTestProcess(t) + svc := &InboundService{} + seedNodeRow(t, db, &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true, Guid: "node-guid"}) + createNodeInboundWithClient(t, db, 1, "keep", 41401, "kept@x") + settings := `{"clients":[{"email":"kept@x","enable":true}]}` + snap := snapshotWithClients(t, "keep", settings, xray.ClientTraffic{Email: "kept@x", Enable: true}) + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { + t.Fatalf("seed sync: %v", err) + } + snap = snapshotWithClients(t, "keep", settings, xray.ClientTraffic{Email: "kept@x", Enable: true}) + snap.OnlineTree = map[string][]string{"node-guid": {"kept@x", "master-only@x", "kept@x"}, "someone-else": {"kept@x"}} + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { + t.Fatalf("sync: %v", err) + } + if got, want := svc.GetOnlineClientsByGuid(), map[string][]string{"node-guid": {"kept@x"}}; !reflect.DeepEqual(got, want) { + t.Fatalf("online by guid = %v, want %v", got, want) + } +} + +// The client-IP reports feed the IP limit and fail2ban; a node must not plant rows for +// clients it does not serve, under other panels' GUIDs, or dated into the future. +func TestNodeClientIpReportsAreScopedToTheNode(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + node := &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true, Guid: "node-guid"} + seedNodeRow(t, db, node) + createNodeInboundWithClient(t, db, 1, "keep", 41402, "kept@x") + settings := `{"clients":[{"email":"kept@x","enable":true}]}` + if _, err := svc.setRemoteTrafficLocked(1, snapshotWithClients(t, "keep", settings, xray.ClientTraffic{Email: "kept@x", Enable: true}), false, false); err != nil { + t.Fatalf("seed sync: %v", err) + } + future := time.Now().Add(240 * time.Hour).Unix() + entries := func(ip string) string { return fmt.Sprintf(`[{"ip":%q,"timestamp":%d}]`, ip, future) } + if err := svc.MergeNodeClientIps(1, []model.InboundClientIps{ + {ClientEmail: "kept@x", Ips: entries("198.51.100.1")}, + {ClientEmail: "master-only@x", Ips: entries("198.51.100.2")}, + }); err != nil { + t.Fatalf("MergeNodeClientIps: %v", err) + } + var flat []model.InboundClientIps + db.Find(&flat) + if len(flat) != 1 || flat[0].ClientEmail != "kept@x" || strings.Contains(flat[0].Ips, strconv.FormatInt(future, 10)) { + t.Fatalf("stored client IPs = %+v, want only kept@x, dated no later than now", flat) + } + tree := map[string]map[string][]model.ClientIpEntry{ + "node-guid": {"kept@x": {{IP: "198.51.100.3", Timestamp: future}}, "master-only@x": {{IP: "198.51.100.4", Timestamp: future}}}, + "someone-else": {"kept@x": {{IP: "198.51.100.5", Timestamp: future}}}, + } + if err := svc.MergeClientIpsByGuid(node, tree); err != nil { + t.Fatalf("MergeClientIpsByGuid: %v", err) + } + var rows []model.NodeClientIp + db.Find(&rows) + if len(rows) != 1 || rows[0].NodeGuid != "node-guid" || rows[0].Email != "kept@x" || strings.Contains(rows[0].Ips, strconv.FormatInt(future, 10)) { + t.Fatalf("stored per-node IPs = %+v, want only kept@x under node-guid, dated no later than now", rows) + } +}