Convert test mocked_reauth to not use private APIs

This also introduces the interception proxy ability to inject messages.
done the interception part; need to implement injection in the proxy
now.

TODO document the injectMessage API; it’s currently in my notes.

test/common/modules/interception_proxy_client.js

@@ -11,6 +11,19 @@ define(['ably', 'json-rpc-2.0'], function (Ably, { JSONRPCClient, JSONRPCServer,
     console.log('Interception proxy client:', ...args);
   }
 
+  function serialize(type, deserialized) {
+    let data;
+
+    if (type === 'binary') {
+      const serialized = msgpack.encode(deserialized);
+      data = BufferUtils.base64Encode(serialized);
+    } else if (type === 'text') {
+      data = JSON.stringify(deserialized);
+    }
+
+    return { type, data };
+  }
+
   class InterceptionProxyClient {
     currentContext = null;
 
@@ -78,7 +91,7 @@ define(['ably', 'json-rpc-2.0'], function (Ably, { JSONRPCClient, JSONRPCServer,
       });
     }
 
-    // TODO explain motivation for this API (so that a lingering test can’t accidentally override the interception in your test; of course, the interception in your test might accidentally _intercept_ messages sent by a lingering test but that’s a separate issue)
+    // TODO explain motivation for this API (so that a lingering test can’t accidentally override the interception in your test; of course, the interception in your test might accidentally _intercept_ messages sent by a lingering test but that’s a separate issue). More broadly it’s a way of ensuring a test case’s effects don’t outlive its execution; perhaps we could do this using hooks instead
     //
     // This is written as (done, action) for compatibility with the way our tests are currently written; a promise-based version would be good to have too
     //
@@ -113,10 +126,7 @@ define(['ably', 'json-rpc-2.0'], function (Ably, { JSONRPCClient, JSONRPCServer,
         );
       }
 
-      this.currentContext = {
-        transformClientMessage: null,
-        transformServerMessage: null,
-      };
+      this.currentContext = new InterceptionContext(this.jsonRPC);
 
       const newDone = (error) => {
         this.currentContext = null;
@@ -127,6 +137,8 @@ define(['ably', 'json-rpc-2.0'], function (Ably, { JSONRPCClient, JSONRPCServer,
     }
 
     async transformInterceptedMessage(paramsDTO) {
+      this.currentContext?._recordSeenConnection(paramsDTO);
+
       let deserialized;
       if (paramsDTO.type === 'binary') {
         const data = BufferUtils.base64Decode(paramsDTO.data);
@@ -156,17 +168,57 @@ define(['ably', 'json-rpc-2.0'], function (Ably, { JSONRPCClient, JSONRPCServer,
       if (result === null) {
         return { action: 'drop' };
       } else {
-        let data;
+        return { action: 'replace', ...serialize(paramsDTO.type, result) };
+      }
+    }
+  }
+
+  class InterceptionContext {
+    transformClientMessage = null;
+    transformServerMessage = null;
+    // TODO this is a temporary API until I figure out what the right thing to do is (probably to add an interception proxy notification when a new connection is intercepted, and then infer it from the query param), but document it anyway
+    // elements are { type: 'binary' | 'text' }
+    //
+    // keyed by connection ID, ordered oldest-to-newest connection
+    #seenConnections = new Map();
+
+    constructor(jsonRPC) {
+      this.jsonRPC = jsonRPC;
+    }
 
-        if (paramsDTO.type === 'binary') {
-          const serialized = msgpack.encode(result);
-          data = BufferUtils.base64Encode(serialized);
-        } else if (paramsDTO.type === 'text') {
-          data = JSON.stringify(result);
-        }
+    _recordSeenConnection(transformInterceptedMessageParamsDTO) {
+      const { connectionID, type } = transformInterceptedMessageParamsDTO;
 
-        return { action: 'replace', type: paramsDTO.type, data };
+      if (this.#seenConnections.has(connectionID)) {
+        return;
+      }
+
+      this.#seenConnections.set(connectionID, { type });
+    }
+
+    // TODO the term "connection ID" is a bit overloaded (becuse it’s an Ably concept too)
+    get latestConnectionID() {
+      if (this.#seenConnections.size === 0) {
+        return null;
+      }
+
+      return Array.from(this.#seenConnections.keys()).pop();
+    }
+
+    async injectMessage(connectionID, deserialized, fromClient) {
+      const seenConnection = this.#seenConnections.get(connectionID);
+      if (!seenConnection) {
+        throw new Error(`Cannot inject message — have not seen a connection with ID ${connectionID}`);
       }
+
+      const params = {
+        connectionID,
+        fromClient,
+        ...serialize(seenConnection.type, deserialized),
+      };
+
+      log('sending injectMessage request with params', params);
+      await this.jsonRPC.request('injectMessage', params);
     }
   }
 

test/interception-proxy/ControlRPC.ts

@@ -3,6 +3,8 @@ import { WebSocketMessageData } from './WebSocketMessageData';
 
 export type ServerMethods = {
   startInterception(params: InterceptionModeDTO): {};
+  injectMessage(params: InjectMessageParamsDTO): InjectMessageResultDTO;
+
   // TODO how to represent a notification?
   mitmproxyReady(): void;
 };
@@ -87,3 +89,26 @@ export function createTransformInterceptedMessageResult(
 
   return { action };
 }
+
+export type InjectMessageParams = {
+  connectionID: string;
+  data: WebSocketMessageData;
+  fromClient: boolean;
+};
+
+export type InjectMessageParamsDTO = {
+  connectionID: string;
+  fromClient: boolean;
+} & WebSocketMessageDataDTO;
+
+export function createInjectMessageParams(dto: InjectMessageParamsDTO): InjectMessageParams {
+  return {
+    connectionID: dto.connectionID,
+    data: createWebSocketMessageData(dto),
+    fromClient: dto.fromClient,
+  };
+}
+
+export type InjectMessageResultDTO = {
+  id: string;
+};

test/interception-proxy/InterceptionContext.ts

@@ -1,7 +1,10 @@
 import {
   ClientMethods,
+  createInjectMessageParams,
   createTransformInterceptedMessageParamsDTO,
   createTransformInterceptedMessageResult,
+  InjectMessageParamsDTO,
+  InjectMessageResultDTO,
   InterceptionModeDTO,
   ServerMethods,
   TransformInterceptedMessageResult,
@@ -42,6 +45,7 @@ export class InterceptionContext {
     this.jsonRPC.addMethod('startInterception', (params, serverParams) =>
       this.startInterception(params, serverParams.controlServerConnection),
     );
+    this.jsonRPC.addMethod('injectMessage', (params) => this.injectMessage(params));
     this.jsonRPC.addMethod('mitmproxyReady', () => this.mitmproxyLauncher?.onMitmproxyReady());
   }
 
@@ -156,4 +160,26 @@ export class InterceptionContext {
       );
     }
   }
+
+  injectMessage(paramsDTO: InjectMessageParamsDTO): InjectMessageResultDTO {
+    const params = createInjectMessageParams(paramsDTO);
+    console.log('context received injectMessage with params', params);
+
+    const interceptedConnection = this.proxyServer!.getInterceptedConnection(params.connectionID);
+    if (!interceptedConnection) {
+      throw new Error(`No connection exists with ID ${params.connectionID}`);
+    }
+
+    // This ProxyMessage is a bit pointless; it’s just so I can generate an ID to return to clients for them to correlate with proxy logs
+    const message = new ProxyMessage(params.data, params.fromClient);
+    console.log(
+      `Injecting user-injected message ${message.id}, with type ${
+        message.data.type
+      } and data (${webSocketMessageDataLoggingDescription(message.data)})`,
+    );
+    // TODO consider whether injecting immediately is indeed the right thing to, or whether it should actually go at the end of the queue of messages awaiting a `transformInterceptedMessage` response
+    interceptedConnection.inject(message.fromClient, message.data);
+
+    return { id: message.id };
+  }
 }

test/interception-proxy/ProxyServer.ts

@@ -52,4 +52,8 @@ export class ProxyServer {
 
     console.log(`Started interception proxy server to receive traffic from mitmproxy on port ${port}`);
   }
+
+  getInterceptedConnection(id: string): InterceptedConnection | null {
+    return this.interceptedConnections.get(id) ?? null;
+  }
 }

test/realtime/auth.test.js

@@ -1,6 +1,12 @@
 'use strict';
 
-define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, Helper, async, chai) {
+define(['ably', 'shared_helper', 'async', 'chai', 'interception_proxy_client'], function (
+  Ably,
+  Helper,
+  async,
+  chai,
+  interceptionProxyClient,
+) {
   var currentTime;
   var exampleTokenDetails;
   var exports = {};
@@ -1033,42 +1039,51 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, Helper, async
      * Inject a fake AUTH message from realtime, check that we reauth and send our own in reply
      */
     it('mocked_reauth', function (done) {
-      var helper = this.test.helper,
-        rest = helper.AblyRest(),
-        authCallback = function (tokenParams, callback) {
-          // Request a token (should happen twice)
-          Helper.whenPromiseSettles(rest.auth.requestToken(tokenParams, null), function (err, tokenDetails) {
-            if (err) {
-              helper.closeAndFinish(done, realtime, err);
-              return;
-            }
-            callback(null, tokenDetails);
-          });
-        },
-        realtime = helper.AblyRealtime({ authCallback: authCallback, transports: [helper.bestTransport] });
+      interceptionProxyClient.intercept(done, (done, interceptionContext) => {
+        var helper = this.test.helper,
+          rest = helper.AblyRest(),
+          authCallback = function (tokenParams, callback) {
+            // Request a token (should happen twice)
+            Helper.whenPromiseSettles(rest.auth.requestToken(tokenParams, null), function (err, tokenDetails) {
+              if (err) {
+                helper.closeAndFinish(done, realtime, err);
+                return;
+              }
+              callback(null, tokenDetails);
+            });
+          },
+          realtime = helper.AblyRealtime({ authCallback: authCallback, transports: [helper.bestTransport] });
 
-      realtime.connection.once('connected', function () {
-        helper.recordPrivateApi('read.connectionManager.activeProtocol.transport');
-        var transport = realtime.connection.connectionManager.activeProtocol.transport,
-          originalSend = transport.send;
-        helper.recordPrivateApi('replace.transport.send');
-        /* Spy on transport.send to detect the outgoing AUTH */
-        transport.send = function (message) {
-          if (message.action === 17) {
-            try {
-              expect(message.auth.accessToken, 'Check AUTH message structure is as expected').to.be.ok;
-              helper.closeAndFinish(done, realtime);
-            } catch (err) {
-              helper.closeAndFinish(done, realtime, err);
+        realtime.connection.once('connected', function () {
+          /* Spy on client messages to detect the outgoing AUTH */
+          interceptionContext.transformClientMessage = ({ deserialized: message }) => {
+            if (message.action === 17) {
+              // TODO return value? the original code didn’t call originalSend. We should either:
+              // - make sure that we always return something (i.e. force it on to callers)
+              // - make sure that if nothing is returned then the interception proxy client makes this very obvious
+              // - make sure to clean up outstanding messages when the `intercept`-created `done` is called
+              //
+              // I think this is what’s causing this in the logs:
+              // Interception proxy client: got result of transforming message d814955d-8a15-4c2f-b873-1bd0c3448635 undefined
+              // and what's hence causing Realtime to send
+              // message: 'Invalid websocket message (decode failure). (See https://help.ably.io/error/40000 for help.)',
+              //
+              // TODO
+              // should we have a separate "spy" interception proxy API that doesn’t require a return value?
+              try {
+                expect(message.auth.accessToken, 'Check AUTH message structure is as expected').to.be.ok;
+                helper.closeAndFinish(done, realtime);
+              } catch (err) {
+                helper.closeAndFinish(done, realtime, err);
+              }
+              return null;
+            } else {
+              return message;
             }
-          } else {
-            helper.recordPrivateApi('call.transport.send');
-            originalSend.call(this, message);
-          }
-        };
-        /* Inject a fake AUTH from realtime */
-        helper.recordPrivateApi('call.transport.onProtocolMessage');
-        transport.onProtocolMessage({ action: 17 });
+          };
+          /* Inject a fake AUTH from realtime */
+          interceptionContext.injectMessage(interceptionContext.latestConnectionID, { action: 17 }, false);
+        });
       });
     });