commit 64cedde4779cc54ae89c689318fd6e1eeb10aaf1
parent 5ab40b5f5ebc8b88d982764d554f657447668d5e
Author: finwo <finwo@pm.me>
Date: Sun, 20 Sep 2026 04:17:48 +0200
Let anyone read a public task back without a worker token
Diffstat:
5 files changed, 522 insertions(+), 308 deletions(-)
diff --git a/src/conductor/app.js b/src/conductor/app.js
@@ -20,6 +20,7 @@ import { createAuth } from '../lib/auth/index.js';
import { createScheduler } from './scheduler.js';
import { createRetention } from './retention.js';
+import workerRoutes from './routes/workers.js';
import taskRoutes from './routes/tasks.js';
import jobRoutes from './routes/jobs.js';
import artifactRoutes from './routes/artifacts.js';
@@ -82,9 +83,14 @@ export async function buildServer(services, options = {}) {
}));
// The only HTTP surface besides the interface: creating and reading jobs,
- // the task edge workers poll, and artifact downloads. Everything else
- // is the UI.
+ // reading tasks back, the edge workers poll, and artifact downloads.
+ // Everything else is the UI.
+ //
+ // workerRoutes and taskRoutes share the /api/v1/tasks prefix deliberately.
+ // They stay separate plugins because the worker token hook covers its own
+ // plugin entirely, and reading a task back is not behind it.
await fastify.register(jobRoutes, { ...services, prefix: '/api/v1' });
+ await fastify.register(workerRoutes, { ...services, prefix: '/api/v1' });
await fastify.register(taskRoutes, { ...services, prefix: '/api/v1' });
await fastify.register(artifactRoutes, { ...services, prefix: '/api/v1' });
diff --git a/src/conductor/routes/jobs.js b/src/conductor/routes/jobs.js
@@ -18,10 +18,10 @@
// without one accepts unsigned requests, which is convenient on a private
// network and a bad idea anywhere else, so it is logged.
-import crypto from 'node:crypto';
import { SHA_PATTERN } from '../../lib/git.js';
import { PipelineError } from '../../lib/pipeline/index.js';
import { canManageProject } from '../../lib/projects.js';
+import { verifySignature } from '../../lib/signature.js';
const ZERO_SHA = '0'.repeat(40);
@@ -176,32 +176,6 @@ export default async function jobRoutes(fastify, { auth, db, projects, scheduler
}
}
-function verifySignature(req, secret) {
- const raw = req.rawBody ?? Buffer.alloc(0);
-
- // GitHub and Gitea: sha256=<hex> over the body.
- const hubSignature = req.headers['x-hub-signature-256'];
- if (typeof hubSignature === 'string' && hubSignature.length > 0) {
- const expected = `sha256=${crypto.createHmac('sha256', secret).update(raw).digest('hex')}`;
- return timingSafeEqual(expected, hubSignature);
- }
-
- // GitLab: the secret itself, compared rather than signed.
- const gitlabToken = req.headers['x-gitlab-token'];
- if (typeof gitlabToken === 'string' && gitlabToken.length > 0) {
- return timingSafeEqual(secret, gitlabToken);
- }
-
- return false;
-}
-
-function timingSafeEqual(a, b) {
- const left = Buffer.from(String(a), 'utf8');
- const right = Buffer.from(String(b), 'utf8');
- if (left.length !== right.length) return false;
- return crypto.timingSafeEqual(left, right);
-}
-
// Reduces the supported payload shapes to { sha, base, ref, actor }.
function normalizePush(body) {
if (!body || typeof body !== 'object') return null;
diff --git a/src/conductor/routes/tasks.js b/src/conductor/routes/tasks.js
@@ -1,307 +1,188 @@
-// src/conductor/routes/tasks.js - the API workers poll and report to
+// src/conductor/routes/tasks.js - reading a task back
//
-// Every route here requires a worker token. A worker may only touch a task it
-// currently holds, which is checked against worker_token_id rather than
-// trusting the task id it sends.
+// A task is the standalone unit a worker runs, and this is how everyone who
+// is not that worker looks at one: its state, and the artifacts it stored.
+// routes/workers.js holds the protocol the worker itself speaks, behind a
+// worker token. These two files share a URL prefix and nothing else, which
+// is why they are separate plugins: the token hook over there covers its
+// whole scope, and must not reach these routes.
//
-// A worker is handed a context and a script and nothing else. It is never
-// told which project or job a task belongs to, and the task id it receives
-// carries no structure it could read that out of. The identifiers a build
-// legitimately wants are in the task's environment, where the worker treats
-// them as opaque strings on their way into the container.
-
-import { pipeline as streamPipeline } from 'node:stream/promises';
-import { LogOffsetError } from '../../lib/log.js';
-import { hashingTransform } from '../../lib/stream.js';
-import { sanitizeRelativePath, keys as storageKeys } from '../../lib/storage/index.js';
-import { maskBuffer } from '../../lib/variables.js';
-import { newArtifactId } from '../../lib/ids.js';
-
-// Log chunks are small and frequent; artifacts are streamed, so this only
-// bounds a single log append.
-const LOG_CHUNK_LIMIT = 1024 * 1024;
-
-export default async function taskRoutes(fastify, { cfg, db, git, logs, storage, projects, scheduler, workerTokens, variables }) {
- // Raw bodies: log chunks and artifacts arrive as octet streams and must
- // not be parsed.
- fastify.addContentTypeParser('application/octet-stream', (req, payload, done) => done(null, payload));
-
- async function requireWorker(req, reply) {
- const header = req.headers.authorization || '';
- const presented = header.startsWith('Bearer ') ? header.slice(7).trim() : '';
- const token = await workerTokens.verify(presented, { ip: req.ip });
- if (!token) {
- reply.code(401).send({ error: 'invalid or missing worker token' });
- return;
- }
- req.worker = token;
- }
-
- // Confirms the task exists and is held by the calling worker.
- async function heldTask(req, reply) {
- const task = await db.get(
- `SELECT t.id, t.job_id, t.name, t.state, t.worker_token_id, t.log_size, t.spec,
- j.project_id, j.head_sha
- FROM tasks t JOIN jobs j ON j.id = t.job_id
- WHERE t.id = {id}`,
- { id: req.params.task }
- );
- if (!task) {
- reply.code(404).send({ error: 'unknown task' });
- return null;
- }
- if (task.worker_token_id !== req.worker.id) {
- reply.code(403).send({ error: 'task is not held by this worker' });
+// Two paths, one answer:
+//
+// GET /tasks/:task a task id is unique on its own
+// GET /projects/:project/tasks/:task the same task, said in full
+//
+// They differ only in what may reach a private task. A task id is enough to
+// prove nothing, so the bare path accepts a public task or a session that
+// may manage the project. The scoped path can look up the project's trigger
+// secret, so it also accepts a signature, matching the job read-back it sits
+// beside: one credential covers starting a build, watching the job and
+// reading any task in it.
+//
+// Absent and forbidden are both 404, so neither can be used to find out
+// which tasks exist.
+//
+// What is deliberately not here: the task's environment, and its services,
+// which carry an environment of their own. Both can hold project variables,
+// and a public job is still built from a repository that may be private. The
+// script is left out for the same reason. What remains describes what the
+// task was and how it went, which is what a caller asking after a task is
+// actually asking.
+
+import { canManageProject } from '../../lib/projects.js';
+import { verifySignature } from '../../lib/signature.js';
+
+export default async function taskRoutes(fastify, { cfg, db, projects, auth = null }) {
+ const base = cfg.server.public_url.replace(/\/+$/, '');
+
+ // Explicit columns rather than t.*, which would pull spec and
+ // worker_token_id into scope and leave one careless spread away from
+ // publishing them.
+ const TASK_QUERY = `
+ SELECT t.id, t.name, t.base_name, t.state, t.arch, t.image,
+ t.attempt, t.max_attempts, t.allow_failure, t.exit_code, t.error,
+ t.worker_name, t.timeout, t.spec, t.log_size,
+ t.created_at, t.started_at, t.finished_at,
+ j.id AS job_id, j.number AS job_number, j.ref, j.head_sha,
+ j.visibility, j.project_id
+ FROM tasks t
+ JOIN jobs j ON j.id = t.job_id`;
+
+ async function currentUser(req) {
+ if (!auth) return null;
+ try {
+ return await auth.identify(req);
+ } catch {
return null;
}
- return task;
}
- fastify.addHook('preHandler', requireWorker);
-
- // Ask for work. The worker describes what it can run; the scheduler
- // decides. Returns 204 when there is nothing to do.
- fastify.post('/tasks/claim', async (req, reply) => {
- const body = req.body ?? {};
- const arches = splitList(body.arches);
- const features = splitList(body.features);
- const name = typeof body.name === 'string' ? body.name.slice(0, 255) : req.worker.name;
+ // Everything a caller is allowed to see, assembled once so the two routes
+ // cannot drift into answering differently.
+ async function represent(reply, row) {
+ const spec = readSpec(row.spec);
- const task = await scheduler.claim({
- arches,
- features,
- tokenId: req.worker.id,
- workerName: name,
- // A worker registered by a user only ever sees that user's projects.
- ownerId: req.worker.owner_id ?? null,
- });
- if (!task) return reply.code(204).send();
-
- const base = cfg.server.public_url.replace(/\/+$/, '');
- const at = (suffix) => `${base}/api/v1/tasks/${task.id}/${suffix}`;
+ const artifacts = await db.all(
+ `SELECT id, path, size, sha256, created_at, expires_at
+ FROM artifacts WHERE task_id = {task} ORDER BY path`,
+ { task: row.id }
+ );
return reply.send({
task: {
- id: task.id,
- name: task.name,
- arch: task.arch,
- image: task.image,
- // The worker maps these onto its own local mounts, environment and
- // privileges. Without them no feature is applied at all.
- requires: task.requires,
- script: task.script,
- env: task.env,
- services: task.services,
- artifacts: task.artifacts,
- // Where the tree is unpacked and the script runs. Resolved when
- // the job was created, so it does not shift under a retry.
- workdir: task.workdir,
- timeout: task.timeout,
- attempt: task.attempt,
- sha: task.sha,
- ref: task.ref,
- // Values the worker must redact from the log stream. Populated
- // once project variables exist; masking also happens on ingest.
- masked: task.masked ?? [],
- // Derived from the server's own timeout so a worker never has to
- // guess how often it is expected to check in.
- heartbeat_interval: Math.max(5, Math.floor(cfg.scheduler.heartbeat_timeout / 4)),
- endpoints: {
- // The tree comes from the conductor, so a worker never needs
- // credentials for the repository and cannot reach any commit
- // other than the one it was given work for.
- source: at('source.tar.gz'),
- log: at('log'),
- artifact: at('artifacts'),
- heartbeat: at('heartbeat'),
- done: at('complete'),
- },
+ id: row.id,
+ name: row.name,
+ base_name: row.base_name,
+ state: row.state,
+ arch: row.arch,
+ image: row.image,
+ attempt: row.attempt,
+ max_attempts: row.max_attempts,
+ allow_failure: row.allow_failure === 1,
+ exit_code: row.exit_code,
+ error: row.error,
+ worker_name: row.worker_name,
+ timeout: row.timeout,
+ // Settled when the job was created, so it is what the task ran in
+ // rather than what the pipeline says today.
+ workdir: spec.workdir,
+ needs: spec.needs,
+ matrix: spec.matrix,
+ // What the task was told to collect, which is not the same as what
+ // it managed to store. The list below is what exists.
+ artifact_paths: spec.artifactPaths,
+ log_size: row.log_size,
+ created_at: row.created_at,
+ started_at: row.started_at,
+ finished_at: row.finished_at,
+ project_id: row.project_id,
+ job_id: row.job_id,
+ job_number: row.job_number,
+ ref: row.ref,
+ sha: row.head_sha,
+ visibility: row.visibility,
},
+ artifacts: artifacts.map((a) => ({
+ id: a.id,
+ path: a.path,
+ size: a.size,
+ sha256: a.sha256,
+ created_at: a.created_at,
+ expires_at: a.expires_at,
+ // Ready to fetch. The download route checks visibility again for
+ // itself, so this is a convenience and not the permission.
+ url: `${base}/api/v1/projects/${encodeURIComponent(row.project_id)}/jobs/${row.job_id}`
+ + `/tasks/${row.id}/artifacts/${a.id}`,
+ })),
});
- });
-
- // The tree at the task's commit, as a gzipped tar.
- fastify.get('/tasks/:task/source.tar.gz', async (req, reply) => {
- const task = await heldTask(req, reply);
- if (!task) return reply;
-
- const project = await projects.get(task.project_id);
- if (!project) return reply.code(404).send({ error: 'unknown project' });
-
- // The tree rooted at the archive, with no wrapping directory: a
- // worker extracts it at whatever path the task is to run in, and
- // docker creates that path on the way.
- reply.header('content-type', 'application/gzip');
- reply.header('content-disposition', `attachment; filename="${task.head_sha.slice(0, 12)}.tar.gz"`);
- return reply.send(git.archiveStream(project, task.head_sha));
- });
-
- // Append to the live log. X-Log-Offset makes a retry after a dropped
- // connection safe.
- fastify.post('/tasks/:task/log', { bodyLimit: LOG_CHUNK_LIMIT }, async (req, reply) => {
- const task = await heldTask(req, reply);
- if (!task) return reply;
-
- const chunks = [];
- let total = 0;
- for await (const chunk of req.body) {
- total += chunk.length;
- if (total > LOG_CHUNK_LIMIT) return reply.code(413).send({ error: 'log chunk too large' });
- chunks.push(chunk);
- }
-
- const rawOffset = req.headers['x-log-offset'];
- const offset = rawOffset === undefined ? undefined : Number(rawOffset);
- if (offset !== undefined && !Number.isInteger(offset)) {
- return reply.code(400).send({ error: 'x-log-offset must be an integer' });
- }
+ }
- // The worker masks before sending; this repeats it on ingest so a
- // worker that does not still cannot write a secret to disk. It works
- // within a chunk only, which is why the worker holds back a boundary.
- let payload = Buffer.concat(chunks);
- if (variables) {
- try {
- payload = maskBuffer(payload, await variables.maskedValues(task.project_id));
- } catch (e) {
- req.log.warn({ err: e }, `could not mask log output for ${task.id}`);
+ // A task id is unique by itself, so the project need not be named. Without
+ // one there is no trigger secret to check against, leaving a public task
+ // or a session that may manage the project.
+ fastify.get('/tasks/:task', async (req, reply) => {
+ const row = await db.get(`${TASK_QUERY} WHERE t.id = {id}`, { id: req.params.task });
+ if (!row) return reply.code(404).send({ error: 'unknown task' });
+
+ if (row.visibility !== 'public') {
+ const project = await projects.get(row.project_id);
+ if (!canManageProject(await currentUser(req), project)) {
+ return reply.code(404).send({ error: 'unknown task' });
}
}
- try {
- const result = await logs.append(task.job_id, task.id, payload, offset);
- await db.run('UPDATE tasks SET log_size = {size}, heartbeat_at = {now} WHERE id = {id}',
- { id: task.id, size: result.size, now: Date.now() });
- return reply.send(result);
- } catch (e) {
- if (e instanceof LogOffsetError) {
- // Tell the worker where to resume from.
- return reply.code(409).send({ error: e.message, expected_offset: e.expected });
- }
- throw e;
- }
+ return represent(reply, row);
});
- // artifacts.expire in the pipeline, as an absolute time. Returns null
- // when the task said nothing, leaving the artifact to the project policy.
- function artifactExpiry(specJson, now) {
- if (!specJson) return null;
- try {
- const spec = typeof specJson === 'string' ? JSON.parse(specJson) : specJson;
- const seconds = spec?.artifacts?.expire;
- if (!Number.isFinite(seconds) || seconds <= 0) return null;
- return now + Math.round(seconds * 1000);
- } catch {
- // A spec that will not parse is a problem, but not this request's
- // problem: the artifact is already stored.
- return null;
- }
- }
-
- // Upload one artifact. The path is worker supplied, so it is sanitized
- // before it can influence a storage key.
- fastify.post('/tasks/:task/artifacts', async (req, reply) => {
- const task = await heldTask(req, reply);
- if (!task) return reply;
+ // The same task named in full. The project in the path must be the one the
+ // task belongs to, so a guessed id cannot be read through a project that
+ // happens to be readable.
+ fastify.get('/projects/:project/tasks/:task', async (req, reply) => {
+ const project = await projects.get(req.params.project);
+ if (!project) return reply.code(404).send({ error: 'unknown task' });
- const relPath = sanitizeRelativePath(req.headers['x-artifact-path']);
- if (!relPath) return reply.code(400).send({ error: 'x-artifact-path is missing or unusable' });
-
- const declared = Number(req.headers['content-length']);
- if (!Number.isInteger(declared) || declared < 0) {
- return reply.code(411).send({ error: 'content-length is required for artifact uploads' });
- }
-
- // Hash on the way past, so the digest costs nothing extra even when the
- // backend cannot report one.
- const hasher = hashingTransform();
- const key = storageKeys.artifact(task.job_id, task.id, relPath);
-
- let stored;
- try {
- [, stored] = await Promise.all([
- streamPipeline(req.body, hasher),
- storage.put(key, hasher, { size: declared }),
- ]);
- } catch (e) {
- // A body shorter or longer than content-length is the worker's
- // mistake, not a server fault.
- if (/size mismatch/.test(e.message)) {
- return reply.code(400).send({ error: e.message });
- }
- throw e;
- }
-
- const digest = hasher.digest();
- const now = Date.now();
-
- await db.run(
- `INSERT INTO artifacts (id, task_id, job_id, path, storage_key, size, sha256, created_at, expires_at)
- VALUES ({id}, {task}, {job}, {path}, {key}, {size}, {sha}, {now}, {expires})`,
- {
- id: newArtifactId(),
- task: task.id,
- job: task.job_id,
- path: relPath,
- key,
- size: stored.size ?? declared,
- sha: digest,
- now,
- // An explicit deadline from the pipeline. Absent, the project's
- // retention policy decides, which is the usual case.
- expires: artifactExpiry(task.spec, now),
- }
+ const row = await db.get(
+ `${TASK_QUERY} WHERE t.id = {id} AND j.project_id = {project}`,
+ { id: req.params.task, project: project.id }
);
+ if (!row) return reply.code(404).send({ error: 'unknown task' });
- return reply.send({ path: relPath, size: stored.size ?? declared, sha256: digest });
- });
+ if (row.visibility !== 'public' && !(await mayRead(req, project))) {
+ return reply.code(404).send({ error: 'unknown task' });
+ }
- // Keeps the task alive, and is how a worker learns it should stop.
- fastify.post('/tasks/:task/heartbeat', async (req, reply) => {
- const result = await scheduler.heartbeat(req.params.task, req.worker.id);
- if (!result.known) return reply.code(404).send({ error: 'unknown task for this worker' });
- return reply.send({ cancelled: result.cancelled, state: result.state });
+ return represent(reply, row);
});
- // Report the outcome. The log is copied to storage here, whatever the
- // result, so a failed task keeps its output.
- fastify.post('/tasks/:task/complete', async (req, reply) => {
- const task = await heldTask(req, reply);
- if (!task) return reply;
-
- const body = req.body ?? {};
- const success = body.success === true;
- const exitCode = Number.isInteger(body.exit_code) ? body.exit_code : null;
- const error = typeof body.error === 'string' ? body.error.slice(0, 4096) : null;
-
- const outcome = await scheduler.complete(task.id, {
- success,
- exitCode,
- error,
- tokenId: req.worker.id,
- });
- if (!outcome.ok) return reply.code(409).send({ error: outcome.reason });
-
- // A retry keeps writing to the same log, so only archive it once the
- // task has actually stopped.
- if (outcome.state !== 'queued') {
- try {
- await scheduler.finalizeLog(task.job_id, task.id);
- } catch (e) {
- req.log.error({ err: e }, `failed to archive log for ${task.id}`);
- }
- }
+ async function mayRead(req, project) {
+ const secret = projects.triggerSecret(project);
+ if (secret && verifySignature(req, secret)) return true;
- return reply.send({
- state: outcome.state,
- retry: outcome.retry === true,
- skipped: outcome.skipped ?? [],
- job_state: outcome.jobState ?? null,
- });
- });
+ return canManageProject(await currentUser(req), project);
+ }
}
-function splitList(value) {
- const items = Array.isArray(value) ? value : typeof value === 'string' ? value.split(',') : [];
- return items.map((s) => String(s).trim()).filter(Boolean).slice(0, 64);
+// The stored spec, reduced to the keys that are safe to publish. env and
+// services are never read: both can carry project variables.
+//
+// A spec that will not parse is not this request's problem, and reporting
+// the task without its detail beats refusing to report it at all.
+function readSpec(raw) {
+ let spec;
+ try {
+ spec = typeof raw === 'string' ? JSON.parse(raw) : raw;
+ } catch {
+ spec = null;
+ }
+ if (!spec || typeof spec !== 'object') {
+ return { workdir: null, needs: [], matrix: {}, artifactPaths: [] };
+ }
+
+ const paths = spec.artifacts?.paths;
+ return {
+ workdir: typeof spec.workdir === 'string' ? spec.workdir : null,
+ needs: Array.isArray(spec.needs) ? spec.needs : [],
+ matrix: spec.matrix && typeof spec.matrix === 'object' ? spec.matrix : {},
+ artifactPaths: Array.isArray(paths) ? paths : [],
+ };
}
diff --git a/src/conductor/routes/workers.js b/src/conductor/routes/workers.js
@@ -0,0 +1,313 @@
+// src/conductor/routes/workers.js - the API workers poll and report to
+//
+// Named for who calls it rather than what it addresses. The paths are under
+// /api/v1/tasks because a task is what a worker deals in, but everything in
+// this file is the worker protocol, and routes/tasks.js next door is the
+// task resource as anyone else reads it.
+//
+// Every route here requires a worker token: the preHandler below covers the
+// whole plugin, so nothing can be added to this file that is not behind it.
+// A worker may only touch a task it currently holds, which is checked
+// against worker_token_id rather than trusting the task id it sends.
+//
+// A worker is handed a context and a script and nothing else. It is never
+// told which project or job a task belongs to, and the task id it receives
+// carries no structure it could read that out of. The identifiers a build
+// legitimately wants are in the task's environment, where the worker treats
+// them as opaque strings on their way into the container.
+
+import { pipeline as streamPipeline } from 'node:stream/promises';
+import { LogOffsetError } from '../../lib/log.js';
+import { hashingTransform } from '../../lib/stream.js';
+import { sanitizeRelativePath, keys as storageKeys } from '../../lib/storage/index.js';
+import { maskBuffer } from '../../lib/variables.js';
+import { newArtifactId } from '../../lib/ids.js';
+
+// Log chunks are small and frequent; artifacts are streamed, so this only
+// bounds a single log append.
+const LOG_CHUNK_LIMIT = 1024 * 1024;
+
+export default async function workerRoutes(fastify, { cfg, db, git, logs, storage, projects, scheduler, workerTokens, variables }) {
+ // Raw bodies: log chunks and artifacts arrive as octet streams and must
+ // not be parsed.
+ fastify.addContentTypeParser('application/octet-stream', (req, payload, done) => done(null, payload));
+
+ async function requireWorker(req, reply) {
+ const header = req.headers.authorization || '';
+ const presented = header.startsWith('Bearer ') ? header.slice(7).trim() : '';
+ const token = await workerTokens.verify(presented, { ip: req.ip });
+ if (!token) {
+ reply.code(401).send({ error: 'invalid or missing worker token' });
+ return;
+ }
+ req.worker = token;
+ }
+
+ // Confirms the task exists and is held by the calling worker.
+ async function heldTask(req, reply) {
+ const task = await db.get(
+ `SELECT t.id, t.job_id, t.name, t.state, t.worker_token_id, t.log_size, t.spec,
+ j.project_id, j.head_sha
+ FROM tasks t JOIN jobs j ON j.id = t.job_id
+ WHERE t.id = {id}`,
+ { id: req.params.task }
+ );
+ if (!task) {
+ reply.code(404).send({ error: 'unknown task' });
+ return null;
+ }
+ if (task.worker_token_id !== req.worker.id) {
+ reply.code(403).send({ error: 'task is not held by this worker' });
+ return null;
+ }
+ return task;
+ }
+
+ fastify.addHook('preHandler', requireWorker);
+
+ // Ask for work. The worker describes what it can run; the scheduler
+ // decides. Returns 204 when there is nothing to do.
+ fastify.post('/tasks/claim', async (req, reply) => {
+ const body = req.body ?? {};
+ const arches = splitList(body.arches);
+ const features = splitList(body.features);
+ const name = typeof body.name === 'string' ? body.name.slice(0, 255) : req.worker.name;
+
+ const task = await scheduler.claim({
+ arches,
+ features,
+ tokenId: req.worker.id,
+ workerName: name,
+ // A worker registered by a user only ever sees that user's projects.
+ ownerId: req.worker.owner_id ?? null,
+ });
+ if (!task) return reply.code(204).send();
+
+ const base = cfg.server.public_url.replace(/\/+$/, '');
+ const at = (suffix) => `${base}/api/v1/tasks/${task.id}/${suffix}`;
+
+ return reply.send({
+ task: {
+ id: task.id,
+ name: task.name,
+ arch: task.arch,
+ image: task.image,
+ // The worker maps these onto its own local mounts, environment and
+ // privileges. Without them no feature is applied at all.
+ requires: task.requires,
+ script: task.script,
+ env: task.env,
+ services: task.services,
+ artifacts: task.artifacts,
+ // Where the tree is unpacked and the script runs. Resolved when
+ // the job was created, so it does not shift under a retry.
+ workdir: task.workdir,
+ timeout: task.timeout,
+ attempt: task.attempt,
+ sha: task.sha,
+ ref: task.ref,
+ // Values the worker must redact from the log stream. Populated
+ // once project variables exist; masking also happens on ingest.
+ masked: task.masked ?? [],
+ // Derived from the server's own timeout so a worker never has to
+ // guess how often it is expected to check in.
+ heartbeat_interval: Math.max(5, Math.floor(cfg.scheduler.heartbeat_timeout / 4)),
+ endpoints: {
+ // The tree comes from the conductor, so a worker never needs
+ // credentials for the repository and cannot reach any commit
+ // other than the one it was given work for.
+ source: at('source.tar.gz'),
+ log: at('log'),
+ artifact: at('artifacts'),
+ heartbeat: at('heartbeat'),
+ done: at('complete'),
+ },
+ },
+ });
+ });
+
+ // The tree at the task's commit, as a gzipped tar.
+ fastify.get('/tasks/:task/source.tar.gz', async (req, reply) => {
+ const task = await heldTask(req, reply);
+ if (!task) return reply;
+
+ const project = await projects.get(task.project_id);
+ if (!project) return reply.code(404).send({ error: 'unknown project' });
+
+ // The tree rooted at the archive, with no wrapping directory: a
+ // worker extracts it at whatever path the task is to run in, and
+ // docker creates that path on the way.
+ reply.header('content-type', 'application/gzip');
+ reply.header('content-disposition', `attachment; filename="${task.head_sha.slice(0, 12)}.tar.gz"`);
+ return reply.send(git.archiveStream(project, task.head_sha));
+ });
+
+ // Append to the live log. X-Log-Offset makes a retry after a dropped
+ // connection safe.
+ fastify.post('/tasks/:task/log', { bodyLimit: LOG_CHUNK_LIMIT }, async (req, reply) => {
+ const task = await heldTask(req, reply);
+ if (!task) return reply;
+
+ const chunks = [];
+ let total = 0;
+ for await (const chunk of req.body) {
+ total += chunk.length;
+ if (total > LOG_CHUNK_LIMIT) return reply.code(413).send({ error: 'log chunk too large' });
+ chunks.push(chunk);
+ }
+
+ const rawOffset = req.headers['x-log-offset'];
+ const offset = rawOffset === undefined ? undefined : Number(rawOffset);
+ if (offset !== undefined && !Number.isInteger(offset)) {
+ return reply.code(400).send({ error: 'x-log-offset must be an integer' });
+ }
+
+ // The worker masks before sending; this repeats it on ingest so a
+ // worker that does not still cannot write a secret to disk. It works
+ // within a chunk only, which is why the worker holds back a boundary.
+ let payload = Buffer.concat(chunks);
+ if (variables) {
+ try {
+ payload = maskBuffer(payload, await variables.maskedValues(task.project_id));
+ } catch (e) {
+ req.log.warn({ err: e }, `could not mask log output for ${task.id}`);
+ }
+ }
+
+ try {
+ const result = await logs.append(task.job_id, task.id, payload, offset);
+ await db.run('UPDATE tasks SET log_size = {size}, heartbeat_at = {now} WHERE id = {id}',
+ { id: task.id, size: result.size, now: Date.now() });
+ return reply.send(result);
+ } catch (e) {
+ if (e instanceof LogOffsetError) {
+ // Tell the worker where to resume from.
+ return reply.code(409).send({ error: e.message, expected_offset: e.expected });
+ }
+ throw e;
+ }
+ });
+
+ // artifacts.expire in the pipeline, as an absolute time. Returns null
+ // when the task said nothing, leaving the artifact to the project policy.
+ function artifactExpiry(specJson, now) {
+ if (!specJson) return null;
+ try {
+ const spec = typeof specJson === 'string' ? JSON.parse(specJson) : specJson;
+ const seconds = spec?.artifacts?.expire;
+ if (!Number.isFinite(seconds) || seconds <= 0) return null;
+ return now + Math.round(seconds * 1000);
+ } catch {
+ // A spec that will not parse is a problem, but not this request's
+ // problem: the artifact is already stored.
+ return null;
+ }
+ }
+
+ // Upload one artifact. The path is worker supplied, so it is sanitized
+ // before it can influence a storage key.
+ fastify.post('/tasks/:task/artifacts', async (req, reply) => {
+ const task = await heldTask(req, reply);
+ if (!task) return reply;
+
+ const relPath = sanitizeRelativePath(req.headers['x-artifact-path']);
+ if (!relPath) return reply.code(400).send({ error: 'x-artifact-path is missing or unusable' });
+
+ const declared = Number(req.headers['content-length']);
+ if (!Number.isInteger(declared) || declared < 0) {
+ return reply.code(411).send({ error: 'content-length is required for artifact uploads' });
+ }
+
+ // Hash on the way past, so the digest costs nothing extra even when the
+ // backend cannot report one.
+ const hasher = hashingTransform();
+ const key = storageKeys.artifact(task.job_id, task.id, relPath);
+
+ let stored;
+ try {
+ [, stored] = await Promise.all([
+ streamPipeline(req.body, hasher),
+ storage.put(key, hasher, { size: declared }),
+ ]);
+ } catch (e) {
+ // A body shorter or longer than content-length is the worker's
+ // mistake, not a server fault.
+ if (/size mismatch/.test(e.message)) {
+ return reply.code(400).send({ error: e.message });
+ }
+ throw e;
+ }
+
+ const digest = hasher.digest();
+ const now = Date.now();
+
+ await db.run(
+ `INSERT INTO artifacts (id, task_id, job_id, path, storage_key, size, sha256, created_at, expires_at)
+ VALUES ({id}, {task}, {job}, {path}, {key}, {size}, {sha}, {now}, {expires})`,
+ {
+ id: newArtifactId(),
+ task: task.id,
+ job: task.job_id,
+ path: relPath,
+ key,
+ size: stored.size ?? declared,
+ sha: digest,
+ now,
+ // An explicit deadline from the pipeline. Absent, the project's
+ // retention policy decides, which is the usual case.
+ expires: artifactExpiry(task.spec, now),
+ }
+ );
+
+ return reply.send({ path: relPath, size: stored.size ?? declared, sha256: digest });
+ });
+
+ // Keeps the task alive, and is how a worker learns it should stop.
+ fastify.post('/tasks/:task/heartbeat', async (req, reply) => {
+ const result = await scheduler.heartbeat(req.params.task, req.worker.id);
+ if (!result.known) return reply.code(404).send({ error: 'unknown task for this worker' });
+ return reply.send({ cancelled: result.cancelled, state: result.state });
+ });
+
+ // Report the outcome. The log is copied to storage here, whatever the
+ // result, so a failed task keeps its output.
+ fastify.post('/tasks/:task/complete', async (req, reply) => {
+ const task = await heldTask(req, reply);
+ if (!task) return reply;
+
+ const body = req.body ?? {};
+ const success = body.success === true;
+ const exitCode = Number.isInteger(body.exit_code) ? body.exit_code : null;
+ const error = typeof body.error === 'string' ? body.error.slice(0, 4096) : null;
+
+ const outcome = await scheduler.complete(task.id, {
+ success,
+ exitCode,
+ error,
+ tokenId: req.worker.id,
+ });
+ if (!outcome.ok) return reply.code(409).send({ error: outcome.reason });
+
+ // A retry keeps writing to the same log, so only archive it once the
+ // task has actually stopped.
+ if (outcome.state !== 'queued') {
+ try {
+ await scheduler.finalizeLog(task.job_id, task.id);
+ } catch (e) {
+ req.log.error({ err: e }, `failed to archive log for ${task.id}`);
+ }
+ }
+
+ return reply.send({
+ state: outcome.state,
+ retry: outcome.retry === true,
+ skipped: outcome.skipped ?? [],
+ job_state: outcome.jobState ?? null,
+ });
+ });
+}
+
+function splitList(value) {
+ const items = Array.isArray(value) ? value : typeof value === 'string' ? value.split(',') : [];
+ return items.map((s) => String(s).trim()).filter(Boolean).slice(0, 64);
+}
diff --git a/src/lib/signature.js b/src/lib/signature.js
@@ -0,0 +1,40 @@
+// src/lib/signature.js - checking a request against a project's trigger secret
+//
+// Shared because the same secret authenticates both starting a job and
+// reading one back, and a comparison this easy to get subtly wrong should
+// exist once rather than once per caller.
+//
+// Two shapes are accepted, matching what the forges actually send:
+// GitHub and Gitea sign the body, GitLab presents the secret itself.
+//
+// The HMAC is computed over the exact bytes received, which is why callers
+// keep the raw body around rather than re-serializing the parsed one. A GET
+// carries no body, so its signature is an HMAC over zero bytes.
+
+import crypto from 'node:crypto';
+
+export function verifySignature(req, secret) {
+ const raw = req.rawBody ?? Buffer.alloc(0);
+
+ // GitHub and Gitea: sha256=<hex> over the body.
+ const hubSignature = req.headers['x-hub-signature-256'];
+ if (typeof hubSignature === 'string' && hubSignature.length > 0) {
+ const expected = `sha256=${crypto.createHmac('sha256', secret).update(raw).digest('hex')}`;
+ return timingSafeEqual(expected, hubSignature);
+ }
+
+ // GitLab: the secret itself, compared rather than signed.
+ const gitlabToken = req.headers['x-gitlab-token'];
+ if (typeof gitlabToken === 'string' && gitlabToken.length > 0) {
+ return timingSafeEqual(secret, gitlabToken);
+ }
+
+ return false;
+}
+
+export function timingSafeEqual(a, b) {
+ const left = Buffer.from(String(a), 'utf8');
+ const right = Buffer.from(String(b), 'utf8');
+ if (left.length !== right.length) return false;
+ return crypto.timingSafeEqual(left, right);
+}
