penguinehis/DragonCoreSSH-NewWEB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix xray

Browse Source
This commit is contained in:
penguinehis
2026-07-04 20:24:20 -03:00
parent 4866f0cf10
commit ea15f1bfa1
10 changed files with 1059 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
admin/assets/app.js
+82
View File
@@ -2434,6 +2434,86 @@ function toggleUdpgwFields(on) {
el.style.pointerEvents = on ? "" : "none"; 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() { async function loadServerConfig() {
const st = document.getElementById("srvCfgStatus"); const st = document.getElementById("srvCfgStatus");
st.textContent = "Loading…"; st.textContent = "Loading…";
@@ -2503,6 +2583,7 @@ async function loadServerConfig() {
const x = c.xray || {}; const x = c.xray || {};
document.getElementById("cfgXrayEnabled").checked = !!x.enabled; document.getElementById("cfgXrayEnabled").checked = !!x.enabled;
document.getElementById("cfgXrayMode").value = xrayModeFromConfig(x); document.getElementById("cfgXrayMode").value = xrayModeFromConfig(x);
writeXrayNativeTuning(x.native_tuning || XRAY_NATIVE_TUNING_DEFAULTS.safe);
st.textContent = "Config loaded."; st.textContent = "Config loaded.";
} catch (e) { } catch (e) {
@@ -2576,6 +2657,7 @@ async function saveServerConfig() {
api_server: "127.0.0.1:10085", api_server: "127.0.0.1:10085",
online_window_seconds: 90, online_window_seconds: 90,
stats_poll_seconds: 15, stats_poll_seconds: 15,
native_tuning: readXrayNativeTuning(),
}, },
}; };
admin/index.html
+25 -1
View File
@@ -1188,6 +1188,30 @@
</label> </label>
<div class="field" style="margin-top:8px;"><label>Runtime mode</label><select id="cfgXrayMode"><option value="native">Internal native emulator</option><option value="external">External xray binary</option></select></div> <div class="field" style="margin-top:8px;"><label>Runtime mode</label><select id="cfgXrayMode"><option value="native">Internal native emulator</option><option value="external">External xray binary</option></select></div>
<div class="hint" style="margin-top:6px;color:var(--muted);">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.</div> <div class="hint" style="margin-top:6px;color:var(--muted);">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.</div>
<details style="margin-top:10px;" open>
<summary style="cursor:pointer;font-size:.76rem;font-weight:700;color:var(--text);">Native Xray scale tuning</summary>
<div class="grid2" style="margin-top:10px;gap:8px;">
<div class="field"><label>Mux sessions per connection</label><input type="number" min="1" id="cfgXrayMuxMaxSessions" placeholder="128"/></div>
<div class="field"><label>Global mux backend sessions</label><input type="number" min="1" id="cfgXrayMuxGlobalSessions" placeholder="32768"/></div>
<div class="field"><label>Mux UDP idle ms</label><input type="number" min="1000" id="cfgXrayMuxUdpIdleMs" placeholder="15000"/></div>
<div class="field"><label>Mux UDP read buffer bytes</label><input type="number" min="4096" id="cfgXrayMuxUdpRbuf" placeholder="262144"/></div>
<div class="field"><label>Mux UDP write buffer bytes</label><input type="number" min="4096" id="cfgXrayMuxUdpWbuf" placeholder="262144"/></div>
<div class="field"><label>XHTTP max active sessions</label><input type="number" min="1" id="cfgXrayXhttpMaxSessions" placeholder="16384"/></div>
<div class="field"><label>XHTTP buffered posts</label><input type="number" min="1" id="cfgXrayXhttpBufferedPosts" placeholder="64"/></div>
<div class="field"><label>XHTTP queue timeout ms</label><input type="number" min="10" id="cfgXrayXhttpQueueTimeoutMs" placeholder="250"/></div>
<div class="field"><label>XHTTP flush ms</label><input type="number" min="1" id="cfgXrayXhttpFlushMs" placeholder="5"/></div>
<div class="field"><label>XHTTP flush bytes</label><input type="number" min="1024" id="cfgXrayXhttpFlushBytes" placeholder="131072"/></div>
<div class="field"><label>HTTP/2 max streams</label><input type="number" min="1" id="cfgXrayH2MaxStreams" placeholder="1024"/></div>
<div class="field"><label>HTTP/2 upload buffer / conn</label><input type="number" min="65536" id="cfgXrayH2UploadConn" placeholder="1048576"/></div>
<div class="field"><label>HTTP/2 upload buffer / stream</label><input type="number" min="32768" id="cfgXrayH2UploadStream" placeholder="262144"/></div>
<label style="font-size:.73rem;display:flex;align-items:center;gap:5px;cursor:pointer;grid-column:1/-1"><input type="checkbox" id="cfgXrayTracePackets"/> Trace every XHTTP/mux packet <span class="hint">debug only, slows QUIC</span></label>
<div class="card-actions" style="grid-column:1/-1;">
<button class="btn btn-ghost btn-sm" type="button" onclick="setXrayNativeTuningDefaults('2k')">Apply 2K defaults</button>
<button class="btn btn-ghost btn-sm" type="button" onclick="setXrayNativeTuningDefaults('safe')">Apply safe defaults</button>
</div>
<div class="hint" style="grid-column:1/-1;margin-top:-4px;">These values are saved in the panel config and applied live on restart/reload. No systemd Environment lines are needed.</div>
</div>
</details>
</div> </div>
</div><!-- /right --> </div><!-- /right -->
@@ -1211,6 +1235,6 @@
</div><!-- /shell --> </div><!-- /shell -->
</div><!-- /app --> </div><!-- /app -->
<script defer src="assets/app.js?v=20260622dnsttfakednsfast"></script> <script defer src="assets/app.js?v=20260704xrayadmintune2"></script>
</body> </body>
</html> </html>
admin_script.js
+85
View File
@@ -1143,6 +1143,85 @@ function toggleUdpgwFields(on) {
el.style.pointerEvents = on ? "" : "none"; 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() { async function loadServerConfig() {
const st = document.getElementById("srvCfgStatus"); const st = document.getElementById("srvCfgStatus");
st.textContent = "Loading…"; st.textContent = "Loading…";
@@ -1211,6 +1290,8 @@ async function loadServerConfig() {
// Xray // Xray
const x = c.xray || {}; const x = c.xray || {};
document.getElementById("cfgXrayEnabled").checked = !!x.enabled; 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."; st.textContent = "Config loaded.";
} catch (e) { } catch (e) {
@@ -1276,8 +1357,12 @@ async function saveServerConfig() {
tls_forwarders: tlsArr, tls_forwarders: tlsArr,
xray: { xray: {
enabled: document.getElementById("cfgXrayEnabled").checked, enabled: document.getElementById("cfgXrayEnabled").checked,
mode: document.getElementById("cfgXrayMode")?.value || "native",
native: (document.getElementById("cfgXrayMode")?.value || "native") !== "external",
bin_path: "/opt/sshpanel/xray", bin_path: "/opt/sshpanel/xray",
config_file: "/opt/sshpanel/xray_config.json", config_file: "/opt/sshpanel/xray_config.json",
native_config_file: "/opt/sshpanel/xray_native_config.json",
native_tuning: readXrayNativeTuning(),
}, },
}; };
go.mod
+7 -10
View File
@@ -2,22 +2,19 @@ module shell2
go 1.25.4 go 1.25.4
require ( require golang.org/x/crypto v0.45.0
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 ( require (
github.com/flynn/noise v1.0.0 // indirect github.com/flynn/noise v1.0.0 // indirect
github.com/klauspost/cpuid/v2 v2.2.6 // indirect github.com/klauspost/cpuid/v2 v2.2.6 // indirect
github.com/klauspost/reedsolomon v1.12.0 // 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/pkg/errors v0.9.1 // indirect
github.com/tjfoc/gmsm v1.4.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/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
) )
go.sum
-17
View File
@@ -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/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= 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/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.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98= 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= 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/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 h1:I5FEp3xSwVCcEh3F5A7dofEfhXdF/bWhQWPH+XwBFno=
github.com/klauspost/reedsolomon v1.12.0/go.mod h1:EPLZJeh4l27pUGC3aXOjheaoh1I9yut7xTURiW3LQ9Y= 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/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= 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/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 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= 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 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= 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/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 h1:aMe1GlZb+0bLjn+cKTPEvvn9oUEBlJitaZiiBwsbgho=
github.com/tjfoc/gmsm v1.4.1/go.mod h1:j4INPkHWMrhJb38G+J6W4Tw0AbuN8Thu3PbdVYhVcTE= 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 h1:ajm12pGuWO+GWQNusPyPESC7Rq0yTC2rEXVYkM8ExOg=
github.com/xtaci/kcp-go/v5 v5.6.61/go.mod h1:9O3D8WR+cyyUjGiTILYfg17vn72otWuXK2AFfqIe6CM= 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 h1:y/1DlWQC9bnMeZzsyk4oL2hbLK6uVk4BKTz5BeQqUEA=
github.com/xtaci/smux v1.5.50/go.mod h1:IGQ9QYrBphmb/4aTnLEcJby0TNr3NV+OslIOMrX825Q= 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= 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 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=
golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= 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.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.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= 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 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= 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= 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.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.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= 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/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-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/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= www.bamsoftware.com/git/dnstt.git v1.20241021.0 h1:Xi0lmT+5kcgzY7P+r726eBXKMZKgGoD8GTNKrlh8TuE=
xray_integration.go
+186 -32
View File
@@ -6,7 +6,6 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"io" "io"
"log"
"net/http" "net/http"
"net/url" "net/url"
"os" "os"
@@ -49,6 +48,10 @@ type XrayConfig struct {
// clients are preserved when the server has IPv6 connectivity. force_ipv4 is // clients are preserved when the server has IPv6 connectivity. force_ipv4 is
// only kept as an explicit legacy override. // only kept as an explicit legacy override.
NativeIPStrategy string `json:"native_ip_strategy,omitempty"` // auto | force_ipv4 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 ( const (
@@ -94,6 +97,9 @@ func (c *XrayConfig) NormalizeDefaults() {
default: default:
c.NativeIPStrategy = xrayNativeIPStrategyAuto c.NativeIPStrategy = xrayNativeIPStrategyAuto
} }
tuning := normalizeNativeXrayTuning(c.NativeTuning)
c.NativeTuning = &tuning
applyNativeXrayTuning(c.NativeTuning)
} }
func (c *XrayConfig) NativeForceIPv4() bool { func (c *XrayConfig) NativeForceIPv4() bool {
@@ -155,7 +161,12 @@ type xrayLogRing struct {
pos int pos int
} }
const xrayLogCap = 200 const (
xrayLogCap = 5000
xrayForcedLogLevel = "debug"
xrayForcedAccessLogPath = "/dev/stdout"
xrayForcedErrorLogPath = "/dev/stderr"
)
func (r *xrayLogRing) add(line string) { func (r *xrayLogRing) add(line string) {
r.mu.Lock() r.mu.Lock()
@@ -200,6 +211,38 @@ func (w xrayWriter) Write(p []byte) (int, error) {
return os.Stderr.Write(p) 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. // XrayManager manages the lifecycle of the external xray subprocess.
type XrayManager struct { type XrayManager struct {
mu sync.Mutex mu sync.Mutex
@@ -249,7 +292,7 @@ func initXrayManager(cfg *XrayConfig) {
xrayMgr.mu.Lock() xrayMgr.mu.Lock()
xrayMgr.cfg = cfg xrayMgr.cfg = cfg
if err := xrayMgr.bootstrapConfigStoreLocked(); err != nil { 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() xrayMgr.mu.Unlock()
@@ -263,7 +306,7 @@ func initXrayManager(cfg *XrayConfig) {
if cfg.Enabled { if cfg.Enabled {
if err := xrayMgr.Start(); err != nil { 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 { if m.cfg == nil {
return fmt.Errorf("xray not configured") return fmt.Errorf("xray not configured")
} }
m.cfg.NormalizeDefaults()
if err := m.syncConfigFileFromStoreLocked(); err != nil { if err := m.syncConfigFileFromStoreLocked(); err != nil {
m.lastErr = err.Error() m.lastErr = err.Error()
return err return err
} }
configFile := m.activeConfigFileLocked() 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. // Native mode: run the in-process emulator instead of the subprocess.
if m.cfg.UseNative() { if m.cfg.UseNative() {
@@ -319,7 +367,7 @@ func (m *XrayManager) Start() error {
if changed, err := m.ensureStatsAPIConfigLocked(); err != nil { if changed, err := m.ensureStatsAPIConfigLocked(); err != nil {
return fmt.Errorf("xray stats api check failed: %w", err) return fmt.Errorf("xray stats api check failed: %w", err)
} else if changed { } 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"} args := []string{"run"}
@@ -350,10 +398,10 @@ func (m *XrayManager) Start() error {
m.lastErr = err.Error() m.lastErr = err.Error()
} }
m.mu.Unlock() 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 return nil
} }
@@ -386,7 +434,7 @@ func (m *XrayManager) Stop() error {
case <-time.After(2 * time.Second): case <-time.After(2 * time.Second):
} }
} }
log.Printf("xray: stopped") xrayLogf("xray: stopped")
return nil return nil
} }
@@ -423,13 +471,13 @@ func (m *XrayManager) recordNativeConnect(uuid, email string) {
m.statsMu.Unlock() m.statsMu.Unlock()
if statsStore != nil && uuid != "" { if statsStore != nil && uuid != "" {
go func() { xrayGo("native xray stats active increment", func() {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel() defer cancel()
if err := statsStore.UpdateXrayClientActive(ctx, uuid, email, 1); err != nil { 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() m.statsMu.Unlock()
if statsStore != nil && uuid != "" { if statsStore != nil && uuid != "" {
go func() { xrayGo("native xray stats active decrement", func() {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel() defer cancel()
if err := statsStore.UpdateXrayClientActive(ctx, uuid, email, -1); err != nil { 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.nativeStatsFlushStarted = true
m.nativeDBMu.Unlock() m.nativeDBMu.Unlock()
go func() { xrayGo("native xray stats flusher", func() {
ticker := time.NewTicker(5 * time.Second) ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop() defer ticker.Stop()
for range ticker.C { for range ticker.C {
m.flushNativeStatsToDB() m.flushNativeStatsToDB()
} }
}() })
} }
func (m *XrayManager) flushNativeStatsToDB() { func (m *XrayManager) flushNativeStatsToDB() {
@@ -534,7 +582,7 @@ func (m *XrayManager) flushNativeStatsToDB() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel() defer cancel()
if err := statsStore.AddXrayClientTrafficBatch(ctx, pending); err != nil { 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. // Put deltas back so a transient DB failure does not lose accounting.
m.nativeDBMu.Lock() m.nativeDBMu.Lock()
if m.nativeTrafficPending == nil { if m.nativeTrafficPending == nil {
@@ -964,6 +1012,55 @@ func normalizeJSONIndent(data []byte) ([]byte, error) {
return json.MarshalIndent(raw, "", " ") 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) { func (m *XrayManager) readConfigLocked() ([]byte, error) {
configFile := m.activeConfigFileLocked() configFile := m.activeConfigFileLocked()
if m.cfg == nil || configFile == "" { if m.cfg == nil || configFile == "" {
@@ -971,8 +1068,15 @@ func (m *XrayManager) readConfigLocked() ([]byte, error) {
} }
if statsStore != nil { if statsStore != nil {
if data, ok, err := statsStore.GetXrayConfig(context.Background(), m.configStoreKeyLocked()); err != 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 { } 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) pretty, err := normalizeJSONIndent(data)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -981,7 +1085,18 @@ func (m *XrayManager) readConfigLocked() ([]byte, error) {
return pretty, nil 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 { func (m *XrayManager) writeConfigLocked(data []byte) error {
@@ -989,6 +1104,12 @@ func (m *XrayManager) writeConfigLocked(data []byte) error {
if m.cfg == nil || configFile == "" { if m.cfg == nil || configFile == "" {
return fmt.Errorf("xray config file not configured") 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) pretty, err := normalizeJSONIndent(data)
if err != nil { if err != nil {
return err return err
@@ -1011,11 +1132,11 @@ func (m *XrayManager) importConfigClientsLocked(data []byte, source string) {
} }
n, err := statsStore.ImportXrayClientsFromConfig(context.Background(), data) n, err := statsStore.ImportXrayClientsFromConfig(context.Background(), data)
if err != nil { 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 return
} }
if n > 0 { 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) data, err := os.ReadFile(configFile)
if err != nil { if err != nil {
if !os.IsNotExist(err) { 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 return
} }
pretty, err := normalizeJSONIndent(data) pretty, err := normalizeJSONIndent(data)
if err != nil { 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 return
} }
m.importConfigClientsLocked(pretty, source) m.importConfigClientsLocked(pretty, source)
@@ -1048,7 +1169,7 @@ func (m *XrayManager) restartIfExternalRunning() {
return return
} }
if err := m.Restart(); err != nil { 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 { if data, ok, err := statsStore.GetXrayConfig(ctx, key); err != nil {
return err return err
} else if ok { } 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) pretty, err := normalizeJSONIndent(data)
if err != nil { if err != nil {
return err return err
@@ -1086,6 +1216,12 @@ func (m *XrayManager) bootstrapConfigStoreLocked() error {
if migrated, migErr := m.seedNativeConfigFromExternalLocked(configFile); migErr != nil { if migrated, migErr := m.seedNativeConfigFromExternalLocked(configFile); migErr != nil {
return migErr return migErr
} else if len(migrated) > 0 { } 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) pretty, err := normalizeJSONIndent(migrated)
if err != nil { if err != nil {
return err return err
@@ -1102,6 +1238,12 @@ func (m *XrayManager) bootstrapConfigStoreLocked() error {
} }
return err 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) pretty, err := normalizeJSONIndent(data)
if err != nil { if err != nil {
return err return err
@@ -1128,7 +1270,7 @@ func (m *XrayManager) seedNativeConfigFromExternalLocked(nativeConfigFile string
if _, err := normalizeJSONIndent(data); err != nil { if _, err := normalizeJSONIndent(data); err != nil {
return nil, err 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 return data, nil
} }
@@ -1141,6 +1283,15 @@ func (m *XrayManager) syncConfigFileFromStoreLocked() error {
if err != nil || !ok { if err != nil || !ok {
return err 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) pretty, err := normalizeJSONIndent(data)
if err != nil { if err != nil {
return err return err
@@ -1166,7 +1317,7 @@ func (m *XrayManager) SetConfig(data []byte) error {
return err return err
} }
if changed { 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) 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") return nil, false, fmt.Errorf("xray config must be a JSON object")
} }
changed, _ := ensureXrayStatsAPIConfig(raw) changed, _ := ensureXrayStatsAPIConfig(raw)
if ensureXrayForcedDebugLogConfig(raw) {
changed = true
}
if !changed { if !changed {
return data, false, nil return data, false, nil
} }
@@ -1666,7 +1820,7 @@ func handleXrayStatus(w http.ResponseWriter, r *http.Request) {
wasRunning := xrayMgr.isRunningSnapshot() wasRunning := xrayMgr.isRunningSnapshot()
if changed, err := xrayMgr.EnsureStatsAPIConfig(); err == nil && changed && wasRunning { if changed, err := xrayMgr.EnsureStatsAPIConfig(); err == nil && changed && wasRunning {
if err := xrayMgr.Restart(); err != nil { 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 err == nil && m.cfg != nil && m.cfg.UseNative() {
if hotErr := nativeXray.addClient(inboundTag, uuid, email); hotErr != nil { 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 return err
@@ -2044,7 +2198,7 @@ func (m *XrayManager) RemoveXrayClient(inboundTag, uuid string) error {
}) })
if err == nil && m.cfg != nil && m.cfg.UseNative() { if err == nil && m.cfg != nil && m.cfg.UseNative() {
if hotErr := nativeXray.removeClient(inboundTag, uuid); hotErr != nil { 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 return err
@@ -2090,7 +2244,7 @@ func (m *XrayManager) UpdateXrayClientEmail(uuid, email string) error {
}) })
if err == nil && m.cfg != nil && m.cfg.UseNative() { if err == nil && m.cfg != nil && m.cfg.UseNative() {
if hotErr := nativeXray.updateClientEmail(uuid, email); hotErr != nil { 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 return err
@@ -2359,7 +2513,7 @@ func handleXrayClientAdd(w http.ResponseWriter, r *http.Request) {
} }
} }
if err := statsStore.UpsertXrayClientMeta(r.Context(), meta); err != nil { 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() xrayMgr.restartIfExternalRunning()
@@ -2445,7 +2599,7 @@ func handleXrayClientUpdate(w http.ResponseWriter, r *http.Request) {
} }
if req.Email != "" { if req.Email != "" {
if err := xrayMgr.UpdateXrayClientEmail(req.UUID, req.Email); err != nil { 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 { } else {
xrayMgr.restartIfExternalRunning() xrayMgr.restartIfExternalRunning()
} }
xray_native.go
+146 -56
View File
@@ -9,11 +9,12 @@ package main
// //
// Native emulator scope: // Native emulator scope:
// - Protocols : VLESS and VMess AEAD (TCP + UDP commands) // - 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 // - Transports: raw TCP, WebSocket (RFC 6455), XHTTP/SplitHTTP
// - Security : TLS, none // - Security : TLS, none
// //
// Mux, REALITY, gRPC and HTTPUpgrade are still deferred; unsupported commands // REALITY, gRPC and HTTPUpgrade are still deferred; unsupported commands are
// are rejected explicitly instead of silently falling back. // rejected explicitly instead of silently falling back.
// //
// Native mode has its own DB-backed config/runtime path. It does not spawn or // 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 // query the external xray binary and does not require /opt/sshpanel/xray to be
@@ -34,7 +35,6 @@ import (
"errors" "errors"
"fmt" "fmt"
"io" "io"
"log"
"net" "net"
"net/http" "net/http"
"os" "os"
@@ -160,13 +160,13 @@ func (s *nativeXrayServer) start(configFile string) error {
serveLn = tls.NewListener(ln, ib.tlsConfig) serveLn = tls.NewListener(ln, ib.tlsConfig)
} }
opened = append(opened, serveLn) opened = append(opened, serveLn)
go ib.serveXHTTPListener(serveLn) xrayGo(fmt.Sprintf("native xray xhttp listener %s", addr), func() { ib.serveXHTTPListener(serveLn) })
} else { } else {
opened = append(opened, serveLn) 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 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()) ib.protocol, ib.transport, addr, ib.tag, orNone(ib.security), ib.clientCount())
} }
@@ -189,25 +189,27 @@ func (s *nativeXrayServer) stop() {
s.listeners = nil s.listeners = nil
s.inboundsByTag = nil s.inboundsByTag = nil
s.running = false s.running = false
log.Printf("native xray: stopped") xrayLogf("native xray: stopped")
} }
func (ib *nativeInbound) acceptLoop(ln net.Listener) { func (ib *nativeInbound) acceptLoop(ln net.Listener) {
defer xrayRecover(fmt.Sprintf("native xray accept loop inbound=%q", ib.tag))
for { for {
c, err := ln.Accept() c, err := ln.Accept()
if err != nil { if err != nil {
if isListenerClosed(err) { if isListenerClosed(err) {
return 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 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. // serve terminates TLS + transport, then dispatches on protocol.
func (ib *nativeInbound) serve(raw net.Conn) { 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() defer raw.Close()
if tc, ok := raw.(*net.TCPConn); ok { 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 := tls.Server(raw, ib.tlsConfig)
_ = tconn.SetDeadline(time.Now().Add(tlsHandshakeTimeout)) _ = tconn.SetDeadline(time.Now().Add(tlsHandshakeTimeout))
if err := tconn.Handshake(); err != nil { 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 return
} }
_ = tconn.SetDeadline(time.Time{}) _ = tconn.SetDeadline(time.Time{})
@@ -237,15 +239,15 @@ func (ib *nativeInbound) serve(raw net.Conn) {
case "ws", "websocket": case "ws", "websocket":
ws, err := wsServerHandshake(conn, ib.path) ws, err := wsServerHandshake(conn, ib.path)
if err != nil { 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 return
} }
stream = ws stream = ws
case "xhttp", "splithttp": 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 return
default: 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()) ib.tag, ib.transport, raw.RemoteAddr())
return return
} }
@@ -257,7 +259,7 @@ func (ib *nativeInbound) serve(raw net.Conn) {
case "vmess": case "vmess":
ib.handleVMess(stream, raw.RemoteAddr()) ib.handleVMess(stream, raw.RemoteAddr())
default: 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()) ib.tag, ib.protocol, raw.RemoteAddr())
} }
} }
@@ -270,9 +272,12 @@ func (ib *nativeInbound) serve(raw net.Conn) {
// 1 byte addon length M // 1 byte addon length M
// M bytes addons (flow etc.) — skipped // M bytes addons (flow etc.) — skipped
// 1 byte command (1=TCP, 2=UDP, 3=Mux) // 1 byte command (1=TCP, 2=UDP, 3=Mux)
// 2 bytes port (big endian) // for TCP/UDP only:
// 1 byte address type (1=IPv4, 2=domain, 3=IPv6) // 2 bytes port (big endian)
// ... address // 1 byte address type (1=IPv4, 2=domain, 3=IPv6)
// ... address
// for Mux:
// ... Mux.Cool/XUDP frames immediately after the command byte
// ... payload // ... payload
// Response (server -> client): 1 byte version echo, 1 byte addon length (0). // 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) { 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)) _ = stream.SetReadDeadline(time.Now().Add(30 * time.Second))
head := make([]byte, 1+16+1) // version + uuid + addonLen head := make([]byte, 1+16+1) // version + uuid + addonLen
if _, err := io.ReadFull(stream, head); err != nil { 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 return
} }
version := head[0] version := head[0]
@@ -301,39 +311,59 @@ func (ib *nativeInbound) handleVLESS(stream net.Conn, remote net.Addr) {
client := ib.getNativeClient(id) client := ib.getNativeClient(id)
if client == nil { 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 return
} }
if addonLen := int(head[17]); addonLen > 0 { if addonLen := int(head[17]); addonLen > 0 {
if _, err := io.CopyN(io.Discard, stream, int64(addonLen)); err != nil { 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 return
} }
} }
var cmd [1]byte var cmd [1]byte
if _, err := io.ReadFull(stream, cmd[:]); err != nil { 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 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) var host string
if err != nil { var port uint16
log.Printf("native xray: inbound %q VLESS bad address from %s: %v", ib.tag, remote, err) if cmd[0] == vlessCmdTCP || cmd[0] == vlessCmdUDP {
return 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{}) _ = 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 { 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 return
} }
@@ -341,24 +371,79 @@ func (ib *nativeInbound) handleVLESS(stream net.Conn, remote net.Addr) {
case vlessCmdTCP: case vlessCmdTCP:
backend, target, err := ib.nativeDialTCP(host, port) backend, target, err := ib.nativeDialTCP(host, port)
if err != nil { 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 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()) nativeTunnel(stream, backend, client.uuid, client.email, ib.upLimiter(), ib.downLimiter())
case vlessCmdUDP: case vlessCmdUDP:
backend, target, err := ib.nativeDialUDP(host, port) backend, target, err := ib.nativeDialUDP(host, port)
if err != nil { 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 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()) 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: 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). // readProxyAddress reads a VMess/VLESS-style address (type byte + address).
func readProxyAddress(r io.Reader) (string, error) { func readProxyAddress(r io.Reader) (string, error) {
var t [1]byte var t [1]byte
@@ -445,13 +530,13 @@ func nativeDialTargetWithSource(network, host string, port uint16, sourceHost st
cancel() cancel()
if err == nil { if err == nil {
if i > 0 && len(attempts) > 1 { 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 return conn, target, nil
} }
lastErr = err lastErr = err
if local != nil { 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 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) { func nativeTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, email string, up, down *rate.Limiter) {
xrayMgr.recordNativeConnect(uuid, email) xrayMgr.recordNativeConnect(uuid, email)
defer xrayMgr.recordNativeDisconnect(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} upMeter := &trafficMeter{uuid: uuid, email: email, uplink: true}
downMeter := &trafficMeter{uuid: uuid, email: email, uplink: false} downMeter := &trafficMeter{uuid: uuid, email: email, uplink: false}
var wg sync.WaitGroup var wg sync.WaitGroup
var closeOnce sync.Once
closeAll := func() { closeAll := func() {
_ = backend.Close() closeOnce.Do(func() {
_ = client.Close() _ = backend.Close()
_ = client.Close()
})
} }
wg.Add(1) wg.Add(1)
go func() { // client -> backend (uplink) xrayGo("native xray TCP uplink", func() { // client -> backend
defer wg.Done() defer wg.Done()
defer closeAll()
_, _ = copyWithRateLimit(meteredWriter{w: backend, meter: upMeter}, client, up) _, _ = copyWithRateLimit(meteredWriter{w: backend, meter: upMeter}, client, up)
if cw, ok := backend.(interface{ CloseWrite() error }); ok { if cw, ok := backend.(interface{ CloseWrite() error }); ok {
_ = cw.CloseWrite() _ = cw.CloseWrite()
} }
closeAll() })
}()
wg.Add(1) wg.Add(1)
go func() { // backend -> client (downlink) xrayGo("native xray TCP downlink", func() { // backend -> client
defer wg.Done() defer wg.Done()
defer closeAll()
_, _ = copyWithRateLimit(meteredWriter{w: client, meter: downMeter}, backend, down) _, _ = copyWithRateLimit(meteredWriter{w: client, meter: downMeter}, backend, down)
closeAll() })
}()
wg.Wait() wg.Wait()
upMeter.flush() upMeter.flush()
@@ -853,7 +943,7 @@ func parseNativeInbounds(configFile string) ([]*nativeInbound, error) {
} }
port, ok := parseSinglePort(in.Port) port, ok := parseSinglePort(in.Port)
if !ok { 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 continue
} }
@@ -894,7 +984,7 @@ func parseNativeInbounds(configFile string) ([]*nativeInbound, error) {
} }
ib.xhttpMaxBufferedPosts = xh.ScMaxBufferedPosts ib.xhttpMaxBufferedPosts = xh.ScMaxBufferedPosts
if ib.xhttpMaxBufferedPosts <= 0 { if ib.xhttpMaxBufferedPosts <= 0 {
ib.xhttpMaxBufferedPosts = 30 ib.xhttpMaxBufferedPosts = nativeXHTTPBufferedPostLimit()
} }
ib.xhttpSessions = make(map[string]*nativeXHTTPSession) ib.xhttpSessions = make(map[string]*nativeXHTTPSession)
} }
@@ -904,7 +994,7 @@ func parseNativeInbounds(configFile string) ([]*nativeInbound, error) {
if ib.security == "tls" { if ib.security == "tls" {
tc, err := buildInboundTLS(in) tc, err := buildInboundTLS(in)
if err != nil { 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 continue
} }
ib.tlsConfig = tc ib.tlsConfig = tc
@@ -920,24 +1010,24 @@ func parseNativeInbounds(configFile string) ([]*nativeInbound, error) {
raw = c.Password // some protocols reuse password as id raw = c.Password // some protocols reuse password as id
} }
if err := ib.addNativeClient(proto, raw, c.Email); err != nil { 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 != "" { if statsStore != nil && in.Tag != "" {
metas, err := statsStore.ListXrayClientsByInbound(context.Background(), in.Tag) metas, err := statsStore.ListXrayClientsByInbound(context.Background(), in.Tag)
if err != nil { 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 { } else {
for _, m := range metas { for _, m := range metas {
if err := ib.addNativeClient(proto, m.UUID, firstNonEmpty(m.Email, m.Name, m.UUID)); err != nil { 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 { 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 continue
} }
out = append(out, ib) out = append(out, ib)
xray_native_test.go
+300
View File
@@ -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) { func TestVLESSRejectsUnknownUUID(t *testing.T) {
echoPort, stopEcho := startEchoServer(t) echoPort, stopEcho := startEchoServer(t)
defer stopEcho() defer stopEcho()
@@ -578,3 +638,243 @@ func TestNativeLocalAddrForIPv6Tunnel(t *testing.T) {
t.Fatalf("must not bind IPv4 source to IPv6 target: %#v", local) 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)
}
}
xray_native_udp.go
+38 -27
View File
@@ -5,7 +5,6 @@ import (
"encoding/binary" "encoding/binary"
"fmt" "fmt"
"io" "io"
"log"
"net" "net"
"sync" "sync"
"time" "time"
@@ -29,24 +28,29 @@ const (
func nativeVLESSUDPTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, email string, up, down *rate.Limiter) { func nativeVLESSUDPTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, email string, up, down *rate.Limiter) {
xrayMgr.recordNativeConnect(uuid, email) xrayMgr.recordNativeConnect(uuid, email)
defer xrayMgr.recordNativeDisconnect(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} upMeter := &trafficMeter{uuid: uuid, email: email, uplink: true}
downMeter := &trafficMeter{uuid: uuid, email: email, uplink: false} downMeter := &trafficMeter{uuid: uuid, email: email, uplink: false}
var wg sync.WaitGroup var wg sync.WaitGroup
var closeOnce sync.Once
closeAll := func() { closeAll := func() {
_ = backend.Close() closeOnce.Do(func() {
_ = client.Close() _ = backend.Close()
_ = client.Close()
})
} }
wg.Add(1) wg.Add(1)
go func() { xrayGo("native xray VLESS UDP uplink", func() {
defer wg.Done() defer wg.Done()
defer closeAll()
for { for {
payload, err := readVLESSLengthPacket(client) payload, err := readVLESSLengthPacket(client)
if err != nil { if err != nil {
if err != io.EOF { 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 return
} }
@@ -61,15 +65,16 @@ func nativeVLESSUDPTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, ema
upMeter.add(n) upMeter.add(n)
} }
if err != nil { 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 return
} }
} }
}() })
wg.Add(1) wg.Add(1)
go func() { xrayGo("native xray VLESS UDP downlink", func() {
defer wg.Done() defer wg.Done()
defer closeAll()
buf := make([]byte, nativeUDPBufferSize) buf := make([]byte, nativeUDPBufferSize)
for { for {
_ = backend.SetReadDeadline(time.Now().Add(nativeUDPIdle)) _ = backend.SetReadDeadline(time.Now().Add(nativeUDPIdle))
@@ -79,7 +84,7 @@ func nativeVLESSUDPTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, ema
return return
} }
if err != io.EOF { 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 return
} }
@@ -90,12 +95,12 @@ func nativeVLESSUDPTunnel(client io.ReadWriteCloser, backend net.Conn, uuid, ema
return return
} }
if err := writeVLESSLengthPacket(client, buf[:n]); err != nil { 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 return
} }
downMeter.add(n) downMeter.add(n)
} }
}() })
wg.Wait() wg.Wait()
upMeter.flush() upMeter.flush()
@@ -198,12 +203,12 @@ func writeVLESSLengthPacket(w io.Writer, payload []byte) error {
if len(payload) > nativeUDPMaxPacket { if len(payload) > nativeUDPMaxPacket {
return fmt.Errorf("udp packet too large: %d", len(payload)) return fmt.Errorf("udp packet too large: %d", len(payload))
} }
var lenBuf [2]byte // One Write is important for XHTTP because the response writer flushes once per
binary.BigEndian.PutUint16(lenBuf[:], uint16(len(payload))) // Write. Two writes per UDP packet doubles flush/syscall pressure.
if _, err := w.Write(lenBuf[:]); err != nil { frame := make([]byte, 0, 2+len(payload))
return err frame = appendUint16(frame, uint16(len(payload)))
} frame = append(frame, payload...)
_, err := w.Write(payload) _, err := w.Write(frame)
return err 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) { func nativeVMessUDPTunnel(client nativeVMessStream, backend net.Conn, uuid, email string, up, down *rate.Limiter) {
xrayMgr.recordNativeConnect(uuid, email) xrayMgr.recordNativeConnect(uuid, email)
defer xrayMgr.recordNativeDisconnect(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} upMeter := &trafficMeter{uuid: uuid, email: email, uplink: true}
downMeter := &trafficMeter{uuid: uuid, email: email, uplink: false} downMeter := &trafficMeter{uuid: uuid, email: email, uplink: false}
var wg sync.WaitGroup var wg sync.WaitGroup
var closeOnce sync.Once
closeAll := func() { closeAll := func() {
_ = backend.Close() closeOnce.Do(func() {
_ = client.Close() _ = backend.Close()
_ = client.Close()
})
} }
wg.Add(1) wg.Add(1)
go func() { xrayGo("native xray VMess UDP uplink", func() {
defer wg.Done() defer wg.Done()
defer closeAll()
for { for {
pkt, err := client.ReadPacket() pkt, err := client.ReadPacket()
if err != nil { if err != nil {
if err != io.EOF { 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 return
} }
@@ -331,15 +341,16 @@ func nativeVMessUDPTunnel(client nativeVMessStream, backend net.Conn, uuid, emai
upMeter.add(n) upMeter.add(n)
} }
if err != nil { 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 return
} }
} }
}() })
wg.Add(1) wg.Add(1)
go func() { xrayGo("native xray VMess UDP downlink", func() {
defer wg.Done() defer wg.Done()
defer closeAll()
buf := make([]byte, nativeUDPBufferSize) buf := make([]byte, nativeUDPBufferSize)
for { for {
_ = backend.SetReadDeadline(time.Now().Add(nativeUDPIdle)) _ = backend.SetReadDeadline(time.Now().Add(nativeUDPIdle))
@@ -349,7 +360,7 @@ func nativeVMessUDPTunnel(client nativeVMessStream, backend net.Conn, uuid, emai
return return
} }
if err != io.EOF { 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 return
} }
@@ -360,12 +371,12 @@ func nativeVMessUDPTunnel(client nativeVMessStream, backend net.Conn, uuid, emai
return return
} }
if err := client.WritePacket(buf[:n]); err != nil { 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 return
} }
downMeter.add(n) downMeter.add(n)
} }
}() })
wg.Wait() wg.Wait()
upMeter.flush() upMeter.flush()
xray_xhttp.go
+190 -47
View File
@@ -2,11 +2,11 @@ package main
import ( import (
"container/heap" "container/heap"
"context"
"encoding/base64" "encoding/base64"
"errors" "errors"
"fmt" "fmt"
"io" "io"
"log"
"net" "net"
"net/http" "net/http"
"os" "os"
@@ -98,7 +98,12 @@ func mergeNativeXHTTPSettings(primary, fallback nativeXHTTPSettingsJSON) nativeX
} }
func (ib *nativeInbound) serveXHTTPListener(ln net.Listener) { 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) handler := http.Handler(ib)
// Official Xray accepts plaintext HTTP/1.1 and h2c on non-TLS XHTTP // Official Xray accepts plaintext HTTP/1.1 and h2c on non-TLS XHTTP
// listeners, and negotiates h2/http1 through ALPN on TLS listeners. Without // 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) _ = http2.ConfigureServer(srv, h2s)
} }
if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) && !isListenerClosed(err) { 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 // ServeHTTP terminates the XHTTP/SplitHTTP transport and exposes the decoded
// byte stream to the VLESS/VMess handlers as a net.Conn. // byte stream to the VLESS/VMess handlers as a net.Conn.
func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) { 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() { 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) xhttpBadRequest(w)
return return
} }
if !ib.xhttpHostAllowed(r.Host) { 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) xhttpBadRequest(w)
return return
} }
base, ok := ib.matchXHTTPPath(r.URL.Path) base, ok := ib.matchXHTTPPath(r.URL.Path)
if !ok { 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) xhttpBadRequest(w)
return return
} }
@@ -158,7 +164,7 @@ func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) {
sessionID, seqStr := ib.extractXHTTPMeta(r, base) sessionID, seqStr := ib.extractXHTTPMeta(r, base)
mode := ib.normalizedXHTTPMode() 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 // 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 // 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 // treated GET as download and dropped those packets, so normal sites such as
// fast.com could authenticate but then stall with no upstream data. // fast.com could authenticate but then stall with no upstream data.
if r.Method == http.MethodGet && sessionID != "" && seqStr != "" { 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) ib.handleXHTTPPacketUpload(w, r, sess, seqStr)
return return
} }
@@ -179,7 +188,10 @@ func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) {
xhttpBadRequest(w) xhttpBadRequest(w)
return return
} }
sess := ib.upsertXHTTPSession(sessionID) sess := ib.upsertXHTTPSession(w, sessionID)
if sess == nil {
return
}
ib.handleXHTTPDownload(w, r, sess, sessionID) ib.handleXHTTPDownload(w, r, sess, sessionID)
return return
} }
@@ -203,7 +215,10 @@ func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return return
} }
sess := ib.upsertXHTTPSession(sessionID) sess := ib.upsertXHTTPSession(w, sessionID)
if sess == nil {
return
}
if seqStr == "" { if seqStr == "" {
ib.handleXHTTPStreamUpload(w, r, sess) ib.handleXHTTPStreamUpload(w, r, sess)
return return
@@ -385,39 +400,70 @@ func extractXHTTPValue(r *http.Request, placement, key string) string {
return "" return ""
} }
func (ib *nativeInbound) upsertXHTTPSession(id string) *nativeXHTTPSession { func (ib *nativeInbound) upsertXHTTPSession(w http.ResponseWriter, id string) *nativeXHTTPSession {
ib.xhttpMu.Lock() ib.xhttpMu.Lock()
defer ib.xhttpMu.Unlock() defer ib.xhttpMu.Unlock()
if ib.xhttpSessions == nil { if ib.xhttpSessions == nil {
ib.xhttpSessions = make(map[string]*nativeXHTTPSession) ib.xhttpSessions = make(map[string]*nativeXHTTPSession)
} }
if s := ib.xhttpSessions[id]; s != nil { if s := ib.xhttpSessions[id]; s != nil {
s.touch()
return s 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{ s := &nativeXHTTPSession{
id: id, id: id,
queue: newNativeXHTTPUploadQueue(ib.xhttpMaxBufferedPosts), queue: newNativeXHTTPUploadQueue(ib.xhttpMaxBufferedPosts),
done: make(chan struct{}), done: make(chan struct{}),
lastSeen: time.Now(),
} }
ib.xhttpSessions[id] = s ib.xhttpSessions[id] = s
log.Printf("native xray: xhttp session created inbound=%q session=%q", ib.tag, id) xrayTracef("native xray: xhttp session created inbound=%q session=%q active=%d", ib.tag, id, len(ib.xhttpSessions))
go ib.reapUnconnectedXHTTPSession(id, s) xrayGo(fmt.Sprintf("native xray XHTTP session reaper inbound=%q session=%q", ib.tag, id), func() { ib.reapUnconnectedXHTTPSession(id, s) })
return s return s
} }
func (ib *nativeInbound) xhttpMaxActiveSessions() int {
if nativeXHTTPMaxSessionLimit() > 0 {
return nativeXHTTPMaxSessionLimit()
}
return defaultNativeXHTTPMaxSessions
}
func (ib *nativeInbound) reapUnconnectedXHTTPSession(id string, s *nativeXHTTPSession) { func (ib *nativeInbound) reapUnconnectedXHTTPSession(id string, s *nativeXHTTPSession) {
t := time.NewTimer(30 * time.Second) // Keep the cheap unconnected cleanup, but also reap stale sessions that never
defer t.Stop() // receive their paired download/close because a mobile network or CDN path died.
select { unconnected := time.NewTimer(20 * time.Second)
case <-t.C: stale := time.NewTicker(30 * time.Second)
s.mu.Lock() defer unconnected.Stop()
connected := s.connected defer stale.Stop()
s.mu.Unlock() for {
if !connected { select {
ib.deleteXHTTPSession(id, s) case <-unconnected.C:
s.close() 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) { 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" { 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) http.Error(w, "xhttp stream-up mode is not allowed", http.StatusBadRequest)
return return
} }
if err := sess.queue.push(nativeXHTTPPacket{Reader: r.Body}); err != nil { if err := sess.queue.pushContext(r.Context(), nativeXHTTPPacket{Reader: r.Body}, nativeXHTTPQueuePushTimeoutDuration()); err != nil {
http.Error(w, err.Error(), http.StatusConflict) status := http.StatusConflict
if errors.Is(err, errNativeXHTTPQueueFull) {
status = http.StatusTooManyRequests
}
http.Error(w, err.Error(), status)
return return
} }
w.Header().Set("X-Accel-Buffering", "no") 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) { 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" { 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) http.Error(w, "xhttp packet-up mode is not allowed", http.StatusBadRequest)
return return
@@ -464,10 +516,14 @@ func (ib *nativeInbound) handleXHTTPPacketUpload(w http.ResponseWriter, r *http.
http.Error(w, err.Error(), http.StatusBadRequest) http.Error(w, err.Error(), http.StatusBadRequest)
return 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) 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.push(nativeXHTTPPacket{Payload: payload, Seq: seq}); err != nil { if err := sess.queue.pushContext(r.Context(), nativeXHTTPPacket{Payload: payload, Seq: seq}, nativeXHTTPQueuePushTimeoutDuration()); err != nil {
log.Printf("native xray: xhttp packet-up push failed inbound=%q session=%q seq=%d: %v", ib.tag, sess.id, seq, err) status := http.StatusConflict
http.Error(w, err.Error(), 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 return
} }
if len(payload) == 0 { if len(payload) == 0 {
@@ -576,7 +632,8 @@ func (ib *nativeInbound) xhttpMaxPostBytes() int64 {
} }
func (ib *nativeInbound) handleXHTTPStreamOne(w http.ResponseWriter, r *http.Request) { 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("X-Accel-Buffering", "no")
w.Header().Set("Cache-Control", "no-store") w.Header().Set("Cache-Control", "no-store")
if !ib.xhttpNoSSEHeader { if !ib.xhttpNoSSEHeader {
@@ -586,12 +643,14 @@ func (ib *nativeInbound) handleXHTTPStreamOne(w http.ResponseWriter, r *http.Req
flushHTTP(w) flushHTTP(w)
remote := remoteAddrFromHTTPRequest(r) remote := remoteAddrFromHTTPRequest(r)
resp := newNativeXHTTPResponseWriter(w)
xc := &nativeXHTTPConn{ xc := &nativeXHTTPConn{
reader: r.Body, reader: r.Body,
writer: &nativeXHTTPResponseWriter{w: w}, writer: resp,
remote: remote, remote: remote,
local: dummyLocalAddr(r), local: dummyLocalAddr(r),
onClose: func() { onClose: func() {
resp.close()
_ = r.Body.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) { 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() sess.markConnected()
defer ib.deleteXHTTPSession(sessionID, sess) defer ib.deleteXHTTPSession(sessionID, sess)
@@ -614,27 +675,31 @@ func (ib *nativeInbound) handleXHTTPDownload(w http.ResponseWriter, r *http.Requ
remote := remoteAddrFromHTTPRequest(r) remote := remoteAddrFromHTTPRequest(r)
var reader io.Reader = sess.queue var reader io.Reader = sess.queue
resp := newNativeXHTTPResponseWriter(w)
xc := &nativeXHTTPConn{ xc := &nativeXHTTPConn{
reader: reader, reader: reader,
writer: &nativeXHTTPResponseWriter{w: w}, writer: resp,
remote: remote, remote: remote,
local: dummyLocalAddr(r), local: dummyLocalAddr(r),
} }
xc.onClose = sess.close xc.onClose = func() {
resp.close()
sess.close()
}
ib.dispatchXHTTPConn(xc, remote) ib.dispatchXHTTPConn(xc, remote)
_ = xc.Close() _ = xc.Close()
} }
func (ib *nativeInbound) dispatchXHTTPConn(xc net.Conn, remote net.Addr) { 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 { switch ib.protocol {
case "vless": case "vless":
ib.handleVLESS(xc, remote) ib.handleVLESS(xc, remote)
case "vmess": case "vmess":
ib.handleVMess(xc, remote) ib.handleVMess(xc, remote)
default: 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 closeOnce sync.Once
mu sync.Mutex mu sync.Mutex
connected bool connected bool
lastSeen time.Time
}
func (s *nativeXHTTPSession) touch() {
s.mu.Lock()
s.lastSeen = time.Now()
s.mu.Unlock()
} }
func (s *nativeXHTTPSession) markConnected() { func (s *nativeXHTTPSession) markConnected() {
s.mu.Lock() s.mu.Lock()
s.connected = true s.connected = true
s.lastSeen = time.Now()
s.mu.Unlock() s.mu.Unlock()
} }
@@ -736,9 +809,21 @@ func (c *nativeXHTTPConn) SetReadDeadline(t time.Time) error {
func (c *nativeXHTTPConn) SetWriteDeadline(time.Time) error { return nil } func (c *nativeXHTTPConn) SetWriteDeadline(time.Time) error { return nil }
type nativeXHTTPResponseWriter struct { type nativeXHTTPResponseWriter struct {
mu sync.Mutex mu sync.Mutex
w http.ResponseWriter w http.ResponseWriter
closed bool 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) { func (w *nativeXHTTPResponseWriter) Write(p []byte) (int, error) {
@@ -748,15 +833,62 @@ func (w *nativeXHTTPResponseWriter) Write(p []byte) (int, error) {
return 0, io.ErrClosedPipe return 0, io.ErrClosedPipe
} }
n, err := w.w.Write(p) n, err := w.w.Write(p)
if n > 0 {
w.buffered += n
}
if err == nil { if err == nil {
flushHTTP(w.w) w.flushMaybeLocked(false)
} }
return n, err 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() { func (w *nativeXHTTPResponseWriter) close() {
w.mu.Lock() 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() 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 { select {
case q.pushedPackets <- p: case q.pushedPackets <- p:
return nil return nil
case <-q.closed: case <-q.closed:
return io.ErrClosedPipe return io.ErrClosedPipe
case <-ctx.Done():
return ctx.Err()
case <-t.C:
return errNativeXHTTPQueueFull
} }
} }