mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-11-02 03:05:02 +03:00
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.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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"))
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
+11
-10
@@ -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/<hash>.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 {
|
||||
|
||||
Reference in New Issue
Block a user