Use proxy reporting endpoint to subscribe to blocked events

packages/safe-chain/src/registryProxy/builtInProxy/createBuiltInProxyServer.js

@@ -3,23 +3,24 @@ import { tunnelRequest } from "./tunnelRequestHandler.js";
 import { mitmConnect } from "./mitmRequestHandler.js";
 import { handleHttpProxyRequest } from "./plainHttpProxy.js";
 import { ui } from "../../environment/userInteraction.js";
-import chalk from "chalk";
 import { createInterceptorForUrl } from "./interceptors/createInterceptorForEcoSystem.js";
 import { getHasSuppressedVersions } from "./interceptors/npm/modifyNpmInfo.js";
 import { getCaCertPath } from "./certUtils.js";
 import { readFileSync } from "fs";
+import EventEmitter from "events";
 
 /** *
  * @returns {import("../registryProxy.js").SafeChainProxy} */
 export function createBuiltInProxyServer() {
   const SERVER_STOP_TIMEOUT_MS = 1000;
   /**
-   * @type {{port: number | null, blockedRequests: {packageName: string, version: string, url: string}[]}}
+   * @type {{port: number | null}}
    */
   const state = {
     port: null,
-    blockedRequests: [],
   };
+  /** @type {EventEmitter<import("../registryProxy.js").ProxyServerEvents>} */
+  const emitter = new EventEmitter();
 
   const server = http.createServer(
     // This handles direct HTTP requests (non-CONNECT requests)
@@ -31,14 +32,13 @@ export function createBuiltInProxyServer() {
   // This handles HTTPS requests via the CONNECT method
   server.on("connect", handleConnect);
 
-  return {
+  return Object.assign(emitter, {
     startServer: () => startServer(server),
     stopServer: () => stopServer(server),
-    verifyNoMaliciousPackages,
     hasSuppressedVersions: getHasSuppressedVersions,
     getServerPort: () => state.port,
     getCaCert,
-  };
+  });
 
   /**
    * @param {import("http").Server} server
@@ -102,7 +102,10 @@ export function createBuiltInProxyServer() {
         (
           /** @type {import("./interceptors/interceptorBuilder.js").MalwareBlockedEvent} */ event,
         ) => {
-          onMalwareBlocked(event.packageName, event.version, event.targetUrl);
+          emitter.emit("malwareBlocked", {
+            packageName: event.packageName,
+            packageVersion: event.version
+          });
         },
       );
 
@@ -114,41 +117,6 @@ export function createBuiltInProxyServer() {
     }
   }
 
-  /**
-   *
-   * @param {string} packageName
-   * @param {string} version
-   * @param {string} url
-   */
-  function onMalwareBlocked(packageName, version, url) {
-    state.blockedRequests.push({ packageName, version, url });
-  }
-
-  function verifyNoMaliciousPackages() {
-    if (state.blockedRequests.length === 0) {
-      // No malicious packages were blocked, so nothing to block
-      return true;
-    }
-
-    ui.emptyLine();
-
-    ui.writeInformation(
-      `Safe-chain: ${chalk.bold(
-        `blocked ${state.blockedRequests.length} malicious package downloads`,
-      )}:`,
-    );
-
-    for (const req of state.blockedRequests) {
-      ui.writeInformation(` - ${req.packageName}@${req.version} (${req.url})`);
-    }
-
-    ui.emptyLine();
-    ui.writeExitWithoutInstallingMaliciousPackages();
-    ui.emptyLine();
-
-    return false;
-  }
-
   function getCaCert() {
     try {
       const safeChainPath = getCaCertPath();

packages/safe-chain/src/registryProxy/ramaProxy/createRamaProxy.js

@@ -6,6 +6,8 @@ import { dirname, join } from "node:path";
 import { promisify } from "node:util";
 import { ui } from "../../environment/userInteraction.js";
 import { getLoggingLevel, LOGGING_VERBOSE } from "../../config/settings.js";
+import { getReportingServer } from "./reportingServer.js";
+import EventEmitter from "node:events";
 
 const readFilePromise = promisify(readFile);
 
@@ -37,23 +39,40 @@ export function getRamaPath() {
  * @returns {import("../registryProxy.js").SafeChainProxy} */
 export function createRamaProxy(ramaPath) {
   const tempDir = mkdtempSync(join(tmpdir(), "safe-chain-proxy-"));
+  const reportingServer = getReportingServer();
+  /** @type {EventEmitter<import("../registryProxy.js").ProxyServerEvents>} */
+  const emitter = new EventEmitter();
   /** @type {RamaProxyInstance | null} */
   let ramaInstance = null;
 
-  return {
+  return Object.assign(emitter, {
     startServer: async () => {
-      ramaInstance = await startRama(ramaPath, tempDir);
+      await reportingServer.start();
+      reportingServer.addListener("blockReceived", (ev) =>
+        emitter.emit("malwareBlocked", {
+          packageName: ev.artifact.identifier,
+          packageVersion: ev.artifact.version,
+        }),
+      );
+      ui.writeVerbose(
+        `Started reporting server at ${reportingServer.getAddress()}`,
+      );
+      ramaInstance = await startRama(
+        ramaPath,
+        tempDir,
+        reportingServer.getAddress(),
+      );
       ui.writeVerbose(
         `Proxy started at address "${ramaInstance.proxyAddress}"`,
       );
     },
     stopServer: async () => {
+      await reportingServer.stop();
       if (ramaInstance) {
         ramaInstance.process.kill();
       }
       return Promise.resolve();
     },
-    verifyNoMaliciousPackages: () => true,
     hasSuppressedVersions: () => false,
     getServerPort: () => {
       if (!ramaInstance) return null;
@@ -61,17 +80,25 @@ export function createRamaProxy(ramaPath) {
       return url.port ? parseInt(url.port, 10) : null;
     },
     getCaCert: () => ramaInstance?.caCert ?? null,
-  };
+  });
 }
 
 /**
  * @param {string} ramaPath
  * @param {string} dataFolder
+ * @param {string} reportingUrl
  * @returns {Promise<RamaProxyInstance>}
  */
-async function startRama(ramaPath, dataFolder) {
+async function startRama(ramaPath, dataFolder, reportingUrl) {
   const startTime = Date.now();
-  const args = ["--secrets", "memory", "--data", dataFolder];
+  const args = [
+    "--secrets",
+    "memory",
+    "--data",
+    dataFolder,
+    "--reporting-endpoint",
+    reportingUrl,
+  ];
   const process =
     getLoggingLevel() === LOGGING_VERBOSE
       ? spawn(ramaPath, args, {

packages/safe-chain/src/registryProxy/ramaProxy/reportingServer.js

@@ -0,0 +1,99 @@
+import * as http from "node:http";
+import { EventEmitter } from "node:events";
+
+const SERVER_STOP_TIMEOUT_MS = 1000;
+
+/**
+ * @typedef {Object} BlockEvent
+ * @property {number} ts_ms
+ * @property {{ product: string, identifier: string, version: string }} artifact
+ */
+
+/**
+ * @typedef {{ blockReceived: [BlockEvent] }} ReportingServerEvents
+ */
+
+/**
+ * @typedef {EventEmitter<ReportingServerEvents> & {
+ *   start: () => Promise<void>,
+ *   stop: () => Promise<void>,
+ *   getAddress: () => string,
+ * }} ReportingServer
+ */
+
+/**
+ * @returns {ReportingServer}
+ */
+export function getReportingServer() {
+  /** @type {EventEmitter<ReportingServerEvents>} */
+  const emitter = new EventEmitter();
+
+  /** @type {{server: http.Server | null, address: string }} */
+  let state = {server: null, address: ""};
+
+  return Object.assign(emitter, {
+    start: async () => {
+      state = await startServer(async (req, res) => {
+        if (req.method == "POST" && req.url?.startsWith("/events/block")) {
+          const blockEvent = await parseBlockEventFromRequest(req);
+          emitter.emit("blockReceived", blockEvent);
+        }
+        res.writeHead(200);
+        res.end();
+      });
+    },
+    stop: () => {
+      return /** @type {Promise<void>} */ (new Promise((resolve) => {
+        try {
+          if (!state.server) {
+            resolve();
+            return;
+          }
+          state.server.close(() => {
+            resolve();
+          });
+        } catch {
+          resolve();
+        }
+        setTimeout(() => resolve(), SERVER_STOP_TIMEOUT_MS);
+      }));
+    },
+    getAddress: () => state.address,
+  });
+}
+
+/**
+ * @param {http.IncomingMessage} req
+ * @returns {Promise<BlockEvent>}
+ */
+function parseBlockEventFromRequest(req) {
+  return new Promise((resolve, reject) => {
+    /** @type {Buffer[]} */
+    const chunks = [];
+    req.on("data", (chunk) => chunks.push(chunk));
+    req.on("end", () => resolve(JSON.parse(Buffer.concat(chunks).toString())));
+    req.on("error", reject);
+  });
+}
+
+/**
+ * @param {http.RequestListener} requestListener
+ * @returns {Promise<{server: http.Server, address: string}>}
+ */
+async function startServer(requestListener) {
+  let server = http.createServer(requestListener);
+
+  return await new Promise((resolve, reject) => {
+    server.listen(0, '127.0.0.1', () => {
+      const address = server.address();
+      if (address && typeof address === "object") {
+        resolve({
+          address: `http://${address.address}:${address.port}`,
+          server: server,
+        });
+      } else {
+        reject(new Error("Failed to start proxy server"));
+      }
+    });
+  });
+}

packages/safe-chain/src/registryProxy/registryProxy.connect-tunnel.spec.js

@@ -111,6 +111,9 @@ describe("registryProxy.connectTunnel", () => {
 
   describe("Error Handling", () => {
     it("should return 502 Bad Gateway for invalid hostname", async () => {
+      // We need to make sure we're not running behind an existing safe-chain installation to allow this test to work
+      const https_proxy = process.env.HTTPS_PROXY;
+      delete process.env.HTTPS_PROXY;
       const socket = await connectToProxy(proxyHost, proxyPort);
       const connectRequest = `CONNECT invalid.hostname.that.does.not.exist:443 HTTP/1.1\r\nHost: invalid.hostname.that.does.not.exist:443\r\n\r\n`;
       socket.write(connectRequest);
@@ -123,8 +126,11 @@ describe("registryProxy.connectTunnel", () => {
         });
       });
 
-      assert.ok(responseData.includes("HTTP/1.1 502 Bad Gateway"));
+      assert.ok(responseData.includes("HTTP/1.1 502 Bad Gateway"), responseData);
       socket.destroy();
+      if (https_proxy) {
+        process.env.HTTPS_PROXY = https_proxy;
+      }
     });
 
     it("should handle client disconnect during tunnel establishment", async () => {

packages/safe-chain/src/registryProxy/registryProxy.js

@@ -4,13 +4,20 @@ import { createBuiltInProxyServer } from "./builtInProxy/createBuiltInProxyServe
 import { getCombinedCaBundlePath } from "./certBundle.js";
 
 /**
- * @typedef {Object} SafeChainProxy
- * @prop {() => Promise<void>} startServer
- * @prop {() => Promise<void>} stopServer
- * @prop {() => boolean} verifyNoMaliciousPackages
- * @prop {() => boolean} hasSuppressedVersions
- * @prop {() => Number | null} getServerPort
- * @prop {() => string | null} getCaCert
+ * @typedef {Object} MalwareBlockedEvent
+ * @prop {string} packageName
+ * @prop {string} packageVersion
+ * 
+ * @typedef {{ malwareBlocked: [MalwareBlockedEvent] }} ProxyServerEvents
+ * 
+ * @import { EventEmitter } from "node:stream"
+ * @typedef {EventEmitter<ProxyServerEvents> & {
+ *   startServer: () => Promise<void>
+ *   stopServer: () => Promise<void>
+ *   getServerPort: () => Number | null
+ *   getCaCert: () => string | null
+ *   hasSuppressedVersions: () => boolean
+ * }} SafeChainProxy
  *
  * @typedef {Object} ProxySettings
  * @prop {string | null} proxyUrl
@@ -27,9 +34,10 @@ export function createSafeChainProxy() {
 
   let ramaPath = getRamaPath();
   if (ramaPath) {
-    ui.writeInformation("Starting safe-chain rama proxy");
+    ui.writeVerbose("Starting safe-chain rama proxy");
     server = createRamaProxy(ramaPath);
   } else {
+    ui.writeVerbose("Starting built-in proxy");
     server = createBuiltInProxyServer();
   }