fix(ssh): port upstream non-retryable failure handling / 移植上游 SSH 确定性失败处理

Adapt cmliu/CF-Workers-WebSSH b0265afcaf40681066d24cd5d859dc32d73c8863 without importing unrelated network-monitoring changes. Classify asynchronous read-loop failures, stop retries on control messages or close code 4001, and preserve manual reconnect and auxiliary-channel behavior. Add focused backend, reconnect, and browser regressions; accept CRLF in the existing deployment workflow test.
This commit is contained in:
gdz66601 committed 2026-10-02 13:05:19 +08:00
1 parent 7e1f657852
commit 4efa997ad1
7 files changed
+226 -9

No files matched your search

+20 -1
View File
@@ -62,6 +62,7 @@ interface ServerMessage {
type?: string;
event?: string;
message?: string;
retryable?: boolean;
fingerprint?: string;
expectedFingerprint?: string;
keyType?: string;
@@ -109,6 +110,7 @@ const LANGUAGE_STORAGE_KEY = 'workers-webssh.language';
const MAX_KEY_BYTES = 65_536;
const PING_INTERVAL_MS = 25_000;
const CLIENT_CLOSE_SESSION_ERROR = 4000;
const SERVER_CLOSE_AUTH_DEFECT = 4001;
const CLIENT_CLOSE_PROTOCOL_ERROR = 4002;
function loadLanguage(): Language {
@@ -1435,6 +1437,12 @@ function handleServerMessage(message: ServerMessage): void {
event(text, message.event ?? 'error', true);
showFormError(text);
toast(text, 'error');
if (message.retryable === false) {
// 客户端随后以 4000 关闭会遮住服务端关闭码,先停止重连才能避免重试旧凭据。
sshReconnectManager?.reset();
sshReconnectManager = null;
reconnectParams = null;
}
failActiveConnection(failedSocket, 'SSH session failed', messageTranslation(text));
return;
}
@@ -1592,7 +1600,17 @@ function createSshReconnectFactory(): (attempt: number) => Promise<WebSocket> {
/** Handles reconnect lifecycle events and updates the SSH UI accordingly. */
function handleSshReconnectLog(entry: ReconnectLogEntry): void {
if (entry.event === 'give_up') {
if (entry.event === 'disconnect' && entry.code === SERVER_CLOSE_AUTH_DEFECT) {
// 控制帧丢失时以关闭码兜底;统一覆盖首次连接和重连工厂创建的 socket。
sshReconnectManager?.reset();
sshReconnectManager = null;
reconnectParams = null;
const reason = bilingual('SSH 认证或主机密钥确认失败,请检查后重试。', 'SSH authentication or host key approval failed; check the settings and try again.');
showFormError(reason);
event(reason, 'disconnect', true);
toast(reason, 'error');
failActiveConnection(socket, 'SSH authentication failed', messageTranslation(reason));
} else if (entry.event === 'give_up') {
// Reconnect exhausted — perform full cleanup that was deferred.
const reason = bilingual('SSH 重连失败,已达最大重试次数。', 'SSH reconnect failed; maximum retries reached.');
event(reason, 'disconnect', true);
@@ -1699,6 +1717,7 @@ async function connect(): Promise<void> {
id: 'SSH',
onConnect: createSshReconnectFactory(),
onLog: handleSshReconnectLog,
nonRetryableCloseCodes: [SERVER_CLOSE_AUTH_DEFECT],
});
sshReconnectManager.attach(activeSocket);
+10
View File
@@ -25,6 +25,8 @@ export interface ReconnectConfig {
onConnect: (attempt: number) => WebSocket | Promise<WebSocket>;
/** Optional callback invoked for every lifecycle event. */
onLog?: (entry: ReconnectLogEntry) => void;
/** 确定性失败不能靠重复连接恢复;默认不改变其他通道的重连策略。 */
nonRetryableCloseCodes?: number[];
}
export interface ReconnectLogEntry {
@@ -53,6 +55,7 @@ export class WebSocketReconnectManager {
private delays: number[];
private onConnect: (attempt: number) => WebSocket | Promise<WebSocket>;
private onLog?: (entry: ReconnectLogEntry) => void;
private nonRetryableCloseCodes: number[];
private socket: WebSocket | null = null;
private destroyed = false;
private reconnecting = false;
@@ -65,6 +68,7 @@ export class WebSocketReconnectManager {
this.delays = config.delays?.length ? config.delays : [...DEFAULT_DELAYS];
this.onConnect = config.onConnect;
this.onLog = config.onLog;
this.nonRetryableCloseCodes = config.nonRetryableCloseCodes ?? [];
}
/**
@@ -128,6 +132,12 @@ export class WebSocketReconnectManager {
return;
}
if (this.nonRetryableCloseCodes.includes(event.code)) {
// 重连中的 socket 也可能被拒绝;reset 同时取消计时器与尚未完成的连接工厂。
this.reset();
return;
}
if (this.reconnecting) {
// The current reconnect attempt's socket closed before the `open`
// event confirmed success (e.g. CONNECTING → error). Continue
+18 -7
View File
@@ -52,6 +52,15 @@ import { parseSystemProbe, SYSTEM_PROBE_COMMAND } from './system-info';
import { ForwardChannel } from '../forwarding/channel';
type Cipher = SSHAESGCMCipher | SSHAESCTRCipher;
// 凭据或主机信任已被拒绝时,重复连接不能解决问题,还可能触发远端账户锁定。
export class SSHAuthDefectError extends Error {
constructor(message: string) {
super(message);
this.name = 'SSHAuthDefectError';
}
}
type Phase = 'version' | 'kex' | 'host-confirm' | 'auth' | 'pty' | 'shell' | 'ready' | 'closed';
interface PendingSFTPChannelOpen {
readonly channelID: number;
@@ -273,7 +282,7 @@ export class SSHSession {
throw new Error('Unsupported WebSocket message');
}
close(normal = false): void {
close(normal = false, failureCode = 1011): void {
if (this.phase === 'closed') return;
this.phase = 'closed';
for (const forward of this.forwards.values()) forward.abort(new Error('SSH session closed'));
@@ -309,7 +318,7 @@ export class SSHSession {
try { this.writer?.releaseLock(); } catch { /* already released */ }
this.writer = null;
try { this.socket.close(); } catch { /* already closed */ }
try { this.ws.close(normal ? 1000 : 1011, normal ? 'Session closed' : 'SSH session failed'); } catch { /* already closed */ }
try { this.ws.close(normal ? 1000 : failureCode, normal ? 'Session closed' : 'SSH session failed'); } catch { /* already closed */ }
}
private async readLoop(): Promise<void> {
@@ -500,12 +509,12 @@ export class SSHSession {
const keyType = this.decoder.decode(hostKey.subarray(4, 4 + keyTypeLength));
if (!this.isHostKeyAlgorithmCompatible(this.hostKeyAlgorithm, keyType)) throw new Error(`Server used ${keyType}, but negotiated ${this.hostKeyAlgorithm ?? 'no host key algorithm'}`);
const fingerprint = `SHA256:${this.base64(new Uint8Array(await crypto.subtle.digest('SHA-256', toBufferSource(hostKey))))}`;
if (!await this.verifyHostSignature(hostKey, signature, hash)) throw new Error('SSH host key signature verification failed');
if (!await this.verifyHostSignature(hostKey, signature, hash)) throw new SSHAuthDefectError('SSH host key signature verification failed');
const trust = classifyHostKey(this.config.expectedFingerprint, fingerprint);
if (trust === 'trusted') {
this.sendJson({ type: 'host_key', fingerprint, keyType, trusted: true });
} else if (!await this.confirmHostKey(fingerprint, keyType, trust === 'changed' ? this.config.expectedFingerprint : undefined)) {
throw new Error('Host key was not accepted');
throw new SSHAuthDefectError('Host key was not accepted');
}
this.status('host_key_verified', `Host key verified (${keyType})`);
@@ -686,7 +695,7 @@ export class SSHSession {
return;
}
this.config.password = undefined;
throw new Error('SSH authentication failed');
throw new SSHAuthDefectError('SSH authentication failed');
}
}
@@ -1732,7 +1741,9 @@ export class SSHSession {
private sendJson(value: unknown): void { if (this.ws.readyState === WebSocket.OPEN) this.ws.send(JSON.stringify(value)); }
private fail(error: unknown, event: string): void {
const message = error instanceof Error ? error.message : String(error);
this.sendJson({ type: 'error', event, message });
this.close();
// 握手由 readLoop 异步推进,必须在这里分类,start() 的调用方无法捕获这些失败。
const retryable = !(error instanceof SSHAuthDefectError);
this.sendJson({ type: 'error', event, message, ...(retryable ? {} : { retryable: false }) });
this.close(false, retryable ? 1011 : 4001);
}
}
+39
View File
@@ -0,0 +1,39 @@
import { test, expect } from '@playwright/test';
import type { WebSocketRoute } from '@playwright/test';
import { connectFiles, fileFixture } from './file-fixture';
for (const signal of ['error-message', 'close-code', 'reconnect-close-code']) {
test(`SSH deterministic failure stops automatic reconnect: ${signal}`, async ({ page }) => {
await fileFixture(page);
const sockets: WebSocketRoute[] = [];
await page.routeWebSocket('**/api/ssh?*', (ws) => {
sockets.push(ws);
ws.onMessage((raw) => {
if (typeof raw !== 'string' || JSON.parse(raw).type !== 'connect') return;
ws.send(JSON.stringify({ type: 'ready' }));
});
});
await connectFiles(page);
await expect(page.locator('#files-connection-state')).toHaveText('SSH 已连接');
if (signal === 'reconnect-close-code') {
sockets[0].close({ code: 1011, reason: 'Transport interrupted' });
await expect.poll(() => sockets.length).toBe(2);
await expect(page.locator('#files-connection-state')).toHaveText('SSH 已连接');
}
const active = sockets.at(-1)!;
if (signal === 'error-message') {
active.send(JSON.stringify({ type: 'error', message: 'SSH authentication failed', retryable: false }));
} else {
active.close({ code: 4001, reason: 'SSH session failed' });
}
await expect(page.locator('#files-connect')).toBeEnabled();
await expect(page.locator('#files-connection-state')).not.toHaveText('SSH 已连接');
const count = sockets.length;
// 首次自动重连延迟为 1 秒,跨过该期限核验不再使用失败凭据。
await page.waitForTimeout(1400);
expect(sockets.length).toBe(count);
await page.locator('#files-connect').click();
await expect.poll(() => sockets.length).toBe(count + 1);
await expect(page.locator('#files-connection-state')).toHaveText('SSH 已连接');
});
}
+2 -1
View File
@@ -226,7 +226,8 @@ test('Actions asks for email, keeps Token private and uses the shared deploy ent
assert.equal(workflow.includes('secrets.ACCESS_TEAM_DOMAIN'), false);
assert.ok(workflow.includes('CUSTOM_DOMAIN: ${{ secrets.CUSTOM_DOMAIN || vars.CUSTOM_DOMAIN }}'));
assert.ok(workflow.includes('run: npm run deploy:validate'));
assert.ok(workflow.includes('run: npm run deploy\n'));
// Windows 检出可能使用 CRLF,部署命令契约不应依赖本地换行格式。
assert.match(workflow, /run: npm run deploy\r?\n/);
assert.equal(workflow.includes('wrangler d1 migrations apply'), false);
});
+83
View File
@@ -0,0 +1,83 @@
import assert from 'node:assert/strict';
import { test } from 'node:test';
import { fileURLToPath } from 'node:url';
import { build } from 'esbuild';
// 与 Worker 构建一样解析扩展名省略的导入,在内存中测试真实会话代码。
const result = await build({
stdin: {
contents: "export * from './src/backend/session.ts'; export { SSHPacketBuilder } from './src/ssh/packet.ts';",
resolveDir: fileURLToPath(new URL('..', import.meta.url)),
loader: 'ts',
},
bundle: true, format: 'esm', platform: 'neutral', write: false,
});
const { SSHSession, SSHAuthDefectError, SSHPacketBuilder } = await import(
`data:text/javascript;base64,${Buffer.from(result.outputFiles[0].text).toString('base64')}`
);
function fixture(authMethod = 'password', mode?: string) {
const messages: any[] = [];
const closes: number[] = [];
let tcpClosed = false;
const ws = { readyState: WebSocket.OPEN, send: (raw: string) => messages.push(JSON.parse(raw)), close: (code: number) => closes.push(code) };
let controller: ReadableStreamDefaultController<Uint8Array>;
const socket = {
readable: new ReadableStream<Uint8Array>({ start(value) { controller = value; } }),
close() { tcpClosed = true; },
};
const config = { host: '192.0.2.10', port: 22, username: 'root', authMethod, password: 'test-only', privateKey: 'test-only', mode };
const session = new SSHSession(ws, socket, config);
session.phase = 'auth';
session.authRequestSent = true;
return { session, config, messages, closes, push: (data: Uint8Array) => controller!.enqueue(data), tcpClosed: () => tcpClosed };
}
function authFailure(methods: string) {
const value = new TextEncoder().encode(methods);
const payload = new Uint8Array(value.length + 6);
payload[0] = 51;
new DataView(payload.buffer).setUint32(1, value.length);
payload.set(value, 5);
return payload;
}
for (const authMethod of ['password', 'publickey']) {
for (const mode of [undefined, 'forward']) {
test(`async ${authMethod} rejection stops retries in ${mode ?? 'terminal'} mode`, async () => {
const f = fixture(authMethod, mode);
f.push(await SSHPacketBuilder.build(authFailure('publickey,password'), 8, null, 0));
await f.session.readLoop();
assert.deepEqual(f.closes, [4001]);
assert.equal(f.messages.at(-1).retryable, false);
assert.equal(f.messages.at(-1).message, 'SSH authentication failed');
assert.equal(f.tcpClosed(), true);
assert.equal(f.config.password, undefined);
assert.equal(f.config.privateKey, undefined);
});
}
}
test('password authentication still falls back to keyboard-interactive before stopping', async () => {
const f = fixture();
const sent: Uint8Array[] = [];
f.session.sendEncrypted = async (payload: Uint8Array) => { sent.push(payload); };
await f.session.handleAuth(51, authFailure('keyboard-interactive'));
assert.equal(f.session.passwordAuthMethod, 'keyboard-interactive');
assert.equal(sent.length, 1);
assert.equal(f.config.password, 'test-only');
await assert.rejects(f.session.handleAuth(51, authFailure('keyboard-interactive')), SSHAuthDefectError);
});
test('host trust failures stop retries but transport failures keep their existing policy', () => {
for (const message of ['Host key was not accepted', 'SSH host key signature verification failed']) {
const f = fixture();
f.session.fail(new SSHAuthDefectError(message), 'read_error');
assert.equal(f.messages[0].retryable, false);
assert.deepEqual(f.closes, [4001]);
}
const f = fixture();
f.session.fail(new Error('Transport interrupted'), 'read_error');
assert.equal(f.messages[0].retryable, undefined);
assert.deepEqual(f.closes, [1011]);
});
+54
View File
@@ -0,0 +1,54 @@
import assert from 'node:assert/strict';
import { test } from 'node:test';
import { WebSocketReconnectManager } from '../frontend/src/ws-reconnect.ts';
class MockSocket extends EventTarget {
readyState = WebSocket.CONNECTING;
close(code: number) {
this.readyState = WebSocket.CLOSED;
const event = new Event('close');
Object.assign(event, { code, reason: 'test' });
this.dispatchEvent(event);
}
}
test('deterministic close never schedules a reconnect', (t) => {
t.mock.timers.enable({ apis: ['setTimeout'] });
let calls = 0;
const manager = new WebSocketReconnectManager({ id: 'SSH', nonRetryableCloseCodes: [4001], onConnect: () => { calls++; return new MockSocket() as any; } });
const socket = new MockSocket();
manager.attach(socket as any);
socket.close(4001);
t.mock.timers.tick(10_000);
assert.equal(calls, 0);
});
test('a rejected reconnect clears its state and cannot schedule another attempt', async (t) => {
t.mock.timers.enable({ apis: ['setTimeout'] });
let calls = 0;
const replacement = new MockSocket();
const manager = new WebSocketReconnectManager({ id: 'SSH', delays: [1], nonRetryableCloseCodes: [4001], onConnect: () => { calls++; return replacement as any; } });
const socket = new MockSocket();
manager.attach(socket as any);
socket.close(1011);
t.mock.timers.tick(1);
await Promise.resolve();
replacement.close(4001);
t.mock.timers.tick(10_000);
assert.equal(calls, 1);
assert.equal((manager as any).reconnecting, false);
assert.equal((manager as any).timer, null);
});
test('other channels retain their existing retry policy', async (t) => {
t.mock.timers.enable({ apis: ['setTimeout'] });
let calls = 0;
const manager = new WebSocketReconnectManager({ id: 'SFTP', delays: [1], onConnect: () => { calls++; return new MockSocket() as any; } });
const socket = new MockSocket();
manager.attach(socket as any);
socket.close(4001);
t.mock.timers.tick(1);
await Promise.resolve();
assert.equal(calls, 1);
manager.reset();
});