| 1 | import { Hono } from "hono"; |
| 2 | import type { Env } from "../env.js"; |
| 3 | import type { AppVars } from "../auth/principal.js"; |
| 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"; |
| 7 | import { importPushAllowed } from "../imports/service.js"; |
| 8 | import { loadRepo, repoStub } from "./common.js"; |
| 9 | import type { Caller } from "../repo/types.js"; |
| 10 | import { resetByDeploy } from "../git/protocol.js"; |
| 11 | |
| 12 | /** |
| 13 | * Git smart HTTP (protocol v2 for fetch/clone, v0 for push) and the Git LFS |
| 14 | * batch API, at https://diffinery.com/<owner>/<repo>.git |
| 15 | * |
| 16 | * A transferred or renamed repository is served under its old name too (`follow`), |
| 17 | * as GitHub does, and a push there says where it moved. A redirect would not do: |
| 18 | * git follows one only on info/refs (never on its POSTs), and then rebuilds its |
| 19 | * credentials from the new URL, which drops a token embedded in the remote URL. |
| 20 | */ |
| 21 | export const git = new Hono<{ Bindings: Env; Variables: AppVars }>(); |
| 22 | |
| 23 | const noCache = { "cache-control": "no-cache, max-age=0, must-revalidate", pragma: "no-cache" }; |
| 24 | |
| 25 | git.get("/:owner/:repo/info/refs", async (c) => { |
| 26 | const service = c.req.query("service"); |
| 27 | if (service !== "git-upload-pack" && service !== "git-receive-pack") throw new HttpError(403, "dumb HTTP protocol is not supported; update git"); |
| 28 | // Git only offers credentials after a 401, so an anonymous push must be asked for them (as GitHub does), even to a public repository. |
| 29 | if (service === "git-receive-pack" && c.get("principal").type === "anonymous") throw new HttpError(401, "authentication required"); |
| 30 | const { stub, repo } = await loadRepo(c, c.req.param("owner"), c.req.param("repo"), service === "git-receive-pack" ? "write" : "read", { follow: true }); |
| 31 | if (service === "git-receive-pack" && repo.archived) throw new HttpError(403, "repository is archived (read-only)"); |
| 32 | if (service === "git-upload-pack" && !/version=2/.test(c.req.header("git-protocol") ?? "")) { |
| 33 | throw new HttpError(400, "Diffinery requires Git protocol v2 (git 2.26 or newer, or run: git config --global protocol.version 2)"); |
| 34 | } |
| 35 | const body = await stub.gitInfoRefs(service); |
| 36 | return new Response(body, { headers: { "content-type": `application/x-${service}-advertisement`, ...noCache } }); |
| 37 | }); |
| 38 | |
| 39 | git.post("/:owner/:repo/git-upload-pack", async (c) => { |
| 40 | const { stub } = await loadRepo(c, c.req.param("owner"), c.req.param("repo"), "read", { readOnlyPost: true, follow: true }); |
| 41 | const body = await readBody(c.req.raw, MAX_UPLOAD_PACK_REQUEST, "fetch request exceeds 16 MB"); |
| 42 | const out = await stub.gitUploadPack(body); |
| 43 | return new Response(out, { headers: { "content-type": "application/x-git-upload-pack-result", ...noCache } }); |
| 44 | }); |
| 45 | |
| 46 | git.post("/:owner/:repo/git-receive-pack", async (c) => { |
| 47 | // During someone else's import the repository refuses the push itself, with a message the client prints. |
| 48 | const { stub, caller, repo, moved } = await loadRepo(c, c.req.param("owner"), c.req.param("repo"), "write", { follow: true, whileImporting: true }); |
| 49 | // A GitHub import's own token, while that import is running and filling this repository, may push in import mode. |
| 50 | const p = c.get("principal"); |
| 51 | if (p.type === "account" && p.importId && p.tokenId && (await importPushAllowed(c.env, p.importId, repo.id, p.tokenId))) caller.importId = p.importId; |
| 52 | const { body, release } = await readPushBody(c.env, repo.id, c.req.raw); |
| 53 | const out = await receiveWithRetry(c.env, repo.id, stub, body, caller, moved).finally(release); |
| 54 | c.executionCtx.waitUntil(c.get("dir").touchPushed(repo.id)); |
| 55 | return new Response(out, { headers: { "content-type": "application/x-git-receive-pack-result", ...noCache } }); |
| 56 | }); |
| 57 | |
| 58 | /** |
| 59 | * A deploy resets every Durable Object, and one doing a push then fails it mid-way: git doesn't |
| 60 | * retry a push, so it would print a 500. The push is handed over once more (to the new code; a |
| 61 | * large one is still staged in R2: only this function deletes it, once the push is done either way). |
| 62 | * Nothing is lost or moved twice: objects are stored by their ids, and refs move only at the end, |
| 63 | * each by compare-and-swap; one the cut-off attempt already moved to the pushed commit is reported |
| 64 | * as updated. |
| 65 | */ |
| 66 | export async function receiveWithRetry(env: Env, repoId: string, stub: ReturnType<typeof repoStub>, body: Uint8Array | SpilledBody, caller: Caller, moved: boolean): Promise<Uint8Array> { |
| 67 | try { |
| 68 | for (let attempt = 1; ; attempt++) { |
| 69 | try { |
| 70 | return await stub.gitReceivePack(body, caller, { moved, retry: attempt > 1 }); |
| 71 | } catch (e) { |
| 72 | if (attempt >= PUSH_TRIES || !(resetByDeploy(e) || resetForMemory(e))) throw e; |
| 73 | // A stub that saw its object reset is broken; a new one reaches the new code. |
| 74 | stub = repoStub(env, repoId); |
| 75 | } |
| 76 | } |
| 77 | } finally { |
| 78 | if (!(body instanceof Uint8Array)) await env.BUCKET.delete(body.spilled).catch(() => {}); |
| 79 | } |
| 80 | } |
| 81 | |
| 82 | const PUSH_TRIES = 2; |
| 83 | |
| 84 | /** |
| 85 | * The repository's object was reset for going past its memory. Pushes to one repository take turns |
| 86 | * for memory (RepoDO.gitReceivePack), so a push cut off this way was most likely caught up in another |
| 87 | * request's reset, not the cause: like a deploy's reset, it is handed over once more, which moves the |
| 88 | * refs the cut-off attempt didn't and sends the push events it may not have sent. |
| 89 | */ |
| 90 | export const resetForMemory = (e: unknown) => e instanceof Error && /exceeded its memory limit and was reset/i.test(e.message); |
| 91 | |
| 92 | export { resetByDeploy }; |
| 93 | |
| 94 | /** A fetch's wants and haves; git's own negotiation stays far below this. */ |
| 95 | const MAX_UPLOAD_PACK_REQUEST = 16 * 1024 * 1024; |
| 96 | /** The repository refuses pushes over 100 MB with a clear message; this stops anything far beyond. */ |
| 97 | const MAX_RECEIVE_PACK_REQUEST = 101 * 1024 * 1024; |
| 98 | |
| 99 | /** |
| 100 | * The request body, gunzipped when the client compressed it. Reading stops at `max` bytes |
| 101 | * (after decompression), so neither a huge body nor a gzip bomb is ever held in memory. |
| 102 | */ |
| 103 | async function readBody(req: Request, max: number, tooLarge: string): Promise<Uint8Array> { |
| 104 | let stream = req.body; |
| 105 | if (!stream) return new Uint8Array(0); |
| 106 | if (req.headers.get("content-encoding") === "gzip") stream = stream.pipeThrough(new DecompressionStream("gzip")); |
| 107 | const chunks: Uint8Array[] = []; |
| 108 | let total = 0; |
| 109 | const reader = stream.getReader(); |
| 110 | for (;;) { |
| 111 | const { done, value } = await reader.read(); |
| 112 | if (done) break; |
| 113 | total += value.length; |
| 114 | if (total > max) { |
| 115 | await reader.cancel().catch(() => {}); |
| 116 | throw new HttpError(413, tooLarge); |
| 117 | } |
| 118 | chunks.push(value); |
| 119 | } |
| 120 | const out = new Uint8Array(total); |
| 121 | let at = 0; |
| 122 | for (const ch of chunks) { |
| 123 | out.set(ch, at); |
| 124 | at += ch.length; |
| 125 | } |
| 126 | return out; |
| 127 | } |
| 128 | |
| 129 | /** Pushes up to this size are handed to the repository in memory; larger ones go through R2. */ |
| 130 | const IN_MEMORY_PUSH = 16 * 1024 * 1024; |
| 131 | /** |
| 132 | * Push bodies one Worker isolate holds in memory at once, across its requests (it has 128 MB, and a |
| 133 | * body is held twice while it is joined): past this, a push goes through R2 even under IN_MEMORY_PUSH. |
| 134 | */ |
| 135 | const ISOLATE_PUSH_BYTES = 32 * 1024 * 1024; |
| 136 | let bufferedPushBytes = 0; |
| 137 | |
| 138 | /** |
| 139 | * A push's body. A large one is staged in R2 as it arrives (a Worker and a Durable Object each |
| 140 | * have 128 MB of memory, and holding a 70 MB push twice, then copying it to the repository, does |
| 141 | * not fit); the repository reads it from there once and deletes it. |
| 142 | */ |
| 143 | async function readPushBody(env: Env, repoId: string, req: Request): Promise<{ body: Uint8Array | SpilledBody; release: () => void }> { |
| 144 | let counted = 0; |
| 145 | const release = () => { |
| 146 | bufferedPushBytes -= counted; |
| 147 | counted = 0; |
| 148 | }; |
| 149 | try { |
| 150 | return { body: await readPushBodyCounted(env, repoId, req, (n) => ((counted += n), (bufferedPushBytes += n))), release }; |
| 151 | } catch (e) { |
| 152 | release(); |
| 153 | throw e; |
| 154 | } |
| 155 | } |
| 156 | |
| 157 | /** readPushBody's body; `count` is told of every byte held in memory (ISOLATE_PUSH_BYTES). */ |
| 158 | async function readPushBodyCounted(env: Env, repoId: string, req: Request, count: (bytes: number) => void): Promise<Uint8Array | SpilledBody> { |
| 159 | const tooLarge = "push exceeds 100 MB; split it into smaller pushes"; |
| 160 | let stream = req.body; |
| 161 | if (!stream) return new Uint8Array(0); |
| 162 | if (req.headers.get("content-encoding") === "gzip") stream = stream.pipeThrough(new DecompressionStream("gzip")); |
| 163 | const chunks: Uint8Array[] = []; |
| 164 | let total = 0; |
| 165 | const reader = stream.getReader(); |
| 166 | for (;;) { |
| 167 | const { done, value } = await reader.read(); |
| 168 | if (done) break; |
| 169 | total += value.length; |
| 170 | chunks.push(value); |
| 171 | count(value.length); |
| 172 | if (total > IN_MEMORY_PUSH || bufferedPushBytes > ISOLATE_PUSH_BYTES) { |
| 173 | // Too big to pass along in memory (or this isolate holds enough already): what came so far, then the rest, goes to R2. |
| 174 | count(-total); |
| 175 | const key = pushBodyKey(repoId); |
| 176 | const rest = new ReadableStream<Uint8Array>({ |
| 177 | start(controller) { |
| 178 | for (const ch of chunks.splice(0)) controller.enqueue(ch); |
| 179 | }, |
| 180 | async pull(controller) { |
| 181 | const next = await reader.read(); |
| 182 | if (next.done) controller.close(); |
| 183 | else controller.enqueue(next.value); |
| 184 | }, |
| 185 | cancel: () => reader.cancel(), |
| 186 | }); |
| 187 | try { |
| 188 | const size = await putStream(env.BUCKET as unknown as BlobBucket, key, rest, MAX_RECEIVE_PACK_REQUEST); |
| 189 | return { spilled: key, size }; |
| 190 | } catch (e) { |
| 191 | await env.BUCKET.delete(key).catch(() => {}); |
| 192 | if (e instanceof BodyTooLarge) throw new HttpError(413, tooLarge); |
| 193 | throw e; |
| 194 | } |
| 195 | } |
| 196 | } |
| 197 | const out = new Uint8Array(total); |
| 198 | let at = 0; |
| 199 | for (const ch of chunks) { |
| 200 | out.set(ch, at); |
| 201 | at += ch.length; |
| 202 | } |
| 203 | return out; |
| 204 | } |
| 205 | |
| 206 | // ---------------------------------------------------------------- Git LFS |
| 207 | |
| 208 | const lfsKey = (repoId: string, oid: string) => `repos/${repoId}/lfs/${oid.slice(0, 2)}/${oid.slice(2, 4)}/${oid}`; |
| 209 | const LFS_TYPE = "application/vnd.git-lfs+json"; |
| 210 | |
| 211 | git.post("/:owner/:repo/info/lfs/objects/batch", async (c) => { |
| 212 | const req = (await c.req.json()) as { operation: "upload" | "download"; objects: Array<{ oid: string; size: number }> }; |
| 213 | const { repo, stub } = await loadRepo(c, c.req.param("owner"), c.req.param("repo"), req.operation === "upload" ? "write" : "read", { readOnlyPost: req.operation !== "upload", follow: true }); |
| 214 | const base = `${c.env.PUBLIC_URL}/${c.req.param("owner")}/${c.req.param("repo")}/info/lfs/objects`; |
| 215 | const auth = c.req.header("authorization"); |
| 216 | const objects = []; |
| 217 | const uploads: Array<{ oid: string; size: number }> = []; |
| 218 | for (const o of req.objects.slice(0, 1000)) { |
| 219 | if (!/^[0-9a-f]{64}$/.test(o.oid)) { |
| 220 | objects.push({ oid: o.oid, size: o.size, error: { code: 422, message: "invalid oid" } }); |
| 221 | continue; |
| 222 | } |
| 223 | const head = await c.env.BUCKET.head(lfsKey(repo.id, o.oid)); |
| 224 | if (req.operation === "download") { |
| 225 | if (!head) objects.push({ oid: o.oid, size: o.size, error: { code: 404, message: "object not found" } }); |
| 226 | else objects.push({ oid: o.oid, size: head.size, authenticated: true, actions: { download: { href: `${base}/${o.oid}`, header: auth ? { Authorization: auth } : {}, expires_in: 3600 } } }); |
| 227 | } else if (head) objects.push({ oid: o.oid, size: o.size, authenticated: true }); |
| 228 | else uploads.push(o); |
| 229 | } |
| 230 | // Past the owner's storage quota (lib/quotas.ts) no new object is taken: each gets the reason instead. |
| 231 | const full = uploads.length ? await stub.storageRefusal(uploads.reduce((n, o) => n + (Number(o.size) || 0), 0)) : null; |
| 232 | for (const o of uploads) objects.push(full ? { oid: o.oid, size: o.size, error: { code: 507, message: full } } : { oid: o.oid, size: o.size, authenticated: true, actions: { upload: { href: `${base}/${o.oid}`, header: auth ? { Authorization: auth } : {}, expires_in: 3600 } } }); |
| 233 | return c.body(JSON.stringify({ transfer: "basic", objects, hash_algo: "sha256" }), 200, { "content-type": LFS_TYPE }); |
| 234 | }); |
| 235 | |
| 236 | git.get("/:owner/:repo/info/lfs/objects/:oid", async (c) => { |
| 237 | const { repo } = await loadRepo(c, c.req.param("owner"), c.req.param("repo"), "read", { follow: true }); |
| 238 | const obj = await c.env.BUCKET.get(lfsKey(repo.id, c.req.param("oid"))); |
| 239 | if (!obj) throw new HttpError(404, "object not found"); |
| 240 | return new Response(obj.body, { headers: { "content-type": "application/octet-stream", "content-length": String(obj.size) } }); |
| 241 | }); |
| 242 | |
| 243 | git.put("/:owner/:repo/info/lfs/objects/:oid", async (c) => { |
| 244 | const { repo, stub } = await loadRepo(c, c.req.param("owner"), c.req.param("repo"), "write", { follow: true }); |
| 245 | const oid = c.req.param("oid"); |
| 246 | if (!/^[0-9a-f]{64}$/.test(oid)) throw new HttpError(422, "invalid oid"); |
| 247 | const length = Number(c.req.header("content-length")); |
| 248 | const full = await stub.storageRefusal(Number.isFinite(length) ? length : 0); |
| 249 | if (full) throw new HttpError(507, full); |
| 250 | // Verify the content hash while storing: R2 checks the SHA-256 we pass. |
| 251 | const body = await c.req.arrayBuffer(); |
| 252 | const digest = [...new Uint8Array(await crypto.subtle.digest("SHA-256", body))].map((b) => b.toString(16).padStart(2, "0")).join(""); |
| 253 | if (digest !== oid) throw new HttpError(422, "content does not match oid"); |
| 254 | await c.env.BUCKET.put(lfsKey(repo.id, oid), body); |
| 255 | await stub.recordLfs(oid, body.byteLength); |
| 256 | return c.body(null, 200); |
| 257 | }); |