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