Add raw-passthrough-data events to monitor tunnel contents

Author: pimterry

src/admin/mockttp-admin-model.ts

@@ -34,6 +34,7 @@ const TLS_CLIENT_ERROR_TOPIC = 'tls-client-error';
 const CLIENT_ERROR_TOPIC = 'client-error';
 const RAW_PASSTHROUGH_OPENED_TOPIC = 'raw-passthrough-opened';
 const RAW_PASSTHROUGH_CLOSED_TOPIC = 'raw-passthrough-closed';
+const RAW_PASSTHROUGH_DATA_TOPIC = 'raw-passthrough-data';
 const RULE_EVENT_TOPIC = 'rule-event';
 
 async function buildMockedEndpointData(endpoint: ServerMockedEndpoint): Promise<MockedEndpointData> {
@@ -146,6 +147,12 @@ export function buildAdminServerModel(
         })
     });
 
+    mockServer.on('raw-passthrough-data', (evt) => {
+        pubsub.publish(RAW_PASSTHROUGH_DATA_TOPIC, {
+            rawPassthroughData: evt
+        })
+    });
+
     mockServer.on('rule-event', (evt) => {
         pubsub.publish(RULE_EVENT_TOPIC, {
             ruleEvent: evt
@@ -257,6 +264,9 @@ export function buildAdminServerModel(
             rawPassthroughClosed: {
                 subscribe: () => pubsub.asyncIterator(RAW_PASSTHROUGH_CLOSED_TOPIC)
             },
+            rawPassthroughData: {
+                subscribe: () => pubsub.asyncIterator(RAW_PASSTHROUGH_DATA_TOPIC)
+            },
             ruleEvent: {
                 subscribe: () => pubsub.asyncIterator(RULE_EVENT_TOPIC)
             }

src/admin/mockttp-schema.ts

@@ -33,6 +33,7 @@ export const MockttpSchema = gql`
         failedTlsRequest: TlsHandshakeFailure!
         rawPassthroughOpened: RawPassthroughEvent!
         rawPassthroughClosed: RawPassthroughEvent!
+        rawPassthroughData: RawPassthroughDataEvent!
         failedClientRequest: ClientError!
         ruleEvent: RuleEvent!
     }
@@ -130,6 +131,13 @@ export const MockttpSchema = gql`
         timingEvents: Json!
     }
 
+    type RawPassthroughDataEvent {
+        id: String!
+        direction: String!
+        content: Buffer!
+        eventTimestamp: Float!
+    }
+
     type RuleEvent {
         requestId: ID!
         ruleId: ID!

src/client/mockttp-admin-request-builder.ts

@@ -481,6 +481,14 @@ export class MockttpAdminRequestBuilder {
                     timingEvents
                 }
             }`,
+            'raw-passthrough-data': gql`subscription OnRawPassthroughData {
+                rawPassthroughData {
+                    id
+                    direction
+                    content
+                    eventTimestamp
+                }
+            }`,
             'rule-event': gql`subscription OnRuleEvent {
                 ruleEvent {
                     requestId
@@ -510,6 +518,8 @@ export class MockttpAdminRequestBuilder {
                     }
                 } else if (event === 'websocket-message-received' || event === 'websocket-message-sent') {
                     normalizeWebSocketMessage(data);
+                } else if (event === 'raw-passthrough-data') {
+                    data.content = Buffer.from(data.content, 'base64');
                 } else if (event === 'abort') {
                     normalizeHttpMessage(data, event);
                     data.error = data.error ? JSON.parse(data.error) : undefined;

src/mockttp.ts

@@ -21,7 +21,8 @@ import {
     WebSocketClose,
     AbortedRequest,
     RuleEvent,
-    RawPassthroughEvent
+    RawPassthroughEvent,
+    RawPassthroughDataEvent
 } from "./types";
 import type { RequestRuleData } from "./rules/requests/request-rule";
 import type { WebSocketRuleData } from "./rules/websockets/websocket-rule";
@@ -570,6 +571,15 @@ export interface Mockttp {
      */
     on(event: 'raw-passthrough-closed', callback: (req: RawPassthroughEvent) => void): Promise<void>;
 
+    /**
+     * Subscribe to hear about each chunk of data that is passed through the raw passthrough
+     * non-intercepted tunnels, due to the `passthrough` option. See `raw-passthrough-opened`
+     * for more details.
+     *
+     * @category Events
+     */
+    on(event: 'raw-passthrough-data', callback: (req: RawPassthroughDataEvent) => void): Promise<void>;
+
     /**
      * Some rules may emit events with metadata about request processing. For example,
      * passthrough rules may emit events about upstream server interactions.
@@ -884,6 +894,7 @@ export type SubscribableEvent =
     | 'client-error'
     | 'raw-passthrough-opened'
     | 'raw-passthrough-closed'
+    | 'raw-passthrough-data'
     | 'rule-event';
 
 /**

src/server/mockttp-server.ts

@@ -29,7 +29,8 @@ import {
     TlsPassthroughEvent,
     RuleEvent,
     RawTrailers,
-    RawPassthroughEvent
+    RawPassthroughEvent,
+    RawPassthroughDataEvent
 } from "../types";
 import { DestroyableServer } from "destroyable-server";
 import {
@@ -325,6 +326,7 @@ export class MockttpServer extends AbstractMockttp implements Mockttp {
     public on(event: 'client-error', callback: (error: ClientError) => void): Promise<void>;
     public on(event: 'raw-passthrough-opened', callback: (req: RawPassthroughEvent) => void): Promise<void>;
     public on(event: 'raw-passthrough-closed', callback: (req: RawPassthroughEvent) => void): Promise<void>;
+    public on(event: 'raw-passthrough-data', callback: (req: RawPassthroughDataEvent) => void): Promise<void>;
     public on<T = unknown>(event: 'rule-event', callback: (event: RuleEvent<T>) => void): Promise<void>;
     public on(event: string, callback: (...args: any[]) => void): Promise<void> {
         this.eventEmitter.on(event, callback);
@@ -1135,6 +1137,29 @@ ${await this.suggestRule(request)}`
         socket.pipe(upstreamSocket);
         upstreamSocket.pipe(socket);
 
+        if (type === 'raw') {
+            socket.on('data', (data) => {
+                setImmediate(() => {
+                    this.eventEmitter.emit('raw-passthrough-data', {
+                        id: eventData.id,
+                        direction: 'received',
+                        content: data,
+                        eventTimestamp: now()
+                    } satisfies RawPassthroughDataEvent);
+                });
+            });
+            upstreamSocket.on('data', (data) => {
+                setImmediate(() => {
+                    this.eventEmitter.emit('raw-passthrough-data', {
+                        id: eventData.id,
+                        direction: 'sent',
+                        content: data,
+                        eventTimestamp: now()
+                    } satisfies RawPassthroughDataEvent);
+                });
+            });
+        }
+
         socket.on('error', () => upstreamSocket.destroy());
         upstreamSocket.on('error', () => socket.destroy());
         upstreamSocket.on('close', () => socket.destroy());

src/types.ts

@@ -130,6 +130,32 @@ export interface RawPassthroughEvent {
     timingEvents: ConnectionTimingEvents;
 }
 
+export interface RawPassthroughDataEvent {
+    /**
+     * The id of the passthrough tunnel.
+     */
+    id: string;
+
+    /**
+     * The direction of the message, from the downstream perspective (received from the client,
+     * or sent back to the client).
+     */
+    direction: 'sent' | 'received';
+
+    /**
+     * The contents of the message as a raw buffer.
+     */
+    content: Uint8Array;
+
+    /**
+     * A high-precision floating-point monotonically increasing timestamp.
+     * Comparable and precise, but not related to specific current time.
+     *
+     * To link this to the current time, compare it to `timingEvents.startTime`.
+     */
+    eventTimestamp: number;
+}
+
 export interface ConnectionTimingEvents {
     /**
      * When the socket initially connected, in MS since the unix

test/integration/subscriptions/raw-passthrough-events.spec.ts

@@ -1,13 +1,14 @@
 import * as net from 'net';
 import { expect } from "chai";
 
-import { getAdminServer, getLocal, getRemote } from "../../..";
+import { getAdminServer, getLocal, getRemote, RawPassthroughDataEvent, RawPassthroughEvent } from "../../..";
 import {
     sendRawRequest,
     openSocksSocket,
     makeDestroyable,
     nodeOnly,
-    delay
+    delay,
+    getDeferred
 } from "../../test-utils";
 
 nodeOnly(() => {
@@ -21,7 +22,7 @@ nodeOnly(() => {
         // Simple TCP echo server:
         let remoteServer = makeDestroyable(net.createServer((socket) => {
             socket.on('data', (data) => {
-                socket.end(data);
+                socket.write(data);
             });
         }));
         let remotePort!: number;
@@ -65,6 +66,50 @@ nodeOnly(() => {
             expect(openEvent.upstreamPort).to.equal(remotePort);
         });
 
+        it("should expose sent & received data", async () => {
+            const openDeferred = getDeferred<RawPassthroughEvent>();
+            let dataEvents = [] as RawPassthroughDataEvent[];
+
+            await server.on('raw-passthrough-opened', (e) => openDeferred.resolve(e));
+            await server.on('raw-passthrough-data', (e) => dataEvents.push(e));
+
+            const socksSocket = await openSocksSocket(server, 'localhost', remotePort);
+
+            socksSocket.write('hello');
+
+            const openEvent = await openDeferred;
+            await delay(10);
+
+            expect(dataEvents.length).to.equal(2);
+            const [firstDataEvent, secondDataEvent] = dataEvents;
+            dataEvents = [];
+
+            expect(firstDataEvent.id).to.equal(openEvent.id);
+            expect(firstDataEvent.direction).to.equal('received');
+            expect(firstDataEvent.content.toString()).to.equal('hello');
+
+            expect(secondDataEvent.id).to.equal(openEvent.id);
+            expect(secondDataEvent.direction).to.equal('sent');
+            expect(secondDataEvent.content.toString()).to.equal('hello');
+            expect(secondDataEvent.eventTimestamp).to.be.greaterThan(firstDataEvent.eventTimestamp);
+
+            socksSocket.write('world');
+            await delay(10);
+
+            expect(dataEvents.length).to.equal(2);
+            const [thirdDataEvent, fourthDataEvent] = dataEvents;
+
+            expect(thirdDataEvent.id).to.equal(openEvent.id);
+            expect(thirdDataEvent.direction).to.equal('received');
+            expect(thirdDataEvent.content.toString()).to.equal('world');
+            expect(thirdDataEvent.eventTimestamp).to.be.greaterThan(secondDataEvent.eventTimestamp);
+
+            expect(fourthDataEvent.id).to.equal(openEvent.id);
+            expect(fourthDataEvent.direction).to.equal('sent');
+            expect(fourthDataEvent.content.toString()).to.equal('world');
+            expect(fourthDataEvent.eventTimestamp).to.be.greaterThan(thirdDataEvent.eventTimestamp);
+        });
+
         describe("with a remote client", () => {
             const adminServer = getAdminServer();
             const remoteClient = getRemote({
@@ -84,6 +129,7 @@ nodeOnly(() => {
             it("should fire for raw sockets that are passed through SOCKS", async () => {
                 const events: any[] = [];
                 await remoteClient.on('raw-passthrough-opened', (e) => events.push(e));
+                await remoteClient.on('raw-passthrough-data', (e) => events.push(e));
                 await remoteClient.on('raw-passthrough-closed', (e) => events.push(e));
 
                 const socksSocket = await openSocksSocket(remoteClient, 'localhost', remotePort);
@@ -92,12 +138,21 @@ nodeOnly(() => {
 
                 await delay(10);
 
-                expect(events.length).to.equal(2);
-                const [openEvent, closeEvent] = events;
+                expect(events.length).to.equal(4);
+                const [openEvent, receivedEvent, sentEvent, closeEvent] = events;
+                expect(receivedEvent.id).to.equal(openEvent.id);
+                expect(sentEvent.id).to.equal(openEvent.id);
                 expect(openEvent.id).to.equal(closeEvent.id);
 
                 expect(openEvent.upstreamHost).to.equal('localhost');
                 expect(openEvent.upstreamPort).to.equal(remotePort);
+
+                expect(receivedEvent.content.toString()).to.equal('123456789');
+                expect(receivedEvent.direction).to.equal('received');
+                expect(receivedEvent.eventTimestamp).to.be.greaterThan(openEvent.timingEvents.connectTimestamp);
+                expect(sentEvent.content.toString()).to.equal('123456789');
+                expect(sentEvent.direction).to.equal('sent');
+                expect(sentEvent.eventTimestamp).to.be.greaterThan(receivedEvent.eventTimestamp);
             });
         });