mirror of
https://github.com/tdurieux/anonymous_github.git
synced 2026-11-14 00:07:00 +01:00
merge: reconcile paid hosting with latest main
This commit is contained in:
Thomas Durieux committed
commit
ded6f905fd
180 files changed
+21709
-3614
No files matched your search
@@ -1,3 +1,4 @@
|
||||
import { notifyOwnerAccessProblem } from "./owner-notifications";
|
||||
import { streamResponse } from "./response-stream";
|
||||
import { join, basename, dirname } from "path";
|
||||
import { Response } from "express";
|
||||
@@ -152,8 +153,18 @@ export default class AnonymizedFile {
|
||||
return this._file.size;
|
||||
}
|
||||
|
||||
private visibleFile(file: IFile): IFile {
|
||||
if (this.repository.model?.source?.type !== "Zip" && file.name === ".gitmodules") {
|
||||
throw new AnonymousError("file_not_found", { httpStatus: 404 });
|
||||
}
|
||||
return file;
|
||||
}
|
||||
|
||||
async getFileInfo(): Promise<IFile> {
|
||||
if (this._file) return this._file;
|
||||
if (this.repository.model?.source?.type !== "Zip" && basename(this.anonymizedPath) === ".gitmodules") {
|
||||
throw new AnonymousError("file_not_found", { httpStatus: 404 });
|
||||
}
|
||||
if (this._file) return this.visibleFile(this._file);
|
||||
let fileDir = dirname(this.anonymizedPath);
|
||||
if (fileDir == ".") fileDir = "";
|
||||
if (fileDir.endsWith("/")) fileDir = fileDir.slice(0, -1);
|
||||
@@ -179,7 +190,7 @@ export default class AnonymizedFile {
|
||||
if (exact) {
|
||||
if (exact.metadataPending) await this.repository.completeRecoveredFileMetadata(exact);
|
||||
this._file = exact;
|
||||
return exact;
|
||||
return this.visibleFile(exact);
|
||||
}
|
||||
|
||||
const terms = this.repository.options.terms || [];
|
||||
@@ -212,7 +223,7 @@ export default class AnonymizedFile {
|
||||
const compiled = compileTerms(terms);
|
||||
if (typeof this.repository.findAnonymizedPath === "function") {
|
||||
const indexed = await this.repository.findAnonymizedPath(this.anonymizedPath);
|
||||
if (indexed) { this._file = indexed; return indexed; }
|
||||
if (indexed) { this._file = indexed; return this.visibleFile(indexed); }
|
||||
if (indexed === null) {
|
||||
if (anonymizePathCompiled(this.anonymizedPath, compiled) === this.anonymizedPath) {
|
||||
const recovered = await this.recoverTruncatedFile(fileDir);
|
||||
@@ -236,7 +247,7 @@ export default class AnonymizedFile {
|
||||
) {
|
||||
if (candidate.metadataPending) await this.repository.completeRecoveredFileMetadata(candidate);
|
||||
this._file = candidate;
|
||||
return candidate;
|
||||
return this.visibleFile(candidate);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -270,7 +281,10 @@ export default class AnonymizedFile {
|
||||
fetchFileInfoFromPath?: (filePath: string) => Promise<IFile | null>;
|
||||
};
|
||||
if (typeof source.fetchFileInfoFromPath !== "function") return null;
|
||||
const recovered = await source.fetchFileInfoFromPath(this.anonymizedPath);
|
||||
const recovered = await source.fetchFileInfoFromPath(this.anonymizedPath).catch(error => {
|
||||
void notifyOwnerAccessProblem(this.repository.owner.id, error, { kind: "repository", id: String(this.repository.model._id) });
|
||||
throw error;
|
||||
});
|
||||
if (!recovered) return null;
|
||||
recovered.repoId = this.repository.repoId;
|
||||
recovered.treeGeneration = this.repository.model.treeGeneration;
|
||||
@@ -309,6 +323,7 @@ export default class AnonymizedFile {
|
||||
if (!this._file) {
|
||||
this._file = await this.getFileInfo();
|
||||
}
|
||||
this.visibleFile(this._file);
|
||||
return join(this._file.path, this._file.name);
|
||||
}
|
||||
extension() {
|
||||
@@ -356,7 +371,11 @@ export default class AnonymizedFile {
|
||||
httpStatus: 413,
|
||||
});
|
||||
}
|
||||
const content = await this.repository.source?.getFileContent(this);
|
||||
const content = await this.repository.source?.getFileContent(this).catch(error => {
|
||||
void notifyOwnerAccessProblem(this.repository.owner.id, error, { kind: "repository", id: String(this.repository.model._id) });
|
||||
throw error;
|
||||
});
|
||||
content.on("error", error => { void notifyOwnerAccessProblem(this.repository.owner.id, error, { kind: "repository", id: String(this.repository.model._id) }); });
|
||||
const cacheWasReset = this.repository.model.isReseted;
|
||||
if (cacheWasReset) {
|
||||
await this.repository.markCachePresent();
|
||||
@@ -379,7 +398,7 @@ export default class AnonymizedFile {
|
||||
}
|
||||
|
||||
// use the streamer service
|
||||
return got.stream(join(config.STREAMER_ENTRYPOINT, "api"), {
|
||||
const content = got.stream(join(config.STREAMER_ENTRYPOINT, "api"), {
|
||||
method: "POST",
|
||||
headers: requestHeaders(),
|
||||
json: {
|
||||
@@ -391,11 +410,17 @@ export default class AnonymizedFile {
|
||||
branch: this.repository.model.source.branch,
|
||||
repoId: this.repository.repoId,
|
||||
filePath: this.filePath,
|
||||
upstream: (await this.getFileInfo()).upstream,
|
||||
sha: await this.sha(),
|
||||
size: await this.size(),
|
||||
anonymizerOptions: anonymizer.opt,
|
||||
},
|
||||
});
|
||||
content.on("error", err => {
|
||||
const { error } = streamerErrorToAnonymous(err, { repoId: this.repository.repoId, filePath: this.anonymizedPath });
|
||||
void notifyOwnerAccessProblem(this.repository.owner.id, error, { kind: "repository", id: String(this.repository.model._id) });
|
||||
});
|
||||
return content;
|
||||
}
|
||||
|
||||
get filePath() {
|
||||
@@ -434,6 +459,7 @@ export default class AnonymizedFile {
|
||||
json: {
|
||||
sha,
|
||||
size,
|
||||
upstream: (await this.getFileInfo()).upstream,
|
||||
token: await githubTokenForStreamer(token, this.repository.model.source.repositoryName),
|
||||
repoFullName: this.repository.model.source.repositoryName,
|
||||
commit: this.repository.model.source.commit,
|
||||
@@ -473,6 +499,7 @@ export default class AnonymizedFile {
|
||||
filePath: this.anonymizedPath,
|
||||
}
|
||||
);
|
||||
void notifyOwnerAccessProblem(this.repository.owner.id, error, { kind: "repository", id: String(this.repository.model._id) });
|
||||
error.value = this;
|
||||
handleError(error, res);
|
||||
if (res.headersSent && !res.writableEnded) res.destroy();
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyOwnerAccessProblem, withOwnerAccessRecovery } from "./owner-notifications";
|
||||
import { registerGitHubToken } from "./github-token-context";
|
||||
import { isConnected } from "../server/database";
|
||||
import { expireEmbeddedContent } from "./content-expiration";
|
||||
@@ -67,6 +68,16 @@ export default class Gist {
|
||||
}
|
||||
|
||||
async download() {
|
||||
try { await withOwnerAccessRecovery(this.owner.id,
|
||||
{ kind: "gist", id: String(this.model._id) }, () => this.downloadWithAccess(),
|
||||
{ "source.gistId": this.model.source.gistId }); }
|
||||
catch (error) {
|
||||
void notifyOwnerAccessProblem(this.owner.id, error, { kind: "gist", id: String(this.model._id) });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async downloadWithAccess() {
|
||||
const access = await this.getAccess();
|
||||
try {
|
||||
await this.downloadContent(access.token);
|
||||
|
||||
+37
-90
@@ -1,3 +1,6 @@
|
||||
import { isGitHubRateLimitError, OctokitRequestErrorLike } from "./github-rate-limit";
|
||||
import { refreshLegacyToken } from "./legacy-token-refresh";
|
||||
import { notifyOwnerAccessProblem } from "./owner-notifications";
|
||||
import AnonymizedRepositoryModel from "./model/anonymizedRepositories/anonymizedRepositories.model";
|
||||
import { isConnected } from "../server/database";
|
||||
import { githubQuotaKey, githubTokenContext } from "./github-token-context";
|
||||
@@ -8,42 +11,15 @@ import { createClient, RedisClientType } from "redis";
|
||||
|
||||
import AnonymousError from "./AnonymousError";
|
||||
import Repository from "./Repository";
|
||||
import { getCredential, replaceCredential, getCredentialToken } from "./credentials";
|
||||
import { getCredential, getCredentialToken } from "./credentials";
|
||||
import config from "../config";
|
||||
import { createLogger } from "./logger";
|
||||
import { measureStage } from "./request-monitoring";
|
||||
import { setTimeout as delay } from "timers/promises";
|
||||
|
||||
const logger = createLogger("github");
|
||||
|
||||
// Octokit RequestError shape (subset we care about for rate-limit detection).
|
||||
interface OctokitRequestErrorLike {
|
||||
status?: number;
|
||||
message?: string;
|
||||
response?: {
|
||||
headers?: Record<string, string | undefined>;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect GitHub rate-limit / abuse responses (primary 5k/h or undocumented
|
||||
* "secondary" limits) and rewrap them as a translatable AnonymousError so
|
||||
* the UI can show a friendly message instead of a raw HttpError stack. The
|
||||
* GitHub `x-github-request-id` header is preserved in `detail` so users can
|
||||
* cite it if they reach out to GitHub Support.
|
||||
*/
|
||||
export function isGitHubRateLimitError(err: unknown): boolean {
|
||||
if (!err || typeof err !== "object") return false;
|
||||
const e = err as OctokitRequestErrorLike;
|
||||
const msg = (e.message ?? "").toLowerCase();
|
||||
// Primary limits return 403 with "x-ratelimit-remaining: 0"; secondary
|
||||
// limits return 403 (sometimes 429) with "secondary rate limit" in the
|
||||
// body. Match on either signal so we catch both.
|
||||
const status = e.status ?? 0;
|
||||
if (status !== 403 && status !== 429) return false;
|
||||
if (msg.includes("rate limit") || msg.includes("abuse")) return true;
|
||||
const remaining = e.response?.headers?.["x-ratelimit-remaining"];
|
||||
return remaining === "0";
|
||||
}
|
||||
export { isGitHubRateLimitError } from "./github-rate-limit";
|
||||
|
||||
function rateLimitDetail(err: OctokitRequestErrorLike): string {
|
||||
const headers = err.response?.headers ?? {};
|
||||
@@ -110,7 +86,18 @@ export function getTokenGateResetAt(token: string): number {
|
||||
return gate.resetAt;
|
||||
}
|
||||
|
||||
async function waitForTokenGate(token: string): Promise<void> {
|
||||
function abortableGateLookup(lookup: Promise<number>, signal: AbortSignal): Promise<number> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const cancelled = () => { signal.removeEventListener("abort", cancelled); reject(signal.reason); };
|
||||
if (signal.aborted) cancelled();
|
||||
else signal.addEventListener("abort", cancelled, { once: true });
|
||||
lookup.then(value => { signal.removeEventListener("abort", cancelled); resolve(value); },
|
||||
error => { signal.removeEventListener("abort", cancelled); reject(error); });
|
||||
});
|
||||
}
|
||||
|
||||
async function waitForTokenGate(token: string, signal?: AbortSignal): Promise<void> {
|
||||
signal?.throwIfAborted();
|
||||
const key = githubQuotaKey(token);
|
||||
const localGate = tokenGates.get(key);
|
||||
let waitMs = 0;
|
||||
@@ -121,7 +108,9 @@ async function waitForTokenGate(token: string): Promise<void> {
|
||||
waitMs = resetAt - Date.now();
|
||||
}
|
||||
|
||||
const redisResetAt = await getRedisGateResetAt(key);
|
||||
const lookup = getRedisGateResetAt(key);
|
||||
const redisResetAt = await (signal ? abortableGateLookup(lookup, signal) : lookup);
|
||||
signal?.throwIfAborted();
|
||||
if (redisResetAt > resetAt) {
|
||||
resetAt = redisResetAt;
|
||||
waitMs = resetAt - Date.now();
|
||||
@@ -137,7 +126,7 @@ async function waitForTokenGate(token: string): Promise<void> {
|
||||
waitMs,
|
||||
resetAt: new Date(resetAt).toISOString(),
|
||||
});
|
||||
await new Promise((resolve) => setTimeout(resolve, waitMs));
|
||||
await delay(waitMs, undefined, { signal });
|
||||
if (localGate) tokenGates.delete(key);
|
||||
}
|
||||
|
||||
@@ -255,7 +244,7 @@ export function octokit(token: string) {
|
||||
const prefix = `/repos/${context.publicRepository.split("/").map(encodeURIComponent).join("/")}`.toLowerCase();
|
||||
const path = url.pathname.toLowerCase();
|
||||
const suffix = path.slice(prefix.length);
|
||||
const readable = /^(?:\/?|\/branches(?:\/[^/]+)?|\/commits(?:\/[^/]+)?|\/readme|\/pages|\/zipball\/[^/]+|\/git\/(?:trees|blobs)\/[^/]+|\/pulls\/\d+|\/issues\/\d+\/comments)$/.test(suffix);
|
||||
const readable = /^(?:\/?|\/branches(?:\/[^/]+)?|\/commits(?:\/[^/]+)?|\/contents(?:\/[^/]+)*|\/readme|\/pages|\/zipball\/[^/]+|\/git\/(?:trees|blobs)\/[^/]+|\/pulls\/\d+|\/issues\/\d+\/comments)$/.test(suffix);
|
||||
if (!readable || (options.method !== "GET" && !(options.method === "HEAD" && suffix.startsWith("/zipball/"))) || url.origin !== "https://api.github.com" ||
|
||||
(path !== prefix && !path.startsWith(prefix + "/"))) {
|
||||
throw new AnonymousError("github_app_access_required", { httpStatus: 403 });
|
||||
@@ -297,12 +286,9 @@ export async function checkToken(token: string) {
|
||||
else await oct.users.getAuthenticated();
|
||||
return true;
|
||||
} catch (err) {
|
||||
if (
|
||||
err instanceof AnonymousError &&
|
||||
err.message === "github_rate_limit_exceeded"
|
||||
) {
|
||||
throw err;
|
||||
}
|
||||
// Only a confirmed invalid credential permits fallback. Network failures,
|
||||
// permission failures and rate limits must retain their original meaning.
|
||||
if ((err as { status?: number })?.status !== 401) throw err;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -310,7 +296,12 @@ export async function checkToken(token: string) {
|
||||
const checkedRepositoryTokens = new WeakMap<Repository, string>();
|
||||
|
||||
export async function getToken(repository: Repository) {
|
||||
return measureStage("authorization", () => resolveRepositoryToken(repository));
|
||||
try {
|
||||
return await measureStage("authorization", () => resolveRepositoryToken(repository));
|
||||
} catch (error) {
|
||||
void notifyOwnerAccessProblem(repository.owner.id, error, { kind: "repository", id: String(repository.model._id) });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function resolveRepositoryToken(repository: Repository) {
|
||||
@@ -337,62 +328,18 @@ async function resolveRepositoryToken(repository: Repository) {
|
||||
credential?.persisted &&
|
||||
(!tokenAge || tokenAge < new Date(Date.now() - 1000 * 60 * 60 * 24 * 7))
|
||||
) {
|
||||
const url = `https://api.github.com/applications/${config.CLIENT_ID}/token`;
|
||||
const headers = {
|
||||
Accept: "application/vnd.github+json",
|
||||
"X-GitHub-Api-Version": "2022-11-28",
|
||||
};
|
||||
|
||||
const res = await fetch(url, {
|
||||
method: "PATCH",
|
||||
body: JSON.stringify({
|
||||
access_token: ownerAccessToken,
|
||||
}),
|
||||
credentials: "include",
|
||||
headers: {
|
||||
...headers,
|
||||
Authorization:
|
||||
"Basic " +
|
||||
Buffer.from(
|
||||
config.CLIENT_ID + ":" + config.CLIENT_SECRET
|
||||
).toString("base64"),
|
||||
},
|
||||
});
|
||||
// Only persist a refreshed token if GitHub actually returned a
|
||||
// valid one. Without this guard, a 4xx/5xx error body (revoked
|
||||
// OAuth, rate limit, transient outage) silently overwrites the
|
||||
// user's stored token with `undefined`, which then propagates as
|
||||
// `Authorization: token undefined` to every subsequent API call —
|
||||
// 401 even on public repos, and the config.GITHUB_TOKEN fallback
|
||||
// below is unreachable because the token field is no longer falsy.
|
||||
if (res.ok) {
|
||||
const resBody = (await res.json().catch(() => null)) as
|
||||
| { token?: unknown }
|
||||
| null;
|
||||
const refreshed =
|
||||
resBody && typeof resBody.token === "string" && resBody.token.length > 0
|
||||
? resBody.token
|
||||
: null;
|
||||
if (refreshed) {
|
||||
if (await replaceCredential(repository.owner.id, ownerAccessToken, refreshed)) {
|
||||
checkedRepositoryTokens.set(repository, refreshed);
|
||||
return refreshed;
|
||||
}
|
||||
return (await getCredentialToken(repository.owner.id)) || config.GITHUB_TOKEN;
|
||||
}
|
||||
const refreshed = await refreshLegacyToken(repository.owner.id, ownerAccessToken);
|
||||
if (refreshed !== null) {
|
||||
checkedRepositoryTokens.set(repository, refreshed);
|
||||
return refreshed;
|
||||
}
|
||||
logger.warn("token refresh failed; falling back", {
|
||||
code: "token_refresh_failed",
|
||||
httpStatus: res.status,
|
||||
username: repository.owner.model.username,
|
||||
});
|
||||
// fall through to the checkToken path / config.GITHUB_TOKEN
|
||||
}
|
||||
const check = await checkToken(ownerAccessToken);
|
||||
if (check) {
|
||||
checkedRepositoryTokens.set(repository, ownerAccessToken);
|
||||
return ownerAccessToken;
|
||||
}
|
||||
void notifyOwnerAccessProblem(repository.owner.id, "token_expired", { kind: "repository", id: String(repository.model._id) });
|
||||
return config.GITHUB_TOKEN;
|
||||
}
|
||||
return (await getCredentialToken(repository.owner.id, "github", {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyOwnerAccessProblem, withOwnerAccessRecovery } from "./owner-notifications";
|
||||
import { isConnected } from "../server/database";
|
||||
import { expireEmbeddedContent } from "./content-expiration";
|
||||
import { boundAppToken } from "./github-app";
|
||||
@@ -33,6 +34,17 @@ export default class PullRequest {
|
||||
}
|
||||
|
||||
async download() {
|
||||
try { await withOwnerAccessRecovery(this.owner.id,
|
||||
{ kind: "pull-request", id: String(this.model._id) }, () => this.downloadContent(),
|
||||
{ "source.repositoryFullName": this.model.source.repositoryFullName, "source.pullRequestId": this.model.source.pullRequestId,
|
||||
"githubAccess.revision": this.model.githubAccess?.revision ?? { $exists: false } }); }
|
||||
catch (error) {
|
||||
void notifyOwnerAccessProblem(this.owner.id, error, { kind: "pull-request", id: String(this.model._id) });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async downloadContent() {
|
||||
logger.info("downloading pull request", {
|
||||
pullRequestId: this._model.source.pullRequestId,
|
||||
});
|
||||
|
||||
+51
-15
@@ -1,4 +1,5 @@
|
||||
import { allowance, reserveRepository } from "./billing/quotas";
|
||||
import { notifyOwnerAccessProblem, withOwnerAccessRecovery } from "./owner-notifications";
|
||||
import AnonymizedPathModel from "./model/anonymized-path";
|
||||
import storage from "./storage";
|
||||
import { contentGenerationPrefix, contentRetirementMarker } from "./content-generation";
|
||||
@@ -8,6 +9,7 @@ import { RepositoryStatus } from "./types";
|
||||
import { Readable } from "stream";
|
||||
import * as sha1 from "crypto-js/sha1";
|
||||
import User from "./User";
|
||||
import { SubmoduleMount } from "./source/github-submodules";
|
||||
import GitHubStream from "./source/GitHubStream";
|
||||
import Zip from "./source/Zip";
|
||||
import {
|
||||
@@ -43,14 +45,14 @@ async function buildPathIndex(repoId: string, key: string, terms: string[], gene
|
||||
const task = (async () => {
|
||||
const compiled = compileTerms(terms);
|
||||
const cursor = FileModel.find({ repoId, treeGeneration: generation || { $exists: false } })
|
||||
.select("name path sha size").sort({ path: 1, name: 1 }).lean().cursor();
|
||||
.select("name path sha size upstream").sort({ path: 1, name: 1 }).lean().cursor();
|
||||
let batch: Parameters<typeof AnonymizedPathModel.bulkWrite>[0] = [];
|
||||
try {
|
||||
for await (const file of cursor) {
|
||||
const name = anonymizePathCompiled(file.name, compiled);
|
||||
const directory = anonymizePathCompiled(file.path, compiled);
|
||||
batch.push({ updateOne: { filter: { repoId, key, originalId: file._id }, update: { $set: {
|
||||
repoId, key, originalId: file._id, name: file.name, path: file.path, sha: file.sha, size: file.size,
|
||||
repoId, key, originalId: file._id, name: file.name, path: file.path, sha: file.sha, size: file.size, upstream: file.upstream,
|
||||
anonymousName: name, anonymousDirectory: directory, createdAt: new Date(),
|
||||
searchName: name.toLowerCase(), searchDirectory: directory.toLowerCase(),
|
||||
anonymousPath: anonymizePathCompiled(file.path ? `${file.path}/${file.name}` : file.name, compiled),
|
||||
@@ -202,6 +204,17 @@ export default class Repository {
|
||||
return await FileModel.find(query).exec();
|
||||
}
|
||||
|
||||
async submoduleMounts(): Promise<SubmoduleMount[]> {
|
||||
if (this.model.source.type === "Zip") return [];
|
||||
await this.ensureFileTree();
|
||||
const mounts = await FileModel.find({ repoId: this.repoId,
|
||||
treeGeneration: this.model.treeGeneration || { $exists: false }, submodule: "included",
|
||||
}).select("name path upstream").lean().exec();
|
||||
return mounts.filter(file => file.upstream).map(file => ({
|
||||
...file.upstream!, path: [file.path, file.name].filter(Boolean).join("/"),
|
||||
}));
|
||||
}
|
||||
|
||||
private async ensureFileTree(opt: { force?: boolean; progress?: (status: string) => void } = {}) {
|
||||
this.assertNotArchived();
|
||||
let hasFile = this.hasEmptyTree() || await FileModel.exists({ repoId: this.repoId, treeGeneration: this.model.treeGeneration || { $exists: false } }).exec();
|
||||
@@ -219,12 +232,22 @@ export default class Repository {
|
||||
hasFile = null;
|
||||
}
|
||||
}
|
||||
if (hasFile && this.source instanceof GitHubStream && await FileModel.exists({
|
||||
repoId: this.repoId, treeGeneration: this.model.treeGeneration || { $exists: false }, name: ".gitmodules",
|
||||
}).exec()) hasFile = null;
|
||||
if (!hasFile || opt.force) {
|
||||
const previousGeneration = this.model.treeGeneration;
|
||||
const previousPrefix = contentGenerationPrefix(`${previousGeneration || "legacy"}:${this.model.anonymizeDate?.toISOString() || ""}`);
|
||||
const streamed = this.model.source.type !== "Zip";
|
||||
const source = this.source;
|
||||
const files = await source.getFiles(opt.progress);
|
||||
const readFiles = () => source.getFiles(opt.progress);
|
||||
const files = await (streamed ? withOwnerAccessRecovery(this.owner.id,
|
||||
{ kind: "repository", id: String(this.model._id) }, readFiles,
|
||||
{ "source.repositoryName": this.model.source.repositoryName, "source.commit": this.model.source.commit,
|
||||
"githubAccess.revision": this.model.githubAccess?.revision ?? { $exists: false } }) : readFiles()).catch(error => {
|
||||
void notifyOwnerAccessProblem(this.owner.id, error, { kind: "repository", id: String(this.model._id) });
|
||||
throw error;
|
||||
});
|
||||
if (this.model.billingManaged) await reserveRepository(String(this.model.owner), this.repoId, files.reduce((sum, file) => sum + (file.size || 0), 0), this.model.conference);
|
||||
const sourceWithTruncation = source as unknown as { truncatedFolderList?: string[] };
|
||||
const truncatedFolders = Array.isArray(sourceWithTruncation.truncatedFolderList)
|
||||
@@ -550,8 +573,8 @@ export default class Repository {
|
||||
async isReady() {
|
||||
if (this.status !== RepositoryStatus.READY) return false;
|
||||
if (!this.hasEmptyTree() && !(await FileModel.exists({ repoId: this.repoId, treeGeneration: this.model.treeGeneration || { $exists: false } }).exec())) {
|
||||
this.model.status = RepositoryStatus.PREPARING;
|
||||
await this.updateIfNeeded({ force: true });
|
||||
// Missing-tree recovery competes with owner refreshes for the same job.
|
||||
await this.refresh();
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
@@ -626,6 +649,14 @@ export default class Repository {
|
||||
|
||||
/** Update the repository if a new commit exists. */
|
||||
async updateIfNeeded(opt?: { force: boolean }): Promise<void> {
|
||||
try { await this.refreshIfNeeded(opt); }
|
||||
catch (error) {
|
||||
void notifyOwnerAccessProblem(this.owner.id, error, { kind: "repository", id: String(this.model._id) });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async refreshIfNeeded(opt?: { force: boolean }): Promise<void> {
|
||||
this.assertNotArchived();
|
||||
if (
|
||||
this._model.options.expirationMode !== "never" &&
|
||||
@@ -677,12 +708,14 @@ export default class Repository {
|
||||
this.checkRefreshWrite(result);
|
||||
}
|
||||
}
|
||||
const branches = await ghRepo.branches({
|
||||
force: true,
|
||||
accessToken: token,
|
||||
});
|
||||
const branchName =
|
||||
this.model.source.branch || ghRepo.model.defaultBranch;
|
||||
const branchName = this.model.source.branch || ghRepo.model.defaultBranch;
|
||||
const branches = await withOwnerAccessRecovery(this.owner.id,
|
||||
{ kind: "repository", id: String(this.model._id) },
|
||||
() => ghRepo.branches({ force: true, accessToken: token }),
|
||||
{ "source.repositoryName": this.model.source.repositoryName,
|
||||
"githubAccess.revision": this.model.githubAccess?.revision ?? { $exists: false } },
|
||||
result => !opt?.force && this.status === RepositoryStatus.READY
|
||||
&& result.some(branch => branch.name === branchName && branch.commit === this.model.source.commit));
|
||||
const newCommit = branches.filter((f) => f.name == branchName)[0]
|
||||
?.commit;
|
||||
|
||||
@@ -703,7 +736,8 @@ export default class Repository {
|
||||
}
|
||||
if (
|
||||
this.model.source.commit == newCommit &&
|
||||
this.status == RepositoryStatus.READY
|
||||
this.status == RepositoryStatus.READY &&
|
||||
!opt?.force
|
||||
) {
|
||||
logger.info("up to date", { repoId: this._model.repoId });
|
||||
return;
|
||||
@@ -723,9 +757,11 @@ export default class Repository {
|
||||
}
|
||||
}
|
||||
this._model.source.commit = newCommit;
|
||||
const commitInfo = await ghRepo.getCommitInfo(newCommit, {
|
||||
accessToken: token,
|
||||
});
|
||||
const commitInfo = await withOwnerAccessRecovery(this.owner.id,
|
||||
{ kind: "repository", id: String(this.model._id) },
|
||||
() => ghRepo.getCommitInfo(newCommit, { accessToken: token }),
|
||||
{ "source.repositoryName": this.model.source.repositoryName,
|
||||
"githubAccess.revision": this.model.githubAccess?.revision ?? { $exists: false } });
|
||||
if (
|
||||
commitInfo.commit?.author?.date ||
|
||||
commitInfo.commit?.committer?.date
|
||||
|
||||
@@ -531,11 +531,12 @@ export class ContentAnonimizer {
|
||||
const replaced = replaceTerm(content, c, asciiContent);
|
||||
if (replaced !== content) this.wasAnonymized = true;
|
||||
content = replaced;
|
||||
// Masks and changed boundaries can introduce later matches. Re-index
|
||||
// only after a real change, never once per absent term.
|
||||
// Masks and changed boundaries can introduce later matches. Invalidate
|
||||
// the original-text index and check remaining rules directly; rebuilding
|
||||
// the index after every change costs more than those presence checks.
|
||||
if (content !== previous) {
|
||||
asciiContent = asciiContent && /^\p{ASCII}*$/u.test(c.mask);
|
||||
candidates = this.termPresence?.candidates(content, index);
|
||||
candidates = undefined;
|
||||
}
|
||||
}
|
||||
return content;
|
||||
|
||||
@@ -1,5 +1,17 @@
|
||||
import { Archiver, EntryData } from "archiver";
|
||||
import { Archiver, EntryData, ZipEntryData } from "archiver";
|
||||
import { extname } from "path";
|
||||
import { Readable } from "stream";
|
||||
|
||||
const compressedExtensions = new Set([
|
||||
".png", ".jpg", ".jpeg", ".gif", ".webp", ".avif", ".heif", ".heic",
|
||||
".mp4", ".m4v", ".mov", ".webm", ".mp3", ".m4a", ".aac", ".ogg", ".flac",
|
||||
".zip", ".gz", ".tgz", ".bz2", ".xz", ".7z", ".rar", ".zst", ".woff", ".woff2",
|
||||
]);
|
||||
|
||||
/** Store already compressed assets; every entry still passes through anonymization. */
|
||||
export function zipEntryOptions(name: string, originalName = name): ZipEntryData {
|
||||
return { name, store: compressedExtensions.has(extname(originalName).toLowerCase()) };
|
||||
}
|
||||
/** Allow only one active entry; compression and the response provide backpressure. */
|
||||
export function appendArchiveEntry(archive: Archiver, source: Readable, options: EntryData): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
/** A bounded cache for short-lived failures. Keys must include access and content versions. */
|
||||
export class ExpiringMap<T> {
|
||||
private entries = new Map<string, { value: T; expires: number }>();
|
||||
constructor(private capacity: number, private now = () => Date.now()) {}
|
||||
get(key: string): T | undefined {
|
||||
const entry = this.entries.get(key);
|
||||
if (!entry) return undefined;
|
||||
if (entry.expires <= this.now()) { this.entries.delete(key); return undefined; }
|
||||
return entry.value;
|
||||
}
|
||||
set(key: string, value: T, ttl: number) {
|
||||
for (const [key, entry] of this.entries) if (entry.expires <= this.now()) this.entries.delete(key);
|
||||
this.entries.delete(key);
|
||||
if (this.entries.size >= this.capacity) this.entries.delete(this.entries.keys().next().value!);
|
||||
this.entries.set(key, { value, expires: this.now() + ttl });
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
// Octokit RequestError shape (subset we care about for rate-limit detection).
|
||||
export interface OctokitRequestErrorLike {
|
||||
status?: number;
|
||||
httpStatus?: number;
|
||||
message?: string;
|
||||
response?: {
|
||||
statusCode?: number;
|
||||
headers?: Record<string, string | undefined>;
|
||||
};
|
||||
}
|
||||
|
||||
/** Detect GitHub primary and secondary rate limits in API and stream errors. */
|
||||
export function isGitHubRateLimitError(err: unknown): boolean {
|
||||
if (!err || typeof err !== "object") return false;
|
||||
const e = err as OctokitRequestErrorLike;
|
||||
const msg = (e.message ?? "").toLowerCase();
|
||||
// Primary limits return 403 with "x-ratelimit-remaining: 0"; secondary
|
||||
// limits return 403 (sometimes 429) with "secondary rate limit" in the
|
||||
// body. Match on either signal so we catch both.
|
||||
const status = e.status ?? e.response?.statusCode ?? e.httpStatus ?? 0;
|
||||
if (status !== 403 && status !== 429) return false;
|
||||
if (msg.includes("rate limit") || msg.includes("abuse")) return true;
|
||||
const remaining = e.response?.headers?.["x-ratelimit-remaining"];
|
||||
return remaining === "0";
|
||||
}
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
import { createHash } from "crypto";
|
||||
import config from "../config";
|
||||
import { getCredentialToken, replaceCredential } from "./credentials";
|
||||
import { ExpiringMap } from "./expiring-map";
|
||||
import { createLogger } from "./logger";
|
||||
|
||||
const logger = createLogger("github");
|
||||
const cooldowns = new ExpiringMap<boolean>(1024);
|
||||
const pending = new Map<string, Promise<string | null>>();
|
||||
|
||||
/** Only legacy OAuth tokens belong to the OAuth application's reset endpoint. */
|
||||
export async function refreshLegacyToken(ownerId: string, previous: string): Promise<string | null> {
|
||||
if (!config.GITHUB_OAUTH_ENABLED || !config.CLIENT_ID || !config.CLIENT_SECRET ||
|
||||
config.CLIENT_ID === "CLIENT_ID" || config.CLIENT_SECRET === "CLIENT_SECRET" ||
|
||||
/^(gh[psur]_|github_pat_)/.test(previous)) return null;
|
||||
const key = createHash("sha256").update(JSON.stringify([ownerId, previous, config.CLIENT_ID, config.CLIENT_SECRET])).digest("hex");
|
||||
if (cooldowns.get(key)) return null;
|
||||
const running = pending.get(key);
|
||||
if (running) return running;
|
||||
if (pending.size >= 1024) return null;
|
||||
const refresh = async () => {
|
||||
let status = 502;
|
||||
try {
|
||||
const res = await fetch(`https://api.github.com/applications/${encodeURIComponent(config.CLIENT_ID)}/token`, {
|
||||
method: "PATCH", body: JSON.stringify({ access_token: previous }),
|
||||
signal: AbortSignal.timeout(5000),
|
||||
headers: { Accept: "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28",
|
||||
Authorization: "Basic " + Buffer.from(config.CLIENT_ID + ":" + config.CLIENT_SECRET).toString("base64") },
|
||||
});
|
||||
status = res.status;
|
||||
if (res.ok) {
|
||||
const body = await res.json().catch(() => null) as { token?: unknown } | null;
|
||||
if (typeof body?.token === "string" && body.token.length) {
|
||||
if (await replaceCredential(ownerId, previous, body.token)) return body.token;
|
||||
return (await getCredentialToken(ownerId)) || config.GITHUB_TOKEN;
|
||||
}
|
||||
} else {
|
||||
// Release the response connection even when only the status matters.
|
||||
await res.body?.cancel();
|
||||
}
|
||||
} catch {
|
||||
// The existing token can still work during a transient reset-endpoint failure.
|
||||
}
|
||||
cooldowns.set(key, true, status === 401 || status === 404 ? 30 * 60_000 : 60_000);
|
||||
logger.warn("token refresh failed; backing off", { code: "token_refresh_failed", httpStatus: status });
|
||||
return null;
|
||||
};
|
||||
const work = refresh();
|
||||
pending.set(key, work);
|
||||
try { return await work; }
|
||||
finally { if (pending.get(key) === work) pending.delete(key); }
|
||||
}
|
||||
+4
-1
@@ -320,7 +320,10 @@ export function serializeError(err: unknown): Record<string, unknown> {
|
||||
|
||||
// AnonymousError carries an httpStatus and an inner cause.
|
||||
if (typeof e.httpStatus === "number") out.httpStatus = e.httpStatus;
|
||||
if (e.code !== undefined && e.code !== e.message) out.code = e.code;
|
||||
// Octokit's deprecated `code` getter emits a stack on every read. Only
|
||||
// inspect data properties so logging never invokes that getter.
|
||||
const code = Object.getOwnPropertyDescriptor(e, "code")?.value;
|
||||
if (code !== undefined && code !== e.message) out.code = code;
|
||||
if (typeof e.url === "function") {
|
||||
try {
|
||||
const u = e.url();
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import { UpstreamFileSchema } from "./files/files.schema";
|
||||
import { IFile } from "./files/files.types";
|
||||
import { model, Schema } from "mongoose";
|
||||
const schema = new Schema({ repoId: { type: String, required: true }, key: { type: String, required: true }, originalId: Schema.Types.ObjectId,
|
||||
name: { type: String, required: true }, path: { type: String, default: "", required: true }, sha: String, size: Number,
|
||||
upstream: { type: UpstreamFileSchema, default: undefined },
|
||||
anonymousName: String, anonymousDirectory: String, anonymousPath: String,
|
||||
searchName: String, searchDirectory: String,
|
||||
createdAt: { type: Date, default: Date.now } });
|
||||
|
||||
@@ -1,6 +1,11 @@
|
||||
import { ownerAlertLifecycle } from "../owner-alert-lifecycle.schema";
|
||||
import { Schema } from "mongoose";
|
||||
|
||||
const AnonymizedGistSchema = new Schema({
|
||||
accessAlertGeneration: { type: String, select: false },
|
||||
accessAlertClaimedAt: { type: Date, select: false },
|
||||
accessAlertAttemptId: { type: String, select: false },
|
||||
accessAlertClaimRevision: { type: Number, select: false },
|
||||
gistId: {
|
||||
type: String,
|
||||
index: { unique: true },
|
||||
@@ -70,4 +75,6 @@ const AnonymizedGistSchema = new Schema({
|
||||
|
||||
AnonymizedGistSchema.index({ status: 1, "options.expirationDate": 1 });
|
||||
|
||||
AnonymizedGistSchema.plugin(ownerAlertLifecycle);
|
||||
|
||||
export default AnonymizedGistSchema;
|
||||
@@ -1,7 +1,12 @@
|
||||
import { ownerAlertLifecycle } from "../owner-alert-lifecycle.schema";
|
||||
import { repositoryAccessSchema } from "../repository-access.schema";
|
||||
import { Schema } from "mongoose";
|
||||
|
||||
const AnonymizedPullRequestSchema = new Schema({
|
||||
accessAlertGeneration: { type: String, select: false },
|
||||
accessAlertClaimedAt: { type: Date, select: false },
|
||||
accessAlertAttemptId: { type: String, select: false },
|
||||
accessAlertClaimRevision: { type: Number, select: false },
|
||||
pullRequestId: {
|
||||
type: String,
|
||||
index: { unique: true },
|
||||
@@ -67,4 +72,6 @@ const AnonymizedPullRequestSchema = new Schema({
|
||||
|
||||
AnonymizedPullRequestSchema.index({ status: 1, "options.expirationDate": 1 });
|
||||
|
||||
AnonymizedPullRequestSchema.plugin(ownerAlertLifecycle);
|
||||
|
||||
export default AnonymizedPullRequestSchema;
|
||||
@@ -1,9 +1,15 @@
|
||||
import { ownerAlertLifecycle } from "../owner-alert-lifecycle.schema";
|
||||
import { repositoryAccessSchema } from "../repository-access.schema";
|
||||
import { Schema } from "mongoose";
|
||||
|
||||
const AnonymizedRepositorySchema = new Schema({
|
||||
billingManaged: Boolean,
|
||||
billingNeverExpiresUntil: Date,
|
||||
accessAlertGeneration: { type: String, select: false },
|
||||
accessAlertClaimedAt: { type: Date, select: false },
|
||||
accessAlertAttemptId: { type: String, select: false },
|
||||
accessAlertClaimRevision: { type: Number, select: false },
|
||||
reviewConsentRevision: { type: Number, select: false },
|
||||
repoId: {
|
||||
type: String,
|
||||
index: { unique: true, collation: { locale: "en", strength: 2 } },
|
||||
@@ -120,4 +126,6 @@ AnonymizedRepositorySchema.index({
|
||||
"options.expirationDate": 1,
|
||||
});
|
||||
|
||||
AnonymizedRepositorySchema.plugin(ownerAlertLifecycle);
|
||||
|
||||
export default AnonymizedRepositorySchema;
|
||||
@@ -1,5 +1,7 @@
|
||||
import { Schema } from "mongoose";
|
||||
|
||||
export const UpstreamFileSchema = new Schema({ organization: String, repoName: String, commit: String, path: String }, { _id: false });
|
||||
|
||||
const FileSchema = new Schema({
|
||||
name: { type: String, index: true },
|
||||
path: { type: String, index: true },
|
||||
@@ -7,6 +9,8 @@ const FileSchema = new Schema({
|
||||
treeGeneration: String,
|
||||
treeStaged: Boolean,
|
||||
metadataPending: Boolean,
|
||||
upstream: { type: UpstreamFileSchema, default: undefined },
|
||||
submodule: { type: String, enum: ["included", "unsupported", "unavailable", "limit", "nested"] },
|
||||
sha: {
|
||||
type: String,
|
||||
},
|
||||
|
||||
@@ -1,6 +1,16 @@
|
||||
import { Document, Model } from "mongoose";
|
||||
|
||||
export interface UpstreamFile {
|
||||
organization: string;
|
||||
repoName: string;
|
||||
commit: string;
|
||||
path: string;
|
||||
}
|
||||
|
||||
export interface IFile {
|
||||
/** Internal source metadata, never returned by the public tree API. */
|
||||
upstream?: UpstreamFile;
|
||||
submodule?: "included" | "unsupported" | "unavailable" | "limit" | "nested";
|
||||
name: string;
|
||||
path: string;
|
||||
repoId: string;
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
import { randomUUID } from "crypto";
|
||||
import { Document, Query, Schema, Types, mongo } from "mongoose";
|
||||
|
||||
const inactive = new Set(["archived", "expired", "expiring", "removed", "removing"]);
|
||||
type Owner = { _id: Types.ObjectId; accessAlertResetRevision?: number; accessAlertDeactivations?: string[] };
|
||||
type Fence = { users: mongo.Collection<Owner>; owners: Types.ObjectId[]; token: string };
|
||||
|
||||
/** Lifecycle mutations and email dispatch compete on the owner document. */
|
||||
export function ownerAlertLifecycle(schema: Schema): void {
|
||||
const fences = new WeakMap<object, Fence>();
|
||||
const begin = async (operation: object, db: { collection(name: string): unknown }, owners: Types.ObjectId[]) => {
|
||||
if (!owners.length) return;
|
||||
const users = db.collection("users") as mongo.Collection<Owner>;
|
||||
const token = randomUUID();
|
||||
// Do not expire this reservation: an interrupted database mutation may still
|
||||
// commit. Offline reconciliation can clear it after all writers have stopped.
|
||||
await users.updateMany({ _id: { $in: owners } }, {
|
||||
$inc: { accessAlertResetRevision: 1 }, $addToSet: { accessAlertDeactivations: token },
|
||||
});
|
||||
fences.set(operation, { users, owners, token });
|
||||
};
|
||||
const finish = async (operation: object) => {
|
||||
const fence = fences.get(operation);
|
||||
if (!fence) return;
|
||||
await fence.users.updateMany({ _id: { $in: fence.owners } }, { $pull: { accessAlertDeactivations: fence.token } });
|
||||
fences.delete(operation);
|
||||
};
|
||||
schema.pre(["updateOne", "updateMany", "findOneAndUpdate"], async function (this: Query<unknown, unknown>) {
|
||||
const update = this.getUpdate();
|
||||
if (!update) return;
|
||||
const stages = (Array.isArray(update) ? update : [update]) as { $set?: { status?: unknown }; status?: unknown }[];
|
||||
if (!stages.some(stage => { const status = stage.$set?.status ?? stage.status; return typeof status === "string" && inactive.has(status); })) return;
|
||||
const filter = this.getFilter();
|
||||
const lookup = this.model.distinct("owner", filter);
|
||||
const collation = this.getOptions().collation;
|
||||
if (collation) lookup.collation(collation);
|
||||
const owners = (await lookup.exec()).filter(id => id && Types.ObjectId.isValid(id));
|
||||
await begin(this, this.model.db, owners);
|
||||
// Preserve the owner snapshot: reassignment must not move this mutation to
|
||||
// an account whose dispatch fence was not advanced.
|
||||
this.setQuery({ $and: [filter, { $or: [{ owner: { $in: owners } }, { owner: null }] }] });
|
||||
});
|
||||
schema.post(["updateOne", "updateMany", "findOneAndUpdate"], async function () { await finish(this); });
|
||||
schema.pre("save", async function (this: Document & { status?: string; owner?: Types.ObjectId; $where?: Record<string, unknown> }) {
|
||||
if (!inactive.has(this.status || "") || !this.isModified("status")) return;
|
||||
const current = this.isNew ? null : await this.collection.findOne({ _id: this._id }, { projection: { owner: 1 } });
|
||||
const owners = [current?.owner, this.owner].filter((id): id is Types.ObjectId => Boolean(id && Types.ObjectId.isValid(id)));
|
||||
await begin(this, this.db, owners);
|
||||
if (current) this.$where = { $and: [this.$where || {}, { owner: current.owner ?? null }] };
|
||||
});
|
||||
schema.post("save", async function () { await finish(this); });
|
||||
// Failed or ambiguous mutations retain their reservation. Releasing one here
|
||||
// could permit dispatch while the resource write subsequently completes.
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
import { Schema } from "mongoose";
|
||||
|
||||
const UserSchema = new Schema({
|
||||
reviewConsentRevision: { type: Number, select: false },
|
||||
accessTokens: {
|
||||
github: { type: String, select: false },
|
||||
},
|
||||
@@ -14,6 +15,14 @@ const UserSchema = new Schema({
|
||||
type: String,
|
||||
index: { unique: true },
|
||||
},
|
||||
notificationEmail: { type: String },
|
||||
emailPromptNever: { type: Boolean, default: false },
|
||||
accessAlertAfter: { type: Date, select: false },
|
||||
accessAlertResetRevision: { type: Number, select: false },
|
||||
accessAlertResetUntil: { type: Date, select: false },
|
||||
accessAlertFenceRevision: { type: Number, select: false },
|
||||
accessAlertDeactivations: { type: [String], default: undefined, select: false },
|
||||
accessAlertDispatches: { type: Schema.Types.Mixed, select: false },
|
||||
emails: [
|
||||
{
|
||||
email: { type: String },
|
||||
@@ -27,6 +36,9 @@ const UserSchema = new Schema({
|
||||
name: { type: String },
|
||||
createdAt: { type: Date, default: Date.now },
|
||||
lastUsedAt: { type: Date },
|
||||
scope: { type: String, enum: ["repository-maintenance"] },
|
||||
repoIds: { type: [String], default: undefined },
|
||||
expiresAt: { type: Date },
|
||||
},
|
||||
],
|
||||
photo: String,
|
||||
|
||||
@@ -19,7 +19,12 @@ export interface IUser {
|
||||
name?: string;
|
||||
createdAt?: Date;
|
||||
lastUsedAt?: Date;
|
||||
scope?: "repository-maintenance";
|
||||
repoIds?: string[];
|
||||
expiresAt?: Date;
|
||||
}[];
|
||||
notificationEmail?: string;
|
||||
emailPromptNever?: boolean;
|
||||
emails: {
|
||||
email: string;
|
||||
default: boolean;
|
||||
|
||||
@@ -0,0 +1,623 @@
|
||||
import { createHash, randomUUID } from "crypto";
|
||||
import { mongo, Types } from "mongoose";
|
||||
import { isGitHubRateLimitError } from "./github-rate-limit";
|
||||
import config from "../config";
|
||||
import UserModel from "./model/users/users.model";
|
||||
import { createLogger } from "./logger";
|
||||
|
||||
const logger = createLogger("owner-notifications");
|
||||
const resourceCollections = {
|
||||
repository: "anonymizedrepositories",
|
||||
"pull-request": "anonymizedpullrequests",
|
||||
gist: "anonymizedgists",
|
||||
} as const;
|
||||
export interface AlertResource { kind: keyof typeof resourceCollections; id: string; }
|
||||
|
||||
// Only explicit owner actions call this; background fetches never reset alerts.
|
||||
export async function resetOwnerAccessAlerts(ownerId: string, resource?: AlertResource): Promise<void> {
|
||||
if (UserModel.db.readyState !== 1 || !Types.ObjectId.isValid(ownerId)) return;
|
||||
try {
|
||||
const users = UserModel.db.collection("users");
|
||||
const owner = new Types.ObjectId(ownerId);
|
||||
const reset = await users.findOneAndUpdate({ _id: owner }, { $inc: { accessAlertResetRevision: 1 },
|
||||
$set: { accessAlertResetUntil: new Date(Date.now() + 60000) } }, { returnDocument: "after" });
|
||||
if (!reset.value) return;
|
||||
const revision = reset.value.accessAlertResetRevision;
|
||||
const heartbeat = setInterval(() => {
|
||||
void users.updateOne({ _id: owner, accessAlertResetRevision: revision, accessAlertResetUntil: { $exists: true } },
|
||||
{ $set: { accessAlertResetUntil: new Date(Date.now() + 60000) } }).catch(() => {});
|
||||
}, 10000);
|
||||
try {
|
||||
const filter = { owner: new Types.ObjectId(ownerId),
|
||||
...(resource ? { _id: new Types.ObjectId(resource.id) } : {}) };
|
||||
const claim = { owner: new Types.ObjectId(ownerId),
|
||||
...(resource ? { resourceKind: resource.kind, resourceId: new Types.ObjectId(resource.id) } : {}) };
|
||||
await alertCollection().updateMany({ ...claim, legacyClaims: { $exists: false } }, { $inc: { resetRevision: 1 } });
|
||||
await alertCollection().updateMany({ legacyClaims: { $elemMatch: claim } }, [{ $set: {
|
||||
resetRevision: { $add: [{ $ifNull: ["$resetRevision", 0] }, 1] },
|
||||
legacyClaims: { $map: { input: "$legacyClaims", as: "claim", in: { $cond: [
|
||||
{ $and: Object.entries(claim).map(([key, value]) => ({ $eq: [`$$claim.${key}`, value] })) },
|
||||
{ $mergeObjects: ["$$claim", { recovered: false }] }, "$$claim",
|
||||
] } } },
|
||||
state: { $cond: [{ $eq: ["$state", "recovered"] }, "legacy", "$state"] },
|
||||
} }]);
|
||||
for (const collection of resource ? [resourceCollections[resource.kind]] : Object.values(resourceCollections)) {
|
||||
await UserModel.db.collection(collection).updateMany(filter, { $unset: { accessAlertClaimedAt: "", accessAlertAttemptId: "" }, $set: { accessAlertGeneration: randomUUID() } });
|
||||
}
|
||||
} finally {
|
||||
clearInterval(heartbeat);
|
||||
const finished = await users.updateOne({ _id: owner, accessAlertResetRevision: revision }, { $unset: { accessAlertResetUntil: "" } });
|
||||
if (finished.matchedCount) {
|
||||
for await (const candidate of alertCollection().find({ deferredRecoveries: { $elemMatch: { owner } } })) {
|
||||
await applyDeferredRecoveries(candidate._id, candidate.attemptId);
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch { logger.warn("owner access alert reset unavailable"); }
|
||||
}
|
||||
export const RECIPIENT_ALERTS = "owner_access_alert_recipients";
|
||||
type LegacyClaim = { owner: Types.ObjectId; resourceKind: AlertResource["kind"]; resourceId: Types.ObjectId; generation: string | null; recovered?: boolean };
|
||||
type DeferredRecovery = { owner: Types.ObjectId; resourceKind: AlertResource["kind"]; resourceId: Types.ObjectId;
|
||||
generation: string | null; resetRevision: number | null; ownerRevision: number | null; bindingFields: string[]; bindingHash: string };
|
||||
type AlertRecord = { _id: string; state?: string; lease?: string; leaseUntil?: Date;
|
||||
owner?: Types.ObjectId; resourceKind?: string; resourceId?: Types.ObjectId;
|
||||
generation?: string | null; attemptId?: string; claimedAt?: Date; completedAt?: Date;
|
||||
recipientKey?: string; resetRevision?: number; httpStatus?: number; startedAt?: Date;
|
||||
progress?: Record<string, { claimedAt: Date; id: Types.ObjectId }>; legacyClaims?: LegacyClaim[];
|
||||
resourceRevision?: number; deferredRecoveries?: DeferredRecovery[]; dispatchPrepared?: boolean;
|
||||
deliveryProgress?: { completedAt: Date; id: string }; failureCode?: string };
|
||||
const alertCollection = (name = RECIPIENT_ALERTS) => UserModel.db.collection(name) as unknown as mongo.Collection<AlertRecord>;
|
||||
const MIGRATION = "recipient-alerts-v2";
|
||||
const PENDING_LEASE_MS = 60000;
|
||||
const expiredPending = () => ({ state: "pending", claimedAt: { $lte: new Date(Date.now() - PENDING_LEASE_MS) } });
|
||||
const expiredPreparation = () => ({ state: { $in: ["pending", "sending"] }, claimedAt: { $lte: new Date(Date.now() - PENDING_LEASE_MS) } });
|
||||
const revisionFilter = (revision: number) => revision === 0 ? { $in: [0, null] } : revision;
|
||||
const generationFilter = (generation: unknown) => generation ?? { $exists: false };
|
||||
export const recipientAlertKey = (email: string) => createHash("sha256").update(email.trim().toLowerCase()).digest("hex");
|
||||
const bindingHash = (row: Record<string, unknown>, fields: string[]) => createHash("sha256").update(JSON.stringify(fields.map(field => {
|
||||
const value = field.split(".").reduce<unknown>((value, key) => value && typeof value === "object" ? (value as Record<string, unknown>)[key] : undefined, row);
|
||||
return [field, value !== undefined, value];
|
||||
}))).digest("hex");
|
||||
|
||||
async function prepareOwnerDispatch(record: AlertRecord, revision: number | null): Promise<boolean> {
|
||||
const path = `accessAlertDispatches.${record.attemptId}`;
|
||||
const prepared = await UserModel.db.collection("users").updateOne({ _id: record.owner,
|
||||
accessAlertResetRevision: generationFilter(revision),
|
||||
"accessAlertDeactivations.0": { $exists: false },
|
||||
$or: [{ accessAlertResetUntil: { $exists: false } }, { accessAlertResetUntil: { $lte: new Date() } }],
|
||||
}, [{ $set: { accessAlertDispatches: { $arrayToObject: { $filter: {
|
||||
input: { $objectToArray: { $ifNull: ["$accessAlertDispatches", {}] } }, as: "dispatch",
|
||||
cond: { $or: [{ $eq: ["$$dispatch.v.state", "committed"] },
|
||||
{ $gt: ["$$dispatch.v.claimedAt", new Date(Date.now() - PENDING_LEASE_MS)] }] },
|
||||
} } } } }, { $set: { [path]: { state: "prepared", claimedAt: record.claimedAt } } }]);
|
||||
return Boolean(prepared.matchedCount);
|
||||
}
|
||||
|
||||
async function clearOwnerDispatch(record: AlertRecord, state: string): Promise<void> {
|
||||
if (!record.dispatchPrepared || !record.owner || !record.attemptId) return;
|
||||
const path = `accessAlertDispatches.${record.attemptId}`;
|
||||
await UserModel.db.collection("users").updateOne({ _id: record.owner, [`${path}.state`]: state }, { $unset: { [path]: "" } });
|
||||
}
|
||||
|
||||
async function commitOwnerRevision(owner: Types.ObjectId, revision: number | null | undefined, record?: AlertRecord): Promise<boolean> {
|
||||
const path = `accessAlertDispatches.${record?.attemptId}`;
|
||||
const committed = await UserModel.db.collection("users").updateOne({ _id: owner,
|
||||
accessAlertResetRevision: generationFilter(revision),
|
||||
"accessAlertDeactivations.0": { $exists: false },
|
||||
...(record ? { [`${path}.state`]: "prepared" } : {}),
|
||||
$or: [{ accessAlertResetUntil: { $exists: false } }, { accessAlertResetUntil: { $lte: new Date() } }],
|
||||
}, record ? { $set: { [`${path}.state`]: "committed" } } : { $inc: { accessAlertFenceRevision: 1 } });
|
||||
return Boolean(committed.matchedCount);
|
||||
}
|
||||
|
||||
async function releaseUnsentAlert(record: AlertRecord, expected: mongo.Filter<AlertRecord>): Promise<boolean> {
|
||||
if (!record.attemptId) return false;
|
||||
const alerts = alertCollection();
|
||||
if (record.dispatchPrepared && record.owner) {
|
||||
const users = UserModel.db.collection("users");
|
||||
const path = `accessAlertDispatches.${record.attemptId}`;
|
||||
// This CAS competes with the final dispatch commitment on the owner document.
|
||||
const cancelled = await users.updateOne({ _id: record.owner, [`${path}.state`]: "prepared" },
|
||||
{ $set: { [`${path}.state`]: "cancelled" } });
|
||||
if (!cancelled.matchedCount) {
|
||||
const owner = await users.findOne({ _id: record.owner }, { projection: { [path]: 1 } });
|
||||
if (!owner || owner.accessAlertDispatches?.[record.attemptId]?.state === "committed") return false;
|
||||
}
|
||||
}
|
||||
// A durable cancellation prevents a paused sender from starting dispatch.
|
||||
const cancelled = await alerts.updateOne({ _id: record._id, attemptId: record.attemptId, ...expected }, { $set: { state: "cancelled" } });
|
||||
if (!cancelled.matchedCount) return false;
|
||||
if (record.owner && record.resourceKind && record.resourceKind in resourceCollections && record.resourceId) {
|
||||
const resource = UserModel.db.collection(resourceCollections[record.resourceKind as AlertResource["kind"]]);
|
||||
const revision = record.resourceRevision ?? 0;
|
||||
const matching = {
|
||||
_id: record.resourceId, owner: record.owner, accessAlertGeneration: generationFilter(record.generation),
|
||||
$or: [{ accessAlertClaimRevision: revisionFilter(revision) },
|
||||
{ accessAlertClaimRevision: revision + 1, accessAlertAttemptId: record.attemptId }],
|
||||
};
|
||||
await retryAlertWrite(() => resource.updateOne(matching, [{ $set: {
|
||||
accessAlertClaimRevision: { $add: [{ $ifNull: ["$accessAlertClaimRevision", 0] }, 1] },
|
||||
accessAlertClaimedAt: { $cond: [{ $eq: ["$accessAlertAttemptId", record.attemptId] }, "$$REMOVE", "$accessAlertClaimedAt"] },
|
||||
accessAlertAttemptId: { $cond: [{ $eq: ["$accessAlertAttemptId", record.attemptId] }, "$$REMOVE", "$accessAlertAttemptId"] },
|
||||
} }]));
|
||||
}
|
||||
const removed = await alerts.deleteOne({ _id: record._id, attemptId: record.attemptId, state: "cancelled" });
|
||||
if (removed.deletedCount) await clearOwnerDispatch(record, "cancelled");
|
||||
return Boolean(removed.deletedCount);
|
||||
}
|
||||
|
||||
async function reconcileUnsentAlerts(writersStopped = false): Promise<void> {
|
||||
const pending = writersStopped ? { state: "pending" } : expiredPending();
|
||||
const preparing = { ...(writersStopped ? { state: "sending" } : { ...expiredPreparation(), state: "sending" }), dispatchPrepared: true };
|
||||
for await (const record of alertCollection().find({ $or: [pending, preparing, { state: "cancelled" }] })) {
|
||||
await releaseUnsentAlert(record, record.state === "cancelled" ? { state: "cancelled" }
|
||||
: writersStopped ? { state: record.state } : expiredPreparation());
|
||||
}
|
||||
}
|
||||
|
||||
/** Call only after all application writers have stopped. */
|
||||
export async function prepareLegacyOwnerAlertMigration(): Promise<void> {
|
||||
await reconcileUnsentAlerts(true);
|
||||
// Interrupted lifecycle mutations cannot resume once every writer is stopped.
|
||||
await UserModel.db.collection("users").updateMany({ accessAlertDeactivations: { $exists: true } },
|
||||
{ $unset: { accessAlertDeactivations: "" }, $inc: { accessAlertResetRevision: 1 } });
|
||||
await alertCollection().updateOne({ _id: MIGRATION }, { $set: { state: "new" },
|
||||
$unset: { progress: "", deliveryProgress: "", lease: "", leaseUntil: "" } }, { upsert: true });
|
||||
}
|
||||
|
||||
async function reconcileCompletedSend(record: AlertRecord): Promise<void> {
|
||||
if (record.state !== "sending" || !record.attemptId) return;
|
||||
const completion = await alertCollection("owner_access_alert_events").findOne({ _id: `${record.attemptId}/completion` });
|
||||
if (!completion || !["accepted", "failed"].includes(completion.state || "")) return;
|
||||
await alertCollection().updateOne({ _id: record._id, attemptId: record.attemptId, state: "sending" },
|
||||
{ $set: { state: completion.state, completedAt: completion.completedAt, httpStatus: completion.httpStatus } });
|
||||
await clearOwnerDispatch(record, "committed");
|
||||
await applyDeferredRecoveries(record._id, record.attemptId);
|
||||
}
|
||||
|
||||
async function reconcileCompletedSends(): Promise<void> {
|
||||
for await (const record of alertCollection().find({ state: "sending" })) await reconcileCompletedSend(record);
|
||||
for await (const record of alertCollection().find({ state: { $in: ["accepted", "failed", "legacy"] },
|
||||
$or: [{ deferredRecoveries: { $exists: true } }, { dispatchPrepared: true }] })) {
|
||||
await clearOwnerDispatch(record, "committed");
|
||||
await applyDeferredRecoveries(record._id, record.attemptId);
|
||||
}
|
||||
}
|
||||
|
||||
async function retryAlertWrite(work: () => Promise<unknown>): Promise<void> {
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
try { await work(); return; }
|
||||
catch (error) {
|
||||
if (attempt === 2) throw error;
|
||||
await new Promise(resolve => setTimeout(resolve, 50 * (attempt + 1)));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Seed the recipient guard once before any new application instance serves traffic. */
|
||||
export async function initializeOwnerAccessAlerts(): Promise<void> {
|
||||
const alerts = alertCollection();
|
||||
await alerts.createIndex({ owner: 1, resourceKind: 1, resourceId: 1 });
|
||||
await alerts.createIndex({ "legacyClaims.owner": 1, "legacyClaims.resourceKind": 1, "legacyClaims.resourceId": 1 });
|
||||
await alerts.createIndex({ claimedAt: 1 }, { name: "owner_access_alert_pending", partialFilterExpression: { state: "pending" } });
|
||||
await alerts.createIndex({ state: 1 });
|
||||
await alerts.createIndex({ "deferredRecoveries.owner": 1 });
|
||||
// Clear known-unsent legacy fields before they can be mistaken for deliveries.
|
||||
await reconcileUnsentAlerts();
|
||||
try { await alerts.updateOne({ _id: MIGRATION }, { $setOnInsert: { state: "new" } }, { upsert: true }); }
|
||||
catch (error) { if ((error as { code?: number }).code !== 11000) throw error; }
|
||||
const lease = randomUUID();
|
||||
for (;;) {
|
||||
if ((await alerts.findOne({ _id: MIGRATION }))?.state === "ready") {
|
||||
await reconcileCompletedSends();
|
||||
await reconcileUnsentAlerts();
|
||||
return;
|
||||
}
|
||||
const now = new Date();
|
||||
const claim = await alerts.updateOne({ _id: MIGRATION, state: { $ne: "ready" },
|
||||
$or: [{ leaseUntil: { $lt: now } }, { leaseUntil: { $exists: false } }],
|
||||
}, { $set: { lease, leaseUntil: new Date(Date.now() + 60000) } });
|
||||
if (!claim.modifiedCount) { await new Promise(resolve => setTimeout(resolve, 250)); continue; }
|
||||
let heartbeatError: unknown;
|
||||
const heartbeat = setInterval(() => {
|
||||
void alerts.updateOne({ _id: MIGRATION, lease }, { $set: { leaseUntil: new Date(Date.now() + 60000) } })
|
||||
.then(result => { if (!result.matchedCount) heartbeatError = new Error("owner_alert_migration_lease_lost"); })
|
||||
.catch(error => { heartbeatError = error; });
|
||||
}, 5000);
|
||||
try {
|
||||
// Delivery evidence survives v1 guard recovery and resource deletion. Restore
|
||||
// missing guards once, before traffic, so those addresses cannot be emailed again.
|
||||
const events = alertCollection("owner_access_alert_events");
|
||||
const index = "owner_access_alert_delivery_history";
|
||||
await events.createIndex({ completedAt: 1, _id: 1 }, { name: index,
|
||||
partialFilterExpression: { completedAt: { $exists: true } } });
|
||||
const checkpoint = (await alerts.findOne({ _id: MIGRATION, lease }))?.deliveryProgress;
|
||||
const filter = { state: { $in: ["accepted", "failed"] }, completedAt: { $exists: true },
|
||||
...(checkpoint ? { $or: [{ completedAt: { $gt: checkpoint.completedAt } },
|
||||
{ completedAt: checkpoint.completedAt, _id: { $gt: checkpoint.id } }] } : {}) };
|
||||
let lastDelivery: { completedAt: Date; id: string } | undefined;
|
||||
let delivered = 0;
|
||||
const saveDeliveries = async () => {
|
||||
if (!lastDelivery) return;
|
||||
const saved = await alerts.updateOne({ _id: MIGRATION, lease }, { $set: { deliveryProgress: lastDelivery } });
|
||||
if (!saved.matchedCount) throw new Error("owner_alert_migration_lease_lost");
|
||||
};
|
||||
for await (const event of events.find(filter).sort({ completedAt: 1, _id: 1 }).hint(index).batchSize(100)) {
|
||||
if (heartbeatError) throw heartbeatError;
|
||||
if (event.recipientKey && /^[0-9a-f]{64}$/.test(event.recipientKey)) {
|
||||
const previous = await alerts.findOne({ _id: event.recipientKey });
|
||||
if (previous && (["pending", "cancelled"].includes(previous.state || "")
|
||||
|| (previous.state === "sending" && previous.dispatchPrepared))) {
|
||||
// A v1 repeat may still be preparing after its earlier guard was
|
||||
// released. Cancel known-unsent work even while its lease is fresh;
|
||||
// committed or ambiguous dispatches remain blocked instead.
|
||||
await releaseUnsentAlert(previous, { state: previous.state });
|
||||
}
|
||||
await alerts.updateOne({ _id: event.recipientKey }, { $setOnInsert: {
|
||||
state: "recipient_notified", attemptId: event.attemptId ?? randomUUID(),
|
||||
claimedAt: event.completedAt, completedAt: event.completedAt, httpStatus: event.httpStatus,
|
||||
} }, { upsert: true });
|
||||
}
|
||||
lastDelivery = { completedAt: event.completedAt!, id: event._id };
|
||||
if (++delivered % 100 === 0) await saveDeliveries();
|
||||
}
|
||||
await saveDeliveries();
|
||||
for (const [kind, name] of Object.entries(resourceCollections)) {
|
||||
const resources = UserModel.db.collection(name);
|
||||
const index = "owner_access_alert_backfill";
|
||||
await resources.createIndex({ accessAlertClaimedAt: 1, _id: 1 },
|
||||
{ name: index, partialFilterExpression: { accessAlertClaimedAt: { $exists: true } } });
|
||||
const checkpoint = (await alerts.findOne({ _id: MIGRATION, lease }))?.progress?.[kind];
|
||||
const filter = { accessAlertClaimedAt: { $exists: true }, ...(checkpoint ? { $or: [
|
||||
{ accessAlertClaimedAt: { $gt: checkpoint.claimedAt } },
|
||||
{ accessAlertClaimedAt: checkpoint.claimedAt, _id: { $gt: checkpoint.id } },
|
||||
] } : {}) };
|
||||
const cursor = resources.find(filter,
|
||||
{ projection: { owner: 1, accessAlertGeneration: 1, accessAlertClaimedAt: 1 } })
|
||||
.sort({ accessAlertClaimedAt: 1, _id: 1 }).hint(index).batchSize(100);
|
||||
let count = 0;
|
||||
let last: { claimedAt: Date; id: Types.ObjectId } | undefined;
|
||||
const saveProgress = async () => {
|
||||
if (!last) return;
|
||||
const saved = await alerts.updateOne({ _id: MIGRATION, lease }, { $set: { [`progress.${kind}`]: last } });
|
||||
if (!saved.matchedCount) throw new Error("owner_alert_migration_lease_lost");
|
||||
};
|
||||
for await (const row of cursor) {
|
||||
if (heartbeatError) throw heartbeatError;
|
||||
last = { claimedAt: row.accessAlertClaimedAt, id: row._id };
|
||||
const owner = await UserModel.findById(row.owner).select("emails notificationEmail status").lean();
|
||||
const email = notificationEmail(owner?.emails, owner?.notificationEmail);
|
||||
const legacyClaim: LegacyClaim = { owner: row.owner, resourceKind: kind as AlertResource["kind"],
|
||||
resourceId: row._id, generation: row.accessAlertGeneration ?? null, recovered: false };
|
||||
if (email && owner && !["removed", "banned"].includes(owner.status || "")) {
|
||||
const key = recipientAlertKey(email);
|
||||
const previous = await alerts.findOne({ _id: key });
|
||||
const claims = [legacyClaim];
|
||||
if (previous && !previous.legacyClaims && previous.owner && previous.resourceKind && previous.resourceId) {
|
||||
claims.push({ owner: previous.owner, resourceKind: previous.resourceKind as AlertResource["kind"],
|
||||
resourceId: previous.resourceId, generation: previous.generation ?? null, recovered: false });
|
||||
}
|
||||
await alerts.updateOne({ _id: key, legacyClaims: { $elemMatch: { owner: legacyClaim.owner,
|
||||
resourceKind: legacyClaim.resourceKind, resourceId: legacyClaim.resourceId, generation: legacyClaim.generation } } },
|
||||
{ $set: { "legacyClaims.$.recovered": false } });
|
||||
await alerts.updateOne({ _id: key }, { $setOnInsert: { owner: legacyClaim.owner, resourceKind: kind,
|
||||
resourceId: row._id, generation: legacyClaim.generation, attemptId: randomUUID(),
|
||||
claimedAt: row.accessAlertClaimedAt, state: "legacy",
|
||||
}, $addToSet: { legacyClaims: { $each: claims } } }, { upsert: true });
|
||||
await alerts.updateOne({ _id: key, state: { $in: ["recovered", "pending"] } }, { $set: { state: "legacy" } });
|
||||
}
|
||||
if (++count % 100 === 0) await saveProgress();
|
||||
}
|
||||
await saveProgress();
|
||||
}
|
||||
if (heartbeatError) throw heartbeatError;
|
||||
const completed = await alerts.updateOne({ _id: MIGRATION, lease }, { $set: { state: "ready" }, $unset: { lease: "", leaseUntil: "" } });
|
||||
if (!completed.matchedCount) throw new Error("owner_alert_migration_lease_lost");
|
||||
await reconcileCompletedSends();
|
||||
await reconcileUnsentAlerts();
|
||||
return;
|
||||
} finally { clearInterval(heartbeat); }
|
||||
}
|
||||
}
|
||||
|
||||
async function completeRecovery(candidate: AlertRecord, recovery: DeferredRecovery): Promise<void> {
|
||||
if (candidate.state !== "pending") await clearOwnerDispatch(candidate, "committed");
|
||||
const fence = { _id: candidate._id, attemptId: candidate.attemptId, resetRevision: generationFilter(recovery.resetRevision),
|
||||
state: { $in: ["pending", "accepted", "failed", "legacy"] } };
|
||||
if (candidate.legacyClaims) {
|
||||
await alertCollection().updateOne(fence, [{ $set: { legacyClaims: { $map: {
|
||||
input: "$legacyClaims", as: "claim", in: { $cond: [{ $and: [
|
||||
{ $eq: ["$$claim.owner", recovery.owner] }, { $eq: ["$$claim.resourceKind", recovery.resourceKind] },
|
||||
{ $eq: ["$$claim.resourceId", recovery.resourceId] }, { $ne: ["$$claim.generation", recovery.generation] },
|
||||
] }, { $mergeObjects: ["$$claim", { recovered: true }] }, "$$claim"] },
|
||||
} } } }, { $set: { state: { $cond: [{ $allElementsTrue: [{ $map: {
|
||||
input: "$legacyClaims", as: "claim", in: { $eq: ["$$claim.recovered", true] },
|
||||
} }] }, "recovered", "$state"] } } }]);
|
||||
} else if (candidate.state === "pending") {
|
||||
await releaseUnsentAlert(candidate, { ...fence, state: "pending" });
|
||||
} else {
|
||||
// Recovery does not authorize another generic email to the same address.
|
||||
await alertCollection().updateOne(fence, { $set: { state: "recovered" } });
|
||||
}
|
||||
}
|
||||
|
||||
const delayedRecoveries = new Map<string, NodeJS.Timeout>();
|
||||
function deferRecoveryUntil(key: string, attemptId: string | undefined, until: Date): void {
|
||||
const timerKey = `${key}/${attemptId}`;
|
||||
if (delayedRecoveries.has(timerKey)) return;
|
||||
const timer = setTimeout(() => {
|
||||
delayedRecoveries.delete(timerKey);
|
||||
void retryAlertWrite(() => applyDeferredRecoveries(key, attemptId))
|
||||
.catch(() => logger.warn("owner access email deferred recovery unavailable"));
|
||||
}, Math.max(1, until.getTime() - Date.now() + 1));
|
||||
timer.unref();
|
||||
delayedRecoveries.set(timerKey, timer);
|
||||
}
|
||||
|
||||
async function applyDeferredRecoveries(key: string, attemptId?: string): Promise<void> {
|
||||
const timerKey = `${key}/${attemptId}`;
|
||||
clearTimeout(delayedRecoveries.get(timerKey));
|
||||
delayedRecoveries.delete(timerKey);
|
||||
const candidate = await alertCollection().findOne({ _id: key, attemptId });
|
||||
if (!candidate || !["pending", "accepted", "failed", "legacy"].includes(candidate.state || "")) return;
|
||||
for (const recovery of candidate.deferredRecoveries ?? []) {
|
||||
if (recovery.resetRevision !== (candidate.resetRevision ?? null)) continue;
|
||||
const row = await UserModel.db.collection(resourceCollections[recovery.resourceKind]).findOne({
|
||||
_id: recovery.resourceId, owner: recovery.owner, accessAlertGeneration: generationFilter(recovery.generation),
|
||||
});
|
||||
if (!row || bindingHash(row, recovery.bindingFields) !== recovery.bindingHash) continue;
|
||||
if (!await commitOwnerRevision(recovery.owner, recovery.ownerRevision)) {
|
||||
const owner = await UserModel.db.collection("users").findOne({ _id: recovery.owner,
|
||||
accessAlertResetRevision: generationFilter(recovery.ownerRevision) }, { projection: { accessAlertResetUntil: 1 } });
|
||||
if (!owner) continue;
|
||||
if (owner.accessAlertResetUntil?.getTime() > Date.now()) {
|
||||
deferRecoveryUntil(key, attemptId, owner.accessAlertResetUntil);
|
||||
continue;
|
||||
}
|
||||
// The lease may have cleared between the failed CAS and this read.
|
||||
if (!await commitOwnerRevision(recovery.owner, recovery.ownerRevision)) continue;
|
||||
}
|
||||
await completeRecovery(candidate, recovery);
|
||||
}
|
||||
}
|
||||
|
||||
/** Confirm recovery after owner action; only known-unsent claims may be released. */
|
||||
export async function withOwnerAccessRecovery<T>(ownerId: string, resource: AlertResource, work: () => Promise<T>, expectedSource: Record<string, unknown> = {}, confirmed: (result: T) => boolean = () => true): Promise<T> {
|
||||
let snapshot: { accessAlertGeneration?: string } | null = null;
|
||||
let candidates: AlertRecord[] = [];
|
||||
let ownerRevision: number | null = null;
|
||||
const valid = UserModel.db.readyState === 1 && Types.ObjectId.isValid(ownerId) && Types.ObjectId.isValid(resource.id);
|
||||
const identity = valid ? { _id: new Types.ObjectId(resource.id), owner: new Types.ObjectId(ownerId), ...expectedSource } : null;
|
||||
if (identity) {
|
||||
try {
|
||||
const row = await UserModel.db.collection(resourceCollections[resource.kind]).findOne(identity, { projection: { accessAlertGeneration: 1 } });
|
||||
snapshot = row ? { accessAlertGeneration: row.accessAlertGeneration } : null;
|
||||
if (snapshot) {
|
||||
const claim = { owner: identity.owner, resourceKind: resource.kind, resourceId: identity._id,
|
||||
generation: { $ne: snapshot.accessAlertGeneration ?? null } };
|
||||
candidates = await alertCollection().find({ state: { $in: ["pending", "accepted", "failed", "legacy", "sending"] },
|
||||
$or: [{ ...claim, legacyClaims: { $exists: false } }, { legacyClaims: { $elemMatch: { ...claim, recovered: { $ne: true } } } }],
|
||||
}).toArray();
|
||||
if (candidates.length) ownerRevision = (await UserModel.db.collection("users").findOne({ _id: identity.owner },
|
||||
{ projection: { accessAlertResetRevision: 1 } }))?.accessAlertResetRevision ?? null;
|
||||
}
|
||||
}
|
||||
catch { /* Email bookkeeping must not change upstream behavior. */ }
|
||||
}
|
||||
const result = await work();
|
||||
if (!identity || !snapshot || !confirmed(result)) return result;
|
||||
try {
|
||||
const current = await UserModel.db.collection(resourceCollections[resource.kind]).findOne({ ...identity,
|
||||
accessAlertGeneration: generationFilter(snapshot.accessAlertGeneration),
|
||||
});
|
||||
if (!current) return result;
|
||||
let recorded = 0;
|
||||
for (const candidate of candidates) {
|
||||
const fence = { _id: candidate._id, attemptId: candidate.attemptId, resetRevision: generationFilter(candidate.resetRevision),
|
||||
state: { $in: ["pending", "accepted", "failed", "legacy", "sending"] } };
|
||||
const fields = Object.keys(expectedSource);
|
||||
const recovery: DeferredRecovery = { owner: identity.owner, resourceKind: resource.kind, resourceId: identity._id,
|
||||
generation: snapshot.accessAlertGeneration ?? null, resetRevision: candidate.resetRevision ?? null, ownerRevision,
|
||||
bindingFields: fields, bindingHash: bindingHash(current, fields) };
|
||||
const saved = await alertCollection().updateOne(fence, [{ $set: { deferredRecoveries: { $concatArrays: [
|
||||
{ $filter: { input: { $ifNull: ["$deferredRecoveries", []] }, as: "recovery", cond: { $not: [{ $and: [
|
||||
{ $eq: ["$$recovery.owner", identity.owner] }, { $eq: ["$$recovery.resourceKind", resource.kind] },
|
||||
{ $eq: ["$$recovery.resourceId", identity._id] },
|
||||
] }] } } }, [recovery],
|
||||
] } } }]);
|
||||
if (saved.matchedCount) recorded++;
|
||||
await applyDeferredRecoveries(candidate._id, candidate.attemptId);
|
||||
}
|
||||
if (recorded) logger.info("owner access email recovery recorded", { resources: recorded });
|
||||
} catch { logger.warn("owner access email recovery unavailable"); }
|
||||
return result;
|
||||
}
|
||||
|
||||
const accessErrors = new Set([
|
||||
"token_expired", "github_oauth_required", "github_app_reconnect_required",
|
||||
"github_app_access_required", "repo_not_found", "repository_not_found",
|
||||
"file_not_accessible", "file_not_found", "pull_request_not_found", "gist_not_found", "repo_not_accessible", "repo_access_limited", "repo_saml_enforcement",
|
||||
]);
|
||||
|
||||
export function normalizeEmail(value: unknown): string | null {
|
||||
if (typeof value !== "string") return null;
|
||||
const email = value.trim();
|
||||
// Accept one plain mailbox, never a display name or recipient list.
|
||||
if (email.length > 254 || !/^[a-zA-Z0-9.!#$%&'*+/=?^_`{|}~-]+@[a-zA-Z0-9](?:[a-zA-Z0-9.-]*[a-zA-Z0-9])?\.[a-zA-Z]{2,}$/.test(email)) return null;
|
||||
const local = email.split("@")[0];
|
||||
if (local.length > 64 || local.startsWith(".") || local.endsWith(".") || email.includes("..")) return null;
|
||||
return email;
|
||||
}
|
||||
|
||||
export function notificationEmail(emails?: { email: string; default: boolean }[], preferred?: string): string {
|
||||
return normalizeEmail(preferred) || normalizeEmail(emails?.find(e => e.default)?.email) ||
|
||||
emails?.map(e => normalizeEmail(e.email)).find(Boolean) || "";
|
||||
}
|
||||
|
||||
export function isAccessFailure(error: unknown): boolean {
|
||||
if (isGitHubRateLimitError(error)) return false;
|
||||
const code = typeof error === "string" ? error : error instanceof Error ? error.message : "";
|
||||
const rawStatus = error instanceof Error ? (error as Error & { status?: number }).status : undefined;
|
||||
if (!accessErrors.has(code) && ![401, 403, 404].includes(rawStatus || 0)) return false;
|
||||
if (error instanceof Error) {
|
||||
const upstream = (error as Error & { cause?: { status?: number; httpStatus?: number; response?: { statusCode?: number } } }).cause;
|
||||
// Some legacy paths wrap timeouts or 5xx responses as repo_not_found.
|
||||
// Only a confirmed access failure should prompt the owner to reconnect.
|
||||
if (isGitHubRateLimitError(upstream)) return false;
|
||||
if (upstream && ![401, 403, 404].includes(upstream.status || upstream.response?.statusCode || upstream.httpStatus || 0)) return false;
|
||||
const status = (error as Error & { httpStatus?: number }).httpStatus;
|
||||
if (status && (status === 429 || status >= 500)) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Best effort; mail and database failures must not change repository behavior. */
|
||||
export async function notifyOwnerAccessProblem(ownerId: string, error: unknown, resource: AlertResource): Promise<void> {
|
||||
if (!config.RESEND_API_KEY || !config.EMAIL_FROM || !isAccessFailure(error)) return;
|
||||
let unsent: AlertRecord | undefined;
|
||||
let dispatchRecord: AlertRecord | undefined;
|
||||
let providerStarted = false;
|
||||
try {
|
||||
if (!resource || !Types.ObjectId.isValid(resource.id) || !Types.ObjectId.isValid(ownerId)) return;
|
||||
const collection = UserModel.db.collection(resourceCollections[resource.kind]);
|
||||
const identity = { _id: new Types.ObjectId(resource.id), owner: new Types.ObjectId(ownerId),
|
||||
status: { $nin: ["archived", "expired", "expiring", "removed", "removing"] } };
|
||||
const ownerState = await UserModel.db.collection("users").findOne({ _id: identity.owner },
|
||||
{ projection: { accessAlertResetRevision: 1, accessAlertResetUntil: 1, accessAlertDeactivations: 1 } });
|
||||
if (!ownerState || ownerState.accessAlertDeactivations?.length || ownerState.accessAlertResetUntil?.getTime() > Date.now()) return;
|
||||
const ownerRevision = ownerState.accessAlertResetRevision ?? null;
|
||||
const snapshot = await collection.findOne(identity, { projection: { accessAlertGeneration: 1, accessAlertClaimedAt: 1, accessAlertClaimRevision: 1 } });
|
||||
if (!snapshot) return;
|
||||
const owner = await UserModel.findOne({ _id: ownerId, status: { $nin: ["removed", "banned"] } }).lean();
|
||||
if (!owner) return;
|
||||
const to = notificationEmail(owner.emails, owner.notificationEmail);
|
||||
if (!to) return;
|
||||
const alerts = alertCollection();
|
||||
const key = recipientAlertKey(to);
|
||||
const attemptId = randomUUID();
|
||||
const record: AlertRecord = { _id: key, owner: identity.owner, resourceKind: resource.kind,
|
||||
resourceId: identity._id, generation: snapshot.accessAlertGeneration ?? null,
|
||||
resourceRevision: snapshot.accessAlertClaimRevision ?? 0, attemptId, claimedAt: new Date(), state: "pending", dispatchPrepared: true };
|
||||
dispatchRecord = record;
|
||||
if (!await prepareOwnerDispatch(record, ownerRevision)) return;
|
||||
unsent = record;
|
||||
try { await alerts.insertOne(record); }
|
||||
catch (error) {
|
||||
if ((error as { code?: number }).code !== 11000) throw error;
|
||||
unsent = undefined;
|
||||
let previous = await alerts.findOne({ _id: key });
|
||||
if (previous?.state === "sending") {
|
||||
await reconcileCompletedSend(previous);
|
||||
previous = await alerts.findOne({ _id: key });
|
||||
}
|
||||
if (previous && ["accepted", "failed"].includes(previous.state || "")) {
|
||||
await clearOwnerDispatch(previous, "committed");
|
||||
await applyDeferredRecoveries(previous._id, previous.attemptId);
|
||||
previous = await alerts.findOne({ _id: key });
|
||||
}
|
||||
if (previous?.state === "sending" && previous.dispatchPrepared && previous.claimedAt
|
||||
&& previous.claimedAt.getTime() <= Date.now() - PENDING_LEASE_MS) {
|
||||
if (await releaseUnsentAlert(previous, expiredPreparation())) previous = await alerts.findOne({ _id: key });
|
||||
}
|
||||
if (previous && (previous.state === "sending" || !previous.owner || !previous.resourceId
|
||||
|| !previous.resourceKind || !(previous.resourceKind in resourceCollections))) return;
|
||||
if (previous) {
|
||||
const retirement: mongo.Filter<AlertRecord> = { _id: key, attemptId: previous.attemptId, state: previous.state,
|
||||
resetRevision: generationFilter(previous.resetRevision) };
|
||||
let retired: boolean;
|
||||
if (previous.state === "pending" || previous.state === "cancelled") {
|
||||
if (previous.state === "pending") {
|
||||
if (!previous.claimedAt || previous.claimedAt.getTime() > Date.now() - PENDING_LEASE_MS) return;
|
||||
}
|
||||
retired = await releaseUnsentAlert(previous, previous.state === "cancelled" ? retirement : { ...retirement, ...expiredPending() });
|
||||
} else {
|
||||
// Delivered, legacy, recovered and ambiguous sends stay blocked permanently,
|
||||
// even if their original resource or account no longer exists.
|
||||
return;
|
||||
}
|
||||
if (!retired) return;
|
||||
}
|
||||
unsent = record;
|
||||
try { await alerts.insertOne(record); }
|
||||
catch (replacementError) { if ((replacementError as { code?: number }).code === 11000) { unsent = undefined; return; } throw replacementError; }
|
||||
}
|
||||
// An earlier claim may have been reclaimed after the initial snapshot, even
|
||||
// when insertOne succeeded. Persist the current fence before resource preparation.
|
||||
const current = await collection.findOne({ ...identity, accessAlertGeneration: generationFilter(record.generation) },
|
||||
{ projection: { accessAlertClaimRevision: 1 } });
|
||||
if (!current) return;
|
||||
const revision = current.accessAlertClaimRevision ?? 0;
|
||||
const refreshed = await alerts.updateOne({ _id: key, attemptId, state: "pending",
|
||||
resetRevision: generationFilter(record.resetRevision) }, { $set: { resourceRevision: revision } });
|
||||
if (!refreshed.matchedCount) return;
|
||||
record.resourceRevision = revision;
|
||||
const base = /^https?:\/\//.test(config.APP_HOSTNAME) ? config.APP_HOSTNAME : `https://${config.APP_HOSTNAME}`;
|
||||
const url = new URL("/connections", base).href;
|
||||
await alertCollection("owner_access_alert_events").insertOne({ _id: attemptId, attemptId,
|
||||
recipientKey: key, state: "prepared", startedAt: new Date(), owner: identity.owner,
|
||||
resourceKind: resource.kind, resourceId: identity._id, generation: record.generation,
|
||||
failureCode: accessErrors.has(typeof error === "string" ? error : error instanceof Error ? error.message : "")
|
||||
? typeof error === "string" ? error : (error as Error).message : "upstream_access_failure" });
|
||||
const claim = await collection.updateOne({ ...identity,
|
||||
accessAlertGeneration: generationFilter(record.generation),
|
||||
accessAlertClaimRevision: revisionFilter(record.resourceRevision ?? 0),
|
||||
}, { $set: { accessAlertClaimedAt: new Date(), accessAlertAttemptId: attemptId }, $inc: { accessAlertClaimRevision: 1 } });
|
||||
if (!claim.matchedCount || !await collection.findOne({ ...identity,
|
||||
accessAlertGeneration: generationFilter(record.generation),
|
||||
accessAlertClaimRevision: (record.resourceRevision ?? 0) + 1, accessAlertAttemptId: attemptId,
|
||||
})) return;
|
||||
// Reserve the recipient transition, but it remains reclaimable until the owner CAS commits dispatch.
|
||||
const dispatch = await alerts.updateOne({ _id: key, attemptId, state: "pending",
|
||||
resetRevision: generationFilter(record.resetRevision) }, { $set: { state: "sending" } });
|
||||
if (!dispatch.matchedCount) return;
|
||||
// Reset and dispatch commit on the same owner document. There is no asynchronous
|
||||
// preparation between this authoritative commitment and the provider request.
|
||||
if (!await commitOwnerRevision(identity.owner, ownerRevision, record)) return;
|
||||
let accepted = false;
|
||||
let httpStatus: number | undefined;
|
||||
try {
|
||||
providerStarted = true;
|
||||
const response = await fetch("https://api.resend.com/emails", {
|
||||
method: "POST",
|
||||
headers: { Authorization: `Bearer ${config.RESEND_API_KEY}`, "Content-Type": "application/json",
|
||||
"Idempotency-Key": `owner-access-alert/${attemptId}` },
|
||||
signal: AbortSignal.timeout(15000),
|
||||
body: JSON.stringify({
|
||||
from: config.EMAIL_FROM, to: [to],
|
||||
subject: "Your Anonymous GitHub connection needs attention",
|
||||
text: `Anonymous GitHub could not access GitHub using your connection, or could not read a repository.\n\nOpen your connections page to reconnect your account or restore repository access:\n${url}\n\nYou are receiving this email because you own content on Anonymous GitHub. You can change your alert address on the connections page.\n`,
|
||||
}),
|
||||
});
|
||||
httpStatus = response.status;
|
||||
accepted = response.ok;
|
||||
} catch { /* Delivery can be ambiguous; retain the recipient guard. */ }
|
||||
const state = accepted ? "accepted" : "failed";
|
||||
if (accepted) logger.info("owner access email accepted", { attemptId, httpStatus });
|
||||
else logger.warn("owner access email delivery failed", { attemptId, httpStatus });
|
||||
const completedAt = new Date();
|
||||
try {
|
||||
await retryAlertWrite(async () => {
|
||||
try { await alertCollection("owner_access_alert_events").insertOne({ _id: `${attemptId}/completion`, attemptId,
|
||||
recipientKey: key, state, httpStatus, completedAt, owner: identity.owner,
|
||||
resourceKind: resource.kind, resourceId: identity._id, generation: record.generation }); }
|
||||
catch (error) { if ((error as { code?: number }).code !== 11000) throw error; }
|
||||
});
|
||||
} catch { logger.warn("owner access email audit unavailable", { attemptId }); }
|
||||
try {
|
||||
await retryAlertWrite(async () => {
|
||||
await alerts.updateOne({ _id: key, attemptId, state: "sending" }, { $set: { state, completedAt, httpStatus } });
|
||||
await clearOwnerDispatch(record, "committed");
|
||||
await applyDeferredRecoveries(key, attemptId);
|
||||
});
|
||||
} catch { logger.warn("owner access email result reconciliation required", { attemptId }); }
|
||||
} catch {
|
||||
logger.warn("owner access email unavailable");
|
||||
} finally {
|
||||
if (unsent && !providerStarted) {
|
||||
await releaseUnsentAlert(unsent, { state: { $in: ["pending", "sending", "cancelled"] } })
|
||||
.catch(() => logger.warn("owner access email claim cleanup unavailable"));
|
||||
}
|
||||
if (dispatchRecord) await clearOwnerDispatch(dispatchRecord, "prepared")
|
||||
.catch(() => logger.warn("owner access email preparation cleanup unavailable"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
import { createLogger, serializeError } from "./logger";
|
||||
|
||||
/** Observe fatal errors without overriding Node's default exit behavior. */
|
||||
export function installFatalErrorLogging(service: "api" | "streamer") {
|
||||
const logger = createLogger("process");
|
||||
const report = (error: Error, origin: string) => logger.error("process fatal error", {
|
||||
...serializeError(error), code: "process_fatal_error", service, origin,
|
||||
});
|
||||
process.on("uncaughtExceptionMonitor", report);
|
||||
return () => process.removeListener("uncaughtExceptionMonitor", report);
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
const sensitive = /^(?:authorization|proxy-authorization|cookie|set-cookie|token|access_?tokens?|refresh_?token|encryptedToken|encryptedRefreshToken|private_?key|GITHUB_APP_PRIVATE_KEY|GITHUB_APP_CLIENT_SECRET|GITHUB_APP_WEBHOOK_SECRET|ciphertext|nonce|tag|password|client_?secret|CREDENTIAL_KEYS)$/i;
|
||||
const sensitive = /^(?:authorization|proxy-authorization|cookie|set-cookie|token|access_?tokens?|refresh_?token|encryptedToken|encryptedRefreshToken|private_?key|GITHUB_APP_PRIVATE_KEY|GITHUB_APP_CLIENT_SECRET|GITHUB_APP_WEBHOOK_SECRET|ciphertext|nonce|tag|password|client_?secret|CREDENTIAL_KEYS|RESEND_API_KEY)$/i;
|
||||
export function redactSecrets(value: unknown, seen = new WeakSet<object>()): unknown {
|
||||
if (typeof value === "string") return value
|
||||
.replace(/\b(?:gh[pousr]_[A-Za-z0-9_]+|github_pat_[A-Za-z0-9_]+)\b/g, "[REDACTED]")
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
import UserModel from "./model/users/users.model";
|
||||
import { IUserDocument } from "./model/users/users.types";
|
||||
import { normalizeEmail, notificationEmail } from "./owner-notifications";
|
||||
|
||||
function usableEmail(value: unknown): string | null {
|
||||
const email = normalizeEmail(value);
|
||||
if (!email) return null;
|
||||
const domain = email.split("@")[1].toLowerCase();
|
||||
return ["users.noreply.github.com", "noreply.github.com"].includes(domain) ? null : email;
|
||||
}
|
||||
|
||||
/** Fill a missing address after the GitHub identity has been authenticated. */
|
||||
export async function saveRegistrationEmail(user: IUserDocument, token: string): Promise<void> {
|
||||
if (notificationEmail(user.emails, user.notificationEmail)) return;
|
||||
// Compare the original fields so a concurrent settings save always wins.
|
||||
const previousEmails = user.toObject().emails;
|
||||
const previousPreference = user.notificationEmail ?? null;
|
||||
try {
|
||||
const response = await fetch("https://api.github.com/user/emails?per_page=100", {
|
||||
headers: { Accept: "application/vnd.github+json", Authorization: `Bearer ${token}`,
|
||||
"X-GitHub-Api-Version": "2022-11-28", "User-Agent": "Anonymous-GitHub" },
|
||||
signal: AbortSignal.timeout(5000),
|
||||
});
|
||||
if (!response.ok) return;
|
||||
const data: unknown = await response.json();
|
||||
if (!Array.isArray(data)) return;
|
||||
const verified = data.filter((entry): entry is { email: string; primary?: boolean } =>
|
||||
!!entry && entry.verified === true && !!usableEmail(entry.email));
|
||||
const email = usableEmail((verified.find(entry => entry.primary === true) || verified[0])?.email);
|
||||
if (!email) return;
|
||||
const emails = [{ email, default: true }];
|
||||
const result = await UserModel.updateOne({ _id: user._id, notificationEmail: previousPreference,
|
||||
$or: [{ emails: previousEmails ?? [] },
|
||||
...(!previousEmails?.length ? [{ emails: { $exists: false } }] : [])],
|
||||
}, { $set: { emails } });
|
||||
if (result.modifiedCount) user.emails = emails;
|
||||
} catch {
|
||||
// Missing App email permission or an upstream outage must not prevent login.
|
||||
// The existing email form remains available when automatic collection fails.
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
/** Persisted repository failures need the same client status as the original error. */
|
||||
export function repositoryFailureStatus(code?: string): number {
|
||||
switch (code) {
|
||||
case "repo_not_found": case "branch_not_found": case "commit_not_found": return 404;
|
||||
case "repo_not_accessible": case "repository_not_accessible": return 403;
|
||||
case "token_expired": case "not_connected": return 401;
|
||||
case "github_rate_limit_exceeded": return 429;
|
||||
case "github_unavailable": return 502;
|
||||
case "repo_empty": return 409;
|
||||
default: return 500;
|
||||
}
|
||||
}
|
||||
@@ -148,8 +148,12 @@ export function monitorRequests(service: Service, metrics = requestMetrics,
|
||||
const slow = ms > (transfer ? 10000 : 1000);
|
||||
metrics.observe("request", route, method, ms, { bytes, aborted, error: res.statusCode >= 500, slow });
|
||||
if (firstByteMs != null) metrics.observe("first_byte", route, method, firstByteMs);
|
||||
const responseMs = firstByteMs == null ? null : Math.max(0, ms - firstByteMs);
|
||||
const detail = { requestId: id, service, method, route, status: res.statusCode,
|
||||
outcome: aborted ? "interrupted" : "completed", ms: Math.round(ms), firstByteMs: firstByteMs == null ? null : Math.round(firstByteMs), bytes, stages: trace.stages };
|
||||
outcome: aborted ? "interrupted" : "completed", ms: Math.round(ms), firstByteMs: firstByteMs == null ? null : Math.round(firstByteMs),
|
||||
responseMs: responseMs == null ? null : Math.round(responseMs),
|
||||
bytesPerSecond: responseMs != null && responseMs >= 1 && bytes > 0 ? Math.round(bytes * 1000 / responseMs) : null,
|
||||
bytes, stages: trace.stages };
|
||||
if (slow || aborted) log.warn(aborted ? "request interrupted" : "slow request", detail);
|
||||
else if (!route.includes("healthcheck")) log.info("request", detail);
|
||||
};
|
||||
|
||||
@@ -52,7 +52,9 @@ export async function classifyGitHubMissError(
|
||||
repo: data.repoName,
|
||||
});
|
||||
return "commit_not_found";
|
||||
} catch {
|
||||
} catch (probeError) {
|
||||
if ((probeError as { status?: number; httpStatus?: number }).status !== 404 &&
|
||||
(probeError as { httpStatus?: number }).httpStatus !== 404) throw probeError;
|
||||
// Repo no longer exists at owner/repo. Try to recover via the cached
|
||||
// numeric GitHub id — if the repo was renamed, GET /repositories/{id}
|
||||
// resolves to its new full_name. See #409.
|
||||
@@ -87,7 +89,9 @@ export async function classifyGitHubMissError(
|
||||
return "repo_renamed";
|
||||
}
|
||||
return "repo_not_found";
|
||||
} catch {
|
||||
} catch (probeError) {
|
||||
if ((probeError as { status?: number; httpStatus?: number }).status !== 404 &&
|
||||
(probeError as { httpStatus?: number }).httpStatus !== 404) throw probeError;
|
||||
return "repo_not_found";
|
||||
}
|
||||
}
|
||||
|
||||
@@ -77,7 +77,8 @@ export class GitHubRepository {
|
||||
try {
|
||||
await oct.repos.get({ owner: this.owner, repo: this.repo });
|
||||
repoExists = true;
|
||||
} catch {
|
||||
} catch (probeError) {
|
||||
if ((probeError as { status?: number }).status !== 404) throw probeError;
|
||||
repoExists = false;
|
||||
}
|
||||
throw new AnonymousError(
|
||||
|
||||
+244
-127
@@ -1,3 +1,5 @@
|
||||
import { withOwnerAccessRecovery } from "../owner-notifications";
|
||||
import { ExpiringMap } from "../expiring-map";
|
||||
import { contentGenerationPrefix, contentRetirementMarker } from "../content-generation";
|
||||
import { createHash } from "crypto";
|
||||
import { coordinatedFill } from "../cache-coordination";
|
||||
@@ -14,21 +16,30 @@ import { basename, dirname } from "path";
|
||||
|
||||
import * as stream from "stream";
|
||||
import AnonymousError from "../AnonymousError";
|
||||
import { octokit, waitForTokenGate } from "../GitHubUtils";
|
||||
import { octokit, waitForTokenGate, RateLimitDelayError } from "../GitHubUtils";
|
||||
import FileModel from "../model/files/files.model";
|
||||
import { IFile } from "../model/files/files.types";
|
||||
import { expandSubmodules, validUpstream } from "./github-submodules";
|
||||
import { IFile, UpstreamFile } from "../model/files/files.types";
|
||||
import { createLogger, serializeError } from "../logger";
|
||||
import { measureStage, recordCacheHit } from "../request-monitoring";
|
||||
import config from "../../config";
|
||||
import { isConnected } from "../../server/database";
|
||||
import AnonymizedRepositoryModel from "../model/anonymizedRepositories/anonymizedRepositories.model";
|
||||
import { RepositoryStatus } from "../types";
|
||||
import { cachedContentMiss, classifyContentMiss } from "./github-content-miss";
|
||||
|
||||
|
||||
const logger = createLogger("gh-stream");
|
||||
|
||||
const GH_API_CONCURRENCY = 6;
|
||||
const cacheFills = new Map<string, Promise<void>>();
|
||||
const missingContent = new ExpiringMap<string>(2048);
|
||||
|
||||
/** Stop upstream before Readable.from waits for the pending iterator to finish. */
|
||||
function stopSourceOnDestroy(output: stream.Readable, stop: () => void): void {
|
||||
const destroy = output._destroy;
|
||||
output._destroy = (error, callback) => { stop(); destroy.call(output, error, callback); };
|
||||
}
|
||||
|
||||
async function pMap<T, R>(
|
||||
items: T[],
|
||||
@@ -64,6 +75,16 @@ export function githubRawFileUrl(
|
||||
)}/raw/${encodeURIComponent(commit)}/${encodedPath}`;
|
||||
}
|
||||
|
||||
function downloadError(error: Error, filePath: string): AnonymousError {
|
||||
if (error instanceof AnonymousError) return error;
|
||||
if (error instanceof RateLimitDelayError) return new AnonymousError("github_rate_limit_exceeded", { httpStatus: 429, cause: error });
|
||||
const response = error as Error & { response?: { statusCode?: number }; status?: number; httpStatus?: number };
|
||||
const status = response.response?.statusCode ?? response.status ?? response.httpStatus;
|
||||
const code = status === 401 || status === 403 ? "file_not_accessible"
|
||||
: status === 404 ? "file_not_found" : status === 422 ? "file_too_big" : "upstream_error";
|
||||
return new AnonymousError(code, { httpStatus: Number.isInteger(status) && status! >= 400 && status! < 500 ? status : 502, cause: error, object: filePath });
|
||||
}
|
||||
|
||||
export default class GitHubStream extends GitHubBase {
|
||||
type: "GitHubDownload" | "GitHubStream" | "Zip" = "GitHubStream";
|
||||
|
||||
@@ -104,10 +125,9 @@ export default class GitHubStream extends GitHubBase {
|
||||
}
|
||||
}
|
||||
|
||||
// GitHub's web raw URL auto-resolves Git LFS pointers via redirect to
|
||||
// media.githubusercontent.com, with the auth header carried through. The
|
||||
// blob endpoint above returns the raw pointer text instead, so we use this
|
||||
// as the fallback for LFS files (#95).
|
||||
// The web URL resolves public LFS pointers. Private web URLs can return
|
||||
// 404 even when the REST API can read the file, so use the scoped contents
|
||||
// API to obtain a short-lived download URL in that case.
|
||||
private downloadFileViaRaw(token: string, filePath: string) {
|
||||
const url = githubRawFileUrl(
|
||||
this.data.organization,
|
||||
@@ -116,11 +136,84 @@ export default class GitHubStream extends GitHubBase {
|
||||
filePath
|
||||
);
|
||||
logger.debug("downloading via raw URL (LFS)", { url });
|
||||
return got.stream(url, {
|
||||
hooks: { beforeRequest: [async () => { await githubTokenContext(token)?.renew(); }] },
|
||||
headers: !token || githubTokenContext(token)?.publicRepository ? {} : { authorization: `token ${token}` },
|
||||
followRedirect: true,
|
||||
});
|
||||
const data = this.data;
|
||||
const cancellation = new AbortController();
|
||||
let active: stream.Readable | undefined;
|
||||
const output = stream.Readable.from((async function* () {
|
||||
try {
|
||||
active = got.stream(url, {
|
||||
hooks: { beforeRequest: [async () => { await githubTokenContext(token)?.renew(); }] },
|
||||
headers: !token || githubTokenContext(token)?.publicRepository ? {} : { authorization: `token ${token}` },
|
||||
followRedirect: true,
|
||||
});
|
||||
for await (const chunk of active) yield chunk;
|
||||
return;
|
||||
} catch (error) {
|
||||
if (!token || githubTokenContext(token)?.publicRepository ||
|
||||
(error as { response?: { statusCode?: number } }).response?.statusCode !== 404) throw error;
|
||||
await waitForTokenGate(token, cancellation.signal);
|
||||
const response = await octokit(token).repos.getContent({
|
||||
owner: data.organization, repo: data.repoName,
|
||||
path: filePath, ref: data.commit || "HEAD", mediaType: { format: "object" },
|
||||
request: { signal: cancellation.signal, timeout: 30_000 },
|
||||
}).catch(apiError => {
|
||||
if (apiError instanceof AnonymousError) throw apiError;
|
||||
const upstream = apiError as { status?: number; response?: { data?: { message?: string } } };
|
||||
if (upstream.status === 422) {
|
||||
const missingCommit = /^No commit found for SHA\b/.test(upstream.response?.data?.message || "");
|
||||
throw new AnonymousError(missingCommit ? "commit_not_found" : "upstream_error", {
|
||||
httpStatus: missingCommit ? 404 : 502, cause: apiError as Error,
|
||||
});
|
||||
}
|
||||
throw apiError;
|
||||
});
|
||||
if (cancellation.signal.aborted) return;
|
||||
const metadata = response.data as { type?: string; download_url?: string };
|
||||
if (metadata.type !== "file" || !metadata.download_url) throw error;
|
||||
const assertDownloadUrl = (value: URL) => {
|
||||
if (value.protocol !== "https:" || value.username || value.password || value.port ||
|
||||
!["raw.githubusercontent.com", "media.githubusercontent.com"].includes(value.hostname)) {
|
||||
throw new AnonymousError("upstream_error", { httpStatus: 502 });
|
||||
}
|
||||
};
|
||||
assertDownloadUrl(new URL(metadata.download_url));
|
||||
// The API URL supplies its own authorization. Never forward the
|
||||
// owner's token to the download host or a redirect destination.
|
||||
active = got.stream(metadata.download_url, {
|
||||
hooks: { beforeRedirect: [options => { assertDownloadUrl(options.url); }] },
|
||||
});
|
||||
let probe = Buffer.alloc(0);
|
||||
let checked = false;
|
||||
for await (const chunk of active) {
|
||||
if (checked) { yield chunk; continue; }
|
||||
probe = Buffer.concat([probe, Buffer.from(chunk)]);
|
||||
if (probe.length < 150) continue;
|
||||
if (probe.toString("utf8").startsWith("version https://git-lfs.github.com/spec/")) {
|
||||
throw new AnonymousError("upstream_error", { httpStatus: 502 });
|
||||
}
|
||||
checked = true; yield probe;
|
||||
}
|
||||
if (!checked && probe.length) {
|
||||
if (probe.toString("utf8").startsWith("version https://git-lfs.github.com/spec/")) {
|
||||
throw new AnonymousError("upstream_error", { httpStatus: 502 });
|
||||
}
|
||||
yield probe;
|
||||
}
|
||||
}
|
||||
})(), { objectMode: false });
|
||||
stopSourceOnDestroy(output, () => { cancellation.abort(); active?.destroy(); });
|
||||
return output;
|
||||
}
|
||||
|
||||
private deferredRawDownload(token: string, filePath: string): stream.Readable {
|
||||
let active: stream.Readable | undefined;
|
||||
const download = () => this.downloadFileViaRaw(token, filePath);
|
||||
const output = stream.Readable.from((async function* () {
|
||||
active = download();
|
||||
for await (const chunk of active) yield chunk;
|
||||
})(), { objectMode: false });
|
||||
stopSourceOnDestroy(output, () => active?.destroy());
|
||||
return output;
|
||||
}
|
||||
|
||||
// Try the blob API, then fall back to the raw URL on statuses where the
|
||||
@@ -135,9 +228,9 @@ export default class GitHubStream extends GitHubBase {
|
||||
// Public raw downloads need no bearer token and do not consume the
|
||||
// unauthenticated REST API quota. GitHub also resolves LFS pointers here.
|
||||
if (!token || githubTokenContext(token)?.publicRepository) {
|
||||
return Promise.resolve(this.downloadFileViaRaw(token, filePath));
|
||||
return Promise.resolve(this.deferredRawDownload(token, filePath));
|
||||
}
|
||||
return new Promise<stream.Readable>((resolve) => {
|
||||
return new Promise<stream.Readable>((resolve, reject) => {
|
||||
const blobStream = this.downloadFile(token, sha);
|
||||
let settled = false;
|
||||
|
||||
@@ -148,7 +241,7 @@ export default class GitHubStream extends GitHubBase {
|
||||
filePath,
|
||||
statusCode,
|
||||
});
|
||||
resolve(this.downloadFileViaRaw(token, filePath));
|
||||
resolve(this.deferredRawDownload(token, filePath));
|
||||
};
|
||||
|
||||
blobStream.on("error", (err) => {
|
||||
@@ -160,13 +253,10 @@ export default class GitHubStream extends GitHubBase {
|
||||
fallbackToRaw(statusCode);
|
||||
return;
|
||||
}
|
||||
// Other errors: let the normal pipeline handle them.
|
||||
// Defer destroy so callers can attach error listeners before
|
||||
// the error event fires, avoiding an uncaughtException crash.
|
||||
// Reject before handing a stream to the caller. nextTick runs before
|
||||
// await continuations, so emitting on a returned stream here can crash.
|
||||
settled = true;
|
||||
const passthrough = new stream.PassThrough();
|
||||
resolve(passthrough);
|
||||
process.nextTick(() => passthrough.destroy(err));
|
||||
reject(err);
|
||||
});
|
||||
|
||||
blobStream.on("response", (response) => {
|
||||
@@ -191,9 +281,7 @@ export default class GitHubStream extends GitHubBase {
|
||||
token: string,
|
||||
filePath: string
|
||||
): stream.Readable {
|
||||
const out = new stream.PassThrough();
|
||||
let active = blobStream;
|
||||
out.once("close", () => active.destroy());
|
||||
// The generator only reads upstream when the output has capacity.
|
||||
const downloadRaw = () => this.downloadFileViaRaw(token, filePath);
|
||||
async function* resolve() {
|
||||
@@ -208,7 +296,7 @@ export default class GitHubStream extends GitHubBase {
|
||||
probe = Buffer.concat([probe, bytes]);
|
||||
if (probe.length < 150) continue;
|
||||
decided = true;
|
||||
if (probe.toString("utf8", 0, 39) === "version https://git-lfs.github.com/spec/") {
|
||||
if (probe.toString("utf8").startsWith("version https://git-lfs.github.com/spec/")) {
|
||||
active = downloadRaw();
|
||||
for await (const raw of active) yield raw;
|
||||
return;
|
||||
@@ -225,28 +313,37 @@ export default class GitHubStream extends GitHubBase {
|
||||
}
|
||||
}
|
||||
}
|
||||
const resolved = stream.Readable.from(resolve());
|
||||
out.once("close", () => resolved.destroy());
|
||||
resolved.on("error", (error) => out.destroy(error));
|
||||
resolved.pipe(out);
|
||||
|
||||
return out;
|
||||
const resolved = stream.Readable.from(resolve(), { objectMode: false });
|
||||
// Keep the generator lazy until the consumer has installed its handlers.
|
||||
stopSourceOnDestroy(resolved, () => active.destroy());
|
||||
return resolved;
|
||||
}
|
||||
|
||||
async getFileContentCache(
|
||||
filePath: string,
|
||||
repoId: string,
|
||||
fileMeta: () =>
|
||||
| Promise<{ sha: string; size?: number }>
|
||||
| { sha: string; size?: number }
|
||||
| Promise<{ sha: string; size?: number; upstream?: UpstreamFile }>
|
||||
| { sha: string; size?: number; upstream?: UpstreamFile }
|
||||
| Promise<string>
|
||||
| string
|
||||
) {
|
||||
const meta = await fileMeta();
|
||||
const expected: { sha: string; size?: number } =
|
||||
const expected: { sha: string; size?: number; upstream?: UpstreamFile } =
|
||||
typeof meta === "string" ? { sha: meta } : meta;
|
||||
if (expected.upstream && !validUpstream(expected.upstream)) throw new AnonymousError("invalid_request", { httpStatus: 400 });
|
||||
// Hydrated Mongoose subdocuments expose source fields through getters,
|
||||
// so spreading them would retain the parent's repository coordinates.
|
||||
const origin = expected.upstream ? new GitHubStream({
|
||||
...this.data,
|
||||
organization: expected.upstream.organization,
|
||||
repoName: expected.upstream.repoName,
|
||||
commit: expected.upstream.commit,
|
||||
getToken: () => "",
|
||||
}) : this;
|
||||
const sourcePath = expected.upstream?.path || filePath;
|
||||
const generation = createHash("sha256").update(JSON.stringify([
|
||||
this.data.commit, expected.sha, this.data.cacheGeneration, this.data.cacheRevision,
|
||||
this.data.commit, expected.sha, this.data.cacheGeneration, this.data.cacheRevision, ...(expected.upstream ? [expected.upstream] : []),
|
||||
])).digest("hex");
|
||||
const cachePath = this.data.cacheGeneration ? `${contentGenerationPrefix(this.data.cacheGeneration)}/${generation}/${filePath}` : filePath;
|
||||
const assertActive = async () => {
|
||||
@@ -298,35 +395,51 @@ export default class GitHubStream extends GitHubBase {
|
||||
if (cacheFills.size >= 64) throw new AnonymousError("cache_busy", { httpStatus: 503 });
|
||||
filling = coordinatedFill(key, cached, async () => {
|
||||
await assertActive();
|
||||
const token = await this.data.getToken();
|
||||
return measureStage("upstream", async () => {
|
||||
const content = await this.downloadWithFallback(token, expected.sha, filePath);
|
||||
let count = 0;
|
||||
const limited = new stream.Transform({ transform(chunk, encoding, callback) {
|
||||
count += chunk.length;
|
||||
callback(count > config.MAX_FILE_SIZE ? new AnonymousError("file_too_big", { httpStatus: 413 }) : null, chunk);
|
||||
} });
|
||||
content.on("error", error => {
|
||||
if (error instanceof AnonymousError) { limited.destroy(error); return; }
|
||||
const status = (error as { response?: { statusCode?: number }; status?: number; httpStatus?: number });
|
||||
const httpStatus = status.response?.statusCode ?? status.status ?? status.httpStatus;
|
||||
const code = httpStatus === 403 ? "file_not_accessible"
|
||||
: httpStatus === 404 ? "file_not_found"
|
||||
: httpStatus === 422 ? "file_too_big" : "upstream_error";
|
||||
limited.destroy(new AnonymousError(code, { httpStatus: httpStatus || 502, cause: error, object: filePath }));
|
||||
const parentToken = await this.data.getToken();
|
||||
const token = expected.upstream ? "" : parentToken;
|
||||
const sourceMiss = cachedContentMiss(this.data, parentToken);
|
||||
if (sourceMiss && sourceMiss !== "file_not_found") throw new AnonymousError(sourceMiss, { httpStatus: 404 });
|
||||
const missKey = createHash("sha256").update(JSON.stringify([
|
||||
key, this.data.organization, this.data.repoName, token,
|
||||
])).digest("hex");
|
||||
const missed = missingContent.get(missKey);
|
||||
if (missed) throw new AnonymousError(missed, { httpStatus: 404 });
|
||||
try {
|
||||
const fetchContent = () => measureStage("upstream", async () => {
|
||||
const content = await origin.downloadWithFallback(token, expected.sha, sourcePath)
|
||||
.catch(error => { throw downloadError(error, filePath); });
|
||||
let count = 0;
|
||||
const limited = new stream.Transform({ transform(chunk, encoding, callback) {
|
||||
count += chunk.length;
|
||||
callback(count > config.MAX_FILE_SIZE ? new AnonymousError("file_too_big", { httpStatus: 413 }) : null, chunk);
|
||||
} });
|
||||
content.on("error", error => limited.destroy(downloadError(error, filePath)));
|
||||
limited.once("close", () => content.destroy());
|
||||
limited.on("error", () => {});
|
||||
content.pipe(limited);
|
||||
const deadline = setTimeout(() => limited.destroy(new AnonymousError("upstream_error", { httpStatus: 502 })), 120_000);
|
||||
try {
|
||||
await storage.write(repoId, cachePath, limited, this.type, expected.size);
|
||||
try { await assertActive(); }
|
||||
catch (error) { await storage.rm(repoId, cachePath); throw error; }
|
||||
}
|
||||
finally { clearTimeout(deadline); limited.destroy(); content.destroy(); }
|
||||
return true;
|
||||
});
|
||||
limited.once("close", () => content.destroy());
|
||||
limited.on("error", () => {});
|
||||
content.pipe(limited);
|
||||
const deadline = setTimeout(() => limited.destroy(new AnonymousError("upstream_error", { httpStatus: 502 })), 120_000);
|
||||
try {
|
||||
await storage.write(repoId, cachePath, limited, this.type, expected.size);
|
||||
try { await assertActive(); }
|
||||
catch (error) { await storage.rm(repoId, cachePath); throw error; }
|
||||
return expected.upstream ? await fetchContent() : await this.confirmOwnerAccess(repoId, fetchContent);
|
||||
} catch (error) {
|
||||
if (expected.upstream && error instanceof AnonymousError &&
|
||||
["file_not_found", "file_not_accessible"].includes(error.message)) {
|
||||
// A removed public dependency is not a lost parent credential.
|
||||
throw new AnonymousError("submodule_not_available", { httpStatus: 404, cause: error });
|
||||
}
|
||||
finally { clearTimeout(deadline); limited.destroy(); content.destroy(); }
|
||||
return true;
|
||||
});
|
||||
if (!expected.upstream && error instanceof AnonymousError && error.message === "file_not_found" && error.httpStatus === 404) {
|
||||
const code = await classifyContentMiss(this.data, token);
|
||||
missingContent.set(missKey, code, 30_000);
|
||||
if (code !== "file_not_found") throw new AnonymousError(code, { httpStatus: 404, cause: error });
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}).then(() => {});
|
||||
cacheFills.set(key, filling);
|
||||
void filling.finally(() => { if (cacheFills.get(key) === filling) cacheFills.delete(key); }).catch(() => {});
|
||||
@@ -353,14 +466,14 @@ export default class GitHubStream extends GitHubBase {
|
||||
object: file,
|
||||
});
|
||||
}
|
||||
return { sha: fileSha, size: await file.size() };
|
||||
return { sha: fileSha, size: await file.size(), upstream: (await file.getFileInfo()).upstream };
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
async getFiles(progress?: (status: string) => void) {
|
||||
this._truncatedFolders = [];
|
||||
return this.getTruncatedTree(this.data.commit, progress);
|
||||
return expandSubmodules(await this.getTruncatedTree(this.data.commit, progress), this.data);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -371,6 +484,26 @@ export default class GitHubStream extends GitHubBase {
|
||||
* files above the 1MB contents-API inline limit.
|
||||
*/
|
||||
async fetchFileInfoFromPath(filePath: string): Promise<IFile | null> {
|
||||
return this.confirmOwnerAccess(this.data.repoId, () => this.fetchFileMetadata(filePath), result => result !== null);
|
||||
}
|
||||
|
||||
private async confirmOwnerAccess<T>(repoId: string, work: () => Promise<T>, confirmed?: (result: T) => boolean): Promise<T> {
|
||||
let repository;
|
||||
try {
|
||||
repository = isConnected ? await AnonymizedRepositoryModel.findOne({ repoId })
|
||||
.select("owner source githubAccess").lean().exec() : null;
|
||||
} catch {
|
||||
// Email recovery bookkeeping must not prevent the upstream read.
|
||||
return work();
|
||||
}
|
||||
return repository?.owner ? withOwnerAccessRecovery(String(repository.owner),
|
||||
{ kind: "repository", id: String(repository._id) }, work,
|
||||
{ "source.repositoryName": `${this.data.organization}/${this.data.repoName}`, "source.commit": this.data.commit,
|
||||
"githubAccess.revision": repository.githubAccess?.revision ?? { $exists: false } }, confirmed) : work();
|
||||
}
|
||||
|
||||
private async fetchFileMetadata(filePath: string): Promise<IFile | null> {
|
||||
if (basename(filePath) === ".gitmodules") return null;
|
||||
const token = await this.data.getToken();
|
||||
const oct = octokit(token);
|
||||
try {
|
||||
@@ -393,10 +526,13 @@ export default class GitHubStream extends GitHubBase {
|
||||
size: data.size,
|
||||
};
|
||||
} catch (error) {
|
||||
logger.debug("fetchFileInfoFromPath miss", {
|
||||
filePath,
|
||||
error: serializeError(error),
|
||||
});
|
||||
if ((error as { status?: number }).status !== 404) throw error;
|
||||
// A missing file and lost repository access both return 404. Only
|
||||
// report a normal miss after the repository probe remains readable.
|
||||
const code = await classifyGitHubMissError(error, this.data);
|
||||
if (code === "repo_not_found") {
|
||||
throw new AnonymousError(code, { httpStatus: 404, cause: error as Error });
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -408,19 +544,37 @@ export default class GitHubStream extends GitHubBase {
|
||||
count = { request: 0, file: 0 },
|
||||
opt = { recursive: true, callback: () => {} }
|
||||
) {
|
||||
await waitForTokenGate(token);
|
||||
const ghRes = await oct.git.getTree({
|
||||
owner: this.data.organization,
|
||||
repo: this.data.repoName,
|
||||
tree_sha: sha,
|
||||
recursive: opt.recursive === true ? "1" : undefined,
|
||||
});
|
||||
count.request++;
|
||||
count.file += ghRes.data.tree.length;
|
||||
if (opt.callback) {
|
||||
opt.callback();
|
||||
try {
|
||||
await waitForTokenGate(token);
|
||||
const ghRes = await oct.git.getTree({
|
||||
owner: this.data.organization,
|
||||
repo: this.data.repoName,
|
||||
tree_sha: sha,
|
||||
recursive: opt.recursive === true ? "1" : undefined,
|
||||
});
|
||||
count.request++;
|
||||
count.file += ghRes.data.tree.length;
|
||||
if (opt.callback) opt.callback();
|
||||
return ghRes.data;
|
||||
} catch (error) {
|
||||
// Keep quota, credential and lifecycle errors intact. A failed tree
|
||||
// request does not imply that the repository has disappeared.
|
||||
if (error instanceof AnonymousError || error instanceof RateLimitDelayError) throw error;
|
||||
const upstream = error as { status?: number; response?: { data?: { message?: string } } };
|
||||
const status = upstream.status;
|
||||
const invalidObject = status === 422 && /^Invalid object requested\. SHA must identify a commit or a tree\b/.test(upstream.response?.data?.message || "");
|
||||
const code = status === 409 ? "repo_empty"
|
||||
: status === 404 ? await classifyGitHubMissError(error, this.data)
|
||||
: invalidObject ? "commit_not_found"
|
||||
: status === 401 ? "token_expired"
|
||||
: status === 403 ? "repo_not_accessible" : "github_unavailable";
|
||||
throw new AnonymousError(code, {
|
||||
httpStatus: status === 409 ? 409 : status === 404 || invalidObject ? 404
|
||||
: status === 401 || status === 403 ? status : 502,
|
||||
object: this.data,
|
||||
cause: error as Error,
|
||||
});
|
||||
}
|
||||
return ghRes.data;
|
||||
}
|
||||
|
||||
private async getTruncatedTree(
|
||||
@@ -435,55 +589,17 @@ export default class GitHubStream extends GitHubBase {
|
||||
file: 0,
|
||||
};
|
||||
const output: IFile[] = [];
|
||||
let data;
|
||||
try {
|
||||
data = await this.getGHTree(oct, token, sha, count, {
|
||||
recursive: true,
|
||||
callback: () => {
|
||||
if (progress) {
|
||||
progress("List file: " + count.file);
|
||||
}
|
||||
},
|
||||
});
|
||||
if (!data.truncated) return this.tree2Tree(data.tree, parentPath);
|
||||
data = await this.getGHTree(oct, token, sha, count, {
|
||||
recursive: false,
|
||||
callback: () => {
|
||||
if (progress) {
|
||||
progress("List file: " + count.file);
|
||||
}
|
||||
},
|
||||
});
|
||||
if (data.truncated) {
|
||||
this._truncatedFolders.push(parentPath);
|
||||
}
|
||||
output.push(...this.tree2Tree(data.tree, parentPath));
|
||||
} catch (error) {
|
||||
const status = (error as { status?: number }).status;
|
||||
if (status === 409) {
|
||||
logger.debug("getTree empty repo", serializeError(error));
|
||||
throw new AnonymousError("repo_empty", {
|
||||
httpStatus: 409,
|
||||
object: this.data,
|
||||
cause: error as Error,
|
||||
});
|
||||
}
|
||||
if (status === 404) {
|
||||
logger.debug("getTree miss", serializeError(error));
|
||||
const code = await classifyGitHubMissError(error, this.data);
|
||||
throw new AnonymousError(code, {
|
||||
httpStatus: 404,
|
||||
object: this.data,
|
||||
cause: error as Error,
|
||||
});
|
||||
}
|
||||
logger.warn("getTree failed", serializeError(error));
|
||||
throw new AnonymousError("repo_not_found", {
|
||||
httpStatus: status || 404,
|
||||
object: this.data,
|
||||
cause: error as Error,
|
||||
});
|
||||
}
|
||||
let data = await this.getGHTree(oct, token, sha, count, {
|
||||
recursive: true,
|
||||
callback: () => { progress?.("List file: " + count.file); },
|
||||
});
|
||||
if (!data.truncated) return this.tree2Tree(data.tree, parentPath);
|
||||
data = await this.getGHTree(oct, token, sha, count, {
|
||||
recursive: false,
|
||||
callback: () => { progress?.("List file: " + count.file); },
|
||||
});
|
||||
if (data.truncated) this._truncatedFolders.push(parentPath);
|
||||
output.push(...this.tree2Tree(data.tree, parentPath));
|
||||
const subtrees: { sha: string; parentPath: string }[] = [];
|
||||
for (const file of data.tree) {
|
||||
if (file.type == "tree" && file.path && file.sha) {
|
||||
@@ -568,6 +684,7 @@ export default class GitHubStream extends GitHubBase {
|
||||
path: pathFile,
|
||||
repoId: this.data.repoId,
|
||||
size: elem.size,
|
||||
...(elem.type === "commit" ? { submodule: "unsupported" } : {}),
|
||||
sha: elem.sha,
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
import { createHash } from "crypto";
|
||||
import AnonymousError from "../AnonymousError";
|
||||
import { ExpiringMap } from "../expiring-map";
|
||||
import { octokit, waitForTokenGate, RateLimitDelayError } from "../GitHubUtils";
|
||||
import { GitHubBaseData } from "./GitHubBase";
|
||||
|
||||
type Miss = "file_not_found" | "repo_not_found" | "commit_not_found";
|
||||
const results = new ExpiringMap<Miss>(2048);
|
||||
const pending = new Map<string, Promise<Miss>>();
|
||||
|
||||
function missKey(data: GitHubBaseData, token: string): string {
|
||||
return createHash("sha256").update(JSON.stringify([
|
||||
data.repoId, data.organization, data.repoName, data.commit, data.cacheGeneration, data.cacheRevision, token,
|
||||
])).digest("hex");
|
||||
}
|
||||
|
||||
export function cachedContentMiss(data: GitHubBaseData, token: string): Miss | undefined {
|
||||
return results.get(missKey(data, token));
|
||||
}
|
||||
|
||||
/** A raw 404 can mean a missing file, lost repository access, or a removed revision. */
|
||||
export async function classifyContentMiss(data: GitHubBaseData, token: string): Promise<Miss> {
|
||||
const key = missKey(data, token);
|
||||
const cached = results.get(key);
|
||||
if (cached) return cached;
|
||||
const active = pending.get(key);
|
||||
if (active) return active;
|
||||
if (pending.size >= 64) throw new AnonymousError("cache_busy", { httpStatus: 503 });
|
||||
const probe = (async (): Promise<Miss> => {
|
||||
const oct = octokit(token);
|
||||
const options = { owner: data.organization, repo: data.repoName, request: { timeout: 15_000 } };
|
||||
try {
|
||||
await waitForTokenGate(token);
|
||||
await oct.repos.get(options);
|
||||
} catch (error) {
|
||||
if ((error as { status?: number }).status === 404) return "repo_not_found";
|
||||
throw probeError(error);
|
||||
}
|
||||
try {
|
||||
await waitForTokenGate(token);
|
||||
// Only check that the revision resolves; do not download a potentially
|
||||
// large commit diff just to classify a missing file.
|
||||
await oct.repos.getCommit({ ...options, ref: data.commit || "HEAD", mediaType: { format: "sha" } });
|
||||
} catch (error) {
|
||||
const upstream = error as { status?: number; response?: { data?: { message?: string } } };
|
||||
if (upstream.status === 404 || upstream.status === 409 || (upstream.status === 422 &&
|
||||
/^No commit found for SHA\b/.test(upstream.response?.data?.message || ""))) return "commit_not_found";
|
||||
throw probeError(error);
|
||||
}
|
||||
return "file_not_found";
|
||||
})().then(result => { results.set(key, result, 30_000); return result; });
|
||||
pending.set(key, probe);
|
||||
try { return await probe; }
|
||||
finally { if (pending.get(key) === probe) pending.delete(key); }
|
||||
}
|
||||
|
||||
function probeError(error: unknown): AnonymousError {
|
||||
if (error instanceof AnonymousError) return error;
|
||||
if (error instanceof RateLimitDelayError) return new AnonymousError("github_rate_limit_exceeded", { httpStatus: 429, cause: error });
|
||||
const status = (error as { status?: number }).status;
|
||||
return new AnonymousError(status === 401 ? "token_expired" : status === 403 ? "repo_not_accessible" : "github_unavailable", {
|
||||
httpStatus: status === 401 || status === 403 ? status : 502,
|
||||
cause: error as Error,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
import { octokit, waitForTokenGate } from "../GitHubUtils";
|
||||
import AnonymousError from "../AnonymousError";
|
||||
import { IFile, UpstreamFile } from "../model/files/files.types";
|
||||
import { GitHubBaseData } from "./GitHubBase";
|
||||
|
||||
export const MAX_SUBMODULES = 20;
|
||||
export const MAX_SUBMODULE_ENTRIES = 10000;
|
||||
export interface SubmoduleMount extends Omit<UpstreamFile, "path"> { path: string; }
|
||||
|
||||
export function safeRepositoryPath(value: unknown): value is string {
|
||||
return typeof value === "string" && value.length > 0 &&
|
||||
!value.includes("\\") && !value.includes("\0") &&
|
||||
value.split("/").every(part => part !== "" && part !== "." && part !== "..");
|
||||
}
|
||||
|
||||
export function validUpstream(value: unknown): value is UpstreamFile {
|
||||
if (!value || typeof value !== "object") return false;
|
||||
const v = value as UpstreamFile;
|
||||
return typeof v.organization === "string" && /^[a-zA-Z0-9-]+$/.test(v.organization) &&
|
||||
typeof v.repoName === "string" && /^[a-zA-Z0-9_.-]+$/.test(v.repoName) && ![".", ".."].includes(v.repoName) &&
|
||||
typeof v.commit === "string" && /^[a-f0-9]{40}$/.test(v.commit) && safeRepositoryPath(v.path);
|
||||
}
|
||||
|
||||
export function submoduleRepository(url: unknown, parent: GitHubBaseData): { organization: string; repoName: string } | null {
|
||||
if (typeof url !== "string") return null;
|
||||
const normalized = url.replace(/^git@github\.com:/, "https://github.com/").replace(/^ssh:\/\/git@github\.com\//, "https://github.com/").replace(/^git:\/\/github\.com\//, "https://github.com/");
|
||||
try {
|
||||
const parsed = new URL(normalized, url.startsWith("./") || url.startsWith("../")
|
||||
? `https://github.com/${parent.organization}/${parent.repoName}/` : undefined);
|
||||
if (parsed.protocol !== "https:" || parsed.hostname !== "github.com" || parsed.port || parsed.username || parsed.password || parsed.search || parsed.hash) return null;
|
||||
const parts = parsed.pathname.replace(/\.git\/?$/, "").replace(/\/$/, "").split("/");
|
||||
if (parts.length !== 3) return null;
|
||||
const candidate = { organization: parts[1], repoName: parts[2], commit: "a".repeat(40), path: "file" };
|
||||
return validUpstream(candidate) ? { organization: candidate.organization, repoName: candidate.repoName } : null;
|
||||
} catch { return null; }
|
||||
}
|
||||
|
||||
/** Resolve only direct public GitHub dependencies. Never send the parent token to a child. */
|
||||
export async function expandSubmodules(files: IFile[], data: GitHubBaseData): Promise<IFile[]> {
|
||||
const output = files.filter(file => file.name !== ".gitmodules");
|
||||
const modules = output.filter(file => file.submodule);
|
||||
let remaining = MAX_SUBMODULE_ENTRIES;
|
||||
for (const [index, mount] of modules.entries()) {
|
||||
if (index >= MAX_SUBMODULES) { mount.submodule = "limit"; continue; }
|
||||
const mountPath = [mount.path, mount.name].filter(Boolean).join("/");
|
||||
if (!safeRepositoryPath(mountPath) || !mount.sha || !/^[a-f0-9]{40}$/.test(mount.sha)) continue;
|
||||
const token = await data.getToken();
|
||||
await waitForTokenGate(token);
|
||||
try {
|
||||
const response = await octokit(token).repos.getContent({
|
||||
owner: data.organization, repo: data.repoName, path: mountPath, ref: data.commit,
|
||||
});
|
||||
const metadata = response.data as { submodule_git_url?: string; sha?: string };
|
||||
const repository = submoduleRepository(metadata.submodule_git_url, data);
|
||||
if (!repository || metadata.sha !== mount.sha) continue;
|
||||
// An unauthenticated request proves public access, even for an owner whose
|
||||
// parent token can also read private dependencies. No credentials cross repos.
|
||||
await waitForTokenGate("");
|
||||
const child = await octokit("").git.getTree({
|
||||
owner: repository.organization, repo: repository.repoName, tree_sha: mount.sha, recursive: "1",
|
||||
});
|
||||
if (child.data.truncated || child.data.tree.length > remaining) { mount.submodule = "limit"; continue; }
|
||||
const entries = child.data.tree;
|
||||
if (entries.some(entry => !safeRepositoryPath(entry.path))) { mount.submodule = "unsupported"; continue; }
|
||||
remaining -= entries.length;
|
||||
mount.submodule = "included";
|
||||
mount.upstream = { ...repository, commit: mount.sha, path: "" };
|
||||
for (const entry of entries) {
|
||||
const relative = entry.path!;
|
||||
if (relative.split("/").pop() === ".gitmodules") continue;
|
||||
const fullPath = `${mountPath}/${relative}`;
|
||||
const slash = fullPath.lastIndexOf("/");
|
||||
output.push({
|
||||
repoId: data.repoId, name: fullPath.slice(slash + 1), path: fullPath.slice(0, slash),
|
||||
sha: entry.sha, size: entry.size,
|
||||
...(entry.type === "commit" ? { submodule: "nested" as const } : {}),
|
||||
...(entry.type === "blob" ? { upstream: { ...repository, commit: mount.sha, path: relative } } : {}),
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
if ((error as { status?: number }).status === 404) { mount.submodule = "unavailable"; continue; }
|
||||
// Do not persist a partial tree on rate limits, outages or credential errors.
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
return output;
|
||||
}
|
||||
|
||||
export function validateMounts(value: unknown): SubmoduleMount[] {
|
||||
if (value === undefined) return [];
|
||||
if (!Array.isArray(value) || value.length > MAX_SUBMODULES || value.some(mount => !validUpstream(mount))) {
|
||||
throw new AnonymousError("invalid_request", { httpStatus: 400 });
|
||||
}
|
||||
const paths = value.map(mount => mount.path as string);
|
||||
if (paths.some((path, i) => paths.some((other, j) => i !== j && (path === other || path.startsWith(other + "/"))))) {
|
||||
throw new AnonymousError("invalid_request", { httpStatus: 400 });
|
||||
}
|
||||
return value;
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
import { appendArchiveEntry } from "../archive-entry";
|
||||
import { appendArchiveEntry, zipEntryOptions } from "../archive-entry";
|
||||
import config from "../../config";
|
||||
import * as fs from "fs";
|
||||
import { Extract } from "unzip-stream";
|
||||
@@ -280,7 +280,7 @@ export default class FileSystem extends StorageBase {
|
||||
output = source.pipe(transformer);
|
||||
active = output;
|
||||
}
|
||||
await appendArchiveEntry(archive, output, { name: filename.slice(dir ? dir.length + 1 : 0) });
|
||||
await appendArchiveEntry(archive, output, zipEntryOptions(filename.slice(dir ? dir.length + 1 : 0), filename));
|
||||
}
|
||||
if (!stopped) await archive.finalize();
|
||||
})().catch(error => archive.destroy(error));
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { appendArchiveEntry } from "../archive-entry";
|
||||
import { appendArchiveEntry, zipEntryOptions } from "../archive-entry";
|
||||
import {
|
||||
GetObjectCommand,
|
||||
ListObjectsV2CommandOutput,
|
||||
@@ -398,7 +398,7 @@ export default class S3Storage extends StorageBase {
|
||||
output = source.pipe(transformer);
|
||||
active = output;
|
||||
}
|
||||
await appendArchiveEntry(archive, output, { name: filename.slice(dir.length) });
|
||||
await appendArchiveEntry(archive, output, zipEntryOptions(filename.slice(dir.length), filename));
|
||||
}
|
||||
if (listing.IsTruncated && !next) throw new Error("Invalid S3 continuation token");
|
||||
} while (next && !stopped);
|
||||
|
||||
+11
-17
@@ -4,6 +4,7 @@
|
||||
*/
|
||||
export class TermPresenceIndex {
|
||||
private groups: { indices: number[]; characters: string[] | null }[] = [];
|
||||
private searches: { indices: number[]; pattern: RegExp }[];
|
||||
|
||||
constructor(private readonly patterns: RegExp[]) {
|
||||
for (let index = 0; index < patterns.length; index++) {
|
||||
@@ -23,27 +24,20 @@ export class TermPresenceIndex {
|
||||
}
|
||||
this.groups.push(group);
|
||||
}
|
||||
this.searches = this.groups.map(({ indices }) => ({
|
||||
indices,
|
||||
pattern: new RegExp(indices.map(index => `(?:${patterns[index].source})`).join("|"), "iu"),
|
||||
}));
|
||||
}
|
||||
|
||||
candidates(content: string, after = -1): Set<number> {
|
||||
const found = new Set<number>();
|
||||
const remaining = this.groups.map(group => group.indices.filter(index => index > after))
|
||||
.filter(indices => indices.length);
|
||||
let offset = 0;
|
||||
while (remaining.length) {
|
||||
// Fixed-width inputs contain no captures or quantifiers. Removing a
|
||||
// discovered group avoids revisiting millions of identical CSV values.
|
||||
const search = new RegExp(remaining.map(indices =>
|
||||
`(${indices.map(index => this.patterns[index].source).join("|")})`).join("|"), "giu");
|
||||
search.lastIndex = offset;
|
||||
const match = search.exec(content);
|
||||
if (!match) break;
|
||||
const selected = match.slice(1).findIndex(value => value !== undefined);
|
||||
for (const index of remaining[selected]) found.add(index);
|
||||
remaining.splice(selected, 1);
|
||||
// Another group's match may start inside this one. Search from the next
|
||||
// code point, not from the end of the match, to retain nested candidates.
|
||||
offset = match.index + (content.codePointAt(match.index)! > 0xffff ? 2 : 1);
|
||||
// Each immutable group is tested independently, so overlapping matches
|
||||
// cannot hide candidates. Non-global regexes retain no search position.
|
||||
for (const { indices, pattern } of this.searches) {
|
||||
if (indices.some(index => index > after) && pattern.test(content)) {
|
||||
for (const index of indices) if (index > after) found.add(index);
|
||||
}
|
||||
}
|
||||
return found;
|
||||
}
|
||||
|
||||
+87
-57
@@ -1,4 +1,6 @@
|
||||
import got from "got";
|
||||
import config from "../config";
|
||||
import { MAX_SUBMODULE_ENTRIES, SubmoduleMount, safeRepositoryPath, validateMounts } from "./source/github-submodules";
|
||||
import { Readable, Transform, Writable } from "stream";
|
||||
import { Parse } from "unzip-stream";
|
||||
import archiver = require("archiver");
|
||||
@@ -13,11 +15,13 @@ import {
|
||||
} from "./anonymize-utils";
|
||||
import { createLogger, serializeError } from "./logger";
|
||||
import { startStage } from "./request-monitoring";
|
||||
import { zipEntryOptions } from "./archive-entry";
|
||||
|
||||
const logger = createLogger("zip-stream");
|
||||
|
||||
export interface StreamAnonymizedZipOptions {
|
||||
repoId: string;
|
||||
submodules?: SubmoduleMount[];
|
||||
organization: string;
|
||||
repoName: string;
|
||||
commit: string;
|
||||
@@ -77,6 +81,7 @@ export async function streamAnonymizedZip(
|
||||
on(event: string, listener: (...args: unknown[]) => void): unknown;
|
||||
}
|
||||
): Promise<void> {
|
||||
const submodules = validateMounts(opt.submodules);
|
||||
let response;
|
||||
try {
|
||||
const token = await opt.getToken();
|
||||
@@ -112,14 +117,10 @@ export async function streamAnonymizedZip(
|
||||
},
|
||||
});
|
||||
}
|
||||
const upstreamDone = startStage("upstream");
|
||||
const downloadStream = got.stream(response.url);
|
||||
downloadStream.once("end", () => upstreamDone());
|
||||
downloadStream.once("error", () => upstreamDone(true));
|
||||
downloadStream.once("close", () => upstreamDone(!downloadStream.readableEnded));
|
||||
|
||||
let downloadStream: Readable;
|
||||
const archive = archiver("zip", {});
|
||||
const parser = Parse() as Transform;
|
||||
let parser: Transform;
|
||||
let childEntries = 0;
|
||||
const activeStreams = new Set<Readable>();
|
||||
const compiledTerms = compileTerms(opt.anonymizerOptions.terms || []);
|
||||
let stopped = false;
|
||||
@@ -159,8 +160,8 @@ export async function streamAnonymizedZip(
|
||||
stopped = true;
|
||||
cleanupDrain?.();
|
||||
cleanupDrain = undefined;
|
||||
downloadStream.destroy();
|
||||
parser.destroy();
|
||||
downloadStream?.destroy();
|
||||
parser?.destroy();
|
||||
for (const stream of activeStreams) stream.destroy();
|
||||
activeStreams.clear();
|
||||
output.destroy();
|
||||
@@ -186,53 +187,82 @@ export async function streamAnonymizedZip(
|
||||
archive.on("error", fail);
|
||||
archive.pipe(output);
|
||||
|
||||
downloadStream
|
||||
.on("error", fail)
|
||||
.pipe(parser)
|
||||
.on("entry", (entry: Readable & { type: string; path: string; autodrain: () => void }) => {
|
||||
if (stopped) {
|
||||
entry.destroy();
|
||||
return;
|
||||
}
|
||||
entry.on("error", fail);
|
||||
activeStreams.add(entry);
|
||||
entry.once("close", () => activeStreams.delete(entry));
|
||||
if (entry.type === "File") {
|
||||
try {
|
||||
const fileName = anonymizePathCompiled(
|
||||
entry.path.substring(entry.path.indexOf("/") + 1),
|
||||
compiledTerms
|
||||
);
|
||||
if (!isEntryAllowed(entry.path, opt.contentOptions)) {
|
||||
entry.autodrain();
|
||||
return;
|
||||
}
|
||||
// Pass filePath via the constructor — AnonymizeTransformer reads it
|
||||
// there to decide whether the entry is text (and therefore should be
|
||||
// anonymized) vs binary (passthrough). Assigning afterwards leaves
|
||||
// isText=false for every file, so the zip ships unanonymized.
|
||||
parser.pause();
|
||||
const onArchived = () => { entryWaiting = true; setImmediate(resumeParser); };
|
||||
archive.once("entry", onArchived);
|
||||
const anonymizer = new AnonymizeTransformer({
|
||||
...opt.anonymizerOptions,
|
||||
filePath: entry.path,
|
||||
});
|
||||
anonymizer.on("error", fail);
|
||||
activeStreams.add(anonymizer);
|
||||
anonymizer.once("close", () => activeStreams.delete(anonymizer));
|
||||
const st = entry.pipe(anonymizer);
|
||||
archive.append(st, { name: fileName });
|
||||
} catch (error) {
|
||||
fail(error as Error);
|
||||
const startArchive = (url: string, prefix: string, next: number) => {
|
||||
if (stopped) return;
|
||||
const upstreamDone = startStage("upstream");
|
||||
const input = got.stream(url);
|
||||
downloadStream = input;
|
||||
input.once("end", () => upstreamDone());
|
||||
input.once("error", () => upstreamDone(true));
|
||||
input.once("close", () => upstreamDone(!input.readableEnded));
|
||||
parser = Parse() as Transform;
|
||||
downloadStream
|
||||
.on("error", fail)
|
||||
.pipe(parser)
|
||||
.on("entry", (entry: Readable & { type: string; path: string; autodrain: () => void }) => {
|
||||
if (stopped) {
|
||||
entry.destroy();
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
entry.autodrain();
|
||||
}
|
||||
})
|
||||
.on("error", fail)
|
||||
.on("end", () => {
|
||||
if (stopped) return;
|
||||
void archive.finalize().catch(fail);
|
||||
});
|
||||
entry.on("error", fail);
|
||||
activeStreams.add(entry);
|
||||
entry.once("close", () => activeStreams.delete(entry));
|
||||
if (entry.type === "File") {
|
||||
try {
|
||||
const relativePath = entry.path.substring(entry.path.indexOf("/") + 1);
|
||||
if (!safeRepositoryPath(relativePath)) throw new AnonymousError("invalid_file_path", { httpStatus: 502 });
|
||||
if (relativePath.split("/").pop() === ".gitmodules") { entry.autodrain(); return; }
|
||||
if (prefix && ++childEntries > MAX_SUBMODULE_ENTRIES) throw new AnonymousError("file_too_big", { httpStatus: 413 });
|
||||
const mountedPath = prefix ? `${prefix}/${relativePath}` : relativePath;
|
||||
const fileName = anonymizePathCompiled(mountedPath, compiledTerms);
|
||||
if (!isEntryAllowed(entry.path, opt.contentOptions)) {
|
||||
entry.autodrain();
|
||||
return;
|
||||
}
|
||||
// Pass filePath via the constructor — AnonymizeTransformer reads it
|
||||
// there to decide whether the entry is text (and therefore should be
|
||||
// anonymized) vs binary (passthrough). Assigning afterwards leaves
|
||||
// isText=false for every file, so the zip ships unanonymized.
|
||||
parser.pause();
|
||||
const onArchived = () => { entryWaiting = true; setImmediate(resumeParser); };
|
||||
archive.once("entry", onArchived);
|
||||
const anonymizer = new AnonymizeTransformer({
|
||||
...opt.anonymizerOptions,
|
||||
filePath: mountedPath,
|
||||
});
|
||||
anonymizer.on("error", fail);
|
||||
activeStreams.add(anonymizer);
|
||||
anonymizer.once("close", () => activeStreams.delete(anonymizer));
|
||||
let source: Readable = entry;
|
||||
if (prefix) {
|
||||
let bytes = 0;
|
||||
const limited = new Transform({ transform(chunk, encoding, callback) {
|
||||
bytes += chunk.length;
|
||||
callback(bytes > config.MAX_FILE_SIZE ? new AnonymousError("file_too_big", { httpStatus: 413 }) : null, chunk);
|
||||
} });
|
||||
limited.on("error", fail);
|
||||
activeStreams.add(limited);
|
||||
limited.once("close", () => activeStreams.delete(limited));
|
||||
source = entry.pipe(limited);
|
||||
}
|
||||
const st = source.pipe(anonymizer);
|
||||
archive.append(st, zipEntryOptions(fileName, mountedPath));
|
||||
} catch (error) {
|
||||
fail(error as Error);
|
||||
}
|
||||
} else {
|
||||
entry.autodrain();
|
||||
}
|
||||
})
|
||||
.on("error", fail)
|
||||
.on("end", () => {
|
||||
if (stopped) return;
|
||||
const mount = submodules[next];
|
||||
if (mount) {
|
||||
// Public child archives never receive parent credentials.
|
||||
startArchive(`https://codeload.github.com/${encodeURIComponent(mount.organization)}/${encodeURIComponent(mount.repoName)}/zip/${mount.commit}`, mount.path, next + 1);
|
||||
} else void archive.finalize().catch(fail);
|
||||
});
|
||||
};
|
||||
startArchive(response.url, "", 0);
|
||||
}
|
||||
Reference in new issue
Block a user