diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 3c543b68c..986e16e2f 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -708,6 +708,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi mergedCentral := make(map[int]struct{}, len(central)) snapTags := make(map[string]struct{}, len(snap.Inbounds)) + oversizeTags := map[string]string{} for snapIdx, snapIb := range snap.Inbounds { if snapIb == nil { continue @@ -759,6 +760,11 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } mergedCentral[c.Id] = struct{}{} } + if reason := oversizeNodeInbound(snapIb); reason != "" { + oversizeTags[snapIb.Tag] = reason + snap.Inbounds[snapIdx] = nil + continue + } if !ok { if dirty { continue @@ -915,6 +921,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi } } + logOversizeNodeInbounds(nodeID, oversizeTags) if hostGroupsSkipped > 0 { logger.Warningf( "setRemoteTraffic: node %d: left out %d host group adoption(s) — a group is limited to %d bytes and the node's inbounds to %d host rows", diff --git a/internal/web/service/inbound_node_claims.go b/internal/web/service/inbound_node_claims.go index 29c21dc93..1f2b5736b 100644 --- a/internal/web/service/inbound_node_claims.go +++ b/internal/web/service/inbound_node_claims.go @@ -2,6 +2,7 @@ package service import ( "encoding/json" + "fmt" "reflect" "strings" "sync" @@ -42,6 +43,9 @@ func loadNodeClientClaims(tx *gorm.DB, nodeID int, snap *runtime.TrafficSnapshot storedSubID: map[string]string{}, } subIDByEmail := map[string]string{} + // A subId selects a whole subscription, so within one snapshot it may name one client only. + firstEmailBySub := map[string]string{} + sharesSubID := map[string]struct{}{} for _, snapIb := range snap.Inbounds { if snapIb == nil { continue @@ -63,6 +67,11 @@ func loadNodeClientClaims(tx *gorm.DB, nodeID int, snap *runtime.TrafficSnapshot var subID string _ = json.Unmarshal(c.SubID, &subID) subIDByEmail[c.Email] = subID + if first, taken := firstEmailBySub[subID]; subID != "" && taken && !strings.EqualFold(first, c.Email) { + sharesSubID[c.Email] = struct{}{} + } else if subID != "" && !taken { + firstEmailBySub[subID] = c.Email + } } } } @@ -105,13 +114,10 @@ func loadNodeClientClaims(tx *gorm.DB, nodeID int, snap *runtime.TrafficSnapshot holders = append(holders, rows...) } - ids := make([]int, 0, len(records)+len(holders)) + ids := make([]int, 0, len(records)) 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 @@ -120,9 +126,12 @@ func loadNodeClientClaims(tx *gorm.DB, nodeID int, snap *runtime.TrafficSnapshot for _, h := range holders { holdersBySub[h.SubID] = append(holdersBySub[h.SubID], h) } - subHeldElsewhere := func(email, subID string) bool { + subIDTaken := func(email, subID string) bool { + if _, shared := sharesSubID[email]; shared { + return true + } for _, h := range holdersBySub[subID] { - if !strings.EqualFold(h.Email, email) && scopes[h.Id] != (clientLinkScope{here: true}) { + if !strings.EqualFold(h.Email, email) { return true } } @@ -134,8 +143,8 @@ func loadNodeClientClaims(tx *gorm.DB, nodeID int, snap *runtime.TrafficSnapshot 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" + } else if subID != "" && subIDTaken(email, subID) { + claims.refused[email] = "its subId belongs to another client" } continue } @@ -147,7 +156,7 @@ func loadNodeClientClaims(tx *gorm.DB, nodeID int, snap *runtime.TrafficSnapshot 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): + case subID != rec.SubID && subID != "" && subIDTaken(email, subID): claims.storedSubID[email] = rec.SubID } } @@ -283,3 +292,45 @@ func sharedRecordWouldChange(rec *model.ClientRecord, reported *model.Client) bo keepLinkOwnedFields(&merged, rec, false) return !reflect.DeepEqual(merged, *rec) } + +// A node's stream, sniffing and remark are rendered into every subscription link of the +// inbound, once per client and external proxy, so their size bounds a request's cost. +const ( + maxNodeStreamSettingsBytes = 64 << 10 + maxNodeSniffingBytes = 16 << 10 + maxNodeRemarkBytes = 1 << 10 + maxNodeExternalProxies = 64 +) + +// reportedOversizeInbounds holds, per node, the tags its last snapshot left out as oversized. +var reportedOversizeInbounds sync.Map + +// oversizeNodeInbound names the field that keeps a reported inbound from being taken in. +func oversizeNodeInbound(ib *model.Inbound) string { + switch { + case len(ib.StreamSettings) > maxNodeStreamSettingsBytes: + return fmt.Sprintf("streamSettings exceeds %d bytes", maxNodeStreamSettingsBytes) + case len(ib.Sniffing) > maxNodeSniffingBytes: + return fmt.Sprintf("sniffing exceeds %d bytes", maxNodeSniffingBytes) + case len(ib.Remark) > maxNodeRemarkBytes: + return fmt.Sprintf("remark exceeds %d bytes", maxNodeRemarkBytes) + } + var stream struct { + ExternalProxy []json.RawMessage `json:"externalProxy"` + } + if json.Unmarshal([]byte(ib.StreamSettings), &stream) == nil && len(stream.ExternalProxy) > maxNodeExternalProxies { + return fmt.Sprintf("more than %d external proxies", maxNodeExternalProxies) + } + return "" +} + +func logOversizeNodeInbounds(nodeID int, tags map[string]string) { + prev, _ := reportedOversizeInbounds.Load(nodeID) + logged, _ := prev.(map[string]string) + for tag, reason := range tags { + if _, seen := logged[tag]; !seen { + logger.Warningf("setRemoteTraffic: node %d reported inbound %.64q, not taking it in: %s", nodeID, tag, reason) + } + } + reportedOversizeInbounds.Store(nodeID, tags) +} diff --git a/internal/web/service/node_foreign_client_test.go b/internal/web/service/node_foreign_client_test.go index 097260625..0d36eebb0 100644 --- a/internal/web/service/node_foreign_client_test.go +++ b/internal/web/service/node_foreign_client_test.go @@ -383,3 +383,59 @@ func TestNodeRefusedClaimLogStateIsBounded(t *testing.T) { t.Fatalf("refused-claim log state holds %d emails for one node, want 1", got) } } + +// A subscription renders every client holding its subId, so a node creating many clients +// under one subId multiplied the cost of a single request without bound. +func TestNodeClientsCannotShareASubID(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}) + createNodeInbound(t, db, 1, "n1-in", 41001) + + syncNodeWithSettings(t, svc, 1, "n1-in", + `{"clients":[{"email":"a@x","subId":"one-sub","enable":true},{"email":"b@x","subId":"one-sub","enable":true},{"email":"c@x","subId":"c-sub","enable":true}]}`) + syncNodeWithSettings(t, svc, 1, "n1-in", + `{"clients":[{"email":"a@x","subId":"one-sub","enable":true},{"email":"c@x","subId":"c-sub","enable":true},{"email":"d@x","subId":"one-sub","enable":true}]}`) + + if n := countClientsWhere(t, db, "sub_id = ?", "one-sub"); n != 1 { + t.Fatalf("%d records hold one subId, want 1", n) + } + if !nodeHostsClient(t, db, "n1-in", "c@x") { + t.Fatal("a client with its own subId was not adopted") + } +} + +// Every reported stream, sniffing and remark is rendered once per client and external +// proxy of a subscription, so an oversized one let a node make one request cost gigabytes. +func TestNodeOversizeInboundIsNotTakenIn(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}) + createNodeInbound(t, db, 1, "n1-in", 41001) + + proxies := make([]string, maxNodeExternalProxies+1) + for i := range proxies { + proxies[i] = fmt.Sprintf(`{"dest":"h%d.example","port":443,"forceTls":"same"}`, i) + } + stream := `{"network":"tcp","externalProxy":[` + strings.Join(proxies, ",") + `]}` + snap := &runtime.TrafficSnapshot{Inbounds: []*model.Inbound{ + {Tag: "n1-in", Enable: true, Port: 41001, Protocol: model.VLESS, Settings: `{"clients":[]}`, StreamSettings: stream}, + {Tag: "n1-new", Enable: true, Port: 41002, Protocol: model.VLESS, Settings: `{"clients":[]}`, Remark: strings.Repeat("r", maxNodeRemarkBytes+1)}, + }} + if _, err := svc.setRemoteTrafficLocked(1, snap, false, false); err != nil { + t.Fatalf("sync: %v", err) + } + + var stored model.Inbound + if err := db.Where("tag = ?", "n1-in").First(&stored).Error; err != nil { + t.Fatalf("the oversized report removed the existing inbound: %v", err) + } + if strings.Contains(stored.StreamSettings, "externalProxy") { + t.Fatal("an inbound with too many external proxies was taken in") + } + var created int64 + db.Model(&model.Inbound{}).Where("tag = ?", "n1-new").Count(&created) + if created != 0 { + t.Fatal("an inbound with an oversized remark was adopted") + } +}