| 1 | import { StreamedPack, fromUtf8, inflateSized, type ExternalLookup, type ObjectType, type PackSource, type PackedObject } from "@diffinery/git-core"; |
| 2 | import { INLINE_LIMIT, type RepoObjectStore } from "../repo/objects.js"; |
| 3 | import { fsckGitmodules, fsckObjects, fsckSize, fsckTree } from "./fsck.js"; |
| 4 | import { parseReceivePackHead, type ReceiveRequest } from "./protocol.js"; |
| 5 | import { scanFeed, scanForSecrets, type SecretHit } from "./secret-scan.js"; |
| 6 | |
| 7 | /** |
| 8 | * What a push brings, as receive-pack checks and stores it: `fsck`, then the secret scan, then |
| 9 | * `write` once every check passed (nothing is stored before, like git's quarantine). |
| 10 | */ |
| 11 | export interface PushContents { |
| 12 | /** The pack's size, for the storage quota. */ |
| 13 | bytes: number; |
| 14 | fsck(): string | null; |
| 15 | /** Credentials found; opted-out ones go to `marked`. */ |
| 16 | scan(marked: SecretHit[]): SecretHit[]; |
| 17 | /** Whether the push stores anything the repository doesn't have. */ |
| 18 | anyNew(): boolean; |
| 19 | write(): Promise<void>; |
| 20 | } |
| 21 | |
| 22 | const scanned = (type: string) => type === "blob" || type === "commit" || type === "tag"; |
| 23 | |
| 24 | /** |
| 25 | * Storing order. A push can be cut off while it is stored (a deploy, a limit), and git then pushes |
| 26 | * again asking which commits the server has (push.negotiate): it won't send what such a commit |
| 27 | * reaches. So files and trees are stored first, and commits after them, each after its parents, |
| 28 | * then tags: a commit that is stored never lacks its files or history. |
| 29 | */ |
| 30 | const last = (type: string) => type === "commit" || type === "tag"; |
| 31 | |
| 32 | /** A commit's parents, from its header. */ |
| 33 | export function commitParents(content: Uint8Array): string[] { |
| 34 | const end = headerEnd(content); |
| 35 | const out: string[] = []; |
| 36 | for (const line of fromUtf8(content.subarray(0, end)).split("\n")) { |
| 37 | const m = /^parent ([0-9a-f]{40})$/.exec(line); |
| 38 | if (m) out.push(m[1]!); |
| 39 | } |
| 40 | return out; |
| 41 | } |
| 42 | |
| 43 | /** Where a commit's header ends (the first empty line). */ |
| 44 | function headerEnd(content: Uint8Array): number { |
| 45 | for (let i = 0; i + 1 < content.length; i++) if (content[i] === 10 && content[i + 1] === 10) return i; |
| 46 | return content.length; |
| 47 | } |
| 48 | |
| 49 | /** `items` reordered so each commit comes after those of its parents that are among them; tags last. */ |
| 50 | export function historyOrder<T extends { oid: string; type: string }>(items: T[], parents: (t: T) => string[]): T[] { |
| 51 | const commits = new Map(items.filter((t) => t.type === "commit").map((t) => [t.oid, t])); |
| 52 | const out: T[] = []; |
| 53 | const entered = new Set<string>(); |
| 54 | for (const start of commits.values()) { |
| 55 | if (entered.has(start.oid)) continue; |
| 56 | entered.add(start.oid); |
| 57 | // Depth first without recursion (a history can be thousands of commits deep). |
| 58 | const stack: Array<{ item: T; ps: string[]; i: number }> = [{ item: start, ps: parents(start), i: 0 }]; |
| 59 | while (stack.length) { |
| 60 | const top = stack[stack.length - 1]!; |
| 61 | if (top.i < top.ps.length) { |
| 62 | const p = commits.get(top.ps[top.i++]!); |
| 63 | if (p && !entered.has(p.oid)) { |
| 64 | entered.add(p.oid); |
| 65 | stack.push({ item: p, ps: parents(p), i: 0 }); |
| 66 | } |
| 67 | continue; |
| 68 | } |
| 69 | out.push(top.item); |
| 70 | stack.pop(); |
| 71 | } |
| 72 | } |
| 73 | for (const t of items) if (t.type !== "commit") out.push(t); |
| 74 | return out; |
| 75 | } |
| 76 | |
| 77 | /** A pack parsed whole in memory (pushes up to 16 MB). */ |
| 78 | export function packedContents(store: RepoObjectStore, objects: PackedObject[], bytes: number): PushContents { |
| 79 | return { |
| 80 | bytes, |
| 81 | fsck: () => fsckObjects(objects), |
| 82 | // The objects themselves, not copies: a large blob is scanned from its compressed bytes. |
| 83 | scan: (marked) => scanForSecrets(objects.filter((o) => scanned(o.type)), marked), |
| 84 | anyNew: () => objects.some((o) => !store.has(o.oid)), |
| 85 | write: async () => { |
| 86 | const put = async (o: PackedObject) => { |
| 87 | if (o.deflated) await store.writeDeflated(o.oid, o.type, o.size!, o.deflated); |
| 88 | // An object that isn't a delta keeps the zlib data it came with: compressing it again cost |
| 89 | // CPU on every push for the same bytes. |
| 90 | else if (o.zlib) await store.writeDeflated(o.oid, o.type, o.content.length, o.zlib, o.content); |
| 91 | else await store.writeKnown(o.oid, o.type, o.content); |
| 92 | }; |
| 93 | for (const o of objects) if (!last(o.type)) await put(o); |
| 94 | for (const o of historyOrder(objects.filter((o) => last(o.type)), (o) => commitParents(o.content))) await put(o); |
| 95 | }, |
| 96 | }; |
| 97 | } |
| 98 | |
| 99 | /** |
| 100 | * Bases a thin pack's deltas name, from the repository: what reading one takes is checked first, |
| 101 | * so a small delta against a huge stored object is refused before it is inflated. |
| 102 | */ |
| 103 | export function repositoryBases(store: RepoObjectStore): ExternalLookup { |
| 104 | return async (oid, maxBytes) => { |
| 105 | const cost = store.readCost(oid); |
| 106 | if (cost === null) return null; |
| 107 | return cost > maxBytes ? cost : store.read(oid); |
| 108 | }; |
| 109 | } |
| 110 | |
| 111 | /** A pack already in memory, as a source to unpack in pieces. */ |
| 112 | export function memorySource(pack: Uint8Array): PackSource { |
| 113 | return { size: pack.length, read: async (offset, length) => pack.subarray(offset, offset + length) }; |
| 114 | } |
| 115 | |
| 116 | /** Bytes of an R2 object from `start` on, read in ranges. */ |
| 117 | function r2Source(bucket: R2Bucket, key: string, start: number, size: number): PackSource { |
| 118 | return { |
| 119 | size, |
| 120 | read: async (offset, length) => { |
| 121 | const o = await bucket.get(key, { range: { offset: start + offset, length } }); |
| 122 | if (!o) throw new Error("the push's data is missing; push again"); |
| 123 | return new Uint8Array(await o.arrayBuffer()); |
| 124 | }, |
| 125 | }; |
| 126 | } |
| 127 | |
| 128 | /** Where a push staged in R2 has its commands, push options and pack. */ |
| 129 | export async function readStagedHead(bucket: R2Bucket, key: string, size: number): Promise<{ req: ReceiveRequest; packOffset: number }> { |
| 130 | // Commands for many refs can run long; read more until they end. |
| 131 | for (let n = 64 * 1024; ; n *= 4) { |
| 132 | const o = await bucket.get(key, { range: { offset: 0, length: Math.min(n, size) } }); |
| 133 | if (!o) throw new Error("the push's data is missing; push again"); |
| 134 | const head = parseReceivePackHead(new Uint8Array(await o.arrayBuffer())); |
| 135 | if (head) return head; |
| 136 | if (n >= size) throw new Error("truncated push: its commands never end"); |
| 137 | if (n >= 16 * 1024 * 1024) throw new Error("the push names too many refs; split it into smaller pushes"); |
| 138 | } |
| 139 | } |
| 140 | |
| 141 | class FsckRefusal extends Error {} |
| 142 | |
| 143 | /** |
| 144 | * A pack staged in R2, unpacked without holding it whole (see StreamedPack): every object is |
| 145 | * checked as it is resolved, and read again from R2 to be stored. |
| 146 | */ |
| 147 | export function stagedContents(store: RepoObjectStore, bucket: R2Bucket, key: string, packOffset: number, packBytes: number, o: { largeObjectBytes: number; maxHeldBytes?: number }): Promise<PushContents> { |
| 148 | return streamedContents(store, r2Source(bucket, key, packOffset, packBytes), o); |
| 149 | } |
| 150 | |
| 151 | /** A pack unpacked from `src` in pieces, one object at a time (see StreamedPack). */ |
| 152 | export async function streamedContents(store: RepoObjectStore, src: PackSource, o: { largeObjectBytes: number; maxHeldBytes?: number }): Promise<PushContents> { |
| 153 | const packBytes = src.size; |
| 154 | const hits: SecretHit[] = []; |
| 155 | const opted: SecretHit[] = []; |
| 156 | const gitmodules = new Set<string>(); |
| 157 | let refusal: string | null = null; |
| 158 | let pack: StreamedPack | null = null; |
| 159 | try { |
| 160 | const p = (pack = await StreamedPack.open(src, { external: repositoryBases(store), largeObjectBytes: o.largeObjectBytes, maxHeldBytes: o.maxHeldBytes, writesOver: INLINE_LIMIT }, async (obj, content, feed) => { |
| 161 | const why = obj.type === "tree" ? fsckTree(obj.oid, content!, gitmodules) : content ? fsckSize(obj.oid, obj.type, content.length) : null; |
| 162 | if (why) throw new FsckRefusal(why); |
| 163 | if (scanned(obj.type)) { |
| 164 | if (content) scanForSecrets([{ oid: obj.oid, content }], opted).forEach((h) => hits.push(h)); |
| 165 | else await scanFeed(obj.oid, feed!, hits, opted); |
| 166 | } |
| 167 | })); |
| 168 | // Like fsckObjects: a `.gitmodules` blob in this push that a pushed tree names is checked too. |
| 169 | for (const [oid, entry] of p.entriesOf(gitmodules)) { |
| 170 | const why = fsckGitmodules(oid, await p.content(entry)); |
| 171 | if (why) throw new FsckRefusal(why); |
| 172 | } |
| 173 | } catch (e) { |
| 174 | if (!(e instanceof FsckRefusal)) throw e; |
| 175 | refusal = e.message; |
| 176 | } |
| 177 | const refused = refusal; |
| 178 | const objects = () => (pack ? pack.objects() : []); |
| 179 | return { |
| 180 | // What storing adds, at most: a delta is stored compressed on its own, not as in the pack. |
| 181 | bytes: pack ? (pack as StreamedPack).storedBytes : packBytes, |
| 182 | fsck: () => refused, |
| 183 | scan: (marked) => { |
| 184 | marked.push(...opted); |
| 185 | return hits; |
| 186 | }, |
| 187 | anyNew: () => { |
| 188 | for (const x of objects()) if (!store.has(x.oid)) return true; |
| 189 | return false; |
| 190 | }, |
| 191 | write: async () => { |
| 192 | const p = pack!; |
| 193 | // Commits and tags wait (see `last`): their parents are noted, and a rebuilt one kept. |
| 194 | const later: Array<{ oid: string; type: string; parents: string[]; slice?: { at: number; length: number; size: number }; content?: Uint8Array }> = []; |
| 195 | const readSlices = async (at: number, length: number) => { |
| 196 | const out = new Uint8Array(length); |
| 197 | let n = 0; |
| 198 | for await (const s of p.slices(at, length)) { |
| 199 | out.set(s, n); |
| 200 | n += s.length; |
| 201 | } |
| 202 | return out; |
| 203 | }; |
| 204 | // An object that isn't a delta is stored from its zlib data in the pack, as it is. |
| 205 | for (const x of p.objects()) { |
| 206 | if (!x.deflated || store.has(x.oid)) continue; |
| 207 | if (last(x.type)) { |
| 208 | const content = x.type === "commit" ? inflateSized(await readSlices(x.deflated.at, x.deflated.length), x.size) : null; |
| 209 | later.push({ oid: x.oid, type: x.type, parents: content ? commitParents(content) : [], slice: { ...x.deflated, size: x.size } }); |
| 210 | continue; |
| 211 | } |
| 212 | await store.writeDeflatedSlices(x.oid, x.type as ObjectType, x.size, x.deflated.length, () => p.slices(x.deflated!.at, x.deflated!.length)); |
| 213 | } |
| 214 | // Deltas are rebuilt exactly as the checks rebuilt them, so no limit can stop the push |
| 215 | // halfway through being stored. Not kept in the store's memory cache: the push would fill |
| 216 | // it for nothing. |
| 217 | await p.replayDeltas(async (entry, content) => { |
| 218 | const oid = p.oidOf(entry); |
| 219 | if (store.has(oid)) return; |
| 220 | const type = p.typeOf(entry); |
| 221 | if (last(type)) later.push({ oid, type, parents: type === "commit" ? commitParents(content) : [], content }); |
| 222 | else await store.writeKnown(oid, type, content, false, p.zlibSizeOf(entry)); |
| 223 | }); |
| 224 | for (const c of historyOrder(later, (c) => c.parents)) { |
| 225 | if (c.content) await store.writeKnown(c.oid, c.type as ObjectType, c.content, false); |
| 226 | else await store.writeDeflatedSlices(c.oid, c.type as ObjectType, c.slice!.size, c.slice!.length, () => p.slices(c.slice!.at, c.slice!.length)); |
| 227 | } |
| 228 | }, |
| 229 | }; |
| 230 | } |