Enhance Mihomo WebSocket functionality and testing for connections
Publish telemt-api gateway Docker image / test (push) Successful in 25s
Publish telemt-api gateway Docker image / build-and-push (push) Successful in 2m18s

- Added support for WebSocket connections to the `/connections` endpoint, allowing for improved handling of WebSocket requests.
- Updated the `useMihomoWebSocketTunnel` function to accept `Sec-WebSocket-Key` for `/connections`, enhancing compatibility with various proxies.
- Introduced new test cases in `mihomo_test.go` to validate WebSocket tunnel behavior for connections, ensuring comprehensive coverage.
- Enhanced the `applyMihomoWSDialAuth` function to manage token handling for WebSocket connections, improving security and functionality.
- Updated Svelte components to implement native WebSocket loops for connections, providing real-time data updates and improved user experience.
This commit is contained in:
Denozordec
2026-03-31 11:54:18 +07:00
parent 49f36d2b8b
commit c3a0a771a3
5 changed files with 160 additions and 22 deletions
+26 -6
View File
@@ -42,7 +42,7 @@ func NewMihomoForward(
}
// useMihomoWebSocketTunnel is true when the client is doing a WS handshake.
// For /traffic and /memory we also accept Sec-WebSocket-Key alone: some hops strip
// For /traffic, /memory, /connections we also accept Sec-WebSocket-Key alone: some hops strip
// Connection/Upgrade but leave Sec-WebSocket-Key; plain streaming GET has no key.
func useMihomoWebSocketTunnel(rest string, r *http.Request) bool {
if r == nil {
@@ -55,13 +55,36 @@ func useMihomoWebSocketTunnel(rest string, r *http.Request) bool {
return false
}
switch rest {
case "traffic", "memory":
case "traffic", "memory", "connections":
return strings.TrimSpace(r.Header.Get("Sec-WebSocket-Key")) != ""
default:
return false
}
}
// applyMihomoWSDialAuth prepares upstream WS URL and Dial headers.
// Mihomo: /connections WebSocket expects ?token=<secret> (raw secret, no "Bearer " prefix);
// /traffic, /memory use Authorization: Bearer … like REST.
func applyMihomoWSDialAuth(rest string, dest *url.URL, auth string) http.Header {
hdr := make(http.Header)
a := strings.TrimSpace(auth)
if a == "" || dest == nil {
return hdr
}
if rest == "connections" {
token := a
if len(token) > 7 && strings.EqualFold(token[:7], "Bearer ") {
token = strings.TrimSpace(token[7:])
}
q := dest.Query()
q.Set("token", token)
dest.RawQuery = q.Encode()
return hdr
}
hdr.Set("Authorization", a)
return hdr
}
func mihomoWebSocketURL(dest *url.URL) (string, error) {
if dest == nil {
return "", fmt.Errorf("nil destination URL")
@@ -111,16 +134,13 @@ func newMihomoWSTunnel(
du.RawQuery = r.URL.RawQuery
dest = &du
dialHdr := applyMihomoWSDialAuth(rest, dest, auth)
wsURL, err := mihomoWebSocketURL(dest)
if err != nil {
errHandler(w, r, err)
return
}
dialHdr := make(http.Header)
if auth != "" {
dialHdr.Set("Authorization", auth)
}
d := websocket.Dialer{
HandshakeTimeout: 15 * time.Second,
Proxy: func(*http.Request) (*url.URL, error) { return nil, nil },
+44
View File
@@ -82,6 +82,50 @@ func TestUseMihomoWebSocketTunnel(t *testing.T) {
t.Fatal("proxies must not use WS tunnel")
}
})
t.Run("connections with key only", func(t *testing.T) {
r := httptest.NewRequest(http.MethodGet, "/api/x/mihomo/connections", nil)
r.Header.Set("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
if !useMihomoWebSocketTunnel("connections", r) {
t.Fatal("expected tunnel for connections+Sec-WebSocket-Key")
}
})
t.Run("connections REST GET without key", func(t *testing.T) {
r := httptest.NewRequest(http.MethodGet, "/api/x/mihomo/connections", nil)
if useMihomoWebSocketTunnel("connections", r) {
t.Fatal("plain GET /connections must use HTTP forwarder")
}
})
}
func TestApplyMihomoWSDialAuth(t *testing.T) {
t.Run("connections strips Bearer for token query", func(t *testing.T) {
u, _ := url.Parse("http://127.0.0.1:9090/connections")
hdr := applyMihomoWSDialAuth("connections", u, "Bearer mysecret")
if hdr.Get("Authorization") != "" {
t.Fatal("connections WS must not set Authorization")
}
if u.RawQuery != "token=mysecret" {
t.Fatalf("query: %q", u.RawQuery)
}
})
t.Run("connections raw token", func(t *testing.T) {
u, _ := url.Parse("http://127.0.0.1:9090/connections")
_ = applyMihomoWSDialAuth("connections", u, "rawonly")
if u.RawQuery != "token=rawonly" {
t.Fatalf("query: %q", u.RawQuery)
}
})
t.Run("traffic uses Authorization", func(t *testing.T) {
u, _ := url.Parse("http://127.0.0.1:9090/traffic")
hdr := applyMihomoWSDialAuth("traffic", u, "Bearer t")
if hdr.Get("Authorization") != "Bearer t" {
t.Fatalf("Authorization: %q", hdr.Get("Authorization"))
}
})
}
func TestMihomoWebSocketURL(t *testing.T) {
+1 -1
View File
@@ -23,7 +23,7 @@ export function mihomoUrl(alias: string, path: string): string {
/**
* WebSocket к шлюзу: `ws(s)://…/api/{alias}/mihomo/{path}`.
* Authorization к Mihomo подставляет шлюз; браузеру токен не нужен.
* Обзор Mihomo: нативный WS для /traffic и /memory + резервный потоковый GET при необходимости.
* Обзор Mihomo: WS для /traffic, /memory, /connections (токен на upstream подставляет шлюз; для /connections — ?token=…).
*/
export function mihomoWsUrl(alias: string, path: string): string {
const p = path.replace(/^\/+/, '');
+4
View File
@@ -18,6 +18,10 @@ export type MihomoProxyEntry = {
export type MihomoConnectionsResponse = {
total?: number;
connections?: MihomoConnection[];
/** Снимок по WebSocket /connections (Mihomo). */
downloadTotal?: number;
uploadTotal?: number;
memory?: number;
};
export type MihomoConnection = {
@@ -371,6 +371,7 @@
// Только нативный WS: без HTTP-обхода, чтобы сразу видеть проблемы апгрейда.
void trafficWsLoop(a, signal);
void memoryWsLoop(a, signal);
void connectionsWsLoop(a, signal);
}
function topProxyCounts(conns: MihomoConnectionsResponse['connections']): { name: string; n: number }[] {
@@ -387,28 +388,96 @@
.slice(0, 8);
}
/** Снимок как у WebSocket /connections у Mihomo (downloadTotal, uploadTotal, memory, connections). */
function applyConnectionsSnapshot(j: MihomoConnectionsResponse) {
const list = j.connections ?? [];
connTotal = typeof j.total === 'number' ? j.total : list.length;
if (typeof j.uploadTotal === 'number' && typeof j.downloadTotal === 'number') {
totalUp = j.uploadTotal;
totalDown = j.downloadTotal;
}
if (typeof j.memory === 'number') {
memKB = j.memory;
histMem = [...histMem, memKB];
while (histMem.length > MAX_POINTS) histMem.shift();
}
let t = 0,
u = 0;
for (const c of list) {
const net = (c.metadata?.network ?? '').toLowerCase();
if (net === 'tcp') t++;
else if (net === 'udp') u++;
}
tcpN = t;
udpN = u;
histConn = [...histConn, connTotal];
while (histConn.length > MAX_POINTS) histConn.shift();
topList = topProxyCounts(list);
}
async function pollConnections(a: string) {
try {
const j = await fetchMihomoJson<MihomoConnectionsResponse>(a, 'connections');
const list = j.connections ?? [];
connTotal = typeof j.total === 'number' ? j.total : list.length;
let t = 0,
u = 0;
for (const c of list) {
const net = (c.metadata?.network ?? '').toLowerCase();
if (net === 'tcp') t++;
else if (net === 'udp') u++;
}
tcpN = t;
udpN = u;
histConn = [...histConn, connTotal];
while (histConn.length > MAX_POINTS) histConn.shift();
topList = topProxyCounts(list);
applyConnectionsSnapshot(j);
} catch {
/* ignore */
}
}
/**
* WebSocket /connections: один JSON на кадр с полным снимком (как wss://…/connections?token=… у Mihomo).
* Шлюз сам подставляет token на upstream; браузер без секрета.
*/
async function connectionsWsLoop(a: string, signal: AbortSignal) {
while (!signal.aborted) {
await new Promise<void>((resolve) => {
let settled = false;
const done = () => {
if (settled) return;
settled = true;
resolve();
};
let ws: WebSocket;
try {
ws = new WebSocket(mihomoWsUrl(a, 'connections'));
} catch {
done();
return;
}
const onAbort = () => {
try {
ws.close();
} catch {
/* ignore */
}
done();
};
signal.addEventListener('abort', onAbort, { once: true });
ws.onmessage = (ev) => {
try {
const j = JSON.parse(String(ev.data)) as MihomoConnectionsResponse;
applyConnectionsSnapshot(j);
} catch {
/* ignore frame */
}
};
ws.onerror = () => {
try {
ws.close();
} catch {
/* ignore */
}
};
ws.onclose = () => {
signal.removeEventListener('abort', onAbort);
done();
};
});
if (signal.aborted) break;
await new Promise((r) => setTimeout(r, 1500));
}
}
$effect(() => {
const a = alias;
if (!a || !browser) return;
@@ -425,7 +494,8 @@
const streamAc = new AbortController();
runTrafficMemoryRealtime(a, streamAc.signal);
if (pollTimer) clearInterval(pollTimer);
pollTimer = setInterval(() => void pollConnections(a), 2000);
// Резервный HTTP, если WS /connections недоступен.
pollTimer = setInterval(() => void pollConnections(a), 15000);
void pollConnections(a);
return () => {
streamAc.abort();