diff --git a/admin/assets/app.js b/admin/assets/app.js index 8a9ec9b..7ed774d 100644 --- a/admin/assets/app.js +++ b/admin/assets/app.js @@ -2434,6 +2434,86 @@ function toggleUdpgwFields(on) { el.style.pointerEvents = on ? "" : "none"; } + +const XRAY_NATIVE_TUNING_DEFAULTS = { + safe: { + mux_max_sessions: 64, + mux_global_sessions: 8192, + mux_udp_idle_ms: 30000, + mux_udp_read_buffer: 131072, + mux_udp_write_buffer: 131072, + xhttp_max_sessions: 4096, + xhttp_buffered_posts: 32, + xhttp_queue_timeout_ms: 500, + xhttp_flush_ms: 2, + xhttp_flush_bytes: 32768, + h2_max_concurrent_streams: 256, + h2_upload_buffer_conn: 1048576, + h2_upload_buffer_stream: 262144, + trace_packets: false, + }, + "2k": { + mux_max_sessions: 128, + mux_global_sessions: 32768, + mux_udp_idle_ms: 15000, + mux_udp_read_buffer: 262144, + mux_udp_write_buffer: 262144, + xhttp_max_sessions: 16384, + xhttp_buffered_posts: 64, + xhttp_queue_timeout_ms: 250, + xhttp_flush_ms: 5, + xhttp_flush_bytes: 131072, + h2_max_concurrent_streams: 1024, + h2_upload_buffer_conn: 1048576, + h2_upload_buffer_stream: 262144, + trace_packets: false, + }, +}; + +const XRAY_NATIVE_TUNING_FIELDS = { + mux_max_sessions: "cfgXrayMuxMaxSessions", + mux_global_sessions: "cfgXrayMuxGlobalSessions", + mux_udp_idle_ms: "cfgXrayMuxUdpIdleMs", + mux_udp_read_buffer: "cfgXrayMuxUdpRbuf", + mux_udp_write_buffer: "cfgXrayMuxUdpWbuf", + xhttp_max_sessions: "cfgXrayXhttpMaxSessions", + xhttp_buffered_posts: "cfgXrayXhttpBufferedPosts", + xhttp_queue_timeout_ms: "cfgXrayXhttpQueueTimeoutMs", + xhttp_flush_ms: "cfgXrayXhttpFlushMs", + xhttp_flush_bytes: "cfgXrayXhttpFlushBytes", + h2_max_concurrent_streams: "cfgXrayH2MaxStreams", + h2_upload_buffer_conn: "cfgXrayH2UploadConn", + h2_upload_buffer_stream: "cfgXrayH2UploadStream", +}; + +function setXrayNativeTuningDefaults(profile = "2k") { + const t = XRAY_NATIVE_TUNING_DEFAULTS[profile] || XRAY_NATIVE_TUNING_DEFAULTS.safe; + writeXrayNativeTuning(t); +} + +function writeXrayNativeTuning(t = {}) { + const defaults = XRAY_NATIVE_TUNING_DEFAULTS.safe; + Object.entries(XRAY_NATIVE_TUNING_FIELDS).forEach(([key, id]) => { + const el = document.getElementById(id); + if (el) el.value = Number(t[key] || defaults[key] || 0); + }); + const trace = document.getElementById("cfgXrayTracePackets"); + if (trace) trace.checked = !!t.trace_packets; +} + +function readXrayNativeTuning() { + const out = {}; + Object.entries(XRAY_NATIVE_TUNING_FIELDS).forEach(([key, id]) => { + const el = document.getElementById(id); + out[key] = parseInt(el?.value || "0", 10) || 0; + }); + out.trace_packets = !!document.getElementById("cfgXrayTracePackets")?.checked; + return out; +} + +// Inline onclick handlers in index.html need this on window when the bundled JS is served from admin/assets/app.js. +window.setXrayNativeTuningDefaults = setXrayNativeTuningDefaults; + async function loadServerConfig() { const st = document.getElementById("srvCfgStatus"); st.textContent = "Loading…"; @@ -2503,6 +2583,7 @@ async function loadServerConfig() { const x = c.xray || {}; document.getElementById("cfgXrayEnabled").checked = !!x.enabled; document.getElementById("cfgXrayMode").value = xrayModeFromConfig(x); + writeXrayNativeTuning(x.native_tuning || XRAY_NATIVE_TUNING_DEFAULTS.safe); st.textContent = "Config loaded."; } catch (e) { @@ -2576,6 +2657,7 @@ async function saveServerConfig() { api_server: "127.0.0.1:10085", online_window_seconds: 90, stats_poll_seconds: 15, + native_tuning: readXrayNativeTuning(), }, }; diff --git a/admin/index.html b/admin/index.html index 584d1d5..20bf827 100644 --- a/admin/index.html +++ b/admin/index.html @@ -1188,6 +1188,30 @@
Native runs inside DragonCoreSSH. External uses /opt/sshpanel/xray with DB-backed /opt/sshpanel/xray_config.json and Stats API on 127.0.0.1:10085.
+
+ Native Xray scale tuning +
+
+
+
+
+
+
+
+
+
+
+
+
+
+ +
+ + +
+
These values are saved in the panel config and applied live on restart/reload. No systemd Environment lines are needed.
+
+
@@ -1211,6 +1235,6 @@ - + diff --git a/admin_script.js b/admin_script.js index bd1b309..e3c6e5d 100644 --- a/admin_script.js +++ b/admin_script.js @@ -1143,6 +1143,85 @@ function toggleUdpgwFields(on) { el.style.pointerEvents = on ? "" : "none"; } + +const XRAY_NATIVE_TUNING_DEFAULTS = { + safe: { + mux_max_sessions: 128, + mux_global_sessions: 32768, + mux_udp_idle_ms: 15000, + mux_udp_read_buffer: 262144, + mux_udp_write_buffer: 262144, + xhttp_max_sessions: 16384, + xhttp_buffered_posts: 64, + xhttp_queue_timeout_ms: 250, + xhttp_flush_ms: 5, + xhttp_flush_bytes: 131072, + h2_max_concurrent_streams: 1024, + h2_upload_buffer_conn: 1048576, + h2_upload_buffer_stream: 262144, + trace_packets: false, + }, + "2k": { + mux_max_sessions: 128, + mux_global_sessions: 32768, + mux_udp_idle_ms: 15000, + mux_udp_read_buffer: 262144, + mux_udp_write_buffer: 262144, + xhttp_max_sessions: 16384, + xhttp_buffered_posts: 64, + xhttp_queue_timeout_ms: 250, + xhttp_flush_ms: 5, + xhttp_flush_bytes: 131072, + h2_max_concurrent_streams: 1024, + h2_upload_buffer_conn: 1048576, + h2_upload_buffer_stream: 262144, + trace_packets: false, + }, +}; + +const XRAY_NATIVE_TUNING_FIELDS = { + mux_max_sessions: "cfgXrayMuxMaxSessions", + mux_global_sessions: "cfgXrayMuxGlobalSessions", + mux_udp_idle_ms: "cfgXrayMuxUdpIdleMs", + mux_udp_read_buffer: "cfgXrayMuxUdpRbuf", + mux_udp_write_buffer: "cfgXrayMuxUdpWbuf", + xhttp_max_sessions: "cfgXrayXhttpMaxSessions", + xhttp_buffered_posts: "cfgXrayXhttpBufferedPosts", + xhttp_queue_timeout_ms: "cfgXrayXhttpQueueTimeoutMs", + xhttp_flush_ms: "cfgXrayXhttpFlushMs", + xhttp_flush_bytes: "cfgXrayXhttpFlushBytes", + h2_max_concurrent_streams: "cfgXrayH2MaxStreams", + h2_upload_buffer_conn: "cfgXrayH2UploadConn", + h2_upload_buffer_stream: "cfgXrayH2UploadStream", +}; + +function setXrayNativeTuningDefaults(profile = "2k") { + const t = XRAY_NATIVE_TUNING_DEFAULTS[profile] || XRAY_NATIVE_TUNING_DEFAULTS.safe; + writeXrayNativeTuning(t); +} + +function writeXrayNativeTuning(t = {}) { + const defaults = XRAY_NATIVE_TUNING_DEFAULTS.safe; + Object.entries(XRAY_NATIVE_TUNING_FIELDS).forEach(([key, id]) => { + const el = document.getElementById(id); + if (el) el.value = Number(t[key] || defaults[key] || 0); + }); + const trace = document.getElementById("cfgXrayTracePackets"); + if (trace) trace.checked = !!t.trace_packets; +} + +function readXrayNativeTuning() { + const out = {}; + Object.entries(XRAY_NATIVE_TUNING_FIELDS).forEach(([key, id]) => { + const el = document.getElementById(id); + out[key] = parseInt(el?.value || "0", 10) || 0; + }); + out.trace_packets = !!document.getElementById("cfgXrayTracePackets")?.checked; + return out; +} + +window.setXrayNativeTuningDefaults = setXrayNativeTuningDefaults; + async function loadServerConfig() { const st = document.getElementById("srvCfgStatus"); st.textContent = "Loading…"; @@ -1211,6 +1290,8 @@ async function loadServerConfig() { // Xray const x = c.xray || {}; document.getElementById("cfgXrayEnabled").checked = !!x.enabled; + if (document.getElementById("cfgXrayMode")) document.getElementById("cfgXrayMode").value = x.mode || (x.native === false ? "external" : "native"); + writeXrayNativeTuning(x.native_tuning || XRAY_NATIVE_TUNING_DEFAULTS.safe); st.textContent = "Config loaded."; } catch (e) { @@ -1276,8 +1357,12 @@ async function saveServerConfig() { tls_forwarders: tlsArr, xray: { enabled: document.getElementById("cfgXrayEnabled").checked, + mode: document.getElementById("cfgXrayMode")?.value || "native", + native: (document.getElementById("cfgXrayMode")?.value || "native") !== "external", bin_path: "/opt/sshpanel/xray", config_file: "/opt/sshpanel/xray_config.json", + native_config_file: "/opt/sshpanel/xray_native_config.json", + native_tuning: readXrayNativeTuning(), }, }; diff --git a/go.mod b/go.mod index 5908e01..204406e 100644 --- a/go.mod +++ b/go.mod @@ -2,22 +2,19 @@ module shell2 go 1.25.4 -require ( - github.com/lib/pq v1.10.9 - github.com/xtaci/kcp-go/v5 v5.6.61 - github.com/xtaci/smux v1.5.50 - golang.org/x/crypto v0.45.0 - golang.org/x/net v0.47.0 - golang.org/x/time v0.14.0 - www.bamsoftware.com/git/dnstt.git v1.20241021.0 -) +require golang.org/x/crypto v0.45.0 require ( github.com/flynn/noise v1.0.0 // indirect github.com/klauspost/cpuid/v2 v2.2.6 // indirect github.com/klauspost/reedsolomon v1.12.0 // indirect + github.com/lib/pq v1.10.9 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/tjfoc/gmsm v1.4.1 // indirect + github.com/xtaci/kcp-go/v5 v5.6.61 // indirect + github.com/xtaci/smux v1.5.50 // indirect + golang.org/x/net v0.47.0 // indirect golang.org/x/sys v0.38.0 // indirect - golang.org/x/text v0.31.0 // indirect + golang.org/x/time v0.14.0 // indirect + www.bamsoftware.com/git/dnstt.git v1.20241021.0 // indirect ) diff --git a/go.sum b/go.sum index b277c72..abce5e9 100644 --- a/go.sum +++ b/go.sum @@ -3,8 +3,6 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc= -github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= -github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98= github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= @@ -29,26 +27,18 @@ github.com/klauspost/cpuid/v2 v2.2.6 h1:ndNyv040zDGIDh8thGkXYjnFtiN02M1PVVF+JE/4 github.com/klauspost/cpuid/v2 v2.2.6/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws= github.com/klauspost/reedsolomon v1.12.0 h1:I5FEp3xSwVCcEh3F5A7dofEfhXdF/bWhQWPH+XwBFno= github.com/klauspost/reedsolomon v1.12.0/go.mod h1:EPLZJeh4l27pUGC3aXOjheaoh1I9yut7xTURiW3LQ9Y= -github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= -github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= -github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= -github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0= -github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/tjfoc/gmsm v1.4.1 h1:aMe1GlZb+0bLjn+cKTPEvvn9oUEBlJitaZiiBwsbgho= github.com/tjfoc/gmsm v1.4.1/go.mod h1:j4INPkHWMrhJb38G+J6W4Tw0AbuN8Thu3PbdVYhVcTE= github.com/xtaci/kcp-go/v5 v5.6.61 h1:ajm12pGuWO+GWQNusPyPESC7Rq0yTC2rEXVYkM8ExOg= github.com/xtaci/kcp-go/v5 v5.6.61/go.mod h1:9O3D8WR+cyyUjGiTILYfg17vn72otWuXK2AFfqIe6CM= -github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae h1:J0GxkO96kL4WF+AIT3M4mfUVinOCPgf2uUWYFUzN0sM= -github.com/xtaci/lossyconn v0.0.0-20190602105132-8df528c0c9ae/go.mod h1:gXtu8J62kEgmN++bm9BVICuT/e8yiLI2KFobd/TRFsE= github.com/xtaci/smux v1.5.50 h1:y/1DlWQC9bnMeZzsyk4oL2hbLK6uVk4BKTz5BeQqUEA= github.com/xtaci/smux v1.5.50/go.mod h1:IGQ9QYrBphmb/4aTnLEcJby0TNr3NV+OslIOMrX825Q= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= @@ -83,12 +73,8 @@ golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= -golang.org/x/term v0.37.0 h1:8EGAD0qCmHYZg6J17DvsMy9/wJ7/D/4pV/wfnld5lTU= -golang.org/x/term v0.37.0/go.mod h1:5pB4lxRNYYVZuTLmy8oR2BH8dflOR+IbTYFD8fi3254= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= -golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= @@ -111,10 +97,7 @@ google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQ google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE= google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo= google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= -gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= -gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= www.bamsoftware.com/git/dnstt.git v1.20241021.0 h1:Xi0lmT+5kcgzY7P+r726eBXKMZKgGoD8GTNKrlh8TuE= diff --git a/xray_integration.go b/xray_integration.go index fc10c6b..ecd0e69 100644 --- a/xray_integration.go +++ b/xray_integration.go @@ -6,7 +6,6 @@ import ( "encoding/json" "fmt" "io" - "log" "net/http" "net/url" "os" @@ -49,6 +48,10 @@ type XrayConfig struct { // clients are preserved when the server has IPv6 connectivity. force_ipv4 is // only kept as an explicit legacy override. NativeIPStrategy string `json:"native_ip_strategy,omitempty"` // auto | force_ipv4 + + // NativeTuning exposes native-emulator scale/performance knobs in the admin panel. + // These replace the old XRAY_NATIVE_* systemd environment overrides. + NativeTuning *XrayNativeTuning `json:"native_tuning,omitempty"` } const ( @@ -94,6 +97,9 @@ func (c *XrayConfig) NormalizeDefaults() { default: c.NativeIPStrategy = xrayNativeIPStrategyAuto } + tuning := normalizeNativeXrayTuning(c.NativeTuning) + c.NativeTuning = &tuning + applyNativeXrayTuning(c.NativeTuning) } func (c *XrayConfig) NativeForceIPv4() bool { @@ -155,7 +161,12 @@ type xrayLogRing struct { pos int } -const xrayLogCap = 200 +const ( + xrayLogCap = 5000 + xrayForcedLogLevel = "debug" + xrayForcedAccessLogPath = "/dev/stdout" + xrayForcedErrorLogPath = "/dev/stderr" +) func (r *xrayLogRing) add(line string) { r.mu.Lock() @@ -200,6 +211,38 @@ func (w xrayWriter) Write(p []byte) (int, error) { return os.Stderr.Write(p) } +// xrayLogf writes native/runtime Xray debug messages directly to stderr and +// to the in-memory Xray log ring. It intentionally bypasses the global +// standard logger because the panel can run in quiet mode and call +// log.SetOutput(io.Discard). Xray debug logs must remain visible while +// troubleshooting transport issues. +func xrayLogf(format string, args ...interface{}) { + msg := strings.TrimRight(fmt.Sprintf(format, args...), "\n") + if strings.TrimSpace(msg) == "" { + return + } + ts := time.Now().Format("2006/01/02 15:04:05") + for _, line := range strings.Split(msg, "\n") { + if line == "" { + continue + } + full := ts + " " + line + xrayLogBuf.add(full) + _, _ = fmt.Fprintln(os.Stderr, full) + } +} + +// xrayTracef is for very hot transport-level traces such as every XHTTP packet. +// Leaving those on under many QUIC users can bottleneck on journald/stderr and +// make the tunnel appear slow. Enable only when packet-level tracing is needed: +// Enable packet-level tracing from Admin Panel -> Xray Native Scale -> Trace packets. +func xrayTracef(format string, args ...interface{}) { + if !nativeTracePacketsEnabled() { + return + } + xrayLogf(format, args...) +} + // XrayManager manages the lifecycle of the external xray subprocess. type XrayManager struct { mu sync.Mutex @@ -249,7 +292,7 @@ func initXrayManager(cfg *XrayConfig) { xrayMgr.mu.Lock() xrayMgr.cfg = cfg if err := xrayMgr.bootstrapConfigStoreLocked(); err != nil { - log.Printf("xray: database config bootstrap failed: %v", err) + xrayLogf("xray: database config bootstrap failed: %v", err) } xrayMgr.mu.Unlock() @@ -263,7 +306,7 @@ func initXrayManager(cfg *XrayConfig) { if cfg.Enabled { if err := xrayMgr.Start(); err != nil { - log.Printf("xray: auto-start failed: %v", err) + xrayLogf("xray: auto-start failed: %v", err) } } } @@ -290,11 +333,16 @@ func (m *XrayManager) Start() error { if m.cfg == nil { return fmt.Errorf("xray not configured") } + m.cfg.NormalizeDefaults() if err := m.syncConfigFileFromStoreLocked(); err != nil { m.lastErr = err.Error() return err } configFile := m.activeConfigFileLocked() + if _, err := m.readConfigLocked(); err != nil && !os.IsNotExist(err) { + m.lastErr = err.Error() + return err + } // Native mode: run the in-process emulator instead of the subprocess. if m.cfg.UseNative() { @@ -319,7 +367,7 @@ func (m *XrayManager) Start() error { if changed, err := m.ensureStatsAPIConfigLocked(); err != nil { return fmt.Errorf("xray stats api check failed: %w", err) } else if changed { - log.Printf("xray: repaired Stats API support in config before start") + xrayLogf("xray: repaired Stats API support and forced debug logs in config before start") } args := []string{"run"} @@ -350,10 +398,10 @@ func (m *XrayManager) Start() error { m.lastErr = err.Error() } m.mu.Unlock() - log.Printf("xray: process exited: %v", err) + xrayLogf("xray: process exited: %v", err) }() - log.Printf("xray: started (pid %d)", cmd.Process.Pid) + xrayLogf("xray: started (pid %d)", cmd.Process.Pid) return nil } @@ -386,7 +434,7 @@ func (m *XrayManager) Stop() error { case <-time.After(2 * time.Second): } } - log.Printf("xray: stopped") + xrayLogf("xray: stopped") return nil } @@ -423,13 +471,13 @@ func (m *XrayManager) recordNativeConnect(uuid, email string) { m.statsMu.Unlock() if statsStore != nil && uuid != "" { - go func() { + xrayGo("native xray stats active increment", func() { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() if err := statsStore.UpdateXrayClientActive(ctx, uuid, email, 1); err != nil { - log.Printf("xray native stats: active +1 for %s failed: %v", uuid, err) + xrayLogf("xray native stats: active +1 for %s failed: %v", uuid, err) } - }() + }) } } @@ -453,13 +501,13 @@ func (m *XrayManager) recordNativeDisconnect(uuid, email string) { m.statsMu.Unlock() if statsStore != nil && uuid != "" { - go func() { + xrayGo("native xray stats active decrement", func() { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() if err := statsStore.UpdateXrayClientActive(ctx, uuid, email, -1); err != nil { - log.Printf("xray native stats: active -1 for %s failed: %v", uuid, err) + xrayLogf("xray native stats: active -1 for %s failed: %v", uuid, err) } - }() + }) } } @@ -511,13 +559,13 @@ func (m *XrayManager) startNativeStatsFlusher() { m.nativeStatsFlushStarted = true m.nativeDBMu.Unlock() - go func() { + xrayGo("native xray stats flusher", func() { ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() for range ticker.C { m.flushNativeStatsToDB() } - }() + }) } func (m *XrayManager) flushNativeStatsToDB() { @@ -534,7 +582,7 @@ func (m *XrayManager) flushNativeStatsToDB() { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := statsStore.AddXrayClientTrafficBatch(ctx, pending); err != nil { - log.Printf("xray native stats: db traffic flush failed: %v", err) + xrayLogf("xray native stats: db traffic flush failed: %v", err) // Put deltas back so a transient DB failure does not lose accounting. m.nativeDBMu.Lock() if m.nativeTrafficPending == nil { @@ -964,6 +1012,55 @@ func normalizeJSONIndent(data []byte) ([]byte, error) { return json.MarshalIndent(raw, "", " ") } +func forceXrayDebugLogBytes(data []byte) ([]byte, bool, error) { + if !json.Valid(data) { + return nil, false, fmt.Errorf("invalid JSON") + } + var raw map[string]interface{} + if err := json.Unmarshal(data, &raw); err != nil { + return nil, false, fmt.Errorf("parse xray config: %w", err) + } + if raw == nil { + return nil, false, fmt.Errorf("xray config must be a JSON object") + } + changed := ensureXrayForcedDebugLogConfig(raw) + if !changed { + return data, false, nil + } + out, err := json.MarshalIndent(raw, "", " ") + if err != nil { + return nil, false, err + } + return out, true, nil +} + +func ensureXrayForcedDebugLogConfig(raw map[string]interface{}) bool { + changed := false + logObj := asObject(raw["log"]) + if logObj == nil { + logObj = map[string]interface{}{} + raw["log"] = logObj + changed = true + } + if v, _ := logObj["access"].(string); v != xrayForcedAccessLogPath { + logObj["access"] = xrayForcedAccessLogPath + changed = true + } + if v, _ := logObj["error"].(string); v != xrayForcedErrorLogPath { + logObj["error"] = xrayForcedErrorLogPath + changed = true + } + if v, _ := logObj["loglevel"].(string); !strings.EqualFold(v, xrayForcedLogLevel) { + logObj["loglevel"] = xrayForcedLogLevel + changed = true + } + if v, _ := logObj["dnsLog"].(bool); !v { + logObj["dnsLog"] = true + changed = true + } + return changed +} + func (m *XrayManager) readConfigLocked() ([]byte, error) { configFile := m.activeConfigFileLocked() if m.cfg == nil || configFile == "" { @@ -971,8 +1068,15 @@ func (m *XrayManager) readConfigLocked() ([]byte, error) { } if statsStore != nil { if data, ok, err := statsStore.GetXrayConfig(context.Background(), m.configStoreKeyLocked()); err != nil { - log.Printf("xray: database config read failed, falling back to file: %v", err) + xrayLogf("xray: database config read failed, falling back to file: %v", err) } else if ok { + if forced, changed, err := forceXrayDebugLogBytes(data); err != nil { + return nil, err + } else if changed { + data = forced + _ = statsStore.UpsertXrayConfig(context.Background(), m.configStoreKeyLocked(), forced) + xrayLogf("xray: forced debug log output in database config") + } pretty, err := normalizeJSONIndent(data) if err != nil { return nil, err @@ -981,7 +1085,18 @@ func (m *XrayManager) readConfigLocked() ([]byte, error) { return pretty, nil } } - return os.ReadFile(configFile) + data, err := os.ReadFile(configFile) + if err != nil { + return nil, err + } + if forced, changed, err := forceXrayDebugLogBytes(data); err != nil { + return nil, err + } else if changed { + data = forced + _ = os.WriteFile(configFile, forced, 0o600) + xrayLogf("xray: forced debug log output in file config") + } + return data, nil } func (m *XrayManager) writeConfigLocked(data []byte) error { @@ -989,6 +1104,12 @@ func (m *XrayManager) writeConfigLocked(data []byte) error { if m.cfg == nil || configFile == "" { return fmt.Errorf("xray config file not configured") } + if forced, changed, err := forceXrayDebugLogBytes(data); err != nil { + return err + } else if changed { + data = forced + xrayLogf("xray: forced debug log output while saving config") + } pretty, err := normalizeJSONIndent(data) if err != nil { return err @@ -1011,11 +1132,11 @@ func (m *XrayManager) importConfigClientsLocked(data []byte, source string) { } n, err := statsStore.ImportXrayClientsFromConfig(context.Background(), data) if err != nil { - log.Printf("xray: import clients from %s failed: %v", source, err) + xrayLogf("xray: import clients from %s failed: %v", source, err) return } if n > 0 { - log.Printf("xray: imported/synced %d client UUIDs from %s into database", n, source) + xrayLogf("xray: imported/synced %d client UUIDs from %s into database", n, source) } } @@ -1027,13 +1148,13 @@ func (m *XrayManager) importRuntimeConfigFileClientsLocked(source string) { data, err := os.ReadFile(configFile) if err != nil { if !os.IsNotExist(err) { - log.Printf("xray: read %s for client import failed: %v", configFile, err) + xrayLogf("xray: read %s for client import failed: %v", configFile, err) } return } pretty, err := normalizeJSONIndent(data) if err != nil { - log.Printf("xray: cannot import clients from %s: invalid JSON: %v", configFile, err) + xrayLogf("xray: cannot import clients from %s: invalid JSON: %v", configFile, err) return } m.importConfigClientsLocked(pretty, source) @@ -1048,7 +1169,7 @@ func (m *XrayManager) restartIfExternalRunning() { return } if err := m.Restart(); err != nil { - log.Printf("xray: external restart after client/config change failed: %v", err) + xrayLogf("xray: external restart after client/config change failed: %v", err) } } @@ -1069,6 +1190,15 @@ func (m *XrayManager) bootstrapConfigStoreLocked() error { if data, ok, err := statsStore.GetXrayConfig(ctx, key); err != nil { return err } else if ok { + if forced, changed, err := forceXrayDebugLogBytes(data); err != nil { + return err + } else if changed { + data = forced + if err := statsStore.UpsertXrayConfig(ctx, key, forced); err != nil { + return err + } + xrayLogf("xray: forced debug log output during config bootstrap") + } pretty, err := normalizeJSONIndent(data) if err != nil { return err @@ -1086,6 +1216,12 @@ func (m *XrayManager) bootstrapConfigStoreLocked() error { if migrated, migErr := m.seedNativeConfigFromExternalLocked(configFile); migErr != nil { return migErr } else if len(migrated) > 0 { + if forced, changed, err := forceXrayDebugLogBytes(migrated); err != nil { + return err + } else if changed { + migrated = forced + xrayLogf("xray: forced debug log output during native config migration") + } pretty, err := normalizeJSONIndent(migrated) if err != nil { return err @@ -1102,6 +1238,12 @@ func (m *XrayManager) bootstrapConfigStoreLocked() error { } return err } + if forced, changed, err := forceXrayDebugLogBytes(data); err != nil { + return err + } else if changed { + data = forced + xrayLogf("xray: forced debug log output while importing runtime config") + } pretty, err := normalizeJSONIndent(data) if err != nil { return err @@ -1128,7 +1270,7 @@ func (m *XrayManager) seedNativeConfigFromExternalLocked(nativeConfigFile string if _, err := normalizeJSONIndent(data); err != nil { return nil, err } - log.Printf("xray native: cloned %s to independent native config %s once", ext, nativeConfigFile) + xrayLogf("xray native: cloned %s to independent native config %s once", ext, nativeConfigFile) return data, nil } @@ -1141,6 +1283,15 @@ func (m *XrayManager) syncConfigFileFromStoreLocked() error { if err != nil || !ok { return err } + if forced, changed, err := forceXrayDebugLogBytes(data); err != nil { + return err + } else if changed { + data = forced + if err := statsStore.UpsertXrayConfig(context.Background(), m.configStoreKeyLocked(), forced); err != nil { + return err + } + xrayLogf("xray: forced debug log output while syncing config") + } pretty, err := normalizeJSONIndent(data) if err != nil { return err @@ -1166,7 +1317,7 @@ func (m *XrayManager) SetConfig(data []byte) error { return err } if changed { - log.Printf("xray: added/repaired Stats API support while saving config") + xrayLogf("xray: added/repaired Stats API support and forced debug logs while saving config") } return m.writeConfigLocked(patched) } @@ -1238,6 +1389,9 @@ func patchXrayStatsAPIBytes(data []byte) ([]byte, bool, error) { return nil, false, fmt.Errorf("xray config must be a JSON object") } changed, _ := ensureXrayStatsAPIConfig(raw) + if ensureXrayForcedDebugLogConfig(raw) { + changed = true + } if !changed { return data, false, nil } @@ -1666,7 +1820,7 @@ func handleXrayStatus(w http.ResponseWriter, r *http.Request) { wasRunning := xrayMgr.isRunningSnapshot() if changed, err := xrayMgr.EnsureStatsAPIConfig(); err == nil && changed && wasRunning { if err := xrayMgr.Restart(); err != nil { - log.Printf("xray: auto stats repair restart failed: %v", err) + xrayLogf("xray: auto stats repair restart failed: %v", err) } } } @@ -1992,7 +2146,7 @@ func (m *XrayManager) AddXrayClient(inboundTag, uuid, email string) error { }) if err == nil && m.cfg != nil && m.cfg.UseNative() { if hotErr := nativeXray.addClient(inboundTag, uuid, email); hotErr != nil { - log.Printf("native xray: hot-add client %s to %s failed: %v", uuid, inboundTag, hotErr) + xrayLogf("native xray: hot-add client %s to %s failed: %v", uuid, inboundTag, hotErr) } } return err @@ -2044,7 +2198,7 @@ func (m *XrayManager) RemoveXrayClient(inboundTag, uuid string) error { }) if err == nil && m.cfg != nil && m.cfg.UseNative() { if hotErr := nativeXray.removeClient(inboundTag, uuid); hotErr != nil { - log.Printf("native xray: hot-remove client %s from %s failed: %v", uuid, inboundTag, hotErr) + xrayLogf("native xray: hot-remove client %s from %s failed: %v", uuid, inboundTag, hotErr) } } return err @@ -2090,7 +2244,7 @@ func (m *XrayManager) UpdateXrayClientEmail(uuid, email string) error { }) if err == nil && m.cfg != nil && m.cfg.UseNative() { if hotErr := nativeXray.updateClientEmail(uuid, email); hotErr != nil { - log.Printf("native xray: hot-update client %s email failed: %v", uuid, hotErr) + xrayLogf("native xray: hot-update client %s email failed: %v", uuid, hotErr) } } return err @@ -2359,7 +2513,7 @@ func handleXrayClientAdd(w http.ResponseWriter, r *http.Request) { } } if err := statsStore.UpsertXrayClientMeta(r.Context(), meta); err != nil { - log.Printf("xray: save meta for %s: %v", req.UUID, err) + xrayLogf("xray: save meta for %s: %v", req.UUID, err) } } xrayMgr.restartIfExternalRunning() @@ -2445,7 +2599,7 @@ func handleXrayClientUpdate(w http.ResponseWriter, r *http.Request) { } if req.Email != "" { if err := xrayMgr.UpdateXrayClientEmail(req.UUID, req.Email); err != nil { - log.Printf("xray: update config email for %s: %v", req.UUID, err) + xrayLogf("xray: update config email for %s: %v", req.UUID, err) } else { xrayMgr.restartIfExternalRunning() } diff --git a/xray_native.go b/xray_native.go index c95ce95..a09213a 100644 --- a/xray_native.go +++ b/xray_native.go @@ -9,11 +9,12 @@ package main // // Native emulator scope: // - Protocols : VLESS and VMess AEAD (TCP + UDP commands) +// - VLESS Mux : Mux.Cool child TCP/UDP sessions, including XUDP metadata // - Transports: raw TCP, WebSocket (RFC 6455), XHTTP/SplitHTTP // - Security : TLS, none // -// Mux, REALITY, gRPC and HTTPUpgrade are still deferred; unsupported commands -// are rejected explicitly instead of silently falling back. +// REALITY, gRPC and HTTPUpgrade are still deferred; unsupported commands are +// rejected explicitly instead of silently falling back. // // Native mode has its own DB-backed config/runtime path. It does not spawn or // query the external xray binary and does not require /opt/sshpanel/xray to be @@ -34,7 +35,6 @@ import ( "errors" "fmt" "io" - "log" "net" "net/http" "os" @@ -160,13 +160,13 @@ func (s *nativeXrayServer) start(configFile string) error { serveLn = tls.NewListener(ln, ib.tlsConfig) } opened = append(opened, serveLn) - go ib.serveXHTTPListener(serveLn) + xrayGo(fmt.Sprintf("native xray xhttp listener %s", addr), func() { ib.serveXHTTPListener(serveLn) }) } else { opened = append(opened, serveLn) - go ib.acceptLoop(serveLn) + xrayGo(fmt.Sprintf("native xray accept loop %s", addr), func() { ib.acceptLoop(serveLn) }) } active[ib.tag] = ib - log.Printf("native xray: serving %s/%s on %s (inbound %q, security=%s, %d clients)", + xrayLogf("native xray: serving %s/%s on %s (inbound %q, security=%s, %d clients)", ib.protocol, ib.transport, addr, ib.tag, orNone(ib.security), ib.clientCount()) } @@ -189,25 +189,27 @@ func (s *nativeXrayServer) stop() { s.listeners = nil s.inboundsByTag = nil s.running = false - log.Printf("native xray: stopped") + xrayLogf("native xray: stopped") } func (ib *nativeInbound) acceptLoop(ln net.Listener) { + defer xrayRecover(fmt.Sprintf("native xray accept loop inbound=%q", ib.tag)) for { c, err := ln.Accept() if err != nil { if isListenerClosed(err) { return } - log.Printf("native xray: accept error on %s: %v", ln.Addr(), err) + xrayLogf("native xray: accept error on %s: %v", ln.Addr(), err) continue } - go ib.serve(c) + xrayGo(fmt.Sprintf("native xray connection remote=%s", c.RemoteAddr()), func() { ib.serve(c) }) } } // serve terminates TLS + transport, then dispatches on protocol. func (ib *nativeInbound) serve(raw net.Conn) { + defer xrayRecover(fmt.Sprintf("native xray serve inbound=%q remote=%s", ib.tag, raw.RemoteAddr())) defer raw.Close() if tc, ok := raw.(*net.TCPConn); ok { @@ -222,7 +224,7 @@ func (ib *nativeInbound) serve(raw net.Conn) { tconn := tls.Server(raw, ib.tlsConfig) _ = tconn.SetDeadline(time.Now().Add(tlsHandshakeTimeout)) if err := tconn.Handshake(); err != nil { - log.Printf("native xray: tls handshake from %s failed: %v", raw.RemoteAddr(), err) + xrayLogf("native xray: tls handshake from %s failed: %v", raw.RemoteAddr(), err) return } _ = tconn.SetDeadline(time.Time{}) @@ -237,15 +239,15 @@ func (ib *nativeInbound) serve(raw net.Conn) { case "ws", "websocket": ws, err := wsServerHandshake(conn, ib.path) if err != nil { - log.Printf("native xray: ws handshake from %s failed: %v", raw.RemoteAddr(), err) + xrayLogf("native xray: ws handshake from %s failed: %v", raw.RemoteAddr(), err) return } stream = ws case "xhttp", "splithttp": - log.Printf("native xray: inbound %q got raw connection for XHTTP; this transport is served by http.Server", ib.tag) + xrayLogf("native xray: inbound %q got raw connection for XHTTP; this transport is served by http.Server", ib.tag) return default: - log.Printf("native xray: inbound %q transport %q not supported yet; dropping conn from %s", + xrayLogf("native xray: inbound %q transport %q not supported yet; dropping conn from %s", ib.tag, ib.transport, raw.RemoteAddr()) return } @@ -257,7 +259,7 @@ func (ib *nativeInbound) serve(raw net.Conn) { case "vmess": ib.handleVMess(stream, raw.RemoteAddr()) default: - log.Printf("native xray: inbound %q protocol %q not supported yet; dropping conn from %s", + xrayLogf("native xray: inbound %q protocol %q not supported yet; dropping conn from %s", ib.tag, ib.protocol, raw.RemoteAddr()) } } @@ -270,9 +272,12 @@ func (ib *nativeInbound) serve(raw net.Conn) { // 1 byte addon length M // M bytes addons (flow etc.) — skipped // 1 byte command (1=TCP, 2=UDP, 3=Mux) -// 2 bytes port (big endian) -// 1 byte address type (1=IPv4, 2=domain, 3=IPv6) -// ... address +// for TCP/UDP only: +// 2 bytes port (big endian) +// 1 byte address type (1=IPv4, 2=domain, 3=IPv6) +// ... address +// for Mux: +// ... Mux.Cool/XUDP frames immediately after the command byte // ... payload // Response (server -> client): 1 byte version echo, 1 byte addon length (0). @@ -287,12 +292,17 @@ const ( ) func (ib *nativeInbound) handleVLESS(stream net.Conn, remote net.Addr) { - log.Printf("native xray: vless handshake start inbound=%q transport=%s remote=%s", ib.tag, ib.transport, remote) + defer xrayRecover(fmt.Sprintf("native xray VLESS inbound=%q remote=%s", ib.tag, remote)) + if ib.isXHTTP() { + xrayTracef("native xray: vless handshake start inbound=%q transport=%s remote=%s", ib.tag, ib.transport, remote) + } else { + xrayLogf("native xray: vless handshake start inbound=%q transport=%s remote=%s", ib.tag, ib.transport, remote) + } _ = stream.SetReadDeadline(time.Now().Add(30 * time.Second)) head := make([]byte, 1+16+1) // version + uuid + addonLen if _, err := io.ReadFull(stream, head); err != nil { - log.Printf("native xray: vless handshake failed inbound=%q transport=%s remote=%s: %v", ib.tag, ib.transport, remote, err) + ib.logVLESSReadFailure("handshake", remote, "", err) return } version := head[0] @@ -301,39 +311,59 @@ func (ib *nativeInbound) handleVLESS(stream net.Conn, remote net.Addr) { client := ib.getNativeClient(id) if client == nil { - log.Printf("native xray: inbound %q rejected unknown VLESS uuid from %s", ib.tag, remote) + xrayLogf("native xray: inbound %q rejected unknown VLESS uuid from %s", ib.tag, remote) return } if addonLen := int(head[17]); addonLen > 0 { if _, err := io.CopyN(io.Discard, stream, int64(addonLen)); err != nil { - log.Printf("native xray: vless addon read failed inbound=%q user=%s: %v", ib.tag, client.email, err) + xrayLogf("native xray: vless addon read failed inbound=%q user=%s: %v", ib.tag, client.email, err) return } } var cmd [1]byte if _, err := io.ReadFull(stream, cmd[:]); err != nil { - log.Printf("native xray: vless command read failed inbound=%q user=%s: %v", ib.tag, client.email, err) + ib.logVLESSReadFailure("command", remote, client.email, err) return } - var portBuf [2]byte - if _, err := io.ReadFull(stream, portBuf[:]); err != nil { - log.Printf("native xray: vless port read failed inbound=%q user=%s: %v", ib.tag, client.email, err) - return - } - port := binary.BigEndian.Uint16(portBuf[:]) - host, err := readProxyAddress(stream) - if err != nil { - log.Printf("native xray: inbound %q VLESS bad address from %s: %v", ib.tag, remote, err) - return + var host string + var port uint16 + if cmd[0] == vlessCmdTCP || cmd[0] == vlessCmdUDP { + var portBuf [2]byte + if _, err := io.ReadFull(stream, portBuf[:]); err != nil { + xrayLogf("native xray: vless port read failed inbound=%q user=%s: %v", ib.tag, client.email, err) + return + } + port = binary.BigEndian.Uint16(portBuf[:]) + + var err error + host, err = readProxyAddress(stream) + if err != nil { + xrayLogf("native xray: inbound %q VLESS bad address from %s: %v", ib.tag, remote, err) + return + } + if isNativeDNSSinkTarget(host) { + _ = stream.SetReadDeadline(time.Time{}) + _ = stream.SetWriteDeadline(time.Now().Add(time.Second)) + _, _ = stream.Write([]byte{version, 0}) + xrayTracef("native xray: inbound %q fast-ignored DNS sink target cmd=%d user=%s host=%q port=%d remote=%s", ib.tag, cmd[0], client.email, host, port, remote) + return + } + if invalidNativeDestination(host, port) { + xrayTracef("native xray: inbound %q rejected invalid VLESS target cmd=%d user=%s host=%q port=%d remote=%s", ib.tag, cmd[0], client.email, host, port, remote) + return + } } _ = stream.SetReadDeadline(time.Time{}) - // VLESS response header must be sent before relaying payload. + // VLESS response header must be sent before relaying payload. CommandMux is + // special: official Xray does not read a target from the VLESS header for it; + // the following bytes are Mux.Cool/XUDP frames. Reading port/address here + // deadlocks muxed UDP clients and shows up as QUIC/YouTube stalls. if _, err := stream.Write([]byte{version, 0}); err != nil { - log.Printf("native xray: vless response write failed inbound=%q user=%s: %v", ib.tag, client.email, err) + xrayLogf("native xray: vless response write failed inbound=%q user=%s: %v", ib.tag, client.email, err) return } @@ -341,24 +371,79 @@ func (ib *nativeInbound) handleVLESS(stream net.Conn, remote net.Addr) { case vlessCmdTCP: backend, target, err := ib.nativeDialTCP(host, port) if err != nil { - log.Printf("native xray: inbound %q VLESS TCP dial %s failed: %v", ib.tag, target, err) + xrayLogf("native xray: inbound %q VLESS TCP dial %s failed: %v", ib.tag, target, err) return } - log.Printf("native xray: vless/tcp user=%s src=%s -> %s (inbound %q)", client.email, backend.LocalAddr(), target, ib.tag) + ib.nativeSuccessLogf("native xray: vless/tcp user=%s src=%s -> %s (inbound %q)", client.email, backend.LocalAddr(), target, ib.tag) nativeTunnel(stream, backend, client.uuid, client.email, ib.upLimiter(), ib.downLimiter()) case vlessCmdUDP: backend, target, err := ib.nativeDialUDP(host, port) if err != nil { - log.Printf("native xray: inbound %q VLESS UDP dial %s failed: %v", ib.tag, target, err) + xrayLogf("native xray: inbound %q VLESS UDP dial %s failed: %v", ib.tag, target, err) return } - log.Printf("native xray: vless/udp user=%s src=%s -> %s (inbound %q)", client.email, backend.LocalAddr(), target, ib.tag) + ib.nativeSuccessLogf("native xray: vless/udp user=%s src=%s -> %s (inbound %q)", client.email, backend.LocalAddr(), target, ib.tag) nativeVLESSUDPTunnel(stream, backend, client.uuid, client.email, ib.upLimiter(), ib.downLimiter()) + case vlessCmdMux: + ib.nativeSuccessLogf("native xray: vless/mux user=%s remote=%s (inbound %q)", client.email, remote, ib.tag) + ib.nativeVLESSMuxTunnel(stream, client.uuid, client.email) default: - log.Printf("native xray: inbound %q VLESS command %d not supported yet", ib.tag, cmd[0]) + xrayLogf("native xray: inbound %q VLESS command %d not supported yet", ib.tag, cmd[0]) } } +func (ib *nativeInbound) logVLESSReadFailure(stage string, remote net.Addr, email string, err error) { + if ib.isXHTTP() && isNativeDeadlineError(err) { + if email == "" { + xrayTracef("native xray: vless %s timed out inbound=%q transport=%s remote=%s: %v", stage, ib.tag, ib.transport, remote, err) + } else { + xrayTracef("native xray: vless %s timed out inbound=%q transport=%s user=%s remote=%s: %v", stage, ib.tag, ib.transport, email, remote, err) + } + return + } + if email == "" { + xrayLogf("native xray: vless %s failed inbound=%q transport=%s remote=%s: %v", stage, ib.tag, ib.transport, remote, err) + } else { + xrayLogf("native xray: vless %s failed inbound=%q transport=%s user=%s remote=%s: %v", stage, ib.tag, ib.transport, email, remote, err) + } +} + +func isNativeDeadlineError(err error) bool { + if errors.Is(err, os.ErrDeadlineExceeded) { + return true + } + var ne net.Error + return errors.As(err, &ne) && ne.Timeout() +} + +func invalidNativeDestination(host string, port uint16) bool { + host = strings.TrimSpace(normalizeNativeTargetHost(host)) + if host == "" || port == 0 { + return true + } + if ip := net.ParseIP(stripNativeIPZone(host)); ip != nil { + return ip.IsUnspecified() + } + return false +} + +func isNativeDNSSinkTarget(host string) bool { + host = strings.TrimSpace(normalizeNativeTargetHost(host)) + if host == "" { + return false + } + ip := net.ParseIP(stripNativeIPZone(host)) + return ip != nil && ip.IsUnspecified() +} + +func (ib *nativeInbound) nativeSuccessLogf(format string, args ...interface{}) { + if ib != nil && ib.isXHTTP() { + xrayTracef(format, args...) + return + } + xrayLogf(format, args...) +} + // readProxyAddress reads a VMess/VLESS-style address (type byte + address). func readProxyAddress(r io.Reader) (string, error) { var t [1]byte @@ -445,13 +530,13 @@ func nativeDialTargetWithSource(network, host string, port uint16, sourceHost st cancel() if err == nil { if i > 0 && len(attempts) > 1 { - log.Printf("native xray: outbound dial recovered target=%s network=%s using auto source after bound source failed", target, dialNetwork) + xrayLogf("native xray: outbound dial recovered target=%s network=%s using auto source after bound source failed", target, dialNetwork) } return conn, target, nil } lastErr = err if local != nil { - log.Printf("native xray: outbound dial target=%s network=%s source=%s failed, retrying auto source: %v", target, dialNetwork, local.String(), err) + xrayLogf("native xray: outbound dial target=%s network=%s source=%s failed, retrying auto source: %v", target, dialNetwork, local.String(), err) } } return nil, target, lastErr @@ -537,31 +622,36 @@ func normalizeNativeTargetHost(raw string) string { func nativeTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, email string, up, down *rate.Limiter) { xrayMgr.recordNativeConnect(uuid, email) defer xrayMgr.recordNativeDisconnect(uuid, email) + defer xrayRecover(fmt.Sprintf("native xray TCP tunnel user=%s", email)) + upMeter := &trafficMeter{uuid: uuid, email: email, uplink: true} downMeter := &trafficMeter{uuid: uuid, email: email, uplink: false} var wg sync.WaitGroup + var closeOnce sync.Once closeAll := func() { - _ = backend.Close() - _ = client.Close() + closeOnce.Do(func() { + _ = backend.Close() + _ = client.Close() + }) } wg.Add(1) - go func() { // client -> backend (uplink) + xrayGo("native xray TCP uplink", func() { // client -> backend defer wg.Done() + defer closeAll() _, _ = copyWithRateLimit(meteredWriter{w: backend, meter: upMeter}, client, up) if cw, ok := backend.(interface{ CloseWrite() error }); ok { _ = cw.CloseWrite() } - closeAll() - }() + }) wg.Add(1) - go func() { // backend -> client (downlink) + xrayGo("native xray TCP downlink", func() { // backend -> client defer wg.Done() + defer closeAll() _, _ = copyWithRateLimit(meteredWriter{w: client, meter: downMeter}, backend, down) - closeAll() - }() + }) wg.Wait() upMeter.flush() @@ -853,7 +943,7 @@ func parseNativeInbounds(configFile string) ([]*nativeInbound, error) { } port, ok := parseSinglePort(in.Port) if !ok { - log.Printf("native xray: inbound %q has unsupported port form; skipping", in.Tag) + xrayLogf("native xray: inbound %q has unsupported port form; skipping", in.Tag) continue } @@ -894,7 +984,7 @@ func parseNativeInbounds(configFile string) ([]*nativeInbound, error) { } ib.xhttpMaxBufferedPosts = xh.ScMaxBufferedPosts if ib.xhttpMaxBufferedPosts <= 0 { - ib.xhttpMaxBufferedPosts = 30 + ib.xhttpMaxBufferedPosts = nativeXHTTPBufferedPostLimit() } ib.xhttpSessions = make(map[string]*nativeXHTTPSession) } @@ -904,7 +994,7 @@ func parseNativeInbounds(configFile string) ([]*nativeInbound, error) { if ib.security == "tls" { tc, err := buildInboundTLS(in) if err != nil { - log.Printf("native xray: inbound %q TLS disabled: %v; skipping", in.Tag, err) + xrayLogf("native xray: inbound %q TLS disabled: %v; skipping", in.Tag, err) continue } ib.tlsConfig = tc @@ -920,24 +1010,24 @@ func parseNativeInbounds(configFile string) ([]*nativeInbound, error) { raw = c.Password // some protocols reuse password as id } if err := ib.addNativeClient(proto, raw, c.Email); err != nil { - log.Printf("native xray: inbound %q skipping client %q: %v", in.Tag, raw, err) + xrayLogf("native xray: inbound %q skipping client %q: %v", in.Tag, raw, err) } } if statsStore != nil && in.Tag != "" { metas, err := statsStore.ListXrayClientsByInbound(context.Background(), in.Tag) if err != nil { - log.Printf("native xray: inbound %q database clients unavailable: %v", in.Tag, err) + xrayLogf("native xray: inbound %q database clients unavailable: %v", in.Tag, err) } else { for _, m := range metas { if err := ib.addNativeClient(proto, m.UUID, firstNonEmpty(m.Email, m.Name, m.UUID)); err != nil { - log.Printf("native xray: inbound %q skipping DB client %q: %v", in.Tag, m.UUID, err) + xrayLogf("native xray: inbound %q skipping DB client %q: %v", in.Tag, m.UUID, err) } } } } if ib.clientCount() == 0 { - log.Printf("native xray: inbound %q has no valid clients; skipping", in.Tag) + xrayLogf("native xray: inbound %q has no valid clients; skipping", in.Tag) continue } out = append(out, ib) diff --git a/xray_native_test.go b/xray_native_test.go index f25f2c4..1878756 100644 --- a/xray_native_test.go +++ b/xray_native_test.go @@ -115,6 +115,66 @@ func TestVLESSOverTCP(t *testing.T) { } } +func vlessDNSSinkHeader(id [16]byte, cmd byte, host net.IP, port uint16) []byte { + var b bytes.Buffer + b.WriteByte(0) + b.Write(id[:]) + b.WriteByte(0) + b.WriteByte(cmd) + b.WriteByte(byte(port >> 8)) + b.WriteByte(byte(port)) + if ip4 := host.To4(); ip4 != nil { + b.WriteByte(atypIPv4) + b.Write(ip4) + } else { + b.WriteByte(atypIPv6) + b.Write(host.To16()) + } + return b.Bytes() +} + +func TestVLESSDNSSinkTargetFastIgnored(t *testing.T) { + _, port, id, stop := newTestInbound(t, "tcp", "") + defer stop() + + conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", itoa(port))) + if err != nil { + t.Fatalf("dial inbound: %v", err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(2 * time.Second)) + + if _, err := conn.Write(vlessDNSSinkHeader(id, vlessCmdTCP, net.IPv4(0, 0, 0, 0), 0)); err != nil { + t.Fatalf("write sink header: %v", err) + } + + resp := make([]byte, 2) + if _, err := io.ReadFull(conn, resp); err != nil { + t.Fatalf("read VLESS sink response: %v", err) + } + if resp[0] != 0 || resp[1] != 0 { + t.Fatalf("bad VLESS sink response: %v", resp) + } + + buf := make([]byte, 1) + if _, err := conn.Read(buf); err == nil { + t.Fatalf("sink target should close immediately after response") + } +} + +func TestNativeDNSSinkTargetDetection(t *testing.T) { + for _, host := range []string{"0.0.0.0", "::", "[::]", "::%lo"} { + if !isNativeDNSSinkTarget(host) { + t.Fatalf("%q should be detected as DNS sink", host) + } + } + for _, host := range []string{"", "127.0.0.1", "1.1.1.1", "example.com"} { + if isNativeDNSSinkTarget(host) { + t.Fatalf("%q should not be detected as DNS sink", host) + } + } +} + func TestVLESSRejectsUnknownUUID(t *testing.T) { echoPort, stopEcho := startEchoServer(t) defer stopEcho() @@ -578,3 +638,243 @@ func TestNativeLocalAddrForIPv6Tunnel(t *testing.T) { t.Fatalf("must not bind IPv4 source to IPv6 target: %#v", local) } } + +func startUDPEchoServer(t *testing.T) (int, func()) { + t.Helper() + pc, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + t.Fatalf("udp echo listen: %v", err) + } + go func() { + buf := make([]byte, 64*1024) + for { + n, addr, err := pc.ReadFrom(buf) + if err != nil { + return + } + _, _ = pc.WriteTo(buf[:n], addr) + } + }() + return pc.LocalAddr().(*net.UDPAddr).Port, func() { pc.Close() } +} + +func vlessMuxHeader(id [16]byte) []byte { + var b bytes.Buffer + b.WriteByte(0) + b.Write(id[:]) + b.WriteByte(0) + b.WriteByte(vlessCmdMux) + return b.Bytes() +} + +func buildMuxUDPFrame(sessionID uint16, host string, port int, payload []byte) []byte { + meta := []byte{byte(sessionID >> 8), byte(sessionID), nativeMuxStatusNew, nativeMuxOptionData, nativeMuxNetworkUDP} + meta = appendNativeMuxAddressPort(meta, host, uint16(port)) + var out bytes.Buffer + binary.Write(&out, binary.BigEndian, uint16(len(meta))) + out.Write(meta) + binary.Write(&out, binary.BigEndian, uint16(len(payload))) + out.Write(payload) + return out.Bytes() +} + +func buildMuxTCPFrame(sessionID uint16, host string, port int, payload []byte) []byte { + meta := []byte{byte(sessionID >> 8), byte(sessionID), nativeMuxStatusNew, nativeMuxOptionData, nativeMuxNetworkTCP} + meta = appendNativeMuxAddressPort(meta, host, uint16(port)) + var out bytes.Buffer + binary.Write(&out, binary.BigEndian, uint16(len(meta))) + out.Write(meta) + binary.Write(&out, binary.BigEndian, uint16(len(payload))) + out.Write(payload) + return out.Bytes() +} + +func buildMuxXUDPFrame(sessionID uint16, host string, port int, payload []byte, gid [8]byte) []byte { + meta := []byte{byte(sessionID >> 8), byte(sessionID), nativeMuxStatusNew, nativeMuxOptionData, nativeMuxNetworkUDP} + meta = appendNativeMuxAddressPort(meta, host, uint16(port)) + meta = append(meta, gid[:]...) + var out bytes.Buffer + binary.Write(&out, binary.BigEndian, uint16(len(meta))) + out.Write(meta) + // Official Mux.Cool/XUDP carries GlobalID in the outer New metadata. The + // following packet block is the UDP datagram itself, not another XUDP metadata + // stream. + binary.Write(&out, binary.BigEndian, uint16(len(payload))) + out.Write(payload) + return out.Bytes() +} + +func TestVLESSMuxUDPDoesNotStall(t *testing.T) { + udpPort, stopUDP := startUDPEchoServer(t) + defer stopUDP() + _, port, id, stop := newTestInbound(t, "tcp", "") + defer stop() + + conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", itoa(port))) + if err != nil { + t.Fatalf("dial inbound: %v", err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(5 * time.Second)) + + if _, err := conn.Write(vlessMuxHeader(id)); err != nil { + t.Fatalf("write mux header: %v", err) + } + resp := make([]byte, 2) + if _, err := io.ReadFull(conn, resp); err != nil { + t.Fatalf("read mux response header: %v", err) + } + if resp[0] != 0 || resp[1] != 0 { + t.Fatalf("bad mux vless response: %v", resp) + } + + want := []byte("quic-over-mux") + if _, err := conn.Write(buildMuxUDPFrame(7, "127.0.0.1", udpPort, want)); err != nil { + t.Fatalf("write mux udp frame: %v", err) + } + meta, err := readNativeMuxMetadata(conn) + if err != nil { + t.Fatalf("read mux response meta: %v", err) + } + if meta.sessionID != 7 || meta.status != nativeMuxStatusKeep || meta.option&nativeMuxOptionData == 0 { + t.Fatalf("bad mux response metadata: %#v", meta) + } + got, err := readNativeMuxDataBlock(conn) + if err != nil { + t.Fatalf("read mux response payload: %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("mux udp echo mismatch: got %q want %q", got, want) + } +} + +func TestVLESSMuxXUDPDoesNotStall(t *testing.T) { + udpPort, stopUDP := startUDPEchoServer(t) + defer stopUDP() + _, port, id, stop := newTestInbound(t, "tcp", "") + defer stop() + + conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", itoa(port))) + if err != nil { + t.Fatalf("dial inbound: %v", err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(5 * time.Second)) + + if _, err := conn.Write(vlessMuxHeader(id)); err != nil { + t.Fatalf("write mux header: %v", err) + } + resp := make([]byte, 2) + if _, err := io.ReadFull(conn, resp); err != nil { + t.Fatalf("read mux response header: %v", err) + } + + want := []byte("quic-over-xudp") + gid := [8]byte{1, 2, 3, 4, 5, 6, 7, 8} + if _, err := conn.Write(buildMuxXUDPFrame(9, "127.0.0.1", udpPort, want, gid)); err != nil { + t.Fatalf("write mux xudp frame: %v", err) + } + meta, err := readNativeMuxMetadata(conn) + if err != nil { + t.Fatalf("read mux xudp response meta: %v", err) + } + if meta.sessionID != 9 || meta.status != nativeMuxStatusKeep || meta.option&nativeMuxOptionData == 0 { + t.Fatalf("bad mux xudp response metadata: %#v", meta) + } + if meta.host != "127.0.0.1" || int(meta.port) != udpPort { + t.Fatalf("xudp response did not preserve UDP endpoint: %#v", meta) + } + got, err := readNativeMuxDataBlock(conn) + if err != nil { + t.Fatalf("read xudp response payload: %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("mux xudp echo mismatch: got %q want %q", got, want) + } +} + +func TestVLESSMuxXUDPPayloadLookingLikeMetadataDoesNotStall(t *testing.T) { + udpPort, stopUDP := startUDPEchoServer(t) + defer stopUDP() + _, port, id, stop := newTestInbound(t, "tcp", "") + defer stop() + + conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", itoa(port))) + if err != nil { + t.Fatalf("dial inbound: %v", err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(5 * time.Second)) + + if _, err := conn.Write(vlessMuxHeader(id)); err != nil { + t.Fatalf("write mux header: %v", err) + } + resp := make([]byte, 2) + if _, err := io.ReadFull(conn, resp); err != nil { + t.Fatalf("read mux response header: %v", err) + } + + // This payload intentionally looks like old inner-XUDP metadata. Mux.Cool + // packet mode must still forward it as one UDP datagram and must not block + // waiting for another fake payload block. + want := []byte{0, 0, 2, 1, 'q', 'u', 'i', 'c'} + gid := [8]byte{8, 7, 6, 5, 4, 3, 2, 1} + if _, err := conn.Write(buildMuxXUDPFrame(10, "127.0.0.1", udpPort, want, gid)); err != nil { + t.Fatalf("write mux xudp frame: %v", err) + } + meta, err := readNativeMuxMetadata(conn) + if err != nil { + t.Fatalf("read mux xudp response meta: %v", err) + } + if meta.sessionID != 10 || meta.status != nativeMuxStatusKeep || meta.option&nativeMuxOptionData == 0 { + t.Fatalf("bad mux xudp response metadata: %#v", meta) + } + got, err := readNativeMuxDataBlock(conn) + if err != nil { + t.Fatalf("read xudp response payload: %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("mux xudp metadata-looking payload changed: got %q want %q", got, want) + } +} + +func TestVLESSMuxTCPDoesNotStall(t *testing.T) { + tcpPort, stopTCP := startEchoServer(t) + defer stopTCP() + _, port, id, stop := newTestInbound(t, "tcp", "") + defer stop() + + conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", itoa(port))) + if err != nil { + t.Fatalf("dial inbound: %v", err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(5 * time.Second)) + + if _, err := conn.Write(vlessMuxHeader(id)); err != nil { + t.Fatalf("write mux header: %v", err) + } + resp := make([]byte, 2) + if _, err := io.ReadFull(conn, resp); err != nil { + t.Fatalf("read mux response header: %v", err) + } + + want := []byte("tcp-over-mux") + if _, err := conn.Write(buildMuxTCPFrame(11, "127.0.0.1", tcpPort, want)); err != nil { + t.Fatalf("write mux tcp frame: %v", err) + } + meta, err := readNativeMuxMetadata(conn) + if err != nil { + t.Fatalf("read mux tcp response meta: %v", err) + } + if meta.sessionID != 11 || meta.status != nativeMuxStatusKeep || meta.option&nativeMuxOptionData == 0 { + t.Fatalf("bad mux tcp response metadata: %#v", meta) + } + got, err := readNativeMuxDataBlock(conn) + if err != nil { + t.Fatalf("read mux tcp response payload: %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("mux tcp echo mismatch: got %q want %q", got, want) + } +} diff --git a/xray_native_udp.go b/xray_native_udp.go index 3371f9b..e90a854 100644 --- a/xray_native_udp.go +++ b/xray_native_udp.go @@ -5,7 +5,6 @@ import ( "encoding/binary" "fmt" "io" - "log" "net" "sync" "time" @@ -29,24 +28,29 @@ const ( func nativeVLESSUDPTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, email string, up, down *rate.Limiter) { xrayMgr.recordNativeConnect(uuid, email) defer xrayMgr.recordNativeDisconnect(uuid, email) + defer xrayRecover(fmt.Sprintf("native xray VLESS UDP tunnel user=%s", email)) upMeter := &trafficMeter{uuid: uuid, email: email, uplink: true} downMeter := &trafficMeter{uuid: uuid, email: email, uplink: false} var wg sync.WaitGroup + var closeOnce sync.Once closeAll := func() { - _ = backend.Close() - _ = client.Close() + closeOnce.Do(func() { + _ = backend.Close() + _ = client.Close() + }) } wg.Add(1) - go func() { + xrayGo("native xray VLESS UDP uplink", func() { defer wg.Done() + defer closeAll() for { payload, err := readVLESSLengthPacket(client) if err != nil { if err != io.EOF { - log.Printf("native xray: VLESS UDP client read failed: %v", err) + xrayLogf("native xray: VLESS UDP client read failed: %v", err) } return } @@ -61,15 +65,16 @@ func nativeVLESSUDPTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, ema upMeter.add(n) } if err != nil { - log.Printf("native xray: VLESS UDP backend write failed: %v", err) + xrayLogf("native xray: VLESS UDP backend write failed: %v", err) return } } - }() + }) wg.Add(1) - go func() { + xrayGo("native xray VLESS UDP downlink", func() { defer wg.Done() + defer closeAll() buf := make([]byte, nativeUDPBufferSize) for { _ = backend.SetReadDeadline(time.Now().Add(nativeUDPIdle)) @@ -79,7 +84,7 @@ func nativeVLESSUDPTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, ema return } if err != io.EOF { - log.Printf("native xray: VLESS UDP backend read failed: %v", err) + xrayLogf("native xray: VLESS UDP backend read failed: %v", err) } return } @@ -90,12 +95,12 @@ func nativeVLESSUDPTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, ema return } if err := writeVLESSLengthPacket(client, buf[:n]); err != nil { - log.Printf("native xray: VLESS UDP client write failed: %v", err) + xrayLogf("native xray: VLESS UDP client write failed: %v", err) return } downMeter.add(n) } - }() + }) wg.Wait() upMeter.flush() @@ -198,12 +203,12 @@ func writeVLESSLengthPacket(w io.Writer, payload []byte) error { if len(payload) > nativeUDPMaxPacket { return fmt.Errorf("udp packet too large: %d", len(payload)) } - var lenBuf [2]byte - binary.BigEndian.PutUint16(lenBuf[:], uint16(len(payload))) - if _, err := w.Write(lenBuf[:]); err != nil { - return err - } - _, err := w.Write(payload) + // One Write is important for XHTTP because the response writer flushes once per + // Write. Two writes per UDP packet doubles flush/syscall pressure. + frame := make([]byte, 0, 2+len(payload)) + frame = appendUint16(frame, uint16(len(payload))) + frame = append(frame, payload...) + _, err := w.Write(frame) return err } @@ -299,24 +304,29 @@ func writeVLESSXUDPPacket(w io.Writer, payload []byte) error { func nativeVMessUDPTunnel(client nativeVMessStream, backend net.Conn, uuid, email string, up, down *rate.Limiter) { xrayMgr.recordNativeConnect(uuid, email) defer xrayMgr.recordNativeDisconnect(uuid, email) + defer xrayRecover(fmt.Sprintf("native xray VMess UDP tunnel user=%s", email)) upMeter := &trafficMeter{uuid: uuid, email: email, uplink: true} downMeter := &trafficMeter{uuid: uuid, email: email, uplink: false} var wg sync.WaitGroup + var closeOnce sync.Once closeAll := func() { - _ = backend.Close() - _ = client.Close() + closeOnce.Do(func() { + _ = backend.Close() + _ = client.Close() + }) } wg.Add(1) - go func() { + xrayGo("native xray VMess UDP uplink", func() { defer wg.Done() + defer closeAll() for { pkt, err := client.ReadPacket() if err != nil { if err != io.EOF { - log.Printf("native xray: VMess UDP client read failed: %v", err) + xrayLogf("native xray: VMess UDP client read failed: %v", err) } return } @@ -331,15 +341,16 @@ func nativeVMessUDPTunnel(client nativeVMessStream, backend net.Conn, uuid, emai upMeter.add(n) } if err != nil { - log.Printf("native xray: VMess UDP backend write failed: %v", err) + xrayLogf("native xray: VMess UDP backend write failed: %v", err) return } } - }() + }) wg.Add(1) - go func() { + xrayGo("native xray VMess UDP downlink", func() { defer wg.Done() + defer closeAll() buf := make([]byte, nativeUDPBufferSize) for { _ = backend.SetReadDeadline(time.Now().Add(nativeUDPIdle)) @@ -349,7 +360,7 @@ func nativeVMessUDPTunnel(client nativeVMessStream, backend net.Conn, uuid, emai return } if err != io.EOF { - log.Printf("native xray: VMess UDP backend read failed: %v", err) + xrayLogf("native xray: VMess UDP backend read failed: %v", err) } return } @@ -360,12 +371,12 @@ func nativeVMessUDPTunnel(client nativeVMessStream, backend net.Conn, uuid, emai return } if err := client.WritePacket(buf[:n]); err != nil { - log.Printf("native xray: VMess UDP client write failed: %v", err) + xrayLogf("native xray: VMess UDP client write failed: %v", err) return } downMeter.add(n) } - }() + }) wg.Wait() upMeter.flush() diff --git a/xray_xhttp.go b/xray_xhttp.go index bc6a6cb..3b74014 100644 --- a/xray_xhttp.go +++ b/xray_xhttp.go @@ -2,11 +2,11 @@ package main import ( "container/heap" + "context" "encoding/base64" "errors" "fmt" "io" - "log" "net" "net/http" "os" @@ -98,7 +98,12 @@ func mergeNativeXHTTPSettings(primary, fallback nativeXHTTPSettingsJSON) nativeX } func (ib *nativeInbound) serveXHTTPListener(ln net.Listener) { - h2s := &http2.Server{} + defer xrayRecover(fmt.Sprintf("native xray XHTTP listener inbound=%q addr=%s", ib.tag, ln.Addr())) + h2s := &http2.Server{ + MaxConcurrentStreams: uint32(nativeH2MaxConcurrentStreams()), + MaxUploadBufferPerConnection: int32(nativeH2UploadBufferConn()), + MaxUploadBufferPerStream: int32(nativeH2UploadBufferStream()), + } handler := http.Handler(ib) // Official Xray accepts plaintext HTTP/1.1 and h2c on non-TLS XHTTP // listeners, and negotiates h2/http1 through ALPN on TLS listeners. Without @@ -117,7 +122,7 @@ func (ib *nativeInbound) serveXHTTPListener(ln net.Listener) { _ = http2.ConfigureServer(srv, h2s) } if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) && !isListenerClosed(err) { - log.Printf("native xray: XHTTP server for inbound %q stopped: %v", ib.tag, err) + xrayLogf("native xray: XHTTP server for inbound %q stopped: %v", ib.tag, err) } } @@ -133,19 +138,20 @@ func (ib *nativeInbound) xhttpServerMaxHeaderBytes() int { // ServeHTTP terminates the XHTTP/SplitHTTP transport and exposes the decoded // byte stream to the VLESS/VMess handlers as a net.Conn. func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) { + defer xrayRecover(fmt.Sprintf("native xray XHTTP request inbound=%q method=%s path=%q remote=%s", ib.tag, r.Method, r.URL.RequestURI(), r.RemoteAddr)) if !ib.isXHTTP() { - log.Printf("native xray: xhttp reject inbound=%q reason=not-xhttp method=%s path=%q host=%q remote=%s", ib.tag, r.Method, r.URL.RequestURI(), r.Host, r.RemoteAddr) + xrayLogf("native xray: xhttp reject inbound=%q reason=not-xhttp method=%s path=%q host=%q remote=%s", ib.tag, r.Method, r.URL.RequestURI(), r.Host, r.RemoteAddr) xhttpBadRequest(w) return } if !ib.xhttpHostAllowed(r.Host) { - log.Printf("native xray: xhttp reject inbound=%q reason=host method=%s path=%q host=%q want=%q remote=%s", ib.tag, r.Method, r.URL.RequestURI(), r.Host, ib.xhttpHost, r.RemoteAddr) + xrayLogf("native xray: xhttp reject inbound=%q reason=host method=%s path=%q host=%q want=%q remote=%s", ib.tag, r.Method, r.URL.RequestURI(), r.Host, ib.xhttpHost, r.RemoteAddr) xhttpBadRequest(w) return } base, ok := ib.matchXHTTPPath(r.URL.Path) if !ok { - log.Printf("native xray: xhttp reject inbound=%q reason=path method=%s path=%q want=%q host=%q remote=%s", ib.tag, r.Method, r.URL.RequestURI(), ib.path, r.Host, r.RemoteAddr) + xrayLogf("native xray: xhttp reject inbound=%q reason=path method=%s path=%q want=%q host=%q remote=%s", ib.tag, r.Method, r.URL.RequestURI(), ib.path, r.Host, r.RemoteAddr) xhttpBadRequest(w) return } @@ -158,7 +164,7 @@ func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) { sessionID, seqStr := ib.extractXHTTPMeta(r, base) mode := ib.normalizedXHTTPMode() - log.Printf("native xray: xhttp request inbound=%q method=%s proto=%s path=%q host=%q session=%q seq=%q len=%d mode=%s remote=%s", ib.tag, r.Method, r.Proto, r.URL.RequestURI(), r.Host, sessionID, seqStr, r.ContentLength, mode, r.RemoteAddr) + xrayTracef("native xray: xhttp request inbound=%q method=%s proto=%s path=%q host=%q session=%q seq=%q len=%d mode=%s remote=%s", ib.tag, r.Method, r.Proto, r.URL.RequestURI(), r.Host, sessionID, seqStr, r.ContentLength, mode, r.RemoteAddr) // Xray's SplitHTTP treats GET with a sequence id as an uplink packet, not as // stream-down. Some clients use this when the upload payload is carried in @@ -166,7 +172,10 @@ func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) { // treated GET as download and dropped those packets, so normal sites such as // fast.com could authenticate but then stall with no upstream data. if r.Method == http.MethodGet && sessionID != "" && seqStr != "" { - sess := ib.upsertXHTTPSession(sessionID) + sess := ib.upsertXHTTPSession(w, sessionID) + if sess == nil { + return + } ib.handleXHTTPPacketUpload(w, r, sess, seqStr) return } @@ -179,7 +188,10 @@ func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) { xhttpBadRequest(w) return } - sess := ib.upsertXHTTPSession(sessionID) + sess := ib.upsertXHTTPSession(w, sessionID) + if sess == nil { + return + } ib.handleXHTTPDownload(w, r, sess, sessionID) return } @@ -203,7 +215,10 @@ func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - sess := ib.upsertXHTTPSession(sessionID) + sess := ib.upsertXHTTPSession(w, sessionID) + if sess == nil { + return + } if seqStr == "" { ib.handleXHTTPStreamUpload(w, r, sess) return @@ -385,39 +400,70 @@ func extractXHTTPValue(r *http.Request, placement, key string) string { return "" } -func (ib *nativeInbound) upsertXHTTPSession(id string) *nativeXHTTPSession { +func (ib *nativeInbound) upsertXHTTPSession(w http.ResponseWriter, id string) *nativeXHTTPSession { ib.xhttpMu.Lock() defer ib.xhttpMu.Unlock() if ib.xhttpSessions == nil { ib.xhttpSessions = make(map[string]*nativeXHTTPSession) } if s := ib.xhttpSessions[id]; s != nil { + s.touch() return s } + if max := ib.xhttpMaxActiveSessions(); max > 0 && len(ib.xhttpSessions) >= max { + xrayTracef("native xray: xhttp session rejected inbound=%q active=%d limit=%d", ib.tag, len(ib.xhttpSessions), max) + http.Error(w, "xhttp session limit reached", http.StatusTooManyRequests) + return nil + } s := &nativeXHTTPSession{ - id: id, - queue: newNativeXHTTPUploadQueue(ib.xhttpMaxBufferedPosts), - done: make(chan struct{}), + id: id, + queue: newNativeXHTTPUploadQueue(ib.xhttpMaxBufferedPosts), + done: make(chan struct{}), + lastSeen: time.Now(), } ib.xhttpSessions[id] = s - log.Printf("native xray: xhttp session created inbound=%q session=%q", ib.tag, id) - go ib.reapUnconnectedXHTTPSession(id, s) + xrayTracef("native xray: xhttp session created inbound=%q session=%q active=%d", ib.tag, id, len(ib.xhttpSessions)) + xrayGo(fmt.Sprintf("native xray XHTTP session reaper inbound=%q session=%q", ib.tag, id), func() { ib.reapUnconnectedXHTTPSession(id, s) }) return s } +func (ib *nativeInbound) xhttpMaxActiveSessions() int { + if nativeXHTTPMaxSessionLimit() > 0 { + return nativeXHTTPMaxSessionLimit() + } + return defaultNativeXHTTPMaxSessions +} + func (ib *nativeInbound) reapUnconnectedXHTTPSession(id string, s *nativeXHTTPSession) { - t := time.NewTimer(30 * time.Second) - defer t.Stop() - select { - case <-t.C: - s.mu.Lock() - connected := s.connected - s.mu.Unlock() - if !connected { - ib.deleteXHTTPSession(id, s) - s.close() + // Keep the cheap unconnected cleanup, but also reap stale sessions that never + // receive their paired download/close because a mobile network or CDN path died. + unconnected := time.NewTimer(20 * time.Second) + stale := time.NewTicker(30 * time.Second) + defer unconnected.Stop() + defer stale.Stop() + for { + select { + case <-unconnected.C: + s.mu.Lock() + connected := s.connected + s.mu.Unlock() + if !connected { + ib.deleteXHTTPSession(id, s) + s.close() + return + } + case <-stale.C: + s.mu.Lock() + idle := time.Since(s.lastSeen) + s.mu.Unlock() + if idle > 5*time.Minute { + ib.deleteXHTTPSession(id, s) + s.close() + return + } + case <-s.done: + return } - case <-s.done: } } @@ -430,13 +476,18 @@ func (ib *nativeInbound) deleteXHTTPSession(id string, s *nativeXHTTPSession) { } func (ib *nativeInbound) handleXHTTPStreamUpload(w http.ResponseWriter, r *http.Request, sess *nativeXHTTPSession) { - log.Printf("native xray: xhttp stream-up inbound=%q session=%q len=%d remote=%s", ib.tag, sess.id, r.ContentLength, r.RemoteAddr) + sess.touch() + xrayTracef("native xray: xhttp stream-up inbound=%q session=%q len=%d remote=%s", ib.tag, sess.id, r.ContentLength, r.RemoteAddr) if ib.xhttpMode != "" && ib.xhttpMode != "auto" && ib.xhttpMode != "stream-up" && ib.xhttpMode != "stream-down" { http.Error(w, "xhttp stream-up mode is not allowed", http.StatusBadRequest) return } - if err := sess.queue.push(nativeXHTTPPacket{Reader: r.Body}); err != nil { - http.Error(w, err.Error(), http.StatusConflict) + if err := sess.queue.pushContext(r.Context(), nativeXHTTPPacket{Reader: r.Body}, nativeXHTTPQueuePushTimeoutDuration()); err != nil { + status := http.StatusConflict + if errors.Is(err, errNativeXHTTPQueueFull) { + status = http.StatusTooManyRequests + } + http.Error(w, err.Error(), status) return } w.Header().Set("X-Accel-Buffering", "no") @@ -450,6 +501,7 @@ func (ib *nativeInbound) handleXHTTPStreamUpload(w http.ResponseWriter, r *http. } func (ib *nativeInbound) handleXHTTPPacketUpload(w http.ResponseWriter, r *http.Request, sess *nativeXHTTPSession, seqStr string) { + sess.touch() if ib.xhttpMode != "" && ib.xhttpMode != "auto" && ib.xhttpMode != "packet-up" && ib.xhttpMode != "stream-down" { http.Error(w, "xhttp packet-up mode is not allowed", http.StatusBadRequest) return @@ -464,10 +516,14 @@ func (ib *nativeInbound) handleXHTTPPacketUpload(w http.ResponseWriter, r *http. http.Error(w, err.Error(), http.StatusBadRequest) return } - log.Printf("native xray: xhttp packet-up inbound=%q session=%q seq=%d payload=%d remote=%s", ib.tag, sess.id, seq, len(payload), r.RemoteAddr) - if err := sess.queue.push(nativeXHTTPPacket{Payload: payload, Seq: seq}); err != nil { - log.Printf("native xray: xhttp packet-up push failed inbound=%q session=%q seq=%d: %v", ib.tag, sess.id, seq, err) - http.Error(w, err.Error(), http.StatusConflict) + xrayTracef("native xray: xhttp packet-up inbound=%q session=%q seq=%d payload=%d remote=%s", ib.tag, sess.id, seq, len(payload), r.RemoteAddr) + if err := sess.queue.pushContext(r.Context(), nativeXHTTPPacket{Payload: payload, Seq: seq}, nativeXHTTPQueuePushTimeoutDuration()); err != nil { + status := http.StatusConflict + if errors.Is(err, errNativeXHTTPQueueFull) { + status = http.StatusTooManyRequests + } + xrayTracef("native xray: xhttp packet-up push failed inbound=%q session=%q seq=%d: %v", ib.tag, sess.id, seq, err) + http.Error(w, err.Error(), status) return } if len(payload) == 0 { @@ -576,7 +632,8 @@ func (ib *nativeInbound) xhttpMaxPostBytes() int64 { } func (ib *nativeInbound) handleXHTTPStreamOne(w http.ResponseWriter, r *http.Request) { - log.Printf("native xray: xhttp stream-one inbound=%q len=%d remote=%s", ib.tag, r.ContentLength, r.RemoteAddr) + defer xrayRecover(fmt.Sprintf("native xray XHTTP stream-one inbound=%q remote=%s", ib.tag, r.RemoteAddr)) + xrayTracef("native xray: xhttp stream-one inbound=%q len=%d remote=%s", ib.tag, r.ContentLength, r.RemoteAddr) w.Header().Set("X-Accel-Buffering", "no") w.Header().Set("Cache-Control", "no-store") if !ib.xhttpNoSSEHeader { @@ -586,12 +643,14 @@ func (ib *nativeInbound) handleXHTTPStreamOne(w http.ResponseWriter, r *http.Req flushHTTP(w) remote := remoteAddrFromHTTPRequest(r) + resp := newNativeXHTTPResponseWriter(w) xc := &nativeXHTTPConn{ reader: r.Body, - writer: &nativeXHTTPResponseWriter{w: w}, + writer: resp, remote: remote, local: dummyLocalAddr(r), onClose: func() { + resp.close() _ = r.Body.Close() }, } @@ -600,7 +659,9 @@ func (ib *nativeInbound) handleXHTTPStreamOne(w http.ResponseWriter, r *http.Req } func (ib *nativeInbound) handleXHTTPDownload(w http.ResponseWriter, r *http.Request, sess *nativeXHTTPSession, sessionID string) { - log.Printf("native xray: xhttp stream-down inbound=%q session=%q proto=%s remote=%s", ib.tag, sessionID, r.Proto, r.RemoteAddr) + sess.touch() + defer xrayRecover(fmt.Sprintf("native xray XHTTP download inbound=%q session=%q remote=%s", ib.tag, sessionID, r.RemoteAddr)) + xrayTracef("native xray: xhttp stream-down inbound=%q session=%q proto=%s remote=%s", ib.tag, sessionID, r.Proto, r.RemoteAddr) sess.markConnected() defer ib.deleteXHTTPSession(sessionID, sess) @@ -614,27 +675,31 @@ func (ib *nativeInbound) handleXHTTPDownload(w http.ResponseWriter, r *http.Requ remote := remoteAddrFromHTTPRequest(r) var reader io.Reader = sess.queue + resp := newNativeXHTTPResponseWriter(w) xc := &nativeXHTTPConn{ reader: reader, - writer: &nativeXHTTPResponseWriter{w: w}, + writer: resp, remote: remote, local: dummyLocalAddr(r), } - xc.onClose = sess.close + xc.onClose = func() { + resp.close() + sess.close() + } ib.dispatchXHTTPConn(xc, remote) _ = xc.Close() } func (ib *nativeInbound) dispatchXHTTPConn(xc net.Conn, remote net.Addr) { - log.Printf("native xray: xhttp dispatch inbound=%q protocol=%s remote=%s", ib.tag, ib.protocol, remote) + xrayTracef("native xray: xhttp dispatch inbound=%q protocol=%s remote=%s", ib.tag, ib.protocol, remote) switch ib.protocol { case "vless": ib.handleVLESS(xc, remote) case "vmess": ib.handleVMess(xc, remote) default: - log.Printf("native xray: inbound %q XHTTP protocol %q not supported", ib.tag, ib.protocol) + xrayLogf("native xray: inbound %q XHTTP protocol %q not supported", ib.tag, ib.protocol) } } @@ -666,11 +731,19 @@ type nativeXHTTPSession struct { closeOnce sync.Once mu sync.Mutex connected bool + lastSeen time.Time +} + +func (s *nativeXHTTPSession) touch() { + s.mu.Lock() + s.lastSeen = time.Now() + s.mu.Unlock() } func (s *nativeXHTTPSession) markConnected() { s.mu.Lock() s.connected = true + s.lastSeen = time.Now() s.mu.Unlock() } @@ -736,9 +809,21 @@ func (c *nativeXHTTPConn) SetReadDeadline(t time.Time) error { func (c *nativeXHTTPConn) SetWriteDeadline(time.Time) error { return nil } type nativeXHTTPResponseWriter struct { - mu sync.Mutex - w http.ResponseWriter - closed bool + mu sync.Mutex + w http.ResponseWriter + closed bool + pendingFlush bool + lastFlush time.Time + buffered int + flushTimer *time.Timer + timerActive bool +} + +func newNativeXHTTPResponseWriter(w http.ResponseWriter) *nativeXHTTPResponseWriter { + // The handler writes/flushed headers before the proxy stream is dispatched. + // Starting lastFlush at now prevents the first tiny mux packet from forcing an + // immediate extra flush for every user. + return &nativeXHTTPResponseWriter{w: w, lastFlush: time.Now()} } func (w *nativeXHTTPResponseWriter) Write(p []byte) (int, error) { @@ -748,15 +833,62 @@ func (w *nativeXHTTPResponseWriter) Write(p []byte) (int, error) { return 0, io.ErrClosedPipe } n, err := w.w.Write(p) + if n > 0 { + w.buffered += n + } if err == nil { - flushHTTP(w.w) + w.flushMaybeLocked(false) } return n, err } +func (w *nativeXHTTPResponseWriter) flushMaybeLocked(force bool) { + now := time.Now() + if force || w.buffered >= nativeXHTTPFlushByteLimit() || now.Sub(w.lastFlush) >= nativeXHTTPFlushIntervalDuration() { + flushHTTP(w.w) + w.lastFlush = now + w.buffered = 0 + w.pendingFlush = false + w.timerActive = false + return + } + if w.pendingFlush { + return + } + w.pendingFlush = true + if w.timerActive { + return + } + w.timerActive = true + if w.flushTimer == nil { + w.flushTimer = time.AfterFunc(nativeXHTTPFlushIntervalDuration(), w.fireFlushTimer) + } else { + w.flushTimer.Reset(nativeXHTTPFlushIntervalDuration()) + } +} + +func (w *nativeXHTTPResponseWriter) fireFlushTimer() { + w.mu.Lock() + defer w.mu.Unlock() + w.timerActive = false + if w.closed || !w.pendingFlush { + return + } + flushHTTP(w.w) + w.lastFlush = time.Now() + w.buffered = 0 + w.pendingFlush = false +} + func (w *nativeXHTTPResponseWriter) close() { w.mu.Lock() - w.closed = true + if !w.closed { + if w.flushTimer != nil { + w.flushTimer.Stop() + } + w.flushMaybeLocked(true) + w.closed = true + } w.mu.Unlock() } @@ -789,12 +921,23 @@ func newNativeXHTTPUploadQueue(maxPackets int) *nativeXHTTPUploadQueue { } } -func (q *nativeXHTTPUploadQueue) push(p nativeXHTTPPacket) error { +var errNativeXHTTPQueueFull = errors.New("xhttp upload queue full") + +func (q *nativeXHTTPUploadQueue) pushContext(ctx context.Context, p nativeXHTTPPacket, timeout time.Duration) error { + if timeout <= 0 { + timeout = nativeXHTTPQueuePushTimeoutDuration() + } + t := time.NewTimer(timeout) + defer t.Stop() select { case q.pushedPackets <- p: return nil case <-q.closed: return io.ErrClosedPipe + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + return errNativeXHTTPQueueFull } }