From 1d8a244ba42bcea8def1751b75fb6de5c8bc28a1 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 14 Oct 2026 22:16:50 +0200 Subject: [PATCH] Say every forge event on the bus through a signed system webhook, and announce a head or merge at once (novox/hq issue 729) The gitea module keeps one system webhook for every event family, made or corrected at start with the admin account (POST /admin/hooks, forge 1.27.3) and a secret it mints and keeps. Its receiver, at the docker bridge's gateway, refuses an unsigned request before reading it, a body over 5 MB, and a signature that does not verify; it answers a verified delivery at once, drops one already heard by its delivery id, and publishes each as forge.. on the module's events (its own stream, ADR 0363, follows issue 725). A pull request opened, pushed to or merged is announced at once. The polls look every five minutes only while deliveries arrive, and every 30 s again, saying so loudly, when one finds what no event said or ten minutes pass without a delivery. The announced records keep both paths from announcing twice. --- modules/gitea/client.ts | 5 + modules/gitea/index.ts | 216 ++++++++++++++------ modules/gitea/module.json | 58 +++++- modules/gitea/test/webhook.test.ts | 120 +++++++++++ modules/gitea/webhook.ts | 314 +++++++++++++++++++++++++++++ 5 files changed, 645 insertions(+), 68 deletions(-) create mode 100644 modules/gitea/test/webhook.test.ts create mode 100644 modules/gitea/webhook.ts diff --git a/modules/gitea/client.ts b/modules/gitea/client.ts index 55bba9dc..389f3b62 100644 --- a/modules/gitea/client.ts +++ b/modules/gitea/client.ts @@ -914,6 +914,11 @@ export class GiteaAdmin { return { status: res.status, body }; } + /** One admin-API call with a JSON body, answering the status and body (the system webhook, novox/hq issue 729). */ + call(method: string, path: string, body?: unknown): Promise { + return this.request(path, { method, ...(body === undefined ? {} : { body: JSON.stringify(body) }) }); + } + /** Fail with the forge's own message when a status the caller did not expect comes back. */ private static fail(path: string, res: AdminResponse): never { const detail = typeof res.body === "string" ? res.body : JSON.stringify(res.body); diff --git a/modules/gitea/index.ts b/modules/gitea/index.ts index d64045a7..8efe7c73 100644 --- a/modules/gitea/index.ts +++ b/modules/gitea/index.ts @@ -40,10 +40,11 @@ import { emit, on } from "@novox/mesh-sdk/events"; import { registerModuleTools } from "@novox/mesh-sdk/tools"; import { commitStatusTool } from "./delivery.js"; import { mergeTool } from "./merge.js"; -import { GiteaClient, movedSince } from "./client.js"; +import { GiteaAdmin, GiteaClient, movedSince } from "./client.js"; import { CHECK_CONTEXT, REPO_CHECK_CONTEXT, commentFor, headsToAnnounce, statusesFor, type Announced, type Checked, type RepoCheckFacts, reposToLook } from "./pulls.js"; import { GateWriter, clonedAs, type BuildStarted } from "./gate.js"; +import { HOOK_PATH, HOOK_PORT, Pace, claimHead, ensureSystemHook, keptSecret, receiver, type ForgeEvent } from "./webhook.js"; import { announcementOf, keepTook, lateVerdictComment, outOfDateWords, sameCommit, tookFile, weighTrunks } from "./trunk.js"; // Without a way to a token — configured, or mintable with the admin account (token.ts) — there is @@ -145,57 +146,9 @@ async function pollMerged(client: GiteaClient): Promise { // at once. const fresh = !!pull.merged_at && !!since && pull.merged_at > since; if (primedMerges && fresh) { - // What it changed, asked for only now: a module is rebuilt because a file inside its own - // directory moved, and without this every module built from a repository is rebuilt for a - // change to any of them (novox/hq 04-ISSUES/131). - const changed = await client.listPullFiles(repo.owner, repo.name, pull.number); - // Which of the directories they are in hold a module at the merge commit (novox/hq issue 278): a - // change inside one is that module's, held or not, and only a file in none is shared code. Not - // said when the list is cut or the forge could not be asked; the mesh then keeps its old rule. - let moduleDirs: string[] | null = null; - if (!changed.truncated) { - try { - moduleDirs = await client.moduleDirsAt(repo.owner, repo.name, pull.merge_commit_sha, changed.paths); - } catch (err) { - console.error(`[gitea] ${repo.full_name}#${pull.number}: which directories hold a module could not be read, ` + - `so the mesh reads its files by the old rule — ${err instanceof Error ? err.message : String(err)}`); - } - } - // The head it merged, and what its head was checked as (novox/hq ADR 0239): the delivery it was, made - // from the forge's word when its owner never heard the head. - let headChecks: Record | null = null; - if (pull.head_sha) { - try { - headChecks = await client.commitStatuses(repo.owner, repo.name, pull.head_sha); - } catch (err) { - console.error(`[gitea] ${repo.full_name}#${pull.number}: its head's statuses could not be read — ${err instanceof Error ? err.message : err}`); - } - } - await emit("pull.merged", { - owner: repo.owner, - repo: repo.name, - number: pull.number, - title: pull.title, - body: pull.body, - head_sha: pull.head_sha, - ...(headChecks ? { head_checks: headChecks } : {}), - head: pull.head, - base: pull.base, - merge_commit_sha: pull.merge_commit_sha, - merged_at: pull.merged_at, - clone_url: repo.clone_url, - html_url: pull.html_url, - paths: changed.paths, - paths_truncated: changed.truncated, - // Which of them the merge deleted (novox/hq ADR 0236): a module whose manifest went is forgotten, - // not built. - removed: changed.removed, - ...(moduleDirs ? { module_dirs: moduleDirs, module_dirs_said: true } : {}), - }); - // Said, because a trigger that fires silently is indistinguishable from one that did not - // fire (novox/hq 04-ISSUES/131) — this line is how an operator knows the mesh was told. - console.log(`[gitea] announced merge ${repo.full_name}#${pull.number} (${pull.merge_commit_sha.slice(0, 8)}) into ${pull.base}`); + if (await announceMerge(client, repo, pull)) pace.miss(`the merge of ${repo.full_name}#${pull.number}`); } + else announced.add(pull.merge_commit_sha); announced.add(pull.merge_commit_sha); changed = true; } @@ -206,6 +159,73 @@ async function pollMerged(client: GiteaClient): Promise { lastLook = began; } +type RepoOf = { owner: string; name: string; full_name: string; clone_url?: string }; +type PullOf = Awaited>[number]; + +/** **A merge, announced once, whichever path saw it first** (novox/hq issue 729): the poll or the forge's event. */ +async function announceMerge(client: GiteaClient, repo: RepoOf, pull: PullOf): Promise { + if (!pull.merged || !pull.merge_commit_sha || announced.has(pull.merge_commit_sha)) return false; + // Recorded before the first wait, so the other path finds it taken while this one is still asking the forge. + announced.add(pull.merge_commit_sha); + try { + // What it changed, asked for only now: a module is rebuilt because a file inside its own + // directory moved, and without this every module built from a repository is rebuilt for a + // change to any of them (novox/hq 04-ISSUES/131). + const changed = await client.listPullFiles(repo.owner, repo.name, pull.number); + // Which of the directories they are in hold a module at the merge commit (novox/hq issue 278): a + // change inside one is that module's, held or not, and only a file in none is shared code. Not + // said when the list is cut or the forge could not be asked; the mesh then keeps its old rule. + let moduleDirs: string[] | null = null; + if (!changed.truncated) { + try { + moduleDirs = await client.moduleDirsAt(repo.owner, repo.name, pull.merge_commit_sha, changed.paths); + } catch (err) { + console.error(`[gitea] ${repo.full_name}#${pull.number}: which directories hold a module could not be read, ` + + `so the mesh reads its files by the old rule — ${err instanceof Error ? err.message : String(err)}`); + } + } + // The head it merged, and what its head was checked as (novox/hq ADR 0239): the delivery it was, made + // from the forge's word when its owner never heard the head. + let headChecks: Record | null = null; + if (pull.head_sha) { + try { + headChecks = await client.commitStatuses(repo.owner, repo.name, pull.head_sha); + } catch (err) { + console.error(`[gitea] ${repo.full_name}#${pull.number}: its head's statuses could not be read — ${err instanceof Error ? err.message : err}`); + } + } + await emit("pull.merged", { + owner: repo.owner, + repo: repo.name, + number: pull.number, + title: pull.title, + body: pull.body, + head_sha: pull.head_sha, + ...(headChecks ? { head_checks: headChecks } : {}), + head: pull.head, + base: pull.base, + merge_commit_sha: pull.merge_commit_sha, + merged_at: pull.merged_at, + clone_url: repo.clone_url, + html_url: pull.html_url, + paths: changed.paths, + paths_truncated: changed.truncated, + // Which of them the merge deleted (novox/hq ADR 0236): a module whose manifest went is forgotten, + // not built. + removed: changed.removed, + ...(moduleDirs ? { module_dirs: moduleDirs, module_dirs_said: true } : {}), + }); + // Said, because a trigger that fires silently is indistinguishable from one that did not + // fire (novox/hq 04-ISSUES/131) — this line is how an operator knows the mesh was told. + console.log(`[gitea] announced merge ${repo.full_name}#${pull.number} (${pull.merge_commit_sha.slice(0, 8)}) into ${pull.base}`); + } catch (err) { + announced.delete(pull.merge_commit_sha); + throw err; + } + keepAnnounced(); + return true; +} + // **Every new head of an open pull request is announced, once** (novox/hq to-be 45 §9): the mesh checks it // against every machine of its facts before it merges. Kept beside the merges' record, so a restart // announces nothing twice; the first look on a machine with no record announces only what moved in the @@ -228,6 +248,31 @@ function keepHeads(): void { writeFileSync(tmp, JSON.stringify({ heads })); renameSync(tmp, pullsRecord); } +/** **A head, announced once, whichever path saw it first** (novox/hq issue 729): the poll or the forge's event. */ +async function announceHead(client: GiteaClient, repo: RepoOf, pull: OpenOf): Promise { + const key = `${repo.full_name}#${pull.number}`; + const sha = String(pull.head_sha ?? ""); + const before = heads[key]; + // Claimed before the first wait, so the other path finds it taken while this one is still announcing. + if (!claimHead(heads, key, sha)) return false; + try { + // The announcement the merge tool makes too when it asks a check (trunk.ts, novox/hq ADR 0323 §5). + await emit("pull.updated", await announcementOf(client, repo.owner, repo.name, repo.clone_url ?? "", pull)); + } catch (err) { + if (before === undefined) delete heads[key]; + else heads[key] = before; + throw err; + } + keepHeads(); + // Said, so an operator knows the mesh was asked to check it. + console.log(`[gitea] announced ${key} at ${sha.slice(0, 8)} to be checked before it merges`); + // Pending until the verdict comes, so the pull request says a check is running rather than nothing — and + // where it is: in the build queue from now, until a build machine takes it (novox/hq issue 446). + await gates.pending(repo.owner, repo.name, sha); + return true; +} +type OpenOf = Awaited>[number]; + let lastPullLook = ""; // **A sweep now and then looks at every repository** (novox/hq issue 347), whatever moved: the floor is a // heuristic, and a heuristic that misses is caught by the next sweep rather than never. @@ -259,15 +304,8 @@ async function pollPulls(client: GiteaClient): Promise { const key = `${repo.full_name}#${pull.number}`; const fresh = primedPulls || (!!pull.updated_at && pull.updated_at > dayAgo); if (fresh) { - // The announcement the merge tool makes too when it asks a check (trunk.ts, novox/hq ADR 0323 §5). - await emit("pull.updated", await announcementOf(client, repo.owner, repo.name, repo.clone_url, pull)); - // Said, so an operator knows the mesh was asked to check it. - console.log(`[gitea] announced ${key} at ${String(pull.head_sha).slice(0, 8)} to be checked before it merges`); - // Pending until the verdict comes, so the pull request says a check is running rather than nothing — and - // where it is: in the build queue from now, until a build machine takes it (novox/hq issue 446). - await gates.pending(repo.owner, repo.name, String(pull.head_sha)); - } - heads[key] = String(pull.head_sha); + if (await announceHead(client, repo, pull)) pace.miss(`${key} at ${String(pull.head_sha).slice(0, 8)}`); + } else heads[key] = String(pull.head_sha); changed = true; } if (await trunksMoved(client, repo.owner, repo.name)) changedTrunks = true; @@ -413,12 +451,59 @@ async function buildStarted(event: { body: unknown }): Promise { } } +/** How fast the polls look: 30 s until the forge's events are heard, five minutes while they are (webhook.ts). */ +const pace = new Pace(); + +/** What the forge's event asks of the announcer at once (issue 729): a new head is announced, and a merge. */ +async function actOn(client: GiteaClient, event: ForgeEvent): Promise { + const b = event.body as { owner?: string; repo?: string; number?: number }; + if (!b.owner || !b.repo || !b.number) return; + const head = ["forge.pull-request.opened", "forge.pull-request.synchronized", "forge.pull-request.reopened"].includes(event.name); + const merged = event.name === "forge.pull-request.merged"; + if (!head && !merged) return; + const [repo, pull] = await Promise.all([client.getRepo(b.owner, b.repo), client.getPullRequest(b.owner, b.repo, b.number)]); + if (head && pull.state === "open") await announceHead(client, repo, pull); + if (merged) await announceMerge(client, repo, pull); +} + +/** + * **The receiver, and the forge's one system webhook pointing at it** (issue 729). The secret is minted and kept beside + * the token; the webhook is made or corrected with the admin account (ADR 0266: the tools' token holds no admin scope). + * Answers whether events are heard; a forge that refuses is said, and the polls keep their pace. + */ +async function listenForForge(client: GiteaClient): Promise { + const stateDir = process.env.MESH_GITEA_STATE_DIR; + const url = process.env.MESH_GITEA_HOOK_URL; + if (!stateDir || !url) { + console.log("[gitea] no webhook receiver: MESH_GITEA_STATE_DIR or MESH_GITEA_HOOK_URL unset, so the polls alone watch the forge"); + return false; + } + try { + const secret = keptSecret(join(stateDir, "webhook.secret")); + const port = Number(process.env.MESH_GITEA_HOOK_PORT ?? HOOK_PORT); + const server = receiver({ secret, port, heard: () => pace.heard(), + publish: (event) => emit(event.name, event.body), + act: (event) => actOn(client, event) }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(port, process.env.MESH_GITEA_HOOK_HOST ?? "0.0.0.0", () => resolve()); + }); + const did = await ensureSystemHook(GiteaAdmin.fromEnv(), url.replace(/\/+$/, "") + HOOK_PATH, secret); + console.log(`[gitea] listening for the forge's events on port ${port}; its system webhook ${did}, every event, signed`); + return true; + } catch (err) { + console.error(`[gitea] the forge's events are not heard, so the polls keep their pace — ${err instanceof Error ? err.message : err}`); + return false; + } +} + if (gitea) { const client = gitea; // A poll that fails says so once, not once a minute: the same reason repeating (the forge not up // yet, the admin account refused on a restored forge) is one fact, and a recovery is worth a line. let failing: string | null = null; - const tick = (fn: () => Promise, everyMs: number): void => { + const tick = (fn: () => Promise, every: number | (() => number)): void => { + const everyMs = (): number => (typeof every === "number" ? every : every()); // **One pass at a time** (novox/hq issue 250): the next pass is scheduled when this one has ended, so a // pass that outlasts its interval delays the next instead of running beside it — two passes at once // could each announce the same merge before either recorded it. @@ -433,12 +518,15 @@ if (gitea) { if (why !== failing) console.error(`[gitea] not watching until this clears — ${why}`); failing = why; }) - .finally(() => setTimeout(run, everyMs)); + .finally(() => setTimeout(run, everyMs())); run(); }; + // **The forge's events first, the polls as the safety net** (novox/hq issue 729): while verified deliveries arrive the + // polls slow to every five minutes and announce only what no event said (Pace); otherwise they keep 30 s. + await listenForForge(client); tick(() => pollRepos(client), 60_000); - tick(() => pollMerged(client), 30_000); - tick(() => pollPulls(client), 30_000); + tick(() => pollMerged(client), () => pace.every()); + tick(() => pollPulls(client), () => pace.every()); await on("mesh-controller.checked", (event) => { const c = (event.body ?? {}) as Checked; // A head's own verdict takes its turn after every write of that head's gate already asked for (gate.ts). diff --git a/modules/gitea/module.json b/modules/gitea/module.json index 069d7163..210c2437 100644 --- a/modules/gitea/module.json +++ b/modules/gitea/module.json @@ -42,7 +42,47 @@ "issue.opened", "pull.merged", "pull.updated", - "pull.closed" + "pull.closed", + "forge.push", + "forge.create", + "forge.delete", + "forge.fork", + "forge.status", + "forge.pull-request.opened", + "forge.pull-request.reopened", + "forge.pull-request.synchronized", + "forge.pull-request.edited", + "forge.pull-request.closed", + "forge.pull-request.merged", + "forge.pull-request.labeled", + "forge.pull-request.unlabeled", + "forge.pull-request.assigned", + "forge.pull-request.unassigned", + "forge.pull-request.reviewed", + "forge.pull-request.review-requested", + "forge.pull-request.commented", + "forge.pull-request.changed", + "forge.issue-comment.created", + "forge.issue-comment.edited", + "forge.issue-comment.deleted", + "forge.issues.opened", + "forge.issues.reopened", + "forge.issues.closed", + "forge.issues.edited", + "forge.issues.labeled", + "forge.issues.unlabeled", + "forge.issues.assigned", + "forge.issues.unassigned", + "forge.issues.changed", + "forge.release.published", + "forge.release.updated", + "forge.release.deleted", + "forge.release.changed", + "forge.repository.created", + "forge.repository.deleted", + "forge.repository.changed", + "forge.wiki.changed", + "forge.other" ], "consumes": [ "mesh-controller.checked", @@ -61,7 +101,14 @@ "port": 22, "protocol": "tcp", "from": "mesh", - "why": "git over ssh, gitea's own unmodified sshd. Published on the machine's own side at 222, the mesh's fixed public convention \u2014 not 22, which the machine's own daemon holds and a module does not take" + "why": "git over ssh, gitea's own unmodified sshd. Published on the machine's own side at 222, the mesh's fixed public convention — not 22, which the machine's own daemon holds and a module does not take" + }, + { + "name": "forge-events", + "port": 3047, + "protocol": "tcp", + "from": "machine", + "why": "the forge's system webhook, every event signed with a secret the module keeps, from the forge's container on the docker bridge to the module's receiver at the bridge's gateway 172.17.0.1 (novox/hq issue 729)" } ], "serves": { @@ -153,7 +200,7 @@ "type": "file", "path": "${dir:state}/server.env", "mode": "0600", - "content": "GITEA__security__INTERNAL_TOKEN=${secret:internal-token}\nGITEA__database__DB_TYPE=postgres\nGITEA__database__HOST=${bound:postgres-database:at}:${bound:postgres-database:port}\nGITEA__database__NAME=${bound:postgres-database:as}\nGITEA__database__USER=${bound:postgres-database:as}\nGITEA__database__PASSWD=${secret:postgres-database}\n" + "content": "GITEA__security__INTERNAL_TOKEN=${secret:internal-token}\nGITEA__database__DB_TYPE=postgres\nGITEA__database__HOST=${bound:postgres-database:at}:${bound:postgres-database:port}\nGITEA__database__NAME=${bound:postgres-database:as}\nGITEA__database__USER=${bound:postgres-database:as}\nGITEA__database__PASSWD=${secret:postgres-database}\nGITEA__webhook__ALLOWED_HOST_LIST=172.17.0.1\n" }, { "id": "data", @@ -263,7 +310,10 @@ "MESH_GITEA_ADMIN_PASSWORD_FILE": "${dir:state}/admin.secret", "MESH_GITEA_STATE_DIR": "${dir:runtime-state}", "MESH_RECEIVES": "${dir:grants}/npm.json", - "MESH_GITEA_READY_AUTHORS_FILE": "${dir:mesh-state}/ready-authors" + "MESH_GITEA_READY_AUTHORS_FILE": "${dir:mesh-state}/ready-authors", + "MESH_GITEA_HOOK_URL": "http://172.17.0.1:3047", + "MESH_GITEA_HOOK_PORT": "3047", + "MESH_GITEA_HOOK_HOST": "172.17.0.1" } }, { diff --git a/modules/gitea/test/webhook.test.ts b/modules/gitea/test/webhook.test.ts new file mode 100644 index 00000000..453ff2d0 --- /dev/null +++ b/modules/gitea/test/webhook.test.ts @@ -0,0 +1,120 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { createHmac } from "node:crypto"; +import type { AddressInfo } from "node:net"; + +// **The forge says every event on the bus** (novox/hq issue 729): a signed delivery is published once and acted on, +// an unsigned or unverified one is refused, and an event and the poll seeing the same head announce it once. + +const secret = "s".repeat(64); +const sign = (body: string, key = secret) => createHmac("sha256", key).update(body).digest("hex"); +const opened = JSON.stringify({ action: "opened", number: 7, pull_request: { number: 7, title: "t", user: { login: "jo" }, + head: { ref: "feat/x", sha: "a".repeat(40) }, base: { ref: "main" }, merged: false }, + repository: { name: "hq", full_name: "novox/hq", owner: { login: "novox" } }, sender: { login: "jo" } }); + +async function serve(published: { name: string }[], acted: { name: string }[]) { + const { receiver, HOOK_PATH } = await import("../webhook.ts"); + const server = receiver({ secret, port: 0, log: () => {}, + publish: async (e) => { published.push(e); }, act: async (e) => { acted.push(e); } }); + await new Promise((r) => server.listen(0, "127.0.0.1", () => r())); + const url = `http://127.0.0.1:${(server.address() as AddressInfo).port}${HOOK_PATH}`; + return { server, url }; +} + +test("a signed event is published on the bus and acted on once", async () => { + const published: { name: string }[] = [], acted: { name: string }[] = []; + const { server, url } = await serve(published, acted); + try { + const headers = { "X-Gitea-Event": "pull_request", "X-Gitea-Signature": sign(opened), "X-Gitea-Delivery": "d-1" }; + const res = await fetch(url, { method: "POST", body: opened, headers }); + assert.equal(res.status, 200); + // The same delivery sent again is answered and published nowhere. + const again = await fetch(url, { method: "POST", body: opened, headers }); + assert.equal(again.status, 200); + await server.idle(); + assert.deepEqual(published.map((e) => e.name), ["forge.pull-request.opened"]); + assert.deepEqual(acted.map((e) => e.name), ["forge.pull-request.opened"]); + } finally { server.close(); } +}); + +test("an unsigned event, or one whose signature does not verify, is refused and published nowhere", async () => { + const published: { name: string }[] = [], acted: { name: string }[] = []; + const { server, url } = await serve(published, acted); + try { + const unsigned = await fetch(url, { method: "POST", body: opened, headers: { "X-Gitea-Event": "pull_request" } }); + assert.equal(unsigned.status, 401); + assert.match(await unsigned.text(), /unsigned/); + const forged = await fetch(url, { method: "POST", body: opened, + headers: { "X-Gitea-Event": "pull_request", "X-Gitea-Signature": sign(opened, "another secret") } }); + assert.equal(forged.status, 401); + assert.match(await forged.text(), /does not verify/); + const big = "x".repeat(5 * 1024 * 1024 + 1); + const large = await fetch(url, { method: "POST", body: big, + headers: { "X-Gitea-Event": "push", "X-Gitea-Signature": sign(big) } }); + assert.equal(large.status, 413); + await server.idle(); + assert.deepEqual(published, []); + assert.deepEqual(acted, []); + } finally { server.close(); } +}); + +test("an event and the poll seeing the same head announce it once", async () => { + const { claimHead } = await import("../webhook.ts"); + const heads: Record = {}; + const sha = "a".repeat(40); + assert.equal(claimHead(heads, "novox/hq#7", sha), true, "the event claims the head and announces it"); + assert.equal(claimHead(heads, "novox/hq#7", sha), false, "the poll finds it taken"); + assert.equal(claimHead(heads, "novox/hq#7", "b".repeat(40)), true, "a new head is announced again"); +}); + +test("each forge event is named in the module's stream and trimmed", async () => { + const { eventOf } = await import("../webhook.ts"); + const merged = eventOf("pull_request", { action: "closed", pull_request: { number: 3, merged: true, merge_commit_sha: "c".repeat(40), + head: { sha: "a".repeat(40) }, base: { ref: "main" } }, repository: { name: "hq", owner: { login: "novox" } } }); + assert.equal(merged.name, "forge.pull-request.merged"); + assert.equal(merged.body.merge_commit_sha, "c".repeat(40)); + assert.equal(eventOf("pull_request", { action: "label_updated", pull_request: { number: 3 }, label: { name: "mesh/ready" } }).name, + "forge.pull-request.labeled"); + const comment = eventOf("issue_comment", { action: "created", is_pull: true, issue: { number: 3 }, + comment: { id: 9, body: "ready at abc", user: { login: "jo" } } }); + assert.equal(comment.name, "forge.pull-request.commented"); + assert.equal(comment.body.comment, "ready at abc"); + assert.equal(eventOf("push", { ref: "refs/heads/main", after: "d".repeat(40) }).name, "forge.push"); + assert.equal(eventOf("pull_request_review_approved", {}).name, "forge.other"); +}); + +test("the system webhook is made when none points at the receiver, and corrected when one does", async () => { + const { ensureSystemHook } = await import("../webhook.ts"); + const calls: string[] = []; + let hooks: unknown[] = []; + const admin = { call: async (method: string, path: string, body?: any) => { + calls.push(`${method} ${path}`); + if (method === "GET") return { status: 200, body: hooks }; + if (method === "POST") { assert.ok(body.events.includes("pull_request") && body.events.includes("push")); assert.equal(body.config.secret, secret); return { status: 201, body: {} }; } + return { status: 200, body: {} }; + } }; + assert.equal(await ensureSystemHook(admin, "http://h:3047/forge-event", secret), "created"); + hooks = [{ id: 4, config: { url: "http://h:3047/forge-event" } }]; + assert.equal(await ensureSystemHook(admin, "http://h:3047/forge-event", secret), "corrected"); + assert.ok(calls.includes("PATCH /admin/hooks/4")); +}); + +test("the polls look every 30 s until events are heard, every five minutes while they are, and 30 s again on a miss", async () => { + const { Pace } = await import("../webhook.ts"); + let now = 1_000_000; + const said: string[] = []; + const pace = new Pace(() => now, (l) => said.push(l)); + assert.equal(pace.every(), 30_000, "no delivery yet"); + pace.heard(); + assert.equal(pace.every(), 300_000, "a delivery heard"); + now += 11 * 60_000; + assert.equal(pace.every(), 30_000, "ten minutes without a delivery"); + pace.heard(); + now += 1000; + pace.miss("novox/hq#7 at aaaaaaaa"); + assert.equal(pace.every(), 30_000, "the poll found what no event said"); + assert.match(said.join("\n"), /MISSED novox\/hq#7/); + now += 1000; + pace.heard(); + assert.equal(pace.every(), 300_000, "the next delivery"); +}); diff --git a/modules/gitea/webhook.ts b/modules/gitea/webhook.ts new file mode 100644 index 00000000..130a4e01 --- /dev/null +++ b/modules/gitea/webhook.ts @@ -0,0 +1,314 @@ +// **The forge says every event on the bus** (novox/hq issue 729; to-be 10: an event makes it fast, nothing makes it +// necessary). The module keeps one system webhook on its forge for every event, pointing at its own receiver, signed +// with a secret it mints and keeps beside its token; the receiver refuses whatever is unsigned or does not verify, and +// publishes each event, trimmed, on the module's events as forge.. (forge.pull-request.opened, forge.push, …), +// on the shared EVENTS path its other events take; its own stream (ADR 0363) is a follow-up once issue 725 ships. +// The polls stay as the safety net, slowed; the announced-heads and announced-merges records keep anything from +// being announced twice, whichever path saw it first. +// +// The secret is never logged: lines say it was minted or reused and where it is kept, never what it is. + +import { createHmac, randomBytes, timingSafeEqual } from "node:crypto"; +import { chmodSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs"; +import { createServer, type Server } from "node:http"; +import { dirname } from "node:path"; + +/** The name the receiver answers on, and the port it listens on unless the manifest says otherwise. */ +export const HOOK_PATH = "/forge-event"; +export const HOOK_PORT = 3047; + +/** The secret the forge signs with: read from its 0600 file, or minted and kept there. */ +export function keptSecret(file: string, log: (line: string) => void = (l) => console.log(`[gitea] ${l}`)): string { + try { + const kept = readFileSync(file, "utf8").trim(); + if (kept.length >= 32) { + log(`reusing the webhook secret kept at ${file}`); + return kept; + } + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err; + } + const secret = randomBytes(32).toString("hex"); + mkdirSync(dirname(file), { recursive: true, mode: 0o700 }); + const tmp = `${file}.tmp`; + writeFileSync(tmp, secret + "\n", { mode: 0o600 }); + chmodSync(tmp, 0o600); + renameSync(tmp, file); + log(`minted a webhook secret, kept at ${file}`); + return secret; +} + +/** Whether a body carries the forge's HMAC-SHA256 signature (X-Gitea-Signature, hex) under the secret. */ +export function verified(secret: string, body: Buffer | string, signature: string | undefined): boolean { + if (!secret || !signature || !/^[0-9a-f]{64}$/i.test(signature.trim())) return false; + const want = createHmac("sha256", secret).update(body).digest(); + const got = Buffer.from(signature.trim(), "hex"); + return got.length === want.length && timingSafeEqual(got, want); +} + +/** One forge event as the bus carries it: its name inside the stream, and what is needed of it — no secrets. */ +export interface ForgeEvent { + name: string; + body: Record; +} + +/** Every event name the module emits for the forge, without its forge. prefix (module.json `emits`). */ +export const EVENTS: readonly string[] = ["push", "create", "delete", "fork", "status", "pull-request.opened", "pull-request.reopened", "pull-request.synchronized", "pull-request.edited", "pull-request.closed", "pull-request.merged", "pull-request.labeled", "pull-request.unlabeled", "pull-request.assigned", "pull-request.unassigned", "pull-request.reviewed", "pull-request.review-requested", "pull-request.commented", "pull-request.changed", "issue-comment.created", "issue-comment.edited", "issue-comment.deleted", "issues.opened", "issues.reopened", "issues.closed", "issues.edited", "issues.labeled", "issues.unlabeled", "issues.assigned", "issues.unassigned", "issues.changed", "release.published", "release.updated", "release.deleted", "release.changed", "repository.created", "repository.deleted", "repository.changed", "wiki.changed", "other"]; + +/** The families the module's stream carries (module.json `streams`); any other kind is said as other.; underscores become hyphens. */ +export const FAMILIES = ["push", "create", "delete", "fork", "status", "pull_request", "issue_comment", "issues", + "release", "repository", "wiki"] as const; + +const ACTIONS: Record = { label_updated: "labeled", label_cleared: "unlabeled", synchronized: "synchronized" }; + +function str(v: unknown): string | undefined { + return typeof v === "string" && v ? v : undefined; +} +function num(v: unknown): number | undefined { + return typeof v === "number" ? v : undefined; +} +function login(u: any): string | undefined { + return str(u?.login) ?? str(u?.username); +} + +/** + * The event a forge delivery is: its kind (X-Gitea-Event) and the payload's action, trimmed to ids, repository, refs, + * shas, author, labels and a comment's body (a ready mark is one, ADR 0364). A merge is pull-request.merged, a comment on + * a pull request pull-request.commented. + */ +export function eventOf(kind: string, payload: any): ForgeEvent { + // The bus's event names take lower-case letters, digits and hyphens: the forge's underscores become hyphens. + const event = forgeEventOf(kind, payload); + const name = event.name.toLowerCase().replace(/_/g, "-").replace(/[^a-z0-9.-]/g, "-"); + // Published on the module's events as it publishes its others, each name declared in its `emits`: one the manifest + // does not name is forge.other, saying what the forge called it. Its own stream (ADR 0363) follows issue 725. + if (EVENTS.includes(name)) return { name: `forge.${name}`, body: event.body }; + return { name: "forge.other", body: { ...event.body, kind, said: name } }; +} + +function forgeEventOf(said: string, payload: any): ForgeEvent { + // The forge names some kinds finer than their family (pull_request_label, pull_request_sync): read by the payload. + let kind = said; + if (said.startsWith("pull_request") && payload?.pull_request) kind = "pull_request"; + if ((said.startsWith("issue") || said === "pull_request_comment") && payload?.comment) kind = "issue_comment"; + if (said.startsWith("issue") && !payload?.comment && payload?.issue) kind = "issues"; + const repo = payload?.repository ?? {}; + const base: Record = { + owner: login(repo.owner), repo: str(repo.name), full_name: str(repo.full_name), clone_url: str(repo.clone_url), + sender: login(payload?.sender), + }; + const family = (FAMILIES as readonly string[]).includes(kind) ? kind : `other`; + let action = str(payload?.action); + const pull = payload?.pull_request; + if (kind === "push") { + return { name: "push", body: { ...base, ref: str(payload.ref), before: str(payload.before), after: str(payload.after), + commits: Array.isArray(payload.commits) ? payload.commits.length : undefined, pusher: login(payload.pusher) } }; + } + if (kind === "pull_request" && pull) { + if (action === "closed" && pull.merged) action = "merged"; + action = ACTIONS[action ?? ""] ?? action ?? "changed"; + return { name: `pull_request.${action}`, body: { ...base, number: num(pull.number ?? payload.number), + title: str(pull.title), author: login(pull.user), head: str(pull.head?.ref), head_sha: str(pull.head?.sha), + base: str(pull.base?.ref), merged: !!pull.merged, merge_commit_sha: str(pull.merge_commit_sha), + merged_at: str(pull.merged_at), html_url: str(pull.html_url), + labels: Array.isArray(pull.labels) ? pull.labels.map((l: any) => str(l?.name)).filter(Boolean) : undefined, + label: str(payload.label?.name) } }; + } + if (kind === "issue_comment") { + const issue = payload.issue ?? {}; + const onPull = !!payload.is_pull || !!issue.pull_request; + return { name: onPull ? "pull_request.commented" : `issue_comment.${action ?? "created"}`, body: { ...base, + number: num(issue.number), comment_id: num(payload.comment?.id), author: login(payload.comment?.user), + comment: str(payload.comment?.body), action } }; + } + if (kind === "create" || kind === "delete") { + return { name: kind, body: { ...base, ref: str(payload.ref), ref_type: str(payload.ref_type), sha: str(payload.sha) } }; + } + if (kind === "status") { + return { name: "status", body: { ...base, sha: str(payload.sha), context: str(payload.context), state: str(payload.state) } }; + } + const name = family === "other" ? `other.${kind.replace(/[^a-z0-9_-]/gi, "_") || "unknown"}` : + ["fork"].includes(family) ? family : `${family}.${action ?? "changed"}`; + return { name, body: { ...base, number: num(payload?.number ?? payload?.issue?.number), action } }; +} + +export interface ReceiverOptions { + secret: string; + port: number; + host?: string; + /** Publishes one event on the bus. */ + publish: (event: ForgeEvent) => Promise; + /** Acts on it at once, beside publishing it (the announcer). */ + act?: (event: ForgeEvent) => Promise; + /** Told of every verified delivery, so the polls know events are heard. */ + heard?: () => void; + log?: (line: string) => void; +} + +/** The most a delivery may carry: the forge's push of many commits stays well under it. */ +export const MAX_BODY = 5 * 1024 * 1024; +/** How many deliveries' ids are remembered, so a delivery sent again is published once. */ +export const REMEMBERED = 2000; + +/** + * The receiver. A request without a signature is refused before its body is read, a body over MAX_BODY is refused, and + * one whose signature does not verify is refused, each with 401 or 413 and a line saying so. A verified delivery is + * answered 200 at once and queued: it is published and acted on after the reply, one at a time, so the forge's + * delivery timeout is never reached. A delivery whose X-Gitea-Delivery id was already heard is answered and dropped. + */ +export function receiver(opts: ReceiverOptions): Server & { idle(): Promise } { + const log = opts.log ?? ((l) => console.log(`[gitea] ${l}`)); + const seen = new Set(); + let queue: Promise = Promise.resolve(); + const server = createServer((req, res) => { + if (req.method !== "POST" || (req.url ?? "").split("?")[0] !== HOOK_PATH) { + res.statusCode = 404; + return res.end(); + } + const from = req.socket.remoteAddress ?? "?"; + const signature = String(req.headers["x-gitea-signature"] ?? "").trim(); + if (!signature) { + log(`refused a forge event from ${from}: it is unsigned`); + res.writeHead(401, { Connection: "close" }); + res.end("unsigned\n"); + req.destroy(); + return; + } + if (Number(req.headers["content-length"] ?? 0) > MAX_BODY) { + log(`refused a forge event from ${from}: its body is over ${MAX_BODY} bytes`); + res.writeHead(413, { Connection: "close" }); + res.end("too large\n"); + req.destroy(); + return; + } + const chunks: Buffer[] = []; + let size = 0; + let over = false; + req.on("data", (c: Buffer) => { + size += c.length; + if (size > MAX_BODY) { + if (!over) { + over = true; + log(`refused a forge event from ${from}: its body is over ${MAX_BODY} bytes`); + res.writeHead(413, { Connection: "close" }); + res.end("too large\n"); + req.destroy(); + } + } else chunks.push(c); + }); + req.on("end", () => { + if (over) return; + const body = Buffer.concat(chunks); + if (!verified(opts.secret, body, signature)) { + log(`refused a forge event from ${from}: its signature does not verify`); + res.statusCode = 401; + return res.end("signature does not verify\n"); + } + let payload: unknown; + try { + payload = JSON.parse(body.toString("utf8")); + } catch { + res.statusCode = 400; + return res.end("not JSON\n"); + } + opts.heard?.(); + const delivery = String(req.headers["x-gitea-delivery"] ?? ""); + if (delivery && seen.has(delivery)) { + res.statusCode = 200; + return res.end("already heard\n"); + } + if (delivery) { + seen.add(delivery); + if (seen.size > REMEMBERED) seen.delete(seen.values().next().value as string); + } + const event = eventOf(String(req.headers["x-gitea-event"] ?? "other"), payload); + res.statusCode = 200; + res.end("queued\n"); + queue = queue.then(async () => { + try { + await opts.publish(event); + if (opts.act) await opts.act(event); + } catch (err) { + log(`a forge event ${event.name} could not be published or acted on; the polls catch it — ${err instanceof Error ? err.message : err}`); + } + }); + }); + }); + return Object.assign(server, { idle: () => queue }); +} + +/** What ensuring the system webhook needs of the forge's administration (GiteaAdmin's login). */ +export interface HookAdmin { + call(method: string, path: string, body?: unknown): Promise<{ status: number; body: any }>; +} + +/** + * **How fast the polls look** (issue 729): every five minutes only while the forge's events are heard — a verified + * delivery within the last ten minutes, and none missed since. Until the first delivery, after ten minutes without one, + * or once the poll found something no event said, every thirty seconds. + */ +export class Pace { + private last = 0; + private missed = 0; + private readonly now: () => number; + private readonly log: (l: string) => void; + constructor(now: () => number = Date.now, log: (l: string) => void = (l) => console.error(`[gitea] ${l}`)) { + this.now = now; + this.log = log; + } + heard(): void { + this.last = this.now(); + } + /** The poll found a head or merge no forge event announced. */ + miss(what: string): void { + if (this.hearing()) this.log(`THE FORGE'S EVENTS MISSED ${what}: the poll found it, so the polls look every 30 s again until the next event`); + this.missed = this.now(); + } + hearing(): boolean { + const t = this.now(); + return this.last > 0 && t - this.last < 10 * 60_000 && this.last > this.missed; + } + every(): number { + return this.hearing() ? 5 * 60_000 : 30_000; + } +} + +/** Every event family the forge (1.27) can send a webhook. */ +export const HOOK_EVENTS = ["create", "delete", "fork", "push", "issues", "issue_comment", "pull_request", + "pull_request_review_request", "wiki", "repository", "release", "package", "status", "workflow_run", "workflow_job"]; + +/** + * **One system webhook, for every event, at the receiver** (issue 729): created when none points there, corrected + * when one does but differs (events, active, content type), and its secret set again each start — the forge never + * hands a secret back, so a corrected secret cannot be told from a kept one. Answers what it did. + */ +export async function ensureSystemHook(admin: HookAdmin, url: string, secret: string): Promise<"created" | "corrected"> { + // POST /admin/hooks makes a system webhook, and its options take no flag for it (the forge's 1.27.3 API schema). The + // forge has no "every event" word: each family is named, and a family name (issues, pull_request) takes all of it. + const want = { type: "gitea", active: true, events: HOOK_EVENTS, branch_filter: "*", + config: { url, content_type: "json", secret } }; + const listed = await admin.call("GET", "/admin/hooks?type=system&limit=50"); + if (listed.status !== 200) throw new Error(`the forge's system webhooks could not be listed: ${listed.status}`); + const hooks: any[] = Array.isArray(listed.body) ? listed.body : []; + const ours = hooks.filter((h) => h?.config?.url === url); + if (ours.length === 0) { + const made = await admin.call("POST", "/admin/hooks", want); + if (made.status !== 201 && made.status !== 200) throw new Error(`the forge's system webhook could not be made: ${made.status}`); + return "created"; + } + for (const extra of ours.slice(1)) await admin.call("DELETE", `/admin/hooks/${extra.id}`); + const fixed = await admin.call("PATCH", `/admin/hooks/${ours[0].id}`, { active: true, events: HOOK_EVENTS, branch_filter: "*", + config: { url, content_type: "json", secret } }); + if (fixed.status !== 200) throw new Error(`the forge's system webhook could not be corrected: ${fixed.status}`); + return "corrected"; +} + +/** + * **Claim a head for announcing, once** (issue 729): true when `sha` is not already the head recorded for `key`, which + * it then is. The poll and the forge's event both claim before they announce, so whichever comes second finds it taken. + */ +export function claimHead(heads: Record, key: string, sha: string): boolean { + if (!sha || heads[key] === sha) return false; + heads[key] = sha; + return true; +}