fix(mitm-socket): fix cpu spiking sockets

Author: blakebyrnes

.github/workflows/lint-and-test.yml

@@ -65,7 +65,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [macos-latest, windows-latest, ubuntu-latest]
-        node-version: [12, 13]
+        node-version: [12, 14]
 
     runs-on: ${{ matrix.os }}
 
@@ -102,6 +102,8 @@ jobs:
       - name: Install mitm
         run: yarn run build
         working-directory: ./mitm-socket
+        env:
+          SA_REBUILD_MITM_SOCKET: true
 
       - name: Copy built mitm
         run: cp -r mitm-socket/dist build/mitm-socket
@@ -114,7 +116,7 @@ jobs:
           SA_REPLAY_SKIP_BINARY_DOWNLOAD: 1
 
       - name: Run tests
-        run: yarn jest --testTimeout=15000 --maxConcurrency=2
+        run: yarn jest --testTimeout=15000 --runInBand
         working-directory: ./build
         env:
           SA_SHOW_REPLAY: false

.gitignore

@@ -26,3 +26,4 @@ emulators/data/DOM-*.json
 /.eslintcache
 /.network.db
 /website/scripts/generateAwaitedDOM.js
+connect

.nvmrc

@@ -1 +1 @@
-12.7.0
+12.19.0

core/lib/Session.ts

@@ -8,6 +8,7 @@ import RequestSession, {
   IRequestSessionRequestEvent,
   IRequestSessionResponseEvent,
   IResourceStateChangeEvent,
+  ISocketEvent,
 } from '@secret-agent/mitm/handlers/RequestSession';
 import * as Os from 'os';
 import IPuppetContext, {
@@ -131,6 +132,8 @@ export default class Session {
     requestSession.on('response', this.onMitmResponse.bind(this));
     requestSession.on('http-error', this.onMitmError.bind(this));
     requestSession.on('resource-state', this.onResourceStates.bind(this));
+    requestSession.on('socket-close', this.onSocketClose.bind(this));
+    requestSession.on('socket-connect', this.onSocketConnect.bind(this));
   }
 
   public async createTab() {
@@ -208,6 +211,14 @@ export default class Session {
     this.sessionState.captureResourceState(event.context.id, event.context.stateChanges);
   }
 
+  private onSocketClose(event: ISocketEvent) {
+    this.sessionState.captureSocketEvent(event);
+  }
+
+  private onSocketConnect(event: ISocketEvent) {
+    this.sessionState.captureSocketEvent(event);
+  }
+
   private async onNewTab(
     parentTab: Tab,
     page: IPuppetPage,

core/test/interact.test.ts

@@ -11,7 +11,7 @@ beforeAll(async () => {
   HumanEmulatorGhost.maxScrollDelayMillis = 0;
   koaServer = await Helpers.runKoaServer();
 });
-afterAll(Helpers.afterAll);
+afterAll(Helpers.afterAll, 30e3);
 afterEach(Helpers.afterEach);
 
 describe.each([['ghost'], ['basic'], ['skipper']])(
@@ -32,6 +32,7 @@ describe.each([['ghost'], ['basic'], ['skipper']])(
       });
       const meta = await Core.createTab({ humanEmulatorId });
       const core = Core.byTabId[meta.tabId];
+      Helpers.needsClosing.push(core);
       // @ts-ignore
       const session = core.session;
       await core.goto(`${koaServer.baseUrl}/mouse`);
@@ -70,6 +71,7 @@ describe.each([['ghost'], ['basic'], ['skipper']])(
       });
       const meta = await Core.createTab({ humanEmulatorId });
       const core = Core.byTabId[meta.tabId];
+      Helpers.needsClosing.push(core);
 
       await core.goto(`${koaServer.baseUrl}/input`);
       await core.execJsPath(['document', ['querySelector', 'input'], ['focus']]);
@@ -114,6 +116,7 @@ describe.each([['ghost'], ['basic'], ['skipper']])(
       });
       const meta = await Core.createTab({ humanEmulatorId });
       const core = Core.byTabId[meta.tabId];
+      Helpers.needsClosing.push(core);
       await core.goto(`${koaServer.baseUrl}/longpage`);
 
       const click = async (selector: string) => {

emulate-browsers/base/injected-scripts/polyfill.changes.ts

@@ -23,8 +23,12 @@ for (const change of args.changes || []) {
     const { descriptor } = descriptorInHierarchy;
 
     if (change.propertyName === '_value') {
-      descriptor.value = change.property;
-      Object.defineProperty(parent, property, descriptor);
+      if (descriptor.get) {
+        descriptor.get = proxyGetter(parent, property, () => change.property);
+      } else {
+        descriptor.value = change.property;
+        Object.defineProperty(parent, property, descriptor);
+      }
     }
 
     if (change.propertyName === '_get') {

mitm-socket/index.ts

@@ -1,36 +1,51 @@
 import { ChildProcess, spawn } from 'child_process';
 import * as net from 'net';
 import { promises as fs, unlink } from 'fs';
-import Log from '@secret-agent/commons/Logger';
-import {  IBoundLog } from '@secret-agent/core-interfaces/ILog';
-import { EventEmitter } from 'events';
-import { createPromise } from '@secret-agent/commons/utils';
 import * as os from 'os';
-
 import { v1 as uuid } from 'uuid';
+import Log from '@secret-agent/commons/Logger';
+import { createPromise } from '@secret-agent/commons/utils';
+import { TypedEventEmitter } from '@secret-agent/commons/eventUtils';
 
 const { log } = Log(module);
 
 const ext = os.platform() === 'win32' ? '.exe' : '';
 const libPath = `${__dirname}/dist/connect${ext}`;
 
-export default class MitmSocket {
+let idCounter = 0;
+
+export default class MitmSocket extends TypedEventEmitter<{
+  connect: void;
+  dial: void;
+  close: void;
+}> {
   public readonly socketPath: string;
   public alpn = 'http/1.1';
   public socket: net.Socket;
   public remoteAddress: string;
   public localAddress: string;
+  public serverName: string;
+
+  public id = (idCounter += 1);
+
+  public spawnTime: Date;
+  public dialTime: Date;
+  public connectTime: Date;
+  public closeTime: Date;
+
+  public get pid() {
+    return this.child?.pid;
+  }
 
   private isClosing = false;
   private isConnected = false;
   private child: ChildProcess;
-  private emitter = new EventEmitter();
   private connectError?: string;
 
-  private readonly logger: IBoundLog;
-
   constructor(readonly sessionId: string, readonly connectOpts: IGoTlsSocketConnectOpts) {
+    super();
     const id = uuid();
+    this.serverName = connectOpts.servername;
     this.socketPath =
       os.platform() === 'win32' ? `\\\\.\\pipe\\sa-${id}` : `${os.tmpdir()}/sa-${id}.sock`;
     this.logger = log.createChild(module, { sessionId });
@@ -55,18 +70,15 @@ export default class MitmSocket {
     this.connectOpts.tcpWindowSize = tcpVars.windowSize;
   }
 
-  public on(event: 'close', listener: (socket: net.Socket) => any) {
-    this.emitter.on(event, listener);
-  }
-
   public isHttp2() {
     return this.alpn === 'h2';
   }
 
   public close() {
     if (this.isClosing) return;
+    this.closeTime = new Date();
     this.isClosing = true;
-    this.emitter.emit('close');
+    this.emit('close');
     this.cleanupSocket();
     this.closeChild();
   }
@@ -94,6 +106,7 @@ export default class MitmSocket {
     const child = spawn(libPath, [this.socketPath, JSON.stringify(this.connectOpts)], {
       stdio: ['pipe', 'pipe', 'pipe'],
     });
+    this.spawnTime = new Date();
     this.child = child;
     child.stdout.setEncoding('utf8');
     child.stderr.setEncoding('utf8');
@@ -168,19 +181,23 @@ export default class MitmSocket {
   private onChildProcessMessage(messages: string) {
     for (const message of messages.split(/\r?\n/)) {
       if (message.startsWith('[DomainSocketPiper.Dialed]')) {
+        this.dialTime = new Date();
         const matches = message.match(/Remote: (.+), Local: (.+)/);
         if (matches?.length) {
           this.remoteAddress = matches[1];
           this.localAddress = matches[2];
         }
+        this.emit('dial');
       } else if (message === '[DomainSocketPiper.ReadyForConnect]') {
         this.onListening();
       } else if (message.startsWith('[DomainSocketPiper.Connected]')) {
         this.isConnected = true;
+        this.connectTime = new Date();
         const matches = message.match(/ALPN: (.+)/);
         if (matches?.length) {
           this.alpn = matches[1];
         }
+        this.emit('connect');
         this.logger.stats('SocketHandler.Connected', {
           alpn: this.alpn,
           host: this.connectOpts?.host,

mitm-socket/install.js

@@ -12,7 +12,7 @@ if (!fs.existsSync(outDir)) {
   fs.mkdirSync(outDir);
 }
 const { version } = packageJson;
-const releasesAssetsUrl = `https://github.com/ulixee/secret-agent/releases/download/v${version}/`;
+const releasesAssetsUrl = `https://github.com/ulixee/secret-agent/releases/download/v${version}`;
 
 // tslint:disable:no-console
 

mitm-socket/lib/domain_socket_piper.go

@@ -52,22 +52,29 @@ func (piper *DomainSocketPiper) Pipe(remoteConn net.Conn, sigc chan os.Signal) {
 	copyUntilTimeout := func(dst io.Writer, src net.Conn, notifyChan chan error, counter *uint32) {
 		var err error
 		var n int
-		var one = make([]byte, 1)
+		var data []byte = make([]byte, 1024)
 
 		for {
 			if atomic.LoadUint32(counter) > 0 {
 				return
 			}
 
-			src.SetReadDeadline(time.Now().Add(readTimeout))
-			n, err = src.Read(one)
+			n, err = src.Read(data)
+
 			if err != nil {
 				notifyChan <- err
-			} else if n > 0 {
-				dst.Write(one)
-				_, err := io.Copy(dst, src)
+				if atomic.LoadUint32(counter) > 0 {
+				    return
+				}
+				if err == io.EOF {
+                    time.Sleep(1 * time.Second)
+                }
+			} else if n == 0 {
+				time.Sleep(200 * time.Millisecond)
+			} else {
+				_, err = dst.Write(data[:n])
 				if err != nil {
-					notifyChan <- err
+				    notifyChan <- err
 				}
 			}
 		}
@@ -84,6 +91,7 @@ func (piper *DomainSocketPiper) Pipe(remoteConn net.Conn, sigc chan os.Signal) {
 	for {
 		select {
 		case <-sigc:
+			atomic.AddUint32(&piper.completeCounter, 1)
 			if piper.debug {
 				fmt.Println("DomainSocket -> Sigc")
 			}
@@ -117,7 +125,7 @@ func (piper *DomainSocketPiper) Pipe(remoteConn net.Conn, sigc chan os.Signal) {
 				return
 			}
 
-		case <-time.After(50 * time.Millisecond):
+		case <-time.After(100 * time.Millisecond):
 			if atomic.LoadUint32(&piper.completeCounter) > 0 {
 				if piper.debug {
 					fmt.Println("DomainSocket -> Closed")

mitm-socket/test/MitmSocket.test.ts

@@ -36,7 +36,7 @@ test('should handle http2 requests', async () => {
   const client = http2.connect('https://secretagent.dev', {
     createConnection: () => tlsConnection.socket,
   });
-  Helpers.onClose(() => new Promise(resolve => client.close(resolve)));
+  closeAfterTest(client);
 
   const request = client.request({ ':path': '/' });
   const httpResponse = await readResponse(request);
@@ -58,7 +58,7 @@ test('should be able to hit google using a Chrome79 Emulator', async () => {
   const client = http2.connect('https://www.google.com', {
     createConnection: () => tlsConnection.socket,
   });
-  Helpers.onClose(() => new Promise(resolve => client.close(resolve)));
+  closeAfterTest(client);
 
   const request = client.request({ ':path': '/' });
   const httpResponse = await readResponse(request);
@@ -67,7 +67,7 @@ test('should be able to hit google using a Chrome79 Emulator', async () => {
 });
 
 test('should be able to hit optimove using a Chrome79 Emulator', async () => {
-  const tlsConnection = new MitmSocket('2', {
+  const tlsConnection = new MitmSocket('optimove', {
     host: 'www.gstatic.com',
     port: '443',
     servername: 'www.gstatic.com',
@@ -82,7 +82,7 @@ test('should be able to hit optimove using a Chrome79 Emulator', async () => {
   const client = http2.connect('https://www.gstatic.com', {
     createConnection: () => tlsConnection.socket,
   });
-  Helpers.onClose(() => new Promise(resolve => client.close(resolve)));
+  closeAfterTest(client);
 
   const request = client.request({
     ':path': '/firebasejs/4.9.1/firebase.js',
@@ -108,7 +108,7 @@ test.skip('should be able to get scripts from unpkg using Chrome79 emulator', as
   const client = http2.connect('https://unpkg.com', {
     createConnection: () => tlsConnection.socket,
   });
-  trackClose(client);
+  closeAfterTest(client);
 
   {
     const request = client.request({ ':path': '/react@16.7.0/umd/react.production.min.js' });
@@ -322,12 +322,12 @@ async function startProxy() {
   });
   proxy.unref();
 
-  trackClose(proxy);
+  closeAfterTest(proxy);
   await proxyPromise.promise;
   return proxy;
 }
 
-function trackClose(closable: { close: (...args: any[]) => any }) {
+function closeAfterTest(closable: { close: (...args: any[]) => any }) {
   Helpers.onClose(() => new Promise(resolve => closable.close(() => process.nextTick(resolve))));
 }