diff --git a/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts b/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts new file mode 100644 index 00000000..8e13acb4 --- /dev/null +++ b/packages/cultnet-ts/src/idunn-odin-presence-publisher.ts @@ -0,0 +1,240 @@ +import dgram from "node:dgram"; +import { decode, encode } from "@msgpack/msgpack"; + +import { encodeCultNetMessageForWire, parseCultNetMessage } from "./contracts"; +import type { IdunnRuntimeSigner } from "./idunn-runtime-authority"; +import { CultNetRudpSession, decodeRudpPacket, encodeRudpPacket } from "./rudp"; +import type { CultNetRudpPacket } from "./rudp"; +import type { RuntimePresenceHealth } from "./runtime-presence-health"; + +const ODIN_CAPABILITY = "odin.verse-rendezvous"; +const MAX_LEASE_WAIT_ROUNDS = 720; + +export type IdunnRuntimePresencePublisher = { + readonly runtimeId: string; + readonly runtimeInstanceId: string; + readonly requiresWriteLease: boolean; + publish(state: RuntimePresenceHealth["state"], detail: string): Promise; + waitForWriteLease(options?: { pollIntervalMs?: number; signal?: AbortSignal }): Promise; + assertWriteLease(): Promise; +}; + +export type IdunnRuntimePresencePublisherOptions = { + signer: IdunnRuntimeSigner; + endpoint: string; +}; + +/** + * Publishes the signer's presence to Odin over RUDP, for services that are + * CultMesh-aware and declare `odin.verse-rendezvous` in Expected. The route + * proof never passes through here: `signer.answerRouteObservation` answers + * Idunn's challenge locally and needs no Odin. Capability shortfalls are not + * judged here either; the correlation owner records a typed disagreement. + */ +export function createIdunnRuntimePresencePublisher( + options: IdunnRuntimePresencePublisherOptions, +): IdunnRuntimePresencePublisher { + const { signer } = options; + const endpoint = parseRudpEndpoint(options.endpoint); + const odinDependency = signer.authority.expected.dependencies.find((dependency) => dependency.kind === "shared-infrastructure" + && dependency.capability === ODIN_CAPABILITY); + if (!odinDependency || normalizeRudpEndpoint(odinDependency.providerEndpoint ?? "") !== normalizeRudpEndpoint(options.endpoint)) { + throw new Error("Configured Odin endpoint does not match Idunn Expected dependency authority."); + } + const connectionId = 0x0d1d0002; + let messageSequence = 0; + + const publish = async (state: RuntimePresenceHealth["state"], detail: string) => { + const signed = signer.sign(state, detail); + messageSequence += 1; + await publishDocument(endpoint, connectionId, { + schemaVersion: "cultnet.document_put_raw.v0", + messageId: `runtime-presence:${signer.authority.expected.target}:${signer.runtimeInstanceId}:${messageSequence}`, + document: signed.document, + }); + if (state === "warming") signer.recordPublishedWarming(signed.canonicalSha256); + return signed.canonicalSha256; + }; + + return { + runtimeId: signer.runtimeId, + runtimeInstanceId: signer.runtimeInstanceId, + requiresWriteLease: signer.requiresWriteLease, + publish, + async waitForWriteLease({ pollIntervalMs = 5000, signal } = {}) { + if (!signer.requiresWriteLease) return ""; + let warmingSequence = 0; + while (!signal?.aborted) { + await publish("warming", "waiting-for-process-write-lease"); + warmingSequence += 1; + try { + return signer.assertWriteLease(); + } catch (error) { + if (warmingSequence > MAX_LEASE_WAIT_ROUNDS) throw new Error("Timed out waiting for Idunn process write lease.", { cause: error }); + } + await abortableDelay(pollIntervalMs, signal); + } + throw signal?.reason ?? new Error("Aborted while waiting for Idunn process write lease."); + }, + async assertWriteLease() { + return signer.assertWriteLease(); + }, + }; +} + +async function publishDocument( + endpoint: { host: string; port: number }, + connectionId: number, + message: Record, +): Promise { + const socket = dgram.createSocket(endpoint.host.includes(":") ? "udp6" : "udp4"); + await new Promise((resolve, reject) => { + socket.once("error", reject); + socket.bind(0, endpoint.host.includes(":") ? "::" : "0.0.0.0", () => { + socket.off("error", reject); + resolve(); + }); + }); + const receiver = receivePackets(socket); + const session = new CultNetRudpSession({ connectionId, initialSequence: 1, resendDelayMs: 100 }); + try { + await sendPacket(socket, endpoint, session.createConnect(Date.now(), new Uint8Array())); + await receiveUntil(receiver, session, endpoint, (packet) => packet.packetType === "accept", 5000); + const wirePayload = encode(encodeCultNetMessageForWire(message as never, "cultnet.schema.v0")); + const packets = session.sendMany("schema", wirePayload, { reliable: true, ordered: true, nowMs: Date.now() }); + for (const packet of packets) await sendPacket(socket, endpoint, packet); + await receiveUntil( + receiver, + session, + endpoint, + () => session.outstandingReliablePacketCount === 0, + 2000, + (frame) => { + if (frame.channelId !== "schema") return; + const response = parseCultNetMessage(decode(frame.payload)); + if (response.schemaVersion === "cultnet.error.v0") { + throw new Error(`Odin rejected runtime presence: ${response.error}`); + } + }, + ); + } finally { + receiver.close(); + socket.close(); + } +} + +type PacketReceiver = { + socket: dgram.Socket; + next(timeoutMs: number): Promise; + close(): void; +}; + +function receivePackets(socket: dgram.Socket): PacketReceiver { + const packets: CultNetRudpPacket[] = []; + const waiters: Array<{ resolve: (packet: CultNetRudpPacket) => void; reject: (error: Error) => void; timer: NodeJS.Timeout }> = []; + const errors: Error[] = []; + const drain = () => { + while (waiters.length && (packets.length || errors.length)) { + const waiter = waiters.shift()!; + clearTimeout(waiter.timer); + if (errors.length) waiter.reject(errors.shift()!); + else waiter.resolve(packets.shift()!); + } + }; + const onMessage = (wire: Buffer) => { + try { packets.push(decodeRudpPacket(wire)); } catch (error) { errors.push(error as Error); } + drain(); + }; + const onError = (error: Error) => { errors.push(error); drain(); }; + socket.on("message", onMessage); + socket.on("error", onError); + return { + socket, + next(timeoutMs) { + if (packets.length) return Promise.resolve(packets.shift()!); + if (errors.length) return Promise.reject(errors.shift()!); + return new Promise((resolve, reject) => { + const waiter = { + resolve, + reject, + timer: setTimeout(() => { + const index = waiters.indexOf(waiter); + if (index >= 0) waiters.splice(index, 1); + const error = new Error("Timed out waiting for CultNet RUDP packet.") as NodeJS.ErrnoException; + error.code = "ETIMEDOUT"; + reject(error); + }, Math.max(1, timeoutMs)), + }; + waiters.push(waiter); + }); + }, + close() { + socket.off("message", onMessage); + socket.off("error", onError); + for (const waiter of waiters.splice(0)) { + clearTimeout(waiter.timer); + waiter.reject(new Error("CultNet RUDP publisher closed.")); + } + }, + }; +} + +async function receiveUntil( + receiver: PacketReceiver, + session: CultNetRudpSession, + endpoint: { host: string; port: number }, + predicate: (packet: CultNetRudpPacket) => boolean, + timeoutMs: number, + onDelivered?: (frame: { channelId: string; payload: Uint8Array; sequence: number }) => void, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + try { + const packet = await receiver.next(Math.min(100, deadline - Date.now())); + const received = session.receive(packet, Date.now()); + if (received.reply) throw new Error("Runtime presence received an unexpected reply-required packet."); + for (const frame of received.delivered) onDelivered?.(frame); + if (predicate(packet)) return; + } catch (error) { + if ((error as NodeJS.ErrnoException)?.code !== "ETIMEDOUT") throw error; + } + for (const packet of session.dueResends(Date.now())) await sendPacket(receiver.socket, endpoint, packet); + } + throw new Error(`Timed out waiting for CultNet RUDP response after ${timeoutMs} ms.`); +} + +async function sendPacket(socket: dgram.Socket, endpoint: { host: string; port: number }, packet: CultNetRudpPacket): Promise { + const wire = encodeRudpPacket(packet); + await new Promise((resolve, reject) => { + socket.send(wire, endpoint.port, endpoint.host, (error) => error ? reject(error) : resolve()); + }); +} + +function parseRudpEndpoint(value: string): { host: string; port: number } { + const text = value.replace(/^rudp:\/\//i, ""); + const match = text.match(/^\[([^\]]+)\]:(\d+)$/) ?? text.match(/^([^:]+):(\d+)$/); + if (!match) throw new Error(`Invalid Odin RUDP endpoint: ${value}`); + const port = Number(match[2]); + if (!Number.isInteger(port) || port < 1 || port > 65535) throw new Error(`Invalid Odin RUDP port: ${match[2]}`); + return { host: match[1]!, port }; +} + +function normalizeRudpEndpoint(value: string): string { + const normalized = value.trim().toLowerCase().replace(/^rudp:\/\//, ""); + const parsed = parseRudpEndpoint(normalized); + return `${parsed.host.toLowerCase()}:${parsed.port}`; +} + +function abortableDelay(milliseconds: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + signal?.removeEventListener("abort", onAbort); + resolve(); + }, milliseconds); + const onAbort = () => { + clearTimeout(timer); + reject(signal?.reason ?? new Error("Aborted.")); + }; + signal?.addEventListener("abort", onAbort, { once: true }); + }); +} diff --git a/packages/cultnet-ts/src/idunn-runtime-authority.ts b/packages/cultnet-ts/src/idunn-runtime-authority.ts index 5ea30fff..47e132b3 100644 --- a/packages/cultnet-ts/src/idunn-runtime-authority.ts +++ b/packages/cultnet-ts/src/idunn-runtime-authority.ts @@ -15,10 +15,7 @@ import { type RuntimePresenceHealth, } from "./runtime-presence-health"; import { truncateUtf8Bytes } from "./utf8-bound"; -import { encodeCultNetMessageForWire, parseCultNetMessage, type CultNetRawDocumentRecord, type CultNetSnapshotRequestMessage } from "./contracts"; -import { CultNetRudpSession, decodeRudpPacket, encodeRudpPacket } from "./rudp"; -import type { CultNetRudpPacket } from "./rudp"; -import dgram from "node:dgram"; +import type { CultNetRawDocumentRecord, CultNetSnapshotRequestMessage } from "./contracts"; export const IDUNN_EXPECTED_INCARNATION_SCHEMA = "idunn.expected_incarnation.v2"; export const IDUNN_RUNTIME_ACTIVATION_SCHEMA = "idunn.runtime_activation.v2"; @@ -77,33 +74,50 @@ export type IdunnRuntimeAuthority = { processWriteLeasePath?: string; }; -export type IdunnRuntimePresencePublisher = { +export type SignedRuntimePresence = { + document: CultNetRawDocumentRecord; + canonicalSha256: string; +}; + +/** + * Signs this launch's runtime presence and answers Idunn's route challenge. It + * owns no network: Odin publication, if a service wants it, is a separate + * consumer of `sign`. + */ +export type IdunnRuntimeSigner = { + readonly authority: IdunnRuntimeAuthority; readonly runtimeId: string; readonly runtimeInstanceId: string; readonly requiresWriteLease: boolean; - latestPresenceDocument(): CultNetRawDocumentRecord | null; - publishRouteObservation(request: CultNetSnapshotRequestMessage): Promise; - publish(state: RuntimePresenceHealth["state"], detail: string): Promise; - waitForWriteLease(options?: { pollIntervalMs?: number; signal?: AbortSignal }): Promise; - assertWriteLease(): Promise; + /** Signs one presence document. Every call takes the next publisher sequence. */ + sign(state: RuntimePresenceHealth["state"], detail: string): SignedRuntimePresence; + /** + * Answers an Idunn route challenge from local authority alone: signs a fresh + * presence in the given state, with no network. The caller supplies the + * app's current health (`warming` until first healthy). + */ + answerRouteObservation(request: CultNetSnapshotRequestMessage, state: RuntimePresenceHealth["state"]): CultNetRawDocumentRecord; + /** Reads Idunn's write lease and checks it names this incarnation and a warming proof this process published. */ + assertWriteLease(): string; + /** Marks a warming presence as accepted by its consumer, making it a valid lease target. */ + recordPublishedWarming(canonicalSha256: string): void; }; -export type IdunnRuntimePresencePublisherOptions = { +export type IdunnRuntimeSignerOptions = { authority: IdunnRuntimeAuthority; - endpoint: string; - healthContract: string; capabilities: RuntimePresenceHealth["capabilities"]; }; /** * Loads the root-provisioned Idunn bundle and parent-only signing descriptors. * The expected record and activation witness are accepted only when their - * exact positional contracts and mutual incarnation binding agree. + * exact positional contracts and mutual incarnation binding agree. Nothing + * here reads Odin: whether a target talks to Odin is decided by the + * publisher a service chooses to create. */ export function loadIdunnRuntimeAuthorityFromEnvironment( target: string, healthContract: string, - odinEndpoint: string, ): IdunnRuntimeAuthority { const bundle = requiredEnvironment(IDUNN_RUNTIME_BUNDLE_ENVIRONMENT); const expected = readAuthorityRecord(path.join(bundle, "expected.cc"), "idunn.expected_incarnation", IDUNN_EXPECTED_INCARNATION_SCHEMA); @@ -113,11 +127,6 @@ export function loadIdunnRuntimeAuthorityFromEnvironment( if (expectedValue.target !== target || expectedValue.healthContract !== healthContract) { throw new Error("Idunn Expected target or health contract does not match this runtime."); } - const odinDependency = expectedValue.dependencies.find((dependency) => dependency.kind === "shared-infrastructure" - && dependency.capability === "odin.verse-rendezvous"); - if (!odinDependency || normalizeRudpEndpoint(odinDependency.providerEndpoint ?? "") !== normalizeRudpEndpoint(odinEndpoint)) { - throw new Error("Configured Odin endpoint does not match Idunn Expected dependency authority."); - } if (activationValue.expectedProjectionSha256 !== expectedValue.canonicalSha256 || activationValue.runtimeId !== expectedValue.runtimeId) { throw new Error("Idunn Activation does not bind this exact Expected incarnation."); @@ -167,6 +176,7 @@ export function loadIdunnRuntimeAuthorityFromEnvironment( return authority; } + /** * Signs the fixed Idunn v2 positional runtime-presence contract with the * provider identity and this launch's Idunn-issued activation key. @@ -205,42 +215,26 @@ export function signIdunnRuntimePresence( }; } -export function createIdunnRuntimePresencePublisher( - options: IdunnRuntimePresencePublisherOptions, -): IdunnRuntimePresencePublisher { - const endpoint = parseRudpEndpoint(options.endpoint); - const expected = options.authority.expected; - if (options.healthContract !== expected.healthContract) throw new Error("Runtime presence contract differs from Expected."); - if (options.authority.boundEndpoint !== expected.route?.candidateEndpoint) throw new Error("Runtime presence bind differs from Expected."); - for (const required of expected.capabilities) { - if (!options.capabilities.some((actual) => actual.capability === required.capability - && actual.schema === required.schema - && actual.compatibility === required.compatibility - && actual.capacity >= required.minimumCapacity)) { - throw new Error(`Runtime does not provide Expected capability ${required.capability}/${required.schema}.`); - } - } - const connectionId = 0x0d1d0002; +export function createIdunnRuntimeSigner(options: IdunnRuntimeSignerOptions): IdunnRuntimeSigner { + const { authority } = options; + const expected = authority.expected; let sequence = 0; - let latestLeaseSha256: string | null = null; - let latestPresenceDocument: CultNetRawDocumentRecord | null = null; const recentWarmingProofs: string[] = []; - const assertWriteLease = async () => { + const assertWriteLease = () => { if (!expected.writeLeaseRequired) return ""; - const leasePath = options.authority.processWriteLeasePath; + const leasePath = authority.processWriteLeasePath; if (!leasePath) throw new Error("Idunn process write lease is required but no lease path is configured."); const record = readAuthorityRecord(leasePath, "idunn.process_write_lease", IDUNN_PROCESS_WRITE_LEASE_SCHEMA); const lease = decodeArray(record.payload, 14, "Idunn process write lease"); - assertLeaseMatches(lease, options.authority, recentWarmingProofs); - latestLeaseSha256 = prefixedSha256(Buffer.from(encode(lease))); - return latestLeaseSha256; + assertLeaseMatches(lease, authority, recentWarmingProofs); + return prefixedSha256(Buffer.from(encode(lease))); }; - const publish = async (state: RuntimePresenceHealth["state"], detail: string) => { - const writeLeaseSha256 = state === "warming" || !expected.writeLeaseRequired ? null : await assertWriteLease(); + const sign = (state: RuntimePresenceHealth["state"], detail: string): SignedRuntimePresence => { + const writeLeaseSha256 = state === "warming" || !expected.writeLeaseRequired ? null : assertWriteLease(); sequence += 1; - const signed = signIdunnRuntimePresence(options.authority, { + const signed = signIdunnRuntimePresence(authority, { capabilities: options.capabilities, state, detail: truncateUtf8Bytes(detail, RUNTIME_PRESENCE_DETAIL_MAX_BYTES), @@ -248,44 +242,28 @@ export function createIdunnRuntimePresencePublisher( publisherSequence: sequence, observedAtUnixMillis: Date.now(), }); - const document: CultNetRawDocumentRecord = { - schemaId: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, - recordKey: expected.target, - storedAt: new Date().toISOString(), - payloadEncoding: "messagepack", - payload: new Uint8Array(signed.payload), - sourceRuntimeId: expected.runtimeId, - sourceRole: "runtime-presence-health-publisher", - tags: ["cultnet.transport.rudp.v0", "runtime-presence"], - }; - await publishDocument(endpoint, connectionId, { - schemaVersion: "cultnet.document_put_raw.v0", - messageId: `runtime-presence:${expected.target}:${options.authority.activation.runtimeInstanceId}:${sequence}`, - document, - }); - latestPresenceDocument = document; - if (state === "warming") { - recentWarmingProofs.push(signed.canonicalSha256); - if (recentWarmingProofs.length > MAX_RECENT_WARMING_PROOFS) recentWarmingProofs.shift(); - } - return signed.canonicalSha256; - }; - - const copyLatestPresenceDocument = (): CultNetRawDocumentRecord | null => { - if (!latestPresenceDocument) return null; return { - ...latestPresenceDocument, - payload: new Uint8Array(latestPresenceDocument.payload), - ...(latestPresenceDocument.tags ? { tags: [...latestPresenceDocument.tags] } : {}), + canonicalSha256: signed.canonicalSha256, + document: { + schemaId: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, + recordKey: expected.target, + storedAt: new Date().toISOString(), + payloadEncoding: "messagepack", + payload: new Uint8Array(signed.payload), + sourceRuntimeId: expected.runtimeId, + sourceRole: "runtime-presence-health-publisher", + tags: ["cultnet.transport.rudp.v0", "runtime-presence"], + }, }; }; return { + authority, runtimeId: expected.runtimeId, - runtimeInstanceId: options.authority.activation.runtimeInstanceId, + runtimeInstanceId: authority.activation.runtimeInstanceId, requiresWriteLease: expected.writeLeaseRequired, - latestPresenceDocument: copyLatestPresenceDocument, - async publishRouteObservation(request) { + sign, + answerRouteObservation(request, state) { if (!request.messageId || request.messageId.trim() !== request.messageId) throw new Error("Idunn route observation requires a message id."); if ((request.schemaIds && !request.schemaIds.includes(GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA)) || (request.recordKeys && !request.recordKeys.includes(expected.target)) @@ -293,159 +271,16 @@ export function createIdunnRuntimePresencePublisher( || request.shardEpoch !== undefined) { throw new Error("Idunn route observation does not request this runtime presence document."); } - await publish("active", `route-observation:${request.messageId}`); - const document = copyLatestPresenceDocument(); - if (!document) throw new Error("Idunn route observation was published without a presence document."); - return document; - }, - publish, - async waitForWriteLease({ pollIntervalMs = 5000, signal } = {}) { - if (!expected.writeLeaseRequired) return ""; - let warmingSequence = 0; - while (!signal?.aborted) { - await publish("warming", "waiting-for-process-write-lease"); - warmingSequence += 1; - try { - return await assertWriteLease(); - } catch (error) { - if (warmingSequence > 720) throw new Error("Timed out waiting for Idunn process write lease.", { cause: error }); - } - await abortableDelay(pollIntervalMs, signal); - } - throw signal?.reason ?? new Error("Aborted while waiting for Idunn process write lease."); + return sign(state, `route-observation:${request.messageId}`).document; }, assertWriteLease, - }; -} - -async function publishDocument( - endpoint: { host: string; port: number }, - connectionId: number, - message: Record, -): Promise { - const socket = dgram.createSocket(endpoint.host.includes(":") ? "udp6" : "udp4"); - await new Promise((resolve, reject) => { - socket.once("error", reject); - socket.bind(0, endpoint.host.includes(":") ? "::" : "0.0.0.0", () => { - socket.off("error", reject); - resolve(); - }); - }); - const receiver = receivePackets(socket); - const session = new CultNetRudpSession({ connectionId, initialSequence: 1, resendDelayMs: 100 }); - try { - await sendPacket(socket, endpoint, session.createConnect(Date.now(), new Uint8Array())); - await receiveUntil(receiver, session, endpoint, (packet) => packet.packetType === "accept", 5000); - const wirePayload = encode(encodeCultNetMessageForWire(message as never, "cultnet.schema.v0")); - const packets = session.sendMany("schema", wirePayload, { reliable: true, ordered: true, nowMs: Date.now() }); - for (const packet of packets) await sendPacket(socket, endpoint, packet); - await receiveUntil( - receiver, - session, - endpoint, - () => session.outstandingReliablePacketCount === 0, - 2000, - (frame) => { - if (frame.channelId !== "schema") return; - const response = parseCultNetMessage(decode(frame.payload)); - if (response.schemaVersion === "cultnet.error.v0") { - throw new Error(`Odin rejected runtime presence: ${response.error}`); - } - }, - ); - } finally { - receiver.close(); - socket.close(); - } -} - -type PacketReceiver = { - socket: dgram.Socket; - next(timeoutMs: number): Promise; - close(): void; -}; - -function receivePackets(socket: dgram.Socket): PacketReceiver { - const packets: CultNetRudpPacket[] = []; - const waiters: Array<{ resolve: (packet: CultNetRudpPacket) => void; reject: (error: Error) => void; timer: NodeJS.Timeout }> = []; - const errors: Error[] = []; - const drain = () => { - while (waiters.length && (packets.length || errors.length)) { - const waiter = waiters.shift()!; - clearTimeout(waiter.timer); - if (errors.length) waiter.reject(errors.shift()!); - else waiter.resolve(packets.shift()!); - } - }; - const onMessage = (wire: Buffer) => { - try { packets.push(decodeRudpPacket(wire)); } catch (error) { errors.push(error as Error); } - drain(); - }; - const onError = (error: Error) => { errors.push(error); drain(); }; - socket.on("message", onMessage); - socket.on("error", onError); - return { - socket, - next(timeoutMs) { - if (packets.length) return Promise.resolve(packets.shift()!); - if (errors.length) return Promise.reject(errors.shift()!); - return new Promise((resolve, reject) => { - const waiter = { - resolve, - reject, - timer: setTimeout(() => { - const index = waiters.indexOf(waiter); - if (index >= 0) waiters.splice(index, 1); - const error = new Error("Timed out waiting for CultNet RUDP packet.") as NodeJS.ErrnoException; - error.code = "ETIMEDOUT"; - reject(error); - }, Math.max(1, timeoutMs)), - }; - waiters.push(waiter); - }); - }, - close() { - socket.off("message", onMessage); - socket.off("error", onError); - for (const waiter of waiters.splice(0)) { - clearTimeout(waiter.timer); - waiter.reject(new Error("CultNet RUDP publisher closed.")); - } + recordPublishedWarming(canonicalSha256) { + recentWarmingProofs.push(canonicalSha256); + if (recentWarmingProofs.length > MAX_RECENT_WARMING_PROOFS) recentWarmingProofs.shift(); }, }; } -async function receiveUntil( - receiver: PacketReceiver, - session: CultNetRudpSession, - endpoint: { host: string; port: number }, - predicate: (packet: CultNetRudpPacket) => boolean, - timeoutMs: number, - onDelivered?: (frame: { channelId: string; payload: Uint8Array; sequence: number }) => void, -): Promise { - const deadline = Date.now() + timeoutMs; - while (Date.now() < deadline) { - try { - const packet = await receiver.next(Math.min(100, deadline - Date.now())); - const received = session.receive(packet, Date.now()); - if (received.reply) throw new Error("Runtime presence received an unexpected reply-required packet."); - for (const frame of received.delivered) onDelivered?.(frame); - if (predicate(packet)) return; - } catch (error) { - if ((error as NodeJS.ErrnoException)?.code !== "ETIMEDOUT") throw error; - } - for (const packet of session.dueResends(Date.now())) await sendPacket(receiver.socket, endpoint, packet); - } - throw new Error(`Timed out waiting for CultNet RUDP response after ${timeoutMs} ms.`); -} - -async function sendPacket(socket: dgram.Socket, endpoint: { host: string; port: number }, packet: CultNetRudpPacket): Promise { - const wire = encodeRudpPacket(packet); - await new Promise((resolve, reject) => { - socket.send(wire, endpoint.port, endpoint.host, (error) => error ? reject(error) : resolve()); - }); -} - function readAuthorityRecord(filePath: string, type: string, schemaId: string): { payload: Uint8Array } { const decoded = decode(fs.readFileSync(filePath)); if (!Array.isArray(decoded) || decoded[0] !== CULTCACHE_STORE_FORMAT || !Array.isArray(decoded[1]) || !Array.isArray(decoded[2])) { @@ -495,9 +330,7 @@ function decodeExpected(payload: Uint8Array): ExpectedIncarnation { if (!/^sha256-[0-9a-f]{64}$/u.test(hash)) throw new Error(`Expected ${name} hash is malformed.`); } if (!/^[0-9a-f]{40}$/u.test(sourceRevision)) throw new Error("Expected source revision is malformed."); - if (!route || route.transport !== "http" || !route.candidateEndpoint.startsWith("http://")) { - throw new Error("Idunn Expected must include an HTTP candidate route."); - } + if (route) validateRoute(route); return { schemaVersion: string(values[0], "Expected schema"), target, @@ -575,6 +408,24 @@ function decodeExpectedDependencies(value: unknown): ExpectedIncarnation["depend }); } +// The Rust owner (IdunnExpectedRoute::validate) accepts http, tcp and rudp +// routes; a runtime is not entitled to a narrower admission than Idunn's. +const ROUTE_SCHEMES: Record = { + http: { stable: ["http://", "https://"], candidate: ["http://"] }, + tcp: { stable: ["tcp://"], candidate: ["tcp://"] }, + rudp: { stable: ["rudp://"], candidate: ["rudp://"] }, +}; + +function validateRoute(route: NonNullable): void { + const schemes = ROUTE_SCHEMES[route.transport]; + if (!schemes) throw new Error("Idunn Expected route transport is unsupported."); + if (!schemes.stable.some((scheme) => route.stableEndpoint.startsWith(scheme)) + || !schemes.candidate.some((scheme) => route.candidateEndpoint.startsWith(scheme))) { + throw new Error(`Idunn Expected ${route.transport} route endpoints use an unsupported scheme.`); + } + if (route.stableEndpoint === route.candidateEndpoint) throw new Error("Idunn Expected stable and candidate route endpoints are identical."); +} + function decodeRoute(value: unknown): ExpectedIncarnation["route"] { if (value === null) return null; const route = array(value, 4, "Expected route"); @@ -595,7 +446,7 @@ function openProviderHealthIdentity(credentialBytes: Buffer): crypto.KeyObject { if (enrollmentNonce.length !== 32) throw new Error("Provider-health enrollment nonce has an invalid length."); const publicKey = Buffer.from(providerIdentityByteVector(publicKeyValue, "Provider-health public key")); const protectedSeed = Buffer.from(providerIdentityByteVector(protectedSeedValue, "Provider-health protected seed")); - const machineId = fs.readFileSync("/etc/machine-id", "utf8").trim(); + const machineId = readMachineId(); if (!machineId) throw new Error("Linux machine-id is unavailable for provider-health identity."); const expectedBinding = `${PROVIDER_PROTECTOR_CONTEXT}:machine-id-sha256:${sha256Hex(Buffer.from(machineId))}`; if (protectorKind !== "linux_file_mode_machine_id_binding" || binding !== expectedBinding || assurance !== "os_installation_file_bound_cloneable_baseline") { @@ -616,6 +467,15 @@ function openProviderHealthIdentity(credentialBytes: Buffer): crypto.KeyObject { return key; } +// Same sources, same order as the Rust identity protector. +function readMachineId(): string { + try { + return fs.readFileSync("/etc/machine-id", "utf8").trim(); + } catch { + return fs.readFileSync("/var/lib/dbus/machine-id", "utf8").trim(); + } +} + function readProviderHealthIdentityCredential(bytes: Buffer): Uint8Array { const decoded = decode(bytes); if (!Array.isArray(decoded) || decoded.length !== 1) { @@ -693,40 +553,12 @@ function assertLeaseMatches(values: unknown[], authority: IdunnRuntimeAuthority, if (!matches) throw new Error("Idunn process write lease does not match the current Expected incarnation."); } -function parseRudpEndpoint(value: string): { host: string; port: number } { - const text = value.replace(/^rudp:\/\//i, ""); - const match = text.match(/^\[([^\]]+)\]:(\d+)$/) ?? text.match(/^([^:]+):(\d+)$/); - if (!match) throw new Error(`Invalid Odin RUDP endpoint: ${value}`); - const port = Number(match[2]); - if (!Number.isInteger(port) || port < 1 || port > 65535) throw new Error(`Invalid Odin RUDP port: ${match[2]}`); - return { host: match[1]!, port }; -} - -function normalizeRudpEndpoint(value: string): string { - const normalized = value.trim().toLowerCase().replace(/^rudp:\/\//, ""); - const parsed = parseRudpEndpoint(normalized); - return `${parsed.host.toLowerCase()}:${parsed.port}`; -} - function requiredEnvironment(name: string): string { const value = process.env[name]?.trim(); if (!value) throw new Error(`${name} is required for Idunn runtime presence.`); return value; } -function abortableDelay(milliseconds: number, signal?: AbortSignal): Promise { - return new Promise((resolve, reject) => { - const timer = setTimeout(() => { - signal?.removeEventListener("abort", onAbort); - resolve(); - }, milliseconds); - const onAbort = () => { - clearTimeout(timer); - reject(signal?.reason ?? new Error("Aborted.")); - }; - signal?.addEventListener("abort", onAbort, { once: true }); - }); -} function array(value: unknown, length: number, label: string): unknown[] { if (!Array.isArray(value) || value.length !== length) throw new Error(`${label} is not a ${length}-field positional contract.`); diff --git a/packages/cultnet-ts/src/index.ts b/packages/cultnet-ts/src/index.ts index 3db890ef..8950f638 100644 --- a/packages/cultnet-ts/src/index.ts +++ b/packages/cultnet-ts/src/index.ts @@ -205,9 +205,15 @@ export { IDUNN_RUNTIME_BUNDLE_ENVIRONMENT, IDUNN_RUNTIME_CANDIDATE_BIND_ENVIRONMENT, IDUNN_RUNTIME_ACTIVATION_SCHEMA, - createIdunnRuntimePresencePublisher, + createIdunnRuntimeSigner, loadIdunnRuntimeAuthorityFromEnvironment, type IdunnRuntimeAuthority, + type IdunnRuntimeSigner, + type IdunnRuntimeSignerOptions, + type SignedRuntimePresence, +} from "./idunn-runtime-authority"; +export { + createIdunnRuntimePresencePublisher, type IdunnRuntimePresencePublisher, type IdunnRuntimePresencePublisherOptions, -} from "./idunn-runtime-authority"; +} from "./idunn-odin-presence-publisher"; diff --git a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts index 7969fde2..1639b9aa 100644 --- a/packages/cultnet-ts/test/idunn-runtime-authority.test.ts +++ b/packages/cultnet-ts/test/idunn-runtime-authority.test.ts @@ -1,30 +1,92 @@ -import { mock, test } from "node:test"; +import { test, type TestContext } from "node:test"; import assert from "node:assert/strict"; -import crypto from "node:crypto"; import dgram from "node:dgram"; import fs from "node:fs"; -import os from "node:os"; import path from "node:path"; -import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { decode, encode } from "@msgpack/msgpack"; import { createIdunnRuntimePresencePublisher, + createIdunnRuntimeSigner, loadIdunnRuntimeAuthorityFromEnvironment, - signIdunnRuntimePresence, - systemdListenPidMatches, -} from "../src/idunn-runtime-authority"; + type IdunnRuntimeAuthority, +} from "../src"; +import { systemdListenPidMatches } from "../src/idunn-runtime-authority"; +import type { CultNetSnapshotRequestMessage } from "../src/contracts"; import { CultNetRudpSession, decodeRudpPacket, encodeRudpPacket } from "../src/rudp"; -import { - runtimePresenceActivationSigningMessage, - runtimePresenceProofPayload, - runtimePresenceProviderSigningMessage, - RUNTIME_PRESENCE_SLOT, -} from "../src/runtime-presence-health"; +import { RUNTIME_PRESENCE_SLOT } from "../src/runtime-presence-health"; -const providerIdentityContext = "gamecult-provider-health-identity-v1"; -const providerIdDomain = Buffer.from("gamecult.provider-health.identity.v1\0", "utf8"); -const activationIdDomain = Buffer.from("idunn.runtime-activation.id.v1\0", "utf8"); +// Every input below was written by the Rust owner; see fixtures/idunn-runtime/README.md. +// The only values this file pins are the ones the generator also pins. +const FIXTURES = path.join(__dirname, "..", "..", "test", "fixtures", "idunn-runtime"); +const OBSERVED_AT = 1_790_000_000_000; +const CHALLENGE: CultNetSnapshotRequestMessage = { + schemaVersion: "cultnet.snapshot_request.v0", + messageId: "route-challenge-1", +}; + +type FixtureSet = "web" | "service" | "rudp-route"; + +function fixtureFile(set: FixtureSet, name: string): Buffer { + return fs.readFileSync(path.join(FIXTURES, set, name)); +} + +/** The Expected fields the tests need, read from the Rust-written file rather than re-spelled. */ +function expectedFacts(set: FixtureSet) { + const store = decode(fixtureFile(set, "expected.cc")) as unknown[][][]; + const values = decode(store[2]![0]![3] as Uint8Array) as unknown[]; + const route = values[15] as string[]; + const capabilities = (values[16] as Array<[string, string, string, number]>).map(([capability, schema, compatibility, minimum]) => ({ + capability, schema, compatibility, capacity: minimum, + })); + const dependencies = values[17] as unknown[][]; + return { + target: values[1] as string, + healthContract: values[10] as string, + candidate: route[3]!, + capabilities, + odinEndpoint: dependencies.map((dependency) => dependency[9] as string)[0], + }; +} + +/** Points the process at one Rust-written bundle and its inherited descriptors, as Idunn's systemd unit would. */ +function openAuthority(context: TestContext, set: FixtureSet, options: { machineId?: "etc" | "dbus" } = {}): IdunnRuntimeAuthority { + const facts = expectedFacts(set); + const originalEnvironment = { ...process.env }; + context.after(() => { + for (const key of Object.keys(process.env)) if (!(key in originalEnvironment)) delete process.env[key]; + Object.assign(process.env, originalEnvironment); + }); + process.env.GAMECULT_IDUNN_RUNTIME_BUNDLE = path.join(FIXTURES, set); + process.env.GAMECULT_IDUNN_CANDIDATE_BIND = facts.candidate; + delete process.env.GAMECULT_IDUNN_PROCESS_WRITE_LEASE; + process.env.LISTEN_PID = String(process.pid); + process.env.LISTEN_FDS = "2"; + process.env.LISTEN_FDNAMES = "gamecult-idunn-runtime-activation-key:gamecult-runtime-presence-identity"; + const machineId = fs.readFileSync(path.join(FIXTURES, "machine-id"), "utf8"); + const realReadFileSync = fs.readFileSync.bind(fs); + context.mock.method(fs, "readFileSync", (filePath: fs.PathOrFileDescriptor, ...args: unknown[]) => { + if (filePath === 3) return fixtureFile(set, "activation-key.seed"); + if (filePath === 4) return fixtureFile(set, "provider-identity.credential"); + if (filePath === "/etc/machine-id") { + if (options.machineId === "dbus") throw Object.assign(new Error("ENOENT"), { code: "ENOENT" }); + return machineId; + } + if (filePath === "/var/lib/dbus/machine-id") return machineId; + return realReadFileSync(filePath as never, ...args as never[]); + }); + return loadIdunnRuntimeAuthorityFromEnvironment(facts.target, facts.healthContract); +} + +function signerFor(context: TestContext, set: FixtureSet, capacityDelta = 0) { + const authority = openAuthority(context, set); + const capabilities = expectedFacts(set).capabilities.map((capability) => ({ ...capability, capacity: capability.capacity + capacityDelta })); + return createIdunnRuntimeSigner({ authority, capabilities }); +} + +function freezeClock(context: TestContext): void { + context.mock.method(Date, "now", () => OBSERVED_AT); +} test("matches systemd's host PID for the private PID namespace init only", () => { assert.equal(systemdListenPidMatches(String(process.pid), process.pid), true); @@ -34,216 +96,160 @@ test("matches systemd's host PID for the private PID namespace init only", () => assert.equal(systemdListenPidMatches(undefined, 1), false); }); -test("opens Idunn identity and publishes valid CultNet data while surfacing piggybacked application errors", async (context) => { - const directory = mkdtempSync(path.join(os.tmpdir(), "cultnet-runtime-authority-")); - context.after(() => rmSync(directory, { recursive: true, force: true })); +for (const set of ["web", "service", "rudp-route"] as const) { + test(`opens the Rust-written ${set} bundle with no Odin endpoint, whatever route transport Rust admits`, (context) => { + const authority = openAuthority(context, set); + assert.equal(authority.expected.target, expectedFacts(set).target); + assert.equal(authority.boundEndpoint, expectedFacts(set).candidate); + }); +} - const machineId = "1234567890abcdef1234567890abcdef"; - const providerSeed = Buffer.alloc(32, 0x31); - const activationSeed = Buffer.alloc(32, 0x72); - const providerKey = privateKeyFromSeed(providerSeed); - const activationKey = privateKeyFromSeed(activationSeed); - const providerPublic = rawPublic(crypto.createPublicKey(providerKey)); - const activationPublic = rawPublic(crypto.createPublicKey(activationKey)); - const providerIdentity = sha256Hex(Buffer.concat([providerIdDomain, providerPublic])); - const activationIdentity = sha256Hex(Buffer.concat([activationIdDomain, activationPublic])); - const target = "streampixels-service"; - const contract = "streampixels-service.cultnet-rudp-service-health"; - const candidate = "http://127.0.0.1:18831"; - const expectedValues = [ - "idunn.expected_incarnation.v2", target, `sha256-${"1".repeat(64)}`, "incarnation-1", - `sha256-${"2".repeat(64)}`, "https://github.com/GameCult/StreamPixels.git", "a".repeat(40), - `sha256-${"3".repeat(64)}`, "streampixels-service-yggdrasil", providerIdentity, contract, - `sha256-${"4".repeat(64)}`, "service-boundary-v1", `sha256-${"5".repeat(64)}`, false, - ["streampixels-service-http", "http", "http://127.0.0.1:8831", candidate], - [["streampixels.service.api", "streampixels.api.v1", "v1", 1]], - [["shared-infrastructure", "odin.verse-rendezvous", "odin.verse-topology.v1", "v1", 1, "before-promotion", null, null, null, "rudp://127.0.0.1:17871"]], +test("a target that declares no Odin dependency opens and answers Idunn's challenge without one", (context) => { + assert.equal(expectedFacts("web").odinEndpoint, undefined); + const signer = signerFor(context, "web"); + assert.ok(signer.answerRouteObservation(CHALLENGE, "warming").payload.byteLength > 0); +}); + +test("finds the machine-id at the dbus path when /etc/machine-id is absent, as the Rust protector does", (context) => { + assert.ok(openAuthority(context, "web", { machineId: "dbus" })); +}); + +test("rejects a credential file holding more than one envelope", (context) => { + openAuthority(context, "web"); + const credential = decode(fixtureFile("web", "provider-identity.credential")) as unknown[][]; + const doubled = Buffer.from(encode([...credential, ...credential])); + const realReadFileSync = fs.readFileSync; + context.mock.method(fs, "readFileSync", (filePath: fs.PathOrFileDescriptor, ...args: unknown[]) => + filePath === 4 ? doubled : (realReadFileSync as (...all: unknown[]) => unknown)(filePath, ...args)); + assert.throws( + () => loadIdunnRuntimeAuthorityFromEnvironment(expectedFacts("web").target, expectedFacts("web").healthContract), + /exactly one envelope/, + ); +}); + +test("answers a route challenge byte for byte as the Rust-verified vectors, in the state the app reports", (context) => { + freezeClock(context); + const signer = signerFor(context, "web"); + const warming = signer.answerRouteObservation(CHALLENGE, "warming"); + assert.deepEqual(Buffer.from(warming.payload), fixtureFile("web", "presence-warming.bin")); + assert.equal((decode(warming.payload) as unknown[])[RUNTIME_PRESENCE_SLOT.state], "warming"); + const second = signer.answerRouteObservation(CHALLENGE, "active"); + assert.equal((decode(second.payload) as unknown[])[RUNTIME_PRESENCE_SLOT.publisherSequence], 2); + // Sequence 2 cannot match a vector written at sequence 1; sign the same fields from a fresh signer. + const fresh = signerFor(context, "web"); + assert.deepEqual(Buffer.from(fresh.answerRouteObservation(CHALLENGE, "active").payload), fixtureFile("web", "presence-active.bin")); +}); + +test("a service with an Odin dependency signs the Rust-verified vectors, including a shortfall Rust correlates to a capacity disagreement", (context) => { + freezeClock(context); + assert.deepEqual( + Buffer.from(signerFor(context, "service").answerRouteObservation(CHALLENGE, "active").payload), + fixtureFile("service", "presence-active.bin"), + ); + assert.deepEqual( + Buffer.from(signerFor(context, "service", -1).answerRouteObservation(CHALLENGE, "active").payload), + fixtureFile("service", "presence-active-below-minimum.bin"), + ); +}); + +test("answers locally: synchronously, with no socket opened", (context) => { + const sockets = context.mock.method(dgram, "createSocket"); + const signer = signerFor(context, "web"); + const answer = signer.answerRouteObservation(CHALLENGE, "warming"); + assert.equal(typeof (answer as unknown as { then?: unknown }).then, "undefined"); + assert.equal(sockets.mock.callCount(), 0); +}); + +test("refuses a challenge that does not ask for this runtime's presence", (context) => { + const signer = signerFor(context, "web"); + const cases: CultNetSnapshotRequestMessage[] = [ + { ...CHALLENGE, messageId: "" }, + { ...CHALLENGE, messageId: " padded" }, + { ...CHALLENGE, recordKeys: ["another-target"] }, + { ...CHALLENGE, schemaIds: ["another.schema"] }, + { ...CHALLENGE, shardId: "shard-1" }, ]; - const expectedBytes = Buffer.from(encode(expectedValues)); - const expectedSha = `sha256-${sha256Hex(expectedBytes)}`; - const activationValues = [ - "idunn.runtime_activation.v2", expectedSha, "streampixels-service-yggdrasil", - `sha256-${"6".repeat(64)}`, activationIdentity, activationPublic, Date.now(), - "7".repeat(64), "ed25519", new Uint8Array(64).fill(0x17), - ]; - const bundle = path.join(directory, "bundle"); - fs.mkdirSync(bundle); - writeFileSync(path.join(bundle, "expected.cc"), cultCacheStore( - "idunn.expected_incarnation", "idunn.expected_incarnation.v2", "expected", expectedBytes, - )); - writeFileSync(path.join(bundle, "activation.cc"), cultCacheStore( - "idunn.runtime_activation", "idunn.runtime_activation.v2", "activation", Buffer.from(encode(activationValues)), - )); - - const binding = `${providerIdentityContext}:machine-id-sha256:${sha256Hex(Buffer.from(machineId))}`; - const mask = crypto.createHash("sha256").update(Buffer.concat([ - Buffer.from("gamecult-linux-service-seed-v1\0", "utf8"), - Buffer.from(providerIdentityContext, "utf8"), - Buffer.from(binding, "utf8"), - ])).digest(); - const protectedSeed = Buffer.from(providerSeed.map((value, index) => value ^ mask[index]!)); - const providerCredential = Buffer.from(encode([[ - "gamecult-provider-health-identity", - "gamecult.provider_health_identity.private.v1", - Buffer.from(encode([ - "gamecult.provider_health_identity.private.v1", providerIdentity, Array.from(providerPublic), Array.from(protectedSeed), - "linux_file_mode_machine_id_binding", binding, "v1", "os_installation_file_bound_cloneable_baseline", - "2026-01-01T00:00:00Z", Array(32).fill(0x2a), - ])), - "2026-01-01T00:00:00Z", - "gamecult.provider_health_identity.private.v1", - ]])); - let descriptorProviderCredential = providerCredential; - - const originalEnvironment = { ...process.env }; - context.after(() => { - for (const key of Object.keys(process.env)) if (!(key in originalEnvironment)) delete process.env[key]; - Object.assign(process.env, originalEnvironment); - }); - process.env.GAMECULT_IDUNN_RUNTIME_BUNDLE = bundle; - process.env.GAMECULT_IDUNN_CANDIDATE_BIND = candidate; - delete process.env.GAMECULT_IDUNN_PROCESS_WRITE_LEASE; - process.env.LISTEN_PID = String(process.pid); - process.env.LISTEN_FDS = "2"; - process.env.LISTEN_FDNAMES = "gamecult-idunn-runtime-activation-key:gamecult-runtime-presence-identity"; - const realReadFileSync = fs.readFileSync.bind(fs); - mock.method(fs, "readFileSync", (filePath: fs.PathOrFileDescriptor, ...args: unknown[]) => { - if (filePath === 3) return activationSeed; - if (filePath === 4) return descriptorProviderCredential; - if (filePath === "/etc/machine-id") return machineId; - return realReadFileSync(filePath as never, ...args as never[]); - }); - - const authority = loadIdunnRuntimeAuthorityFromEnvironment(target, contract, "127.0.0.1:17871"); - const signed = signIdunnRuntimePresence(authority, { - capabilities: [{ capability: "streampixels.service.api", schema: "streampixels.api.v1", compatibility: "v1", capacity: 1 }], - state: "warming", - detail: "starting", - writeLeaseSha256: null, - publisherSequence: 1, - observedAtUnixMillis: Date.now(), - }); - const proof = runtimePresenceProofPayload(signed.presence); - assert.equal((decode(expectedBytes) as unknown[]).length, 18); - assert.equal(signed.presence.expectedProjectionSha256, expectedSha); - const signedRecord = decode(signed.payload) as unknown[]; - assert.equal(crypto.verify(null, runtimePresenceProviderSigningMessage(proof), crypto.createPublicKey(providerKey), Buffer.from(signedRecord[21] as Uint8Array)), true); - assert.equal(crypto.verify(null, runtimePresenceActivationSigningMessage(proof), crypto.createPublicKey(activationKey), Buffer.from(signedRecord[23] as Uint8Array)), true); + for (const request of cases) assert.throws(() => signer.answerRouteObservation(request, "active"), /route observation/); +}); +async function startOdinPeer(context: TestContext, endpoint: string) { const peer = dgram.createSocket("udp4"); + const port = Number(endpoint.slice(endpoint.lastIndexOf(":") + 1)); await new Promise((resolve, reject) => { peer.once("error", reject); - peer.bind(0, "127.0.0.1", resolve); + peer.bind(port, "127.0.0.1", resolve); }); context.after(() => peer.close()); - const address = peer.address(); - assert.equal(typeof address, "object"); - const peerPort = (address as { port: number }).port; + const state = { + reject: false, + received: [] as Array<{ schemaVersion?: string; document: { payload: Uint8Array } }>, + responseAck: undefined as number | undefined, + failure: undefined as Error | undefined, + }; let serverSession: CultNetRudpSession | undefined; - let receivedMessage: unknown; - let responseAck: number | undefined; - let peerFailure: Error | undefined; - let rejectOdinAdmission = true; peer.on("message", (wire, remote) => { try { const packet = decodeRudpPacket(wire); if (packet.packetType === "connect") { serverSession = new CultNetRudpSession({ connectionId: packet.connectionId, initialSequence: 100 }); - const accept = serverSession.acceptConnect(packet, Date.now()); - peer.send(encodeRudpPacket(accept), remote.port, remote.address); + peer.send(encodeRudpPacket(serverSession.acceptConnect(packet, Date.now())), remote.port, remote.address); return; } if (!serverSession) throw new Error("Publisher sent data before establishing RUDP."); - const result = serverSession.receive(packet, Date.now()); - const frame = result.delivered.find((candidate) => candidate.channelId === "schema"); + const frame = serverSession.receive(packet, Date.now()).delivered.find((candidate) => candidate.channelId === "schema"); if (!frame) return; - receivedMessage = decode(frame.payload); - const responsePayload = encode(rejectOdinAdmission ? { - schemaVersion: "cultnet.error.v0", - error: "test admission denied", - routingHint: null, - code: null, - details: null, - } : { - schemaVersion: "cultnet.snapshot_response.v0", - messageId: "ack", - documents: [], - }); - const [response] = serverSession.sendMany("schema", responsePayload, { + state.received.push(decode(frame.payload) as never); + const [response] = serverSession.sendMany("schema", encode(state.reject + ? { schemaVersion: "cultnet.error.v0", error: "test admission denied", routingHint: null, code: null, details: null } + : { schemaVersion: "cultnet.snapshot_response.v0", messageId: "ack", documents: [] }), { reliable: true, ordered: true, nowMs: Date.now(), }); assert.ok(response); - responseAck = response.ack; + state.responseAck = response.ack; peer.send(encodeRudpPacket(response), remote.port, remote.address); } catch (error) { - peerFailure = error as Error; + state.failure = error as Error; } }); - const publisher = createIdunnRuntimePresencePublisher({ - authority, - endpoint: `rudp://127.0.0.1:${peerPort}`, - healthContract: contract, - capabilities: [{ capability: "streampixels.service.api", schema: "streampixels.api.v1", compatibility: "v1", capacity: 1 }], - }); - await assert.rejects(publisher.publish("warming", "local RUDP integration"), /Odin rejected runtime presence: test admission denied/); - assert.deepEqual(receivedMessage && (receivedMessage as { schemaVersion?: string }).schemaVersion, "cultnet.document_put_raw.v0"); - assert.ok(Buffer.from(encode(receivedMessage)).byteLength > 0); - assert.equal(responseAck, 2, "the response data packet must acknowledge the publisher's reliable schema packet"); - assert.equal(peerFailure, undefined, peerFailure?.message); - assert.equal(publisher.latestPresenceDocument(), null, "a rejected publication must not become the snapshot source"); + return state; +} + +test("the Odin publisher publishes the signer's document to the endpoint Expected names and surfaces Odin's rejection", async (context) => { + const endpoint = expectedFacts("service").odinEndpoint!; + const peer = await startOdinPeer(context, endpoint); + const signer = signerFor(context, "service"); + const publisher = createIdunnRuntimePresencePublisher({ signer, endpoint }); - rejectOdinAdmission = false; await publisher.publish("warming", "accepted presence"); - const snapshotDocument = await publisher.publishRouteObservation({ - schemaVersion: "cultnet.snapshot_request.v0", - messageId: "route-challenge-1", - schemaIds: ["gamecult.runtime_presence_health.v2"], - recordKeys: [target], - }); - const publishedDocument = (receivedMessage as { document: { payload: Uint8Array; schemaId: string; recordKey: string } }).document; - assert.ok(snapshotDocument); - assert.equal(snapshotDocument.schemaId, publishedDocument.schemaId); - assert.equal(snapshotDocument.recordKey, publishedDocument.recordKey); - assert.deepEqual(Buffer.from(snapshotDocument.payload), Buffer.from(publishedDocument.payload)); - const routePresence = decode(snapshotDocument.payload) as unknown[]; - assert.equal(routePresence[RUNTIME_PRESENCE_SLOT.state], "active"); - assert.equal(routePresence[RUNTIME_PRESENCE_SLOT.detail], "route-observation:route-challenge-1"); - snapshotDocument.payload[0] = snapshotDocument.payload[0]! ^ 0xff; - assert.deepEqual(Buffer.from(publisher.latestPresenceDocument()?.payload ?? []), Buffer.from(publishedDocument.payload), "snapshot readers cannot mutate publisher state"); + assert.equal(peer.received[0]?.schemaVersion, "cultnet.document_put_raw.v0"); + const published = decode(peer.received[0]!.document.payload) as unknown[]; + assert.equal(published[RUNTIME_PRESENCE_SLOT.state], "warming"); + assert.equal(published[RUNTIME_PRESENCE_SLOT.publisherSequence], 1); + assert.equal(peer.responseAck, 2, "the response data packet must acknowledge the publisher's reliable schema packet"); - descriptorProviderCredential = Buffer.from(encode([ - ...(decode(providerCredential) as unknown[][]), - ...(decode(providerCredential) as unknown[][]), - ])); - assert.throws( - () => loadIdunnRuntimeAuthorityFromEnvironment(target, contract, "127.0.0.1:17871"), - /exactly one envelope/, - ); + peer.reject = true; + await assert.rejects(publisher.publish("active", "denied presence"), /Odin rejected runtime presence: test admission denied/); + assert.equal(peer.failure, undefined, peer.failure?.message); }); -function cultCacheStore(type: string, schemaId: string, key: string, payload: Uint8Array): Buffer { - return Buffer.from(encode([ - "cultcache.store.v1", - [[schemaId, type, `${schemaId}.v1`, schemaId, "{}", [schemaId], []]], - [[key, schemaId, "2026-01-01T00:00:00Z", payload]], - ])); -} +test("the Odin publisher leaves a capability shortfall to the Rust correlation instead of refusing to publish", async (context) => { + const endpoint = expectedFacts("service").odinEndpoint!; + const peer = await startOdinPeer(context, endpoint); + const publisher = createIdunnRuntimePresencePublisher({ signer: signerFor(context, "service", -1), endpoint }); + await publisher.publish("active", "below minimum"); + const capabilities = (decode(peer.received[0]!.document.payload) as unknown[])[RUNTIME_PRESENCE_SLOT.capabilities] as unknown[][]; + assert.equal(capabilities[0]![3], expectedFacts("service").capabilities[0]!.capacity - 1); +}); -function privateKeyFromSeed(seed: Uint8Array): crypto.KeyObject { - return crypto.createPrivateKey({ - key: Buffer.concat([Buffer.from("302e020100300506032b657004220420", "hex"), Buffer.from(seed)]), - format: "der", - type: "pkcs8", - }); -} - -function rawPublic(key: crypto.KeyObject): Buffer { - const der = key.export({ type: "spki", format: "der" }) as Buffer; - return Buffer.from(der.subarray(der.length - 32)); -} - -function sha256Hex(bytes: Uint8Array): string { - return crypto.createHash("sha256").update(bytes).digest("hex"); -} +test("the Odin publisher takes only the endpoint Expected names", (context) => { + assert.throws( + () => createIdunnRuntimePresencePublisher({ signer: signerFor(context, "service"), endpoint: "rudp://127.0.0.1:1" }), + /does not match Idunn Expected dependency authority/, + ); + assert.throws( + () => createIdunnRuntimePresencePublisher({ signer: signerFor(context, "web"), endpoint: expectedFacts("service").odinEndpoint! }), + /does not match Idunn Expected dependency authority/, + ); +});