diff --git a/packages/cultmesh-ts/README.md b/packages/cultmesh-ts/README.md index 92e3c49d..9c4dfb01 100644 --- a/packages/cultmesh-ts/README.md +++ b/packages/cultmesh-ts/README.md @@ -280,6 +280,7 @@ const odinTransport = new CultMeshProviderRudpTransport({ endpoint: "rudp://127.0.0.1:17871", runtimeId: "voidbot-worker-7", connectionId: 0x43554c54, + sessionToken: process.env.ODIN_PROVIDER_SESSION_TOKEN, }); const session = new CultMeshProviderSession({ @@ -321,6 +322,10 @@ desired documents and durable receipt outbox. Do not advertise it as a public provider boundary until the surrounding CultNet session authenticates a claim authorizing the four-part provider identity. Source address and successful contact are not authority. +The adapter carries `sessionToken` in a typed RUDP Connect-evidence map beside +a fresh `clientSessionId`. Packet retransmits reuse that id; an actual reconnect +mints another. The receiver must decode and validate the token as authorization +evidence for the requested provider identity. ## RUDP Helpers diff --git a/packages/cultmesh-ts/src/provider-session-broker.ts b/packages/cultmesh-ts/src/provider-session-broker.ts index a8ddc3d5..eb9e18b5 100644 --- a/packages/cultmesh-ts/src/provider-session-broker.ts +++ b/packages/cultmesh-ts/src/provider-session-broker.ts @@ -74,6 +74,8 @@ export class CultMeshProviderSessionBroker { readonly #commands = new Map(); readonly #receipts = new Map(); readonly #publicationOwners = new Map(); + readonly #closedSessions = new Set(); + readonly #inFlightBySession = new Map(); readonly #expiryTimer: NodeJS.Timeout; #work: Promise = Promise.resolve(); #leaseSequence = 0; @@ -101,15 +103,34 @@ export class CultMeshProviderSessionBroker { request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession, ): Promise { + this.#inFlightBySession.set( + session.sessionId, + (this.#inFlightBySession.get(session.sessionId) ?? 0) + 1, + ); const work = this.#work.then(() => this.#handle(request, session)); this.#work = work.then(() => undefined, () => undefined); - return await work; + try { + return await work; + } finally { + const remaining = (this.#inFlightBySession.get(session.sessionId) ?? 1) - 1; + if (remaining > 0) { + this.#inFlightBySession.set(session.sessionId, remaining); + } else { + this.#inFlightBySession.delete(session.sessionId); + this.#closedSessions.delete(session.sessionId); + } + } } async #handle( request: CultNetOperationRequestMessage, session: CultMeshRudpServerSession, ): Promise { + if (this.#closedSessions.has(session.sessionId)) { + return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + "The physical provider session is closed or replaced.", + ]); + } if (request.serviceId !== CULTMESH_PROVIDER_SESSION_SERVICE_ID) { return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ `Unsupported service ${request.serviceId}.`, @@ -148,6 +169,9 @@ export class CultMeshProviderSessionBroker { } public sessionClosed(session: CultMeshRudpServerSession): void { + if ((this.#inFlightBySession.get(session.sessionId) ?? 0) > 0) { + this.#closedSessions.add(session.sessionId); + } for (const leaseId of this.#sessionLeases.get(session.sessionId) ?? []) { const record = this.#leases.get(leaseId); if (record?.session?.sessionId === session.sessionId) record.session = undefined; @@ -207,6 +231,11 @@ export class CultMeshProviderSessionBroker { "The CultNet session does not authorize this provider identity.", ]); } + if (this.#closedSessions.has(session.sessionId)) { + return this.#response(request, "denied", cultMeshProviderSessionSchemas.mutationAcceptance, {}, [ + "The physical provider session closed while registration authority was being checked.", + ]); + } const identityKey = identityTuple(identity); const previousLeaseId = this.#identityLeases.get(identityKey); let publications = new Map(); diff --git a/packages/cultmesh-ts/src/provider-session-wire.ts b/packages/cultmesh-ts/src/provider-session-wire.ts index 45b8cc2c..e89a3148 100644 --- a/packages/cultmesh-ts/src/provider-session-wire.ts +++ b/packages/cultmesh-ts/src/provider-session-wire.ts @@ -31,6 +31,13 @@ export type CultMeshProviderOperationStatus = | "denied" | "invalid"; +/** + * Transport evidence carried by a provider's RUDP Connect packet. + * + * `clientSessionId` distinguishes an actual reconnect from retransmission of + * the same Connect packet. The token remains authorization evidence only; it + * is not a transport-session identifier. + */ export interface CultMeshProviderConnectEvidenceWire { clientSessionId: string; sessionToken: string | null; @@ -123,21 +130,29 @@ export function decodeProviderSessionPayload(payload: string): T { return decode(Buffer.from(payload, "base64")) as T; } -export function encodeProviderConnectEvidence(evidence: CultMeshProviderConnectEvidenceWire): Uint8Array { +export function encodeProviderConnectEvidence( + evidence: CultMeshProviderConnectEvidenceWire, +): Uint8Array { assertProviderConnectEvidence(evidence); return encode(evidence); } -export function decodeProviderConnectEvidence(payload: Uint8Array): CultMeshProviderConnectEvidenceWire { +export function decodeProviderConnectEvidence( + payload: Uint8Array, +): CultMeshProviderConnectEvidenceWire { const evidence = decode(payload) as unknown; assertProviderConnectEvidence(evidence); return evidence; } -export function assertProviderConnectEvidence(value: unknown): asserts value is CultMeshProviderConnectEvidenceWire { +export function assertProviderConnectEvidence( + value: unknown, +): asserts value is CultMeshProviderConnectEvidenceWire { const evidence = requireObject(value, "provider Connect evidence"); requireText(evidence.clientSessionId, "clientSessionId"); - if (evidence.sessionToken !== null) requireText(evidence.sessionToken, "sessionToken"); + if (evidence.sessionToken !== null) { + requireText(evidence.sessionToken, "sessionToken"); + } } export function decodeProviderCommandDocument(document: CultNetRawDocumentRecord): CultMeshProviderCommandWire { diff --git a/packages/cultmesh-ts/test/provider-rudp-transport.test.ts b/packages/cultmesh-ts/test/provider-rudp-transport.test.ts index 6856f1cc..5b44c84b 100644 --- a/packages/cultmesh-ts/test/provider-rudp-transport.test.ts +++ b/packages/cultmesh-ts/test/provider-rudp-transport.test.ts @@ -1,12 +1,25 @@ import test from "node:test"; import assert from "node:assert/strict"; +import { createSocket } from "node:dgram"; import { decode } from "@msgpack/msgpack"; -import { CultNetDocumentRegistry } from "cultnet-ts"; +import { + CultNetDocumentRegistry, + CultNetRudpSession, + decodeRudpPacket, + encodeRudpPacket, + type CultNetOperationRequestMessage, +} from "cultnet-ts"; import { CultMesh, CultMeshProviderRudpTransport, CultMeshProviderSessionBroker, + CULTMESH_PROVIDER_SESSION_SERVICE_ID, + cultMeshProviderSessionOperations, + cultMeshProviderSessionSchemas, + decodeProviderConnectEvidence, + encodeProviderConnectEvidence, + encodeProviderSessionPayload, type CultMeshProviderCommand, type CultMeshProviderIdentity, type CultMeshProviderPublication, @@ -28,7 +41,9 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle" const broker = new CultMeshProviderSessionBroker({ runtimeId: "odin-test", createLeaseId: () => `lease-${++leaseSequence}`, - authorizeRegistration: () => true, + authorizeRegistration: (_identity, session) => + session.connectPayload !== undefined + && decodeProviderConnectEvidence(session.connectPayload).sessionToken === "signed-provider-token", acceptPublication: (_identity, publicationId, document) => { accepted.set(publicationId, decode(document.payload)); acceptedProvenance.set(publicationId, { @@ -54,10 +69,44 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle" }); await server.start(); + const unauthorizedConnection = await new CultMeshProviderRudpTransport({ + endpoint: `rudp://127.0.0.1:${server.bind.port}`, + runtimeId: "unauthorized-provider", + connectionId: 0x43554c54, + sessionToken: "not-authority", + }).connect(identity, new AbortController().signal); + try { + await assert.rejects( + unauthorizedConnection.register( + { identity, requestedLeaseDurationMs: 10_000 }, + new AbortController().signal, + ), + /denied|not authorized/i, + ); + assert.equal( + broker.activeLeaseCount, + 0, + "a successful RUDP handshake does not grant provider authority", + ); + } finally { + unauthorizedConnection.close(); + } + + const providerSocket = createSocket("udp4"); + const rawSend = providerSocket.send.bind(providerSocket); + let originalConnectWire: Buffer | undefined; + providerSocket.send = ((message: Uint8Array, ...args: unknown[]) => { + if (decodeRudpPacket(message).packetType === "connect" && !originalConnectWire) { + originalConnectWire = Buffer.from(message); + } + return (rawSend as (...parameters: unknown[]) => boolean)(message, ...args); + }) as typeof providerSocket.send; const transport = new CultMeshProviderRudpTransport({ endpoint: `rudp://127.0.0.1:${server.bind.port}`, runtimeId: identity.serviceInstanceId, connectionId: 0x43554c54, + sessionToken: "signed-provider-token", + socketFactory: () => providerSocket, }); const connection = await transport.connect(identity, new AbortController().signal); const secondIdentity: CultMeshProviderIdentity = { @@ -70,6 +119,7 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle" endpoint: `rudp://127.0.0.1:${server.bind.port}`, runtimeId: secondIdentity.serviceInstanceId, connectionId: 0x43554c54, + sessionToken: "signed-provider-token", }).connect(secondIdentity, new AbortController().signal); try { broker.enqueueCommand({ @@ -81,6 +131,12 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle" }); const lease = await connection.register({ identity, requestedLeaseDurationMs: 10_000 }, new AbortController().signal); assert.equal(lease.leaseId, "lease-1"); + assert.ok(originalConnectWire); + await new Promise((resolve, reject) => { + rawSend(originalConnectWire!, server.bind.port, "127.0.0.1", error => error ? reject(error) : resolve()); + }); + await new Promise(resolve => setTimeout(resolve, 25)); + assert.equal(broker.activeLeaseCount, 1, "a duplicate Connect cannot revoke the live provider lease"); const secondLease = await secondConnection.register({ identity: secondIdentity, requestedLeaseDurationMs: 10_000, @@ -155,7 +211,124 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle" } }); +test("closed physical sessions cannot finish delayed registration", async () => { + const authorization = deferred(); + const broker = new CultMeshProviderSessionBroker({ + runtimeId: "odin-test", + authorizeRegistration: () => authorization.promise, + acceptPublication: () => undefined, + deletePublications: () => undefined, + acceptReceipt: () => undefined, + }); + const session = { + sessionId: "closed-generation", + remote: { address: "127.0.0.1", family: "IPv4", port: 40000 }, + connectPayload: new Uint8Array(), + send: () => undefined, + }; + const request: CultNetOperationRequestMessage = { + schemaVersion: "cultnet.operation_request.v0", + messageId: "late-register", + serviceId: CULTMESH_PROVIDER_SESSION_SERVICE_ID, + operation: cultMeshProviderSessionOperations.register, + payloadSchema: cultMeshProviderSessionSchemas.registration, + payloadEncoding: "messagepack-base64", + payload: encodeProviderSessionPayload({ ...identity, requestedLeaseDurationMs: 10_000 }), + }; + try { + const pending = broker.handle(request, session); + await new Promise(resolve => setTimeout(resolve, 0)); + broker.sessionClosed(session); + authorization.resolve(true); + const response = await pending; + assert.equal(response.status, "denied"); + assert.equal(broker.activeLeaseCount, 0); + } finally { + broker.close(); + } +}); + +test("a replaced RUDP generation drops operations queued behind an active handler", async () => { + const connectionId = 0x43554c54; + const handlerGate = deferred(); + let handlerCalls = 0; + let closedSessions = 0; + const server = CultMesh.createRudpDocumentServer("odin-queue-test", connectionId, { + bindPort: 0, + documents: new CultNetDocumentRegistry(), + onOperationRequest: async request => { + handlerCalls += 1; + if (handlerCalls === 1) await handlerGate.promise; + return { + schemaVersion: "cultnet.operation_response.v0", + messageId: request.messageId, + serviceId: request.serviceId, + operation: request.operation, + status: "denied", + payloadSchema: cultMeshProviderSessionSchemas.mutationAcceptance, + payloadEncoding: "messagepack-base64", + payload: encodeProviderSessionPayload({}), + }; + }, + onSessionClosed: () => { closedSessions += 1; }, + onError: () => undefined, + }); + await server.start(); + + const socket = createSocket("udp4"); + const rawSend = socket.send.bind(socket); + const connection = await new CultMeshProviderRudpTransport({ + endpoint: `rudp://127.0.0.1:${server.bind.port}`, + runtimeId: identity.serviceInstanceId, + connectionId, + sessionToken: "signed-provider-token", + operationTimeoutMs: 100, + socketFactory: () => socket, + }).connect(identity, new AbortController().signal); + try { + const first = connection.register( + { identity, requestedLeaseDurationMs: 10_000 }, + new AbortController().signal, + ); + const second = connection.register( + { identity, requestedLeaseDurationMs: 10_000 }, + new AbortController().signal, + ); + await waitFor(() => handlerCalls === 1, "first queued provider operation"); + + const replacement = new CultNetRudpSession({ connectionId }); + const replacementWire = encodeRudpPacket(replacement.createConnect( + Date.now(), + Buffer.from(encodeProviderConnectEvidence({ + clientSessionId: "replacement-generation", + sessionToken: "signed-provider-token", + })), + )); + await new Promise((resolve, reject) => { + rawSend(replacementWire, server.bind.port, "127.0.0.1", error => error ? reject(error) : resolve()); + }); + await waitFor(() => closedSessions === 1, "old RUDP generation closure"); + handlerGate.resolve(); + await new Promise(resolve => setTimeout(resolve, 20)); + assert.equal(handlerCalls, 1, "transport-queued work from the closed generation must be dropped"); + + connection.close(); + await Promise.allSettled([first, second]); + } finally { + connection.close(); + server.close(); + } +}); + function deferred(): { promise: Promise; resolve(value: T): void } { let resolve!: (value: T) => void; return { promise: new Promise(done => resolve = done), resolve }; } + +async function waitFor(predicate: () => boolean, description: string): Promise { + const startedAt = Date.now(); + while (!predicate()) { + if (Date.now() - startedAt > 1_000) throw new Error(`Timed out waiting for ${description}.`); + await new Promise(resolve => setTimeout(resolve, 5)); + } +} diff --git a/packages/cultmesh-ts/test/provider-session-wire.test.ts b/packages/cultmesh-ts/test/provider-session-wire.test.ts index bedbfd29..ac607163 100644 --- a/packages/cultmesh-ts/test/provider-session-wire.test.ts +++ b/packages/cultmesh-ts/test/provider-session-wire.test.ts @@ -1,9 +1,15 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { encodeProviderSessionPayload } from "../src/provider-session-wire"; +import { + decodeProviderConnectEvidence, + encodeProviderConnectEvidence, + encodeProviderSessionPayload, +} from "../src/provider-session-wire"; const REGISTRATION_GOLDEN = "hqpwcm92aWRlcklkqGFldGhlcmlhsXNlcnZpY2VJbnN0YW5jZUlkq2FldGhlcmlhLTQyqmVuZHBvaW50SWSvYWV0aGVyaWEtcHVibGljp3ZlcnNlSWSmcHVibGljuHJlcXVlc3RlZExlYXNlRHVyYXRpb25Nc811MLBhdXRob3JpdHlMZWFzZUlkq2F1dGhvcml0eS03"; +const CONNECT_EVIDENCE_GOLDEN = "gq9jbGllbnRTZXNzaW9uSWSyYWV0aGVyaWEtY2xpZW50LTQyrHNlc3Npb25Ub2tlbrJvZGluLXNlc3Npb24tdG9rZW4="; +const TOKENLESS_CONNECT_EVIDENCE_GOLDEN = "gq9jbGllbnRTZXNzaW9uSWSyYWV0aGVyaWEtY2xpZW50LTQyrHNlc3Npb25Ub2tlbsA="; test("provider registration matches the shared C# MessagePack golden payload", () => { assert.equal(encodeProviderSessionPayload({ @@ -15,3 +21,23 @@ test("provider registration matches the shared C# MessagePack golden payload", ( authorityLeaseId: "authority-7", }), REGISTRATION_GOLDEN); }); + +test("provider Connect evidence keeps transport generation separate from authority", () => { + const evidence = { + clientSessionId: "aetheria-client-42", + sessionToken: "odin-session-token", + }; + const encoded = encodeProviderConnectEvidence(evidence); + assert.equal(Buffer.from(encoded).toString("base64"), CONNECT_EVIDENCE_GOLDEN); + assert.deepEqual( + decodeProviderConnectEvidence(encoded), + evidence, + ); + assert.equal( + Buffer.from(encodeProviderConnectEvidence({ + clientSessionId: "aetheria-client-42", + sessionToken: null, + })).toString("base64"), + TOKENLESS_CONNECT_EVIDENCE_GOLDEN, + ); +}); diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 0677d421..46e7cb6f 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -415,6 +415,14 @@ test("rudp session handshake acks reliable connect and accept packets", () => { assert.equal(server.connected, true); assert.deepEqual(server.pendingReliableSequences, [100]); + const retransmittedAccept = server.acceptConnect(connect, 15, Buffer.from("ok", "utf8")); + assert.deepEqual( + encodeRudpPacket(retransmittedAccept), + encodeRudpPacket(accept), + "a retransmitted Connect reuses the pending Accept packet", + ); + assert.deepEqual(server.pendingReliableSequences, [100]); + client.receive(accept, 20); assert.equal(client.connected, true); assert.deepEqual(client.pendingReliableSequences, []); diff --git a/src/GameCult.Mesh/CultMeshProviderSessionWire.cs b/src/GameCult.Mesh/CultMeshProviderSessionWire.cs index 37d991a0..03f3a70c 100644 --- a/src/GameCult.Mesh/CultMeshProviderSessionWire.cs +++ b/src/GameCult.Mesh/CultMeshProviderSessionWire.cs @@ -60,6 +60,19 @@ namespace GameCult.Mesh public const string PayloadEncoding = "messagepack-base64"; } + /// + /// Carries provider authorization evidence and a transport-owned connection + /// generation in an RUDP Connect packet. The token is not a session id. + /// + [MessagePackObject] + public sealed class CultMeshProviderConnectEvidenceWire + { + /// Gets or sets the fresh id minted for this physical connection. + [Key("clientSessionId")] public string ClientSessionId { get; set; } = string.Empty; + /// Gets or sets the optional signed provider authorization token. + [Key("sessionToken")] public string? SessionToken { get; set; } + } + /// Requests a fenced provider lease from the provider-session broker. [MessagePackObject] public sealed class CultMeshProviderRegistrationWire @@ -245,6 +258,24 @@ namespace GameCult.Mesh return value; } + /// Encodes provider RUDP Connect evidence as a named MessagePack map. + public static byte[] EncodeConnectEvidence(CultMeshProviderConnectEvidenceWire evidence) + { + if (evidence == null) throw new ArgumentNullException(nameof(evidence)); + Validate(evidence); + return MessagePackSerializer.Serialize(evidence, Options); + } + + /// Decodes and validates provider RUDP Connect evidence. + public static CultMeshProviderConnectEvidenceWire DecodeConnectEvidence(byte[] payload) + { + if (payload == null) throw new ArgumentNullException(nameof(payload)); + var evidence = MessagePackSerializer.Deserialize(payload, Options); + if (evidence == null) throw new MessagePackSerializationException("Provider Connect evidence decoded to null."); + Validate(evidence); + return evidence; + } + /// Decodes a retained provider command carried as a raw CultNet document. public static CultMeshProviderCommandWire DecodeCommandDocument(CultNetRawDocumentRecord document) { @@ -336,6 +367,10 @@ namespace GameCult.Mesh { switch (value) { + case CultMeshProviderConnectEvidenceWire evidence: + RequireText(evidence.ClientSessionId, nameof(evidence.ClientSessionId)); + RequireOptionalText(evidence.SessionToken, nameof(evidence.SessionToken)); + break; case CultMeshProviderRegistrationWire registration: RequireIdentity(registration.ProviderId, registration.ServiceInstanceId, registration.EndpointId, registration.VerseId); RequirePositive(registration.RequestedLeaseDurationMs, nameof(registration.RequestedLeaseDurationMs)); diff --git a/tests/GameCult.Mesh.Tests/CultMeshProviderSessionWireTests.cs b/tests/GameCult.Mesh.Tests/CultMeshProviderSessionWireTests.cs index 7e6f89a1..75125148 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshProviderSessionWireTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshProviderSessionWireTests.cs @@ -10,6 +10,32 @@ namespace GameCult.Mesh.Tests; public sealed class CultMeshProviderSessionWireTests { private const string RegistrationGolden = "hqpwcm92aWRlcklkqGFldGhlcmlhsXNlcnZpY2VJbnN0YW5jZUlkq2FldGhlcmlhLTQyqmVuZHBvaW50SWSvYWV0aGVyaWEtcHVibGljp3ZlcnNlSWSmcHVibGljuHJlcXVlc3RlZExlYXNlRHVyYXRpb25Nc811MLBhdXRob3JpdHlMZWFzZUlkq2F1dGhvcml0eS03"; + private const string ConnectEvidenceGolden = "gq9jbGllbnRTZXNzaW9uSWSyYWV0aGVyaWEtY2xpZW50LTQyrHNlc3Npb25Ub2tlbrJvZGluLXNlc3Npb24tdG9rZW4="; + private const string TokenlessConnectEvidenceGolden = "gq9jbGllbnRTZXNzaW9uSWSyYWV0aGVyaWEtY2xpZW50LTQyrHNlc3Npb25Ub2tlbsA="; + + [Test] + public void ConnectEvidence_SeparatesTransportGenerationFromAuthorityToken() + { + var encoded = CultMeshProviderSessionWire.EncodeConnectEvidence( + new CultMeshProviderConnectEvidenceWire + { + ClientSessionId = "aetheria-client-42", + SessionToken = "odin-session-token" + }); + var decoded = CultMeshProviderSessionWire.DecodeConnectEvidence(encoded); + + Assert.Multiple(() => + { + Assert.That(Convert.ToBase64String(encoded), Is.EqualTo(ConnectEvidenceGolden), + "C#, TypeScript, and Rust must emit the same Connect evidence map"); + Assert.That(decoded.ClientSessionId, Is.EqualTo("aetheria-client-42")); + Assert.That(decoded.SessionToken, Is.EqualTo("odin-session-token")); + Assert.That( + Convert.ToBase64String(CultMeshProviderSessionWire.EncodeConnectEvidence( + new CultMeshProviderConnectEvidenceWire { ClientSessionId = "aetheria-client-42" })), + Is.EqualTo(TokenlessConnectEvidenceGolden)); + }); + } [Test] public void Registration_RoundTrips_InsideCanonicalCultNetOperationEnvelope()