Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262import { expect } from "@playwright/test";import { spawn, spawnSync } from "node:child_process";import { createHmac } from "node:crypto";import { existsSync, readFileSync, readlinkSync } from "node:fs";import { createServer, type IncomingHttpHeaders } from "node:http";import type { AddressInfo } from "node:net";import { createInterface } from "node:readline";
import { spindleGet, spindlePost } from "./api";import type { Account } from "./config";import { sleep, stamp, until } from "./net";import { resolve, type Ctx, type Lazy } from "./vocabulary";
export const WEBHOOKS = "org.tangled.webhook";
export interface Delivery { method?: string; headers: IncomingHttpHeaders; raw: string; body: any;}
// on linux the local stack has its own network namespace that can't reach the host's loopback,// so the receiver runs inside it and reports over stdout. elsewhere it listens hereconst inboxes = new Map<string, Delivery[]>();let serving: Promise<number> | undefined;
const receiver = `const server = require("node:http").createServer((req, res) => { const chunks = []; req.on("data", (chunk) => chunks.push(chunk)); req.on("end", () => { const raw = Buffer.concat(chunks).toString(); console.log(JSON.stringify({ url: req.url, method: req.method, headers: req.headers, raw })); res.end("ok"); });});server.listen(0, "127.0.0.1", () => console.log(JSON.stringify({ port: server.address().port })));process.stdin.on("end", () => process.exit());process.stdin.resume();`;
const take = ( url: string, method: string | undefined, headers: IncomingHttpHeaders, raw: string) => { const inbox = inboxes.get(url); if (!inbox) return false; let body: unknown; try { body = JSON.parse(raw); } catch {} inbox.push({ method, headers, raw, body }); return true;};
// ns.pid goes stale if linux-ns runs again, so look up process-compose by its socketconst namespaceOfTheStack = () => { const state = process.env.E2E_PC_STATE; if (!state) return undefined; let named: string; try { named = readFileSync(`${state}/run/ns.pid`, "utf8").trim(); } catch { return undefined; } if (existsSync(`/proc/${named}/ns/net`)) return named; // the wrapper matches the same pgrep but is still in the host namespace const here = readlinkSync("/proc/self/ns/net"); const found = spawnSync("pgrep", ["-f", `process-compose up .*${state}/run/pc.sock`], { encoding: "utf8" }) .stdout.split("\n") .find((pid) => pid && readlinkSync(`/proc/${pid}/ns/net`) !== here); if (!found) throw new Error( `${state}/run/ns.pid names ${named}, which is gone, and no process-compose is in the stack's namespace` ); return found;};
const serveInsideTheStack = (ns: string) => new Promise<number>((ready, failed) => { const child = spawn( "nsenter", ["--user", "--preserve-credentials", "--net", "-t", ns, "node", "-e", receiver], { stdio: ["pipe", "pipe", "inherit"] } ); child.on("error", failed); child.on("exit", (code) => failed(new Error(`the receiver in the stack's namespace exited with ${code}`)) ); createInterface({ input: child.stdout }).on("line", (line) => { const said = JSON.parse(line); if (said.port) return ready(said.port); take(said.url, said.method, said.headers, said.raw); }); // unref or the test run waits on the receiver child.unref(); (child.stdout as unknown as { unref(): void }).unref(); (child.stdin as unknown as { unref(): void }).unref(); });
const serveHere = () => new Promise<number>((ready) => { const server = createServer((req, res) => { const chunks: Buffer[] = []; req.on("data", (chunk) => chunks.push(chunk)); req.on("end", () => { const found = take( req.url ?? "", req.method, req.headers, Buffer.concat(chunks).toString() ); if (!found) return void res.writeHead(404).end(); res.end("ok"); }); }); server.unref(); server.listen(0, "127.0.0.1", () => ready((server.address() as AddressInfo).port)); });
const serve = () => { const ns = namespaceOfTheStack(); return (serving ??= ns ? serveInsideTheStack(ns) : serveHere());};
interface Hook { id: string; url: string; owner: Account; secret?: string; inbox: Delivery[]; taken: number;}
const hooks = new WeakMap<Ctx, Hook>();const hookOf = (ctx: Ctx) => { const hook = hooks.get(ctx); if (!hook) throw new Error("this step needs a webhook step before it"); return hook;};
export const hookOn = (events: string[], options: { secret?: string; active?: boolean } = {}) => async (ctx: Ctx) => { const url = `http://127.0.0.1:${await serve()}/${stamp()}`; const inbox: Delivery[] = []; inboxes.set(new URL(url).pathname, inbox); // spindle doesn't know a new repo for a bit, so this retries const id = await until<string>("spindle should take the webhook", 15_000, async () => { const made = await spindlePost(ctx.actor, `${WEBHOOKS}.createWebhook`, { repo: ctx.repo.did, url, events, ...options }); return [ made.status === 200, `it answered ${made.status} ${JSON.stringify(made.body)}`, made.body?.id ]; }); hooks.set(ctx, { id, url, owner: ctx.actor, secret: options.secret, inbox, taken: 0 }); };
export const hookUrl = (ctx: Ctx) => hookOf(ctx).url;export const hookId = (ctx: Ctx) => hookOf(ctx).id;
export const dropsHook = async (ctx: Ctx) => { const hook = hookOf(ctx); await spindlePost(hook.owner, `${WEBHOOKS}.deleteWebhook`, { repo: ctx.repo.did, webhook: hook.id });};
interface Logged { id: string; webhook: string; event: string; url: string; success: boolean; requestBody?: { $bytes: string }; responseCode?: number;}
const logOf = async (ctx: Ctx, hook: Hook) => { const reply = await spindleGet(hook.owner, `${WEBHOOKS}.listWebhookDeliveries`, { repo: ctx.repo.did, webhooks: hook.id, limit: "100" }); expect(reply.status, `spindle answered ${JSON.stringify(reply.body)}`).toBe(200); return reply.body.deliveries as Logged[];};
const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
// deliveries are taken in order, so an extra one fails here as the wrong eventexport const delivers = (event: string, userAgent: Lazy<string>, check: (body: any, ctx: Ctx) => void) => async (ctx: Ctx) => { const hook = hookOf(ctx); await until(`the receiver should get ${event}`, 30_000, async () => [ hook.inbox.length > hook.taken, `it has ${hook.inbox.length} deliveries, ${hook.taken} already checked` ]); const got = hook.inbox[hook.taken++]; const { headers, body } = got;
expect(got.method).toBe("POST"); expect(headers["x-tangled-event"]).toBe(event); expect(headers["content-type"]).toBe("application/json"); expect(headers["user-agent"]).toBe(resolve(userAgent, ctx)); expect(headers["x-tangled-hook-id"]).toBe(String(hook.id)); expect(headers["x-tangled-delivery"]).toMatch(UUID); expect(headers["x-tangled-repo"]).toBe(body.repository.full_name); expect(headers["x-tangled-signature-256"]).toBe( hook.secret ? `sha256=${createHmac("sha256", hook.secret).update(got.raw).digest("hex")}` : undefined ); expect(body.repository.full_name).toMatch(new RegExp(`^${hook.owner.did}/.+`)); expect(body.repository.owner).toEqual({ did: hook.owner.did }); check(body, ctx);
const deliveryId = headers["x-tangled-delivery"]; const logged = await until<Logged>( `spindle's delivery log should list ${deliveryId}`, 15_000, async () => { const entry = (await logOf(ctx, hook)).find((one) => one.id === deliveryId); return [Boolean(entry), `it has no such delivery`, entry]; } ); expect(logged).toMatchObject({ webhook: hook.id, event, url: hook.url, success: true, responseCode: 200 }); expect(Buffer.from(logged.requestBody?.$bytes ?? "", "base64").toString()).toBe(got.raw); };
// a stray duplicate lands within a moment of the first, so sleep 3s firstexport const nothingMore = async (ctx: Ctx) => { const hook = hookOf(ctx); await sleep(3000); expect( hook.inbox.map((one) => one.headers["x-tangled-event"]), "the receiver got more than the flow expected" ).toHaveLength(hook.taken); expect(await logOf(ctx, hook), "spindle logged more than the flow expected").toHaveLength( hook.taken );};