Pushes can carry large files; the Live test waits for a free import slot (#120)
A push over 16 MB is staged in R2 as it arrives, and blobs of 4 MiB or more are hashed, secret-scanned and stored from their compressed pack bytes, so a 70 MB file no longer has to fit in memory several times over. The Live test's import check waits up to 15 minutes for an import slot instead of skipping while a campaign holds them. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AizKXSsengXZRegJxRATkL
Canberk Demirci authored · GitHub committed signed1 parent d228433commit 4d87efb1c1524cc15b1eaa5432e49546cb582bceBrowse files
12 files changed+356−17
Files (12)
- .github/workflows/live-test.yml+15−2
- apps/diffinery/docs/reference/limits.md+1−0
- apps/diffinery/src/git/secret-scan.ts+45−2
- apps/diffinery/src/http/git.ts+57−1
- apps/diffinery/src/lib/storage-keys.ts+9−1
- apps/diffinery/src/repo/RepoDO.ts+37−6
- apps/diffinery/src/repo/objects.ts+28−0
- apps/diffinery/test/e2e/e2e.test.mjs+25−0
- apps/diffinery/test/unit/secret-scan.test.ts+25−0
- packages/git-core/src/pack.ts+45−5
- packages/git-core/src/zlib.ts+27−0
- packages/git-core/test/pack.test.ts+42−0
modified.github/workflows/live-test.yml+15−2View file
| @@ -542,15 +542,28 @@ | ||
| 542 | 542 | if [ -n "${GITHUB_TOKEN_FOR_IMPORT:-}" ]; then echo "importing with IMPORT_GITHUB_TOKEN"; else echo "::warning::IMPORT_GITHUB_TOKEN is not set, so the import runs token-free"; fi |
| 543 | 543 | # Token-free, GitHub allows 60 requests an hour per address: a rate limit skips this check with a warning. |
| 544 | 544 | # Starting the import reads the repository from GitHub; a GitHub outage (502 "could not reach GitHub") |
| 545 | - # is retried, and skips this check with a warning if it outlasts the retries. | |
| 546 | - for try in 1 2 3 4; do | |
| 545 | + # is retried, and skips this check with a warning if it outlasts the retries. While other imports hold | |
| 546 | + # every slot (an import campaign), it waits up to 15 minutes for one, so a busy account doesn't hide this check. | |
| 547 | + tries=0 waits=0 | |
| 548 | + while :; do | |
| 547 | 549 | code=$(curl -sS -o import.json -w '%{http_code}' -X POST "$DIFFINERY/api/v3/diffinery/imports" -H "Authorization: Bearer $TOKEN" -H "content-type: application/json" \ |
| 548 | 550 | --data-binary @- <<<"$body" || true) |
| 549 | 551 | unreachable=$(jq -r '.message // "" | startswith("could not reach GitHub")' import.json 2>/dev/null || echo false) |
| 552 | + busy=$(jq -r '.message // "" | startswith("too many repository imports are already running")' import.json 2>/dev/null || echo false) | |
| 553 | + if [ "$code" = 429 ] && [ "$busy" = true ]; then | |
| 554 | + waits=$((waits + 1)) | |
| 555 | + [ "$waits" -le 30 ] || break | |
| 556 | + [ "$waits" -gt 1 ] || echo "every import slot is taken ($(jq -r .message import.json)); waiting for one, up to 15 minutes" | |
| 557 | + sleep 30 | |
| 558 | + continue | |
| 559 | + fi | |
| 550 | 560 | { [ "$code" = 502 ] && [ "$unreachable" = true ]; } || [ "$code" = 000 ] || break |
| 561 | + tries=$((tries + 1)) | |
| 562 | + [ "$tries" -lt 4 ] || break | |
| 551 | 563 | echo "starting the import returned $code ($(jq -r .message import.json 2>/dev/null)); trying again in 30 seconds" |
| 552 | 564 | sleep 30 |
| 553 | 565 | done |
| 566 | + if [ "$code" = 429 ] && [ "$busy" = true ]; then echo "::warning::no import slot came free in 15 minutes, so the import check is skipped: $(jq -r .message import.json)"; exit 0; fi | |
| 554 | 567 | if [ "$code" = 429 ]; then echo "::warning::GitHub's rate limit stopped the import check: $(jq -r .message import.json)"; exit 0; fi |
| 555 | 568 | if [ "$code" = 502 ] && [ "$unreachable" = true ]; then echo "::warning::GitHub could not be reached, so the import check is skipped: $(jq -r .message import.json)"; exit 0; fi |
| 556 | 569 | [ "$code" = 201 ] || { echo "::error::starting the import returned $code: $(cat import.json)"; exit 1; } |
modifiedapps/diffinery/docs/reference/limits.md+1−0View file
| @@ -125,6 +125,7 @@ | ||
| 125 | 125 | | --- | --- | --- | |
| 126 | 126 | | Push size | 100 MB per push | The whole push request after decompression. A larger push is rejected with `push exceeds 100 MB; split it into smaller pushes`. Push in several steps (for example a range of commits at a time). | |
| 127 | 127 | | Fetch request size | 16 MB | The request a client sends to fetch or clone (its wants and haves), after decompression. A larger one is refused with `413 fetch request exceeds 16 MB`. | |
| 128 | +| File size in a push | Up to the push size | A single file (blob) can be as large as fits in one push (GitHub allows files up to 100 MB). Files of 4 MiB or more are checked and stored from their compressed bytes, so they never need to fit in memory inflated. | | |
| 128 | 129 | | Unpacked size of a push | 512 MiB | What a push's pack inflates to, deltas resolved. A pack that unpacks to more is refused, however small it is on the wire. | |
| 129 | 130 | | Git protocol | Version 2 for fetch and clone | Clients older than git 2.26 must run `git config --global protocol.version 2`, or the server answers `Diffinery requires Git protocol v2`. | |
| 130 | 131 | | Secret scanning coverage | Every new text file | Push protection scans every new text file in a push, whatever its size. The rejection message lists up to 10 findings. | |
modifiedapps/diffinery/src/git/secret-scan.ts+45−2View file
| @@ -1,4 +1,4 @@ | ||
| 1 | -import { fromUtf8 } from "@diffinery/git-core"; | |
| 1 | +import { fromUtf8, inflateEach } from "@diffinery/git-core"; | |
| 2 | 2 | |
| 3 | 3 | /** High-confidence credential patterns. Push protection blocks pushes that add these. */ |
| 4 | 4 | const PATTERNS: Array<{ kind: string; re: RegExp }> = [ |
| @@ -33,9 +33,13 @@ | ||
| 33 | 33 | * too). A file marked `diffinery:allow-secret` is let through, but what it holds goes to `marked` |
| 34 | 34 | * so the caller can put the opt-out on the audit record. |
| 35 | 35 | */ |
| 36 | -export function scanForSecrets(blobs: Array<{ oid: string; content: Uint8Array }>, marked: SecretHit[] = []): SecretHit[] { | |
| 36 | +export function scanForSecrets(blobs: Array<{ oid: string; content: Uint8Array; deflated?: Uint8Array }>, marked: SecretHit[] = []): SecretHit[] { | |
| 37 | 37 | const hits: SecretHit[] = []; |
| 38 | 38 | for (const b of blobs) { |
| 39 | + if (b.deflated) { | |
| 40 | + scanDeflated(b.oid, b.deflated, hits, marked); | |
| 41 | + continue; | |
| 42 | + } | |
| 39 | 43 | const text = textOf(b.content); |
| 40 | 44 | if (text === null) continue; |
| 41 | 45 | const out = /diffinery:allow-secret/.test(text) ? marked : hits; |
| @@ -44,6 +48,45 @@ | ||
| 44 | 48 | return hits; |
| 45 | 49 | } |
| 46 | 50 | |
| 51 | +/** Carried from one slice of a large blob to the next, longer than any pattern can match. */ | |
| 52 | +const OVERLAP = 1024; | |
| 53 | + | |
| 54 | +/** | |
| 55 | + * `scanForSecrets` for a large blob kept compressed: it is inflated and decoded slice by slice, never | |
| 56 | + * whole, with the same rules (binary when it has a NUL byte, unless it starts with a UTF-16 byte | |
| 57 | + * order mark; `diffinery:allow-secret` anywhere moves its findings to `marked`). | |
| 58 | + */ | |
| 59 | +function scanDeflated(oid: string, deflated: Uint8Array, hits: SecretHit[], marked: SecretHit[]): void { | |
| 60 | + const found = new Set<string>(); | |
| 61 | + let decoder: TextDecoder | null = null; | |
| 62 | + let utf8 = true; | |
| 63 | + let binary = false; | |
| 64 | + let allowed = false; | |
| 65 | + let tail = ""; | |
| 66 | + const scan = (text: string) => { | |
| 67 | + const window = tail + text; | |
| 68 | + if (/diffinery:allow-secret/.test(window)) allowed = true; | |
| 69 | + for (const p of PATTERNS) if (p.re.test(window)) found.add(p.kind); | |
| 70 | + tail = window.slice(-OVERLAP); | |
| 71 | + }; | |
| 72 | + inflateEach(deflated, 0, (chunk) => { | |
| 73 | + if (binary) return; | |
| 74 | + if (!decoder) { | |
| 75 | + const bom = chunk[0] === 0xff && chunk[1] === 0xfe ? "utf-16le" : chunk[0] === 0xfe && chunk[1] === 0xff ? "utf-16be" : null; | |
| 76 | + utf8 = !bom; | |
| 77 | + decoder = new TextDecoder(bom ?? "utf-8"); | |
| 78 | + } | |
| 79 | + if (utf8 && chunk.includes(0)) binary = true; | |
| 80 | + else scan(decoder.decode(chunk, { stream: true })); | |
| 81 | + }); | |
| 82 | + // TypeScript can't see the assignment inside the callback. | |
| 83 | + const last = decoder as TextDecoder | null; | |
| 84 | + if (binary) return; | |
| 85 | + if (last) scan(last.decode()); | |
| 86 | + const out = allowed ? marked : hits; | |
| 87 | + for (const p of PATTERNS) if (found.has(p.kind)) out.push({ oid, kind: p.kind }); | |
| 88 | +} | |
| 89 | + | |
| 47 | 90 | /** The reason given with `git push -o allow-secrets=<reason>`, if any. */ |
| 48 | 91 | export function secretBypassReason(pushOptions: string[]): string | null { |
| 49 | 92 | for (const o of pushOptions) { |
modifiedapps/diffinery/src/http/git.ts+57−1View file
| @@ -2,6 +2,8 @@ | ||
| 2 | 2 | import type { Env } from "../env.js"; |
| 3 | 3 | import type { AppVars } from "../auth/principal.js"; |
| 4 | 4 | import { HttpError } from "../lib/util.js"; |
| 5 | +import { BodyTooLarge, putStream, type BlobBucket } from "../lib/blob-store.js"; | |
| 6 | +import { pushBodyKey, type SpilledBody } from "../lib/storage-keys.js"; | |
| 5 | 7 | import { importPushAllowed } from "../imports/service.js"; |
| 6 | 8 | import { loadRepo } from "./common.js"; |
| 7 | 9 | |
| @@ -45,7 +47,7 @@ | ||
| 45 | 47 | // A GitHub import's own token, while that import is running and filling this repository, may push in import mode. |
| 46 | 48 | const p = c.get("principal"); |
| 47 | 49 | if (p.type === "account" && p.importId && p.tokenId && (await importPushAllowed(c.env, p.importId, repo.id, p.tokenId))) caller.importId = p.importId; |
| 48 | - const body = await readBody(c.req.raw, MAX_RECEIVE_PACK_REQUEST, "push exceeds 100 MB; split it into smaller pushes"); | |
| 50 | + const body = await readPushBody(c.env, repo.id, c.req.raw); | |
| 49 | 51 | const out = await stub.gitReceivePack(body, caller, { moved }); |
| 50 | 52 | c.executionCtx.waitUntil(c.get("dir").touchPushed(repo.id)); |
| 51 | 53 | return new Response(out, { headers: { "content-type": "application/x-git-receive-pack-result", ...noCache } }); |
| @@ -86,6 +88,60 @@ | ||
| 86 | 88 | return out; |
| 87 | 89 | } |
| 88 | 90 | |
| 91 | +/** Pushes up to this size are handed to the repository in memory; larger ones go through R2. */ | |
| 92 | +const IN_MEMORY_PUSH = 16 * 1024 * 1024; | |
| 93 | + | |
| 94 | +/** | |
| 95 | + * A push's body. A large one is staged in R2 as it arrives (a Worker and a Durable Object each | |
| 96 | + * have 128 MB of memory, and holding a 70 MB push twice, then copying it to the repository, does | |
| 97 | + * not fit); the repository reads it from there once and deletes it. | |
| 98 | + */ | |
| 99 | +async function readPushBody(env: Env, repoId: string, req: Request): Promise<Uint8Array | SpilledBody> { | |
| 100 | + const tooLarge = "push exceeds 100 MB; split it into smaller pushes"; | |
| 101 | + let stream = req.body; | |
| 102 | + if (!stream) return new Uint8Array(0); | |
| 103 | + if (req.headers.get("content-encoding") === "gzip") stream = stream.pipeThrough(new DecompressionStream("gzip")); | |
| 104 | + const chunks: Uint8Array[] = []; | |
| 105 | + let total = 0; | |
| 106 | + const reader = stream.getReader(); | |
| 107 | + for (;;) { | |
| 108 | + const { done, value } = await reader.read(); | |
| 109 | + if (done) break; | |
| 110 | + total += value.length; | |
| 111 | + chunks.push(value); | |
| 112 | + if (total > IN_MEMORY_PUSH) { | |
| 113 | + // Too big to pass along in memory: what came so far, then the rest, goes to R2. | |
| 114 | + const key = pushBodyKey(repoId); | |
| 115 | + const rest = new ReadableStream<Uint8Array>({ | |
| 116 | + start(controller) { | |
| 117 | + for (const ch of chunks.splice(0)) controller.enqueue(ch); | |
| 118 | + }, | |
| 119 | + async pull(controller) { | |
| 120 | + const next = await reader.read(); | |
| 121 | + if (next.done) controller.close(); | |
| 122 | + else controller.enqueue(next.value); | |
| 123 | + }, | |
| 124 | + cancel: () => reader.cancel(), | |
| 125 | + }); | |
| 126 | + try { | |
| 127 | + const size = await putStream(env.BUCKET as unknown as BlobBucket, key, rest, MAX_RECEIVE_PACK_REQUEST); | |
| 128 | + return { spilled: key, size }; | |
| 129 | + } catch (e) { | |
| 130 | + await env.BUCKET.delete(key).catch(() => {}); | |
| 131 | + if (e instanceof BodyTooLarge) throw new HttpError(413, tooLarge); | |
| 132 | + throw e; | |
| 133 | + } | |
| 134 | + } | |
| 135 | + } | |
| 136 | + const out = new Uint8Array(total); | |
| 137 | + let at = 0; | |
| 138 | + for (const ch of chunks) { | |
| 139 | + out.set(ch, at); | |
| 140 | + at += ch.length; | |
| 141 | + } | |
| 142 | + return out; | |
| 143 | +} | |
| 144 | + | |
| 89 | 145 | // ---------------------------------------------------------------- Git LFS |
| 90 | 146 | |
| 91 | 147 | const lfsKey = (repoId: string, oid: string) => `repos/${repoId}/lfs/${oid.slice(0, 2)}/${oid.slice(2, 4)}/${oid}`; |
modifiedapps/diffinery/src/lib/storage-keys.ts+9−1View file
| @@ -10,7 +10,7 @@ | ||
| 10 | 10 | * | `artifacts/` | workflow artifacts | deleted after 90 days | |
| 11 | 11 | * | `actions-cache/` | Actions caches | none: evicted by the code | |
| 12 | 12 | * | `logs/` | archived job logs | deleted after 90 days | |
| 13 | - * | `actions-staging/` | uploads in progress (blocks, chunks) | deleted after 1 day | | |
| 13 | + * | `actions-staging/` | uploads in progress (blocks, chunks, pushes) | deleted after 1 day | | |
| 14 | 14 | * | `archives/` | zipball/tarball cache (rebuilt on a miss) | deleted after 30 days | |
| 15 | 15 | * | `events/` | the event archive | deleted after 400 days | |
| 16 | 16 | * |
| @@ -26,6 +26,14 @@ | ||
| 26 | 26 | |
| 27 | 27 | export const STAGING_ROOT = "actions-staging/"; |
| 28 | 28 | |
| 29 | +/** A push body staged in R2 (see {@link pushBodyKey}). */ | |
| 30 | +export type SpilledBody = { spilled: string; size: number }; | |
| 31 | + | |
| 32 | +/** A large push body on its way from the Worker to the repository's Durable Object (deleted once read). */ | |
| 33 | +export function pushBodyKey(repoId: string): string { | |
| 34 | + return `${STAGING_ROOT}pushes/${repoId}/${crypto.randomUUID()}`; | |
| 35 | +} | |
| 36 | + | |
| 29 | 37 | /** The object of an Actions runtime cache entry or artifact, by its numeric id. */ |
| 30 | 38 | export function runtimeObjectKey(b: { repo: string; kind: "cache" | "artifact"; id: number }): string { |
| 31 | 39 | return b.kind === "cache" ? `actions-cache/${b.repo}/${b.id}` : `artifacts/${b.repo}/${b.id}`; |
modifiedapps/diffinery/src/repo/RepoDO.ts+37−6View file
| @@ -46,7 +46,7 @@ | ||
| 46 | 46 | import { fsckObjects } from "../git/fsck.js"; |
| 47 | 47 | import { scanForSecrets, secretBypassReason, type SecretHit } from "../git/secret-scan.js"; |
| 48 | 48 | import { HttpError, isValidRefName, now } from "../lib/util.js"; |
| 49 | -import { archivePrefix, deletePrefixes, repoPrefixes } from "../lib/storage-keys.js"; | |
| 49 | +import { archivePrefix, deletePrefixes, repoPrefixes, type SpilledBody } from "../lib/storage-keys.js"; | |
| 50 | 50 | import { Actions } from "./actions.js"; |
| 51 | 51 | import { AutoMerge, isPerson } from "./auto-merge.js"; |
| 52 | 52 | import { Collab } from "./collab.js"; |
| @@ -67,6 +67,8 @@ | ||
| 67 | 67 | import { atLeast, globToRegExp, type Caller, type ProtectionRules, type RepoMeta } from "./types.js"; |
| 68 | 68 | |
| 69 | 69 | const MAX_PUSH_BYTES = 100 * 1024 * 1024; |
| 70 | +/** Pushed blobs this large are hashed, scanned and stored from their compressed bytes, never held inflated. */ | |
| 71 | +const LARGE_OBJECT_BYTES = 4 * 1024 * 1024; | |
| 70 | 72 | /** A purge walks every reachable commit and tree while holding the repository: bounded so it finishes in time. */ |
| 71 | 73 | export const MAX_PURGE_OBJECTS = 300_000; |
| 72 | 74 | /** Storage key that marks a deleted repository whose R2 objects are still being deleted. */ |
| @@ -374,12 +376,37 @@ | ||
| 374 | 376 | return readable; |
| 375 | 377 | } |
| 376 | 378 | |
| 379 | + /** A push the Worker staged in R2, read into one buffer of its exact size (and then deleted). */ | |
| 380 | + private async readSpilledPush(b: SpilledBody): Promise<Uint8Array> { | |
| 381 | + try { | |
| 382 | + const obj = await this.env.BUCKET.get(b.spilled); | |
| 383 | + if (!obj) throw new Error("the push's data is missing; push again"); | |
| 384 | + const out = new Uint8Array(obj.size); | |
| 385 | + let at = 0; | |
| 386 | + const reader = obj.body.getReader(); | |
| 387 | + for (;;) { | |
| 388 | + const { done, value } = await reader.read(); | |
| 389 | + if (done) break; | |
| 390 | + if (at + value.length > out.length) throw new Error("the push's data is longer than announced; push again"); | |
| 391 | + out.set(value, at); | |
| 392 | + at += value.length; | |
| 393 | + } | |
| 394 | + if (at !== out.length) throw new Error("the push's data is shorter than announced; push again"); | |
| 395 | + return out; | |
| 396 | + } finally { | |
| 397 | + await this.env.BUCKET.delete(b.spilled).catch(() => {}); | |
| 398 | + } | |
| 399 | + } | |
| 400 | + | |
| 377 | 401 | /** `moved`: the push came in under an old name (a transfer or rename), so say where the repository lives now, like GitHub. */ |
| 378 | - async gitReceivePack(body: Uint8Array, caller: Caller, o: { moved?: boolean } = {}): Promise<Uint8Array> { | |
| 379 | - if (body.length > MAX_PUSH_BYTES) { | |
| 402 | + async gitReceivePack(input: Uint8Array | SpilledBody, caller: Caller, o: { moved?: boolean } = {}): Promise<Uint8Array> { | |
| 403 | + const size = input instanceof Uint8Array ? input.length : input.size; | |
| 404 | + if (size > MAX_PUSH_BYTES) { | |
| 405 | + if (!(input instanceof Uint8Array)) await this.env.BUCKET.delete(input.spilled).catch(() => {}); | |
| 380 | 406 | const req = { commands: [], capabilities: new Set<string>(), pushOptions: [], pack: null }; |
| 381 | 407 | return reportStatus(req, "push exceeds 100 MB; split it into smaller pushes", []); |
| 382 | 408 | } |
| 409 | + const body = input instanceof Uint8Array ? input : await this.readSpilledPush(input); | |
| 383 | 410 | const req = parseReceivePack(body); |
| 384 | 411 | // A purge of unreachable objects that ran while this push was in flight may have deleted objects |
| 385 | 412 | // the push counted on the repository having: refuse rather than point a ref at them. |
| @@ -401,13 +428,14 @@ | ||
| 401 | 428 | let objects: Awaited<ReturnType<typeof parsePack>> = []; |
| 402 | 429 | if (req.pack && req.pack.length > 0) { |
| 403 | 430 | try { |
| 404 | - objects = await parsePack(req.pack, (oid) => this.store.read(oid)); | |
| 431 | + objects = await parsePack(req.pack, (oid) => this.store.read(oid), { largeObjectBytes: LARGE_OBJECT_BYTES }); | |
| 405 | 432 | } catch (e) { |
| 406 | 433 | unpackError = (e as Error).message; |
| 407 | 434 | } |
| 408 | 435 | } |
| 409 | 436 | // Commit and tag messages are scanned too, not only files. |
| 410 | - const newBlobs = objects.filter((o) => o.type === "blob" || o.type === "commit" || o.type === "tag").map((o) => ({ oid: o.oid, content: o.content })); | |
| 437 | + // The objects themselves, not copies: a large blob is scanned from its compressed bytes. | |
| 438 | + const newBlobs = objects.filter((o) => o.type === "blob" || o.type === "commit" || o.type === "tag"); | |
| 411 | 439 | const results: RefResult[] = []; |
| 412 | 440 | if (unpackError) { |
| 413 | 441 | for (const c of req.commands) results.push({ ref: c.ref, ok: false, reason: "unpacker error" }); |
| @@ -447,7 +475,10 @@ | ||
| 447 | 475 | } |
| 448 | 476 | |
| 449 | 477 | try { |
| 450 | - for (const o of objects) await this.store.writeKnown(o.oid, o.type, o.content); | |
| 478 | + for (const o of objects) { | |
| 479 | + if (o.deflated) await this.store.writeDeflated(o.oid, o.type, o.size!, o.deflated); | |
| 480 | + else await this.store.writeKnown(o.oid, o.type, o.content); | |
| 481 | + } | |
| 451 | 482 | } catch (e) { |
| 452 | 483 | for (const c of req.commands) results.push({ ref: c.ref, ok: false, reason: "unpacker error" }); |
| 453 | 484 | return reportStatus(req, (e as Error).message, results); |
modifiedapps/diffinery/src/repo/objects.ts+28−0View file
| @@ -1,4 +1,5 @@ | ||
| 1 | 1 | import { CODE_TYPES, TYPE_CODES, deflate, hashObject, inflate, type GitObject, type ObjectStore, type ObjectType } from "@diffinery/git-core"; |
| 2 | +import { sized } from "../lib/blob-store.js"; | |
| 2 | 3 | |
| 3 | 4 | /** Objects larger than this (compressed) go to R2 instead of SQLite (rows are capped at 2 MB). */ |
| 4 | 5 | const INLINE_LIMIT = 512 * 1024; |
| @@ -102,6 +103,20 @@ | ||
| 102 | 103 | this.remember(oid, { type, content }); |
| 103 | 104 | } |
| 104 | 105 | |
| 106 | + /** | |
| 107 | + * Store an object from zlib data that is already in hand (a large blob as it sits in a pushed | |
| 108 | + * pack), without inflating or recompressing it. The data goes to R2 in slices, never copied whole. | |
| 109 | + */ | |
| 110 | + async writeDeflated(oid: string, type: ObjectType, size: number, deflated: Uint8Array): Promise<void> { | |
| 111 | + if (this.has(oid)) return; | |
| 112 | + if (deflated.length > INLINE_LIMIT) { | |
| 113 | + await this.bucket.put(this.r2Key(oid), sized(slices(deflated), deflated.length)); | |
| 114 | + this.sql.exec("INSERT OR IGNORE INTO objects (oid, type, size, data, r2) VALUES (?, ?, ?, NULL, 1)", oid, TYPE_CODES[type], size); | |
| 115 | + } else { | |
| 116 | + this.sql.exec("INSERT OR IGNORE INTO objects (oid, type, size, data, r2) VALUES (?, ?, ?, ?, 0)", oid, TYPE_CODES[type], size, deflated.slice()); | |
| 117 | + } | |
| 118 | + } | |
| 119 | + | |
| 105 | 120 | count(): number { |
| 106 | 121 | return this.sql.exec<{ n: number }>("SELECT count(*) AS n FROM objects").one().n; |
| 107 | 122 | } |
| @@ -111,3 +126,16 @@ | ||
| 111 | 126 | return this.sql.exec<{ n: number | null }>("SELECT sum(length(data)) AS n FROM objects").one().n ?? 0; |
| 112 | 127 | } |
| 113 | 128 | } |
| 129 | + | |
| 130 | +/** `data` as a stream of 1 MiB views. */ | |
| 131 | +function slices(data: Uint8Array): ReadableStream<Uint8Array> { | |
| 132 | + let at = 0; | |
| 133 | + return new ReadableStream<Uint8Array>({ | |
| 134 | + pull(controller) { | |
| 135 | + if (at >= data.length) return controller.close(); | |
| 136 | + const end = Math.min(data.length, at + 1024 * 1024); | |
| 137 | + controller.enqueue(data.subarray(at, end)); | |
| 138 | + at = end; | |
| 139 | + }, | |
| 140 | + }); | |
| 141 | +} |
modifiedapps/diffinery/test/e2e/e2e.test.mjs+25−0View file
| @@ -2659,3 +2659,28 @@ | ||
| 2659 | 2659 | git(src, ["commit", "-q", "-am", "no schedule"]); |
| 2660 | 2660 | git(src, ["push", "-q", `${base}/${tickRepo}.git`, "main"]); |
| 2661 | 2661 | }); |
| 2662 | + | |
| 2663 | +test("large files: a 40 MB file is pushed and cloned intact, and push protection still reads big text files", async () => { | |
| 2664 | + const me = await ok("GET", "/user"); | |
| 2665 | + await ok("POST", "/user/repos", { name: "bigfile" }); | |
| 2666 | + const src = join(work, "bigfile"); | |
| 2667 | + mkdirSync(src, { recursive: true }); | |
| 2668 | + git(src, ["init", "-q", "-b", "main"]); | |
| 2669 | + // Random bytes don't compress: the push is over 16 MB, so it goes through R2 on its way in. | |
| 2670 | + const data = randomBytes(40 * 1024 * 1024); | |
| 2671 | + writeFileSync(join(src, "big.bin"), data); | |
| 2672 | + git(src, ["add", "-A"]); | |
| 2673 | + git(src, ["commit", "-q", "-m", "a large file"]); | |
| 2674 | + git(src, ["-c", "http.postBuffer=1048576", "push", "-q", `${base}/${me.login}/bigfile.git`, "main"], token, { maxBuffer: 16 * 1024 * 1024 }); | |
| 2675 | + const out = join(work, "bigfile-clone"); | |
| 2676 | + git(work, ["clone", "-q", `${base}/${me.login}/bigfile.git`, out]); | |
| 2677 | + assert.ok(readFileSync(join(out, "big.bin")).equals(data), "the cloned file is the pushed file"); | |
| 2678 | + const wantBlob = git(src, ["rev-parse", "HEAD:big.bin"]).trim(); | |
| 2679 | + assert.equal(git(out, ["rev-parse", "HEAD:big.bin"]).trim(), wantBlob); | |
| 2680 | + // A large text file (scanned slice by slice) with a key near its end is still refused. | |
| 2681 | + // diffinery:allow-secret (a fake key) | |
| 2682 | + writeFileSync(join(src, "dump.sql"), `${"-- row\n".repeat(1024 * 1024)}aws_key = AKIAIOSFODNN7EXAMPL3\n`); | |
| 2683 | + git(src, ["add", "-A"]); | |
| 2684 | + git(src, ["commit", "-q", "-m", "a large dump"]); | |
| 2685 | + assert.match(gitFails(src, ["push", "-q", `${base}/${me.login}/bigfile.git`, "main"]), /possible secrets detected/); | |
| 2686 | +}); |
modifiedapps/diffinery/test/unit/secret-scan.test.ts+25−0View file
| @@ -1,4 +1,5 @@ | ||
| 1 | 1 | import { describe, expect, it } from "vitest"; |
| 2 | +import { deflate } from "@diffinery/git-core"; | |
| 2 | 3 | import { scanForSecrets, secretBypassReason } from "../../src/git/secret-scan.js"; |
| 3 | 4 | |
| 4 | 5 | const blob = (text: string) => ({ oid: "a".repeat(40), content: new TextEncoder().encode(text) }); |
| @@ -38,6 +39,30 @@ | ||
| 38 | 39 | expect(scanForSecrets([{ oid: "b".repeat(40), content: new Uint8Array([0x41, 0, 0x41]) }])).toEqual([]); |
| 39 | 40 | }); |
| 40 | 41 | |
| 42 | + it("scans large blobs from their compressed bytes, slice by slice, with the same rules", () => { | |
| 43 | + // diffinery:allow-secret (fake keys) | |
| 44 | + const filler = "x".repeat(256 * 1024 - 10); | |
| 45 | + const big = (text: string) => { | |
| 46 | + const content = new TextEncoder().encode(text); | |
| 47 | + return { oid: "c".repeat(40), content: new Uint8Array(0), deflated: deflate(content) }; | |
| 48 | + }; | |
| 49 | + // A key that straddles two inflated slices is still found. | |
| 50 | + expect(scanForSecrets([big(`${filler}\nAKIAIOSFODNN7EXAMPL3\n${filler}`)]).map((h) => h.kind)).toEqual(["AWS access key"]); | |
| 51 | + expect(scanForSecrets([big(`${filler}${filler}${filler}`)])).toEqual([]); | |
| 52 | + // The opt-out anywhere in the file moves its findings to the record. | |
| 53 | + const marked: { oid: string; kind: string }[] = []; | |
| 54 | + expect(scanForSecrets([big(`AKIAIOSFODNN7EXAMPL3 ${filler}${filler} diffinery:allow-secret`)], marked)).toEqual([]); | |
| 55 | + expect(marked.map((h) => h.kind)).toEqual(["AWS access key"]); | |
| 56 | + // A NUL byte anywhere makes it binary, as for a small blob. | |
| 57 | + expect(scanForSecrets([big(`AKIAIOSFODNN7EXAMPL3 ${filler}${filler}\0`)])).toEqual([]); | |
| 58 | + // UTF-16 with a byte order mark is text. | |
| 59 | + const text = `${"y".repeat(300_000)}AWS_ACCESS_KEY_ID=AKIAIOSFODNN7EXAMPL3\n`; | |
| 60 | + const le = new Uint8Array(2 + text.length * 2); | |
| 61 | + le.set([0xff, 0xfe]); | |
| 62 | + for (let i = 0; i < text.length; i++) le[2 + i * 2] = text.charCodeAt(i); | |
| 63 | + expect(scanForSecrets([{ oid: "d".repeat(40), content: new Uint8Array(0), deflated: deflate(le) }]).map((h) => h.kind)).toEqual(["AWS access key"]); | |
| 64 | + }); | |
| 65 | + | |
| 41 | 66 | it("reads the bypass reason from push options", () => { |
| 42 | 67 | expect(secretBypassReason([])).toBeNull(); |
| 43 | 68 | expect(secretBypassReason(["allow-secrets="])).toBeNull(); |
modifiedpackages/git-core/src/pack.ts+45−5View file
| @@ -1,22 +1,41 @@ | ||
| 1 | 1 | import { ByteWriter, equalBytes, fromUtf8, toHex } from "./bytes.js"; |
| 2 | 2 | import { applyDelta } from "./delta.js"; |
| 3 | -import { CODE_TYPES, TYPE_CODES, hashObject, type GitObject, type ObjectType } from "./objects.js"; | |
| 3 | +import { CODE_TYPES, TYPE_CODES, hashObject, objectHeader, type GitObject, type ObjectType } from "./objects.js"; | |
| 4 | 4 | import { Sha1, sha1 } from "./sha1.js"; |
| 5 | -import { inflateAt } from "./zlib.js"; | |
| 5 | +import { inflate, inflateAt, inflateEach } from "./zlib.js"; | |
| 6 | 6 | |
| 7 | 7 | const OFS_DELTA = 6; |
| 8 | 8 | const REF_DELTA = 7; |
| 9 | 9 | |
| 10 | 10 | export interface PackedObject extends GitObject { |
| 11 | 11 | oid: string; |
| 12 | + /** | |
| 13 | + * Set on a large blob parsed with `largeObjectBytes`: its zlib data as it sits in the pack (a view, | |
| 14 | + * not a copy) and its size. Its `content` is then inflated afresh on every read, so callers that | |
| 15 | + * can work from `deflated` should. | |
| 16 | + */ | |
| 17 | + deflated?: Uint8Array; | |
| 18 | + size?: number; | |
| 12 | 19 | } |
| 13 | 20 | |
| 21 | +/** A large blob kept compressed: `content` inflates it when something needs the bytes. */ | |
| 22 | +function largeBlob(oid: string, size: number, deflated: Uint8Array): PackedObject { | |
| 23 | + const obj = { oid, type: "blob" as const, size, deflated } as PackedObject; | |
| 24 | + Object.defineProperty(obj, "content", { enumerable: true, get: () => inflate(deflated) }); | |
| 25 | + return obj; | |
| 26 | +} | |
| 27 | + | |
| 28 | +/** Packs bigger than this get their checksum computed in slices, so the pack is never copied for hashing. */ | |
| 29 | +const SLICED_CHECKSUM_BYTES = 16 * 1024 * 1024; | |
| 30 | + | |
| 14 | 31 | interface RawEntry { |
| 15 | 32 | offset: number; |
| 16 | 33 | typeCode: number; |
| 17 | 34 | data: Uint8Array; // inflated: full content, or delta instructions |
| 18 | 35 | baseOffset?: number; |
| 19 | 36 | baseOid?: string; |
| 37 | + /** A large blob that was hashed while inflating and not kept (see `largeObjectBytes`). */ | |
| 38 | + large?: PackedObject; | |
| 20 | 39 | } |
| 21 | 40 | |
| 22 | 41 | export type ExternalLookup = (oid: string) => Promise<GitObject | null>; |
| @@ -30,7 +49,7 @@ | ||
| 30 | 49 | * Packs come from clients, so unpacking is bounded: each object stops inflating at its |
| 31 | 50 | * declared size, and the whole pack may unpack to at most `maxBytes`. |
| 32 | 51 | */ |
| 33 | -export async function parsePack(pack: Uint8Array, external?: ExternalLookup, opts: { maxBytes?: number } = {}): Promise<PackedObject[]> { | |
| 52 | +export async function parsePack(pack: Uint8Array, external?: ExternalLookup, opts: { maxBytes?: number; largeObjectBytes?: number } = {}): Promise<PackedObject[]> { | |
| 34 | 53 | if (pack.length < 32 || fromUtf8(pack.subarray(0, 4)) !== "PACK") throw new Error("not a packfile"); |
| 35 | 54 | const view = new DataView(pack.buffer, pack.byteOffset, pack.byteLength); |
| 36 | 55 | const version = view.getUint32(4); |
| @@ -38,10 +57,16 @@ | ||
| 38 | 57 | const count = view.getUint32(8); |
| 39 | 58 | |
| 40 | 59 | const trailer = pack.subarray(pack.length - 20); |
| 41 | - const digest = await sha1(pack.subarray(0, pack.length - 20)); | |
| 60 | + let digest: Uint8Array; | |
| 61 | + if (pack.length > SLICED_CHECKSUM_BYTES) { | |
| 62 | + const h = new Sha1(); | |
| 63 | + for (let i = 0; i < pack.length - 20; i += 1024 * 1024) h.update(pack.subarray(i, Math.min(i + 1024 * 1024, pack.length - 20))); | |
| 64 | + digest = h.digest(); | |
| 65 | + } else digest = await sha1(pack.subarray(0, pack.length - 20)); | |
| 42 | 66 | if (!equalBytes(trailer, digest)) throw new Error("pack checksum mismatch"); |
| 43 | 67 | |
| 44 | 68 | const maxBytes = opts.maxBytes ?? MAX_UNPACKED_BYTES; |
| 69 | + const large = opts.largeObjectBytes ?? Infinity; | |
| 45 | 70 | let unpacked = 0; |
| 46 | 71 | const tooBig = () => new Error(`pack unpacks to more than ${maxBytes} bytes`); |
| 47 | 72 | |
| @@ -74,6 +99,17 @@ | ||
| 74 | 99 | throw new Error(`bad object type ${typeCode} at ${offset}`); |
| 75 | 100 | } |
| 76 | 101 | if (unpacked + size > maxBytes) throw tooBig(); |
| 102 | + if (typeCode === TYPE_CODES.blob && size >= large) { | |
| 103 | + // Hash it as it inflates and keep only the compressed bytes, so memory never holds it whole. | |
| 104 | + const h = new Sha1().update(objectHeader("blob", size)); | |
| 105 | + const { consumed, produced } = inflateEach(pack, pos, (chunk) => h.update(chunk), size); | |
| 106 | + if (produced !== size) throw new Error(`size mismatch at ${offset}: ${produced} != ${size}`); | |
| 107 | + entry.large = largeBlob(toHex(h.digest()), size, pack.subarray(pos, pos + consumed)); | |
| 108 | + unpacked += size; | |
| 109 | + pos += consumed; | |
| 110 | + raws.push(entry); | |
| 111 | + continue; | |
| 112 | + } | |
| 77 | 113 | const { out, consumed } = inflateAt(pack, pos, size); |
| 78 | 114 | if (out.length !== size) throw new Error(`size mismatch at ${offset}: ${out.length} != ${size}`); |
| 79 | 115 | unpacked += size; |
| @@ -96,7 +132,11 @@ | ||
| 96 | 132 | let pending: RawEntry[] = []; |
| 97 | 133 | for (const raw of raws) { |
| 98 | 134 | const t = CODE_TYPES[raw.typeCode]; |
| 99 | - if (t) await add(raw, t, raw.data); | |
| 135 | + if (raw.large) { | |
| 136 | + byOffset.set(raw.offset, raw.large); | |
| 137 | + byOid.set(raw.large.oid, raw.large); | |
| 138 | + results.push(raw.large); | |
| 139 | + } else if (t) await add(raw, t, raw.data); | |
| 100 | 140 | else pending.push(raw); |
| 101 | 141 | } |
| 102 | 142 |
modifiedpackages/git-core/src/zlib.ts+27−0View file
| @@ -42,3 +42,30 @@ | ||
| 42 | 42 | const out = inflator.result as Uint8Array; |
| 43 | 43 | return { out: out ?? new Uint8Array(0), consumed }; |
| 44 | 44 | } |
| 45 | + | |
| 46 | +/** | |
| 47 | + * Like `inflateAt`, but hands each inflated chunk to `onChunk` instead of keeping it, so an object | |
| 48 | + * much larger than memory allows can be hashed or scanned. Returns the input bytes consumed and the | |
| 49 | + * bytes produced. | |
| 50 | + */ | |
| 51 | +export function inflateEach(data: Uint8Array, offset: number, onChunk: (chunk: Uint8Array) => void, expectedSize?: number): { consumed: number; produced: number } { | |
| 52 | + const inflator = new pako.Inflate({ chunkSize: 256 * 1024 }) as pako.Inflate & { ended: boolean }; | |
| 53 | + let produced = 0; | |
| 54 | + inflator.onData = (chunk) => { | |
| 55 | + const c = chunk as Uint8Array; | |
| 56 | + produced += c.length; | |
| 57 | + if (expectedSize !== undefined && produced > expectedSize) throw new Error(`object at ${offset} is larger than its declared size (${expectedSize} bytes)`); | |
| 58 | + onChunk(c); | |
| 59 | + }; | |
| 60 | + const window = 1024 * 1024; | |
| 61 | + let pos = offset; | |
| 62 | + while (!inflator.ended && pos < data.length) { | |
| 63 | + const end = Math.min(data.length, pos + window); | |
| 64 | + inflator.push(data.subarray(pos, end), false); | |
| 65 | + if (inflator.err) throw new Error(`zlib error at ${offset}: ${inflator.msg}`); | |
| 66 | + pos = end; | |
| 67 | + } | |
| 68 | + if (!inflator.ended) throw new Error(`truncated zlib stream at offset ${offset}`); | |
| 69 | + const strm = (inflator as unknown as { strm: { avail_in: number } }).strm; | |
| 70 | + return { consumed: pos - offset - strm.avail_in, produced }; | |
| 71 | +} |
modifiedpackages/git-core/test/pack.test.ts+42−0View file
| @@ -21,7 +21,10 @@ | ||
| 21 | 21 | writePack, |
| 22 | 22 | fromUtf8, |
| 23 | 23 | listFiles, |
| 24 | + hashObject, | |
| 25 | + inflate, | |
| 24 | 26 | } from "../src/index.js"; |
| 27 | +import { randomBytes } from "node:crypto"; | |
| 25 | 28 | |
| 26 | 29 | const git = (cwd: string, ...args: string[]) => |
| 27 | 30 | execFileSync("git", args, { cwd, env: { ...process.env, GIT_AUTHOR_NAME: "T", GIT_AUTHOR_EMAIL: "t@x", GIT_COMMITTER_NAME: "T", GIT_COMMITTER_EMAIL: "t@x" } }); |
| @@ -59,6 +62,45 @@ | ||
| 59 | 62 | for (const oid of expected) expect(store.objects.has(oid), oid).toBe(true); |
| 60 | 63 | }); |
| 61 | 64 | |
| 65 | + it("keeps large blobs compressed, with the same oids, and still resolves deltas against them", async () => { | |
| 66 | + const dir = makeRepo(); | |
| 67 | + const expected = git(dir, "rev-list", "--objects", "--all").toString().trim().split("\n").map((l) => l.split(" ")[0]!); | |
| 68 | + const pack = new Uint8Array(execFileSync("git", ["pack-objects", "--stdout", "--delta-base-offset"], { cwd: dir, input: expected.join("\n") + "\n" })); | |
| 69 | + const plain = await parsePack(pack); | |
| 70 | + const lean = await parsePack(pack, undefined, { largeObjectBytes: 1000 }); | |
| 71 | + expect(lean.map((o) => o.oid).sort()).toEqual(plain.map((o) => o.oid).sort()); | |
| 72 | + const large = lean.filter((o) => o.deflated); | |
| 73 | + // a.txt in both commits: one is stored whole, the other is a delta against it. | |
| 74 | + expect(large.length).toBeGreaterThanOrEqual(1); | |
| 75 | + for (const o of large) { | |
| 76 | + const want = plain.find((p) => p.oid === o.oid)!; | |
| 77 | + expect(o.size).toBe(want.content.length); | |
| 78 | + expect(inflate(o.deflated!)).toEqual(want.content); | |
| 79 | + expect(o.content).toEqual(want.content); | |
| 80 | + } | |
| 81 | + for (const o of lean) expect(await hashObject(o.type, o.content)).toBe(o.oid); | |
| 82 | + }); | |
| 83 | + | |
| 84 | + it("parses a pack over 16 MB holding one large blob", async () => { | |
| 85 | + const dir = mkdtempSync(join(tmpdir(), "gc-")); | |
| 86 | + git(dir, "init", "-q", "-b", "main"); | |
| 87 | + const data = randomBytes(17 * 1024 * 1024); | |
| 88 | + writeFileSync(join(dir, "big.bin"), data); | |
| 89 | + git(dir, "add", "-A"); | |
| 90 | + git(dir, "commit", "-q", "-m", "big"); | |
| 91 | + const expected = git(dir, "rev-list", "--objects", "--all").toString().trim().split("\n").map((l) => l.split(" ")[0]!); | |
| 92 | + const pack = new Uint8Array(execFileSync("git", ["pack-objects", "--stdout"], { cwd: dir, input: expected.join("\n") + "\n", maxBuffer: 64 * 1024 * 1024 })); | |
| 93 | + expect(pack.length).toBeGreaterThan(16 * 1024 * 1024); | |
| 94 | + const objs = await parsePack(pack, undefined, { largeObjectBytes: 4 * 1024 * 1024 }); | |
| 95 | + const blob = objs.find((o) => o.type === "blob")!; | |
| 96 | + expect(blob.oid).toBe(git(dir, "rev-parse", "HEAD:big.bin").toString().trim()); | |
| 97 | + expect(blob.size).toBe(data.length); | |
| 98 | + expect(Buffer.from(inflate(blob.deflated!)).equals(data)).toBe(true); | |
| 99 | + // A damaged pack still fails its checksum. | |
| 100 | + pack[pack.length - 30]! ^= 1; | |
| 101 | + await expect(parsePack(pack, undefined, { largeObjectBytes: 4 * 1024 * 1024 })).rejects.toThrow(/checksum/); | |
| 102 | + }); | |
| 103 | + | |
| 62 | 104 | it("writes packs that git accepts", async () => { |
| 63 | 105 | const dir = makeRepo(); |
| 64 | 106 | const { store } = await loadAll(dir); |