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

@@ -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 {

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)
+	}
+}

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

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
+}

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}}},

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)
+	}
+}