diff --git a/internal/database/db.go b/internal/database/db.go index f051e44cd..a8a570ab9 100644 --- a/internal/database/db.go +++ b/internal/database/db.go @@ -405,7 +405,10 @@ func migrateSyncOrphanColumns() error { if !db.Migrator().HasColumn(&model.ClientRecord{}, "sync_orphaned_at") { return nil } - return db.Exec("UPDATE clients SET sync_orphaned_at = 0 WHERE sync_orphaned_at IS NULL").Error + if err := db.Exec("UPDATE clients SET sync_orphaned_at = 0 WHERE sync_orphaned_at IS NULL").Error; err != nil { + return err + } + return db.Exec("UPDATE clients SET sync_orphaned_by = 0 WHERE sync_orphaned_by IS NULL").Error } // The client identity checks match emails case-insensitively; without an diff --git a/internal/database/model/model.go b/internal/database/model/model.go index 2183bde85..284d6d08a 100644 --- a/internal/database/model/model.go +++ b/internal/database/model/model.go @@ -1018,6 +1018,8 @@ type ClientRecord struct { // Owned solely by the node-snapshot sweep, which soft-orphans instead of // deleting; orphans from any other cause stay at zero and are never reaped. SyncOrphanedAt int64 `json:"-" gorm:"column:sync_orphaned_at;default:0"` + // The node whose merge orphaned the record: only that node may take it back. + SyncOrphanedBy int `json:"-" gorm:"column:sync_orphaned_by;default:0"` } func (ClientRecord) TableName() string { return "clients" } diff --git a/internal/web/service/client_group_node_sync_test.go b/internal/web/service/client_group_node_sync_test.go index 28fe18f3e..942d64cad 100644 --- a/internal/web/service/client_group_node_sync_test.go +++ b/internal/web/service/client_group_node_sync_test.go @@ -38,14 +38,12 @@ func TestSetRemoteTraffic_PreservesPanelLocalGroupAndComment(t *testing.T) { t.Fatalf("create node inbound: %v", err) } - if err := db.Create(&model.ClientRecord{ - Email: email, - UUID: uid, - Enable: true, - Group: wantGroup, - Comment: wantComment, - }).Error; err != nil { - t.Fatalf("create client record: %v", err) + centralClients, err := (&InboundService{}).GetClients(central) + if err != nil { + t.Fatalf("parse central clients: %v", err) + } + if err := (&ClientService{}).SyncInbound(db, central.Id, centralClients); err != nil { + t.Fatalf("link client record: %v", err) } snap := &runtime.TrafficSnapshot{ diff --git a/internal/web/service/client_link.go b/internal/web/service/client_link.go index 3a744dd7b..912408bb2 100644 --- a/internal/web/service/client_link.go +++ b/internal/web/service/client_link.go @@ -113,7 +113,7 @@ func (s *ClientService) syncInboundClients(tx *gorm.DB, inboundId int, clients [ } var inbound model.Inbound - if err := tx.Select("protocol").Where("id = ?", inboundId).Take(&inbound).Error; err != nil { + if err := tx.Select("protocol", "node_id").Where("id = ?", inboundId).Take(&inbound).Error; err != nil { return err } if err := s.validateProtocolClients(tx, inbound.Protocol, inboundId, clients, detachEmails, prune); err != nil { @@ -156,7 +156,7 @@ func (s *ClientService) syncInboundClients(tx *gorm.DB, inboundId int, clients [ ids = append(ids, row.Id) } var err error - if onLowerTunnel, err = clientsOnLowerTunnel(tx, inboundId, ids); err != nil { + if onLowerTunnel, err = clientsOnLowerTunnel(tx, inboundId, inbound.NodeID, ids); err != nil { return err } } @@ -408,16 +408,21 @@ func keepLinkOwnedFields(row, before *model.ClientRecord, ownsTunnelPeer bool) { row.KeepAlive = before.KeepAlive } -// clientsOnLowerTunnel returns the clients also attached to a WireGuard/AmneziaWG -// inbound with a lower id than inboundId: that tunnel owns their record's peer fields. -func clientsOnLowerTunnel(tx *gorm.DB, inboundId int, clientIds []int) (map[int]struct{}, error) { +// clientsOnLowerTunnel returns the clients also attached to a higher-ranked WireGuard/AmneziaWG +// inbound, which owns their record's peer fields: a master-local one first, then the lowest id. +func clientsOnLowerTunnel(tx *gorm.DB, inboundId int, nodeID *int, clientIds []int) (map[int]struct{}, error) { + outranks := "inbounds.node_id IS NULL AND client_inbounds.inbound_id < ?" + if nodeID != nil { + outranks = "(inbounds.node_id IS NULL OR client_inbounds.inbound_id < ?)" + } out := make(map[int]struct{}) for _, batch := range chunkInts(clientIds, sqlInChunk) { var ids []int if err := tx.Table("client_inbounds"). Joins("JOIN inbounds ON inbounds.id = client_inbounds.inbound_id"). - Where("client_inbounds.client_id IN ? AND client_inbounds.inbound_id < ? AND inbounds.protocol IN ?", - batch, inboundId, []model.Protocol{model.WireGuard, model.AmneziaWG}). + Where("client_inbounds.client_id IN ? AND inbounds.protocol IN ?", + batch, []model.Protocol{model.WireGuard, model.AmneziaWG}). + Where(outranks, inboundId). Distinct().Pluck("client_inbounds.client_id", &ids).Error; err != nil { return nil, err } diff --git a/internal/web/service/client_locks.go b/internal/web/service/client_locks.go index faf5976fe..00ac0cc1c 100644 --- a/internal/web/service/client_locks.go +++ b/internal/web/service/client_locks.go @@ -173,6 +173,7 @@ func isClientEmailTombstoned(email string) bool { // nest any value deeply, so entries stay raw text and only these fields are decoded. type settingsClientEntry struct { Email string `json:"email"` + SubID json.RawMessage `json:"subId"` ExpiryTime json.RawMessage `json:"expiryTime"` Enable json.RawMessage `json:"enable"` } diff --git a/internal/web/service/client_sync_orphan.go b/internal/web/service/client_sync_orphan.go index 17b780ff8..4aa8d4275 100644 --- a/internal/web/service/client_sync_orphan.go +++ b/internal/web/service/client_sync_orphan.go @@ -25,13 +25,13 @@ func clientRecordExists(tx *gorm.DB, email string) bool { return n > 0 } -func markSyncOrphan(tx *gorm.DB, email string, nowMs int64) error { +func markSyncOrphan(tx *gorm.DB, email string, nodeID int, nowMs int64) error { if email == "" { return nil } return tx.Model(&model.ClientRecord{}). Where("email = ? AND sync_orphaned_at = 0", email). - Update("sync_orphaned_at", nowMs).Error + Updates(map[string]any{"sync_orphaned_at": nowMs, "sync_orphaned_by": nodeID}).Error } // A client that is attached again is not orphaned, whatever an earlier merge @@ -39,7 +39,7 @@ func markSyncOrphan(tx *gorm.DB, email string, nowMs int64) error { func clearSyncOrphanMarks(tx *gorm.DB) error { return tx.Model(&model.ClientRecord{}). Where("sync_orphaned_at > 0 AND EXISTS (SELECT 1 FROM client_inbounds WHERE client_inbounds.client_id = clients.id)"). - Update("sync_orphaned_at", 0).Error + Updates(map[string]any{"sync_orphaned_at": 0, "sync_orphaned_by": 0}).Error } // ReapSyncOrphans deletes the clients the node-snapshot sweep marked and no diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index c47ef8672..a17787e21 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -23,8 +23,6 @@ import ( var reportedRemoteTagConflict sync.Map -var reportedForeignClientClaim sync.Map - // nodeBulkPushThreshold caps how many per-client RPCs a single operation will // stream to a remote node. Above it, the panel marks the node dirty instead and // lets one ReconcileNode push converge the whole inbound — far cheaper than M @@ -494,46 +492,6 @@ func snapshotAttachedEmails(settings string) (map[string]struct{}, bool) { return emails, true } -// clientEmailsOwnedElsewhere returns the emails attached only to inbounds of -// other nodes: email is unique, so adopting one would overwrite a client this -// node does not serve. Attached nowhere means soft-orphaned, hence adoptable. -func clientEmailsOwnedElsewhere(tx *gorm.DB, nodeID int, emails []string) (map[string]struct{}, error) { - attachedEmails := func(nodeScoped bool) ([]string, error) { - q := tx.Table("clients"). - Joins("JOIN client_inbounds ON client_inbounds.client_id = clients.id"). - Joins("JOIN inbounds ON inbounds.id = client_inbounds.inbound_id"). - Where("clients.email IN ?", emails) - if nodeScoped { - q = q.Where("inbounds.node_id = ?", nodeID) - } - var rows []string - err := q.Pluck("clients.email", &rows).Error - return rows, err - } - attached, err := attachedEmails(false) - if err != nil { - return nil, err - } - if len(attached) == 0 { - return nil, nil - } - owned, err := attachedEmails(true) - if err != nil { - return nil, err - } - ownedSet := make(map[string]struct{}, len(owned)) - for _, email := range owned { - ownedSet[email] = struct{}{} - } - foreign := make(map[string]struct{}) - for _, email := range attached { - if _, ok := ownedSet[email]; !ok { - foreign[email] = struct{}{} - } - } - return foreign, nil -} - func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.TrafficSnapshot, dirty, justPushed bool) (bool, error) { if snap == nil || nodeID <= 0 { return false, nil @@ -704,8 +662,23 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } }() + claims, err := loadNodeClientClaims(tx, nodeID, snap) + if err != nil { + return false, err + } + // The raw blob stays for the reconcile fingerprint, so a rewrite leaves a mismatch to + // re-push the master's values against. + rawSettings := make(map[*model.Inbound]string, len(snap.Inbounds)) + for _, snapIb := range snap.Inbounds { + if snapIb != nil { + rawSettings[snapIb] = snapIb.Settings + snapIb.Settings = claims.apply(snapIb.Settings) + } + } + structuralChange := false lifecycleLifted := false + claimsRewritten := false var sweptTags []string var adoptedInbounds []*model.Inbound @@ -713,6 +686,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi central *model.Inbound snapIb *model.Inbound wireSettings string + fingerprint string } var pendingAdopts []pendingAdopt degradedInbounds := map[int]string{} @@ -870,6 +844,11 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi if deduped, changed := dedupeSettingsClients(adoptedSettings); changed { adoptedSettings = deduped } + fingerprint := adoptedSettings + if raw := rawSettings[snapIb]; raw != snapIb.Settings { + fingerprint, _ = stripTombstonedClients(raw) + fingerprint, _ = dedupeSettingsClients(fingerprint) + } updates := map[string]any{} if !dirty && snapshotDropsEveryHubClient(tx, c.Id, adoptedSettings) { degradedInbounds[c.Id] = c.Tag @@ -877,7 +856,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi // Defer lifecycle lift until after client_traffics absorbs this tick's // deltas so quota stale-disable matches SQL (#6228). pendingAdopts = append(pendingAdopts, pendingAdopt{ - central: c, snapIb: snapIb, wireSettings: adoptedSettings, + central: c, snapIb: snapIb, wireSettings: adoptedSettings, fingerprint: fingerprint, }) updates["enable"] = snapIb.Enable updates["remark"] = snapIb.Remark @@ -1287,10 +1266,12 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi // so the adopted rows are served this tick, not after the TTL. InvalidateSubData() } - // The fingerprint stamps the un-lifted wire blob on purpose: a lift must - // leave reconcile a mismatch to re-push against. - if liftChanged || adoptedWireChanged(p.central, p.snapIb, p.wireSettings) { - adoptedInbounds = append(adoptedInbounds, adoptedWireInbound(p.central, p.snapIb, p.wireSettings)) + rewritten := p.fingerprint != p.wireSettings + claimsRewritten = claimsRewritten || rewritten + // The fingerprint stamps the node's own blob on purpose: a lift or a refused claim + // must leave reconcile a mismatch to re-push against. + if liftChanged || rewritten || adoptedWireChanged(p.central, p.snapIb, p.fingerprint) { + adoptedInbounds = append(adoptedInbounds, adoptedWireInbound(p.central, p.snapIb, p.fingerprint)) } } for _, snapIb := range snap.Inbounds { @@ -1374,29 +1355,6 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } } } - if len(localEmails) > 0 { - foreign, err := clientEmailsOwnedElsewhere(tx, nodeID, localEmails) - if err != nil { - return false, err - } - if len(foreign) > 0 { - kept := filtered[:0] - for i := range filtered { - if _, claimed := foreign[filtered[i].Email]; !claimed { - kept = append(kept, filtered[i]) - continue - } - key := fmt.Sprintf("%d:%s", nodeID, filtered[i].Email) - if _, seen := reportedForeignClientClaim.LoadOrStore(key, struct{}{}); !seen { - logger.Warningf( - "setRemoteTraffic: node %d reported client %q, which is attached only to inbounds of another node — not adopting (rename one side to remove the duplicate email)", - nodeID, filtered[i].Email, - ) - } - } - filtered = kept - } - } if err := s.clientService.SyncInbound(tx, c.Id, filtered); err != nil { logger.Warningf("setRemoteTraffic: sync clients for tag %q failed: %v", snapIb.Tag, err) syncFailedInbounds[c.Id] = struct{}{} @@ -1436,7 +1394,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } // "Ended the merge unattached" is true for a real remote deletion and // equally true for a bad merge, so record a strike instead of deleting. - if err := markSyncOrphan(tx, email, now); err != nil { + if err := markSyncOrphan(tx, email, nodeID, now); err != nil { logger.Warningf("setRemoteTraffic: mark orphan %q failed: %v", email, err) continue } @@ -1472,14 +1430,18 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } } - if lifecycleLifted && !dirty { + if (lifecycleLifted || claimsRewritten) && !dirty { + reason := "lifecycle lift" + if !lifecycleLifted { + reason = "client claim rewrite" + } var already model.Node if err := database.GetDB().Select("config_dirty").Where("id = ?", nodeID).First(&already).Error; err == nil && already.ConfigDirty { - logger.Debugf("setRemoteTraffic: node %d lifecycle lift; already dirty", nodeID) + logger.Debugf("setRemoteTraffic: node %d %s; already dirty", nodeID, reason) } else { - logger.Infof("setRemoteTraffic: node %d lifecycle lift; marking dirty for re-push", nodeID) + logger.Infof("setRemoteTraffic: node %d %s; marking dirty for re-push", nodeID, reason) if err := (&NodeService{}).MarkNodeDirty(nodeID); err != nil { - logger.Warningf("setRemoteTraffic: mark node %d dirty after lifecycle lift failed: %v", nodeID, err) + logger.Warningf("setRemoteTraffic: mark node %d dirty after %s failed: %v", nodeID, reason, err) } } } diff --git a/internal/web/service/inbound_node_claims.go b/internal/web/service/inbound_node_claims.go new file mode 100644 index 000000000..b54ff91c5 --- /dev/null +++ b/internal/web/service/inbound_node_claims.go @@ -0,0 +1,275 @@ +package service + +import ( + "encoding/json" + "fmt" + "reflect" + "strings" + "sync" + + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/logger" + "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" + + "gorm.io/gorm" +) + +var reportedRefusedClientClaim sync.Map + +// nodeClientClaims is what one node snapshot may do to the master's client records: a node +// changes only records linked solely to its own inbounds, plus the new ones it creates. +type nodeClientClaims struct { + refused map[string]string + // shared records are also served by the master or another node, so they keep the + // master's values except nodeKeptSharedKeys. + shared map[string]*model.ClientRecord + // storedSubID keeps an own record's subId when the node reports one held by a + // client it does not host. + storedSubID map[string]string +} + +// nodeKeptSharedKeys stay as the node reported them: per-link values, and the lifecycle +// the merge already settles against the master's traffic row. +var nodeKeptSharedKeys = []string{"flow", "privateKey", "publicKey", "allowedIPs", "preSharedKey", "keepAlive", "enable", "expiryTime"} + +type clientLinkScope struct{ here, elsewhere bool } + +func loadNodeClientClaims(tx *gorm.DB, nodeID int, snap *runtime.TrafficSnapshot) (*nodeClientClaims, error) { + claims := &nodeClientClaims{ + refused: map[string]string{}, + shared: map[string]*model.ClientRecord{}, + storedSubID: map[string]string{}, + } + subIDByEmail := map[string]string{} + for _, snapIb := range snap.Inbounds { + if snapIb == nil { + continue + } + _, entries, ok := splitSettingsClients(snapIb.Settings) + if !ok { + continue + } + for _, entry := range entries { + c, ok := readSettingsClient(entry) + if !ok || c.Email == "" { + continue + } + if c.Email != strings.TrimSpace(c.Email) { + claims.refused[c.Email] = "its email carries surrounding white space" + continue + } + if _, seen := subIDByEmail[c.Email]; !seen { + var subID string + _ = json.Unmarshal(c.SubID, &subID) + subIDByEmail[c.Email] = subID + } + } + } + if len(subIDByEmail) == 0 { + return claims, nil + } + + lowered := make([]string, 0, len(subIDByEmail)) + for email := range subIDByEmail { + lowered = append(lowered, strings.ToLower(email)) + } + var records []model.ClientRecord + for _, batch := range chunkStrings(uniqueNonEmptyStrings(lowered), sqlInChunk) { + var rows []model.ClientRecord + if err := tx.Where("LOWER(email) IN ?", batch).Find(&rows).Error; err != nil { + return nil, err + } + records = append(records, rows...) + } + byEmail := make(map[string]*model.ClientRecord, len(records)) + byLower := make(map[string]string, len(records)) + for i := range records { + byEmail[records[i].Email] = &records[i] + byLower[strings.ToLower(records[i].Email)] = records[i].Email + } + + subIDs := make([]string, 0, len(subIDByEmail)) + for _, subID := range subIDByEmail { + if subID != "" { + subIDs = append(subIDs, subID) + } + } + var holders []model.ClientRecord + for _, batch := range chunkStrings(uniqueNonEmptyStrings(subIDs), sqlInChunk) { + var rows []model.ClientRecord + if err := tx.Select("id", "email", "sub_id").Where("sub_id IN ?", batch).Find(&rows).Error; err != nil { + return nil, err + } + holders = append(holders, rows...) + } + + ids := make([]int, 0, len(records)+len(holders)) + for i := range records { + ids = append(ids, records[i].Id) + } + for i := range holders { + ids = append(ids, holders[i].Id) + } + scopes, err := clientLinkScopes(tx, nodeID, ids) + if err != nil { + return nil, err + } + holdersBySub := make(map[string][]model.ClientRecord, len(holders)) + for _, h := range holders { + holdersBySub[h.SubID] = append(holdersBySub[h.SubID], h) + } + subHeldElsewhere := func(email, subID string) bool { + for _, h := range holdersBySub[subID] { + if !strings.EqualFold(h.Email, email) && scopes[h.Id] != (clientLinkScope{here: true}) { + return true + } + } + return false + } + + for email, subID := range subIDByEmail { + rec, exact := byEmail[email] + if !exact { + if _, variant := byLower[strings.ToLower(email)]; variant { + claims.refused[email] = "another client's email differs from it only in case" + } else if subID != "" && subHeldElsewhere(email, subID) { + claims.refused[email] = "its subId belongs to a client this node does not host" + } + continue + } + scope := scopes[rec.Id] + switch { + case scope.elsewhere && scope.here: + claims.shared[email] = rec + case scope.elsewhere: + claims.refused[email] = "it is attached only to inbounds this node does not host" + case !scope.here && (rec.SyncOrphanedAt == 0 || rec.SyncOrphanedBy != nodeID): + claims.refused[email] = "it is detached from every inbound on the master" + case subID != rec.SubID && subID != "" && subHeldElsewhere(email, subID): + claims.storedSubID[email] = rec.SubID + } + } + for email, reason := range claims.refused { + key := fmt.Sprintf("%d:%s", nodeID, email) + if _, seen := reportedRefusedClientClaim.LoadOrStore(key, struct{}{}); !seen { + logger.Warningf("setRemoteTraffic: node %d reported client %q, not adopting: %s", nodeID, email, reason) + } + } + return claims, nil +} + +// clientLinkScopes reports, per client id, whether it is linked to an inbound of this node +// and whether to any inbound outside it (the master's own or another node's). +func clientLinkScopes(tx *gorm.DB, nodeID int, ids []int) (map[int]clientLinkScope, error) { + scopes := make(map[int]clientLinkScope, len(ids)) + for _, batch := range chunkInts(uniqueInts(ids), sqlInChunk) { + var rows []struct { + ClientID int + NodeID *int + } + if err := tx.Table("client_inbounds"). + Select("client_inbounds.client_id AS client_id, inbounds.node_id AS node_id"). + Joins("JOIN inbounds ON inbounds.id = client_inbounds.inbound_id"). + Where("client_inbounds.client_id IN ?", batch). + Scan(&rows).Error; err != nil { + return nil, err + } + for _, r := range rows { + s := scopes[r.ClientID] + if r.NodeID != nil && *r.NodeID == nodeID { + s.here = true + } else { + s.elsewhere = true + } + scopes[r.ClientID] = s + } + } + return scopes, nil +} + +// apply rewrites a snapshot inbound's settings to what this node may claim. +func (c *nodeClientClaims) apply(settings string) string { + if len(c.refused) == 0 && len(c.shared) == 0 && len(c.storedSubID) == 0 { + return settings + } + members, entries, ok := splitSettingsClients(settings) + if !ok { + return settings + } + changed := false + kept := make([]json.RawMessage, 0, len(entries)) + for _, entry := range entries { + sc, ok := readSettingsClient(entry) + if !ok || sc.Email == "" { + kept = append(kept, entry) + continue + } + if _, refused := c.refused[sc.Email]; refused { + changed = true + continue + } + var override map[string]json.RawMessage + if rec, shared := c.shared[sc.Email]; shared { + override = sharedClientEntry(rec, entry) + } else if subID, keep := c.storedSubID[sc.Email]; keep { + if json.Unmarshal(entry, &override) == nil { + override["subId"], _ = json.Marshal(subID) + } + } + if override == nil { + kept = append(kept, entry) + continue + } + rewritten, err := json.Marshal(override) + if err != nil { + kept = append(kept, entry) + continue + } + changed = true + kept = append(kept, rewritten) + } + if !changed { + return settings + } + out, _ := joinSettingsClients(settings, members, kept) + return out +} + +// sharedClientEntry is nil when the node's entry already agrees with the record on every +// master-owned value, so an honest node's differing timestamps cause no re-push. +func sharedClientEntry(rec *model.ClientRecord, nodeEntry json.RawMessage) map[string]json.RawMessage { + var reported model.Client + if json.Unmarshal(nodeEntry, &reported) != nil { + return nil + } + if !sharedRecordWouldChange(rec, &reported) { + return nil + } + var fromRecord, fromNode map[string]json.RawMessage + raw, err := json.Marshal(rec.ToClient()) + if err != nil || json.Unmarshal(raw, &fromRecord) != nil || json.Unmarshal(nodeEntry, &fromNode) != nil { + return nil + } + for _, key := range nodeKeptSharedKeys { + if v, ok := fromNode[key]; ok { + fromRecord[key] = v + } else { + delete(fromRecord, key) + } + } + return fromRecord +} + +// sharedRecordWouldChange merges the reported client onto a copy of the record the way +// SyncInbound does, leaving out what the node may set and what the master restores anyway. +func sharedRecordWouldChange(rec *model.ClientRecord, reported *model.Client) bool { + incoming := reported.ToRecord() + incoming.Enable = rec.Enable + incoming.ExpiryTime = rec.ExpiryTime + incoming.Comment = rec.Comment + incoming.CreatedAt = 0 + merged := *rec + applyClientRecordMerge(&merged, incoming) + keepLinkOwnedFields(&merged, rec, false) + return !reflect.DeepEqual(merged, *rec) +} diff --git a/internal/web/service/node_client_expiry_sync_test.go b/internal/web/service/node_client_expiry_sync_test.go index bf88b14fc..1398a6537 100644 --- a/internal/web/service/node_client_expiry_sync_test.go +++ b/internal/web/service/node_client_expiry_sync_test.go @@ -1030,7 +1030,7 @@ func TestNodeActivationLiftsClientRecordExpiry(t *testing.T) { const activated = int64(1798448344010) negSettings := `{"clients":[{"email":"delayed","enable":true,"expiryTime":-2592000000}]}` - if err := db.Create(&model.ClientRecord{Email: email, Enable: true, ExpiryTime: duration}).Error; err != nil { + if err := db.Model(&model.ClientRecord{}).Where("email = ?", email).Update("expiry_time", duration).Error; err != nil { t.Fatalf("seed client record: %v", err) } diff --git a/internal/web/service/node_client_traffic_sum_test.go b/internal/web/service/node_client_traffic_sum_test.go index dece8e23b..dd8e01fd2 100644 --- a/internal/web/service/node_client_traffic_sum_test.go +++ b/internal/web/service/node_client_traffic_sum_test.go @@ -38,7 +38,7 @@ func createNodeInbound(t *testing.T, db *gorm.DB, nodeID int, tag string, port i } // createNodeInboundWithClient mirrors createNodeInbound but stores the client -// in the settings JSON, which the node sync turns into a client_inbounds link. +// in the settings JSON and links it, as attaching it on the master does. func createNodeInboundWithClient(t *testing.T, db *gorm.DB, nodeID int, tag string, port int, email string) { t.Helper() nid := nodeID @@ -47,6 +47,9 @@ func createNodeInboundWithClient(t *testing.T, db *gorm.DB, nodeID int, tag stri if err := db.Create(ib).Error; err != nil { t.Fatalf("create node inbound %q: %v", tag, err) } + if err := (&ClientService{}).SyncInbound(db, ib.Id, []model.Client{{Email: email, Enable: true}}); err != nil { + t.Fatalf("link %q to %q: %v", email, tag, err) + } } func syncNode(t *testing.T, svc *InboundService, nodeID int, tag string, stats ...xray.ClientTraffic) { diff --git a/internal/web/service/node_foreign_client_test.go b/internal/web/service/node_foreign_client_test.go index ee7c09bee..6f9d71c5e 100644 --- a/internal/web/service/node_foreign_client_test.go +++ b/internal/web/service/node_foreign_client_test.go @@ -2,9 +2,11 @@ package service import ( "fmt" + "strings" "testing" "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" "github.com/mhsanaei/3x-ui/v3/internal/xray" "gorm.io/gorm" @@ -88,3 +90,279 @@ func nodeInboundID(t *testing.T, db *gorm.DB, tag string) int { } return ib.Id } + +func seedMasterInbound(t *testing.T, db *gorm.DB, tag string, port int, clientsJSON string) *model.Inbound { + t.Helper() + ib := &model.Inbound{ + UserId: 1, Tag: tag, Enable: true, Port: port, Protocol: model.VLESS, + Settings: `{"clients":[` + clientsJSON + `]}`, + } + if err := db.Create(ib).Error; err != nil { + t.Fatalf("create inbound %q: %v", tag, err) + } + clients, err := (&InboundService{}).GetClients(ib) + if err != nil { + t.Fatalf("parse clients of %q: %v", tag, err) + } + if err := (&ClientService{}).SyncInbound(db, ib.Id, clients); err != nil { + t.Fatalf("attach clients of %q: %v", tag, err) + } + return ib +} + +func readClientRecord(t *testing.T, db *gorm.DB, email string) model.ClientRecord { + t.Helper() + var row model.ClientRecord + if err := db.Where("email = ?", email).First(&row).Error; err != nil { + t.Fatalf("read client %q: %v", email, err) + } + return row +} + +func countClientsWhere(t *testing.T, db *gorm.DB, query string, args ...any) int64 { + t.Helper() + var n int64 + if err := db.Model(&model.ClientRecord{}).Where(query, args...).Count(&n).Error; err != nil { + t.Fatalf("count clients: %v", err) + } + return n +} + +func storedInboundSettings(t *testing.T, db *gorm.DB, tag string) string { + t.Helper() + var ib model.Inbound + if err := db.Where("tag = ?", tag).First(&ib).Error; err != nil { + t.Fatalf("read inbound %q: %v", tag, err) + } + return ib.Settings +} + +func nodeHostsClient(t *testing.T, db *gorm.DB, tag, email string) bool { + t.Helper() + attached, err := (&ClientService{}).ListForInbound(db, nodeInboundID(t, db, tag)) + if err != nil { + t.Fatalf("list clients of %q: %v", tag, err) + } + for _, c := range attached { + if c.Email == email { + return true + } + } + return false +} + +// A client shared with a master inbound is rendered into the master's own Xray, so the +// node's copy must not move its credential, reverse tag or subscription id. +func TestNodeCannotRewriteClientSharedWithMaster(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + seedNodeRow(t, db, &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true}) + + const legitUUID = "11111111-1111-4111-8111-111111111111" + shared := fmt.Sprintf(`{"email":"shared@x","id":%q,"subId":"shared-sub","enable":true}`, legitUUID) + seedMasterInbound(t, db, "master-in", 40001, shared) + nid := 1 + nodeIb := &model.Inbound{ + UserId: 1, Tag: "n1-in", Enable: true, Port: 41001, Protocol: model.VLESS, NodeID: &nid, + Settings: `{"clients":[` + shared + `]}`, + } + if err := db.Create(nodeIb).Error; err != nil { + t.Fatalf("create node inbound: %v", err) + } + clients, _ := svc.GetClients(nodeIb) + if err := (&ClientService{}).SyncInbound(db, nodeIb.Id, clients); err != nil { + t.Fatalf("attach node clients: %v", err) + } + + syncNodeWithSettings(t, svc, 1, "n1-in", + `{"clients":[{"email":"shared@x","id":"zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz","subId":"picked-by-node","reverse":{"tag":"direct"},"enable":true}]}`, + xray.ClientTraffic{Email: "shared@x", Enable: true}) + + row := readClientRecord(t, db, "shared@x") + if row.UUID != legitUUID || row.SubID != "shared-sub" || row.Reverse != "" { + t.Fatalf("node rewrote a shared client: uuid=%q subId=%q reverse=%q", row.UUID, row.SubID, row.Reverse) + } + stored := storedInboundSettings(t, db, "n1-in") + if strings.Contains(stored, "zzzz") || strings.Contains(stored, "picked-by-node") || strings.Contains(stored, "direct") { + t.Fatalf("node inbound stored the node's values for a shared client: %s", stored) + } + if !nodeHostsClient(t, db, "n1-in", "shared@x") { + t.Fatal("the shared client lost its node link") + } +} + +// A refused claim must leave no trace: the stored settings are re-synced later by +// auto-renew, and its stats would land on the victim's traffic row. +func TestNodeRefusedClaimIsNotStoredOrCounted(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + seedNodeRow(t, db, &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true}) + master := seedMasterInbound(t, db, "master-in", 40001, `{"email":"victim@x","id":"11111111-1111-4111-8111-111111111111","subId":"victim-sub","enable":true}`) + if err := db.Create(&xray.ClientTraffic{InboundId: master.Id, Email: "victim@x", Enable: true}).Error; err != nil { + t.Fatalf("seed victim traffic: %v", err) + } + createNodeInbound(t, db, 1, "n1-in", 41001) + + hostile := `{"clients":[{"email":"victim@x","id":"","subId":"picked-by-node","enable":true},{"email":"victim@x ","subId":"padded","enable":true},{"email":"VICTIM@X","subId":"cased","enable":true},{"email":"own@x","enable":true}]}` + for _, up := range []int64{100, 600} { + syncNodeWithSettings(t, svc, 1, "n1-in", hostile, + xray.ClientTraffic{Email: "victim@x", Up: up, Enable: true}, + xray.ClientTraffic{Email: "own@x", Enable: true}) + } + + if stored := storedInboundSettings(t, db, "n1-in"); strings.Contains(strings.ToLower(stored), "victim") { + t.Fatalf("refused claim stored on the node inbound: %s", stored) + } + if row := readClientRecord(t, db, "victim@x"); row.SubID != "victim-sub" { + t.Fatalf("victim subId = %q, want victim-sub", row.SubID) + } + if n := countClientsWhere(t, db, "LOWER(email) LIKE ?", "victim%"); n != 1 { + t.Fatalf("a padded or case-variant claim created a look-alike record: %d victim rows", n) + } + var counted int64 + db.Model(xray.ClientTraffic{}).Where("LOWER(email) LIKE ? AND up > 0", "victim%").Count(&counted) + if counted != 0 { + t.Fatalf("refused claim's stats reached a victim traffic row: %d rows", counted) + } + if !nodeHostsClient(t, db, "n1-in", "own@x") { + t.Fatal("the node's own client was not adopted") + } +} + +// A record with no inbound is the admin's (detached, or its inbound deleted), not a +// node's; only a node whose own merge orphaned it may take it back. +func TestNodeCannotAdoptDetachedClient(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + seedNodeRow(t, db, &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true}) + seedNodeRow(t, db, &model.Node{Id: 2, Name: "n2", Address: "127.0.0.2", Port: 2096, ApiToken: "tok2", Enable: true}) + master := seedMasterInbound(t, db, "master-in", 40001, + `{"email":"detached@x","id":"11111111-1111-4111-8111-111111111111","subId":"d-sub","enable":true},`+ + `{"email":"mine@x","id":"22222222-2222-4222-8222-222222222222","enable":true},`+ + `{"email":"theirs@x","id":"33333333-3333-4333-8333-333333333333","enable":true}`) + if err := db.Exec("DELETE FROM client_inbounds WHERE inbound_id = ?", master.Id).Error; err != nil { + t.Fatalf("detach: %v", err) + } + for email, node := range map[string]int{"mine@x": 1, "theirs@x": 2} { + if err := db.Model(&model.ClientRecord{}).Where("email = ?", email). + Updates(map[string]any{"sync_orphaned_at": 1, "sync_orphaned_by": node}).Error; err != nil { + t.Fatalf("mark %s orphaned: %v", email, err) + } + } + createNodeInbound(t, db, 1, "n1-in", 41001) + + syncNodeWithSettings(t, svc, 1, "n1-in", + `{"clients":[{"email":"detached@x","subId":"picked","enable":true},{"email":"mine@x","enable":true},{"email":"theirs@x","subId":"picked2","enable":true}]}`) + + if row := readClientRecord(t, db, "detached@x"); row.SubID != "d-sub" || nodeHostsClient(t, db, "n1-in", "detached@x") { + t.Fatalf("node adopted a detached client: subId=%q", row.SubID) + } + if nodeHostsClient(t, db, "n1-in", "theirs@x") { + t.Fatal("node adopted a client another node's merge orphaned") + } + if !nodeHostsClient(t, db, "n1-in", "mine@x") { + t.Fatal("node could not take back a client its own merge orphaned") + } +} + +// The subscription server selects by subId alone, so a node's client must not join +// the subscription of a customer it does not host. +func TestNodeCannotClaimForeignSubID(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + seedNodeRow(t, db, &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true}) + seedMasterInbound(t, db, "master-in", 40001, `{"email":"customer@x","id":"11111111-1111-4111-8111-111111111111","subId":"cust-sub","enable":true}`) + createNodeInbound(t, db, 1, "n1-in", 41001) + + syncNodeWithSettings(t, svc, 1, "n1-in", + `{"clients":[{"email":"node-own@x","id":"66666666-6666-4666-8666-666666666666","subId":"cust-sub","enable":true}]}`) + + if n := countClientsWhere(t, db, "sub_id = ?", "cust-sub"); n != 1 { + t.Fatalf("node joined a customer's subscription: %d records hold cust-sub", n) + } +} + +// A shared client's tunnel keys are rendered into the master's own WireGuard inbound, so a +// node tunnel must not own them even when its inbound id is the lower one. +func TestNodeTunnelDoesNotOwnSharedPeerKeys(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + seedNodeRow(t, db, &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true}) + + const masterKey = "mJ3F8Yb3Kx6cdx6Sx4WbJ3qQDXB5cN3d0pH6m4kq0Ug=" + const nodeKey = "nK9Pq1Zr6s8Lw2Vb4Xc7Dm0Fh5Jt3Ny9Ru1Ea6Go2Is=" + entry := func(key, ip string) string { + return fmt.Sprintf(`{"email":"shared@x","publicKey":%q,"allowedIPs":[%q],"enable":true}`, key, ip) + } + nid := 1 + nodeIb := &model.Inbound{ + UserId: 1, Tag: "n1-wg", Enable: true, Port: 41002, Protocol: model.WireGuard, NodeID: &nid, + Settings: `{"clients":[` + entry(masterKey, "10.1.0.2/32") + `]}`, + } + masterIb := &model.Inbound{ + UserId: 1, Tag: "master-wg", Enable: true, Port: 40002, Protocol: model.WireGuard, + Settings: `{"clients":[` + entry(masterKey, "10.0.0.2/32") + `]}`, + } + for _, ib := range []*model.Inbound{nodeIb, masterIb} { + if err := db.Create(ib).Error; err != nil { + t.Fatalf("create %q: %v", ib.Tag, err) + } + clients, err := svc.GetClients(ib) + if err != nil { + t.Fatalf("parse %q: %v", ib.Tag, err) + } + if err := (&ClientService{}).SyncInbound(db, ib.Id, clients); err != nil { + t.Fatalf("attach %q: %v", ib.Tag, err) + } + } + + snap := &runtime.TrafficSnapshot{Inbounds: []*model.Inbound{{ + Tag: "n1-wg", Enable: true, Port: 41002, Protocol: model.WireGuard, + Settings: `{"clients":[` + entry(nodeKey, "10.1.0.2/32") + `]}`, + ClientStats: []xray.ClientTraffic{{Email: "shared@x", Enable: true}}, + }}} + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { + t.Fatalf("sync: %v", err) + } + if !nodeHostsClient(t, db, "n1-wg", "shared@x") { + t.Fatal("the shared client lost its node tunnel link") + } + if got := readClientRecord(t, db, "shared@x").PublicKey; got != masterKey { + t.Fatalf("node tunnel replaced a shared client's public key: %q", got) + } +} + +// An honest node's copy of a shared client differs only in timestamps and comment; treating +// that as a refused claim would mark the node dirty and re-push it on every tick. +func TestNodeSharedClientInAgreementIsNotRewritten(t *testing.T) { + db := initTrafficTestDB(t) + svc := &InboundService{} + seedNodeRow(t, db, &model.Node{Id: 1, Name: "n1", Address: "127.0.0.1", Port: 2096, ApiToken: "tok", Enable: true}) + + shared := `{"email":"shared@x","id":"11111111-1111-4111-8111-111111111111","subId":"shared-sub","comment":"vip","enable":true}` + seedMasterInbound(t, db, "master-in", 40001, shared) + nid := 1 + nodeIb := &model.Inbound{ + UserId: 1, Tag: "n1-in", Enable: true, Port: 41001, Protocol: model.VLESS, NodeID: &nid, + Settings: `{"clients":[` + shared + `]}`, + } + if err := db.Create(nodeIb).Error; err != nil { + t.Fatalf("create node inbound: %v", err) + } + clients, _ := svc.GetClients(nodeIb) + if err := (&ClientService{}).SyncInbound(db, nodeIb.Id, clients); err != nil { + t.Fatalf("attach node clients: %v", err) + } + + syncNodeWithSettings(t, svc, 1, "n1-in", + `{"clients":[{"email":"shared@x","id":"11111111-1111-4111-8111-111111111111","subId":"shared-sub","enable":true,"created_at":1,"updated_at":2}]}`, + xray.ClientTraffic{Email: "shared@x", Enable: true}) + + var node model.Node + if err := db.Select("config_dirty").Where("id = ?", 1).First(&node).Error; err != nil { + t.Fatalf("read node: %v", err) + } + if node.ConfigDirty { + t.Fatal("a node in agreement with the master was marked dirty for a re-push") + } +}