Allow subscribing to 'rule-event', to monitor rule processing events

pimterry · pimterry · commit 8db095227086 · 2023-07-07T01:13:36.000+02:00
The first and main motivating example here is upstream passthrough
events. This allows you to see exactly what a passthrough rule is send &amp;
receiving from the server, not just what the client is sending &amp;
receiving to Mockttp in general.

This can be useful for low level debugging of proxy traffic and complex
processing or transformations, where the forwarded traffic is
confusingly different to the original input/resulting output.

This does not attempt to do the same for websocket passthrough yet, as
websocket passthrough doesn't support very much in terms of traffic
modification, but this will be added in future as that develops.

This required touching on a few details of how passthrough worked for
server response bodies, and _may_ have fixed some related bugs there
where the body could be read, not modified, but then lost and not
forwarded back to clients.
diff --git a/src/admin/mockttp-admin-model.ts b/src/admin/mockttp-admin-model.ts
@@ -32,6 +32,7 @@ const TLS_PASSTHROUGH_OPENED_TOPIC = 'tls-passthrough-opened';
 const TLS_PASSTHROUGH_CLOSED_TOPIC = 'tls-passthrough-closed';
 const TLS_CLIENT_ERROR_TOPIC = 'tls-client-error';
 const CLIENT_ERROR_TOPIC = 'client-error';
+const RULE_EVENT_TOPIC = 'rule-event';
 
 async function buildMockedEndpointData(endpoint: ServerMockedEndpoint): Promise<MockedEndpointData> {
     return {
@@ -131,6 +132,12 @@ export function buildAdminServerModel(
         })
     });
 
+    mockServer.on('rule-event', (evt) => {
+        pubsub.publish(RULE_EVENT_TOPIC, {
+            ruleEvent: evt
+        })
+    });
+
     return <any> {
         Query: {
             mockedEndpoints: async (): Promise<MockedEndpointData[]> => {
@@ -229,6 +236,9 @@ export function buildAdminServerModel(
             },
             failedClientRequest: {
                 subscribe: () => pubsub.asyncIterator(CLIENT_ERROR_TOPIC)
+            },
+            ruleEvent: {
+                subscribe: () => pubsub.asyncIterator(RULE_EVENT_TOPIC)
             }
         },
 
diff --git a/src/admin/mockttp-schema.ts b/src/admin/mockttp-schema.ts
@@ -32,6 +32,7 @@ export const MockttpSchema = gql`
         tlsPassthroughClosed: TlsPassthroughEvent!
         failedTlsRequest: TlsHandshakeFailure!
         failedClientRequest: ClientError!
+        ruleEvent: RuleEvent!
     }
 
     type MockedEndpoint {
@@ -113,6 +114,13 @@ export const MockttpSchema = gql`
         remotePort: Int!
     }
 
+    type RuleEvent {
+        requestId: ID!
+        ruleId: ID!
+        eventType: String!
+        eventData: Raw!
+    }
+
     type InitiatedRequest {
         id: ID!
         timingEvents: Json!
diff --git a/src/client/mockttp-admin-request-builder.ts b/src/client/mockttp-admin-request-builder.ts
@@ -435,6 +435,14 @@ export class MockttpAdminRequestBuilder {
                         body
                     }
                 }
+            }`,
+            'rule-event': gql`subscription OnRuleEvent {
+                ruleEvent {
+                    requestId
+                    ruleId
+                    eventType
+                    eventData
+                }
             }`
         }[event];
 
@@ -460,6 +468,13 @@ export class MockttpAdminRequestBuilder {
                 } else if (event === 'abort') {
                     normalizeHttpMessage(data, event);
                     data.error = data.error ? JSON.parse(data.error) : undefined;
+                } else if (event === 'rule-event') {
+                    const { eventData } = data;
+
+                    // Events may include raw body data buffers, serialized as base64:
+                    if (eventData.rawBody !== undefined) {
+                        eventData.rawBody = Buffer.from(eventData.rawBody, 'base64');
+                    }
                 } else {
                     normalizeHttpMessage(data, event);
                 }
@@ -477,6 +492,7 @@ export class MockttpAdminRequestBuilder {
                     query GetEndpointData($id: ID!) {
                         mockedEndpoint(id: $id) {
                             seenRequests {
+                                id,
                                 protocol,
                                 method,
                                 url,
diff --git a/src/mockttp.ts b/src/mockttp.ts
@@ -19,7 +19,8 @@ import {
     RulePriority,
     WebSocketMessage,
     WebSocketClose,
-    AbortedRequest
+    AbortedRequest,
+    RuleEvent
 } from "./types";
 import type { RequestRuleData } from "./rules/requests/request-rule";
 import type { WebSocketRuleData } from "./rules/websockets/websocket-rule";
@@ -544,6 +545,25 @@ export interface Mockttp {
      */
     on(event: 'client-error', callback: (error: ClientError) => void): Promise<void>;
 
+    /**
+     * Some rules may emit events with metadata about request processing. For example,
+     * passthrough rules may emit events about upstream server interactions.
+     *
+     * You can listen to rule-event to hear about all these events. When emitted,
+     * this will include the id of the request being processed, the id of the rule
+     * that fired the event, the type of the event, and the event data itself.
+     *
+     * This is only useful in some niche use cases, such as logging all proxied upstream
+     * requests made by the server, separately from the client connections handled.
+     *
+     * The callback will be called asynchronously from request handling. This function
+     * returns a promise, and the callback is not guaranteed to be registered until
+     * the promise is resolved.
+     *
+     * @category Events
+     */
+    on<T = unknown>(event: 'rule-event', callback: (event: RuleEvent<T>) => void): Promise<void>;
+
     /**
      * Adds the given HTTP request rules to the server.
      *
@@ -765,7 +785,8 @@ export type SubscribableEvent =
     | 'tls-passthrough-opened'
     | 'tls-passthrough-closed'
     | 'tls-client-error'
-    | 'client-error';
+    | 'client-error'
+    | 'rule-event';
 
 /**
  * @hidden
diff --git a/src/rules/requests/request-handlers.ts b/src/rules/requests/request-handlers.ts
@@ -140,7 +140,15 @@ function isSerializedBuffer(obj: any): obj is SerializedBuffer {
 }
 
 export interface RequestHandler extends RequestHandlerDefinition {
-    handle(request: OngoingRequest, response: OngoingResponse): Promise<void>;
+    handle(
+        request: OngoingRequest,
+        response: OngoingResponse,
+        options: RequestHandlerOptions
+    ): Promise<void>;
+}
+
+export interface RequestHandlerOptions {
+    emitEventCallback?: (type: string, event: unknown) => void;
 }
 
 export class SimpleHandler extends SimpleHandlerDefinition {
@@ -380,7 +388,11 @@ export class PassThroughHandler extends PassThroughHandlerDefinition {
         return this._trustedCACertificates;
     }
 
-    async handle(clientReq: OngoingRequest, clientRes: OngoingResponse) {
+    async handle(
+        clientReq: OngoingRequest,
+        clientRes: OngoingResponse,
+        options: RequestHandlerOptions
+    ) {
         // Don't let Node add any default standard headers - we want full control
         dropDefaultHeaders(clientRes);
 
@@ -735,8 +747,24 @@ export class PassThroughHandler extends PassThroughHandlerDefinition {
                 let serverStatusCode = serverRes.statusCode!;
                 let serverStatusMessage = serverRes.statusMessage
                 let serverRawHeaders = pairFlatRawHeaders(serverRes.rawHeaders);
+
+                // This is only set if we need to read the body here, for a callback or similar. If so,
+                // we keep the buffer in case we need it afterwards (if the cb doesn't replace it).
+                let originalBody: Buffer | undefined;
+
+                // This is set when we override the body data. Note that this doesn't mean we actually
+                // read & buffered the original data! With a fixed replacement body we can skip that.
                 let resBodyOverride: Uint8Array | undefined;
 
+                if (options.emitEventCallback) {
+                    options.emitEventCallback('passthrough-response-head', {
+                        statusCode: serverStatusCode,
+                        statusMessage: serverStatusMessage,
+                        httpVersion: serverRes.httpVersion,
+                        rawHeaders: serverRawHeaders
+                    });
+                }
+
                 if (isH2Downstream) {
                     serverRawHeaders = h1HeadersToH2(serverRawHeaders);
                 }
@@ -775,8 +803,8 @@ export class PassThroughHandler extends PassThroughHandlerDefinition {
                     } else if (replaceBodyFromFile) {
                         resBodyOverride = await fs.readFile(replaceBodyFromFile);
                     } else if (updateJsonBody) {
-                        const rawBody = await streamToBuffer(serverRes);
-                        const realBody = buildBodyReader(rawBody, serverRes.headers);
+                        originalBody = await streamToBuffer(serverRes);
+                        const realBody = buildBodyReader(originalBody, serverRes.headers);
 
                         if (await realBody.getJson() === undefined) {
                             throw new Error("Can't transform non-JSON response body");
@@ -795,26 +823,27 @@ export class PassThroughHandler extends PassThroughHandlerDefinition {
 
                         resBodyOverride = asBuffer(JSON.stringify(updatedBody));
                     } else if (matchReplaceBody) {
-                        const rawBody = await streamToBuffer(serverRes);
-                        const realBody = buildBodyReader(rawBody, serverRes.headers);
+                        originalBody = await streamToBuffer(serverRes);
+                        const realBody = buildBodyReader(originalBody, serverRes.headers);
 
-                        const originalBody = await realBody.getText();
-                        if (originalBody === undefined) {
+                        const originalBodyText = await realBody.getText();
+                        if (originalBodyText === undefined) {
                             throw new Error("Can't match & replace non-decodeable response body");
                         }
 
-                        let replacedBody = originalBody;
+                        let replacedBody = originalBodyText;
                         for (let [match, result] of matchReplaceBody) {
                             replacedBody = replacedBody!.replace(match, result);
                         }
 
-                        if (replacedBody !== originalBody) {
+                        if (replacedBody !== originalBodyText) {
                             resBodyOverride = asBuffer(replacedBody);
                         }
                     }
 
                     if (resBodyOverride) {
-                        // We always re-encode the body to match the resulting content-encoding header:
+                        // In the above cases, the overriding data is assumed to always be in decoded form,
+                        // so we re-encode the body to match the resulting content-encoding header:
                         resBodyOverride = await encodeBodyBuffer(
                             resBodyOverride,
                             serverHeaders
@@ -833,9 +862,8 @@ export class PassThroughHandler extends PassThroughHandlerDefinition {
                     serverRawHeaders = objectHeadersToRaw(serverHeaders);
                 } else if (this.beforeResponse) {
                     let modifiedRes: CallbackResponseResult | void;
-                    let body: Buffer;
 
-                    body = await streamToBuffer(serverRes);
+                    originalBody = await streamToBuffer(serverRes);
                     let serverHeaders = rawHeadersToObject(serverRawHeaders);
 
                     modifiedRes = await this.beforeResponse({
@@ -844,19 +872,14 @@ export class PassThroughHandler extends PassThroughHandlerDefinition {
                         statusMessage: serverRes.statusMessage,
                         headers: serverHeaders,
                         rawHeaders: _.cloneDeep(serverRawHeaders),
-                        body: buildBodyReader(body, serverHeaders)
+                        body: buildBodyReader(originalBody, serverHeaders)
                     });
 
                     if (modifiedRes === 'close') {
-                        // Dump the real response data and kill the client socket:
-                        serverRes.resume();
                         (clientReq as any).socket.end();
                         throw new AbortError('Connection closed intentionally by rule');
                     } else if (modifiedRes === 'reset') {
                         requireSocketResetSupport();
-
-                        // Dump the real response data and kill the client socket:
-                        serverRes.resume();
                         resetOrDestroy(clientReq);
                         throw new AbortError('Connection reset intentionally by rule');
                     }
@@ -880,11 +903,6 @@ export class PassThroughHandler extends PassThroughHandlerDefinition {
                             modifiedRes?.headers,
                             method === 'HEAD' // HEAD responses are allowed mismatched content-length
                         );
-                    } else {
-                        // If you don't specify a body override, we need to use the real
-                        // body anyway, because as we've read it already streaming it to
-                        // the response won't work
-                        resBodyOverride = body;
                     }
 
                     serverRawHeaders = objectHeadersToRaw(serverHeaders);
@@ -910,14 +928,40 @@ export class PassThroughHandler extends PassThroughHandlerDefinition {
                 if (resBodyOverride) {
                     // Return the override data to the client:
                     clientRes.end(resBodyOverride);
-                    // Dump the real response data:
-                    serverRes.resume();
 
+                    // Dump the real response data, in case that body wasn't read yet:
+                    serverRes.resume();
+                    resolve();
+                } else if (originalBody) {
+                    // If the original body was read, and not overridden, then send it
+                    // onward directly:
+                    clientRes.end(originalBody);
                     resolve();
                 } else {
+                    // Otherwise the body hasn't been read - stream it live:
                     serverRes.pipe(clientRes);
                     serverRes.once('end', resolve);
                 }
+
+                if (options.emitEventCallback) {
+                    if (!!resBodyOverride) {
+                        (originalBody
+                            ? Promise.resolve(originalBody)
+                            : streamToBuffer(serverRes)
+                        ).then((upstreamBody) => {
+                            options.emitEventCallback!('passthrough-response-body', {
+                                overridden: true,
+                                rawBody: upstreamBody
+                            });
+                        });
+                    } else {
+                        options.emitEventCallback('passthrough-response-body', {
+                            overridden: false
+                            // We don't bother buffering & re-sending the body if
+                            // it's the same as the one being sent to the client.
+                        });
+                    }
+                }
             })().catch(reject));
 
             serverReq.once('socket', (socket: net.Socket) => {
@@ -979,6 +1023,30 @@ export class PassThroughHandler extends PassThroughHandlerDefinition {
 
             // For similar reasons, we don't want any buffering on outgoing data at all if possible:
             serverReq.setNoDelay(true);
+
+            // Fire rule events, to allow in-depth debugging of upstream traffic & modifications,
+            // so anybody interested can see _exactly_ what we're sending upstream here:
+            if (options.emitEventCallback) {
+                options.emitEventCallback('passthrough-request-head', {
+                    method,
+                    protocol: protocol!.replace(/:$/, ''),
+                    hostname,
+                    port,
+                    path,
+                    rawHeaders
+                });
+
+                if (!!reqBodyOverride) {
+                    options.emitEventCallback('passthrough-request-body', {
+                        overridden: true,
+                        rawBody: reqBodyOverride
+                    });
+                } else {
+                    options.emitEventCallback!('passthrough-request-body', {
+                        overridden: false
+                    });
+                }
+            }
         })().catch(reject)
         ).catch((e: ErrorLike) => {
             // All errors anywhere above (thrown or from explicit reject()) should end up here.
diff --git a/src/rules/requests/request-rule.ts b/src/rules/requests/request-rule.ts
@@ -18,7 +18,10 @@ export interface RequestRule extends Explainable {
 
     // We don't extend the main interfaces for these, because MockRules are not Serializable
     matches(request: OngoingRequest): MaybePromise<boolean>;
-    handle(request: OngoingRequest, response: OngoingResponse, record: boolean): Promise<void>;
+    handle(request: OngoingRequest, response: OngoingResponse, options: {
+        record: boolean,
+        emitEventCallback?: (type: string, event: unknown) => void
+    }): Promise<void>;
     isComplete(): boolean | null;
 }
 
@@ -64,14 +67,19 @@ export class RequestRule implements RequestRule {
         return matchers.matchesAll(request, this.matchers);
     }
 
-    handle(req: OngoingRequest, res: OngoingResponse, record: boolean): Promise<void> {
+    handle(req: OngoingRequest, res: OngoingResponse, options: {
+        record?: boolean,
+        emitEventCallback?: (type: string, event: unknown) => void
+    }): Promise<void> {
         let handlerPromise = (async () => { // Catch (a)sync errors
-            return this.handler.handle(req, res);
+            return this.handler.handle(req, res, {
+                emitEventCallback: options.emitEventCallback
+            });
         })();
 
         // Requests are added to rule.requests as soon as they start being handled,
         // as promises, which resolve only when the response & request body is complete.
-        if (record) {
+        if (options.record) {
             this.requests.push(
                 Promise.race([
                     // When the handler resolves, the request is completed:
diff --git a/src/server/mockttp-server.ts b/src/server/mockttp-server.ts
diff --git a/src/types.ts b/src/types.ts
diff --git a/test/integration/subscriptions/rule-events.spec.ts b/test/integration/subscriptions/rule-events.spec.ts
