src/vs/base/parts/ipc/node/ipc.net.ts
1025 LOC · 794 covered · 231 uncovered · 168 ranges · 90 concepts · 21 introducers · 52 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
>
/*---------------------------------------------------------------------------------------------
ipc.net.ts ×73
2
>
* Copyright (c) Microsoft Corporation. All rights reserved.
3
>
* Licensed under the MIT License. See License.txt in the project root for license information.
4
>
*--------------------------------------------------------------------------------------------*/
5
>
6
>
import { createHash } from 'crypto';
7
>
import type * as http from 'http';
8
>
import { Server as NetServer, Socket, createConnection, createServer } from 'net';
9
>
import { tmpdir } from 'os';
10
>
import { DeflateRaw, InflateRaw, ZlibOptions, createDeflateRaw, createInflateRaw } from 'zlib';
11
>
import { VSBuffer } from '../../../common/buffer.js';
12
>
import { onUnexpectedError } from '../../../common/errors.js';
13
>
import { Emitter, Event } from '../../../common/event.js';
14
>
import { Disposable, IDisposable } from '../../../common/lifecycle.js';
15
>
import { join } from '../../../common/path.js';
16
>
import { Platform, platform } from '../../../common/platform.js';
17
>
import { generateUuid } from '../../../common/uuid.js';
18
>
import { ClientConnectionEvent, IPCServer } from '../common/ipc.js';
19
>
import { ChunkStream, Client, ISocket, Protocol, SocketCloseEvent, SocketCloseEventType, SocketDiagnostics, SocketDiagnosticsEventType } from '../common/ipc.net.js';
20
>
21
>
export function upgradeToISocket(req: http.IncomingMessage, socket: Socket, {
22
debugLabel,
23
skipWebSocketFrames = false,
24
disableWebSocketCompression = false,
25
enableMessageSplitting = true,
26
}: {
27
debugLabel: string;
28
skipWebSocketFrames?: boolean;
29
disableWebSocketCompression?: boolean;
30
enableMessageSplitting?: boolean;
31
}): NodeSocket | WebSocketNodeSocket | undefined {
32
if (req.headers.upgrade === undefined || req.headers.upgrade.toLowerCase() !== 'websocket') {
33
socket.end('HTTP/1.1 400 Bad Request');
34
return;
35
}
36
37
// https://tools.ietf.org/html/rfc6455#section-4
38
const requestNonce = req.headers['sec-websocket-key'];
39
const hash = createHash('sha1');// CodeQL [SM04514] SHA1 must be used here to respect the WebSocket protocol specification
40
hash.update(requestNonce + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11');
41
const responseNonce = hash.digest('base64');
42
43
const responseHeaders = [
44
`HTTP/1.1 101 Switching Protocols`,
45
`Upgrade: websocket`,
46
`Connection: Upgrade`,
47
`Sec-WebSocket-Accept: ${responseNonce}`
48
];
49
50
// See https://tools.ietf.org/html/rfc7692#page-12
51
let permessageDeflate = false;
52
if (!skipWebSocketFrames && !disableWebSocketCompression && req.headers['sec-websocket-extensions']) {
53
const websocketExtensionOptions = Array.isArray(req.headers['sec-websocket-extensions']) ? req.headers['sec-websocket-extensions'] : [req.headers['sec-websocket-extensions']];
54
for (const websocketExtensionOption of websocketExtensionOptions) {
55
if (/\b((server_max_window_bits)|(server_no_context_takeover)|(client_no_context_takeover))\b/.test(websocketExtensionOption)) {
56
// sorry, the server does not support zlib parameter tweaks
57
continue;
58
}
59
if (/\b(permessage-deflate)\b/.test(websocketExtensionOption)) {
60
permessageDeflate = true;
61
responseHeaders.push(`Sec-WebSocket-Extensions: permessage-deflate`);
62
break;
63
}
64
if (/\b(x-webkit-deflate-frame)\b/.test(websocketExtensionOption)) {
65
permessageDeflate = true;
66
responseHeaders.push(`Sec-WebSocket-Extensions: x-webkit-deflate-frame`);
67
break;
68
}
69
}
70
}
71
72
socket.write(responseHeaders.join('\r\n') + '\r\n\r\n');
73
74
// Never timeout this socket due to inactivity!
75
socket.setTimeout(0);
76
// Disable Nagle's algorithm
77
socket.setNoDelay(true);
78
// Finally!
79
80
if (skipWebSocketFrames) {
81
return new NodeSocket(socket, debugLabel);
82
} else {
83
return new WebSocketNodeSocket(new NodeSocket(socket, debugLabel), permessageDeflate, null, true, enableMessageSplitting);
84
}
85
}
87
>
/**
88
>
* Maximum time to wait for a 'close' event to fire after the socket stream
89
>
* ends. For unix domain sockets, the close event may not fire consistently
90
>
* due to what appears to be a Node.js bug.
91
>
*
92
>
* @see https://github.com/microsoft/vscode/issues/211462#issuecomment-2155471996
93
>
*/
94
>
const socketEndTimeoutMs = 30_000;
95
>
96
>
export class NodeSocket implements ISocket {
97
>
98
>
public readonly debugLabel: string;
99
>
public readonly socket: Socket;
100
>
private readonly _errorListener: (err: NodeJS.ErrnoException) => void;
101
>
private readonly _closeListener: (hadError: boolean) => void;
102
>
private readonly _endListener: () => void;
103
>
private _endTimeoutHandle: Timeout | undefined;
104
>
private _canWrite = true;
105
>
106
>
public traceSocketEvent(type: SocketDiagnosticsEventType, data?: VSBuffer | Uint8Array | ArrayBuffer | ArrayBufferView | unknown): void {
107
>
SocketDiagnostics.traceSocketEvent(this.socket, this.debugLabel, type, data);
108
>
}
109
>
110
>
constructor(socket: Socket, debugLabel = '') {
112
>
this.socket = socket;
113
>
this.traceSocketEvent(SocketDiagnosticsEventType.Created, { type: 'NodeSocket' });
114
>
this._errorListener = (err: NodeJS.ErrnoException) => {
115
this.traceSocketEvent(SocketDiagnosticsEventType.Error, { code: err?.code, message: err?.message });
116
if (err) {
117
if (err.code === 'EPIPE') {
118
// An EPIPE exception at the wrong time can lead to a renderer process crash
119
// so ignore the error since the socket will fire the close event soon anyways:
120
// > https://nodejs.org/api/errors.html#errors_common_system_errors
121
// > EPIPE (Broken pipe): A write on a pipe, socket, or FIFO for which there is no
122
// > process to read the data. Commonly encountered at the net and http layers,
123
// > indicative that the remote side of the stream being written to has been closed.
124
return;
125
}
126
onUnexpectedError(err);
127
}
128
};
130
>
131
>
this._closeListener = (hadError: boolean) => {
133
>
this._canWrite = false;
134
>
if (this._endTimeoutHandle) {
135
>
clearTimeout(this._endTimeoutHandle);
136
>
}
137
>
};
139
>
140
>
this._endListener = () => {
142
>
this._canWrite = false;
143
>
this._endTimeoutHandle = setTimeout(() => socket.destroy(), socketEndTimeoutMs);
144
>
};
146
>
}
148
>
public dispose(destroySocket = true): void {
151
>
this._endTimeoutHandle = undefined;
152
>
}
154
>
this.socket.off('close', this._closeListener);
155
>
this.socket.off('end', this._endListener);
156
>
if (destroySocket) {
158
>
}
161
>
public onData(_listener: (e: VSBuffer) => void): IDisposable {
164
>
_listener(VSBuffer.wrap(buff));
165
>
};
167
>
return {
168
>
dispose: () => this.socket.off('data', listener)
169
>
};
170
>
}
172
>
public onClose(listener: (e: SocketCloseEvent) => void): IDisposable {
175
>
type: SocketCloseEventType.NodeSocketCloseEvent,
176
>
hadError: hadError,
177
>
error: undefined
178
>
});
179
>
};
181
>
return {
182
>
dispose: () => this.socket.off('close', adapter)
183
>
};
184
>
}
186
>
public onEnd(listener: () => void): IDisposable {
187
const adapter = () => {
188
listener();
189
};
190
this.socket.on('end', adapter);
191
return {
192
dispose: () => this.socket.off('end', adapter)
193
};
194
}
196
>
public write(buffer: VSBuffer): void {
198
>
if (this.socket.destroyed || !this._canWrite) {
199
return;
200
}
202
>
// we ignore the returned value from `write` because we would have to cached the data
203
>
// anyways and nodejs is already doing that for us:
204
>
// > https://nodejs.org/api/stream.html#stream_writable_write_chunk_encoding_callback
205
>
// > However, the false return value is only advisory and the writable stream will unconditionally
206
>
// > accept and buffer chunk even if it has not been allowed to drain.
207
>
try {
208
>
this.traceSocketEvent(SocketDiagnosticsEventType.Write, buffer);
209
>
this.socket.write(buffer.buffer, (err: NodeJS.ErrnoException | null | undefined) => {
211
if (err.code === 'EPIPE') {
212
// An EPIPE exception at the wrong time can lead to a renderer process crash
213
// so ignore the error since the socket will fire the close event soon anyways:
214
// > https://nodejs.org/api/errors.html#errors_common_system_errors
215
// > EPIPE (Broken pipe): A write on a pipe, socket, or FIFO for which there is no
216
// > process to read the data. Commonly encountered at the net and http layers,
217
// > indicative that the remote side of the stream being written to has been closed.
218
return;
219
}
220
onUnexpectedError(err);
221
}
223
>
} catch (err) {
224
if (err.code === 'EPIPE') {
225
// An EPIPE exception at the wrong time can lead to a renderer process crash
226
// so ignore the error since the socket will fire the close event soon anyways:
227
// > https://nodejs.org/api/errors.html#errors_common_system_errors
228
// > EPIPE (Broken pipe): A write on a pipe, socket, or FIFO for which there is no
229
// > process to read the data. Commonly encountered at the net and http layers,
230
// > indicative that the remote side of the stream being written to has been closed.
231
return;
232
}
233
onUnexpectedError(err);
234
}
237
>
public end(): void {
238
this.traceSocketEvent(SocketDiagnosticsEventType.NodeEndSent);
239
this.socket.end();
240
}
242
>
public drain(): Promise<void> {
244
>
return new Promise<void>((resolve, reject) => {
245
>
if (this.socket.bufferSize === 0) {
246
>
this.traceSocketEvent(SocketDiagnosticsEventType.NodeDrainEnd);
247
>
resolve();
248
>
return;
249
>
}
250
const finished = () => {
251
this.socket.off('close', finished);
252
this.socket.off('end', finished);
253
this.socket.off('error', finished);
254
this.socket.off('timeout', finished);
255
this.socket.off('drain', finished);
256
this.traceSocketEvent(SocketDiagnosticsEventType.NodeDrainEnd);
257
resolve();
258
};
259
this.socket.on('close', finished);
260
this.socket.on('end', finished);
261
this.socket.on('error', finished);
262
this.socket.on('timeout', finished);
263
this.socket.on('drain', finished);
265
>
}
267
>
268
>
const enum Constants {
269
>
MinHeaderByteSize = 2,
270
>
/**
271
>
* If we need to write a large buffer, we will split it into 256KB chunks and
272
>
* send each chunk as a websocket message. This is to prevent that the sending
273
>
* side is stuck waiting for the entire buffer to be compressed before writing
274
>
* to the underlying socket or that the receiving side is stuck waiting for the
275
>
* entire message to be received before processing the bytes.
276
>
*/
277
>
MaxWebSocketMessageLength = 256 * 1024 // 256 KB
278
>
}
279
>
280
>
const enum ReadState {
281
>
PeekHeader = 1,
282
>
ReadHeader = 2,
283
>
ReadBody = 3,
284
>
Fin = 4
285
>
}
286
>
287
>
interface ISocketTracer {
288
>
traceSocketEvent(type: SocketDiagnosticsEventType, data?: VSBuffer | Uint8Array | ArrayBuffer | ArrayBufferView | unknown): void;
289
>
}
290
>
291
>
interface FrameOptions {
292
>
compressed: boolean;
293
>
opcode: number;
294
>
}
295
>
296
>
/**
297
>
* See https://tools.ietf.org/html/rfc6455#section-5.2
298
>
*/
299
>
export class WebSocketNodeSocket extends Disposable implements ISocket, ISocketTracer {
300
>
301
>
public readonly socket: NodeSocket;
302
>
private readonly _flowManager: WebSocketFlowManager;
303
>
private readonly _incomingData: ChunkStream;
304
>
private readonly _onData = this._register(new Emitter<VSBuffer>());
305
>
private readonly _onClose = this._register(new Emitter<SocketCloseEvent>());
306
>
private readonly _maxSocketMessageLength: number;
307
>
private _isEnded = false;
308
>
309
>
private readonly _state = {
310
>
state: ReadState.PeekHeader,
311
>
readLen: Constants.MinHeaderByteSize,
312
>
fin: 0,
313
>
compressed: false,
314
>
firstFrameOfMessage: true,
315
>
mask: 0,
316
>
opcode: 0
317
>
};
318
>
319
>
public get permessageDeflate(): boolean {
320
>
return this._flowManager.permessageDeflate;
321
>
}
322
>
323
>
public get recordedInflateBytes(): VSBuffer {
325
>
}
327
>
public setRecordInflateBytes(record: boolean): void {
329
>
}
331
>
public traceSocketEvent(type: SocketDiagnosticsEventType, data?: VSBuffer | Uint8Array | ArrayBuffer | ArrayBufferView | unknown): void {
333
>
}
335
>
/**
336
>
* Create a socket which can communicate using WebSocket frames.
337
>
*
338
>
* **NOTE**: When using the permessage-deflate WebSocket extension, if parts of inflating was done
339
>
* in a different zlib instance, we need to pass all those bytes into zlib, otherwise the inflate
340
>
* might hit an inflated portion referencing a distance too far back.
341
>
*
342
>
* @param socket The underlying socket
343
>
* @param permessageDeflate Use the permessage-deflate WebSocket extension
344
>
* @param inflateBytes "Seed" zlib inflate with these bytes.
345
>
* @param recordInflateBytes Record all bytes sent to inflate
346
>
*/
347
>
constructor(socket: NodeSocket, permessageDeflate: boolean, inflateBytes: VSBuffer | null, recordInflateBytes: boolean, enableMessageSplitting = true) {
349
>
this.socket = socket;
350
>
this._maxSocketMessageLength = enableMessageSplitting ? Constants.MaxWebSocketMessageLength : Infinity;
351
>
this.traceSocketEvent(SocketDiagnosticsEventType.Created, { type: 'WebSocketNodeSocket', permessageDeflate, inflateBytesLength: inflateBytes?.byteLength || 0, recordInflateBytes });
352
>
this._flowManager = this._register(new WebSocketFlowManager(
353
>
this,
354
>
permessageDeflate,
355
>
inflateBytes,
356
>
recordInflateBytes,
357
>
this._onData,
358
>
(data, options) => this._write(data, options)
359
>
));
360
>
this._register(this._flowManager.onError((err) => {
361
// zlib errors are fatal, since we have no idea how to recover
362
console.error(err);
363
onUnexpectedError(err);
364
this._onClose.fire({
365
type: SocketCloseEventType.NodeSocketCloseEvent,
366
hadError: true,
367
error: err
368
});
370
>
this._incomingData = new ChunkStream();
371
>
this._register(this.socket.onData(data => this._acceptChunk(data)));
372
>
this._register(this.socket.onClose(async (e) => {
374
>
// and all data has been emitted
375
>
if (this._flowManager.isProcessingReadQueue()) {
376
>
await Event.toPromise(this._flowManager.onDidFinishProcessingReadQueue);
377
>
}
378
>
this._onClose.fire(e);
380
>
}
382
>
public override dispose(): void {
384
// Wait for any outstanding writes to finish before disposing
385
this._register(this._flowManager.onDidFinishProcessingWriteQueue(() => {
386
this.dispose();
387
}));
389
>
this.socket.dispose();
390
>
super.dispose();
391
>
}
392
>
}
394
>
public onData(listener: (e: VSBuffer) => void): IDisposable {
396
>
}
398
>
public onClose(listener: (e: SocketCloseEvent) => void): IDisposable {
400
>
}
402
>
public onEnd(listener: () => void): IDisposable {
403
return this.socket.onEnd(listener);
404
}
406
>
public write(buffer: VSBuffer): void {
407
>
// If we write many logical messages (let's say 1000 messages of 100KB) during a single process tick, we do
ipc.net.ts ×18
408
>
// this thing where we install a process.nextTick timer and group all of them together and we then issue a
409
>
// single WebSocketNodeSocket.write with a 100MB buffer.
410
>
//
411
>
// The first problem is that the actual writing to the underlying node socket will only happen after all of
412
>
// the 100MB have been deflated (due to waiting on zlib flush). The second problem is on the reading side,
413
>
// where we will get a single WebSocketNodeSocket.onData event fired when all the 100MB have arrived,
414
>
// delaying processing the 1000 received messages until all have arrived, instead of processing them as each
415
>
// one arrives.
416
>
//
417
>
// We therefore split the buffer into chunks, and issue a write for each chunk.
418
>
419
>
let start = 0;
420
>
while (start < buffer.byteLength) {
421
>
this._flowManager.writeMessage(buffer.slice(start, Math.min(start + this._maxSocketMessageLength, buffer.byteLength)), { compressed: true, opcode: 0x02 /* Binary frame */ });
422
>
start += this._maxSocketMessageLength;
423
>
}
424
>
}
426
>
private _write(buffer: VSBuffer, { compressed, opcode }: FrameOptions): void {
428
// Avoid ERR_STREAM_WRITE_AFTER_END
429
return;
430
}
432
>
this.traceSocketEvent(SocketDiagnosticsEventType.WebSocketNodeSocketWrite, buffer);
433
>
let headerLen = Constants.MinHeaderByteSize;
434
>
if (buffer.byteLength < 126) {
437
headerLen += 2;
439
>
headerLen += 8;
440
>
}
442
>
443
>
// The RSV1 bit indicates a compressed frame
444
>
const compressedFlag = compressed ? 0b01000000 : 0;
445
>
const opcodeFlag = opcode & 0b00001111;
446
>
header.writeUInt8(0b10000000 | compressedFlag | opcodeFlag, 0);
447
>
if (buffer.byteLength < 126) {
450
header.writeUInt8(126, 1);
451
let offset = 1;
452
header.writeUInt8((buffer.byteLength >>> 8) & 0b11111111, ++offset);
453
header.writeUInt8((buffer.byteLength >>> 0) & 0b11111111, ++offset);
455
>
header.writeUInt8(127, 1);
456
>
let offset = 1;
457
>
header.writeUInt8(0, ++offset);
458
>
header.writeUInt8(0, ++offset);
459
>
header.writeUInt8(0, ++offset);
460
>
header.writeUInt8(0, ++offset);
461
>
header.writeUInt8((buffer.byteLength >>> 24) & 0b11111111, ++offset);
462
>
header.writeUInt8((buffer.byteLength >>> 16) & 0b11111111, ++offset);
463
>
header.writeUInt8((buffer.byteLength >>> 8) & 0b11111111, ++offset);
464
>
header.writeUInt8((buffer.byteLength >>> 0) & 0b11111111, ++offset);
465
>
}
467
>
this.socket.write(VSBuffer.concat([header, buffer]));
468
>
}
470
>
public end(): void {
471
this._isEnded = true;
472
this.socket.end();
473
}
475
>
private _acceptChunk(data: VSBuffer): void {
477
return;
478
}
480
>
this._incomingData.acceptChunk(data);
481
>
482
>
while (this._incomingData.byteLength >= this._state.readLen) {
483
>
484
>
if (this._state.state === ReadState.PeekHeader) {
485
>
// peek to see if we can read the entire header
486
>
const peekHeader = this._incomingData.peek(this._state.readLen);
487
>
const firstByte = peekHeader.readUInt8(0);
488
>
const finBit = (firstByte & 0b10000000) >>> 7;
489
>
const rsv1Bit = (firstByte & 0b01000000) >>> 6;
490
>
const opcode = (firstByte & 0b00001111);
491
>
492
>
const secondByte = peekHeader.readUInt8(1);
493
>
const hasMask = (secondByte & 0b10000000) >>> 7;
494
>
const len = (secondByte & 0b01111111);
495
>
496
>
this._state.state = ReadState.ReadHeader;
497
>
this._state.readLen = Constants.MinHeaderByteSize + (hasMask ? 4 : 0) + (len === 126 ? 2 : 0) + (len === 127 ? 8 : 0);
498
>
this._state.fin = finBit;
499
>
if (this._state.firstFrameOfMessage) {
500
>
// if the frame is compressed, the RSV1 bit is set only for the first frame of the message
501
>
this._state.compressed = Boolean(rsv1Bit);
502
>
}
503
>
this._state.firstFrameOfMessage = Boolean(finBit);
504
>
this._state.mask = 0;
505
>
this._state.opcode = opcode;
506
>
507
>
this.traceSocketEvent(SocketDiagnosticsEventType.WebSocketNodeSocketPeekedHeader, { headerSize: this._state.readLen, compressed: this._state.compressed, fin: this._state.fin, opcode: this._state.opcode });
508
>
509
>
} else if (this._state.state === ReadState.ReadHeader) {
510
>
// read entire header
511
>
const header = this._incomingData.read(this._state.readLen);
512
>
const secondByte = header.readUInt8(1);
513
>
const hasMask = (secondByte & 0b10000000) >>> 7;
514
>
let len = (secondByte & 0b01111111);
515
>
516
>
let offset = 1;
517
>
if (len === 126) {
518
len = (
519
header.readUInt8(++offset) * 2 ** 8
520
+ header.readUInt8(++offset)
521
);
524
>
header.readUInt8(++offset) * 0
525
>
+ header.readUInt8(++offset) * 0
526
>
+ header.readUInt8(++offset) * 0
527
>
+ header.readUInt8(++offset) * 0
528
>
+ header.readUInt8(++offset) * 2 ** 24
529
>
+ header.readUInt8(++offset) * 2 ** 16
530
>
+ header.readUInt8(++offset) * 2 ** 8
531
>
+ header.readUInt8(++offset)
532
>
);
533
>
}
535
>
let mask = 0;
536
>
if (hasMask) {
538
>
header.readUInt8(++offset) * 2 ** 24
539
>
+ header.readUInt8(++offset) * 2 ** 16
540
>
+ header.readUInt8(++offset) * 2 ** 8
541
>
+ header.readUInt8(++offset)
542
>
);
543
>
}
545
>
this._state.state = ReadState.ReadBody;
546
>
this._state.readLen = len;
547
>
this._state.mask = mask;
548
>
549
>
this.traceSocketEvent(SocketDiagnosticsEventType.WebSocketNodeSocketPeekedHeader, { bodySize: this._state.readLen, compressed: this._state.compressed, fin: this._state.fin, mask: this._state.mask, opcode: this._state.opcode });
550
>
551
>
} else if (this._state.state === ReadState.ReadBody) {
552
>
// read body
553
>
554
>
const body = this._incomingData.read(this._state.readLen);
555
>
this.traceSocketEvent(SocketDiagnosticsEventType.WebSocketNodeSocketReadData, body);
556
>
557
>
unmask(body, this._state.mask);
558
>
this.traceSocketEvent(SocketDiagnosticsEventType.WebSocketNodeSocketUnmaskedData, body);
559
>
560
>
this._state.state = ReadState.PeekHeader;
561
>
this._state.readLen = Constants.MinHeaderByteSize;
562
>
this._state.mask = 0;
563
>
564
>
if (this._state.opcode <= 0x02 /* Continuation frame or Text frame or binary frame */) {
565
>
this._flowManager.acceptFrame(body, this._state.compressed, !!this._state.fin);
566
>
} else if (this._state.opcode === 0x09 /* Ping frame */) {
568
>
this._flowManager.writeMessage(body, { compressed: false, opcode: 0x0A /* Pong frame */ });
569
>
}
571
>
}
572
>
}
574
>
public async drain(): Promise<void> {
575
>
this.traceSocketEvent(SocketDiagnosticsEventType.WebSocketNodeSocketDrainBegin);
ipc.net.ts ×18
576
>
if (this._flowManager.isProcessingWriteQueue()) {
577
>
await Event.toPromise(this._flowManager.onDidFinishProcessingWriteQueue);
578
>
}
579
>
await this.socket.drain();
580
>
this.traceSocketEvent(SocketDiagnosticsEventType.WebSocketNodeSocketDrainEnd);
581
>
}
583
>
584
>
class WebSocketFlowManager extends Disposable {
585
>
586
>
private readonly _onError = this._register(new Emitter<Error>());
587
>
public readonly onError = this._onError.event;
588
>
589
>
private readonly _zlibInflateStream: ZlibInflateStream | null;
590
>
private readonly _zlibDeflateStream: ZlibDeflateStream | null;
591
>
private readonly _writeQueue: { data: VSBuffer; options: FrameOptions }[] = [];
592
>
private readonly _readQueue: { data: VSBuffer; isCompressed: boolean; isLastFrameOfMessage: boolean }[] = [];
593
>
594
>
private readonly _onDidFinishProcessingReadQueue = this._register(new Emitter<void>());
595
>
public readonly onDidFinishProcessingReadQueue = this._onDidFinishProcessingReadQueue.event;
596
>
597
>
private readonly _onDidFinishProcessingWriteQueue = this._register(new Emitter<void>());
598
>
public readonly onDidFinishProcessingWriteQueue = this._onDidFinishProcessingWriteQueue.event;
599
>
600
>
public get permessageDeflate(): boolean {
601
>
return Boolean(this._zlibInflateStream && this._zlibDeflateStream);
602
>
}
603
>
604
>
public get recordedInflateBytes(): VSBuffer {
606
>
return this._zlibInflateStream.recordedInflateBytes;
607
>
}
608
return VSBuffer.alloc(0);
611
>
public setRecordInflateBytes(record: boolean): void {
613
>
}
615
>
constructor(
617
>
permessageDeflate: boolean,
618
>
inflateBytes: VSBuffer | null,
619
>
recordInflateBytes: boolean,
620
>
private readonly _onData: Emitter<VSBuffer>,
621
>
private readonly _writeFn: (data: VSBuffer, options: FrameOptions) => void
622
>
) {
623
>
super();
624
>
if (permessageDeflate) {
626
>
// To simplify our logic, we don't negotiate the window size
627
>
// and simply dedicate (2^15) / 32kb per web socket
628
>
this._zlibInflateStream = this._register(new ZlibInflateStream(this._tracer, recordInflateBytes, inflateBytes, { windowBits: 15 }));
629
>
this._zlibDeflateStream = this._register(new ZlibDeflateStream(this._tracer, { windowBits: 15 }));
630
>
this._register(this._zlibInflateStream.onError((err) => this._onError.fire(err)));
631
>
this._register(this._zlibDeflateStream.onError((err) => this._onError.fire(err)));
634
>
this._zlibDeflateStream = null;
635
>
}
638
>
public writeMessage(data: VSBuffer, options: FrameOptions): void {
640
>
this._processWriteQueue();
641
>
}
643
>
private _isProcessingWriteQueue = false;
644
>
private async _processWriteQueue(): Promise<void> {
647
>
}
649
>
while (this._writeQueue.length > 0) {
650
>
const { data, options } = this._writeQueue.shift()!;
651
>
if (this._zlibDeflateStream && options.compressed) {
652
>
const compressedData = await this._deflateMessage(this._zlibDeflateStream, data);
ipc.net.ts ×18
653
>
this._writeFn(compressedData, options);
656
>
}
658
>
this._isProcessingWriteQueue = false;
659
>
this._onDidFinishProcessingWriteQueue.fire();
660
>
}
662
>
public isProcessingWriteQueue(): boolean {
664
>
}
666
>
/**
667
>
* Subsequent calls should wait for the previous `_deflateBuffer` call to complete.
668
>
*/
669
>
private _deflateMessage(zlibDeflateStream: ZlibDeflateStream, buffer: VSBuffer): Promise<VSBuffer> {
671
>
zlibDeflateStream.write(buffer);
672
>
zlibDeflateStream.flush(data => resolve(data));
673
>
});
674
>
}
676
>
public acceptFrame(data: VSBuffer, isCompressed: boolean, isLastFrameOfMessage: boolean): void {
678
>
this._processReadQueue();
679
>
}
681
>
private _isProcessingReadQueue = false;
682
>
private async _processReadQueue(): Promise<void> {
685
>
}
687
>
while (this._readQueue.length > 0) {
688
>
const frameInfo = this._readQueue.shift()!;
689
>
if (this._zlibInflateStream && frameInfo.isCompressed) {
691
>
// Even if permessageDeflate is negotiated, it is possible
692
>
// that the other side might decide to send uncompressed messages
693
>
// So only decompress messages that have the RSV 1 bit set
694
>
const data = await this._inflateFrame(this._zlibInflateStream, frameInfo.data, frameInfo.isLastFrameOfMessage);
695
>
this._onData.fire(data);
698
>
}
700
>
this._isProcessingReadQueue = false;
701
>
this._onDidFinishProcessingReadQueue.fire();
702
>
}
704
>
public isProcessingReadQueue(): boolean {
706
>
}
708
>
/**
709
>
* Subsequent calls should wait for the previous `transformRead` call to complete.
710
>
*/
711
>
private _inflateFrame(zlibInflateStream: ZlibInflateStream, buffer: VSBuffer, isLastFrameOfMessage: boolean): Promise<VSBuffer> {
713
>
// See https://tools.ietf.org/html/rfc7692#section-7.2.2
714
>
zlibInflateStream.write(buffer);
715
>
if (isLastFrameOfMessage) {
716
>
zlibInflateStream.write(VSBuffer.fromByteArray([0x00, 0x00, 0xff, 0xff]));
717
>
}
718
>
zlibInflateStream.flush(data => resolve(data));
719
>
});
720
>
}
722
>
723
>
class ZlibInflateStream extends Disposable {
724
>
725
>
private readonly _onError = this._register(new Emitter<Error>());
726
>
public readonly onError = this._onError.event;
727
>
728
>
private readonly _zlibInflate: InflateRaw;
729
>
private readonly _recordedInflateBytes: VSBuffer[] = [];
730
>
private readonly _pendingInflateData: VSBuffer[] = [];
731
>
private _recordInflateBytes: boolean;
732
>
733
>
public get recordedInflateBytes(): VSBuffer {
734
>
if (this._recordInflateBytes) {
735
>
return VSBuffer.concat(this._recordedInflateBytes);
736
>
}
737
>
return VSBuffer.alloc(0);
738
>
}
739
>
740
>
constructor(
742
>
recordInflateBytes: boolean,
743
>
inflateBytes: VSBuffer | null,
744
>
options: ZlibOptions
745
>
) {
746
>
super();
747
>
this._recordInflateBytes = recordInflateBytes;
748
>
this._zlibInflate = createInflateRaw(options);
749
>
this._zlibInflate.on('error', (err: Error) => {
750
this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibInflateError, { message: err?.message, code: (err as NodeJS.ErrnoException)?.code });
751
this._onError.fire(err);
753
>
this._zlibInflate.on('data', (data: Buffer) => {
754
>
this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibInflateData, data);
ipc.net.ts ×6
755
>
this._pendingInflateData.push(VSBuffer.wrap(data));
757
>
if (inflateBytes) {
758
this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibInflateInitialWrite, inflateBytes.buffer);
759
this._zlibInflate.write(inflateBytes.buffer);
760
this._zlibInflate.flush(() => {
761
this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibInflateInitialFlushFired);
762
this._pendingInflateData.length = 0;
763
});
764
}
767
>
public write(buffer: VSBuffer): void {
770
>
}
771
>
this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibInflateWrite, buffer);
ipc.net.ts ×6
772
>
this._zlibInflate.write(buffer.buffer);
773
>
}
775
>
public setRecordInflateBytes(record: boolean): void {
777
>
if (!record) {
778
>
this._recordedInflateBytes.length = 0;
779
>
}
780
>
}
782
>
public flush(callback: (data: VSBuffer) => void): void {
784
>
this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibInflateFlushFired);
785
>
const data = VSBuffer.concat(this._pendingInflateData);
786
>
this._pendingInflateData.length = 0;
787
>
callback(data);
788
>
});
789
>
}
791
>
public override dispose(): void {
793
>
this._pendingInflateData.length = 0;
794
>
try {
795
>
this._zlibInflate.close();
796
>
} catch {
797
// ignore errors while disposing
798
}
800
>
}
802
>
803
>
class ZlibDeflateStream extends Disposable {
804
>
805
>
private readonly _onError = this._register(new Emitter<Error>());
806
>
public readonly onError = this._onError.event;
807
>
808
>
private readonly _zlibDeflate: DeflateRaw;
809
>
private readonly _pendingDeflateData: VSBuffer[] = [];
810
>
811
>
constructor(
813
>
options: ZlibOptions
814
>
) {
815
>
super();
816
>
817
>
this._zlibDeflate = createDeflateRaw({
818
>
windowBits: 15
819
>
});
820
>
this._zlibDeflate.on('error', (err: Error) => {
821
this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibDeflateError, { message: err?.message, code: (err as NodeJS.ErrnoException)?.code });
822
this._onError.fire(err);
824
>
this._zlibDeflate.on('data', (data: Buffer) => {
825
>
this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibDeflateData, data);
ipc.net.ts ×18
826
>
this._pendingDeflateData.push(VSBuffer.wrap(data));
828
>
}
830
>
public write(buffer: VSBuffer): void {
831
>
this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibDeflateWrite, buffer.buffer);
ipc.net.ts ×18
832
>
this._zlibDeflate.write(<Buffer>buffer.buffer);
833
>
}
835
>
public flush(callback: (data: VSBuffer) => void): void {
837
>
this._zlibDeflate.flush(/*Z_SYNC_FLUSH*/2, () => {
838
>
this._tracer.traceSocketEvent(SocketDiagnosticsEventType.zlibDeflateFlushFired);
839
>
840
>
let data = VSBuffer.concat(this._pendingDeflateData);
841
>
this._pendingDeflateData.length = 0;
842
>
843
>
// See https://tools.ietf.org/html/rfc7692#section-7.2.1
844
>
data = data.slice(0, data.byteLength - 4);
845
>
846
>
callback(data);
847
>
});
848
>
}
850
>
public override dispose(): void {
852
>
try {
853
>
this._zlibDeflate.close();
854
>
} catch {
855
// ignore errors while disposing
856
}
858
>
}
860
>
862
>
if (mask === 0) {
864
>
}
866
>
for (let i = 0; i < cnt; i++) {
867
>
const v = buffer.readUInt32BE(i * 4);
868
>
buffer.writeUInt32BE(v ^ mask, i * 4);
869
>
}
870
>
const offset = cnt * 4;
871
>
const bytesLeft = buffer.byteLength - offset;
872
>
const m3 = (mask >>> 24) & 0b11111111;
873
>
const m2 = (mask >>> 16) & 0b11111111;
874
>
const m1 = (mask >>> 8) & 0b11111111;
875
>
if (bytesLeft >= 1) {
876
>
buffer.writeUInt8(buffer.readUInt8(offset) ^ m3, offset);
877
>
}
878
>
if (bytesLeft >= 2) {
879
buffer.writeUInt8(buffer.readUInt8(offset + 1) ^ m2, offset + 1);
880
}
882
buffer.writeUInt8(buffer.readUInt8(offset + 2) ^ m1, offset + 2);
883
}
886
>
// Read this before there's any chance it is overwritten
887
>
// Related to https://github.com/microsoft/vscode/issues/30624
888
>
export const XDG_RUNTIME_DIR = process.env['XDG_RUNTIME_DIR'];
889
>
890
>
const safeIpcPathLengths: { [platform: number]: number } = {
891
>
[Platform.Linux]: 107,
892
>
[Platform.Mac]: 103
893
>
};
894
>
895
>
export function createRandomIPCHandle(): string {
897
>
898
>
// Windows: use named pipe
899
>
if (process.platform === 'win32') {
900
return `\\\\.\\pipe\\vscode-ipc-${randomSuffix}-sock`;
901
}
903
>
// Mac & Unix: Use socket file
904
>
// Unix: Prefer XDG_RUNTIME_DIR over user data path
905
>
const basePath = process.platform !== 'darwin' && XDG_RUNTIME_DIR ? XDG_RUNTIME_DIR : tmpdir();
906
>
907
>
// As of Node.js 24, socket paths that exceed the
908
>
// platform limit cause an `EINVAL` error at bind time instead of being silently
909
>
// truncated. The suffix only needs to be unique, so trim it (while keeping enough
910
>
// entropy) to make the path fit within the limit.
911
>
// See https://github.com/nodejs/node/commit/75884678d7e7ef228c8f8f82b4c085258c70a823
912
>
const limit = safeIpcPathLengths[platform];
913
>
let suffix = randomSuffix;
914
>
if (typeof limit === 'number') {
915
>
const available = Math.max(0, (limit - 1) - join(basePath, `vscode-ipc-.sock`).length);
916
>
if (available < suffix.length) {
917
suffix = suffix.slice(0, available);
918
}
920
>
921
>
return join(basePath, `vscode-ipc-${suffix}.sock`);
922
>
}
924
>
export function createStaticIPCHandle(directoryPath: string, type: string, version: string): string {
926
>
const scopeForSocket = scope.substr(0, 8);
927
>
928
>
// Windows: use named pipe
929
>
if (process.platform === 'win32') {
930
return `\\\\.\\pipe\\${scopeForSocket}-${version}-${type}-sock`;
931
}
933
>
// Mac & Unix: Use socket file
934
>
// Unix: Prefer XDG_RUNTIME_DIR over user data path, unless portable
935
>
// Trim the version and type values for the socket to prevent too large
936
>
// file names causing issues: https://unix.stackexchange.com/q/367008
937
>
938
>
const versionForSocket = version.substr(0, 4);
939
>
const typeForSocket = type.substr(0, 6);
940
>
941
>
let result: string;
942
>
if (process.platform !== 'darwin' && XDG_RUNTIME_DIR && !process.env['VSCODE_PORTABLE']) {
943
result = join(XDG_RUNTIME_DIR, `vscode-${scopeForSocket}-${versionForSocket}-${typeForSocket}.sock`);
945
>
result = join(directoryPath, `${versionForSocket}-${typeForSocket}.sock`);
946
>
}
947
>
948
>
// Validate length. Unlike `createRandomIPCHandle`, the path here must be derived
949
>
// deterministically from `directoryPath` so that the server and its clients agree
950
>
// on the same socket. There is no random component to trim, so an over-long
951
>
// `--user-data-dir` can still produce a path that exceeds the platform limit.
952
>
validateIPCHandleLength(result);
953
>
954
>
return result;
955
>
}
958
>
const limit = safeIpcPathLengths[platform];
959
>
if (typeof limit === 'number' && handle.length >= limit) {
960
// https://nodejs.org/api/net.html#net_identifying_paths_for_ipc_connections
961
console.warn(`WARNING: IPC handle "${handle}" is longer than ${limit} chars, try a shorter --user-data-dir`);
962
}
965
>
export class Server extends IPCServer {
966
>
967
>
private static toClientConnectionEvent(server: NetServer): Event<ClientConnectionEvent> {
968
const onConnection = Event.fromNodeEventEmitter<Socket>(server, 'connection');
969
970
return Event.map(onConnection, socket => ({
971
protocol: new Protocol(new NodeSocket(socket, 'ipc-server-connection')),
972
onDidClientDisconnect: Event.once(Event.fromNodeEventEmitter<void>(socket, 'close'))
973
}));
974
}
976
>
private server: NetServer | null;
977
>
978
>
constructor(server: NetServer) {
979
super(Server.toClientConnectionEvent(server));
980
this.server = server;
981
}
983
>
override dispose(): void {
984
super.dispose();
985
if (this.server) {
986
this.server.close();
987
this.server = null;
988
}
989
}
991
>
992
>
export function serve(port: number): Promise<Server>;
993
>
export function serve(namedPipe: string): Promise<Server>;
994
>
export function serve(hook: number | string): Promise<Server> {
995
return new Promise<Server>((resolve, reject) => {
996
const server = createServer();
997
998
server.on('error', reject);
999
server.listen(hook, () => {
1000
server.removeListener('error', reject);
1001
resolve(new Server(server));
1002
});
1003
});
1004
}
1006
>
export function connect(options: { host: string; port: number }, clientId: string): Promise<Client>;
1007
>
export function connect(namedPipe: string, clientId: string): Promise<Client>;
1008
>
export function connect(hook: { host: string; port: number } | string, clientId: string): Promise<Client> {
1009
return new Promise<Client>((resolve, reject) => {
1010
let socket: Socket;
1011
1012
const callbackHandler = () => {
1013
socket.removeListener('error', reject);
1014
resolve(Client.fromSocket(new NodeSocket(socket, `ipc-client${clientId}`), clientId));
1015
};
1016
1017
if (typeof hook === 'string') {
1018
socket = createConnection(hook, callbackHandler);
1019
} else {
1020
socket = createConnection(hook, callbackHandler);
1021
}
1022
1023
socket.once('error', reject);
1024
});
1025
}