| 1 | import { Hono } from "hono"; |
| 2 | import type { Env } from "../env.js"; |
| 3 | import type { AppContext, AppVars } from "../auth/principal.js"; |
| 4 | import { blockKey, chunkKey, compose, deleteKeys, finalKey, listStaged, parseBlockList, parseRange, sized, stagingPrefix, type BlobBucket, type ComposeStep } from "../lib/blob-store.js"; |
| 5 | import { signBlobUrl, verifyBlobUrl, verifyRuntimeToken, type BlobRef, type RuntimeClaims } from "../lib/runtime-auth.js"; |
| 6 | import { HttpError, newId } from "../lib/util.js"; |
| 7 | import { headFirst } from "../lib/storage-keys.js"; |
| 8 | import { MAX_UPLOAD_PART, artifactFileName, artifactObjectKeys, type ArtifactEntry, type CacheEntry } from "../repo/runtime.js"; |
| 9 | import { auditRepo, loadRepo, repoStub } from "./common.js"; |
| 10 | import { pageOf, paginate, setLinks } from "./gh.js"; |
| 11 | |
| 12 | /** |
| 13 | * GitHub's Actions runtime services, so the official actions and the toolkit |
| 14 | * (`@actions/cache`, `@actions/artifact`, buildkit's `type=gha` cache) work |
| 15 | * against Diffinery unchanged: |
| 16 | * |
| 17 | * - Twirp `CacheService` (cache v2) and `ArtifactService` (artifacts v4) at |
| 18 | * `ACTIONS_RESULTS_URL`, and the legacy cache REST API at `ACTIONS_CACHE_URL`. |
| 19 | * All authenticate with the job's ACTIONS_RUNTIME_TOKEN and work only while |
| 20 | * that job runs (repo/runtime.ts decides what may be read or written). |
| 21 | * - Signed blob URLs that speak the parts of Azure Blob Storage the clients use |
| 22 | * (Put Block, Put Block List, Put Blob, ranged GET), backed by R2. |
| 23 | * - The REST API's artifacts and caches endpoints (`actionsRest`, under /api/v3). |
| 24 | */ |
| 25 | export const actionsRuntime = new Hono<{ Bindings: Env; Variables: AppVars }>(); |
| 26 | export const actionsRest = new Hono<{ Bindings: Env; Variables: AppVars }>(); |
| 27 | |
| 28 | const READ_URL_TTL_S = 3600; |
| 29 | const ZIP = "application/zip"; |
| 30 | |
| 31 | const bucket = (env: Env) => env.BUCKET as unknown as BlobBucket; |
| 32 | const origin = (env: Env) => new URL(env.PUBLIC_URL).origin; |
| 33 | const nowS = () => Math.floor(Date.now() / 1000); |
| 34 | |
| 35 | async function runtimeClaims(c: AppContext): Promise<RuntimeClaims> { |
| 36 | const m = /^Bearer\s+(\S+)$/i.exec(c.req.header("authorization") ?? ""); |
| 37 | const claims = m ? await verifyRuntimeToken(c.env, m[1]!) : null; |
| 38 | if (!claims) throw new HttpError(401, "invalid or expired runtime token"); |
| 39 | return claims; |
| 40 | } |
| 41 | |
| 42 | /** Status and message of an error thrown here or inside the Durable Object. */ |
| 43 | function failure(e: unknown): { status: number; message: string } { |
| 44 | if (e instanceof HttpError) return { status: e.status, message: e.detail }; |
| 45 | const msg = (e as Error)?.message ?? String(e); |
| 46 | const m = /^(\d{3}): ([\s\S]*)$/.exec(msg); |
| 47 | if (m) return { status: Number(m[1]), message: m[2]! }; |
| 48 | console.error(e); |
| 49 | return { status: 500, message: "internal error" }; |
| 50 | } |
| 51 | |
| 52 | const readUrl = (env: Env, b: BlobRef, fileName?: string) => signBlobUrl(env, origin(env), b, "r", nowS() + READ_URL_TTL_S, fileName); |
| 53 | |
| 54 | /** |
| 55 | * A cache lookup whose object is really there. An entry whose object went missing |
| 56 | * (or was replaced after it was committed) is dropped, and the lookup tries again. |
| 57 | */ |
| 58 | async function liveCacheHit(env: Env, claims: RuntimeClaims, keys: string[], version: string): Promise<CacheEntry | null> { |
| 59 | const stub = repoStub(env, claims.repo); |
| 60 | for (let i = 0; i < 5; i++) { |
| 61 | const hit = await stub.rtCacheLookup(claims.job, keys, version); |
| 62 | if (!hit) return null; |
| 63 | const head = await env.BUCKET.head(finalKey({ repo: claims.repo, kind: "cache", id: hit.id })); |
| 64 | if (head && (!hit.etag || head.etag === hit.etag)) return hit; |
| 65 | await stub.rtCacheMissing(hit.id); |
| 66 | } |
| 67 | return null; |
| 68 | } |
| 69 | |
| 70 | /** Composes an upload's staged pieces, each step in a Durable Object of its own (compose.ts, lib/blob-store.ts). */ |
| 71 | function composeUpload(env: Env, b: BlobRef, segments: Array<{ key: string; size: number }>, contentType: string): Promise<number> { |
| 72 | const stub = repoStub(env, b.repo); |
| 73 | return compose(bucket(env), finalKey(b), segments, contentType, { |
| 74 | scratch: stagingPrefix(b), |
| 75 | run: (step: ComposeStep) => (env.COMPOSE ? env.COMPOSE.get(env.COMPOSE.newUniqueId()).step(b, step, contentType) : stub.rtComposeStep({ kind: b.kind, id: b.id }, step, contentType)), |
| 76 | }); |
| 77 | } |
| 78 | |
| 79 | /** A request's declared length; null without Content-Length. */ |
| 80 | function declaredLength(c: AppContext): number | null { |
| 81 | const v = c.req.header("content-length"); |
| 82 | if (v === undefined) return null; |
| 83 | if (!/^\d{1,15}$/.test(v)) throw new HttpError(400, "invalid Content-Length"); |
| 84 | return Number(v); |
| 85 | } |
| 86 | |
| 87 | /** A small text body, refused past `max` bytes (with or without Content-Length). */ |
| 88 | async function smallText(c: AppContext, max: number): Promise<string> { |
| 89 | const len = declaredLength(c); |
| 90 | if (len !== null && len > max) throw new HttpError(413, "the request body is too large"); |
| 91 | const body = c.req.raw.body; |
| 92 | if (!body) return ""; |
| 93 | const reader = body.getReader(); |
| 94 | const chunks: Uint8Array[] = []; |
| 95 | let n = 0; |
| 96 | for (;;) { |
| 97 | const { done, value } = await reader.read(); |
| 98 | if (done) break; |
| 99 | n += value.length; |
| 100 | if (n > max) { |
| 101 | await reader.cancel().catch(() => {}); |
| 102 | throw new HttpError(413, "the request body is too large"); |
| 103 | } |
| 104 | chunks.push(value); |
| 105 | } |
| 106 | const out = new Uint8Array(n); |
| 107 | let at = 0; |
| 108 | for (const ch of chunks) { |
| 109 | out.set(ch, at); |
| 110 | at += ch.length; |
| 111 | } |
| 112 | return new TextDecoder().decode(out); |
| 113 | } |
| 114 | |
| 115 | // ---------------------------------------------------------------- Twirp services |
| 116 | |
| 117 | const TWIRP_CODES: Record<number, string> = { 400: "invalid_argument", 401: "unauthenticated", 403: "permission_denied", 404: "not_found", 409: "already_exists", 412: "failed_precondition", 429: "resource_exhausted" }; |
| 118 | |
| 119 | type TwirpRequest = Record<string, unknown>; |
| 120 | |
| 121 | /** A request field by its proto name or its JSON (camelCase) name. */ |
| 122 | function field(req: TwirpRequest, name: string): unknown { |
| 123 | return req[name] ?? req[name.replace(/_([a-z])/g, (_, ch: string) => ch.toUpperCase())]; |
| 124 | } |
| 125 | |
| 126 | function str(req: TwirpRequest, name: string, max = 1024): string { |
| 127 | const v = field(req, name) ?? ""; |
| 128 | if (typeof v !== "string" || v.length > max) throw new HttpError(400, `${name} must be a string of at most ${max} characters`); |
| 129 | return v; |
| 130 | } |
| 131 | |
| 132 | /** int64 values arrive as strings (protobuf JSON) or numbers (Go clients). */ |
| 133 | function int64(req: TwirpRequest, name: string): number | undefined { |
| 134 | const v = field(req, name); |
| 135 | if (v === undefined || v === null || v === "") return undefined; |
| 136 | const n = typeof v === "number" ? v : typeof v === "string" && /^\d{1,16}$/.test(v) ? Number(v) : NaN; |
| 137 | if (!Number.isSafeInteger(n) || n < 0) throw new HttpError(400, `${name} must be a non-negative integer`); |
| 138 | return n; |
| 139 | } |
| 140 | |
| 141 | function twirp(service: "CacheService" | "ArtifactService", method: string, fn: (c: AppContext, claims: RuntimeClaims, req: TwirpRequest) => Promise<Record<string, unknown>>) { |
| 142 | actionsRuntime.post(`/twirp/github.actions.results.api.v1.${service}/${method}`, async (c) => { |
| 143 | try { |
| 144 | const claims = await runtimeClaims(c); |
| 145 | let req: unknown; |
| 146 | try { |
| 147 | req = await c.req.json(); |
| 148 | } catch { |
| 149 | throw new HttpError(400, "the request body must be JSON"); |
| 150 | } |
| 151 | if (!req || typeof req !== "object" || Array.isArray(req)) throw new HttpError(400, "the request body must be a JSON object"); |
| 152 | return c.json(await fn(c, claims, req as TwirpRequest)); |
| 153 | } catch (e) { |
| 154 | const { status, message } = failure(e); |
| 155 | return c.json({ code: TWIRP_CODES[status] ?? "internal", msg: message }, status as 400); |
| 156 | } |
| 157 | }); |
| 158 | } |
| 159 | |
| 160 | /** Reads name the run; the job may be the one that uploaded (download-artifact sends the artifact's ids). */ |
| 161 | function ownRun(claims: RuntimeClaims, req: TwirpRequest): void { |
| 162 | const run = field(req, "workflow_run_backend_id"); |
| 163 | if (run !== undefined && run !== claims.run) throw new HttpError(403, "the request names another workflow run"); |
| 164 | } |
| 165 | |
| 166 | /** Writes name the run and job; they must be the token's own. */ |
| 167 | function ownJob(claims: RuntimeClaims, req: TwirpRequest): void { |
| 168 | const run = field(req, "workflow_run_backend_id"); |
| 169 | const job = field(req, "workflow_job_run_backend_id"); |
| 170 | if ((run !== undefined && run !== claims.run) || (job !== undefined && job !== claims.job)) throw new HttpError(403, "the request names another workflow run or job"); |
| 171 | } |
| 172 | |
| 173 | twirp("CacheService", "CreateCacheEntry", async (c, claims, req) => { |
| 174 | const reserved = await repoStub(c.env, claims.repo).rtCacheReserve(claims.job, str(req, "key", 512), str(req, "version", 128)); |
| 175 | if (!reserved) throw new HttpError(409, "a cache entry with this key and version already exists"); |
| 176 | const url = await signBlobUrl(c.env, origin(c.env), { repo: claims.repo, kind: "cache", id: reserved.id }, "w", claims.exp); |
| 177 | return { ok: true, signed_upload_url: url }; |
| 178 | }); |
| 179 | |
| 180 | twirp("CacheService", "FinalizeCacheEntryUpload", async (c, claims, req) => { |
| 181 | const stub = repoStub(c.env, claims.repo); |
| 182 | const entry = await stub.rtCacheReservation(claims.job, { key: str(req, "key", 512), version: str(req, "version", 128) }); |
| 183 | if (!entry) throw new HttpError(404, "no upload of this cache entry is in progress"); |
| 184 | const obj = await c.env.BUCKET.head(finalKey({ repo: claims.repo, kind: "cache", id: entry.id })); |
| 185 | if (!obj) return { ok: false, entry_id: "0", message: "the cache archive was not uploaded" }; |
| 186 | const size = int64(req, "size_bytes"); |
| 187 | if (size !== undefined && size !== obj.size) return { ok: false, entry_id: "0", message: `the uploaded archive has ${obj.size} bytes, not ${size}` }; |
| 188 | const done = await stub.rtCacheCommit(claims.job, entry.id, obj.size, obj.etag); |
| 189 | return { ok: true, entry_id: String(done.id) }; |
| 190 | }); |
| 191 | |
| 192 | twirp("CacheService", "GetCacheEntryDownloadURL", async (c, claims, req) => { |
| 193 | const restore = field(req, "restore_keys"); |
| 194 | if (restore !== undefined && (!Array.isArray(restore) || restore.length > 20 || restore.some((k) => typeof k !== "string" || k.length > 512))) throw new HttpError(400, "restore_keys must be a list of at most 20 keys"); |
| 195 | const keys = [...new Set([str(req, "key", 512), ...((restore as string[] | undefined) ?? [])].filter(Boolean))]; |
| 196 | if (!keys.length) throw new HttpError(400, "key is required"); |
| 197 | const hit = await liveCacheHit(c.env, claims, keys, str(req, "version", 128)); |
| 198 | if (!hit) return { ok: false, signed_download_url: "", matched_key: "" }; |
| 199 | return { ok: true, signed_download_url: await readUrl(c.env, { repo: claims.repo, kind: "cache", id: hit.id }), matched_key: hit.key }; |
| 200 | }); |
| 201 | |
| 202 | const artifactName = (req: TwirpRequest) => str(req, "name", 256); |
| 203 | |
| 204 | twirp("ArtifactService", "CreateArtifact", async (c, claims, req) => { |
| 205 | ownJob(claims, req); |
| 206 | const mime = field(req, "mime_type"); |
| 207 | const mimeType = typeof mime === "string" && /^[\w.+-]{1,64}\/[\w.+-]{1,64}$/.test(mime) ? mime : ZIP; |
| 208 | const expires = field(req, "expires_at"); |
| 209 | const { id } = await repoStub(c.env, claims.repo).rtArtifactCreate(claims.job, artifactName(req), mimeType, typeof expires === "string" ? expires : undefined); |
| 210 | const url = await signBlobUrl(c.env, origin(c.env), { repo: claims.repo, kind: "artifact", id }, "w", claims.exp); |
| 211 | return { ok: true, signed_upload_url: url }; |
| 212 | }); |
| 213 | |
| 214 | twirp("ArtifactService", "FinalizeArtifact", async (c, claims, req) => { |
| 215 | ownJob(claims, req); |
| 216 | const stub = repoStub(c.env, claims.repo); |
| 217 | const name = artifactName(req); |
| 218 | const upload = await stub.rtArtifactUpload(claims.job, name); |
| 219 | if (!upload) throw new HttpError(404, `no upload of artifact '${name}' is in progress`); |
| 220 | const obj = await c.env.BUCKET.head(finalKey({ repo: claims.repo, kind: "artifact", id: upload.id })); |
| 221 | if (!obj) throw new HttpError(412, "the artifact's content was not uploaded"); |
| 222 | const size = int64(req, "size"); |
| 223 | if (size !== undefined && size !== obj.size) throw new HttpError(412, `the uploaded artifact has ${obj.size} bytes, not ${size}`); |
| 224 | const hash = field(req, "hash"); |
| 225 | const digest = typeof hash === "string" && /^sha256:[0-9a-f]{64}$/.test(hash) ? hash : undefined; |
| 226 | const done = await stub.rtArtifactFinalize(claims.job, name, obj.size, digest, obj.etag); |
| 227 | return { ok: true, artifact_id: String(done.id) }; |
| 228 | }); |
| 229 | |
| 230 | twirp("ArtifactService", "ListArtifacts", async (c, claims, req) => { |
| 231 | ownRun(claims, req); |
| 232 | const nameFilter = field(req, "name_filter"); |
| 233 | const list = await repoStub(c.env, claims.repo).rtJobArtifacts(claims.job, { name: typeof nameFilter === "string" ? nameFilter : undefined, id: int64(req, "id_filter") }); |
| 234 | return { |
| 235 | artifacts: list.map((a) => ({ |
| 236 | workflow_run_backend_id: a.runId, |
| 237 | workflow_job_run_backend_id: a.jobId, |
| 238 | database_id: String(a.id), |
| 239 | name: a.name, |
| 240 | size: String(a.size), |
| 241 | created_at: a.createdAt, |
| 242 | ...(a.digest ? { digest: a.digest } : {}), |
| 243 | })), |
| 244 | }; |
| 245 | }); |
| 246 | |
| 247 | twirp("ArtifactService", "GetSignedArtifactURL", async (c, claims, req) => { |
| 248 | ownRun(claims, req); |
| 249 | const name = artifactName(req); |
| 250 | const a = (await repoStub(c.env, claims.repo).rtJobArtifacts(claims.job, { name }))[0]; |
| 251 | if (!a) throw new HttpError(404, `artifact '${name}' not found`); |
| 252 | return { signed_url: await readUrl(c.env, { repo: claims.repo, kind: "artifact", id: a.id }, artifactFileName(a)) }; |
| 253 | }); |
| 254 | |
| 255 | twirp("ArtifactService", "DeleteArtifact", async (c, claims, req) => { |
| 256 | ownJob(claims, req); |
| 257 | const name = artifactName(req); |
| 258 | const deleted = await repoStub(c.env, claims.repo).rtArtifactDeleteByName(claims.job, name); |
| 259 | if (!deleted) throw new HttpError(404, `artifact '${name}' not found`); |
| 260 | return { ok: true, artifact_id: String(deleted.id) }; |
| 261 | }); |
| 262 | |
| 263 | // ---------------------------------------------------------------- legacy cache API (ACTIONS_CACHE_URL) |
| 264 | |
| 265 | const V1 = "/_runtime/_apis/artifactcache"; |
| 266 | |
| 267 | /** Errors in the shape the legacy API used (clients look at `typeKey` and `message`). */ |
| 268 | function v1(fn: (c: AppContext, claims: RuntimeClaims) => Promise<Response>) { |
| 269 | return async (c: AppContext) => { |
| 270 | try { |
| 271 | return await fn(c, await runtimeClaims(c)); |
| 272 | } catch (e) { |
| 273 | const { status, message } = failure(e); |
| 274 | const typeKey = status === 409 ? "ArtifactCacheItemAlreadyExistsException" : status === 404 ? "ArtifactCacheNotFoundException" : "ArtifactCacheException"; |
| 275 | return c.json({ $id: "1", innerException: null, message, typeName: `Microsoft.Azure.DevOps.ArtifactCache.WebApi.${typeKey}, Microsoft.Azure.DevOps.ArtifactCache.WebApi`, typeKey, errorCode: 0, eventId: 3000 }, status as 400); |
| 276 | } |
| 277 | }; |
| 278 | } |
| 279 | |
| 280 | const v1Cache = (e: CacheEntry) => ({ cacheKey: e.key, scope: e.scope, cacheVersion: e.version, creationTime: e.createdAt }); |
| 281 | |
| 282 | actionsRuntime.get( |
| 283 | `${V1}/cache`, |
| 284 | v1(async (c, claims) => { |
| 285 | const keys = (c.req.query("keys") ?? "").split(",").map((k) => k.trim()).filter(Boolean); |
| 286 | if (!keys.length || keys.length > 20 || keys.some((k) => k.length > 512)) throw new HttpError(400, "keys must list 1 to 20 cache keys"); |
| 287 | const hit = await liveCacheHit(c.env, claims, keys, (c.req.query("version") ?? "").slice(0, 128)); |
| 288 | if (!hit) return c.body(null, 204); |
| 289 | return c.json({ ...v1Cache(hit), archiveLocation: await readUrl(c.env, { repo: claims.repo, kind: "cache", id: hit.id }) }); |
| 290 | }), |
| 291 | ); |
| 292 | |
| 293 | actionsRuntime.get( |
| 294 | `${V1}/caches`, |
| 295 | v1(async (c, claims) => { |
| 296 | const stub = repoStub(c.env, claims.repo); |
| 297 | const { scopes } = await stub.rtJob(claims.job); |
| 298 | const key = (c.req.query("key") ?? "").slice(0, 512); |
| 299 | const { caches } = await stub.rtListCaches({ key: key || undefined, perPage: 100 }); |
| 300 | const visible = caches.filter((e) => scopes.includes(e.scope)); |
| 301 | return c.json({ totalCount: visible.length, artifactCaches: visible.map(v1Cache) }); |
| 302 | }), |
| 303 | ); |
| 304 | |
| 305 | actionsRuntime.post( |
| 306 | `${V1}/caches`, |
| 307 | v1(async (c, claims) => { |
| 308 | const b = (await c.req.json().catch(() => null)) as { key?: unknown; version?: unknown; cacheSize?: unknown } | null; |
| 309 | if (!b || typeof b.key !== "string" || typeof b.version !== "string") throw new HttpError(400, "key and version are required"); |
| 310 | if (typeof b.cacheSize === "number" && b.cacheSize > 10 * 1024 ** 3) throw new HttpError(400, "the cache is larger than the 10 GB limit"); |
| 311 | const reserved = await repoStub(c.env, claims.repo).rtCacheReserve(claims.job, b.key, b.version); |
| 312 | if (!reserved) throw new HttpError(409, `Cache already exists. Key: ${b.key}, Version: ${b.version}`); |
| 313 | return c.json({ cacheId: reserved.id }, 201); |
| 314 | }), |
| 315 | ); |
| 316 | |
| 317 | const cacheIdParam = (c: AppContext) => { |
| 318 | const id = Number(c.req.param("id")); |
| 319 | if (!Number.isSafeInteger(id) || id < 1) throw new HttpError(404, "cache not found"); |
| 320 | return id; |
| 321 | }; |
| 322 | |
| 323 | actionsRuntime.patch( |
| 324 | `${V1}/caches/:id`, |
| 325 | v1(async (c, claims) => { |
| 326 | const id = cacheIdParam(c); |
| 327 | const m = /^bytes (\d{1,15})-(\d{1,15})\/(\*|\d+)$/.exec(c.req.header("content-range") ?? ""); |
| 328 | if (!m) throw new HttpError(400, "Content-Range must be bytes <start>-<end>/*"); |
| 329 | const start = Number(m[1]); |
| 330 | const length = Number(m[2]) - start + 1; |
| 331 | if (length < 1) throw new HttpError(400, "invalid Content-Range"); |
| 332 | const declared = declaredLength(c); |
| 333 | if (declared !== null && declared !== length) throw new HttpError(400, `the chunk has ${declared} bytes but Content-Range says ${length}`); |
| 334 | const stub = repoStub(c.env, claims.repo); |
| 335 | if (!(await stub.rtCacheReservation(claims.job, { id }))) throw new HttpError(404, "no upload of this cache entry is in progress"); |
| 336 | // Counted against the budget before anything is stored; the body streams to R2 |
| 337 | // (@actions/cache sends 32 MiB chunks, four at a time, without Content-Length). |
| 338 | if (!c.req.raw.body) throw new HttpError(400, "the chunk is empty"); |
| 339 | await stub.rtStage("cache", id, length); |
| 340 | try { |
| 341 | await c.env.BUCKET.put(chunkKey({ repo: claims.repo, kind: "cache", id }, start), sized(c.req.raw.body, length)); |
| 342 | } catch { |
| 343 | await stub.rtUnstage("cache", id, length); |
| 344 | throw new HttpError(400, `the chunk does not have the ${length} bytes Content-Range says`); |
| 345 | } |
| 346 | return c.body(null, 204); |
| 347 | }), |
| 348 | ); |
| 349 | |
| 350 | actionsRuntime.post( |
| 351 | `${V1}/caches/:id`, |
| 352 | v1(async (c, claims) => { |
| 353 | const id = cacheIdParam(c); |
| 354 | const b = (await c.req.json().catch(() => null)) as { size?: unknown } | null; |
| 355 | const size = b?.size; |
| 356 | if (typeof size !== "number" || !Number.isSafeInteger(size) || size < 0) throw new HttpError(400, "size is required"); |
| 357 | const stub = repoStub(c.env, claims.repo); |
| 358 | if (!(await stub.rtCacheReservation(claims.job, { id }))) throw new HttpError(404, "no upload of this cache entry is in progress"); |
| 359 | const ref: BlobRef = { repo: claims.repo, kind: "cache", id }; |
| 360 | const staged = await listStaged(bucket(c.env), ref); |
| 361 | // Chunks are keyed by their offset; together they must cover [0, size) exactly. |
| 362 | const chunks = [...staged].flatMap(([key, n]) => { |
| 363 | const off = /\/o-(\d{16})$/.exec(key); |
| 364 | return off ? [{ key, size: n, offset: Number(off[1]) }] : []; |
| 365 | }); |
| 366 | chunks.sort((x, y) => x.offset - y.offset); |
| 367 | let at = 0; |
| 368 | for (const ch of chunks) { |
| 369 | if (ch.offset !== at) throw new HttpError(400, `the uploaded chunks do not cover byte ${at} exactly`); |
| 370 | at += ch.size; |
| 371 | } |
| 372 | if (at !== size) throw new HttpError(400, `${at} bytes were uploaded, not ${size}`); |
| 373 | await stub.rtCheckCommitSize("cache", id, size); |
| 374 | await composeUpload(c.env, ref, chunks, "application/octet-stream"); |
| 375 | await deleteKeys(bucket(c.env), [...staged.keys()]); |
| 376 | const obj = await c.env.BUCKET.head(finalKey(ref)); |
| 377 | if (!obj) throw new HttpError(500, "the composed cache archive is missing"); |
| 378 | await stub.rtCacheCommit(claims.job, id, obj.size, obj.etag); |
| 379 | return c.body(null, 204); |
| 380 | }), |
| 381 | ); |
| 382 | |
| 383 | // ---------------------------------------------------------------- signed blob URLs (Azure Blob subset on R2) |
| 384 | |
| 385 | const BLOB_PATH = "/_runtime/blob/:repo/:kind/:id/:file?"; |
| 386 | |
| 387 | function blobRef(c: AppContext): BlobRef | null { |
| 388 | const { repo, kind, id } = c.req.param() as { repo: string; kind: string; id: string }; |
| 389 | if (!/^[A-Za-z0-9_-]{1,64}$/.test(repo) || (kind !== "cache" && kind !== "artifact") || !/^\d{1,15}$/.test(id)) return null; |
| 390 | return { repo, kind, id: Number(id) }; |
| 391 | } |
| 392 | |
| 393 | /** Headers every Azure Blob response carries. */ |
| 394 | const azureHeaders = (extra: Record<string, string> = {}): Record<string, string> => ({ "x-ms-request-id": newId(), "x-ms-version": "2025-01-05", date: new Date().toUTCString(), ...extra }); |
| 395 | |
| 396 | function azureError(status: number, code: string, message: string): Response { |
| 397 | const xml = `<?xml version="1.0" encoding="utf-8"?><Error><Code>${code}</Code><Message>${message.replace(/[<>&]/g, "")}</Message></Error>`; |
| 398 | return new Response(xml, { status, headers: azureHeaders({ "content-type": "application/xml", "x-ms-error-code": code }) }); |
| 399 | } |
| 400 | |
| 401 | const created = (etag?: string) => new Response(null, { status: 201, headers: azureHeaders({ "x-ms-request-server-encrypted": "true", ...(etag ? { etag, "last-modified": new Date().toUTCString() } : {}) }) }); |
| 402 | |
| 403 | /** |
| 404 | * Stores a block or blob body as it streams in. Like Azure, a length is required |
| 405 | * (411 without one); the bytes are counted against the repository's budget first. |
| 406 | */ |
| 407 | async function putUpload(c: AppContext, b: BlobRef, key: string, contentType?: string): Promise<{ etag: string; httpEtag: string } | Response> { |
| 408 | const len = declaredLength(c); |
| 409 | if (len === null) return azureError(411, "MissingContentLengthHeader", "Content-Length is required."); |
| 410 | if (len > MAX_UPLOAD_PART) return azureError(413, "RequestBodyTooLarge", `A single request may carry at most ${MAX_UPLOAD_PART} bytes.`); |
| 411 | const stub = repoStub(c.env, b.repo); |
| 412 | await stub.rtStage(b.kind, b.id, len); |
| 413 | const body = len > 0 && c.req.raw.body ? sized(c.req.raw.body, len) : new Uint8Array(0); |
| 414 | try { |
| 415 | const put = await c.env.BUCKET.put(key, body, contentType ? { httpMetadata: { contentType } } : undefined); |
| 416 | return { etag: put.etag, httpEtag: put.httpEtag }; |
| 417 | } catch (e) { |
| 418 | await stub.rtUnstage(b.kind, b.id, len); |
| 419 | throw e; |
| 420 | } |
| 421 | } |
| 422 | |
| 423 | const BLOCK_LIST_MAX = 8 * 1024 * 1024; |
| 424 | |
| 425 | actionsRuntime.put(BLOB_PATH, async (c) => { |
| 426 | const b = blobRef(c); |
| 427 | if (!b) return azureError(404, "ResourceNotFound", "The specified resource does not exist."); |
| 428 | const q = new URL(c.req.url).searchParams; |
| 429 | if (!(await verifyBlobUrl(c.env, b, "w", q))) return azureError(403, "AuthenticationFailed", "The upload URL is invalid or has expired."); |
| 430 | try { |
| 431 | const comp = q.get("comp"); |
| 432 | if (comp === "block") { |
| 433 | const blockId = q.get("blockid") ?? ""; |
| 434 | if (!blockId || blockId.length > 1024) return azureError(400, "InvalidQueryParameterValue", "blockid is required."); |
| 435 | const put = await putUpload(c, b, blockKey(b, blockId)); |
| 436 | return put instanceof Response ? put : created(); |
| 437 | } |
| 438 | if (comp !== "blocklist" && comp !== null) return azureError(400, "InvalidQueryParameterValue", `comp=${comp} is not supported.`); |
| 439 | const stub = repoStub(c.env, b.repo); |
| 440 | const writable = b.kind === "cache" ? await stub.rtCacheWritable(b.id) : await stub.rtArtifactWritable(b.id); |
| 441 | if (!writable) return azureError(409, "BlobImmutable", "This upload is finished or its job is no longer running."); |
| 442 | const contentType = c.req.header("x-ms-blob-content-type") ?? "application/octet-stream"; |
| 443 | if (comp === null) { |
| 444 | // Put Blob: the whole content in one request. |
| 445 | if ((c.req.header("x-ms-blob-type") ?? "BlockBlob") !== "BlockBlob") return azureError(400, "InvalidHeaderValue", "Only block blobs are supported."); |
| 446 | const put = await putUpload(c, b, finalKey(b), contentType); |
| 447 | return put instanceof Response ? put : created(put.httpEtag); |
| 448 | } |
| 449 | // Put Block List: concatenate the staged blocks in the order given. |
| 450 | const ids = parseBlockList(await smallText(c, BLOCK_LIST_MAX)); |
| 451 | if (ids.length > 50_000) return azureError(400, "InvalidBlockList", "Too many blocks."); |
| 452 | const staged = await listStaged(bucket(c.env), b); |
| 453 | const segments: Array<{ key: string; size: number }> = []; |
| 454 | for (const id of ids) { |
| 455 | const key = blockKey(b, id); |
| 456 | const size = staged.get(key); |
| 457 | if (size === undefined) return azureError(400, "InvalidBlockList", "The specified block list is invalid."); |
| 458 | segments.push({ key, size }); |
| 459 | } |
| 460 | await stub.rtCheckCommitSize(b.kind, b.id, segments.reduce((n, s) => n + s.size, 0)); |
| 461 | await composeUpload(c.env, b, segments, contentType); |
| 462 | await deleteKeys(bucket(c.env), [...staged.keys()]); |
| 463 | const head = await c.env.BUCKET.head(finalKey(b)); |
| 464 | return created(head?.httpEtag ?? `"${newId()}"`); |
| 465 | } catch (e) { |
| 466 | const { status, message } = failure(e); |
| 467 | if (status >= 500) console.error("blob upload failed", b.kind, b.id, q.get("comp") ?? "put", e); |
| 468 | return azureError(status, status === 404 ? "ResourceNotFound" : status === 413 ? "RequestBodyTooLarge" : status === 409 ? "BlobImmutable" : status < 500 ? "InvalidInput" : "InternalError", message); |
| 469 | } |
| 470 | }); |
| 471 | |
| 472 | const serveBlob = async (c: AppContext) => { |
| 473 | const b = blobRef(c); |
| 474 | if (!b) return azureError(404, "ResourceNotFound", "The specified resource does not exist."); |
| 475 | if (!(await verifyBlobUrl(c.env, b, "r", new URL(c.req.url).searchParams))) return azureError(403, "AuthenticationFailed", "The download URL is invalid or has expired."); |
| 476 | const stub = repoStub(c.env, b.repo); |
| 477 | const headers: Record<string, string> = { "accept-ranges": "bytes", "x-ms-blob-type": "BlockBlob", "content-type": "application/octet-stream" }; |
| 478 | let etag: string | null; |
| 479 | let keys = [finalKey(b)]; |
| 480 | if (b.kind === "cache") { |
| 481 | const e = await stub.rtCacheGet(b.id); |
| 482 | if (!e) return azureError(404, "BlobNotFound", "The cache entry does not exist."); |
| 483 | etag = e.etag; |
| 484 | } else { |
| 485 | const a = await stub.rtArtifactGet(b.id); |
| 486 | if (!a) return azureError(404, "BlobNotFound", "The artifact does not exist or has expired."); |
| 487 | etag = a.etag; |
| 488 | const name = artifactFileName(a); |
| 489 | headers["content-type"] = a.mimeType; |
| 490 | headers["content-disposition"] = `attachment; filename="${name.replace(/[^\w.-]+/g, "_")}"; filename*=UTF-8''${encodeURIComponent(name)}`; |
| 491 | if (a.digest) headers["x-ms-meta-digest"] = a.digest; |
| 492 | // An older runner's tarball stays at its legacy key. |
| 493 | keys = artifactObjectKeys(b.repo, a); |
| 494 | } |
| 495 | const head = await headFirst(c.env.BUCKET, keys); |
| 496 | if (!head) return azureError(404, "BlobNotFound", "The specified blob does not exist."); |
| 497 | const key = head.key; |
| 498 | // Only the object that was committed: a write that landed after Finalize is not served. |
| 499 | if (etag && head.etag !== etag) return azureError(409, "BlobModified", "The blob was modified after it was committed."); |
| 500 | Object.assign(headers, { etag: head.httpEtag, "last-modified": head.uploaded.toUTCString() }); |
| 501 | const range = parseRange(c.req.header("x-ms-range") ?? c.req.header("range"), head.size); |
| 502 | if (range === "invalid") return new Response(null, { status: 416, headers: azureHeaders({ ...headers, "content-range": `bytes */${head.size}` }) }); |
| 503 | if (c.req.method === "HEAD") return new Response(null, { status: 200, headers: azureHeaders({ ...headers, "content-length": String(head.size) }) }); |
| 504 | if (!range) { |
| 505 | const obj = await c.env.BUCKET.get(key, { onlyIf: { etagMatches: head.etag } }); |
| 506 | if (!obj || !("body" in obj)) return azureError(404, "BlobNotFound", "The specified blob does not exist."); |
| 507 | return new Response(obj.body, { status: 200, headers: azureHeaders({ ...headers, "content-length": String(obj.size) }) }); |
| 508 | } |
| 509 | const obj = await c.env.BUCKET.get(key, { range: { offset: range.offset, length: range.length }, onlyIf: { etagMatches: head.etag } }); |
| 510 | if (!obj || !("body" in obj)) return azureError(404, "BlobNotFound", "The specified blob does not exist."); |
| 511 | return new Response(obj.body, { |
| 512 | status: 206, |
| 513 | headers: azureHeaders({ ...headers, "content-length": String(range.length), "content-range": `bytes ${range.offset}-${range.offset + range.length - 1}/${head.size}` }), |
| 514 | }); |
| 515 | }; |
| 516 | actionsRuntime.get(BLOB_PATH, serveBlob); |
| 517 | actionsRuntime.on("HEAD", BLOB_PATH, serveBlob); |
| 518 | |
| 519 | // ---------------------------------------------------------------- REST: artifacts and caches (/api/v3) |
| 520 | |
| 521 | const P = (c: AppContext) => ({ owner: c.req.param("owner")!, repo: c.req.param("repo")! }); |
| 522 | |
| 523 | const idParam = (c: AppContext, name: string) => { |
| 524 | const n = Number(c.req.param(name)); |
| 525 | if (!Number.isSafeInteger(n) || n < 1) throw new HttpError(404, "Not Found"); |
| 526 | return n; |
| 527 | }; |
| 528 | |
| 529 | function artifactJson(c: AppContext, repoId: string, a: ArtifactEntry) { |
| 530 | const { owner, repo } = P(c); |
| 531 | const url = `${c.env.PUBLIC_URL}/api/v3/repos/${owner}/${repo}/actions/artifacts/${a.id}`; |
| 532 | return { |
| 533 | id: a.id, |
| 534 | node_id: `MDg6QXJ0aWZhY3Q${a.id}`, |
| 535 | name: a.name, |
| 536 | size_in_bytes: a.size, |
| 537 | url, |
| 538 | archive_download_url: `${url}/zip`, |
| 539 | expired: a.expired, |
| 540 | digest: a.digest, |
| 541 | created_at: a.createdAt, |
| 542 | updated_at: a.updatedAt, |
| 543 | expires_at: a.expiresAt, |
| 544 | workflow_run: { id: a.runNumber, repository_id: repoId, head_repository_id: repoId, head_branch: a.headBranch, head_sha: a.headSha }, |
| 545 | }; |
| 546 | } |
| 547 | |
| 548 | const cacheJson = (e: CacheEntry) => ({ id: e.id, ref: e.scope, key: e.key, version: e.version, last_accessed_at: e.lastUsed, created_at: e.createdAt, size_in_bytes: e.size }); |
| 549 | |
| 550 | actionsRest.get("/repos/:owner/:repo/actions/artifacts", async (c) => { |
| 551 | const { repo, stub } = await loadRepo(c, P(c).owner, P(c).repo); |
| 552 | const p = pageOf(c); |
| 553 | const res = await stub.rtRepoArtifacts({ name: c.req.query("name") || undefined, page: p.page, perPage: p.perPage }); |
| 554 | setLinks(c, p, res.total); |
| 555 | return c.json({ total_count: res.total, artifacts: res.artifacts.map((a) => artifactJson(c, repo.id, a)) }); |
| 556 | }); |
| 557 | |
| 558 | actionsRest.get("/repos/:owner/:repo/actions/runs/:runId/artifacts", async (c) => { |
| 559 | const { repo, stub } = await loadRepo(c, P(c).owner, P(c).repo); |
| 560 | const runId = c.req.param("runId"); |
| 561 | if (!/^[A-Za-z0-9]{1,64}$/.test(runId)) throw new HttpError(404, "Not Found"); |
| 562 | const all = await stub.rtRunArtifacts(runId, { name: c.req.query("name") || undefined, includeExpired: true }); |
| 563 | return c.json({ total_count: all.length, artifacts: paginate(c, all).map((a) => artifactJson(c, repo.id, a)) }); |
| 564 | }); |
| 565 | |
| 566 | actionsRest.get("/repos/:owner/:repo/actions/artifacts/:id", async (c) => { |
| 567 | const { repo, stub } = await loadRepo(c, P(c).owner, P(c).repo); |
| 568 | const a = await stub.rtArtifactGet(idParam(c, "id"), { includeExpired: true }); |
| 569 | if (!a) throw new HttpError(404, "Not Found"); |
| 570 | return c.json(artifactJson(c, repo.id, a)); |
| 571 | }); |
| 572 | |
| 573 | actionsRest.get("/repos/:owner/:repo/actions/artifacts/:id/:format", async (c) => { |
| 574 | const { repo, stub } = await loadRepo(c, P(c).owner, P(c).repo); |
| 575 | if (c.req.param("format") !== "zip") throw new HttpError(404, "Not Found"); |
| 576 | const a = await stub.rtArtifactGet(idParam(c, "id"), { includeExpired: true }); |
| 577 | if (!a) throw new HttpError(404, "Not Found"); |
| 578 | // As on GitHub: an expired artifact is still listed, but its content is gone. |
| 579 | if (a.expired) throw new HttpError(410, "Artifact has expired"); |
| 580 | return c.redirect(await signBlobUrl(c.env, origin(c.env), { repo: repo.id, kind: "artifact", id: a.id }, "r", nowS() + 600, artifactFileName(a)), 302); |
| 581 | }); |
| 582 | |
| 583 | actionsRest.delete("/repos/:owner/:repo/actions/artifacts/:id", async (c) => { |
| 584 | const { repo, caller, stub } = await loadRepo(c, P(c).owner, P(c).repo, "write"); |
| 585 | const deleted = await stub.rtArtifactDelete(idParam(c, "id")); |
| 586 | if (!deleted) throw new HttpError(404, "Not Found"); |
| 587 | auditRepo(c, caller, repo, "actions.artifact.delete", { id: deleted.id, name: deleted.name }); |
| 588 | return c.body(null, 204); |
| 589 | }); |
| 590 | |
| 591 | actionsRest.get("/repos/:owner/:repo/actions/caches", async (c) => { |
| 592 | const { stub } = await loadRepo(c, P(c).owner, P(c).repo); |
| 593 | const p = pageOf(c); |
| 594 | const res = await stub.rtListCaches({ key: c.req.query("key") || undefined, ref: c.req.query("ref") || undefined, sort: c.req.query("sort"), direction: c.req.query("direction"), page: p.page, perPage: p.perPage }); |
| 595 | setLinks(c, p, res.total); |
| 596 | return c.json({ total_count: res.total, actions_caches: res.caches.map(cacheJson) }); |
| 597 | }); |
| 598 | |
| 599 | actionsRest.delete("/repos/:owner/:repo/actions/caches", async (c) => { |
| 600 | const { repo, caller, stub } = await loadRepo(c, P(c).owner, P(c).repo, "write"); |
| 601 | const key = c.req.query("key"); |
| 602 | if (!key) throw new HttpError(422, "key is required"); |
| 603 | const deleted = await stub.rtDeleteCaches({ key, ref: c.req.query("ref") || undefined }); |
| 604 | if (!deleted.length) throw new HttpError(404, "Not Found"); |
| 605 | auditRepo(c, caller, repo, "actions.cache.delete", { key, ids: deleted.map((d) => d.id) }); |
| 606 | return c.json({ total_count: deleted.length, actions_caches: deleted.map(cacheJson) }); |
| 607 | }); |
| 608 | |
| 609 | actionsRest.delete("/repos/:owner/:repo/actions/caches/:id", async (c) => { |
| 610 | const { repo, caller, stub } = await loadRepo(c, P(c).owner, P(c).repo, "write"); |
| 611 | const deleted = await stub.rtDeleteCaches({ id: idParam(c, "id") }); |
| 612 | if (!deleted.length) throw new HttpError(404, "Not Found"); |
| 613 | auditRepo(c, caller, repo, "actions.cache.delete", { key: deleted[0]!.key, ids: [deleted[0]!.id] }); |
| 614 | return c.body(null, 204); |
| 615 | }); |
| 616 | |
| 617 | // The workflow token's defaults: GitHub's "Workflow permissions" settings. |
| 618 | const workflowSettingsJson = (s: { defaultPermissions: "read" | "write"; canApprovePullRequests: boolean; agentRunSecrets: boolean }) => ({ default_workflow_permissions: s.defaultPermissions, can_approve_pull_request_reviews: s.canApprovePullRequests, agent_runs_use_secrets: s.agentRunSecrets }); |
| 619 | |
| 620 | actionsRest.get("/repos/:owner/:repo/actions/permissions/workflow", async (c) => { |
| 621 | const { stub } = await loadRepo(c, P(c).owner, P(c).repo, "admin"); |
| 622 | return c.json(workflowSettingsJson(await stub.workflowSettings())); |
| 623 | }); |
| 624 | |
| 625 | actionsRest.put("/repos/:owner/:repo/actions/permissions/workflow", async (c) => { |
| 626 | const { caller, stub } = await loadRepo(c, P(c).owner, P(c).repo, "admin"); |
| 627 | const b = (await c.req.json().catch(() => null)) as { default_workflow_permissions?: unknown; can_approve_pull_request_reviews?: unknown; agent_runs_use_secrets?: unknown } | null; |
| 628 | if (!b || typeof b !== "object") throw new HttpError(422, "a JSON body is required"); |
| 629 | // Audited by the repository (an audit event per change), marked as made through the API. |
| 630 | await stub.setWorkflowSettings({ defaultPermissions: b.default_workflow_permissions as "read" | "write" | undefined, canApprovePullRequests: b.can_approve_pull_request_reviews as boolean | undefined, agentRunSecrets: b.agent_runs_use_secrets as boolean | undefined }, caller, { via: "api" }); |
| 631 | return c.body(null, 204); |
| 632 | }); |
| 633 | |
| 634 | actionsRest.get("/repos/:owner/:repo/actions/cache/usage", async (c) => { |
| 635 | const { stub } = await loadRepo(c, P(c).owner, P(c).repo); |
| 636 | const u = await stub.rtCacheUsage(); |
| 637 | return c.json({ full_name: `${P(c).owner}/${P(c).repo}`, active_caches_size_in_bytes: u.size, active_caches_count: u.count }); |
| 638 | }); |