Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http";import { createHash, timingSafeEqual } from "node:crypto";import { readFileSync } from "node:fs";import type { Central, NewJob } from "./service";import { bundleFromGit } from "./bundleBuild";import type { DispatchRequest, Dispatcher } from "./dispatcher";
/** * The central system over HTTP (docs/design/job-api.md). Two surfaces on one listener: the device API under /api, authenticated by a device token, and the * operator API under /admin, authenticated by a separate admin token. No CORS headers are sent (the extension reaches it through its host permission), a request whose * Host header is not one this service answers to is refused (so a web page cannot reach a loopback service through DNS rebinding), bodies are size-capped, and * tokens are never logged. */
export interface HttpOptions { central: Central; adminToken: string; /** Host header values this service answers to, for example `127.0.0.1:4790`. */ hosts: string[]; /** The secret the XMPP server presents when it asks who may log in. Without it the internal XMPP routes do not exist. */ xmppSecret?: string; /** The password the central system itself logs in to XMPP with (not the same secret). Without it nobody can log in as the central system. */ xmppCentralPassword?: string; /** The user name the central system itself logs in to XMPP with, and the password it uses (the same secret). */ xmppCentralUser?: string; /** What a device is told about the doorbell (GET /api/config): where to connect and the domain. Absent: there is no doorbell, devices poll. */ xmpp?: { url: string; domain: string; centralJid: string }; /** What the doorbell knows of a device that is connected: shown in the operator's device list. */ presence?(deviceId: string): { online: boolean; resources: string[]; caps?: unknown } | undefined; /** Decides which device runs a test (POST /admin/dispatch). Without it the operator names the device when making a job. */ dispatcher?: Dispatcher; /** The test repository a job's `ref` is bundled from. Never taken from a request. */ testRepo?: string; maxJsonBytes?: number; maxArchiveBytes: number; log?(line: string): void; now?(): number;}
const digest = (s: string): Buffer => createHash("sha256").update(s).digest();const sameSecret = (a: string, b: string): boolean => timingSafeEqual(digest(a), digest(b));
class Refused extends Error { constructor(readonly status: number, message: string) { super(message); }}
async function readBody(req: IncomingMessage, limit: number): Promise<Buffer> { const declared = Number(req.headers["content-length"] ?? 0); if (Number.isFinite(declared) && declared > limit) throw new Refused(413, "the request body is too large"); const chunks: Buffer[] = []; let size = 0; for await (const chunk of req) { size += (chunk as Buffer).length; if (size > limit) throw new Refused(413, "the request body is too large"); chunks.push(chunk as Buffer); } return Buffer.concat(chunks);}
const bearer = (req: IncomingMessage): string | undefined => { const h = req.headers.authorization; return typeof h === "string" && h.startsWith("Bearer ") ? h.slice(7).trim() : undefined;};
export function createCentralServer(o: HttpOptions): Server { const maxJson = o.maxJsonBytes ?? 64 * 1024; /** A signed bundle brought in by the operator: up to the bundle limit, a little more once escaped. */ const MAX_IMPORT_BYTES = 8 * 1024 * 1024; const now = o.now ?? Date.now; /** * Guessing limits, only where a guess is worth making: enrolment codes (80 bits) and refresh tokens, each with its own count per address. A wrong bearer token is not * limited: tokens are 256 random bits, and behind a tunnel every client arrives from 127.0.0.1, so a limit by address would let one party lock out every device and the * operator. Entries older than a minute are dropped, so the table cannot grow without limit. */ const failures = new Map<string, { n: number; since: number }>(); const tooManyFailures = (bucket: string, ip: string): boolean => { const f = failures.get(`${bucket}:${ip}`); return f !== undefined && now() - f.since < 60_000 && f.n >= 20; }; const noteFailure = (bucket: string, ip: string): void => { const key = `${bucket}:${ip}`; const f = failures.get(key); if (f === undefined || now() - f.since >= 60_000) { if (failures.size >= 1000) for (const [k, v] of failures) if (now() - v.since >= 60_000) failures.delete(k); failures.set(key, { n: 1, since: now() }); } else f.n++; };
const json = (res: ServerResponse, status: number, body: unknown) => { const text = JSON.stringify(body); res.writeHead(status, { "content-type": "application/json", "cache-control": "no-store", "x-content-type-options": "nosniff", "content-length": Buffer.byteLength(text) }); res.end(text); }; const parse = (body: Buffer): Record<string, unknown> => { try { const v = body.length === 0 ? {} : JSON.parse(body.toString("utf8")); if (typeof v !== "object" || v === null || Array.isArray(v)) throw new Error(); return v as Record<string, unknown>; } catch { throw new Refused(400, "the body is not a JSON object"); } }; const str = (v: unknown, name: string): string => { if (typeof v !== "string") throw new Refused(400, `${name} must be text`); return v; };
async function admin(req: IncomingMessage, res: ServerResponse, method: string, url: URL): Promise<void> { const c = o.central; const path = url.pathname.replace(/^\/admin/, ""); let m: RegExpExecArray | null; if (method === "GET" && path === "/key") return json(res, 200, { publicKey: c.publicKey, dispatchKey: c.dispatchKey }); // a bundle signed elsewhere with the release key, and a revocation list signed the same way: checked against the public key before they are kept if (method === "POST" && path === "/bundles/import") { const b = parse(await readBody(req, MAX_IMPORT_BYTES)); if (typeof b.files !== "object" || b.files === null || Array.isArray(b.files)) throw new Refused(400, "files must be an object of path to text"); const r = await c.importBundle({ files: b.files as Record<string, string>, commit: typeof b.commit === "string" ? b.commit : "imported", notAfter: typeof b.notAfter === "number" ? b.notAfter : Number.NaN, signature: typeof b.signature === "string" ? b.signature : "" }); return r.ok ? json(res, 201, r) : json(res, 400, { error: r.message }); } if (method === "POST" && path === "/denylist") { const r = await c.importDenylist(parse(await readBody(req, maxJson))); return r.ok ? json(res, 200, r) : json(res, 400, { error: r.message }); } if (method === "POST" && path === "/bundles/revoke") { const b = parse(await readBody(req, maxJson)); if (!Array.isArray(b.hashes) || !b.hashes.every((h) => typeof h === "string" && /^[0-9a-f]{64}$/.test(h))) throw new Refused(400, "hashes must be a list of 64 character hashes"); const r = await c.revokeBundles(b.hashes as string[]); return r.ok ? json(res, 200, r) : json(res, 409, { error: r.message }); } if (method === "GET" && path === "/devices") return json(res, 200, { devices: c.listDevices().map((d) => ({ ...d, ...(o.presence?.(d.id.toLowerCase()) !== undefined ? { presence: o.presence(d.id.toLowerCase()) } : {}) })) }); if (method === "POST" && path === "/enrolments") { const b = parse(await readBody(req, maxJson)); if (!Array.isArray(b.origins) || !b.origins.every((x) => typeof x === "string")) throw new Refused(400, "origins must be a list of text"); try { const e = c.createEnrolment({ person: str(b.person, "person"), origins: b.origins as string[], ...(typeof b.name === "string" ? { name: b.name } : {}), ...(typeof b.ttlMs === "number" ? { ttlMs: b.ttlMs } : {}) }); return json(res, 201, e); } catch (err) { throw new Refused(400, err instanceof Error ? err.message : String(err)); } } if (method === "POST" && (m = /^\/devices\/([^/]+)\/revoke$/.exec(path))) { const done = c.revokeDevice(m[1]!); return json(res, done ? 200 : 404, { ok: done }); } if (method === "GET" && path === "/jobs") { const state = url.searchParams.get("state"); const device = url.searchParams.get("device"); return json(res, 200, { jobs: c.listJobs({ ...(state ? { state: state as never } : {}), ...(device ? { deviceId: device } : {}) }) }); } if (method === "POST" && path === "/jobs") { const b = parse(await readBody(req, maxJson)); let bundle = typeof b.bundle === "string" ? b.bundle : undefined; let commit: string | undefined; if (typeof b.ref === "string") { if (o.testRepo === undefined) throw new Refused(400, "this service has no test repository configured, so a job cannot name a ref; publish a bundle instead"); try { const built = await bundleFromGit(o.testRepo, b.ref); const published = await c.publishBundle(built.bundle.files, built.commit); bundle = published.hash; commit = built.commit; } catch (err) { throw new Refused(400, err instanceof Error ? err.message : String(err)); } } if (bundle === undefined) throw new Refused(400, "give a bundle (its hash) or a ref (a commit of the test repository)"); const input: NewJob = { deviceId: str(b.device, "device"), bundle, file: str(b.file, "file"), start: str(b.start, "start"), allow: Array.isArray(b.allow) ? (b.allow as string[]) : [], ...(typeof b.name === "string" ? { name: b.name } : {}), ...(typeof b.session === "string" ? { session: b.session } : {}), ...(typeof b.totalMs === "number" ? { totalMs: b.totalMs } : {}), ...(typeof b.postsAllowed === "number" ? { postsAllowed: b.postsAllowed } : {}), ...(typeof b.spendsAllowed === "number" ? { spendsAllowed: b.spendsAllowed } : {}), ...(typeof b.captureScreenshots === "boolean" ? { captureScreenshots: b.captureScreenshots } : {}), }; // through the dispatcher's switches and caps when there is one (the real service always has one), so this endpoint is no way round them (#71) const made = o.dispatcher !== undefined ? o.dispatcher.direct(input) : c.createJob(input); if (made.ok) return json(res, 201, { job: made.job, ...(commit !== undefined ? { commit } : {}) }); if ("skipped" in made && made.skipped !== undefined) return json(res, 201, { skipped: made.skipped, message: made.message, ...(commit !== undefined ? { commit } : {}) }); return json(res, /switched off|does not allow tests/.test(made.message) ? 409 : 400, { error: made.message }); } if (method === "POST" && path === "/dispatch") { if (o.dispatcher === undefined) throw new Refused(400, "this service has no dispatcher"); const b = parse(await readBody(req, maxJson)); let bundle = typeof b.bundle === "string" ? b.bundle : undefined; let commit: string | undefined; if (typeof b.ref === "string") { if (o.testRepo === undefined) throw new Refused(400, "this service has no test repository configured, so a dispatch cannot name a ref; publish a bundle instead"); try { const built = await bundleFromGit(o.testRepo, b.ref); bundle = (await c.publishBundle(built.bundle.files, built.commit)).hash; commit = built.commit; } catch (err) { throw new Refused(400, err instanceof Error ? err.message : String(err)); } } if (bundle === undefined) throw new Refused(400, "give a bundle (its hash) or a ref (a commit of the test repository)"); if (b.fanout !== undefined && b.fanout !== "one" && b.fanout !== "all") throw new Refused(400, "fanout is one or all"); const request: DispatchRequest = { bundle, file: str(b.file, "file"), start: str(b.start, "start"), allow: Array.isArray(b.allow) ? (b.allow as string[]) : [], ...(typeof b.name === "string" ? { name: b.name } : {}), ...(typeof b.session === "string" ? { session: b.session } : {}), ...(typeof b.chromeMin === "string" ? { chromeMin: b.chromeMin } : {}), ...(typeof b.totalMs === "number" ? { totalMs: b.totalMs } : {}), ...(typeof b.postsAllowed === "number" ? { postsAllowed: b.postsAllowed } : {}), ...(typeof b.spendsAllowed === "number" ? { spendsAllowed: b.spendsAllowed } : {}), ...(typeof b.captureScreenshots === "boolean" ? { captureScreenshots: b.captureScreenshots } : {}), ...(b.fanout !== undefined ? { fanout: b.fanout as "one" | "all" } : {}), ...(typeof b.device === "string" ? { device: b.device } : {}), }; const r = o.dispatcher.dispatch(request); // nothing made and nothing skipped: no device matched, which the operator must see as a failure; skipped runs are a report, not a failure return json(res, r.jobs.length === 0 && r.skipped.length === 0 ? 409 : 201, { ...r, ...(commit !== undefined ? { commit } : {}), ...(r.jobs.length === 0 && r.skipped.length === 0 ? { error: r.message ?? "no device matches" } : {}) }); } if (method === "GET" && (m = /^\/dispatch\/([^/]+)$/.exec(path))) { const jobs = c.jobsOfDispatch(m[1]!); return jobs.length === 0 ? json(res, 404, { error: "no such dispatch" }) : json(res, 200, { dispatchId: m[1], jobs, ...(o.dispatcher?.lastRetry(m[1]!) !== undefined ? { lastRetry: o.dispatcher.lastRetry(m[1]!) } : {}) }); } if (method === "POST" && path === "/stop") { const b = parse(await readBody(req, maxJson)); try { c.stop(str(b.reason, "reason")); } catch (err) { throw new Refused(400, err instanceof Error ? err.message : String(err)); } return json(res, 200, { stopped: c.stopped() }); } if (method === "POST" && path === "/resume") { c.resume(); return json(res, 200, { stopped: null }); } if (method === "GET" && path === "/status") { return json(res, 200, { stopped: c.stopped() ?? null, postCaps: o.dispatcher?.postCaps ?? null, spendCaps: o.dispatcher?.spendCaps ?? null, devices: c.listDevices().filter((d) => d.revokedAt === undefined).map((d) => ({ id: d.id, name: d.name, person: d.person, presence: o.presence?.(d.id.toLowerCase()) ?? null })), jobs: Object.fromEntries(["offered", "accepted", "done", "failed", "expired", "cancelled", "skipped"].map((st) => [st, c.listJobs({ state: st as never }).length])), }); } if (method === "GET" && (m = /^\/jobs\/([^/]+)$/.exec(path))) { const d = c.jobDetail(m[1]!); return d ? json(res, 200, d) : json(res, 404, { error: "no such job" }); } if (method === "POST" && (m = /^\/jobs\/([^/]+)\/cancel$/.exec(path))) return json(res, c.cancelJob(m[1]!) ? 200 : 409, { ok: true }); if (method === "GET" && (m = /^\/jobs\/([^/]+)\/archive$/.exec(path))) { const file = c.archivePath(m[1]!); if (file === undefined) return json(res, 404, { error: "no archive for this job" }); const bytes = readFileSync(file); res.writeHead(200, { "content-type": "application/zip", "content-length": bytes.length, "cache-control": "no-store", "x-content-type-options": "nosniff", "content-disposition": `attachment; filename="${m[1]}.zip"` }); return void res.end(bytes); } return json(res, 404, { error: "no route" }); }
/** The two calls a device makes before it has an access token. Both count failures, so a guessed code or token is rate limited like a guessed bearer. */ async function anonymous(req: IncomingMessage, res: ServerResponse, method: string, path: string, ip: string): Promise<boolean> { const c = o.central; if (method === "POST" && path === "/api/enrol") { if (tooManyFailures("enrol", ip)) return json(res, 429, { error: "too many wrong enrolment codes from this address" }), true; const b = parse(await readBody(req, maxJson)); const r = c.redeemEnrolment(str(b.code, "code"), typeof b.name === "string" ? b.name : undefined); if (!r.ok) { noteFailure("enrol", ip); json(res, r.status, { error: r.message }); } else json(res, 201, { device: { id: r.device.id, name: r.device.name, person: r.device.person, origins: r.device.origins }, credentials: r.credentials }); return true; } if (method === "POST" && path === "/api/token/refresh") { if (tooManyFailures("refresh", ip)) return json(res, 429, { error: "too many wrong refresh tokens from this address" }), true; const b = parse(await readBody(req, maxJson)); const r = c.refresh(str(b.refreshToken, "refreshToken")); if (!r.ok) { noteFailure("refresh", ip); json(res, r.status, { error: "unauthorized" }); } else json(res, 200, { credentials: r.credentials }); return true; } return false; }
async function device(req: IncomingMessage, res: ServerResponse, method: string, url: URL, token: string | undefined, ip: string): Promise<void> { const c = o.central; const dev = c.authenticate(token); if (dev === undefined) return json(res, 401, { error: "unauthorized" }); const path = url.pathname.replace(/^\/api/, ""); let m: RegExpExecArray | null; if (method === "GET" && path === "/config") return json(res, 200, { xmpp: o.xmpp ?? null }); if (method === "GET" && path === "/jobs") { if ((url.searchParams.get("state") ?? "offered") !== "offered") return json(res, 400, { error: "only state=offered can be listed" }); return json(res, 200, { jobs: c.offered(dev) }); } if (method === "POST" && (m = /^\/jobs\/([A-Za-z0-9_-]{1,64})\/accept$/.exec(path))) { const r = c.accept(dev, m[1]!, { seenFrom: req.socket.remoteAddress ?? "" }); return r.ok ? json(res, 200, { job: r.job }) : json(res, r.status, { error: r.status === 409 ? "taken" : "no such job" }); } // Both name the attempt they are about: a runner that lost its attempt (the job was offered again) cannot extend or release the next one. const attemptOf = (b: Record<string, unknown>): number => { if (typeof b.attempt !== "number" || !Number.isInteger(b.attempt) || b.attempt < 1) throw new Refused(400, "give the attempt this is about (a whole number from 1)"); return b.attempt; }; if (method === "POST" && (m = /^\/jobs\/([A-Za-z0-9_-]{1,64})\/heartbeat$/.exec(path))) { const attempt = attemptOf(parse(await readBody(req, maxJson))); return json(res, c.heartbeat(dev, m[1]!, attempt) ? 200 : 409, { ok: true }); } if (method === "POST" && (m = /^\/jobs\/([A-Za-z0-9_-]{1,64})\/fail$/.exec(path))) { const b = parse(await readBody(req, maxJson)); return json(res, c.fail(dev, m[1]!, typeof b.reason === "string" ? b.reason : "no reason given", attemptOf(b)) ? 200 : 409, { ok: true }); } if (method === "GET" && path === "/denylist") { const list = c.denylist(); return list ? json(res, 200, list) : json(res, 404, { error: "no revocation list has been published" }); } if (method === "GET" && (m = /^\/bundles\/([0-9a-f]{64})$/.exec(path))) { const b = c.bundleFor(dev, m[1]!); return b ? json(res, 200, b) : json(res, 404, { error: "no such bundle" }); } if (method === "PUT" && (m = /^\/jobs\/([A-Za-z0-9_-]{1,64})\/archive$/.exec(path))) { const bytes = await readBody(req, o.maxArchiveBytes); const r = c.putArchive(dev, m[1]!, new Uint8Array(bytes), String(req.headers["x-aite-sha256"] ?? ""), { seenFrom: req.socket.remoteAddress ?? "" }); return r.ok ? json(res, 200, { sha256: r.sha256, bytes: r.bytes }) : json(res, r.status, { error: r.message }); } return json(res, 404, { error: "no route" }); }
/** The XMPP server asks here whether a login is good. Loopback only and secret protected; not rate limited (see the comment on the guessing limits). */ async function internal(req: IncomingMessage, res: ServerResponse, method: string, path: string): Promise<boolean> { if (o.xmppSecret === undefined || path !== "/internal/xmpp/auth") return false; const given = req.headers["x-aite-internal"]; if (method !== "POST" || typeof given !== "string" || !sameSecret(given, o.xmppSecret)) { json(res, 401, { error: "unauthorized" }); return true; } const b = parse(await readBody(req, maxJson)); const user = typeof b.user === "string" ? b.user : ""; const password = typeof b.password === "string" ? b.password : ""; // not rate limited: a device token is 256 random bits, and a limit per user name would let anyone who can reach the XMPP server lock a known device id out of it const good = user === (o.xmppCentralUser ?? "central") ? o.xmppCentralPassword !== undefined && sameSecret(password, o.xmppCentralPassword) : o.central.xmppLogin(user, password); if (!good) json(res, 401, { error: "unauthorized" }); else json(res, 200, { ok: true }); return true; }
const server = createServer((req, res) => { const started = now(); const method = req.method ?? "GET"; const ip = req.socket.remoteAddress ?? "?"; const url = new URL(req.url ?? "/", "http://x"); res.on("finish", () => o.log?.(`${method} ${url.pathname} ${res.statusCode} ${now() - started}ms`)); void (async () => { try { if (!o.hosts.includes(String(req.headers.host ?? ""))) return json(res, 421, { error: "this service does not answer to that host name" }); if (url.pathname === "/healthz") return json(res, 200, { ok: true }); if (url.pathname.startsWith("/internal/")) { if (await internal(req, res, method, url.pathname)) return; return json(res, 404, { error: "no route" }); } if (url.pathname.startsWith("/admin/")) { const t = bearer(req); if (t === undefined || !sameSecret(t, o.adminToken)) return json(res, 401, { error: "unauthorized" }); return await admin(req, res, method, url); } if (url.pathname.startsWith("/api/")) { if (await anonymous(req, res, method, url.pathname, ip)) return; return await device(req, res, method, url, bearer(req), ip); } return json(res, 404, { error: "no route" }); } catch (err) { if (err instanceof Refused) { if (!res.headersSent) json(res, err.status, { error: err.message }); if (err.status === 413) req.destroy(); return; } if (!res.headersSent) json(res, 500, { error: "internal error" }); o.log?.(`error: ${err instanceof Error ? err.message : String(err)}`); } })(); }); server.requestTimeout = 120_000; server.headersTimeout = 20_000; server.maxHeadersCount = 50; return server;}