diff --git a/src/core/AnonymizedFile.ts b/src/core/AnonymizedFile.ts index 73f4782..8d1ef4f 100644 --- a/src/core/AnonymizedFile.ts +++ b/src/core/AnonymizedFile.ts @@ -18,6 +18,7 @@ import FileModel from "./model/files/files.model"; import { IFile } from "./model/files/files.types"; import { FilterQuery } from "mongoose"; import { createLogger, serializeError } from "./logger"; +import { githubTokenForStreamer } from "./github-token-context"; const logger = createLogger("anonymized-file"); @@ -356,7 +357,7 @@ export default class AnonymizedFile { return got.stream(join(config.STREAMER_ENTRYPOINT, "api"), { method: "POST", json: { - token: await this.repository.getToken(), + token: await githubTokenForStreamer(await this.repository.getToken(), this.repository.model.source.repositoryName), repoFullName: this.repository.model.source.repositoryName, commit: this.repository.model.source.commit, branch: this.repository.model.source.branch, @@ -403,7 +404,7 @@ export default class AnonymizedFile { json: { sha, size, - token, + token: await githubTokenForStreamer(token, this.repository.model.source.repositoryName), repoFullName: this.repository.model.source.repositoryName, commit: this.repository.model.source.commit, branch: this.repository.model.source.branch, @@ -507,7 +508,7 @@ export default class AnonymizedFile { resolve(); }); } catch (error) { - handleError(error, res); + reject(error); } }); } diff --git a/src/core/github-token-context.ts b/src/core/github-token-context.ts index 3ea47ae..9f9c6f3 100644 --- a/src/core/github-token-context.ts +++ b/src/core/github-token-context.ts @@ -11,6 +11,17 @@ export function githubTokenContext(token: string) { if (!context && token.startsWith("public-read:")) throw new Error("Public repository access context expired"); return context; } +// Public handles belong to this process. Revalidate here, then let the +// streamer fetch public bytes anonymously without forwarding the owner's token. +export async function githubTokenForStreamer(token: string, repository: string | undefined): Promise { + const context = githubTokenContext(token); + if (!context?.publicRepository) return token; + if (!repository || context.publicRepository.toLowerCase() !== repository.toLowerCase()) { + throw new Error("Public repository access context mismatch"); + } + await context.renew(); + return ""; +} export function githubQuotaKey(token: string) { return contexts.get(token)?.quotaKey || createHash("sha256").update(token).digest("hex").slice(0, 24); } diff --git a/src/core/source/GitHubStream.ts b/src/core/source/GitHubStream.ts index 06fe5ff..c2eabfa 100644 --- a/src/core/source/GitHubStream.ts +++ b/src/core/source/GitHubStream.ts @@ -105,13 +105,13 @@ export default class GitHubStream extends GitHubBase { const url = githubRawFileUrl( this.data.organization, this.data.repoName, - this.data.commit, + this.data.commit || "HEAD", filePath ); logger.debug("downloading via raw URL (LFS)", { url }); return got.stream(url, { hooks: { beforeRequest: [async () => { await githubTokenContext(token)?.renew(); }] }, - headers: githubTokenContext(token)?.publicRepository ? {} : { authorization: `token ${token}` }, + headers: !token || githubTokenContext(token)?.publicRepository ? {} : { authorization: `token ${token}` }, followRedirect: true, }); } @@ -127,7 +127,7 @@ export default class GitHubStream extends GitHubBase { ): Promise { // Public raw downloads need no bearer token and do not consume the // unauthenticated REST API quota. GitHub also resolves LFS pointers here. - if (githubTokenContext(token)?.publicRepository) { + if (!token || githubTokenContext(token)?.publicRepository) { return Promise.resolve(this.downloadFileViaRaw(token, filePath)); } return new Promise((resolve) => { diff --git a/src/core/zipStream.ts b/src/core/zipStream.ts index 2d527c2..51e3baf 100644 --- a/src/core/zipStream.ts +++ b/src/core/zipStream.ts @@ -75,17 +75,23 @@ export async function streamAnonymizedZip( on(event: string, listener: (...args: unknown[]) => void): unknown; } ): Promise { - const source = new GitHubDownload({ - repoId: opt.repoId, - organization: opt.organization, - repoName: opt.repoName, - commit: opt.commit, - getToken: opt.getToken, - }); - let response; try { - response = await source.getZipUrl(); + const token = await opt.getToken(); + if (!token) { + // The API already checked public access. Codeload serves public archives + // directly, avoiding the streamer's shared unauthenticated REST quota. + response = { url: `https://codeload.github.com/${encodeURIComponent(opt.organization)}/${encodeURIComponent(opt.repoName)}/zip/${encodeURIComponent(opt.commit || "HEAD")}` }; + } else { + const source = new GitHubDownload({ + repoId: opt.repoId, + organization: opt.organization, + repoName: opt.repoName, + commit: opt.commit, + getToken: () => token, + }); + response = await source.getZipUrl(); + } } catch (error) { const code = await classifyGitHubMissError(error, { organization: opt.organization, @@ -124,9 +130,14 @@ export async function streamAnonymizedZip( // opens). Destroy the response instead so the client sees a connection // drop and knows the download failed. Same class of silent-truncation // bug as #694. - let upstreamSucceeded = false; + let failed = false; + const parser = Parse(); const fail = (error: Error) => { + if (failed) return; + failed = true; logger.error("upstream zipball failed", serializeError(error)); + downloadStream.destroy(); + downloadStream.unpipe(parser); archive.abort(); const destroyable = res as unknown as { destroy?: (err?: Error) => void; @@ -138,10 +149,13 @@ export async function streamAnonymizedZip( destroyable.end(); } }; + // pipe() returns the destination. Listen on the archive itself as well, + // including errors emitted after the upstream ZIP has finished downloading. + archive.on("error", fail); downloadStream .on("error", fail) - .pipe(Parse()) + .pipe(parser) .on("entry", (entry: NodeJS.ReadableStream & { type: string; path: string; autodrain: () => void }) => { if (entry.type === "File") { try { @@ -161,11 +175,13 @@ export async function streamAnonymizedZip( ...opt.anonymizerOptions, filePath: entry.path, }); + entry.on("error", fail); + anonymizer.on("error", fail); const st = entry.pipe(anonymizer); archive.append(st, { name: fileName }); } catch (error) { entry.autodrain(); - logger.error("entry transform failed", serializeError(error)); + fail(error as Error); } } else { entry.autodrain(); @@ -173,22 +189,13 @@ export async function streamAnonymizedZip( }) .on("error", fail) .on("finish", () => { - upstreamSucceeded = true; + if (failed) return; try { - archive.finalize(); - } catch { - /* ignored */ + archive.finalize().catch(fail); + } catch (error) { + fail(error as Error); } }); - archive.pipe(res).on("error", (error) => { - logger.error("archive pipe error", serializeError(error)); - if (!upstreamSucceeded) { - // archive errored while we were still depending on upstream bytes: - // treat as failure rather than truncating. - fail(error); - return; - } - (res as { end?: () => void }).end?.(); - }); + archive.pipe(res).on("error", fail); } diff --git a/src/server/routes/repository-public.ts b/src/server/routes/repository-public.ts index f4dbd90..092b3ea 100644 --- a/src/server/routes/repository-public.ts +++ b/src/server/routes/repository-public.ts @@ -12,6 +12,7 @@ import User from "../../core/User"; import { streamAnonymizedZip } from "../../core/zipStream"; import FileModel from "../../core/model/files/files.model"; import { createLogger, serializeError } from "../../core/logger"; +import { githubTokenForStreamer } from "../../core/github-token-context"; import gh = require("parse-github-url"); const logger = createLogger("repository-public"); @@ -62,7 +63,7 @@ router.get( .stream(join(config.STREAMER_ENTRYPOINT, "api/download"), { method: "POST", json: { - token, + token: await githubTokenForStreamer(token, repo.model.source.repositoryName), repoFullName: repo.model.source.repositoryName, commit: repo.model.source.commit, branch: repo.model.source.branch, diff --git a/test/streamer-public-access.test.js b/test/streamer-public-access.test.js new file mode 100644 index 0000000..e7eafaf --- /dev/null +++ b/test/streamer-public-access.test.js @@ -0,0 +1,122 @@ +const { expect } = require("chai"); +const express = require("express"); +const got = require("got"); +const { Readable, PassThrough } = 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"); + }); + + it("rejects send when the access recheck fails before opening the streamer", async function () { + const endpoint = config.STREAMER_ENTRYPOINT; + config.STREAMER_ENTRYPOINT = "http://unused.test/"; + const response = new PassThrough(); + try { + registerGitHubToken("public-read:revoked-send", { + quotaKey: "test", publicRepository: "owner/public", renew: async () => { throw new Error("revoked"); }, + }); + const file = new File({ repository: { + options: { terms: [] }, model: { source: { repositoryName: "owner/public" } }, + getToken: async () => "public-read:revoked-send", + generateAnonymizeTransformer: filePath => new AnonymizeTransformer({ terms: [], filePath }), + }, anonymizedPath: "README.md" }); + file._file = { name: "README.md", path: "", sha: "sha", size: 25 }; + await rejects(file.send(response), "revoked"); + } finally { + config.STREAMER_ENTRYPOINT = endpoint; + response.destroy(); + } + }); + + for (const [mode, commit] of [ + ["send", "abc"], ["anonymizedContent", "abc"], ["send", undefined], ["send", ""], + ]) { + it(`serves a public README through ${mode} with commit ${JSON.stringify(commit)}`, 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 } }, + 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/${commit || "HEAD"}/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)))); + } + }); + } +}); diff --git a/test/zip-stream-errors.test.js b/test/zip-stream-errors.test.js new file mode 100644 index 0000000..2903f5c --- /dev/null +++ b/test/zip-stream-errors.test.js @@ -0,0 +1,96 @@ +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 () { + for (const commit of ["abc", undefined, ""]) { + it(`downloads a public ZIP without a REST lookup with commit ${JSON.stringify(commit)}`, 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 () => { throw new Error("must not use the anonymous REST quota"); }; + got.stream = url => { + expect(url).to.equal(`https://codeload.github.com/owner/public/zip/${commit || "HEAD"}`); + return Readable.from([input]); + }; + await streamAnonymizedZip({ + repoId: "test", organization: "owner", repoName: "public", commit, + 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: () => "private-token", 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(); + } + }); +});