From 5a1ea427855640e30168446291c8dacfd0538ed3 Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Sat, 31 Oct 2026 09:19:51 +0000 Subject: [PATCH] fix(node): bound what a node's answers can cost the master Node responses are untrusted input on the master, and a few shapes still cost it without bound under the 64 MiB body cap: - a success=false message became an error (logged, stored as lastError) in full; it is now cut to the 8 KiB the non-200 diagnostic already uses; - each "{}" of a reported list decodes into a whole struct, so lists are now decoded entry by entry with per-endpoint caps (1024 descendants, 65536 inbounds and host groups, 1M client-IP rows), and the GUID-keyed trees and maps are capped at 1M values; - a TLS handshake the node never answers kept a socket and goroutines alive forever, since the transport keeps dialing for its pool after the request gives up; node transports now give up after 10 s. --- internal/web/runtime/remote.go | 47 +++++++------- internal/web/runtime/remote_decode.go | 81 +++++++++++++++++++++++++ internal/web/runtime/remote_test.go | 28 +++++++++ internal/web/runtime/tls_client.go | 5 ++ internal/web/runtime/tls_client_test.go | 37 +++++++++++ 5 files changed, 172 insertions(+), 26 deletions(-) create mode 100644 internal/web/runtime/remote_decode.go diff --git a/internal/web/runtime/remote.go b/internal/web/runtime/remote.go index a156027c7..5a4b9a6f9 100644 --- a/internal/web/runtime/remote.go +++ b/internal/web/runtime/remote.go @@ -327,6 +327,7 @@ func (r *Remote) do(ctx context.Context, method, path string, body any) (*envelo return nil, fmt.Errorf("decode envelope: %w", err) } if !env.Success { + env.Msg = truncateNodeMessage(env.Msg) return &env, &remoteAPIError{msg: env.Msg} } return &env, nil @@ -422,8 +423,8 @@ func (r *Remote) ListInboundOptions(ctx context.Context) ([]RemoteInboundOption, if err != nil { return nil, err } - var list []RemoteInboundOption - if err := json.Unmarshal(env.Obj, &list); err != nil { + list, err := decodeCappedList[RemoteInboundOption](env.Obj, maxNodeInbounds) + if err != nil { return nil, fmt.Errorf("decode inbound list: %w", err) } return list, nil @@ -434,11 +435,11 @@ func (r *Remote) refreshRemoteIDs(ctx context.Context) error { if err != nil { return err } - var list []struct { + list, err := decodeCappedList[struct { Id int `json:"id"` Tag string `json:"tag"` - } - if err := json.Unmarshal(env.Obj, &list); err != nil { + }](env.Obj, maxNodeInbounds) + if err != nil { return fmt.Errorf("decode inbound list: %w", err) } next := make(map[string]int, len(list)) @@ -750,11 +751,9 @@ func (r *Remote) GetDescendants(ctx context.Context) ([]model.NodeSummary, error if err != nil { return nil, err } - var out []model.NodeSummary - if len(env.Obj) > 0 { - if err := json.Unmarshal(env.Obj, &out); err != nil { - return nil, fmt.Errorf("decode descendants: %w", err) - } + out, err := decodeCappedList[model.NodeSummary](env.Obj, maxNodeDescendants) + if err != nil { + return nil, fmt.Errorf("decode descendants: %w", err) } return out, nil } @@ -813,11 +812,9 @@ func (r *Remote) FetchHostGroups(ctx context.Context) ([]*entity.HostGroup, erro if err != nil { return nil, err } - var groups []*entity.HostGroup - if len(env.Obj) > 0 { - if err := json.Unmarshal(env.Obj, &groups); err != nil { - return nil, fmt.Errorf("decode host groups: %w", err) - } + groups, err := decodeCappedList[*entity.HostGroup](env.Obj, maxNodeHostGroups) + if err != nil { + return nil, fmt.Errorf("decode host groups: %w", err) } return groups, nil } @@ -829,7 +826,7 @@ func (r *Remote) FetchTrafficSnapshot(ctx context.Context) (*TrafficSnapshot, er if err != nil { return nil, err } - if err := json.Unmarshal(envList.Obj, &snap.Inbounds); err != nil { + if snap.Inbounds, err = decodeCappedList[*model.Inbound](envList.Obj, maxNodeInbounds); err != nil { return nil, fmt.Errorf("decode inbound list: %w", err) } @@ -837,14 +834,14 @@ func (r *Remote) FetchTrafficSnapshot(ctx context.Context) (*TrafficSnapshot, er // node is an old build without the per-GUID endpoint (#4983). envTree, err := r.do(ctx, http.MethodPost, "panel/api/clients/onlinesByGuid", nil) if err == nil && len(envTree.Obj) > 0 { - _ = json.Unmarshal(envTree.Obj, &snap.OnlineTree) + _ = decodeCappedValues(envTree.Obj, &snap.OnlineTree, maxNodeMapValues) } if len(snap.OnlineTree) == 0 { envOnlines, err := r.do(ctx, http.MethodPost, "panel/api/clients/onlines", nil) if err != nil { logger.Warning("remote", r.node.Name, "onlines fetch failed:", err) } else if len(envOnlines.Obj) > 0 { - _ = json.Unmarshal(envOnlines.Obj, &snap.OnlineEmails) + _ = decodeCappedValues(envOnlines.Obj, &snap.OnlineEmails, maxNodeMapValues) } } @@ -852,14 +849,14 @@ func (r *Remote) FetchTrafficSnapshot(ctx context.Context) (*TrafficSnapshot, er if err != nil { logger.Warning("remote", r.node.Name, "lastOnline fetch failed:", err) } else if len(envLastOnline.Obj) > 0 { - _ = json.Unmarshal(envLastOnline.Obj, &snap.LastOnlineMap) + _ = decodeCappedValues(envLastOnline.Obj, &snap.LastOnlineMap, maxNodeMapValues) } envActiveInbounds, err := r.do(ctx, http.MethodPost, "panel/api/clients/activeInbounds", nil) if err != nil { logger.Debugf("remote %s active inbounds fetch failed: %v", r.node.Name, err) } else if len(envActiveInbounds.Obj) > 0 { - _ = json.Unmarshal(envActiveInbounds.Obj, &snap.ActiveInboundTree) + _ = decodeCappedValues(envActiveInbounds.Obj, &snap.ActiveInboundTree, maxNodeMapValues) } return snap, nil @@ -989,11 +986,9 @@ func (r *Remote) FetchAllClientIps(ctx context.Context) ([]model.InboundClientIp if err != nil { return nil, err } - var ips []model.InboundClientIps - if len(env.Obj) > 0 { - if err := json.Unmarshal(env.Obj, &ips); err != nil { - return nil, fmt.Errorf("decode client ips: %w", err) - } + ips, err := decodeCappedList[model.InboundClientIps](env.Obj, maxNodeClientIPs) + if err != nil { + return nil, fmt.Errorf("decode client ips: %w", err) } return ips, nil } @@ -1014,7 +1009,7 @@ func (r *Remote) FetchClientIpsByGuid(ctx context.Context) (map[string]map[strin } out := map[string]map[string][]model.ClientIpEntry{} if len(env.Obj) > 0 { - if err := json.Unmarshal(env.Obj, &out); err != nil { + if err := decodeCappedValues(env.Obj, &out, maxNodeMapValues); err != nil { return nil, fmt.Errorf("decode client ips by guid: %w", err) } } diff --git a/internal/web/runtime/remote_decode.go b/internal/web/runtime/remote_decode.go new file mode 100644 index 000000000..2bd2be435 --- /dev/null +++ b/internal/web/runtime/remote_decode.go @@ -0,0 +1,81 @@ +package runtime + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "unicode/utf8" +) + +// Per-endpoint caps on node-reported lists: each "{}" decodes into a whole struct, so the +// byte cap alone still lets a node's answer expand into gigabytes. +const ( + maxNodeDescendants = 1024 + maxNodeInbounds = 65536 + maxNodeHostGroups = 65536 + maxNodeClientIPs = 1 << 20 + // maxNodeMapValues bounds every value of the GUID-keyed trees and maps a node reports. + maxNodeMapValues = 1 << 20 +) + +var errRemoteListTooLong = errors.New("remote list exceeds entry limit") + +// decodeCappedList decodes a JSON array entry by entry and stops at the first entry +// past maxEntries; null decodes to a nil slice. +func decodeCappedList[T any](raw json.RawMessage, maxEntries int) ([]T, error) { + raw = bytes.TrimSpace(raw) + if len(raw) == 0 || bytes.Equal(raw, []byte("null")) { + return nil, nil + } + dec := json.NewDecoder(bytes.NewReader(raw)) + if tok, err := dec.Token(); err != nil { + return nil, err + } else if delim, ok := tok.(json.Delim); !ok || delim != '[' { + return nil, fmt.Errorf("expected a JSON array, got %v", tok) + } + var out []T + for dec.More() { + if len(out) >= maxEntries { + return nil, fmt.Errorf("%w (%d)", errRemoteListTooLong, maxEntries) + } + var item T + if err := dec.Decode(&item); err != nil { + return nil, err + } + out = append(out, item) + } + return out, nil +} + +// decodeCappedValues unmarshals raw into v after checking it holds at most maxValues +// JSON values in all, nested ones included. +func decodeCappedValues(raw json.RawMessage, v any, maxValues int) error { + dec := json.NewDecoder(bytes.NewReader(raw)) + for count := 0; ; count++ { + tok, err := dec.Token() + if err != nil { + break + } + if delim, ok := tok.(json.Delim); ok && (delim == ']' || delim == '}') { + continue + } + if count > maxValues { + return fmt.Errorf("%w (%d values)", errRemoteListTooLong, maxValues) + } + } + return json.Unmarshal(raw, v) +} + +// truncateNodeMessage bounds a node-supplied message the master turns into an error, +// which is logged and stored as the node's lastError. +func truncateNodeMessage(msg string) string { + if len(msg) <= errBodyDiagBytes { + return msg + } + cut := errBodyDiagBytes + for cut > 0 && !utf8.RuneStart(msg[cut]) { + cut-- + } + return msg[:cut] + "... (truncated)" +} diff --git a/internal/web/runtime/remote_test.go b/internal/web/runtime/remote_test.go index a3cacf3db..1b79bd577 100644 --- a/internal/web/runtime/remote_test.go +++ b/internal/web/runtime/remote_test.go @@ -510,3 +510,31 @@ func TestRemoteAdoptedAliasYieldsToNodeReportedTag(t *testing.T) { t.Fatalf("central tag resolved to %d via a stale alias, want 5 (the id the node reports)", id) } } + +// A node's success=false message becomes an error the master logs and stores as the +// node's lastError; it must not carry the whole response body there. +func TestRemoteDo_BoundsTheNodeErrorMessage(t *testing.T) { + srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"success":false,"msg":"` + strings.Repeat("m", 1<<20) + `"}`)) + })) + defer srv.Close() + r := NewRemote(nodeForServer(t, srv, "skip", ""), nil) + _, err := r.do(context.Background(), http.MethodGet, "/probe", nil) + if err == nil || len(err.Error()) > errBodyDiagBytes+64 { + t.Fatalf("error = %d bytes, want at most the %d-byte diagnostic", len(err.Error()), errBodyDiagBytes) + } +} + +// Each "{}" of a node-reported list decodes into a whole struct, so a body under the +// byte cap could still expand into gigabytes; lists are capped by entry count. +func TestRemoteNodeListsAreCappedByEntryCount(t *testing.T) { + descendants := "[" + strings.TrimSuffix(strings.Repeat("{},", maxNodeDescendants+1), ",") + "]" + srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"success":true,"obj":` + descendants + `}`)) + })) + defer srv.Close() + r := NewRemote(nodeForServer(t, srv, "skip", ""), nil) + if got, err := r.GetDescendants(context.Background()); !errors.Is(err, errRemoteListTooLong) { + t.Fatalf("GetDescendants = %d entries, %v; want errRemoteListTooLong", len(got), err) + } +} diff --git a/internal/web/runtime/tls_client.go b/internal/web/runtime/tls_client.go index a9db7260a..4836e4b52 100644 --- a/internal/web/runtime/tls_client.go +++ b/internal/web/runtime/tls_client.go @@ -140,8 +140,13 @@ const ( maxIdleNodeConnsPerHost = 8 ) +// nodeTLSHandshakeTimeout ends a handshake the node never answers: the transport keeps +// dialing for its pool after the request gives up, so without it the socket lives forever. +var nodeTLSHandshakeTimeout = remoteHTTPTimeout + func newNodeTransport(tlsCfg *tls.Config) *http.Transport { return &http.Transport{ + TLSHandshakeTimeout: nodeTLSHandshakeTimeout, MaxIdleConns: maxIdleNodeConns, MaxIdleConnsPerHost: maxIdleNodeConnsPerHost, IdleConnTimeout: 60 * time.Second, diff --git a/internal/web/runtime/tls_client_test.go b/internal/web/runtime/tls_client_test.go index 1bae8063f..1b11f175b 100644 --- a/internal/web/runtime/tls_client_test.go +++ b/internal/web/runtime/tls_client_test.go @@ -16,6 +16,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/mhsanaei/3x-ui/v3/internal/database/model" "github.com/mhsanaei/3x-ui/v3/internal/util/crypto" @@ -693,3 +694,39 @@ func TestNodeClientsDoNotFollowARedirectToAnotherHost(t *testing.T) { t.Fatalf("a node's redirect reached another host %d time(s)", otherHit.Load()) } } + +// The transport keeps dialing for its pool after a request gives up, so a node that +// never answers the TLS handshake would hold a socket and two goroutines forever. +func TestNodeTransportGivesUpOnAStalledTLSHandshake(t *testing.T) { + previous := nodeTLSHandshakeTimeout + nodeTLSHandshakeTimeout = 200 * time.Millisecond + t.Cleanup(func() { nodeTLSHandshakeTimeout = previous }) + + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer lis.Close() + closed := make(chan struct{}) + go func() { + conn, err := lis.Accept() + if err != nil { + return + } + defer conn.Close() + _, _ = io.Copy(io.Discard, conn) // returns once the master hangs up + close(closed) + }() + client := &http.Client{Transport: newNodeTransport(&tls.Config{InsecureSkipVerify: true})} + ctx, cancel := context.WithTimeout(netsafe.ContextWithAllowPrivate(context.Background(), true), 50*time.Millisecond) + defer cancel() + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, "https://"+lis.Addr().String()+"/", nil) + if resp, err := client.Do(req); err == nil { + _ = resp.Body.Close() + } + select { + case <-closed: + case <-time.After(3 * time.Second): + t.Fatal("the stalled handshake's connection is still open") + } +}