fix(net): validate wire frames and coalesce reconnects

This commit is contained in:
2026-09-05 04:25:51 +08:00
parent 58a7a2dc68
commit 85db7765ac
3 changed files with 272 additions and 22 deletions
@@ -37,22 +37,112 @@ function assertSwitchTarget(value: string): void {
}
}
function invalidEnvelope(path: string, reason: string): never {
throw new Error(`invalid outbound envelope at ${path}: ${reason}`);
}
function childPath(parent: string, key: string): string {
return /^[A-Za-z_$][A-Za-z0-9_$]*$/.test(key)
? `${parent}.${key}`
: `${parent}[${JSON.stringify(key)}]`;
}
function dataPropertyValue(owner: object, key: string, path: string): unknown {
const descriptor = Object.getOwnPropertyDescriptor(owner, key);
if (!descriptor) return invalidEnvelope(path, 'missing own data property');
if (!descriptor.enumerable) return invalidEnvelope(path, 'property must be enumerable');
if (!Object.prototype.hasOwnProperty.call(descriptor, 'value')) {
return invalidEnvelope(path, 'accessor properties are not allowed');
}
return descriptor.value;
}
function validateJsonValue(value: unknown, path: string, ancestors: Set<object>): void {
if (value === null || typeof value === 'string' || typeof value === 'boolean') return;
if (typeof value === 'number') {
if (!Number.isFinite(value)) invalidEnvelope(path, 'expected finite number');
return;
}
if (typeof value !== 'object') {
invalidEnvelope(path, `${typeof value} is not protocol JSON data`);
}
const objectValue = value as object;
if (ancestors.has(objectValue)) invalidEnvelope(path, 'cycle is not protocol JSON data');
ancestors.add(objectValue);
try {
if (Array.isArray(objectValue)) {
if (Object.getPrototypeOf(objectValue) !== Array.prototype) {
invalidEnvelope(path, 'expected plain array');
}
const allowedKeys = new Set<string>(['length']);
for (let index = 0; index < objectValue.length; index++) {
const key = String(index);
const itemPath = `${path}[${index}]`;
allowedKeys.add(key);
if (!Object.prototype.hasOwnProperty.call(objectValue, key)) {
invalidEnvelope(itemPath, 'array hole is not protocol JSON data');
}
validateJsonValue(
dataPropertyValue(objectValue, key, itemPath),
itemPath,
ancestors,
);
}
for (const key of Reflect.ownKeys(objectValue)) {
if (typeof key === 'symbol') invalidEnvelope(path, 'symbol key is not allowed');
if (!allowedKeys.has(key as string)) {
invalidEnvelope(childPath(path, key as string), 'extra array property is not allowed');
}
}
return;
}
const prototype = Object.getPrototypeOf(objectValue);
if (prototype !== Object.prototype && prototype !== null) {
invalidEnvelope(path, 'expected plain object or null-prototype object');
}
for (const key of Reflect.ownKeys(objectValue)) {
if (typeof key === 'symbol') invalidEnvelope(path, 'symbol key is not allowed');
const propertyPath = childPath(path, key as string);
validateJsonValue(
dataPropertyValue(objectValue, key as string, propertyPath),
propertyPath,
ancestors,
);
}
} finally {
ancestors.delete(objectValue);
}
}
function serializeOutbound(envelope: OutboundEnvelope): string {
if (typeof envelope !== 'object' || envelope === null || Array.isArray(envelope)) {
throw new Error('invalid outbound envelope');
return invalidEnvelope('$', 'expected object');
}
const value = envelope as unknown as Record<string, unknown>;
const owns = (key: string) => Object.prototype.hasOwnProperty.call(value, key);
if (
!owns('app') || value.app !== APP
|| !owns('route') || typeof value.route !== 'string' || value.route.length === 0
|| !owns('rpc') || typeof value.rpc !== 'string' || value.rpc.length === 0
|| !owns('data') || value.data === undefined
) {
throw new Error('invalid outbound envelope');
const prototype = Object.getPrototypeOf(envelope);
if (prototype !== Object.prototype && prototype !== null) {
return invalidEnvelope('$', 'expected plain object or null-prototype object');
}
const required = new Set(['app', 'route', 'rpc', 'data']);
for (const key of Reflect.ownKeys(envelope)) {
if (typeof key === 'symbol') return invalidEnvelope('$', 'top-level symbol key is not allowed');
if (!required.has(key)) return invalidEnvelope(childPath('$', key), 'extra top-level property');
}
const app = dataPropertyValue(envelope, 'app', '$.app');
const route = dataPropertyValue(envelope, 'route', '$.route');
const rpc = dataPropertyValue(envelope, 'rpc', '$.rpc');
const data = dataPropertyValue(envelope, 'data', '$.data');
if (app !== APP) invalidEnvelope('$.app', `expected ${APP}`);
if (typeof route !== 'string' || route.length === 0) {
invalidEnvelope('$.route', 'expected non-empty string');
}
if (typeof rpc !== 'string' || rpc.length === 0) {
invalidEnvelope('$.rpc', 'expected non-empty string');
}
validateJsonValue(data, '$.data', new Set<object>());
const frame = JSON.stringify(envelope);
if (typeof frame !== 'string') throw new Error('invalid outbound envelope');
if (typeof frame !== 'string') return invalidEnvelope('$', 'not JSON serializable');
return frame;
}
@@ -69,6 +159,7 @@ export class WireClient {
private transport: Transport | null = null;
private reconnectTimer: unknown = null;
private reconnectTarget: string | null = null;
private reconnectToken = 0;
private generation = 0;
private started = false;
@@ -117,7 +208,9 @@ export class WireClient {
reconnectCurrent(): void {
this.assertRunning();
this.disconnectAndSchedule(this.policy.current());
const server = this.policy.current();
if (this.reconnectTimer !== null && this.reconnectTarget === server) return;
this.disconnectAndSchedule(server);
}
stop(): void {
@@ -241,30 +334,29 @@ export class WireClient {
this.clearReconnectTimer();
if (this.stopped || this.policy.current() !== server) return;
const token = this.reconnectToken;
this.emit({ type: 'reconnecting', server });
if (
this.stopped
|| token !== this.reconnectToken
|| this.policy.current() !== server
) return;
let handle: unknown = null;
handle = this.clock.setTimeout(() => {
if (
this.stopped
|| token !== this.reconnectToken
|| this.policy.current() !== server
|| this.reconnectTarget !== server
|| this.reconnectTimer !== handle
) return;
this.reconnectTimer = null;
this.reconnectTarget = null;
this.reconnectToken++;
this.connect(server);
}, RECONNECT_INTERVAL_MS);
this.reconnectTimer = handle;
this.reconnectTarget = server;
this.emit({ type: 'reconnecting', server });
}
private clearReconnectTimer(): void {
this.reconnectToken++;
if (this.reconnectTimer !== null) this.clock.clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
this.reconnectTarget = null;
}
}
@@ -11,14 +11,16 @@ export class FakeTransport implements Transport {
private closeCb?: () => void;
private pending: Array<() => void> = [];
private closed = false;
private connectFailure: unknown = null;
private hasConnectFailure = false;
private connectFailure: unknown;
connect(url: string): void {
this.connectCalls++;
this.url = url;
if (this.connectFailure !== null) {
if (this.hasConnectFailure) {
const error = this.connectFailure;
this.connectFailure = null;
this.hasConnectFailure = false;
this.connectFailure = undefined;
throw error;
}
this.pending.push(() => this.openCb?.());
@@ -35,7 +37,10 @@ export class FakeTransport implements Transport {
onClose(cb: () => void): void { this.closeCb = cb; }
/** 测试侧:令下一次 connect 同步失败。 */
failNextConnect(error: unknown): void { this.connectFailure = error; }
failNextConnect(error: unknown): void {
this.hasConnectFailure = true;
this.connectFailure = error;
}
/** 测试侧:模拟服务器推一帧给客户端。 */
serverPush(frame: string): void { this.pending.push(() => this.msgCb?.(frame)); }
@@ -159,6 +159,8 @@ test('send rejects malformed completed envelopes before touching transport', asy
{ app: 'youle', route: '', rpc: 'x', data: {} },
{ app: 'youle', route: 'agent', rpc: '', data: {} },
{ app: 'youle', route: 'agent', rpc: 'x' },
{ app: 'youle', route: 'agent', rpc: 'x', data: {}, extra: true },
{ app: 'youle', route: 'agent', rpc: 'x', data: () => 1 },
null,
];
for (const value of invalid) {
@@ -167,6 +169,101 @@ test('send rejects malformed completed envelopes before touching transport', asy
assert.deepEqual(transport.sent, []);
});
test('send accepts the complete protocol JSON value boundary without rebuilding bytes', async () => {
const { client, transports } = makeHarness();
client.start();
const transport = transports[0]!;
await transport.flush();
const nullPrototype = Object.assign(Object.create(null) as Record<string, unknown>, {
nested: [null, 'text', true, false, 0, -1.5],
});
const valid = {
rpc: 'json_boundary',
data: nullPrototype,
route: 'agent',
app: 'youle',
} as OutboundEnvelope;
client.send(valid);
assert.equal(transport.sent[0], JSON.stringify(valid));
});
test('send rejects every non-JSON data graph edge with its exact path', async () => {
const { client, transports } = makeHarness();
client.start();
const transport = transports[0]!;
await transport.flush();
const sparse: unknown[] = [];
sparse.length = 1;
const arrayWithExtra = [] as unknown[] & { extra?: number };
arrayWithExtra.extra = 1;
const cyclic: Record<string, unknown> = {};
cyclic.self = cyclic;
const withAccessor: Record<string, unknown> = {};
let getterCalls = 0;
Object.defineProperty(withAccessor, 'secret', {
enumerable: true,
get() { getterCalls++; return 1; },
});
const withHidden: Record<string, unknown> = {};
Object.defineProperty(withHidden, 'hidden', { enumerable: false, value: 1 });
const withSymbolKey: Record<PropertyKey, unknown> = { visible: true };
withSymbolKey[Symbol('hidden')] = 1;
class Exotic { value = 1; }
const cases: ReadonlyArray<readonly [unknown, RegExp]> = [
[{ value: undefined }, /\$\.data\.value.*undefined/i],
[{ value: () => 1 }, /\$\.data\.value.*function/i],
[{ value: Symbol('x') }, /\$\.data\.value.*symbol/i],
[{ value: 1n }, /\$\.data\.value.*bigint/i],
[{ value: Number.NaN }, /\$\.data\.value.*finite number/i],
[{ value: Number.POSITIVE_INFINITY }, /\$\.data\.value.*finite number/i],
[{ nested: sparse }, /\$\.data\.nested\[0\].*array hole/i],
[{ nested: arrayWithExtra }, /\$\.data\.nested\.extra.*extra array property/i],
[{ nested: new Date(0) }, /\$\.data\.nested.*plain object/i],
[{ nested: new Exotic() }, /\$\.data\.nested.*plain object/i],
[{ nested: withAccessor }, /\$\.data\.nested\.secret.*accessor/i],
[{ nested: withHidden }, /\$\.data\.nested\.hidden.*enumerable/i],
[{ nested: withSymbolKey }, /\$\.data\.nested.*symbol key/i],
[{ nested: { toJSON: () => ({ changed: true }) } }, /\$\.data\.nested\.toJSON.*function/i],
[cyclic, /\$\.data\.self.*cycle/i],
];
for (const [data, expected] of cases) {
assert.throws(
() => client.send({ app: 'youle', route: 'agent', rpc: 'x', data }),
expected,
);
}
assert.equal(getterCalls, 0, 'accessor validation must not invoke the getter');
assert.deepEqual(transport.sent, []);
});
test('send rejects top-level symbol, non-enumerable and accessor properties', async () => {
const { client, transports } = makeHarness();
client.start();
const transport = transports[0]!;
await transport.flush();
const symbolEnvelope = { app: 'youle', route: 'agent', rpc: 'x', data: {} };
Object.defineProperty(symbolEnvelope, Symbol('extra'), { enumerable: true, value: 1 });
const hiddenEnvelope = { app: 'youle', route: 'agent', rpc: 'x' } as Record<string, unknown>;
Object.defineProperty(hiddenEnvelope, 'data', { enumerable: false, value: {} });
let accessorCalls = 0;
const accessorEnvelope = { app: 'youle', route: 'agent', rpc: 'x' } as Record<string, unknown>;
Object.defineProperty(accessorEnvelope, 'data', {
enumerable: true,
get() { accessorCalls++; return {}; },
});
assert.throws(() => client.send(symbolEnvelope as OutboundEnvelope), /top-level.*symbol/i);
assert.throws(() => client.send(hiddenEnvelope as unknown as OutboundEnvelope), /\$\.data.*enumerable/i);
assert.throws(() => client.send(accessorEnvelope as unknown as OutboundEnvelope), /\$\.data.*accessor/i);
assert.equal(accessorCalls, 0);
assert.deepEqual(transport.sent, []);
});
test('constructor rejects an empty validated server list and start cannot run twice', () => {
assert.throws(
() => new WireClient({ servers: [], transportFactory: () => new FakeTransport() }),
@@ -304,6 +401,35 @@ test('a reconnecting listener can stop without leaving a timer or reviving a cle
assert.equal(transports.length, 1);
});
test('reconnectCurrent coalesces the same pending target before reconnecting listeners run', async () => {
const { client, transports, advance, activeTimers } = makeHarness();
let reconnectEvents = 0;
let reentered = false;
const timerCountsDuringEvent: number[] = [];
client.subscribe((event) => {
if (event.type !== 'reconnecting') return;
reconnectEvents++;
timerCountsDuringEvent.push(activeTimers());
if (!reentered) {
reentered = true;
client.reconnectCurrent();
}
});
client.start();
const first = transports[0]!;
await first.flush();
first.serverClose();
await first.flush();
client.reconnectCurrent();
assert.equal(reconnectEvents, 1);
assert.deepEqual(timerCountsDuringEvent, [1]);
assert.equal(activeTimers(), 1);
advance(10000);
assert.equal(transports.length, 2);
assert.equal(transports[1]!.url, 'ws://a');
});
test('slow listener stop prevents the timeout path from scheduling reconnect', async () => {
const { client, transports, advance, activeTimers } = makeHarness();
client.subscribe((event) => {
@@ -431,6 +557,33 @@ test('transport connect synchronous failure ignores duplicate close/error and co
assert.equal(recovered.url, 'ws://a');
});
test('FakeTransport and WireClient preserve synchronous throw null as one connection failure', async () => {
const timer = fakeClock();
const broken = new FakeTransport();
broken.failNextConnect(null);
const recovered = new FakeTransport();
const transports = [broken, recovered];
let factoryIndex = 0;
const events: WireEvent[] = [];
const client = new WireClient({
servers: ['ws://a'],
clock: timer.clock,
transportFactory: () => transports[factoryIndex++]!,
});
client.subscribe((event) => events.push(event));
client.start();
assert.deepEqual(events, [
{ type: 'close' },
{ type: 'reconnecting', server: 'ws://a' },
]);
assert.equal(broken.closeCalls, 1);
assert.equal(timer.activeTimers(), 1);
timer.advance(10000);
await recovered.flush();
assert.deepEqual(events[events.length - 1], { type: 'open', server: 'ws://a' });
});
test('stale transport generations cannot emit messages or opens', async () => {
const { client, transports, advance } = makeHarness();
const events: WireEvent[] = [];