From 3526a65f5dd58c2ad09f711366cc30dd9ebaf2ed Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Sun, 1 Nov 2026 01:01:38 +0100 Subject: [PATCH] fix(web): bound the panel's open connections and share asset gzip The engine-wide gzip middleware gave every in-flight request its own compressor of about 1 MiB, and the public /assets bundles let an anonymous client keep a handler blocked in Write until WriteTimeout, so a thousand stalled downloads held a gigabyte. The listener itself took any number of connections, each holding buffers and a goroutine before any login. The bundles are now served from a gzip copy made once per file, with the assets prefix left out of the per-request middleware, and the panel listener closes a new connection once 2048 are open or its source (an IPv4 address or IPv6 /64) holds 256; a loopback reverse proxy counts only toward the total. --- internal/web/assets_test.go | 89 +++++++++++++++++++ internal/web/middleware/gzip_static.go | 64 ++++++++++++++ internal/web/middleware/gzip_static_test.go | 88 ++++++++++++++++++ internal/web/network/limit_listener.go | 98 +++++++++++++++++++++ internal/web/network/serve_test.go | 77 ++++++++++++++++ internal/web/routes_contract_test.go | 10 ++- internal/web/web.go | 21 ++--- 7 files changed, 436 insertions(+), 11 deletions(-) create mode 100644 internal/web/assets_test.go create mode 100644 internal/web/middleware/gzip_static.go create mode 100644 internal/web/middleware/gzip_static_test.go create mode 100644 internal/web/network/limit_listener.go diff --git a/internal/web/assets_test.go b/internal/web/assets_test.go new file mode 100644 index 000000000..6258e656b --- /dev/null +++ b/internal/web/assets_test.go @@ -0,0 +1,89 @@ +package web + +import ( + "bytes" + "compress/gzip" + "io" + "io/fs" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "testing" +) + +// Any per-request compressor here is pinned for as long as an anonymous client +// stalls its download, so the bundles must come out as one shared gzip copy. +func TestAssetsServedFromSharedGzipCopy(t *testing.T) { + entries, err := fs.ReadDir(distFS, "dist/assets") + if err != nil || len(entries) == 0 { + t.Skip("no built frontend bundles embedded (dist stub)") + } + name := "" + for _, e := range entries { + if strings.HasSuffix(e.Name(), ".js") { + name = e.Name() + break + } + } + if name == "" { + t.Skip("no .js bundle embedded") + } + raw, err := fs.ReadFile(distFS, "dist/assets/"+name) + if err != nil { + t.Fatal(err) + } + engine := newTestEngine(t) + + get := func(acceptEncoding string) *http.Response { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/assets/"+name, nil) + req.Header.Set("Accept-Encoding", acceptEncoding) + w := httptest.NewRecorder() + engine.ServeHTTP(w, req) + return w.Result() + } + + for range 2 { + res := get("gzip") + body, _ := io.ReadAll(res.Body) + if res.StatusCode != http.StatusOK || res.Header.Get("Content-Encoding") != "gzip" { + t.Fatalf("gzip download: got %d, Content-Encoding %q", res.StatusCode, res.Header.Get("Content-Encoding")) + } + // A precompressed copy is sent with its length; a streamed compressor can't know it. + if got := res.Header.Get("Content-Length"); got != strconv.Itoa(len(body)) { + t.Errorf("Content-Length = %q, want %d: bundle was compressed per request", got, len(body)) + } + if res.Header.Get("Last-Modified") == "" { + t.Error("Last-Modified missing: assets FS no longer stamps startTime") + } + zr, err := gzip.NewReader(bytes.NewReader(body)) + if err != nil { + t.Fatal(err) + } + if plain, _ := io.ReadAll(zr); !bytes.Equal(plain, raw) { + t.Fatal("gzip download decompresses to something other than the bundle") + } + } + + res := get("") + body, _ := io.ReadAll(res.Body) + if res.StatusCode != http.StatusOK || res.Header.Get("Content-Encoding") != "" || !bytes.Equal(body, raw) { + t.Fatalf("plain download: got %d, Content-Encoding %q, %d bytes", res.StatusCode, res.Header.Get("Content-Encoding"), len(body)) + } +} + +// The assets route never listed its directory; serving the bundles from a shared copy +// must not turn that 404 into a 500. +func TestAssetsDirectoryIsNotFound(t *testing.T) { + engine := newTestEngine(t) + for _, encoding := range []string{"", "gzip"} { + req := httptest.NewRequest(http.MethodGet, "/assets/", nil) + req.Header.Set("Accept-Encoding", encoding) + w := httptest.NewRecorder() + engine.ServeHTTP(w, req) + if w.Code != http.StatusNotFound { + t.Errorf("GET /assets/ with Accept-Encoding %q = %d, want 404", encoding, w.Code) + } + } +} diff --git a/internal/web/middleware/gzip_static.go b/internal/web/middleware/gzip_static.go new file mode 100644 index 000000000..d09ff7631 --- /dev/null +++ b/internal/web/middleware/gzip_static.go @@ -0,0 +1,64 @@ +package middleware + +import ( + "bytes" + "compress/gzip" + "io/fs" + "net/http" + "path" + "strconv" + "strings" + "sync" + + "github.com/gin-gonic/gin" +) + +// GzipStatic serves a static route's "*filepath" from a gzip copy made once per +// file, so a stalled download holds shared bytes instead of its own compressor. +func GzipStatic(files fs.FS) gin.HandlerFunc { + var compressed sync.Map // name -> func() []byte + return func(c *gin.Context) { + if c.Request.Method != http.MethodGet && c.Request.Method != http.MethodHead { + return + } + c.Writer.Header().Add("Vary", "Accept-Encoding") + if !strings.Contains(c.GetHeader("Accept-Encoding"), "gzip") { + return + } + name := strings.TrimPrefix(path.Clean("/"+c.Param("filepath")), "/") + // Only names that exist get an entry, so requests cannot grow the map. + info, err := fs.Stat(files, name) + if err != nil || info.IsDir() { + return + } + once, _ := compressed.LoadOrStore(name, sync.OnceValue(func() []byte { + return gzipFile(files, name) + })) + body := once.(func() []byte)() + if body == nil { + return + } + c.Header("Content-Encoding", "gzip") + // ServeContent leaves Content-Length unset once Content-Encoding is present. + c.Header("Content-Length", strconv.Itoa(len(body))) + http.ServeContent(c.Writer, c.Request, name, info.ModTime(), bytes.NewReader(body)) + c.Abort() + } +} + +// gzipFile returns the file compressed, or nil to leave it to the static route. +func gzipFile(files fs.FS, name string) []byte { + raw, err := fs.ReadFile(files, name) + if err != nil { + return nil + } + var buf bytes.Buffer + zw, _ := gzip.NewWriterLevel(&buf, gzip.BestCompression) + if _, err := zw.Write(raw); err != nil { + return nil + } + if err := zw.Close(); err != nil { + return nil + } + return buf.Bytes() +} diff --git a/internal/web/middleware/gzip_static_test.go b/internal/web/middleware/gzip_static_test.go new file mode 100644 index 000000000..acecc1aed --- /dev/null +++ b/internal/web/middleware/gzip_static_test.go @@ -0,0 +1,88 @@ +package middleware + +import ( + "bytes" + "compress/gzip" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "testing/fstest" + + "github.com/gin-gonic/gin" +) + +// readCountingFS counts ReadFile calls, i.e. how often a file gets compressed. +type readCountingFS struct { + fstest.MapFS + reads atomic.Int32 +} + +func (f *readCountingFS) ReadFile(name string) ([]byte, error) { + f.reads.Add(1) + return f.MapFS.ReadFile(name) +} + +func TestGzipStatic(t *testing.T) { + gin.SetMode(gin.TestMode) + raw := strings.Repeat("console.log('bundle');", 4096) + files := &readCountingFS{MapFS: fstest.MapFS{ + "app.js": {Data: []byte(raw)}, + }} + r := gin.New() + r.Group("/assets", GzipStatic(files)).StaticFS("/", http.FS(files)) + + do := func(method, target, acceptEncoding string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(method, target, nil) + req.Header.Set("Accept-Encoding", acceptEncoding) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + return w + } + + t.Run("gzip downloads share one compression", func(t *testing.T) { + for range 3 { + w := do(http.MethodGet, "/assets/app.js", "gzip, deflate") + if w.Code != http.StatusOK || w.Header().Get("Content-Encoding") != "gzip" || w.Header().Get("Vary") != "Accept-Encoding" { + t.Fatalf("got %d, headers %v", w.Code, w.Header()) + } + if ct := w.Header().Get("Content-Type"); !strings.Contains(ct, "javascript") { + t.Errorf("Content-Type = %q, want the bundle's own type", ct) + } + zr, err := gzip.NewReader(bytes.NewReader(w.Body.Bytes())) + if err != nil { + t.Fatal(err) + } + if plain, _ := io.ReadAll(zr); string(plain) != raw { + t.Fatal("decompressed body differs from the bundle") + } + } + if n := files.reads.Load(); n != 1 { + t.Errorf("bundle compressed %d times, want once", n) + } + }) + + t.Run("HEAD sends headers only", func(t *testing.T) { + w := do(http.MethodHead, "/assets/app.js", "gzip") + if w.Code != http.StatusOK || w.Body.Len() != 0 || w.Header().Get("Content-Encoding") != "gzip" { + t.Errorf("got %d, %d body bytes, headers %v", w.Code, w.Body.Len(), w.Header()) + } + }) + + t.Run("without Accept-Encoding the static route serves the plain file", func(t *testing.T) { + w := do(http.MethodGet, "/assets/app.js", "") + if w.Code != http.StatusOK || w.Header().Get("Content-Encoding") != "" || w.Body.String() != raw { + t.Errorf("got %d, Content-Encoding %q, %d body bytes", w.Code, w.Header().Get("Content-Encoding"), w.Body.Len()) + } + }) + + t.Run("unknown name is left to the static route", func(t *testing.T) { + w := do(http.MethodGet, "/assets/none.js", "gzip") + if w.Code != http.StatusNotFound || w.Header().Get("Content-Encoding") != "" { + t.Errorf("got %d, Content-Encoding %q, want 404 plain", w.Code, w.Header().Get("Content-Encoding")) + } + }) +} diff --git a/internal/web/network/limit_listener.go b/internal/web/network/limit_listener.go new file mode 100644 index 000000000..3c3b17398 --- /dev/null +++ b/internal/web/network/limit_listener.go @@ -0,0 +1,98 @@ +package network + +import ( + "net" + "net/netip" + "sync" +) + +// Every open panel connection holds buffers and a goroutine before any login, so their +// number is bounded overall and per source; a loopback reverse proxy counts only overall. +const ( + MaxPanelConns = 2048 + MaxPanelConnsPerSource = 256 +) + +type limitListener struct { + net.Listener + total, perSource int + + mu sync.Mutex + open int + bySource map[netip.Prefix]int +} + +// NewLimitListener closes an accepted connection at once when the total or its source +// (an IPv4 address or IPv6 /64) is at its limit, rather than leaving it in the backlog. +func NewLimitListener(l net.Listener, total, perSource int) net.Listener { + return &limitListener{Listener: l, total: total, perSource: perSource, bySource: map[netip.Prefix]int{}} +} + +func (l *limitListener) Accept() (net.Conn, error) { + for { + c, err := l.Listener.Accept() + if err != nil { + return nil, err + } + source, counted := connSource(c.RemoteAddr()) + if !l.acquire(source, counted) { + _ = c.Close() + continue + } + return &limitConn{Conn: c, release: func() { l.release(source, counted) }}, nil + } +} + +func (l *limitListener) acquire(source netip.Prefix, counted bool) bool { + l.mu.Lock() + defer l.mu.Unlock() + if l.open >= l.total || (counted && l.bySource[source] >= l.perSource) { + return false + } + l.open++ + if counted { + l.bySource[source]++ + } + return true +} + +func (l *limitListener) release(source netip.Prefix, counted bool) { + l.mu.Lock() + defer l.mu.Unlock() + l.open-- + if counted { + if l.bySource[source]--; l.bySource[source] <= 0 { + delete(l.bySource, source) + } + } +} + +// connSource is the per-source bucket of addr; loopback and unparseable sources have none. +func connSource(addr net.Addr) (netip.Prefix, bool) { + ap, err := netip.ParseAddrPort(addr.String()) + if err != nil { + return netip.Prefix{}, false + } + ip := ap.Addr().Unmap() + if ip.IsLoopback() { + return netip.Prefix{}, false + } + bits := 32 + if ip.Is6() { + bits = 64 + } + prefix, err := ip.Prefix(bits) + return prefix, err == nil +} + +type limitConn struct { + net.Conn + once sync.Once + release func() +} + +func (c *limitConn) Close() error { + err := c.Conn.Close() + c.once.Do(c.release) + return err +} diff --git a/internal/web/network/serve_test.go b/internal/web/network/serve_test.go index 8a0c08657..115a38be7 100644 --- a/internal/web/network/serve_test.go +++ b/internal/web/network/serve_test.go @@ -109,3 +109,80 @@ func TestProductionHTTPServersUseServeHTTPWrapper(t *testing.T) { t.Fatalf("scan production Go files: %v", err) } } + +// queueListener hands out conns from a channel, each claiming the given remote address. +type queueListener struct{ conns chan net.Conn } + +func (l queueListener) Accept() (net.Conn, error) { + c, ok := <-l.conns + if !ok { + return nil, net.ErrClosed + } + return c, nil +} +func (queueListener) Close() error { return nil } +func (queueListener) Addr() net.Addr { return testAddr("queue") } + +type remoteConn struct { + net.Conn + remote net.Addr + closed chan struct{} +} + +func (c *remoteConn) RemoteAddr() net.Addr { return c.remote } +func (c *remoteConn) Close() error { close(c.closed); return nil } + +func newRemoteConn(addr string) *remoteConn { + return &remoteConn{remote: &net.TCPAddr{IP: net.ParseIP(addr), Port: 40000}, closed: make(chan struct{})} +} + +// Every panel connection holds buffers before any login, so one source opening +// thousands of them used to take the whole panel's memory. +func TestLimitListenerCapsConnectionsPerSourceAndInTotal(t *testing.T) { + queue := queueListener{conns: make(chan net.Conn, 16)} + l := NewLimitListener(queue, 4, 2) + accept := func(c *remoteConn) net.Conn { + t.Helper() + queue.conns <- c + accepted, err := l.Accept() + if err != nil { + t.Fatalf("Accept: %v", err) + } + return accepted + } + + first := accept(newRemoteConn("203.0.113.1")) + accept(newRemoteConn("203.0.113.1")) + refused := newRemoteConn("203.0.113.1") + queue.conns <- refused + other := accept(newRemoteConn("2001:db8::1")) + select { + case <-refused.closed: + default: + t.Fatal("a source past its share was accepted") + } + if other.RemoteAddr().String() != "[2001:db8::1]:40000" { + t.Fatalf("accepted %s, want the other source", other.RemoteAddr()) + } + + // Loopback is exempt from the per-source share but counts toward the total. + accept(newRemoteConn("127.0.0.1")) + overTotal := newRemoteConn("127.0.0.1") + queue.conns <- overTotal + next := make(chan net.Conn, 1) + go func() { + c, _ := l.Accept() + next <- c + }() + select { + case <-overTotal.closed: + case c := <-next: + t.Fatalf("a connection past the total was accepted: %s", c.RemoteAddr()) + } + _ = first.Close() + queue.conns <- newRemoteConn("203.0.113.1") + again := <-next + if again.RemoteAddr().String() != "203.0.113.1:40000" { + t.Fatalf("a closed connection did not free its slot: accepted %s", again.RemoteAddr()) + } +} diff --git a/internal/web/routes_contract_test.go b/internal/web/routes_contract_test.go index dccdcb42e..7e6af1bf0 100644 --- a/internal/web/routes_contract_test.go +++ b/internal/web/routes_contract_test.go @@ -10,6 +10,7 @@ import ( "testing" "time" + "github.com/gin-gonic/gin" "github.com/robfig/cron/v3" "github.com/mhsanaei/3x-ui/v3/internal/database/dbtest" @@ -42,7 +43,8 @@ func inContractScope(method, path string) bool { return strings.HasPrefix(path, "/panel/api/") || contractExtraRoutes[method+" "+path] } -func registeredContractRoutes(t *testing.T) map[string]bool { +// newTestEngine builds the real panel router on a throwaway database. +func newTestEngine(t *testing.T) *gin.Engine { t.Helper() dbtest.InitDB(t, filepath.Join(t.TempDir(), "x-ui.db")) @@ -59,6 +61,12 @@ func registeredContractRoutes(t *testing.T) map[string]bool { if err != nil { t.Fatalf("init router: %v", err) } + return engine +} + +func registeredContractRoutes(t *testing.T) map[string]bool { + t.Helper() + engine := newTestEngine(t) routes := make(map[string]bool) for _, r := range engine.Routes() { routes[r.Method+" "+r.Path] = true diff --git a/internal/web/web.go b/internal/web/web.go index aaee5dbc6..81790cb8c 100644 --- a/internal/web/web.go +++ b/internal/web/web.go @@ -66,16 +66,14 @@ type cronPanicLogger struct{} func (cronPanicLogger) Printf(format string, args ...any) { logger.Errorf(format, args...) } -// wrapDistFS adapts the embedded `dist/` directory so it can be mounted -// as the panel's `/assets/` static route. Vite emits its bundled JS/CSS -// under `dist/assets/`; serving the FS rooted at `dist/assets` makes -// `/assets/.js` URLs resolve directly. +// wrapDistFS serves the embedded `dist/assets` tree (Vite's bundled JS/CSS) as the panel's +// `/assets/` route; a named field keeps embed.FS's ReadFile, rooted at `dist`, unpromoted. type wrapDistFS struct { - embed.FS + dist embed.FS } func (f *wrapDistFS) Open(name string) (fs.File, error) { - file, err := f.FS.Open("dist/assets/" + name) + file, err := f.dist.Open("dist/assets/" + name) if err != nil { return nil, err } @@ -205,8 +203,10 @@ func (s *Server) initRouter() (*gin.Engine, error) { if err != nil { return nil, err } - engine.Use(gzip.Gzip(gzip.DefaultCompression)) assetsBasePath := basePath + "assets/" + // The bundles are served below from a gzip copy made once per file: compressing them + // per request lets every stalled download pin its own compressor until WriteTimeout. + engine.Use(gzip.Gzip(gzip.DefaultCompression, gzip.WithExcludedPaths([]string{assetsBasePath}))) store := cookie.NewStore(secret) // Configure default session cookie options, including expiration (MaxAge) @@ -244,11 +244,11 @@ func (s *Server) initRouter() (*gin.Engine, error) { // so the Vite watcher's incremental rebuilds show up without // restarting the binary; in prod we serve the embedded dist FS // rooted at `dist/assets/`. + var assetsFS fs.FS = &wrapDistFS{dist: distFS} if config.IsDebug() { - engine.StaticFS(basePath+"assets", http.FS(os.DirFS("internal/web/dist/assets"))) - } else { - engine.StaticFS(basePath+"assets", http.FS(&wrapDistFS{FS: distFS})) + assetsFS = os.DirFS("internal/web/dist/assets") } + engine.Group(basePath+"assets", middleware.GzipStatic(assetsFS)).StaticFS("/", http.FS(assetsFS)) // Hand the embedded `dist/` filesystem to the controller package // before any HTML-serving controller is constructed. Phase 8 @@ -606,6 +606,7 @@ func (s *Server) start(restartXray bool, startTgBot bool) (err error) { if err != nil { return err } + listener = network.NewLimitListener(listener, network.MaxPanelConns, network.MaxPanelConnsPerSource) if certFile != "" || keyFile != "" { cert, err := tls.LoadX509KeyPair(certFile, keyFile) if err == nil {