| 1 | import { Hono } from "hono"; |
| 2 | import type { JobCompletion, LogLine, StepReport } from "@diffinery/shared"; |
| 3 | import type { Env } from "../env.js"; |
| 4 | import type { AppContext, AppVars } from "../auth/principal.js"; |
| 5 | import { credentialFrom } from "../auth/principal.js"; |
| 6 | import { signEs256 } from "../lib/jwt.js"; |
| 7 | import { BodyTooLarge, putStream, sized, type BlobBucket } from "../lib/blob-store.js"; |
| 8 | import { getFirst, headFirst } from "../lib/storage-keys.js"; |
| 9 | import { HttpError } from "../lib/util.js"; |
| 10 | import { mirrorKey, parseActionRef, refreshMirror, type MirrorBucket } from "../lib/action-mirror.js"; |
| 11 | import { parseRunnerInfo } from "../repo/actions.js"; |
| 12 | import { CACHE_LIMIT_BYTES } from "../repo/runtime.js"; |
| 13 | import { repoStub } from "./common.js"; |
| 14 | |
| 15 | /** Endpoints the runner agent inside a job container calls. Auth: the job token. */ |
| 16 | export const runner = new Hono<{ Bindings: Env; Variables: AppVars }>(); |
| 17 | |
| 18 | /** The job token's shape only: which repository to ask. Every handler must still have the repository verify the token. */ |
| 19 | function jobAuth(req: Request, jobId: string): { token: string; repoId: string } { |
| 20 | const token = credentialFrom(req); |
| 21 | if (!token?.startsWith("djt_")) throw new HttpError(401, "job token required"); |
| 22 | const [, repoId, tokJob] = token.split("_"); |
| 23 | if (!repoId || tokJob !== jobId) throw new HttpError(401, "token does not match job"); |
| 24 | return { token, repoId }; |
| 25 | } |
| 26 | |
| 27 | /** Stores a request body as it streams in, refusing more than `max` bytes (older runners' cache and artifact uploads). */ |
| 28 | async function storeBody(c: AppContext, key: string, max: number): Promise<number> { |
| 29 | const body = c.req.raw.body; |
| 30 | if (!body) throw new HttpError(400, "empty body"); |
| 31 | const options = { httpMetadata: { contentType: "application/gzip" } }; |
| 32 | const len = c.req.header("content-length"); |
| 33 | if (len !== undefined && /^\d{1,15}$/.test(len)) { |
| 34 | if (Number(len) > max) throw new HttpError(413, `uploads are limited to ${max} bytes`); |
| 35 | return (await c.env.BUCKET.put(key, sized(body, Number(len)), options)).size; |
| 36 | } |
| 37 | try { |
| 38 | return await putStream(c.env.BUCKET as unknown as BlobBucket, key, body, max, options); |
| 39 | } catch (e) { |
| 40 | if (e instanceof BodyTooLarge) throw new HttpError(413, `uploads are limited to ${max} bytes`); |
| 41 | throw e; |
| 42 | } |
| 43 | } |
| 44 | |
| 45 | runner.get("/runner/jobs/:jobId", async (c) => { |
| 46 | const jobId = c.req.param("jobId"); |
| 47 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 48 | // This Worker's runner image, not the repository object's: after a deploy the object can run the previous version for minutes. |
| 49 | return c.json(await repoStub(c.env, repoId).runnerSpec(jobId, token, parseRunnerInfo(c.req.header("x-diffinery-runner")), c.env.RUNNER_IMAGE ?? null)); |
| 50 | }); |
| 51 | |
| 52 | runner.post("/runner/jobs/:jobId/logs", async (c) => { |
| 53 | const jobId = c.req.param("jobId"); |
| 54 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 55 | const { lines } = (await c.req.json()) as { lines: LogLine[] }; |
| 56 | await repoStub(c.env, repoId).runnerLogs(jobId, token, lines ?? []); |
| 57 | return c.json({ ok: true }); |
| 58 | }); |
| 59 | |
| 60 | runner.post("/runner/jobs/:jobId/steps/:index", async (c) => { |
| 61 | const jobId = c.req.param("jobId"); |
| 62 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 63 | await repoStub(c.env, repoId).runnerStep(jobId, token, Number(c.req.param("index")), (await c.req.json()) as StepReport); |
| 64 | return c.json({ ok: true }); |
| 65 | }); |
| 66 | |
| 67 | runner.post("/runner/jobs/:jobId/complete", async (c) => { |
| 68 | const jobId = c.req.param("jobId"); |
| 69 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 70 | await repoStub(c.env, repoId).runnerComplete(jobId, token, (await c.req.json()) as JobCompletion, c.env.RUNNER_IMAGE ?? null); |
| 71 | return c.json({ ok: true }); |
| 72 | }); |
| 73 | |
| 74 | // ---- cache (actions/cache compatible semantics, backed by R2) |
| 75 | |
| 76 | runner.get("/runner/jobs/:jobId/cache", async (c) => { |
| 77 | const jobId = c.req.param("jobId"); |
| 78 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 79 | const key = c.req.query("key") ?? ""; |
| 80 | const restore = (c.req.query("restore") ?? "").split(",").map((s) => s.trim()).filter(Boolean); |
| 81 | const hit = await repoStub(c.env, repoId).cacheLookup(jobId, token, key, restore); |
| 82 | if (!hit) return c.json({ message: "cache miss" }, 404); |
| 83 | return c.json({ key: hit.key, url: `${new URL(c.req.url).origin}/runner/jobs/${jobId}/cache/blob/${encodeURIComponent(hit.stored)}` }); |
| 84 | }); |
| 85 | |
| 86 | runner.get("/runner/jobs/:jobId/cache/blob/:key", async (c) => { |
| 87 | const jobId = c.req.param("jobId"); |
| 88 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 89 | // Only entries of a scope the job may restore from (its ref, the PR base, the default branch). |
| 90 | const keys = await repoStub(c.env, repoId).cacheReadKeys(jobId, token, c.req.param("key")); |
| 91 | const obj = keys && (await getFirst(c.env.BUCKET, keys)); |
| 92 | if (!obj) throw new HttpError(404, "cache entry missing"); |
| 93 | return new Response(obj.body, { headers: { "content-type": "application/gzip", "content-length": String(obj.size) } }); |
| 94 | }); |
| 95 | |
| 96 | runner.put("/runner/jobs/:jobId/cache/:key", async (c) => { |
| 97 | const jobId = c.req.param("jobId"); |
| 98 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 99 | const key = c.req.param("key"); |
| 100 | if (key.length > 512) throw new HttpError(422, "cache key too long"); |
| 101 | const stub = repoStub(c.env, repoId); |
| 102 | const { r2Keys: keys } = await stub.cacheSaveKeys(jobId, token, key); |
| 103 | if (await headFirst(c.env.BUCKET, keys)) return c.json({ message: "cache entry already exists" }, 409); |
| 104 | const size = await storeBody(c, keys[0]!, CACHE_LIMIT_BYTES); |
| 105 | const evicted = await stub.cacheSaved(jobId, token, key, size); |
| 106 | if (evicted.length) await c.env.BUCKET.delete(evicted); |
| 107 | return c.json({ ok: true }); |
| 108 | }); |
| 109 | |
| 110 | // ---- artifacts |
| 111 | |
| 112 | runner.put("/runner/jobs/:jobId/artifacts/:name", async (c) => { |
| 113 | const jobId = c.req.param("jobId"); |
| 114 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 115 | const stub = repoStub(c.env, repoId); |
| 116 | // Authenticates the job before anything is written: jobAuth only reads the token's shape. |
| 117 | // The artifact is recorded with the artifact service's (numeric ids, retention, the REST API) |
| 118 | // but stored at its legacy key; `room` is what is left of the repository's artifact quota. |
| 119 | const { r2Keys: keys, room } = await stub.artifactUploadKeys(jobId, token, c.req.param("name")); |
| 120 | if (room === 0) throw new HttpError(413, "the repository's artifact storage quota is used up; delete artifacts or wait for them to expire"); |
| 121 | const size = await storeBody(c, keys[0]!, room); |
| 122 | await stub.artifactSaved(jobId, token, c.req.param("name"), size); |
| 123 | return c.json({ ok: true }); |
| 124 | }); |
| 125 | |
| 126 | runner.get("/runner/jobs/:jobId/artifacts/:name", async (c) => { |
| 127 | const jobId = c.req.param("jobId"); |
| 128 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 129 | const obj = await getFirst(c.env.BUCKET, await repoStub(c.env, repoId).artifactKeysFor(jobId, token, c.req.param("name"))); |
| 130 | if (!obj) throw new HttpError(404, "artifact not found"); |
| 131 | return new Response(obj.body, { headers: { "content-type": "application/gzip" } }); |
| 132 | }); |
| 133 | |
| 134 | // ---- Diffinery's copies of GitHub actions, for when GitHub is down (lib/action-mirror.ts) |
| 135 | |
| 136 | /** The runner downloaded this action from GitHub: keep a recent copy of it. */ |
| 137 | runner.post("/runner/jobs/:jobId/action-mirror", async (c) => { |
| 138 | const jobId = c.req.param("jobId"); |
| 139 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 140 | const action = parseActionRef((await c.req.json().catch(() => ({}))) as Record<string, unknown>); |
| 141 | if (!action) throw new HttpError(422, "invalid action"); |
| 142 | await repoStub(c.env, repoId).requireRunningJob(jobId, token); |
| 143 | const work = refreshMirror(c.env.BUCKET as unknown as MirrorBucket, action); |
| 144 | // The runner doesn't wait for this: its job can end, and drop the request, before a large action is stored. |
| 145 | c.executionCtx.waitUntil(work); |
| 146 | return c.json({ result: await work }); |
| 147 | }); |
| 148 | |
| 149 | /** GitHub failed: the stored copy of an action, and when it was fetched from GitHub. */ |
| 150 | runner.get("/runner/jobs/:jobId/action-mirror", async (c) => { |
| 151 | const jobId = c.req.param("jobId"); |
| 152 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 153 | const action = parseActionRef({ owner: c.req.query("owner"), repo: c.req.query("repo"), ref: c.req.query("ref") }); |
| 154 | if (!action) throw new HttpError(422, "invalid action"); |
| 155 | await repoStub(c.env, repoId).requireRunningJob(jobId, token); |
| 156 | const obj = await c.env.BUCKET.get(mirrorKey(action)); |
| 157 | if (!obj) throw new HttpError(404, "no copy of this action"); |
| 158 | return new Response(obj.body, { headers: { "content-type": "application/gzip", "content-length": String(obj.size), "x-diffinery-fetched-at": obj.uploaded.toISOString() } }); |
| 159 | }); |
| 160 | |
| 161 | // ---- OIDC id tokens (actions/core getIDToken compatible: GET ?audience= → { value }) |
| 162 | |
| 163 | const oidc = async (c: AppContext) => { |
| 164 | const jobId = c.req.param("jobId")!; |
| 165 | const { token, repoId } = jobAuth(c.req.raw, jobId); |
| 166 | if (!c.env.OIDC_PRIVATE_KEY) throw new HttpError(501, "OIDC is not configured on this Diffinery instance"); |
| 167 | const claims = await repoStub(c.env, repoId).oidcClaims(jobId, token); |
| 168 | if (!claims.allowed) throw new HttpError(403, "this job lacks permissions: id-token: write"); |
| 169 | const g = claims.github; |
| 170 | const nowS = Math.floor(Date.now() / 1000); |
| 171 | const value = await signEs256(JSON.parse(c.env.OIDC_PRIVATE_KEY), { |
| 172 | iss: c.env.PUBLIC_URL, |
| 173 | aud: c.req.query("audience") ?? c.env.PUBLIC_URL, |
| 174 | // GitHub's forms: a job in an environment, a pull request, or a branch or tag. |
| 175 | sub: g.environment ? `repo:${g.repository}:environment:${g.environment}` : g.event_name === "pull_request" ? `repo:${g.repository}:pull_request` : `repo:${g.repository}:ref:${g.ref}`, |
| 176 | iat: nowS, |
| 177 | nbf: nowS, |
| 178 | exp: nowS + 600, |
| 179 | repository: g.repository, |
| 180 | repository_id: g.repository_id, |
| 181 | repository_owner: g.repository_owner, |
| 182 | repository_owner_id: g.repository_owner_id, |
| 183 | actor_id: g.actor_id, |
| 184 | environment: g.environment, |
| 185 | // Diffinery's own claim: whether the code that runs came from an agent. |
| 186 | agent_origin: g.agent_origin === "true", |
| 187 | ref_type: g.ref_type, |
| 188 | head_ref: g.head_ref || undefined, |
| 189 | base_ref: g.base_ref || undefined, |
| 190 | run_number: g.run_number, |
| 191 | run_attempt: g.run_attempt, |
| 192 | ref: g.ref, |
| 193 | sha: g.sha, |
| 194 | workflow: g.workflow, |
| 195 | event_name: g.event_name, |
| 196 | run_id: g.run_id, |
| 197 | actor: g.actor, |
| 198 | job_workflow_ref: g.workflow_ref, |
| 199 | }); |
| 200 | return c.json({ value }); |
| 201 | }; |
| 202 | runner.get("/runner/jobs/:jobId/oidc", oidc); |
| 203 | runner.post("/runner/jobs/:jobId/oidc", oidc); |