fix: restore public file streaming and contain ZIP failures

test/streamer-public-access.test.js

@@ -0,0 +1,99 @@
+const { expect } = require("chai");
+const express = require("express");
+const got = require("got");
+const { Readable } = require("stream");
+require("ts-node/register/transpile-only");
+const config = require("../src/config").default;
+const { registerGitHubToken, githubTokenForStreamer } = require("../src/core/github-token-context");
+const File = require("../src/core/AnonymizedFile").default;
+const GitHubStream = require("../src/core/source/GitHubStream").default;
+const { AnonymizeTransformer } = require("../src/core/anonymize-utils");
+const streamer = require("../src/streamer/route").default;
+
+async function rejects(promise, message) {
+  try { await promise; } catch (error) { expect(error.message).to.equal(message); return; }
+  throw new Error("Expected rejection");
+}
+
+async function listen(app) {
+  return new Promise(resolve => {
+    const server = app.listen(0, "127.0.0.1", () => resolve(server));
+  });
+}
+
+describe("public repository streamer handoff", function () {
+  it("revalidates public access without exporting handles or owner credentials", async function () {
+    let valid = true;
+    registerGitHubToken("public-read:transport", {
+      quotaKey: "test", publicRepository: "owner/public",
+      renew: async () => { if (!valid) throw new Error("revoked"); return "private-owner-token"; },
+    });
+    expect(await githubTokenForStreamer("public-read:transport", "owner/public")).to.equal("");
+    expect(await githubTokenForStreamer("installation-token", "owner/private")).to.equal("installation-token");
+    await rejects(githubTokenForStreamer("public-read:transport", "other/repo"), "Public repository access context mismatch");
+    valid = false;
+    await rejects(githubTokenForStreamer("public-read:transport", "owner/public"), "revoked");
+    await rejects(githubTokenForStreamer("public-read:missing", "owner/public"), "Public repository access context expired");
+  });
+
+  for (const mode of ["send", "anonymizedContent"]) {
+    it(`serves a public README through ${mode} and a separate HTTP streamer`, async function () {
+      const previous = { endpoint: config.STREAMER_ENTRYPOINT, stream: got.stream, cache: GitHubStream.prototype.getFileContentCache };
+      const servers = [];
+      let payload;
+      const requests = [];
+      try {
+        registerGitHubToken("public-read:http-test", {
+          quotaKey: "test", publicRepository: "owner/public", renew: async () => "private-owner-token",
+        });
+        got.stream = (url, options) => {
+          if (!String(url).startsWith("https://github.com/")) return previous.stream(url, options);
+          requests.push({ url, options });
+          return Readable.from((async function* () {
+            for (const hook of options.hooks.beforeRequest) await hook();
+            yield Buffer.from("# README\nAlice wrote this.");
+          })());
+        };
+        GitHubStream.prototype.getFileContentCache = async function (path) {
+          return this.downloadWithFallback(await this.data.getToken(), "sha", path);
+        };
+        const upstream = express();
+        upstream.use(express.json());
+        upstream.use((req, _res, next) => { payload = req.body; next(); });
+        upstream.use("/api", streamer);
+        servers.push(await listen(upstream));
+        config.STREAMER_ENTRYPOINT = `http://127.0.0.1:${servers[0].address().port}/`;
+        const options = { terms: ["Alice"], image: true, link: true };
+        const repo = {
+          repoId: "test", options,
+          model: { source: { repositoryName: "owner/public", commit: "abc" } },
+          getToken: async () => "public-read:http-test",
+          generateAnonymizeTransformer: path => new AnonymizeTransformer({ ...options, filePath: path }),
+        };
+        const file = new File({ repository: repo, anonymizedPath: "README.md" });
+        file._file = { name: "README.md", path: "", sha: "sha", size: 25 };
+        const api = express();
+        api.get("/file", async (_req, res) => {
+          try {
+            if (mode === "send") await file.send(res);
+            else (await file.anonymizedContent()).pipe(res);
+          } catch (error) { res.status(500).json({ error: error.message }); }
+        });
+        servers.push(await listen(api));
+        const response = await got(`http://127.0.0.1:${servers[1].address().port}/file`);
+        expect(response.body).to.equal("# README\nXXXX-1 wrote this.");
+        expect(payload.token).to.equal("");
+        expect(JSON.stringify(payload)).not.to.include("public-read:");
+        expect(JSON.stringify(payload)).not.to.include("private-owner-token");
+        expect(requests).to.have.length(1);
+        expect(requests[0].url).to.equal("https://github.com/owner/public/raw/abc/README.md");
+        expect(requests[0].options.headers).not.to.have.property("authorization");
+      } finally {
+        got.stream = previous.stream;
+        GitHubStream.prototype.getFileContentCache = previous.cache;
+        config.STREAMER_ENTRYPOINT = previous.endpoint;
+        await Promise.all(servers.map(server => new Promise(resolve => server.close(resolve))));
+      }
+    });
+  }
+});

test/zip-stream-errors.test.js

@@ -0,0 +1,91 @@
+const { expect } = require("chai");
+const { Readable, PassThrough } = require("stream");
+const { setImmediate } = require("timers");
+const { once } = require("events");
+const vm = require("vm");
+const got = require("got");
+const archiver = require("archiver");
+require("ts-node/register/transpile-only");
+const GitHubDownload = require("../src/core/source/GitHubDownload").default;
+const { AnonymizeTransformer } = require("../src/core/anonymize-utils");
+const { streamAnonymizedZip } = require("../src/core/zipStream");
+
+async function fixture() {
+  const zip = archiver("zip");
+  const chunks = [];
+  zip.on("data", chunk => chunks.push(chunk));
+  const finished = once(zip, "end");
+  zip.append("private source content", { name: "repo/README.md" });
+  await zip.finalize();
+  await finished;
+  return Buffer.concat(chunks);
+}
+
+describe("ZIP stream errors", function () {
+  it("finishes a valid ZIP with anonymized file content", async function () {
+    const input = await fixture();
+    const previous = { stream: got.stream, zip: GitHubDownload.prototype.getZipUrl };
+    const response = new PassThrough();
+    const parser = require("unzip-stream").Parse();
+    const files = [];
+    const entries = [];
+    parser.on("entry", entry => {
+      const chunks = [];
+      entry.on("data", chunk => chunks.push(chunk));
+      entries.push(once(entry, "end").then(() => files.push({ name: entry.path, content: Buffer.concat(chunks).toString() })));
+    });
+    const finished = once(parser, "finish");
+    response.pipe(parser);
+    try {
+      GitHubDownload.prototype.getZipUrl = async () => ({ url: "https://example.test/archive.zip" });
+      got.stream = () => Readable.from([input]);
+      await streamAnonymizedZip({
+        repoId: "test", organization: "owner", repoName: "public", commit: "abc",
+        getToken: () => "", anonymizerOptions: { terms: ["private"], image: true, link: true },
+      }, response);
+      await finished;
+      await Promise.all(entries);
+      expect(files).to.deep.equal([{ name: "README.md", content: "XXXX-1 source content" }]);
+    } finally {
+      got.stream = previous.stream;
+      GitHubDownload.prototype.getZipUrl = previous.zip;
+      response.destroy();
+    }
+  });
+
+  it("aborts a download on an asynchronous anonymization timeout without crashing", async function () {
+    const input = await fixture();
+    const previous = { stream: got.stream, zip: GitHubDownload.prototype.getZipUrl, flush: AnonymizeTransformer.prototype._flush };
+    const response = new PassThrough();
+    const errors = [];
+    const chunks = [];
+    response.on("error", error => errors.push(error));
+    response.on("data", chunk => chunks.push(chunk));
+    const closed = new Promise(resolve => response.once("close", resolve));
+    const timeout = vm.runInNewContext('Object.assign(new Error("Script execution timed out after 1000ms"), { code: "ERR_SCRIPT_EXECUTION_TIMEOUT" })');
+    try {
+      GitHubDownload.prototype.getZipUrl = async () => ({ url: "https://example.test/archive.zip" });
+      got.stream = () => Readable.from([input]);
+      // VM timeout errors come from another realm and can arrive after the ZIP
+      // parser finishes, while archiver is still consuming the entry stream.
+      AnonymizeTransformer.prototype._flush = function (callback) {
+        setImmediate(() => callback(timeout));
+      };
+      await streamAnonymizedZip({
+        repoId: "test", organization: "owner", repoName: "public", commit: "abc",
+        getToken: () => "", anonymizerOptions: { terms: ["private"], image: true, link: true },
+      }, response);
+      await closed;
+      await new Promise(resolve => setImmediate(resolve));
+      expect(response.destroyed).to.equal(true);
+      expect(errors).to.have.length(1);
+      expect(errors[0].code).to.equal("ERR_SCRIPT_EXECUTION_TIMEOUT");
+      expect(Buffer.concat(chunks).includes(Buffer.from("private source content"))).to.equal(false);
+    } finally {
+      got.stream = previous.stream;
+      GitHubDownload.prototype.getZipUrl = previous.zip;
+      AnonymizeTransformer.prototype._flush = previous.flush;
+      response.destroy();
+    }
+  });
+});