tests/integration/federation.test.ts
Rama main · réplica de lectura del repositorio original.
import { afterAll, beforeAll, test } from "bun:test";
import assert from "node:assert/strict";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createSqliteDb, openDatabase } from "../../src/runtime/bun/sqlite";
import { serve } from "../../src/runtime/bun/server";
import { signPayload } from "../../src/core/webhooks";
import { provisionWith, TEST_SECRET_CIPHER } from "../contract/harness";
/**
* Federation between two independent deployments, over real HTTP.
*
* Each side is its own process-level service with its own database and its own
* credentials, exactly as two regions would be. Titen ships no transport of its
* own: an operator's scheduler pulls from one deployment and pushes into the
* other. This exercises that composition end to end, which is the part the
* contract suite cannot reach because it runs against a single deployment.
*/
const directory = mkdtempSync(join(tmpdir(), "titen-fed-"));
const PEER_SECRET = "federation-shared-secret-value";
interface Node {
server: Awaited<ReturnType<typeof serve>>;
key: string;
orgId: string;
}
let west: Node;
let east: Node;
/** Registered once: a peer is unique per endpoint, as a real pairing is. */
let outbound: string;
let inbound: string;
async function boot(name: string): Promise<Node> {
const dbPath = join(directory, `${name}.db`);
const server = await serve({
dbPath,
port: 0,
hostname: "127.0.0.1",
quiet: true,
revision: name,
maintenanceIntervalMs: 0,
secretCipher: TEST_SECRET_CIPHER,
});
const provisioned = await provisionWith(createSqliteDb(openDatabase(dbPath)), {
scopes: ["*"],
});
return { server, key: provisioned.key, orgId: provisioned.orgId };
}
async function call(
node: Node,
method: string,
path: string,
body?: unknown,
extraHeaders: Record<string, string> = {},
) {
const serialized = body === undefined ? undefined : JSON.stringify(body);
const res = await fetch(`${node.server.url}${path}`, {
method,
headers: {
"content-type": "application/json",
authorization: `Bearer ${node.key}`,
...extraHeaders,
},
body: serialized,
});
return { status: res.status, body: (await res.json()) as any };
}
/** What an operator's scheduler does: sign the batch as the registered peer. */
async function signedPush(node: Node, payload: unknown) {
const serialized = JSON.stringify(payload);
const signature = await signPayload(PEER_SECRET, serialized);
const res = await fetch(`${node.server.url}/v1/federation/push`, {
method: "POST",
headers: {
"content-type": "application/json",
authorization: `Bearer ${node.key}`,
"x-titen-peer-signature": `sha256=${signature}`,
},
body: serialized,
});
return { status: res.status, body: (await res.json()) as any };
}
async function registerPeer(node: Node, other: Node, direction: string) {
const res = await call(node, "POST", "/v1/federation/peers", {
name: `peer-${direction}`,
endpoint: other.server.url,
shared_secret: PEER_SECRET,
direction,
});
assert.equal(res.status, 201, JSON.stringify(res.body));
return res.body.data.peer_id as string;
}
beforeAll(async () => {
west = await boot("west");
east = await boot("east");
outbound = await registerPeer(west, east, "pull");
inbound = await registerPeer(east, west, "push");
});
afterAll(async () => {
await west.server.stop();
await east.server.stop();
rmSync(directory, { recursive: true, force: true });
});
test("the two deployments are genuinely separate services", async () => {
assert.notEqual(west.server.url, east.server.url, "each side listens on its own port");
assert.notEqual(west.orgId, east.orgId, "each side has its own organization");
// A credential from one deployment is meaningless at the other.
const crossed = await fetch(`${east.server.url}/v1/events`, {
headers: { authorization: `Bearer ${west.key}` },
});
assert.equal(crossed.status, 401, "a foreign credential must not authenticate");
});
test("events cross the network from one deployment into the other", async () => {
// Write real memory on the west side.
const observation = await call(west, "POST", "/v1/observations", {
subject_id: "user_fed",
kind: "tool_result",
content: "Regional rollout evidence recorded in the west deployment.",
source: { type: "tool", ref: "fed#1" },
trust: "verified",
});
assert.equal(observation.status, 201);
// East holds nothing yet.
const before = await call(east, "GET", "/v1/events");
assert.equal(before.body.data.events.length, 0, "the east side starts empty");
// The operator pulls from west over HTTP, then pushes into east over HTTP.
const pulled = await call(west, "POST", "/v1/federation/pull", { peer_id: outbound });
assert.equal(pulled.status, 200);
assert.ok(pulled.body.data.events.length >= 1, "west must offer its events");
const pushed = await signedPush(east, {
peer_id: inbound,
events: pulled.body.data.events,
});
assert.equal(pushed.status, 200, JSON.stringify(pushed.body));
const applied = pushed.body.data.results.filter((r: any) => r.status === "success");
assert.ok(
applied.length >= 1,
`expected accepted events, got ${JSON.stringify(pushed.body.data.results)}`,
);
// Remote identity and canonical pointers remain untrusted. The receiving
// principal sees local wrappers, never remote rows with local authority.
const after = await call(east, "GET", "/v1/events");
assert.equal(after.body.data.events.length, applied.length);
const remoteById = new Map(pulled.body.data.events.map((event: any) => [event.id, event]));
for (const wrapper of after.body.data.events) {
assert.equal(wrapper.kind, "federation.received");
assert.equal(wrapper.resource_type, "federated_event");
assert.equal(wrapper.resource_id, wrapper.id);
assert.deepEqual(wrapper.payload.untrusted_remote_event, remoteById.get(wrapper.id));
}
// Both sides recorded the exchange.
const sent = await call(west, "GET", `/v1/federation/log?peer_id=${outbound}`);
assert.ok((sent.body.data.entries ?? []).length >= 1, "the sender logged the exchange");
const got = await call(east, "GET", `/v1/federation/log?peer_id=${inbound}`);
assert.ok((got.body.data.entries ?? []).length >= 1, "the receiver logged the exchange");
assert.ok(
got.body.data.entries.some((entry: any) => entry.status === "success"),
"the receiver must durably record a successful signed exchange",
);
});
test("a replayed batch is preserved as a conflict rather than duplicated", async () => {
await call(west, "POST", "/v1/observations", {
subject_id: "user_fed_replay",
kind: "tool_result",
content: "Evidence that will be delivered twice on purpose.",
source: { type: "tool", ref: "fed#replay" },
trust: "verified",
});
const pulled = await call(west, "POST", "/v1/federation/pull", { peer_id: outbound });
const batch = pulled.body.data.events;
assert.ok(batch.length >= 1);
const first = await signedPush(east, { peer_id: inbound, events: batch });
assert.equal(first.status, 200);
const countAfterFirst = (await call(east, "GET", "/v1/events?limit=200")).body.data.events
.length;
// The same batch again: nothing new, and the collision is reported, not hidden.
const second = await signedPush(east, { peer_id: inbound, events: batch });
assert.equal(second.status, 200);
assert.ok(
second.body.data.results.every((r: any) => r.status === "conflict"),
`a replay must report conflicts, got ${JSON.stringify(second.body.data.results)}`,
);
const countAfterSecond = (await call(east, "GET", "/v1/events?limit=200")).body.data.events
.length;
assert.equal(countAfterSecond, countAfterFirst, "a replay must not duplicate records");
});
test("an unsigned or wrongly signed batch is refused", async () => {
const payload = {
peer_id: inbound,
events: [
{
id: "evt_forged00000000000000000000000",
kind: "observation.appended",
actor_id: "attacker",
resource_type: "observation",
resource_id: "obs_forged00000000000000000000000",
payload: {},
created_at: new Date().toISOString(),
},
],
};
// No signature at all.
const unsigned = await call(east, "POST", "/v1/federation/push", payload);
assert.equal(unsigned.status, 403, "an unsigned batch must be refused");
// A signature computed with the wrong key.
const serialized = JSON.stringify(payload);
const wrong = await signPayload("not-the-shared-secret-value", serialized);
const forged = await fetch(`${east.server.url}/v1/federation/push`, {
method: "POST",
headers: {
"content-type": "application/json",
authorization: `Bearer ${east.key}`,
"x-titen-peer-signature": `sha256=${wrong}`,
},
body: serialized,
});
assert.equal(forged.status, 403, "a batch signed with the wrong key must be refused");
// Nothing was written by either attempt.
const events = await call(east, "GET", "/v1/events?limit=200");
assert.ok(
!events.body.data.events.some((e: any) => e.actor_id === "attacker"),
"a refused batch must write nothing",
);
});
test("a filter keeps a peer's scope narrower than the sender's", async () => {
// Only claim events may leave for this peer.
const filter = await call(west, "POST", `/v1/federation/peers/${outbound}/filters`, {
resource_type: "claim",
});
assert.equal(filter.status, 201);
await call(west, "POST", "/v1/observations", {
subject_id: "user_fed_filter",
kind: "tool_result",
content: "Observation that policy keeps inside the west deployment.",
source: { type: "tool", ref: "fed#filter" },
trust: "verified",
});
const pulled = await call(west, "POST", "/v1/federation/pull", { peer_id: outbound });
assert.equal(pulled.status, 200);
assert.ok(
pulled.body.data.events.every((e: any) => e.resource_type === "claim"),
`a filtered pull must exclude other resource types, got ${JSON.stringify(
pulled.body.data.events.map((e: any) => e.resource_type),
)}`,
);
});