From 6c848bd6c0d961251c6ed210fa64e80ae0466bea Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 18 Aug 2026 17:21:58 +0200 Subject: [PATCH] Bind CultMesh authority identity to physical routes --- package-lock.json | 6 +- packages/cultmesh-browser/README.md | 11 +- packages/cultmesh-browser/package.json | 4 +- packages/cultmesh-browser/src/index.ts | 153 ++++++++++--- .../test/cultmesh-browser.test.ts | 160 ++++++++++++- .../cultmesh.session-accepted.schema.json | 18 ++ .../cultmesh.session-open.schema.json | 17 ++ ...ultmesh.verse-catalog-response.schema.json | 13 ++ packages/cultnet-ts/package.json | 2 +- packages/cultnet-ts/src/contracts.ts | 45 +++- packages/cultnet-ts/test/cultnet.test.ts | 19 ++ samples/eve-browser-network/Program.cs | 35 ++- samples/eve-browser-network/README.md | 3 + samples/eve-browser-network/browser/main.ts | 6 +- scripts/verify-eve-browser-network.mjs | 33 ++- .../CultMeshNativeQuicRealtimeTransport.cs | 31 ++- .../CultMeshQuicRealtimeTransport.cs | 32 ++- src/GameCult.Mesh/CultMeshClient.cs | 12 +- src/GameCult.Mesh/CultMeshContentSessions.cs | 43 +++- .../CultMeshContentTransports.cs | 2 +- src/GameCult.Mesh/CultMeshDiscoveryService.cs | 3 +- .../CultMeshRealtimeTransports.cs | 2 +- src/GameCult.Mesh/CultMeshSessionIdentity.cs | 213 ++++++++++++++++++ src/GameCult.Mesh/CultMeshSessions.cs | 187 ++++++++++----- src/GameCult.Mesh/CultMeshSnapshots.cs | 2 +- .../CultMeshTcpContentTransport.cs | 113 +++++++++- src/GameCult.Mesh/CultMeshVerse.cs | 98 +++++++- src/GameCult.Mesh/CultMeshVerseDiscovery.cs | 21 +- src/GameCult.Mesh/docs/public-api.md | 8 +- src/GameCult.Mesh/docs/transport-planes.md | 14 +- src/GameCult.Mesh/docs/verses.md | 13 +- .../CultNetOperationServer.cs | 10 + .../CultNetSchemaMessageSerialization.cs | 2 + .../CultNetSchemaMessages.cs | 47 ++++ ...ultMeshNativeQuicRealtimeTransportTests.cs | 10 +- .../CultMeshQuicRealtimeTransportTests.cs | 18 +- .../CultMeshBodySessionTests.cs | 12 +- .../CultMeshClientTests.cs | 49 +++- .../CultMeshContentSessionTests.cs | 57 ++++- .../CultMeshDiscoveryServiceTests.cs | 2 +- .../CultMeshSessionManagerTests.cs | 154 ++++++++++++- .../CultMeshTransportModularityTests.cs | 101 ++++++++- .../CultNetOperationServerTests.cs | 38 +++- .../NetworkingTests.cs | 15 +- tools/GameCult.TransportParity/Program.cs | 29 ++- 45 files changed, 1670 insertions(+), 193 deletions(-) create mode 100644 packages/cultnet-ts/contracts/cultmesh.session-accepted.schema.json create mode 100644 packages/cultnet-ts/contracts/cultmesh.session-open.schema.json create mode 100644 src/GameCult.Mesh/CultMeshSessionIdentity.cs diff --git a/package-lock.json b/package-lock.json index 74a753a5..1e955e9e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -846,11 +846,11 @@ } }, "packages/cultmesh-browser": { - "version": "0.2.0", + "version": "0.3.0", "license": "MIT", "dependencies": { "@msgpack/msgpack": "^3.1.3", - "cultnet-ts": "^0.1.1" + "cultnet-ts": "^0.2.0" }, "devDependencies": { "@types/node": "^24.0.0", @@ -878,7 +878,7 @@ } }, "packages/cultnet-ts": { - "version": "0.1.1", + "version": "0.2.0", "license": "MIT", "dependencies": { "@msgpack/msgpack": "^3.1.3", diff --git a/packages/cultmesh-browser/README.md b/packages/cultmesh-browser/README.md index 4f5c7e8b..59a44016 100644 --- a/packages/cultmesh-browser/README.md +++ b/packages/cultmesh-browser/README.md @@ -43,10 +43,19 @@ selection, reconnect, or resubscription. This is deliberately distinct from an Eve surface's `providerId`, which identifies the UI provider inside the Verse. The physical socket is a replaceable route, not either stable identity. +Odin returns exact authority-route bindings. The browser selects only routes +bound to the requested authority and `cultmesh.documents.v1`, then performs a +session-open handshake. It does not report `connected` until the peer proves +the exact Verse, authority runtime, protocol, and route generation. Operation +responses must also name that authority. Legacy catalogs are accepted only +when one authority runtime makes the endpoint binding unambiguous. + The browser handshake carries authentication through the host application's ordinary secure cookie or equivalent upgrade credential. The C# host adapter rejects endpoints without an authorization predicate unless anonymous local development is explicitly enabled. Run `scripts/verify-eve-browser-network.mjs --eve-root ` for the -real Chromium + C# headless chronology. +real Chromium + C# headless chronology. The witness publishes a lower-priority +decoy endpoint owned by another runtime and proves that neither client connects +to it before and after route rotation. diff --git a/packages/cultmesh-browser/package.json b/packages/cultmesh-browser/package.json index 41833562..c49062b0 100644 --- a/packages/cultmesh-browser/package.json +++ b/packages/cultmesh-browser/package.json @@ -1,6 +1,6 @@ { "name": "cultmesh-browser", - "version": "0.2.0", + "version": "0.3.0", "description": "Browser-safe CultMesh document leases and typed operations over CultNet WebSocket transport.", "type": "module", "main": "dist/index.js", @@ -41,7 +41,7 @@ }, "dependencies": { "@msgpack/msgpack": "^3.1.3", - "cultnet-ts": "^0.1.1" + "cultnet-ts": "^0.2.0" }, "devDependencies": { "@types/node": "^24.0.0", diff --git a/packages/cultmesh-browser/src/index.ts b/packages/cultmesh-browser/src/index.ts index efeb4515..bbc253f4 100644 --- a/packages/cultmesh-browser/src/index.ts +++ b/packages/cultmesh-browser/src/index.ts @@ -12,6 +12,8 @@ import { type CultNetSnapshotResponseRawMessage, type CultMeshVerseCatalogRequestMessage, type CultMeshVerseCatalogResponseMessage, + type CultMeshSessionAcceptedMessage, + type CultMeshSessionOpenMessage, } from "cultnet-ts/contracts"; export interface CultMeshBrowserIdentity { @@ -21,7 +23,8 @@ export interface CultMeshBrowserIdentity { export interface CultMeshBrowserRoute extends CultMeshBrowserIdentity { endpoint: string; - generation?: string; + protocolId?: string; + generation: string; } export interface CultMeshBrowserRendezvous { @@ -173,6 +176,7 @@ export interface CultMeshOperationOptions { payloadSchema: string; payload: unknown; idempotencyKey?: string; + /** @deprecated The connected authority is the operation target. A different value is rejected. */ targetRuntimeId?: string; } @@ -338,6 +342,11 @@ export class CultMeshBrowserClient implements AsyncDisposable { requireText(options.serviceId, "serviceId"); requireText(options.operation, "operation"); requireText(options.payloadSchema, "payloadSchema"); + if (options.targetRuntimeId && options.targetRuntimeId !== this.identity.authorityRuntimeId) { + throw new Error( + `CultMesh operation target '${options.targetRuntimeId}' does not match connected authority '${this.identity.authorityRuntimeId}'.`, + ); + } await this.ensureConnected(); const messageId = options.idempotencyKey ?? this.#options.createId(); requireText(messageId, "operation idempotencyKey"); @@ -353,7 +362,7 @@ export class CultMeshBrowserClient implements AsyncDisposable { payloadEncoding: "messagepack-base64", payload: bytesToBase64(encode(options.payload)), sourceRuntimeId: this.runtimeId, - ...(options.targetRuntimeId ? { targetRuntimeId: options.targetRuntimeId } : {}), + targetRuntimeId: this.identity.authorityRuntimeId, }; const response = new Promise((resolve, reject) => { const timer = setTimeout(() => { @@ -489,13 +498,67 @@ export class CultMeshBrowserClient implements AsyncDisposable { socket.binaryType = "arraybuffer"; try { await new Promise((resolve, reject) => { + let opened = false; + let settled = false; + const handshakeId = this.#options.createId(); + const finish = (error?: Error) => { + if (settled) return; + settled = true; + clearTimeout(timer); + if (error) reject(error); + else resolve(); + }; const timer = setTimeout( - () => reject(new Error(`CultMesh WebSocket open timed out for '${route.endpoint}'.`)), + () => finish(new Error( + opened + ? `CultMesh authority handshake timed out for '${route.endpoint}'.` + : `CultMesh WebSocket open timed out for '${route.endpoint}'.`, + )), this.#options.connectTimeoutMs, ); - socket.onopen = () => { clearTimeout(timer); resolve(); }; - socket.onerror = () => { clearTimeout(timer); reject(new Error(`CultMesh WebSocket could not open '${route.endpoint}'.`)); }; - socket.onclose = () => { clearTimeout(timer); reject(new Error(`CultMesh WebSocket closed while opening '${route.endpoint}'.`)); }; + socket.onopen = () => { + opened = true; + try { + sendSocketMessage(socket, { + schemaVersion: "cultnet.hello.v0", + runtimeId: this.runtimeId, + runtimeKind: "browser", + supportedMessageVersions: [ + "cultmesh.session_open.v1", + "cultmesh.session_accepted.v1", + "cultnet.database_subscribe.v0", + "cultnet.database_unsubscribe.v0", + "cultnet.database_change_raw.v0", + "cultnet.operation_request.v0", + "cultnet.operation_response.v0", + ], + }, this.#options.maxFrameBytes); + const request: CultMeshSessionOpenMessage = { + schemaVersion: "cultmesh.session_open.v1", + messageId: handshakeId, + sourceRuntimeId: this.runtimeId, + verseId: this.identity.verseId, + authorityRuntimeId: this.identity.authorityRuntimeId, + protocolId: route.protocolId ?? "cultmesh.documents.v1", + routeGeneration: route.generation, + }; + sendSocketMessage(socket, request, this.#options.maxFrameBytes); + } catch (error) { + finish(error instanceof Error ? error : new Error(String(error))); + } + }; + socket.onmessage = event => { + void decodeSocketMessage(event, this.#options.maxFrameBytes).then(message => { + if (message.schemaVersion === "cultnet.error.v0") { + finish(new Error(message.error)); + return; + } + if (message.schemaVersion !== "cultmesh.session_accepted.v1" || message.messageId !== handshakeId) return; + finish(validateSessionAcceptance(message, this.identity, route)); + }).catch(error => finish(error instanceof Error ? error : new Error(String(error)))); + }; + socket.onerror = () => finish(new Error(`CultMesh WebSocket could not open '${route.endpoint}'.`)); + socket.onclose = () => finish(new Error(`CultMesh WebSocket closed before proving authority at '${route.endpoint}'.`)); }); } catch (error) { socket.close(1000, "CultMesh route open failed"); @@ -511,18 +574,6 @@ export class CultMeshBrowserClient implements AsyncDisposable { socket.onclose = () => this.handleClosed(generation); this.setState("connected"); try { - this.send({ - schemaVersion: "cultnet.hello.v0", - runtimeId: this.runtimeId, - runtimeKind: "browser", - supportedMessageVersions: [ - "cultnet.database_subscribe.v0", - "cultnet.database_unsubscribe.v0", - "cultnet.database_change_raw.v0", - "cultnet.operation_request.v0", - "cultnet.operation_response.v0", - ], - }); await Promise.all([...this.#leases.values()].map(lease => this.subscribe(lease))); } catch (error) { if (this.#socket === socket) this.#socket = undefined; @@ -576,6 +627,15 @@ export class CultMeshBrowserClient implements AsyncDisposable { case "cultnet.operation_response.v0": { const pending = this.#pendingOperations.get(message.messageId); if (!pending) return; + if (message.sourceRuntimeId !== this.identity.authorityRuntimeId) { + this.#pendingOperations.delete(message.messageId); + clearTimeout(pending.timer); + pending.reject(new Error( + `CultMesh operation response came from '${message.sourceRuntimeId ?? "unknown"}', expected '${this.identity.authorityRuntimeId}'.`, + )); + this.#socket?.close(1008, "CultMesh authority identity changed"); + return; + } this.#pendingOperations.delete(message.messageId); clearTimeout(pending.timer); if (message.payloadSchema === cultNetOperationFailureSchema) { @@ -733,33 +793,72 @@ function selectOdinRoute( response: CultMeshVerseCatalogResponseMessage, identity: CultMeshBrowserIdentity, ): CultMeshBrowserRoute { - const matchingVerses = response.verses.filter(verse => - verse.verseId === identity.verseId && - verse.authorityRuntimeIds.includes(identity.authorityRuntimeId)); + const matchingVerses = response.verses.filter(verse => verse.verseId === identity.verseId); if (matchingVerses.length !== 1) { throw new Error( `Odin returned ${matchingVerses.length} routes for Verse '${identity.verseId}' authority runtime '${identity.authorityRuntimeId}'.`, ); } const verse = matchingVerses[0]; - const endpoints = verse.discoveryEndpoints.filter(endpoint => { + if (verse.authorityRoutes == null && verse.authorityRuntimeIds.length !== 1) { + throw new Error( + `Odin returned ambiguous legacy routes for Verse '${identity.verseId}'. Explicit authority route bindings are required.`, + ); + } + if (verse.authorityRoutes == null && verse.authorityRuntimeIds[0] !== identity.authorityRuntimeId) { + throw new Error( + `Odin advertised no route for authority runtime '${identity.authorityRuntimeId}'.`, + ); + } + const boundRoutes = verse.authorityRoutes ?? verse.discoveryEndpoints.map(endpoint => ({ + authorityRuntimeId: verse.authorityRuntimeIds[0], + endpoint, + protocolIds: ["cultmesh.documents.v1"], + priority: 0, + generation: [verse.compatibility.transportVersion, verse.compatibility.rulesHash, endpoint].join(":"), + })); + const routes = boundRoutes.filter(route => + route.authorityRuntimeId === identity.authorityRuntimeId && + route.protocolIds.includes("cultmesh.documents.v1") && + (() => { try { - requireWebSocketEndpoint(endpoint, "Odin provider route"); + requireWebSocketEndpoint(route.endpoint, "Odin provider route"); return true; } catch { return false; } - }); - if (endpoints.length === 0) { + })()).sort((left, right) => left.priority - right.priority || left.endpoint.localeCompare(right.endpoint)); + if (routes.length === 0) { throw new Error(`Odin advertised no browser-compatible route for authority runtime '${identity.authorityRuntimeId}'.`); } + const route = routes[0]; return { ...identity, - endpoint: endpoints[0], - generation: [verse.compatibility.transportVersion, verse.compatibility.rulesHash, endpoints[0]].join(":"), + endpoint: route.endpoint, + protocolId: "cultmesh.documents.v1", + generation: route.generation, }; } +function validateSessionAcceptance( + message: CultMeshSessionAcceptedMessage, + identity: CultMeshBrowserIdentity, + route: CultMeshBrowserRoute, +): Error | undefined { + if (!message.accepted) return new Error(message.error ?? "CultMesh authority rejected the session."); + const protocolId = route.protocolId ?? "cultmesh.documents.v1"; + if (message.verseId !== identity.verseId || + message.authorityRuntimeId !== identity.authorityRuntimeId || + message.protocolId !== protocolId || + message.routeGeneration !== route.generation) { + return new Error( + `CultMesh route proved '${message.verseId}/${message.authorityRuntimeId}/${message.protocolId}/${message.routeGeneration}', ` + + `expected '${identity.verseId}/${identity.authorityRuntimeId}/${protocolId}/${route.generation}'.`, + ); + } + return undefined; +} + function requireText(value: string, field: string): void { if (!value || value.trim().length === 0) throw new Error(`CultMesh browser ${field} is required.`); } diff --git a/packages/cultmesh-browser/test/cultmesh-browser.test.ts b/packages/cultmesh-browser/test/cultmesh-browser.test.ts index 624bdf17..a785d47c 100644 --- a/packages/cultmesh-browser/test/cultmesh-browser.test.ts +++ b/packages/cultmesh-browser/test/cultmesh-browser.test.ts @@ -9,6 +9,7 @@ import { type CultNetMessage, type CultNetOperationRequestMessage, type CultNetRawDocumentRecord, + type CultMeshSessionOpenMessage, } from "cultnet-ts/contracts"; import { @@ -48,6 +49,46 @@ test("Odin rendezvous resolves stable authority-runtime identity and observes ro ); }); +test("Odin rendezvous never selects another authority's faster endpoint", async () => { + const odin = new FakeOdin("ws://127.0.0.1:4050/mesh"); + odin.decoyEndpoint = "ws://127.0.0.1:4049/decoy"; + const rendezvous = new CultMeshBrowserOdinRendezvous({ + endpoints: ["ws://127.0.0.1:4040/odin"], + runtimeId: "browser-route-binding-test", + socketFactory: () => odin.open(), + }); + + const route = await rendezvous.resolve({ + verseId: "sample.counter", + authorityRuntimeId: "sample.counter-daemon", + }); + assert.equal(route.endpoint, "ws://127.0.0.1:4050/mesh"); +}); + +test("Odin rendezvous accepts one-authority legacy routes and rejects ambiguous legacy catalogs", async () => { + const odin = new FakeOdin("ws://127.0.0.1:4070/mesh"); + odin.omitAuthorityRoutes = true; + const rendezvous = new CultMeshBrowserOdinRendezvous({ + endpoints: ["ws://127.0.0.1:4040/odin"], + runtimeId: "browser-legacy-route-test", + socketFactory: () => odin.open(), + }); + + const route = await rendezvous.resolve({ + verseId: "sample.counter", + authorityRuntimeId: "sample.counter-daemon", + }); + assert.equal(route.endpoint, "ws://127.0.0.1:4070/mesh"); + + odin.decoyEndpoint = "ws://127.0.0.1:4069/decoy"; + await assert.rejects( + rendezvous.resolve({ verseId: "sample.counter", authorityRuntimeId: "sample.counter-daemon" }), + (error: unknown) => + error instanceof AggregateError && + error.errors.some(inner => inner instanceof Error && /ambiguous legacy routes/.test(inner.message)), + ); +}); + test("browser client leases provider state, invokes an operation, and follows route replacement", async () => { const first = new FakeProvider("ws://127.0.0.1:4101/mesh", 4); const second = new FakeProvider("ws://127.0.0.1:4102/mesh", 9); @@ -117,18 +158,33 @@ test("browser client rejects wrong rendezvous identity and has no document write assert.equal("setDocument" in CultMeshBrowserClient.prototype, false); }); +test("browser client rejects a route whose peer proves the wrong authority", async () => { + const provider = new FakeProvider("ws://127.0.0.1:4251/mesh", 1, false, "intruder-runtime"); + await assert.rejects( + CultMeshBrowserClient.connect({ + verseId: "sample.counter", + authorityRuntimeId: "sample.counter-daemon", + runtimeId: "browser-peer-proof-test", + rendezvous: { resolve: async () => ({ ...provider.route, authorityRuntimeId: "sample.counter-daemon" }) }, + socketFactory: () => provider.open(), + }), + /proved .*intruder-runtime.*expected .*sample.counter-daemon/, + ); +}); + test("browser client bounds unanswered provider operations", async () => { const route = { verseId: "sample.counter", authorityRuntimeId: "sample.counter-daemon", endpoint: "ws://127.0.0.1:4301/mesh", + generation: "route-4301", }; const client = await CultMeshBrowserClient.connect({ ...route, runtimeId: "browser-timeout-test", rendezvous: { resolve: async () => route }, requestTimeoutMs: 5, - socketFactory: () => new FakeSocket(() => undefined, () => undefined), + socketFactory: () => handshakeOnlySocket(route), }); await assert.rejects( client.invoke({ @@ -172,11 +228,39 @@ test("browser client rejects a correlated framework failure without waiting for await client.dispose(); }); +test("browser client rejects an operation response from another runtime", async () => { + const provider = new FakeProvider( + "ws://127.0.0.1:4371/mesh", + 1, + false, + "sample.counter-daemon", + "intruder-runtime", + ); + const client = await CultMeshBrowserClient.connect({ + ...provider.route, + runtimeId: "browser-response-authority-test", + rendezvous: { resolve: async () => provider.route }, + socketFactory: () => provider.open(), + }); + + await assert.rejects( + client.invoke({ + serviceId: "sample.counter", + operation: "increment", + payloadSchema: "sample.increment.v1", + payload: { amount: 1 }, + }), + /response came from 'intruder-runtime'/, + ); + await client.dispose(); +}); + test("browser client rejects oversized outbound schema messages", async () => { const route = { verseId: "sample.counter", authorityRuntimeId: "sample.counter-daemon", endpoint: "ws://127.0.0.1:4401/mesh", + generation: "route-4401", }; await assert.rejects( CultMeshBrowserClient.connect({ @@ -220,8 +304,16 @@ class FakeProvider { #count: number; #subscriptions = new Map>(); #rejectOperations: boolean; + #runtimeId: string; + #responseRuntimeId: string; - constructor(endpoint: string, count: number, rejectOperations = false) { + constructor( + endpoint: string, + count: number, + rejectOperations = false, + runtimeId = "sample.counter-daemon", + responseRuntimeId = runtimeId, + ) { this.route = { verseId: "sample.counter", authorityRuntimeId: "sample.counter-daemon", @@ -230,6 +322,8 @@ class FakeProvider { }; this.#count = count; this.#rejectOperations = rejectOperations; + this.#runtimeId = runtimeId; + this.#responseRuntimeId = responseRuntimeId; } get activeSubscriptionCount(): number { @@ -250,6 +344,9 @@ class FakeProvider { private receive(socket: FakeSocket, wire: Uint8Array): void { const message = parseCultNetMessage(decode(wire)); switch (message.schemaVersion) { + case "cultmesh.session_open.v1": + this.acceptSession(socket, message); + return; case "cultnet.database_subscribe.v0": this.subscribe(socket, message); return; @@ -264,6 +361,18 @@ class FakeProvider { } } + private acceptSession(socket: FakeSocket, message: CultMeshSessionOpenMessage): void { + socket.deliver({ + schemaVersion: "cultmesh.session_accepted.v1", + messageId: message.messageId, + accepted: true, + verseId: message.verseId, + authorityRuntimeId: this.#runtimeId, + protocolId: message.protocolId, + routeGeneration: message.routeGeneration, + }); + } + private subscribe(socket: FakeSocket, message: CultNetDatabaseSubscribeMessage): void { this.#subscriptions.get(socket)!.set(message.subscriptionId, message); socket.deliver({ @@ -274,6 +383,7 @@ class FakeProvider { } private invoke(socket: FakeSocket, message: CultNetOperationRequestMessage): void { + assert.equal(message.targetRuntimeId, this.#runtimeId); if (this.#rejectOperations) { socket.deliver({ schemaVersion: "cultnet.operation_response.v0", @@ -288,7 +398,7 @@ class FakeProvider { message: "Expected payload schema 'sample.increment.v1'.", })), diagnostics: ["Request schema did not match."], - sourceRuntimeId: "sample.counter-daemon", + sourceRuntimeId: this.#responseRuntimeId, }); return; } @@ -314,7 +424,7 @@ class FakeProvider { payloadSchema: "sample.increment_receipt.v1", payloadEncoding: "messagepack-base64", payload: bytesToBase64(encode({ count: this.#count })), - sourceRuntimeId: "sample.counter-daemon", + sourceRuntimeId: this.#responseRuntimeId, }); } @@ -328,13 +438,15 @@ class FakeProvider { storedAt: "2026-08-17T00:00:00Z", payloadEncoding: "messagepack", payload: encode({ count: this.#count }), - sourceRuntimeId: "sample.counter-daemon", + sourceRuntimeId: this.#runtimeId, }; } } class FakeOdin { providerEndpoint: string; + decoyEndpoint: string | undefined; + omitAuthorityRoutes = false; constructor(providerEndpoint: string) { this.providerEndpoint = providerEndpoint; @@ -359,7 +471,25 @@ class FakeOdin { optionalPluginIds: [], }, discoveryEndpoints: [this.providerEndpoint], - authorityRuntimeIds: ["sample.counter-daemon"], + authorityRuntimeIds: this.decoyEndpoint + ? ["sample.counter-daemon", "sample.decoy-daemon"] + : ["sample.counter-daemon"], + ...(!this.omitAuthorityRoutes ? { authorityRoutes: [ + ...(this.decoyEndpoint ? [{ + authorityRuntimeId: "sample.decoy-daemon", + endpoint: this.decoyEndpoint, + protocolIds: ["cultmesh.documents.v1"], + priority: 0, + generation: "decoy-generation", + }] : []), + { + authorityRuntimeId: "sample.counter-daemon", + endpoint: this.providerEndpoint, + protocolIds: ["cultmesh.documents.v1"], + priority: 10, + generation: this.providerEndpoint, + }, + ] } : {}), }], }); }, () => undefined); @@ -367,6 +497,24 @@ class FakeOdin { } } +function handshakeOnlySocket(route: CultMeshBrowserRoute): FakeSocket { + let socket: FakeSocket; + socket = new FakeSocket(wire => { + const message = parseCultNetMessage(decode(wire)); + if (message.schemaVersion !== "cultmesh.session_open.v1") return; + socket.deliver({ + schemaVersion: "cultmesh.session_accepted.v1", + messageId: message.messageId, + accepted: true, + verseId: route.verseId, + authorityRuntimeId: route.authorityRuntimeId, + protocolId: message.protocolId, + routeGeneration: message.routeGeneration, + }); + }, () => undefined); + return socket; +} + class FakeSocket implements CultMeshBrowserSocket { binaryType: BinaryType = "arraybuffer"; readyState = 0; diff --git a/packages/cultnet-ts/contracts/cultmesh.session-accepted.schema.json b/packages/cultnet-ts/contracts/cultmesh.session-accepted.schema.json new file mode 100644 index 00000000..55a2f742 --- /dev/null +++ b/packages/cultnet-ts/contracts/cultmesh.session-accepted.schema.json @@ -0,0 +1,18 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/GameCult/cultnet-ts/contracts/cultmesh.session-accepted.schema.json", + "title": "CultMesh Session Accepted", + "type": "object", + "required": ["schemaVersion", "messageId", "accepted", "verseId", "authorityRuntimeId", "protocolId", "routeGeneration"], + "additionalProperties": false, + "properties": { + "schemaVersion": { "const": "cultmesh.session_accepted.v1" }, + "messageId": { "type": "string", "minLength": 1 }, + "accepted": { "type": "boolean" }, + "verseId": { "type": "string", "minLength": 1 }, + "authorityRuntimeId": { "type": "string", "minLength": 1 }, + "protocolId": { "type": "string", "minLength": 1 }, + "routeGeneration": { "type": "string" }, + "error": { "type": "string", "minLength": 1 } + } +} diff --git a/packages/cultnet-ts/contracts/cultmesh.session-open.schema.json b/packages/cultnet-ts/contracts/cultmesh.session-open.schema.json new file mode 100644 index 00000000..14d945ed --- /dev/null +++ b/packages/cultnet-ts/contracts/cultmesh.session-open.schema.json @@ -0,0 +1,17 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/GameCult/cultnet-ts/contracts/cultmesh.session-open.schema.json", + "title": "CultMesh Session Open", + "type": "object", + "required": ["schemaVersion", "messageId", "sourceRuntimeId", "verseId", "authorityRuntimeId", "protocolId", "routeGeneration"], + "additionalProperties": false, + "properties": { + "schemaVersion": { "const": "cultmesh.session_open.v1" }, + "messageId": { "type": "string", "minLength": 1 }, + "sourceRuntimeId": { "type": "string", "minLength": 1 }, + "verseId": { "type": "string", "minLength": 1 }, + "authorityRuntimeId": { "type": "string", "minLength": 1 }, + "protocolId": { "type": "string", "minLength": 1 }, + "routeGeneration": { "type": "string" } + } +} diff --git a/packages/cultnet-ts/contracts/cultmesh.verse-catalog-response.schema.json b/packages/cultnet-ts/contracts/cultmesh.verse-catalog-response.schema.json index 21367d09..6e7551d5 100644 --- a/packages/cultnet-ts/contracts/cultmesh.verse-catalog-response.schema.json +++ b/packages/cultnet-ts/contracts/cultmesh.verse-catalog-response.schema.json @@ -18,6 +18,18 @@ "optionalPluginIds": { "type": "array", "items": { "type": "string", "minLength": 1 }, "uniqueItems": true } } }, + "authorityRoute": { + "type": "object", + "required": ["authorityRuntimeId", "endpoint", "protocolIds", "priority", "generation"], + "additionalProperties": false, + "properties": { + "authorityRuntimeId": { "type": "string", "minLength": 1 }, + "endpoint": { "type": "string", "minLength": 1 }, + "protocolIds": { "type": "array", "items": { "type": "string", "minLength": 1 }, "uniqueItems": true }, + "priority": { "type": "integer", "minimum": 0 }, + "generation": { "type": "string", "minLength": 1 } + } + }, "verse": { "type": "object", "required": ["verseId", "displayName", "authorityModel", "compatibility", "discoveryEndpoints", "authorityRuntimeIds"], @@ -29,6 +41,7 @@ "compatibility": { "$ref": "#/$defs/compatibility" }, "discoveryEndpoints": { "type": "array", "items": { "type": "string", "minLength": 1 }, "uniqueItems": true }, "authorityRuntimeIds": { "type": "array", "items": { "type": "string", "minLength": 1 }, "uniqueItems": true }, + "authorityRoutes": { "type": "array", "items": { "$ref": "#/$defs/authorityRoute" } }, "parentVerseId": { "type": "string", "minLength": 1 }, "description": { "type": "string" } } diff --git a/packages/cultnet-ts/package.json b/packages/cultnet-ts/package.json index 6ff29340..df129a42 100644 --- a/packages/cultnet-ts/package.json +++ b/packages/cultnet-ts/package.json @@ -1,6 +1,6 @@ { "name": "cultnet-ts", - "version": "0.1.1", + "version": "0.2.0", "description": "CultNet-style typed direct-pipe messaging and state replication for TypeScript runtimes.", "main": "dist/index.js", "types": "dist/index.d.ts", diff --git a/packages/cultnet-ts/src/contracts.ts b/packages/cultnet-ts/src/contracts.ts index 1c3998ae..2d86ad27 100644 --- a/packages/cultnet-ts/src/contracts.ts +++ b/packages/cultnet-ts/src/contracts.ts @@ -30,6 +30,8 @@ import operationRequestSchema from "../contracts/cultnet.operation-request.schem import operationResponseSchema from "../contracts/cultnet.operation-response.schema.json"; import verseCatalogRequestSchema from "../contracts/cultmesh.verse-catalog-request.schema.json"; import verseCatalogResponseSchema from "../contracts/cultmesh.verse-catalog-response.schema.json"; +import sessionOpenSchema from "../contracts/cultmesh.session-open.schema.json"; +import sessionAcceptedSchema from "../contracts/cultmesh.session-accepted.schema.json"; import ghostlightAgentStateSchema from "../contracts/ghostlight.agent-state.schema.json"; export type CultNetWireContract = "cultnet.schema.v0" | "gamecult.networking.v0"; @@ -59,7 +61,9 @@ export type CultNetSchemaVersion = | "cultnet.operation_request.v0" | "cultnet.operation_response.v0" | "cultmesh.verse_catalog_request.v0" - | "cultmesh.verse_catalog_response.v0"; + | "cultmesh.verse_catalog_response.v0" + | "cultmesh.session_open.v1" + | "cultmesh.session_accepted.v1"; export type CultNetSchemaKind = "wire_message" | "document_payload" | "shared_contract"; export type CultNetRawPayloadEncoding = "messagepack"; @@ -350,10 +354,40 @@ export interface CultMeshVerseDescriptorMessage { compatibility: CultMeshVerseCompatibilityMessage; discoveryEndpoints: string[]; authorityRuntimeIds: string[]; + authorityRoutes?: CultMeshAuthorityRouteMessage[]; parentVerseId?: string; description?: string; } +export interface CultMeshAuthorityRouteMessage { + authorityRuntimeId: string; + endpoint: string; + protocolIds: string[]; + priority: number; + generation: string; +} + +export interface CultMeshSessionOpenMessage { + schemaVersion: "cultmesh.session_open.v1"; + messageId: string; + sourceRuntimeId: string; + verseId: string; + authorityRuntimeId: string; + protocolId: string; + routeGeneration: string; +} + +export interface CultMeshSessionAcceptedMessage { + schemaVersion: "cultmesh.session_accepted.v1"; + messageId: string; + accepted: boolean; + verseId: string; + authorityRuntimeId: string; + protocolId: string; + routeGeneration: string; + error?: string; +} + export interface CultMeshVerseCatalogRequestMessage { schemaVersion: "cultmesh.verse_catalog_request.v0"; messageId: string; @@ -618,7 +652,9 @@ export type CultNetMessage = | CultNetOperationRequestMessage | CultNetOperationResponseMessage | CultMeshVerseCatalogRequestMessage - | CultMeshVerseCatalogResponseMessage; + | CultMeshVerseCatalogResponseMessage + | CultMeshSessionOpenMessage + | CultMeshSessionAcceptedMessage; const ajv = new Ajv2020({ allErrors: true, @@ -651,6 +687,8 @@ const CULTNET_MESSAGE_SCHEMAS = [ operationResponseSchema, verseCatalogRequestSchema, verseCatalogResponseSchema, + sessionOpenSchema, + sessionAcceptedSchema, ] as const; const cultNetValidators = new Map(); @@ -1198,6 +1236,9 @@ function normalizeCultNetOptionalNulls( } } return; + case "cultmesh.session_accepted.v1": + stripNullProperties(candidate, ["error"]); + return; case "cultnet.schema_catalog_request.v0": stripNullProperties(candidate, ["schemaIds", "kinds"]); return; diff --git a/packages/cultnet-ts/test/cultnet.test.ts b/packages/cultnet-ts/test/cultnet.test.ts index 945e0a3b..f0f01b29 100644 --- a/packages/cultnet-ts/test/cultnet.test.ts +++ b/packages/cultnet-ts/test/cultnet.test.ts @@ -83,10 +83,29 @@ test("CultNet validates the canonical CultMesh Verse catalog messages", () => { }, discoveryEndpoints: ["wss://provider.example/cultmesh"], authorityRuntimeIds: ["sample.counter-provider"], + authorityRoutes: [{ + authorityRuntimeId: "sample.counter-provider", + endpoint: "wss://provider.example/cultmesh", + protocolIds: ["cultmesh.documents.v1"], + priority: 0, + generation: "provider-route-1", + }], }], }); assert.equal(response.schemaVersion, "cultmesh.verse_catalog_response.v0"); assert.equal(response.verses[0].authorityRuntimeIds[0], "sample.counter-provider"); + assert.equal(response.verses[0].authorityRoutes![0].generation, "provider-route-1"); + + const session = parseCultNetMessage({ + schemaVersion: "cultmesh.session_accepted.v1", + messageId: "session-1", + accepted: true, + verseId: "sample.counter", + authorityRuntimeId: "sample.counter-provider", + protocolId: "cultmesh.documents.v1", + routeGeneration: "provider-route-1", + }); + assert.equal(session.schemaVersion, "cultmesh.session_accepted.v1"); assert.throws(() => parseCultNetMessage({ ...response, diff --git a/samples/eve-browser-network/Program.cs b/samples/eve-browser-network/Program.cs index 4b60a18e..db92d3c1 100644 --- a/samples/eve-browser-network/Program.cs +++ b/samples/eve-browser-network/Program.cs @@ -37,13 +37,30 @@ static async Task RunOdinAsync(Args arguments) if (string.IsNullOrWhiteSpace(arguments.ProviderEndpoint)) throw new ArgumentException("The Odin fixture requires --provider-endpoint."); using var catalog = new CultMeshVerseCatalog(); + var routes = new List + { + new( + arguments.AuthorityRuntimeId, + arguments.ProviderEndpoint, + [CultMeshProtocols.Documents.Value], + priority: 10, + generation: arguments.RouteGeneration) + }; + if (!string.IsNullOrWhiteSpace(arguments.DecoyEndpoint)) + { + routes.Add(new CultMeshAuthorityRoute( + arguments.DecoyAuthorityRuntimeId, + arguments.DecoyEndpoint, + [CultMeshProtocols.Documents.Value], + priority: 0, + generation: arguments.DecoyRouteGeneration)); + } catalog.Upsert(new CultMeshVerseDescriptor( arguments.VerseId, arguments.VerseName, CultMeshVerseAuthorityModel.OperatorCluster, new CultMeshVerseCompatibility(arguments.TransportVersion, arguments.RulesHash), - discoveryEndpoints: [arguments.ProviderEndpoint], - authorityRuntimeIds: [arguments.AuthorityRuntimeId], + authorityRoutes: routes, description: "Local executable Odin fixture for the browser/network onboarding witness.")); await using var schemaServer = new CultNetWebSocketSchemaServer(); @@ -97,6 +114,12 @@ static async Task RunProviderAsync(Args arguments) await node.FlushAsync(); await using var schemaServer = new CultNetWebSocketSchemaServer(); + using var identity = new CultMeshSessionIdentityServer( + schemaServer, + arguments.AuthorityRuntimeId, + [arguments.VerseId], + [CultMeshProtocols.Documents.Value], + [arguments.RouteGeneration]); using var subscriptions = new CultNetDatabaseSubscriptionServer(schemaServer, node.Database); using var operationGate = new SemaphoreSlim(1, 1); using var operations = new CultNetOperationServer(schemaServer, arguments.AuthorityRuntimeId) @@ -347,12 +370,16 @@ sealed class Args public string StatePath { get; private init; } = "sample-counter.cc"; public string OdinEndpoint { get; private init; } = ""; public string ProviderEndpoint { get; private init; } = ""; + public string DecoyEndpoint { get; private init; } = ""; public string Token { get; private init; } = "sample-session"; public string VerseId { get; private init; } = "sample.counter"; public string VerseName { get; private init; } = "CultMesh browser counter"; public string AuthorityRuntimeId { get; private init; } = "sample.counter-daemon"; + public string DecoyAuthorityRuntimeId { get; private init; } = "sample.decoy-daemon"; public string TransportVersion { get; private init; } = "cultmesh.v0"; public string RulesHash { get; private init; } = "sample-counter-v1"; + public string RouteGeneration { get; private init; } = "sample-counter-route-1"; + public string DecoyRouteGeneration { get; private init; } = "sample-decoy-route-1"; public int ExpectedCount { get; private init; } = 2; public int BenchmarkOperations { get; private init; } = 10_000; @@ -369,12 +396,16 @@ sealed class Args StatePath = named.GetValueOrDefault("--state", "sample-counter.cc"), OdinEndpoint = named.GetValueOrDefault("--odin", ""), ProviderEndpoint = named.GetValueOrDefault("--provider-endpoint", ""), + DecoyEndpoint = named.GetValueOrDefault("--decoy-endpoint", ""), Token = named.GetValueOrDefault("--token", "sample-session"), VerseId = named.GetValueOrDefault("--verse-id", "sample.counter"), VerseName = named.GetValueOrDefault("--verse-name", "CultMesh browser counter"), AuthorityRuntimeId = named.GetValueOrDefault("--authority-runtime-id", "sample.counter-daemon"), + DecoyAuthorityRuntimeId = named.GetValueOrDefault("--decoy-authority-runtime-id", "sample.decoy-daemon"), TransportVersion = named.GetValueOrDefault("--transport-version", "cultmesh.v0"), RulesHash = named.GetValueOrDefault("--rules-hash", "sample-counter-v1"), + RouteGeneration = named.GetValueOrDefault("--route-generation", "sample-counter-route-1"), + DecoyRouteGeneration = named.GetValueOrDefault("--decoy-route-generation", "sample-decoy-route-1"), ExpectedCount = named.TryGetValue("--expected-count", out var expectedCount) ? int.Parse(expectedCount) : 2, diff --git a/samples/eve-browser-network/README.md b/samples/eve-browser-network/README.md index 7ad20a00..445bfb9d 100644 --- a/samples/eve-browser-network/README.md +++ b/samples/eve-browser-network/README.md @@ -12,6 +12,9 @@ This is the real-process continuation of `samples/eve-two-runtime`: without application transport or reconnect code; - one separate local Odin fixture answers the canonical CultMesh Verse catalog over binary CultNet WebSocket messages; +- one wrong-authority provider is deliberately advertised on a better-priority + route, and both clients must ignore it because the route is bound to another + runtime; - the browser and C# clients speak the same binary CultNet schema-v0 WebSocket messages through the authenticated host adapter. diff --git a/samples/eve-browser-network/browser/main.ts b/samples/eve-browser-network/browser/main.ts index 3947516b..3b0e28f9 100644 --- a/samples/eve-browser-network/browser/main.ts +++ b/samples/eve-browser-network/browser/main.ts @@ -86,7 +86,11 @@ try { window.__sampleCount = decodeCultNetPayload<{ count: number }>(counter.current!).count; window.__sampleReady = true; } catch (error) { - window.__sampleError = error instanceof Error ? error.stack || error.message : String(error); + const details = error instanceof AggregateError + ? error.errors.map(inner => inner instanceof Error ? inner.stack || inner.message : String(inner)).join("\n---\n") + : ""; + window.__sampleError = (error instanceof Error ? error.stack || error.message : String(error)) + + (details ? `\n${details}` : ""); host.textContent = window.__sampleError; throw error; } diff --git a/scripts/verify-eve-browser-network.mjs b/scripts/verify-eve-browser-network.mjs index 3f633dc9..20d819a0 100644 --- a/scripts/verify-eve-browser-network.mjs +++ b/scripts/verify-eve-browser-network.mjs @@ -17,9 +17,11 @@ const eveRoot = resolve(eveFlag >= 0 ? process.argv[eveFlag + 1] : join(repoRoot const sampleRoot = join(repoRoot, "samples", "eve-browser-network"); const workRoot = await mkdtemp(join(tmpdir(), "cultmesh-browser-network-")); const statePath = join(workRoot, "counter.cc"); +const decoyStatePath = join(workRoot, "decoy-counter.cc"); const bundlePath = join(workRoot, "bundle.js"); const token = "sample-session"; let provider; +let decoyProvider; let replacementProvider; let odin; let headless; @@ -41,6 +43,8 @@ try { alias: { "@gamecult/eve-contracts": join(eveRoot, "packages", "eve-contracts", "src", "index.ts"), "@gamecult/eve-browser-lowering": join(eveRoot, "packages", "eve-browser-lowering", "src", "index.ts"), + "cultmesh-browser": join(repoRoot, "packages", "cultmesh-browser", "src", "index.ts"), + "cultnet-ts/contracts": join(repoRoot, "packages", "cultnet-ts", "src", "contracts.ts"), }, logLevel: "warning", }); @@ -54,14 +58,22 @@ try { "--verbosity", "quiet", "-p:NoWarn=1591%3BCS8632", `-p:EveRoot=${eveRoot}`, + `-p:CultLibRoot=${repoRoot}`, ]); const providerPort = await freePort(); + const decoyProviderPort = await freePort(); const replacementProviderPort = await freePort(); const odinPort = await freePort(); const httpPort = await freePort(); provider = await startProvider(providerPort); const endpoint = await provider.waitFor("PROVIDER_READY "); - odin = await startOdin(odinPort, endpoint); + decoyProvider = await startProvider(decoyProviderPort, { + authorityRuntimeId: "sample.decoy-daemon", + routeGeneration: "sample-decoy-route-1", + statePath: decoyStatePath, + }); + const decoyEndpoint = await decoyProvider.waitFor("PROVIDER_READY "); + odin = await startOdin(odinPort, endpoint, decoyEndpoint); const odinEndpoint = await odin.waitFor("ODIN_READY "); headless = startDotnet([ "headless", "--odin", odinEndpoint, "--verse-id", "sample.counter", @@ -91,7 +103,7 @@ try { const restartedEndpoint = await replacementProvider.waitFor("PROVIDER_READY "); assert.notEqual(restartedEndpoint, endpoint); await stop(odin.process); - odin = await startOdin(odinPort, restartedEndpoint); + odin = await startOdin(odinPort, restartedEndpoint, decoyEndpoint); assert.equal(await odin.waitFor("ODIN_READY "), odinEndpoint); await stop(provider.process); provider = replacementProvider; @@ -122,6 +134,7 @@ try { console.log(JSON.stringify({ providerEndpoint: endpoint, replacementProviderEndpoint: restartedEndpoint, + rejectedWrongAuthorityEndpoint: decoyEndpoint, odinEndpoint, browserRuntime: "chromium", headlessRuntime: "csharp", @@ -136,21 +149,31 @@ try { if (browser) await browser.close().catch(() => undefined); if (headless) await stop(headless.process); if (provider) await stop(provider.process); + if (decoyProvider) await stop(decoyProvider.process); if (replacementProvider) await stop(replacementProvider.process); if (odin) await stop(odin.process); if (httpServer) await new Promise(resolve => httpServer.close(resolve)); await rm(workRoot, { recursive: true, force: true }); } -async function startProvider(port) { - return startDotnet(["provider", "--port", String(port), "--state", statePath, "--token", token]); +async function startProvider(port, options = {}) { + return startDotnet([ + "provider", "--port", String(port), + "--state", options.statePath ?? statePath, + "--authority-runtime-id", options.authorityRuntimeId ?? "sample.counter-daemon", + "--route-generation", options.routeGeneration ?? "sample-counter-route-1", + "--token", token, + ]); } -async function startOdin(port, providerEndpoint) { +async function startOdin(port, providerEndpoint, decoyEndpoint) { return startDotnet([ "odin", "--port", String(port), "--provider-endpoint", providerEndpoint, + "--decoy-endpoint", decoyEndpoint, + "--decoy-authority-runtime-id", "sample.decoy-daemon", + "--decoy-route-generation", "sample-decoy-route-1", "--token", token, ]); } diff --git a/src/GameCult.Mesh.Quic.Native/CultMeshNativeQuicRealtimeTransport.cs b/src/GameCult.Mesh.Quic.Native/CultMeshNativeQuicRealtimeTransport.cs index d687d5b6..cb356044 100644 --- a/src/GameCult.Mesh.Quic.Native/CultMeshNativeQuicRealtimeTransport.cs +++ b/src/GameCult.Mesh.Quic.Native/CultMeshNativeQuicRealtimeTransport.cs @@ -42,11 +42,11 @@ namespace GameCult.Mesh.Quic.Native public async Task ConnectAsync( CultMeshTransportCandidate candidate, - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, CancellationToken cancellationToken = default) { if (candidate == null) throw new ArgumentNullException(nameof(candidate)); - if (endpointId == null) throw new ArgumentNullException(nameof(endpointId)); + if (target == null) throw new ArgumentNullException(nameof(target)); if (!TryParseEndpoint(candidate.Endpoint, out var host, out var port, out var pin)) throw new NotSupportedException($"Native QUIC connector does not support '{candidate.Endpoint}'."); @@ -72,7 +72,9 @@ namespace GameCult.Mesh.Quic.Native var transport = new CultMeshNativeQuicRealtimeTransport( candidate.Endpoint, handle, - _options.PollInterval); + _options.PollInterval, + target, + candidate.Generation); try { await transport.WaitUntilConnectedAsync(_options.HandshakeTimeout, cancellationToken).ConfigureAwait(false); @@ -122,23 +124,42 @@ namespace GameCult.Mesh.Quic.Native } } - internal sealed class CultMeshNativeQuicRealtimeTransport : ICultMeshRealtimeTransport + internal sealed class CultMeshNativeQuicRealtimeTransport : ICultMeshRealtimeTransport, ICultMeshVerifiedTransport { private readonly object _gate = new object(); private readonly TimeSpan _pollInterval; private IntPtr _handle; + private readonly CultMeshSessionTarget _target; + private readonly string _routeGeneration; private bool _disposed; - public CultMeshNativeQuicRealtimeTransport(string endpoint, IntPtr handle, TimeSpan pollInterval) + public CultMeshNativeQuicRealtimeTransport( + string endpoint, + IntPtr handle, + TimeSpan pollInterval, + CultMeshSessionTarget target, + string routeGeneration) { Endpoint = endpoint; _handle = handle; _pollInterval = pollInterval; + _target = target; + _routeGeneration = routeGeneration ?? string.Empty; } public string TransportId => "msquic-native-realtime"; public string Endpoint { get; } + public bool IsVerifiedFor( + string verseId, + string authorityRuntimeId, + string protocolId, + string routeGeneration) => + string.Equals(_target.VerseId, verseId, StringComparison.Ordinal) && + string.Equals(_target.AuthorityRuntimeId, authorityRuntimeId, StringComparison.Ordinal) && + string.Equals(CultMeshProtocols.RealtimeState.Value, protocolId, StringComparison.Ordinal) && + string.Equals(_routeGeneration, routeGeneration, StringComparison.Ordinal); + public Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default) => Task.FromException(new NotSupportedException( "The Unity-compatible native QUIC connector currently owns provider-to-client state only.")); diff --git a/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs b/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs index a0198a32..91af6d7c 100644 --- a/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs +++ b/src/GameCult.Mesh.Quic/CultMeshQuicRealtimeTransport.cs @@ -22,7 +22,7 @@ public sealed class CultMeshQuicRealtimeConnectorOptions /// /// Optional provider-aware certificate validator. Standard TLS validation is used when omitted. /// - public Func? + public Func? ValidateProviderCertificate { get; set; } } @@ -55,11 +55,11 @@ public sealed class CultMeshQuicRealtimeTransportConnector : ICultMeshRealtimeTr /// public async Task ConnectAsync( CultMeshTransportCandidate candidate, - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, CancellationToken cancellationToken = default) { if (candidate == null) throw new ArgumentNullException(nameof(candidate)); - if (endpointId == null) throw new ArgumentNullException(nameof(endpointId)); + if (target == null) throw new ArgumentNullException(nameof(target)); if (!QuicConnection.IsSupported) throw new PlatformNotSupportedException("QUIC is unavailable. Install a supported MsQuic runtime."); if (!TryParseEndpoint(candidate.Endpoint, out var host, out var port)) @@ -84,14 +84,15 @@ public sealed class CultMeshQuicRealtimeTransportConnector : ICultMeshRealtimeTr { var providerCertificate = certificate as X509Certificate2 ?? (certificate == null ? null : new X509Certificate2(certificate)); - return validator?.Invoke(endpointId, providerCertificate, chain, errors) ?? + return validator?.Invoke(CultMeshRuntimeId.Parse(target.AuthorityRuntimeId), providerCertificate, chain, errors) ?? errors == SslPolicyErrors.None || MatchesAdvertisedCertificatePin(candidate.Endpoint, providerCertificate); } } }, cancellationToken).ConfigureAwait(false); - return new CultMeshQuicRealtimeTransport(candidate.Endpoint, connection); + return new CultMeshQuicRealtimeTransport( + candidate.Endpoint, connection, target: target, routeGeneration: candidate.Generation); } private static bool TryParseEndpoint(string? endpoint, out string host, out int port) @@ -308,7 +309,7 @@ public sealed class CultMeshQuicRealtimeServer : IAsyncDisposable } } -internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport +internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport, ICultMeshVerifiedTransport { private readonly QuicConnection _connection; private readonly CancellationTokenSource _shutdown = new(); @@ -318,6 +319,8 @@ internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport private readonly SemaphoreSlim _reliableSendGate = new(1, 1); private readonly CultMeshRealtimeInbox _outbound = new(); private readonly Action? _observer; + private readonly CultMeshSessionTarget? _target; + private readonly string _routeGeneration; private readonly Task _acceptLoop; private readonly Task _outboundLoop; private readonly Task _completion; @@ -328,11 +331,15 @@ internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport public CultMeshQuicRealtimeTransport( string endpoint, QuicConnection connection, - Action? observer = null) + Action? observer = null, + CultMeshSessionTarget? target = null, + string routeGeneration = "") { Endpoint = endpoint; _connection = connection; _observer = observer; + _target = target; + _routeGeneration = routeGeneration ?? string.Empty; _acceptLoop = AcceptStreamsAsync(_shutdown.Token); _outboundLoop = SendPublishedFramesAsync(_shutdown.Token); _completion = Task.WhenAll(_acceptLoop, _outboundLoop); @@ -342,6 +349,17 @@ internal sealed class CultMeshQuicRealtimeTransport : ICultMeshRealtimeTransport public string Endpoint { get; } public Task Completion => _completion; + public bool IsVerifiedFor( + string verseId, + string authorityRuntimeId, + string protocolId, + string routeGeneration) => + _target != null && + string.Equals(_target.VerseId, verseId, StringComparison.Ordinal) && + string.Equals(_target.AuthorityRuntimeId, authorityRuntimeId, StringComparison.Ordinal) && + string.Equals(CultMeshProtocols.RealtimeState.Value, protocolId, StringComparison.Ordinal) && + string.Equals(_routeGeneration, routeGeneration, StringComparison.Ordinal); + internal bool TryPublishLatest(CultMeshRealtimeFrame frame) { if (_disposed || frame.Delivery != CultMeshRealtimeDelivery.LatestOnly) return false; diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs index b4667f9c..19a01c54 100644 --- a/src/GameCult.Mesh/CultMeshClient.cs +++ b/src/GameCult.Mesh/CultMeshClient.cs @@ -391,7 +391,8 @@ namespace GameCult.Mesh PayloadSchema = requestSchema.Trim(), PayloadEncoding = "messagepack-base64", Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(request, CultNetSchemaMessageSerialization.Options)), - SourceRuntimeId = sourceRuntimeId.Trim() + SourceRuntimeId = sourceRuntimeId.Trim(), + TargetRuntimeId = target.AuthorityRuntimeId }; using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); timeout.CancelAfter(_operationResponseTimeout); @@ -441,6 +442,11 @@ namespace GameCult.Mesh if (!string.Equals(envelope.ServiceId, expectedService, StringComparison.Ordinal) || !string.Equals(envelope.Operation, expectedOperation, StringComparison.Ordinal)) throw new InvalidOperationException("CultMesh operation response did not match the requested service and operation."); + if (!string.Equals(envelope.SourceRuntimeId, target.AuthorityRuntimeId, StringComparison.Ordinal)) + throw new CultMeshSessionException(new CultMeshSessionFailure( + CultMeshSessionFailureReason.Authority, + $"CultMesh operation response came from runtime '{envelope.SourceRuntimeId ?? "(missing)"}', " + + $"not selected authority '{target.AuthorityRuntimeId}'.")); if (string.Equals(envelope.PayloadSchema, CultNetOperationServer.FailureSchemaId, StringComparison.Ordinal)) { if (!string.Equals(envelope.PayloadEncoding, "messagepack-base64", StringComparison.Ordinal)) @@ -795,7 +801,7 @@ namespace GameCult.Mesh _cache, new CultRecordKey(recordKey), CultMesh.Verse(target.VerseId, "cultmesh-client").Context, - routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.ProviderRuntimeId)); + routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.AuthorityRuntimeId)); } public CultMeshDocumentHandle Handle { get; } @@ -897,7 +903,7 @@ namespace GameCult.Mesh _subscription = new CultNetDatabaseSubscriptionClient(session.OpenSchemaClient(), _cache, networkRegistry); Handle = CultMesh.Collection( _cache, - routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.ProviderRuntimeId)); + routeHint: new CultMeshRouteHint(CultMeshLocalityKind.Network, target.AuthorityRuntimeId)); } public CultMeshCollectionHandle Handle { get; } diff --git a/src/GameCult.Mesh/CultMeshContentSessions.cs b/src/GameCult.Mesh/CultMeshContentSessions.cs index 51570c54..4c562450 100644 --- a/src/GameCult.Mesh/CultMeshContentSessions.cs +++ b/src/GameCult.Mesh/CultMeshContentSessions.cs @@ -122,31 +122,56 @@ namespace GameCult.Mesh public async Task ConnectAsync( CultMeshTransportCandidate candidate, - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, CancellationToken cancellationToken = default) { if (!CanConnect(candidate)) throw new NotSupportedException($"Legacy RUDP content connector does not support '{candidate.Endpoint}'."); var client = await _schemaConnector.ConnectAsync(candidate, CultMeshProtocols.Content, cancellationToken) .ConfigureAwait(false); - return new LegacyRudpContentTransport(candidate.Endpoint, client, _options); + try + { + await CultMeshSessionIdentityClient.VerifyAsync( + client, + "cultmesh-content-client", + target, + CultMeshProtocols.Content, + candidate.Generation, + _options.ResponseTimeout, + cancellationToken).ConfigureAwait(false); + return new LegacyRudpContentTransport( + candidate.Endpoint, client, _options, target, candidate.Generation); + } + catch + { + client.Dispose(); + throw; + } } - private sealed class LegacyRudpContentTransport : ICultMeshContentTransport + private sealed class LegacyRudpContentTransport : ICultMeshContentTransport, ICultMeshVerifiedTransport { private readonly ICultNetSchemaClient _client; private readonly CultMeshLegacyRudpContentTransportOptions _options; + private readonly string _verseId; + private readonly string _authorityRuntimeId; + private readonly string _routeGeneration; private readonly System.Collections.Concurrent.ConcurrentDictionary> _pending = new(StringComparer.Ordinal); private bool _disposed; public LegacyRudpContentTransport( string endpoint, ICultNetSchemaClient client, - CultMeshLegacyRudpContentTransportOptions options) + CultMeshLegacyRudpContentTransportOptions options, + CultMeshSessionTarget target, + string routeGeneration) { Endpoint = endpoint; _client = client; _options = options; + _verseId = target.VerseId; + _authorityRuntimeId = target.AuthorityRuntimeId; + _routeGeneration = routeGeneration ?? string.Empty; _client.OnCultNet(OnResponse); _client.OnCultNet(OnError); } @@ -154,6 +179,16 @@ namespace GameCult.Mesh public string TransportId => "legacy-rudp-content"; public string Endpoint { get; } + public bool IsVerifiedFor( + string verseId, + string authorityRuntimeId, + string protocolId, + string routeGeneration) => + string.Equals(_verseId, verseId, StringComparison.Ordinal) && + string.Equals(_authorityRuntimeId, authorityRuntimeId, StringComparison.Ordinal) && + string.Equals(CultMeshProtocols.Content.Value, protocolId, StringComparison.Ordinal) && + string.Equals(_routeGeneration, routeGeneration, StringComparison.Ordinal); + public async Task CopyChunkToAsync( CultMeshCdnChunkRef chunk, Stream destination, diff --git a/src/GameCult.Mesh/CultMeshContentTransports.cs b/src/GameCult.Mesh/CultMeshContentTransports.cs index 3e2c54b1..891c221a 100644 --- a/src/GameCult.Mesh/CultMeshContentTransports.cs +++ b/src/GameCult.Mesh/CultMeshContentTransports.cs @@ -36,7 +36,7 @@ namespace GameCult.Mesh bool CanConnect(CultMeshTransportCandidate candidate); Task ConnectAsync( CultMeshTransportCandidate candidate, - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, CancellationToken cancellationToken = default); } diff --git a/src/GameCult.Mesh/CultMeshDiscoveryService.cs b/src/GameCult.Mesh/CultMeshDiscoveryService.cs index 93ea6a84..3f108e03 100644 --- a/src/GameCult.Mesh/CultMeshDiscoveryService.cs +++ b/src/GameCult.Mesh/CultMeshDiscoveryService.cs @@ -267,7 +267,8 @@ namespace GameCult.Mesh var observations = results.SelectMany(result => result.Observations) .Where(observation => query.VerseIds.Count == 0 || query.VerseIds.Contains(observation.Candidate.VerseId, StringComparer.Ordinal)) .Where(observation => query.AuthorityRuntimeId == null || - observation.Candidate.AuthorityRuntimeIds.Contains(query.AuthorityRuntimeId, StringComparer.Ordinal)) + observation.Candidate.AuthorityRoutes.Any(route => + string.Equals(route.AuthorityRuntimeId, query.AuthorityRuntimeId, StringComparison.Ordinal))) .ToArray(); foreach (var rejected in observations.Where(observation => observation.Trust == CultMeshDiscoveryTrust.Rejected)) diff --git a/src/GameCult.Mesh/CultMeshRealtimeTransports.cs b/src/GameCult.Mesh/CultMeshRealtimeTransports.cs index 7b32ad6a..d37b7381 100644 --- a/src/GameCult.Mesh/CultMeshRealtimeTransports.cs +++ b/src/GameCult.Mesh/CultMeshRealtimeTransports.cs @@ -59,7 +59,7 @@ namespace GameCult.Mesh bool CanConnect(CultMeshTransportCandidate candidate); Task ConnectAsync( CultMeshTransportCandidate candidate, - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, CancellationToken cancellationToken = default); } diff --git a/src/GameCult.Mesh/CultMeshSessionIdentity.cs b/src/GameCult.Mesh/CultMeshSessionIdentity.cs new file mode 100644 index 00000000..bb17a3d8 --- /dev/null +++ b/src/GameCult.Mesh/CultMeshSessionIdentity.cs @@ -0,0 +1,213 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using GameCult.Networking; + +namespace GameCult.Mesh +{ + /// + /// Optional transport proof surfaced by connectors that already authenticate + /// the peer during their native handshake. Other schema clients must complete + /// the portable CultMesh handshake before use. + /// + public interface ICultMeshVerifiedSchemaClient + { + bool IsVerifiedFor( + string verseId, + string authorityRuntimeId, + string protocolId, + string routeGeneration); + } + + /// + /// Proof exposed by content and realtime transports after their native or + /// portable handshake has bound the physical channel to an Odin route. + /// + public interface ICultMeshVerifiedTransport + { + bool IsVerifiedFor( + string verseId, + string authorityRuntimeId, + string protocolId, + string routeGeneration); + } + + internal static class CultMeshTransportIdentity + { + public static void RequireVerified( + object transport, + CultMeshSessionTarget target, + CultMeshProtocolId protocol, + CultMeshTransportCandidate candidate) + { + if (transport is ICultMeshVerifiedTransport verified && + verified.IsVerifiedFor( + target.VerseId, + target.AuthorityRuntimeId, + protocol.Value, + candidate.Generation)) + return; + throw new CultMeshSessionException(new CultMeshSessionFailure( + CultMeshSessionFailureReason.Authority, + $"Transport did not prove CultMesh authority '{target}' for protocol '{protocol}'.", + candidate.Endpoint)); + } + } + + /// + /// Owns the server side of the transport-neutral CultMesh session identity + /// handshake. Application messages are not authority proof; providers attach + /// this gate to every schema transport they advertise. + /// + public sealed class CultMeshSessionIdentityServer : IDisposable + { + private readonly ICultNetSchemaServer _server; + private readonly string _authorityRuntimeId; + private readonly HashSet _verseIds; + private readonly HashSet _protocolIds; + private readonly HashSet _routeGenerations; + private readonly Func _handler; + private bool _disposed; + + public CultMeshSessionIdentityServer( + ICultNetSchemaServer server, + string authorityRuntimeId, + IEnumerable verseIds, + IEnumerable protocolIds, + IEnumerable routeGenerations) + { + _server = server ?? throw new ArgumentNullException(nameof(server)); + _authorityRuntimeId = Require(authorityRuntimeId, nameof(authorityRuntimeId)); + _verseIds = Clean(verseIds, nameof(verseIds)); + _protocolIds = Clean(protocolIds, nameof(protocolIds)); + _routeGenerations = Clean(routeGenerations, nameof(routeGenerations)); + _handler = HandleAsync; + _server.OnCultNet(_handler); + } + + public void Dispose() + { + if (_disposed) return; + _disposed = true; + _server.RemoveCultNetMessageListener(_handler); + } + + private Task HandleAsync(CultMeshSessionOpenMessage request, ICultNetSchemaServerPeer peer) + { + var error = Validate(request); + peer.SendCultNet(new CultMeshSessionAcceptedMessage + { + MessageId = request.MessageId ?? string.Empty, + Accepted = error == null, + VerseId = request.VerseId ?? string.Empty, + AuthorityRuntimeId = _authorityRuntimeId, + ProtocolId = request.ProtocolId ?? string.Empty, + RouteGeneration = request.RouteGeneration ?? string.Empty, + Error = error + }); + return Task.CompletedTask; + } + + private string? Validate(CultMeshSessionOpenMessage request) + { + if (string.IsNullOrWhiteSpace(request.MessageId)) return "session-message-id-required"; + if (string.IsNullOrWhiteSpace(request.SourceRuntimeId)) return "source-runtime-id-required"; + if (!string.Equals(request.AuthorityRuntimeId, _authorityRuntimeId, StringComparison.Ordinal)) + return "target-runtime-mismatch"; + if (!_verseIds.Contains(request.VerseId ?? string.Empty)) return "verse-not-served"; + if (!_protocolIds.Contains(request.ProtocolId ?? string.Empty)) return "protocol-not-served"; + if (!_routeGenerations.Contains(request.RouteGeneration ?? string.Empty)) return "route-generation-not-served"; + return null; + } + + private static HashSet Clean(IEnumerable values, string name) + { + if (values == null) throw new ArgumentNullException(name); + var result = new HashSet( + values.Where(value => !string.IsNullOrWhiteSpace(value)).Select(value => value.Trim()), + StringComparer.Ordinal); + if (result.Count == 0) throw new ArgumentException("At least one identity is required.", name); + return result; + } + + private static string Require(string value, string name) => + string.IsNullOrWhiteSpace(value) + ? throw new ArgumentException("Value must be non-empty.", name) + : value.Trim(); + } + + internal static class CultMeshSessionIdentityClient + { + public static async Task VerifyAsync( + ICultNetSchemaClient client, + string sourceRuntimeId, + CultMeshSessionTarget target, + CultMeshProtocolId protocol, + string routeGeneration, + TimeSpan timeout, + CancellationToken cancellationToken = default) + { + if (client == null) throw new ArgumentNullException(nameof(client)); + if (client is ICultMeshVerifiedSchemaClient verified) + { + if (!verified.IsVerifiedFor( + target.VerseId, + target.AuthorityRuntimeId, + protocol.Value, + routeGeneration ?? string.Empty)) + { + throw new CultMeshSessionException(new CultMeshSessionFailure( + CultMeshSessionFailureReason.Authority, + $"Native transport identity did not match CultMesh target '{target}' for '{protocol}'.")); + } + return; + } + var messageId = Guid.NewGuid().ToString("N"); + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + client.OnCultNet(response => + { + if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) + completion.TrySetResult(response); + }); + client.SendCultNet(new CultMeshSessionOpenMessage + { + MessageId = messageId, + SourceRuntimeId = sourceRuntimeId, + VerseId = target.VerseId, + AuthorityRuntimeId = target.AuthorityRuntimeId, + ProtocolId = protocol.Value, + RouteGeneration = routeGeneration ?? string.Empty + }); + + using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + deadline.CancelAfter(timeout); + var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using (deadline.Token.Register(() => cancelled.TrySetResult(true))) + { + if (await Task.WhenAny(completion.Task, cancelled.Task).ConfigureAwait(false) != completion.Task) + { + cancellationToken.ThrowIfCancellationRequested(); + throw new CultMeshSessionException(new CultMeshSessionFailure( + CultMeshSessionFailureReason.Timeout, + $"Timed out verifying authority runtime '{target.AuthorityRuntimeId}'.")); + } + } + + var accepted = await completion.Task.ConfigureAwait(false); + if (!accepted.Accepted || + !string.Equals(accepted.VerseId, target.VerseId, StringComparison.Ordinal) || + !string.Equals(accepted.AuthorityRuntimeId, target.AuthorityRuntimeId, StringComparison.Ordinal) || + !string.Equals(accepted.ProtocolId, protocol.Value, StringComparison.Ordinal) || + !string.Equals(accepted.RouteGeneration, routeGeneration ?? string.Empty, StringComparison.Ordinal)) + { + throw new CultMeshSessionException(new CultMeshSessionFailure( + CultMeshSessionFailureReason.Authority, + $"Route did not prove CultMesh authority '{target}' for protocol '{protocol}': " + + (accepted.Error ?? "identity-mismatch"))); + } + } + } +} diff --git a/src/GameCult.Mesh/CultMeshSessions.cs b/src/GameCult.Mesh/CultMeshSessions.cs index c562de83..7db6f06b 100644 --- a/src/GameCult.Mesh/CultMeshSessions.cs +++ b/src/GameCult.Mesh/CultMeshSessions.cs @@ -9,14 +9,14 @@ using R3; namespace GameCult.Mesh { - public sealed class CultMeshEndpointId : IEquatable + public sealed class CultMeshRuntimeId : IEquatable { - private CultMeshEndpointId(string value) { Value = value; } + private CultMeshRuntimeId(string value) { Value = value; } public string Value { get; } - public static CultMeshEndpointId Parse(string value) => new CultMeshEndpointId( - string.IsNullOrWhiteSpace(value) ? throw new FormatException("CultMesh endpoint identity must be non-empty.") : value.Trim()); - public bool Equals(CultMeshEndpointId? other) => other != null && string.Equals(Value, other.Value, StringComparison.Ordinal); - public override bool Equals(object? obj) => Equals(obj as CultMeshEndpointId); + public static CultMeshRuntimeId Parse(string value) => new CultMeshRuntimeId( + string.IsNullOrWhiteSpace(value) ? throw new FormatException("CultMesh runtime identity must be non-empty.") : value.Trim()); + public bool Equals(CultMeshRuntimeId? other) => other != null && string.Equals(Value, other.Value, StringComparison.Ordinal); + public override bool Equals(object? obj) => Equals(obj as CultMeshRuntimeId); public override int GetHashCode() => StringComparer.Ordinal.GetHashCode(Value); public override string ToString() => Value; } @@ -28,31 +28,32 @@ namespace GameCult.Mesh public sealed class CultMeshSessionTarget : IEquatable { /// Creates one stable provider target inside one Verse. - public CultMeshSessionTarget(string verseId, string providerRuntimeId) + public CultMeshSessionTarget(string verseId, string authorityRuntimeId) { VerseId = string.IsNullOrWhiteSpace(verseId) ? throw new ArgumentException("Verse identity is required.", nameof(verseId)) : verseId.Trim(); - ProviderEndpointId = CultMeshEndpointId.Parse(providerRuntimeId); - ProviderRuntimeId = ProviderEndpointId.Value; + AuthorityRuntimeId = CultMeshRuntimeId.Parse(authorityRuntimeId).Value; } /// Gets the stable Verse identity. public string VerseId { get; } - /// Gets the stable provider runtime identity advertised by that Verse. - public string ProviderRuntimeId { get; } - internal CultMeshEndpointId ProviderEndpointId { get; } - internal string SessionKey => VerseId + "\u001f" + ProviderRuntimeId; + /// Gets the stable authority runtime identity advertised by that Verse. + public string AuthorityRuntimeId { get; } + /// Compatibility projection. Use . + [Obsolete("Use AuthorityRuntimeId. Provider identity is not a physical endpoint.")] + public string ProviderRuntimeId => AuthorityRuntimeId; + internal string SessionKey => VerseId + "\u001f" + AuthorityRuntimeId; public bool Equals(CultMeshSessionTarget? other) => other != null && string.Equals(VerseId, other.VerseId, StringComparison.Ordinal) && - string.Equals(ProviderRuntimeId, other.ProviderRuntimeId, StringComparison.Ordinal); + string.Equals(AuthorityRuntimeId, other.AuthorityRuntimeId, StringComparison.Ordinal); public override bool Equals(object? obj) => Equals(obj as CultMeshSessionTarget); public override int GetHashCode() => HashCode.Combine( StringComparer.Ordinal.GetHashCode(VerseId), - StringComparer.Ordinal.GetHashCode(ProviderRuntimeId)); - public override string ToString() => VerseId + "/" + ProviderRuntimeId; + StringComparer.Ordinal.GetHashCode(AuthorityRuntimeId)); + public override string ToString() => VerseId + "/" + AuthorityRuntimeId; } public sealed class CultMeshProtocolId : IEquatable @@ -80,15 +81,24 @@ namespace GameCult.Mesh public sealed class CultMeshTransportCandidate { - public CultMeshTransportCandidate(string endpoint, CultMeshTransportPathKind pathKind = CultMeshTransportPathKind.Direct, int priority = 0) + public CultMeshTransportCandidate( + string endpoint, + CultMeshTransportPathKind pathKind = CultMeshTransportPathKind.Direct, + int priority = 0, + string authorityRuntimeId = "", + string generation = "") { Endpoint = string.IsNullOrWhiteSpace(endpoint) ? throw new ArgumentException("Physical endpoint is required.", nameof(endpoint)) : endpoint; PathKind = pathKind; Priority = priority; + AuthorityRuntimeId = authorityRuntimeId?.Trim() ?? string.Empty; + Generation = generation?.Trim() ?? string.Empty; } public string Endpoint { get; } public CultMeshTransportPathKind PathKind { get; } public int Priority { get; } + public string AuthorityRuntimeId { get; } + public string Generation { get; } } public enum CultMeshSessionStatus { Connecting, Online, Degraded, Reconnecting, Offline } @@ -522,6 +532,8 @@ namespace GameCult.Mesh public ICultMeshClock Clock { get; set; } = CultMeshSystemClock.Instance; public ICultMeshDiagnosticSink Diagnostics { get; set; } = CultMeshNullDiagnosticSink.Instance; public int MaxRacedCandidates { get; set; } = 2; + public string RuntimeId { get; set; } = "cultmesh-client"; + public TimeSpan IdentityHandshakeTimeout { get; set; } = TimeSpan.FromSeconds(5); } public sealed class CultMeshSessionManager : IDisposable @@ -581,6 +593,8 @@ namespace GameCult.Mesh throw new ArgumentException("Realtime transport connectors cannot contain null entries.", nameof(realtimeConnectors)); _options = options ?? new CultMeshSessionManagerOptions(); if (_options.MaxRacedCandidates <= 0) throw new ArgumentOutOfRangeException(nameof(options)); + if (string.IsNullOrWhiteSpace(_options.RuntimeId)) throw new ArgumentException("Client runtime identity is required.", nameof(options)); + if (_options.IdentityHandshakeTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(options)); } public async Task ConnectAsync( @@ -661,14 +675,10 @@ namespace GameCult.Mesh "No streaming content connectors are configured. Register the TCP content connector or the explicit legacy RUDP connector."); var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery( - target.ProviderRuntimeId, + target.AuthorityRuntimeId, new[] { target.VerseId }, - target.ProviderRuntimeId)).ConfigureAwait(false); - var candidates = discovery.Candidates - .SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints) - .Distinct(StringComparer.Ordinal) - .Select(endpoint => new CultMeshTransportCandidate(endpoint)) - .ToArray(); + target.AuthorityRuntimeId)).ConfigureAwait(false); + var candidates = BoundCandidates(discovery, target, CultMeshProtocols.Content); var routes = candidates .SelectMany(candidate => _contentConnectors .Where(connector => connector.CanConnect(candidate)) @@ -685,7 +695,7 @@ namespace GameCult.Mesh foreach (var tier in routes.GroupBy(route => route.Connector.Priority).OrderBy(group => group.Key)) { var attempts = tier.Take(_options.MaxRacedCandidates) - .Select(route => ConnectContentCandidateAsync(target.ProviderEndpointId, route)) + .Select(route => ConnectContentCandidateAsync(target, route)) .ToList(); while (attempts.Count > 0) { @@ -709,18 +719,33 @@ namespace GameCult.Mesh } var last = failures.LastOrDefault(); + var reason = last is CultMeshSessionException typed + ? typed.Failure.Reason + : last is TimeoutException + ? CultMeshSessionFailureReason.Timeout + : CultMeshSessionFailureReason.Transport; throw Failure( - last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport, + reason, $"No streaming content path connected for '{target}'.", last); } private static async Task ConnectContentCandidateAsync( - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, ContentRoute route) { - var transport = await route.Connector.ConnectAsync(route.Candidate, endpointId).ConfigureAwait(false); - return new ConnectedContentPath(route.Candidate, transport); + var transport = await route.Connector.ConnectAsync(route.Candidate, target).ConfigureAwait(false); + try + { + CultMeshTransportIdentity.RequireVerified( + transport, target, CultMeshProtocols.Content, route.Candidate); + return new ConnectedContentPath(route.Candidate, transport); + } + catch + { + transport.Dispose(); + throw; + } } private void InvalidateContentSession(string key, CultMeshContentSession session) @@ -790,13 +815,10 @@ namespace GameCult.Mesh "No realtime state connectors are configured. Register a QUIC connector explicitly."); var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery( - target.ProviderRuntimeId, + target.AuthorityRuntimeId, new[] { target.VerseId }, - target.ProviderRuntimeId)).ConfigureAwait(false); - var routes = discovery.Candidates - .SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints) - .Distinct(StringComparer.Ordinal) - .Select(endpoint => new CultMeshTransportCandidate(endpoint)) + target.AuthorityRuntimeId)).ConfigureAwait(false); + var routes = BoundCandidates(discovery, target, CultMeshProtocols.RealtimeState) .SelectMany(candidate => _realtimeConnectors .Where(connector => connector.CanConnect(candidate)) .Select(connector => new RealtimeRoute(candidate, connector))) @@ -812,7 +834,7 @@ namespace GameCult.Mesh foreach (var tier in routes.GroupBy(route => route.Connector.Priority).OrderBy(group => group.Key)) { var attempts = tier.Take(_options.MaxRacedCandidates) - .Select(route => ConnectRealtimeCandidateAsync(target.ProviderEndpointId, route)) + .Select(route => ConnectRealtimeCandidateAsync(target, route)) .ToList(); while (attempts.Count > 0) { @@ -836,18 +858,33 @@ namespace GameCult.Mesh } var last = failures.LastOrDefault(); + var reason = last is CultMeshSessionException typed + ? typed.Failure.Reason + : last is TimeoutException + ? CultMeshSessionFailureReason.Timeout + : CultMeshSessionFailureReason.Transport; throw Failure( - last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport, + reason, $"No realtime state path connected for '{target}'.", last); } private static async Task ConnectRealtimeCandidateAsync( - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, RealtimeRoute route) { - var transport = await route.Connector.ConnectAsync(route.Candidate, endpointId).ConfigureAwait(false); - return new ConnectedRealtimePath(route.Candidate, transport); + var transport = await route.Connector.ConnectAsync(route.Candidate, target).ConfigureAwait(false); + try + { + CultMeshTransportIdentity.RequireVerified( + transport, target, CultMeshProtocols.RealtimeState, route.Candidate); + return new ConnectedRealtimePath(route.Candidate, transport); + } + catch + { + transport.Dispose(); + throw; + } } private void InvalidateRealtimeSession(string key, CultMeshRealtimeSession session) @@ -880,13 +917,10 @@ namespace GameCult.Mesh private async Task ResolveConnectedPathAsync(CultMeshSessionTarget target, CultMeshProtocolId protocol) { var discovery = await _discovery.ResolveAsync(new CultMeshDiscoveryQuery( - target.ProviderRuntimeId, + target.AuthorityRuntimeId, new[] { target.VerseId }, - target.ProviderRuntimeId)).ConfigureAwait(false); - var candidates = discovery.Candidates.SelectMany(candidate => candidate.Descriptor.DiscoveryEndpoints) - .Distinct(StringComparer.Ordinal) - .Select(endpoint => new CultMeshTransportCandidate(endpoint)) - .ToArray(); + target.AuthorityRuntimeId)).ConfigureAwait(false); + var candidates = BoundCandidates(discovery, target, protocol); var routes = candidates .SelectMany(candidate => _connectors .Where(connector => connector.CanConnect(candidate)) @@ -908,7 +942,7 @@ namespace GameCult.Mesh foreach (var tier in routes.GroupBy(route => route.Connector.Priority).OrderBy(group => group.Key)) { var attempts = tier.Take(_options.MaxRacedCandidates) - .Select(route => ConnectCandidateAsync(route, protocol)) + .Select(route => ConnectCandidateAsync(target, route, protocol)) .ToList(); while (attempts.Count > 0) { @@ -925,7 +959,12 @@ namespace GameCult.Mesh } } var last = failures.LastOrDefault(); - throw Failure(last is TimeoutException ? CultMeshSessionFailureReason.Timeout : CultMeshSessionFailureReason.Transport, + var reason = last is CultMeshSessionException typed + ? typed.Failure.Reason + : last is TimeoutException + ? CultMeshSessionFailureReason.Timeout + : CultMeshSessionFailureReason.Transport; + throw Failure(reason, $"No transport path connected for '{target}' and protocol '{protocol}'.", last); } @@ -960,9 +999,9 @@ namespace GameCult.Mesh try { _discovery.Invalidate(new CultMeshDiscoveryQuery( - session.Target.ProviderRuntimeId, + session.Target.AuthorityRuntimeId, new[] { session.Target.VerseId }, - session.Target.ProviderRuntimeId)); + session.Target.AuthorityRuntimeId)); var replacement = await ResolveConnectedPathAsync(session.Target, session.Protocol).ConfigureAwait(false); session.ReplacePhysicalChannel(replacement.Client); session.Transition(new CultMeshSessionState( @@ -987,10 +1026,28 @@ namespace GameCult.Mesh } } - private static async Task ConnectCandidateAsync(SchemaRoute route, CultMeshProtocolId protocol) + private async Task ConnectCandidateAsync( + CultMeshSessionTarget target, + SchemaRoute route, + CultMeshProtocolId protocol) { var client = await route.Connector.ConnectAsync(route.Candidate, protocol).ConfigureAwait(false); - return new ConnectedPath(route.Candidate, client); + try + { + await CultMeshSessionIdentityClient.VerifyAsync( + client, + _options.RuntimeId, + target, + protocol, + route.Candidate.Generation, + _options.IdentityHandshakeTimeout).ConfigureAwait(false); + return new ConnectedPath(route.Candidate, client); + } + catch + { + client.Dispose(); + throw; + } } private static async Task AwaitForCallerAsync(Task shared, CancellationToken cancellationToken) @@ -1008,12 +1065,36 @@ namespace GameCult.Mesh private CultMeshSessionException Failure(CultMeshSessionFailureReason reason, string message, Exception? inner = null) => new CultMeshSessionException(new CultMeshSessionFailure(reason, message), inner); + private static CultMeshTransportCandidate[] BoundCandidates( + CultMeshDiscoveryState discovery, + CultMeshSessionTarget target, + CultMeshProtocolId protocol) + { + return discovery.Candidates + .SelectMany(candidate => candidate.Descriptor.AuthorityRoutes) + .Where(route => string.Equals( + route.AuthorityRuntimeId, + target.AuthorityRuntimeId, + StringComparison.Ordinal)) + .Where(route => route.Supports(protocol)) + .GroupBy(route => route.Endpoint, StringComparer.Ordinal) + .Select(group => group.OrderBy(route => route.Priority).First()) + .Select(route => new CultMeshTransportCandidate( + route.Endpoint, + priority: route.Priority, + authorityRuntimeId: route.AuthorityRuntimeId, + generation: route.Generation)) + .OrderBy(candidate => candidate.Priority) + .ThenBy(candidate => candidate.Endpoint, StringComparer.Ordinal) + .ToArray(); + } + private void Emit(CultMeshSessionTarget target, CultMeshProtocolId protocol, string state, string path) { _options.Diagnostics.Emit(new CultMeshDiagnosticEvent( Interlocked.Increment(ref _diagnosticSequence), _options.Clock.UtcNow, CultMeshReliabilityOrgan.Session, CultMeshDiagnosticKind.PathChanged, - target.SessionKey + ":" + protocol.Value, target.ProviderRuntimeId, state, + target.SessionKey + ":" + protocol.Value, target.AuthorityRuntimeId, state, sourceId: "session-manager", endpoint: path)); } diff --git a/src/GameCult.Mesh/CultMeshSnapshots.cs b/src/GameCult.Mesh/CultMeshSnapshots.cs index cff94fa4..305fed4a 100644 --- a/src/GameCult.Mesh/CultMeshSnapshots.cs +++ b/src/GameCult.Mesh/CultMeshSnapshots.cs @@ -102,7 +102,7 @@ namespace GameCult.Mesh CultNetDocumentRegistry? registry) { _sharedSession = session ?? throw new ArgumentNullException(nameof(session)); - _endpoint = session.State.Path?.Endpoint ?? session.Target.ProviderRuntimeId; + _endpoint = session.State.Path?.Endpoint ?? session.Target.AuthorityRuntimeId; _defaults = CultMesh.CloneSnapshotRequestOptions(options); _registry = registry ?? new CultNetDocumentRegistry(); _client = session.Channel; diff --git a/src/GameCult.Mesh/CultMeshTcpContentTransport.cs b/src/GameCult.Mesh/CultMeshTcpContentTransport.cs index 0c3eb19e..f47f08bd 100644 --- a/src/GameCult.Mesh/CultMeshTcpContentTransport.cs +++ b/src/GameCult.Mesh/CultMeshTcpContentTransport.cs @@ -11,6 +11,21 @@ using GameCult.Networking; namespace GameCult.Mesh { + /// Identity served by one dedicated content listener. + public sealed class CultMeshTcpContentServerOptions + { + public string VerseId { get; set; } = string.Empty; + public string AuthorityRuntimeId { get; set; } = string.Empty; + public string RouteGeneration { get; set; } = string.Empty; + + internal void Validate() + { + if (string.IsNullOrWhiteSpace(VerseId)) throw new InvalidOperationException("Content Verse identity is required."); + if (string.IsNullOrWhiteSpace(AuthorityRuntimeId)) throw new InvalidOperationException("Content authority runtime identity is required."); + if (string.IsNullOrWhiteSpace(RouteGeneration)) throw new InvalidOperationException("Content route generation is required."); + } + } + /// Creates dedicated TCP streams for immutable CultMesh content. public sealed class CultMeshTcpContentTransportConnector : ICultMeshContentTransportConnector { @@ -33,11 +48,11 @@ namespace GameCult.Mesh public async Task ConnectAsync( CultMeshTransportCandidate candidate, - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, CancellationToken cancellationToken = default) { if (candidate == null) throw new ArgumentNullException(nameof(candidate)); - if (endpointId == null) throw new ArgumentNullException(nameof(endpointId)); + if (target == null) throw new ArgumentNullException(nameof(target)); if (!TryParseEndpoint(candidate.Endpoint, out var host, out var port)) throw new NotSupportedException($"TCP content connector does not support '{candidate.Endpoint}'."); @@ -48,8 +63,10 @@ namespace GameCult.Mesh { await CultMeshTcpContentWire.AwaitAsync(client.ConnectAsync(host, port), timeout.Token) .ConfigureAwait(false); + await CultMeshTcpContentWire.VerifyAuthorityAsync( + client.GetStream(), target, candidate, timeout.Token).ConfigureAwait(false); Interlocked.Increment(ref _connectCount); - return new CultMeshTcpContentTransport(candidate.Endpoint, client); + return new CultMeshTcpContentTransport(candidate.Endpoint, client, target, candidate.Generation); } catch { @@ -72,7 +89,7 @@ namespace GameCult.Mesh } } - internal sealed class CultMeshTcpContentTransport : ICultMeshContentTransport + internal sealed class CultMeshTcpContentTransport : ICultMeshContentTransport, ICultMeshVerifiedTransport { private readonly TcpClient _client; private readonly NetworkStream _stream; @@ -80,12 +97,22 @@ namespace GameCult.Mesh private readonly ConcurrentDictionary _pending = new(StringComparer.Ordinal); private readonly CancellationTokenSource _shutdown = new(); private readonly Task _receiveLoop; + private readonly string _verseId; + private readonly string _authorityRuntimeId; + private readonly string _routeGeneration; private bool _disposed; - public CultMeshTcpContentTransport(string endpoint, TcpClient client) + public CultMeshTcpContentTransport( + string endpoint, + TcpClient client, + CultMeshSessionTarget target, + string routeGeneration) { Endpoint = endpoint; _client = client ?? throw new ArgumentNullException(nameof(client)); + _verseId = target?.VerseId ?? throw new ArgumentNullException(nameof(target)); + _authorityRuntimeId = target.AuthorityRuntimeId; + _routeGeneration = routeGeneration ?? string.Empty; _stream = client.GetStream(); _receiveLoop = ReceiveLoopAsync(_shutdown.Token); } @@ -93,6 +120,16 @@ namespace GameCult.Mesh public string TransportId => "tcp-content"; public string Endpoint { get; } + public bool IsVerifiedFor( + string verseId, + string authorityRuntimeId, + string protocolId, + string routeGeneration) => + string.Equals(_verseId, verseId, StringComparison.Ordinal) && + string.Equals(_authorityRuntimeId, authorityRuntimeId, StringComparison.Ordinal) && + string.Equals(CultMeshProtocols.Content.Value, protocolId, StringComparison.Ordinal) && + string.Equals(_routeGeneration, routeGeneration, StringComparison.Ordinal); + public async Task CopyChunkToAsync( CultMeshCdnChunkRef chunk, Stream destination, @@ -212,6 +249,7 @@ namespace GameCult.Mesh private readonly TcpListener _listener; private readonly Func _resolveChunk; private readonly Func _canServeHash; + private readonly CultMeshTcpContentServerOptions _options; private readonly CancellationTokenSource _shutdown = new(); private readonly ConcurrentDictionary _clients = new(); private readonly Task _acceptLoop; @@ -221,8 +259,9 @@ namespace GameCult.Mesh public CultMeshTcpContentServer( TcpListener listener, CultCache content, + CultMeshTcpContentServerOptions options, Func? canServeHash = null) - : this(listener, CreateCacheResolver(content), canServeHash) + : this(listener, CreateCacheResolver(content), options, canServeHash) { } @@ -234,10 +273,13 @@ namespace GameCult.Mesh public CultMeshTcpContentServer( TcpListener listener, Func resolveChunk, + CultMeshTcpContentServerOptions options, Func? canServeHash = null) { _listener = listener ?? throw new ArgumentNullException(nameof(listener)); _resolveChunk = resolveChunk ?? throw new ArgumentNullException(nameof(resolveChunk)); + _options = options ?? throw new ArgumentNullException(nameof(options)); + _options.Validate(); _canServeHash = canServeHash ?? (_ => true); _listener.Start(); _acceptLoop = AcceptLoopAsync(_shutdown.Token); @@ -278,6 +320,7 @@ namespace GameCult.Mesh try { using var stream = client.GetStream(); + if (!await AcceptAuthorityAsync(stream, cancellationToken).ConfigureAwait(false)) return; while (!cancellationToken.IsCancellationRequested) { var requestBytes = await CultMeshTcpContentWire.ReadFrameAsync(stream, cancellationToken).ConfigureAwait(false); @@ -296,6 +339,32 @@ namespace GameCult.Mesh } } + private async Task AcceptAuthorityAsync(Stream stream, CancellationToken cancellationToken) + { + var requestBytes = await CultMeshTcpContentWire.ReadFrameAsync(stream, cancellationToken).ConfigureAwait(false); + var request = CultNetSchemaMessageSerialization.Deserialize(requestBytes) as CultMeshSessionOpenMessage + ?? throw new InvalidDataException("TCP content session did not begin with an authority handshake."); + var accepted = string.Equals(request.VerseId, _options.VerseId, StringComparison.Ordinal) && + string.Equals(request.AuthorityRuntimeId, _options.AuthorityRuntimeId, StringComparison.Ordinal) && + string.Equals(request.ProtocolId, CultMeshProtocols.Content.Value, StringComparison.Ordinal) && + string.Equals(request.RouteGeneration, _options.RouteGeneration, StringComparison.Ordinal); + var response = new CultMeshSessionAcceptedMessage + { + MessageId = request.MessageId, + Accepted = accepted, + VerseId = _options.VerseId, + AuthorityRuntimeId = _options.AuthorityRuntimeId, + ProtocolId = CultMeshProtocols.Content.Value, + RouteGeneration = _options.RouteGeneration, + Error = accepted ? null : "Content session target does not match this authority route." + }; + await CultMeshTcpContentWire.WriteFrameAsync( + stream, + CultNetSchemaMessageSerialization.Serialize(response), + cancellationToken).ConfigureAwait(false); + return accepted; + } + private async Task SendChunkAsync( Stream stream, CultMeshContentChunkRequestMessage request, @@ -391,6 +460,38 @@ namespace GameCult.Mesh await stream.FlushAsync(cancellationToken).ConfigureAwait(false); } + public static async Task VerifyAuthorityAsync( + Stream stream, + CultMeshSessionTarget target, + CultMeshTransportCandidate candidate, + CancellationToken cancellationToken) + { + var request = new CultMeshSessionOpenMessage + { + MessageId = Guid.NewGuid().ToString("N"), + SourceRuntimeId = "cultmesh-content-client", + VerseId = target.VerseId, + AuthorityRuntimeId = target.AuthorityRuntimeId, + ProtocolId = CultMeshProtocols.Content.Value, + RouteGeneration = candidate.Generation + }; + await WriteFrameAsync(stream, CultNetSchemaMessageSerialization.Serialize(request), cancellationToken) + .ConfigureAwait(false); + var responseBytes = await ReadFrameAsync(stream, cancellationToken).ConfigureAwait(false); + var response = CultNetSchemaMessageSerialization.Deserialize(responseBytes) as CultMeshSessionAcceptedMessage + ?? throw new InvalidDataException("TCP content authority handshake returned an unexpected schema."); + if (!string.Equals(response.MessageId, request.MessageId, StringComparison.Ordinal) || + !response.Accepted || + !string.Equals(response.VerseId, target.VerseId, StringComparison.Ordinal) || + !string.Equals(response.AuthorityRuntimeId, target.AuthorityRuntimeId, StringComparison.Ordinal) || + !string.Equals(response.ProtocolId, CultMeshProtocols.Content.Value, StringComparison.Ordinal) || + !string.Equals(response.RouteGeneration, candidate.Generation, StringComparison.Ordinal)) + throw new CultMeshSessionException(new CultMeshSessionFailure( + CultMeshSessionFailureReason.Authority, + response.Error ?? "TCP content endpoint did not prove the selected authority route.", + candidate.Endpoint)); + } + public static async Task ReadFrameAsync(Stream stream, CancellationToken cancellationToken) { var header = new byte[HeaderBytes]; diff --git a/src/GameCult.Mesh/CultMeshVerse.cs b/src/GameCult.Mesh/CultMeshVerse.cs index 547a46ec..55d6deeb 100644 --- a/src/GameCult.Mesh/CultMeshVerse.cs +++ b/src/GameCult.Mesh/CultMeshVerse.cs @@ -91,6 +91,53 @@ namespace GameCult.Mesh } } + /// + /// Binds one authoritative runtime to one physical route. This relationship is + /// the routing authority; the legacy endpoint/runtime lists are projections only. + /// + public sealed class CultMeshAuthorityRoute + { + public CultMeshAuthorityRoute( + string authorityRuntimeId, + string endpoint, + IEnumerable? protocolIds = null, + int priority = 0, + string? generation = null) + { + AuthorityRuntimeId = RequireNonEmpty(authorityRuntimeId, nameof(authorityRuntimeId)); + Endpoint = RequireNonEmpty(endpoint, nameof(endpoint)); + ProtocolIds = Clean(protocolIds); + if (priority < 0) throw new ArgumentOutOfRangeException(nameof(priority)); + Priority = priority; + Generation = string.IsNullOrWhiteSpace(generation) + ? AuthorityRuntimeId + "@" + Endpoint + : generation!.Trim(); + } + + public string AuthorityRuntimeId { get; } + public string Endpoint { get; } + public IReadOnlyList ProtocolIds { get; } + public int Priority { get; } + public string Generation { get; } + + public bool Supports(CultMeshProtocolId protocol) => + protocol != null && + (ProtocolIds.Count == 0 || ProtocolIds.Contains(protocol.Value, StringComparer.Ordinal)); + + private static string[] Clean(IEnumerable? values) => + values?.Where(value => !string.IsNullOrWhiteSpace(value)) + .Select(value => value.Trim()) + .Distinct(StringComparer.Ordinal) + .OrderBy(value => value, StringComparer.Ordinal) + .ToArray() + ?? Array.Empty(); + + private static string RequireNonEmpty(string value, string paramName) => + string.IsNullOrWhiteSpace(value) + ? throw new ArgumentException("Value must be non-empty.", paramName) + : value.Trim(); + } + /// /// Describes one CultMesh Verse: a rule-bearing consensus graph. /// @@ -107,14 +154,36 @@ namespace GameCult.Mesh IEnumerable? discoveryEndpoints = null, IEnumerable? authorityRuntimeIds = null, string? parentVerseId = null, - string? description = null) + string? description = null, + IEnumerable? authorityRoutes = null) { VerseId = RequireNonEmpty(verseId, nameof(verseId)); DisplayName = RequireNonEmpty(displayName, nameof(displayName)); AuthorityModel = authorityModel; Compatibility = compatibility ?? throw new ArgumentNullException(nameof(compatibility)); - DiscoveryEndpoints = Clean(discoveryEndpoints); - AuthorityRuntimeIds = Clean(authorityRuntimeIds); + var legacyEndpoints = Clean(discoveryEndpoints); + var legacyRuntimeIds = Clean(authorityRuntimeIds); + var suppliedRoutes = authorityRoutes?.ToArray(); + if (suppliedRoutes?.Any(route => route == null) == true) + throw new ArgumentException("Authority routes cannot contain null.", nameof(authorityRoutes)); + if (suppliedRoutes != null && (legacyEndpoints.Length > 0 || legacyRuntimeIds.Length > 0)) + throw new ArgumentException( + "Supply bound authority routes or legacy endpoint/runtime lists, never both.", + nameof(authorityRoutes)); + if (suppliedRoutes == null && legacyEndpoints.Length > 0 && legacyRuntimeIds.Length > 1) + throw new InvalidOperationException( + "Legacy Verse routing with multiple authority runtimes is ambiguous. " + + "Publish explicit CultMeshAuthorityRoute bindings."); + + AuthorityRoutes = NormalizeRoutes(suppliedRoutes ?? + (legacyRuntimeIds.Length == 1 + ? legacyEndpoints.Select(endpoint => new CultMeshAuthorityRoute(legacyRuntimeIds[0], endpoint)) + : Array.Empty())); + DiscoveryEndpoints = AuthorityRoutes.Count > 0 + ? AuthorityRoutes.Select(route => route.Endpoint).Distinct(StringComparer.Ordinal).ToArray() + : legacyEndpoints; + AuthorityRuntimeIds = AuthorityRoutes.Select(route => route.AuthorityRuntimeId) + .Distinct(StringComparer.Ordinal).OrderBy(value => value, StringComparer.Ordinal).ToArray(); ParentVerseId = parentVerseId; Description = description; } @@ -143,6 +212,8 @@ namespace GameCult.Mesh /// Gets known authoritative runtime ids, when authority is cluster-shaped. /// public IReadOnlyList AuthorityRuntimeIds { get; } + /// Gets exact authority-to-route bindings used for identity-sensitive resolution. + public IReadOnlyList AuthorityRoutes { get; } /// /// Gets the parent Verse id for subscribed overlays or branches. /// @@ -177,10 +248,29 @@ namespace GameCult.Mesh private static string[] Clean(IEnumerable? values) { - return values?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray() + return values?.Where(value => !string.IsNullOrWhiteSpace(value)) + .Select(value => value.Trim()) + .Distinct(StringComparer.Ordinal) + .ToArray() ?? Array.Empty(); } + private static CultMeshAuthorityRoute[] NormalizeRoutes(IEnumerable routes) + { + return routes + .GroupBy(route => new + { + route.AuthorityRuntimeId, + route.Endpoint, + Protocols = string.Join("\u001f", route.ProtocolIds) + }) + .Select(group => group.OrderBy(route => route.Priority).First()) + .OrderBy(route => route.Priority) + .ThenBy(route => route.AuthorityRuntimeId, StringComparer.Ordinal) + .ThenBy(route => route.Endpoint, StringComparer.Ordinal) + .ToArray(); + } + private static string RequireNonEmpty(string value, string paramName) { if (string.IsNullOrWhiteSpace(value)) diff --git a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs index 801e4716..774a05f9 100644 --- a/src/GameCult.Mesh/CultMeshVerseDiscovery.cs +++ b/src/GameCult.Mesh/CultMeshVerseDiscovery.cs @@ -33,6 +33,14 @@ namespace GameCult.Mesh }, DiscoveryEndpoints = verse.DiscoveryEndpoints.ToArray(), AuthorityRuntimeIds = verse.AuthorityRuntimeIds.ToArray(), + AuthorityRoutes = verse.AuthorityRoutes.Select(route => new CultMeshAuthorityRouteMessage + { + AuthorityRuntimeId = route.AuthorityRuntimeId, + Endpoint = route.Endpoint, + ProtocolIds = route.ProtocolIds.ToArray(), + Priority = route.Priority, + Generation = route.Generation + }).ToArray(), ParentVerseId = verse.ParentVerseId, Description = verse.Description }; @@ -50,6 +58,12 @@ namespace GameCult.Mesh out var parsed) ? parsed : CultMeshVerseAuthorityModel.SubscribedOverlay; + var routes = message.AuthorityRoutes?.Select(route => new CultMeshAuthorityRoute( + route.AuthorityRuntimeId, + route.Endpoint, + route.ProtocolIds, + route.Priority, + route.Generation)).ToArray(); return new CultMeshVerseDescriptor( message.VerseId, message.DisplayName, @@ -60,10 +74,11 @@ namespace GameCult.Mesh message.Compatibility.CompatibleVerseIds, message.Compatibility.RequiredPluginIds, message.Compatibility.OptionalPluginIds), - message.DiscoveryEndpoints, - message.AuthorityRuntimeIds, + routes is { Length: > 0 } ? null : message.DiscoveryEndpoints, + routes is { Length: > 0 } ? null : message.AuthorityRuntimeIds, message.ParentVerseId, - message.Description); + message.Description, + routes is { Length: > 0 } ? routes : null); } /// Creates the canonical filtered Verse catalog response for any CultNet transport adapter. diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index 88074704..df1735b6 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -50,8 +50,12 @@ not the brand surface. ### Verses `CultMeshVerseDescriptor` describes a rule-bearing consensus graph. It includes -transport compatibility, rules hash, authority model, discovery endpoints, -known authority runtimes, optional parent Verse id, and plugin requirements. +transport compatibility, rules hash, authority model, exact authority-route +bindings, optional parent Verse id, and plugin requirements. Each route binds +an authority runtime to one endpoint, protocol set, priority, and generation. +The old endpoint/runtime arrays are derived projections; a legacy descriptor +with several authority runtimes is rejected because it cannot identify which +runtime owns which route. `CultMeshVerseCatalog` is the local reactive catalog for discovered Verses. It can publish discovery updates and find compatible transfer targets. diff --git a/src/GameCult.Mesh/docs/transport-planes.md b/src/GameCult.Mesh/docs/transport-planes.md index 39be9c59..bd42a071 100644 --- a/src/GameCult.Mesh/docs/transport-planes.md +++ b/src/GameCult.Mesh/docs/transport-planes.md @@ -9,7 +9,8 @@ be equally good at files, realtime state, and same-machine memory access. registered connector for each traffic class. A connector owns only its physical connection and byte movement. -**Inputs:** stable Verse/provider identity, advertised physical candidates, +**Inputs:** stable Verse/authority-runtime identity, exact Odin-bound physical +routes with protocol and generation, connector support and priority, caller cancellation, and locality evidence for body negotiation. @@ -63,13 +64,22 @@ operation. transport diagnostics. None of these grant provider authority or become content identity. +Before a schema, content, or realtime session becomes online, the transport +must prove that the connected peer matches the selected Verse, authority +runtime, protocol, and route generation. TCP schema and content lanes use the +portable CultNet session-open handshake. QUIC uses the exact Odin-bound route +plus its TLS endpoint/certificate proof. A connector that returns bytes without +authority proof is rejected; racing every endpoint in a Verse descriptor is +not route selection. + **Forbidden writers:** applications and renderers do not choose unadvertised endpoints, implement reconnect loops, or bypass verification. Schema messages must not carry bulk bodies. RUDP/LiteNetLib cannot become an implicit default for any traffic class. QUIC does not own schemas or immutable files merely because it can transport bytes. -**Shared paths:** discovery and identity resolution remain common. Each traffic +**Shared paths:** exact route resolution and peer verification remain common. +Each traffic class has its own connector set, connection cache, priority tiers, and failure state. All content connectors feed the same transfer and promotion owner. diff --git a/src/GameCult.Mesh/docs/verses.md b/src/GameCult.Mesh/docs/verses.md index 656dcfaf..fbd9a522 100644 --- a/src/GameCult.Mesh/docs/verses.md +++ b/src/GameCult.Mesh/docs/verses.md @@ -21,8 +21,10 @@ The same transport can carry many Verses: subscribed overlay - `Compatibility`: transport version, rules hash, compatible Verse ids, and plugin requirements -- `DiscoveryEndpoints`: endpoints where nodes for the Verse can be found -- `AuthorityRuntimeIds`: known authoritative runtimes when cluster-shaped +- `AuthorityRoutes`: exact authority-runtime, endpoint, protocol, priority, and + generation bindings. This is the routing authority. +- `DiscoveryEndpoints`: read-only endpoint projection for legacy inspection +- `AuthorityRuntimeIds`: read-only authority projection for legacy inspection - `ParentVerseId`: source Verse for overlays or branches ## Compatibility @@ -57,6 +59,13 @@ static config, cloud discovery, peer gossip, or mod portals. It does not care where the descriptor came from. That keeps discovery pluggable while the Verse model stays stable. +An endpoint and an authority id are not two independent facts. A descriptor +with explicit routes binds them. The legacy parallel arrays are accepted only +when exactly one authority runtime exists; a multi-authority legacy descriptor +is ambiguous and fails closed. Session establishment then requires the selected +physical route to prove the exact Verse, authority runtime, protocol, and route +generation before the session becomes online. + Peer exchange is the torrent-shaped layer under Verse discovery. A Verse catalog says which graph exists; a peer catalog says which nodes can currently serve or observe parts of that graph. Peer cards can advertise discovery, diff --git a/src/GameCult.Networking/CultNetOperationServer.cs b/src/GameCult.Networking/CultNetOperationServer.cs index 6261328e..27b52c55 100644 --- a/src/GameCult.Networking/CultNetOperationServer.cs +++ b/src/GameCult.Networking/CultNetOperationServer.cs @@ -164,6 +164,16 @@ namespace GameCult.Networking if (_disposed) return; if (request == null) throw new ArgumentNullException(nameof(request)); if (peer == null) throw new ArgumentNullException(nameof(peer)); + if (_sourceRuntimeId != null && + !string.Equals(request.TargetRuntimeId, _sourceRuntimeId, StringComparison.Ordinal)) + { + peer.SendCultNet(Failure( + request, + "denied", + "target-runtime-mismatch", + $"Operation targeted runtime '{request.TargetRuntimeId ?? "(missing)"}', not '{_sourceRuntimeId}'.")); + return; + } Route route; try diff --git a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs index 386bc99b..e712beef 100644 --- a/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs +++ b/src/GameCult.Networking/CultNetSchemaMessageSerialization.cs @@ -72,6 +72,8 @@ namespace GameCult.Networking CultNetSchemaVersions.SimulationConsensusCandidate => Deserialize(payload), CultNetSchemaVersions.VerseCatalogRequest => Deserialize(payload), CultNetSchemaVersions.VerseCatalogResponse => Deserialize(payload), + CultNetSchemaVersions.SessionOpen => Deserialize(payload), + CultNetSchemaVersions.SessionAccepted => Deserialize(payload), CultNetSchemaVersions.PeerExchangeRequest => Deserialize(payload), CultNetSchemaVersions.PeerExchangeResponse => Deserialize(payload), CultNetSchemaVersions.ContentChunkRequest => Deserialize(payload), diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 764238ff..2117693d 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -135,6 +135,10 @@ namespace GameCult.Networking /// CultMesh Verse catalog response contract identifier. /// public const string VerseCatalogResponse = "cultmesh.verse_catalog_response.v0"; + /// CultMesh session identity request contract identifier. + public const string SessionOpen = "cultmesh.session_open.v1"; + /// CultMesh verified session identity response contract identifier. + public const string SessionAccepted = "cultmesh.session_accepted.v1"; /// /// CultMesh peer exchange request contract identifier. /// @@ -1179,6 +1183,33 @@ namespace GameCult.Networking } } + /// Requests an identity-bound CultMesh session before application traffic begins. + [MessagePackObject] + public class CultMeshSessionOpenMessage : ICultNetSchemaMessage + { + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.SessionOpen; + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + [Key("sourceRuntimeId")] public string SourceRuntimeId { get; set; } = string.Empty; + [Key("verseId")] public string VerseId { get; set; } = string.Empty; + [Key("authorityRuntimeId")] public string AuthorityRuntimeId { get; set; } = string.Empty; + [Key("protocolId")] public string ProtocolId { get; set; } = string.Empty; + [Key("routeGeneration")] public string RouteGeneration { get; set; } = string.Empty; + } + + /// Accepts or rejects a CultMesh session with the server's actual identity. + [MessagePackObject] + public class CultMeshSessionAcceptedMessage : ICultNetSchemaMessage + { + [Key("schemaVersion")] public string SchemaVersion { get; set; } = CultNetSchemaVersions.SessionAccepted; + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + [Key("accepted")] public bool Accepted { get; set; } + [Key("verseId")] public string VerseId { get; set; } = string.Empty; + [Key("authorityRuntimeId")] public string AuthorityRuntimeId { get; set; } = string.Empty; + [Key("protocolId")] public string ProtocolId { get; set; } = string.Empty; + [Key("routeGeneration")] public string RouteGeneration { get; set; } = string.Empty; + [Key("error")] public string? Error { get; set; } + } + /// /// Requests known CultMesh Verses from a peer. /// @@ -1254,6 +1285,11 @@ namespace GameCult.Networking /// [Key("authorityRuntimeIds")] public string[] AuthorityRuntimeIds { get; set; } = Array.Empty(); /// + /// Gets or sets exact authoritative-runtime route bindings. This is the + /// routing authority; the parallel legacy arrays are projections only. + /// + [Key("authorityRoutes")] public CultMeshAuthorityRouteMessage[] AuthorityRoutes { get; set; } = Array.Empty(); + /// /// Gets or sets the parent Verse id for overlays or branches. /// [Key("parentVerseId")] public string? ParentVerseId { get; set; } @@ -1263,6 +1299,17 @@ namespace GameCult.Networking [Key("description")] public string? Description { get; set; } } + /// Wire representation of one authority-bound CultMesh route. + [MessagePackObject] + public class CultMeshAuthorityRouteMessage + { + [Key("authorityRuntimeId")] public string AuthorityRuntimeId { get; set; } = string.Empty; + [Key("endpoint")] public string Endpoint { get; set; } = string.Empty; + [Key("protocolIds")] public string[] ProtocolIds { get; set; } = Array.Empty(); + [Key("priority")] public int Priority { get; set; } + [Key("generation")] public string Generation { get; set; } = string.Empty; + } + /// /// Wire representation of CultMesh Verse compatibility. /// diff --git a/tests/GameCult.Mesh.Quic.Native.Tests/CultMeshNativeQuicRealtimeTransportTests.cs b/tests/GameCult.Mesh.Quic.Native.Tests/CultMeshNativeQuicRealtimeTransportTests.cs index afa74426..7eb99edc 100644 --- a/tests/GameCult.Mesh.Quic.Native.Tests/CultMeshNativeQuicRealtimeTransportTests.cs +++ b/tests/GameCult.Mesh.Quic.Native.Tests/CultMeshNativeQuicRealtimeTransportTests.cs @@ -22,7 +22,7 @@ public sealed class CultMeshNativeQuicRealtimeTransportTests var connector = new CultMeshNativeQuicRealtimeTransportConnector(); using var client = await connector.ConnectAsync( new CultMeshTransportCandidate(endpoint!), - CultMeshEndpointId.Parse("aetheria.local")); + new CultMeshSessionTarget("aetheria", "aetheria.local")); using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); var frame = await client.ReceiveAsync(timeout.Token); @@ -49,7 +49,7 @@ public sealed class CultMeshNativeQuicRealtimeTransportTests using var client = await connector.ConnectAsync( new CultMeshTransportCandidate(endpoint), - CultMeshEndpointId.Parse("aetheria.local")); + new CultMeshSessionTarget("aetheria", "aetheria.local")); var expected = new CultMeshRealtimeFrame { ChannelId = "aetheria.entities", @@ -92,7 +92,7 @@ public sealed class CultMeshNativeQuicRealtimeTransportTests var connect = () => connector.ConnectAsync( new CultMeshTransportCandidate(endpoint), - CultMeshEndpointId.Parse("aetheria.local")); + new CultMeshSessionTarget("aetheria", "aetheria.local")); await connect.Should().ThrowAsync() .WithMessage("*certificate*pin*"); @@ -114,10 +114,10 @@ public sealed class CultMeshNativeQuicRealtimeTransportTests var connector = new CultMeshNativeQuicRealtimeTransportConnector(); var departedClient = await connector.ConnectAsync( new CultMeshTransportCandidate(endpoint), - CultMeshEndpointId.Parse("aetheria.local")); + new CultMeshSessionTarget("aetheria", "aetheria.local")); using var healthyClient = await connector.ConnectAsync( new CultMeshTransportCandidate(endpoint), - CultMeshEndpointId.Parse("aetheria.local")); + new CultMeshSessionTarget("aetheria", "aetheria.local")); await WaitUntilAsync(() => server.ConnectionCount == 2); await server.BroadcastAsync(Frame(sequence: 1)); using var receiveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(5)); diff --git a/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs b/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs index 8d90f147..68afda10 100644 --- a/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs +++ b/tests/GameCult.Mesh.Quic.Tests/CultMeshQuicRealtimeTransportTests.cs @@ -40,7 +40,7 @@ public sealed class CultMeshQuicRealtimeTransportTests }); using var client = await connector.ConnectAsync( new CultMeshTransportCandidate($"cultmesh-state+quic://127.0.0.1:{listener.LocalEndPoint.Port}"), - CultMeshEndpointId.Parse("service:aetheria.daemon")); + new CultMeshSessionTarget("aetheria", "service:aetheria.daemon")); await using var serverConnection = await accept; client.TransportId.Should().Be("msquic-realtime"); @@ -62,7 +62,7 @@ public sealed class CultMeshQuicRealtimeTransportTests using var client = await connector.ConnectAsync( new CultMeshTransportCandidate(endpoint), - CultMeshEndpointId.Parse("service:aetheria.daemon")); + new CultMeshSessionTarget("aetheria", "service:aetheria.daemon")); client.TransportId.Should().Be("msquic-realtime"); await WaitUntilAsync(() => server.ConnectionCount == 1); @@ -78,8 +78,8 @@ public sealed class CultMeshQuicRealtimeTransportTests ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0), ServerCertificate = certificate }); - var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon"); - var target = new CultMeshSessionTarget("aetheria", endpointId.Value); + var runtimeId = CultMeshRuntimeId.Parse("service:aetheria.daemon"); + var target = new CultMeshSessionTarget("aetheria", runtimeId.Value); using var sessions = CreateSessions(server, out var validatedIdentity); CultMeshRealtimeSession session; try @@ -105,7 +105,7 @@ public sealed class CultMeshQuicRealtimeTransportTests new[] { firstAtServer.Sequence, secondAtServer.Sequence }.Should().Equal(1, 2); new[] { firstAtClient.Sequence, secondAtClient.Sequence }.Should().Equal(3, 4); firstAtClient.Payload.ToArray().Should().Equal(3, 4, 5); - validatedIdentity.Value.Should().Be(endpointId.Value); + validatedIdentity.Value.Should().Be(runtimeId.Value); } [Test] @@ -118,8 +118,8 @@ public sealed class CultMeshQuicRealtimeTransportTests ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0), ServerCertificate = certificate }); - var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon"); - var target = new CultMeshSessionTarget("aetheria", endpointId.Value); + var runtimeId = CultMeshRuntimeId.Parse("service:aetheria.daemon"); + var target = new CultMeshSessionTarget("aetheria", runtimeId.Value); using var sessions = CreateSessions(server, out _); var session = await sessions.ConnectRealtimeAsync(target); await WaitUntilAsync(() => server.ConnectionCount == 1); @@ -148,8 +148,8 @@ public sealed class CultMeshQuicRealtimeTransportTests ListenEndPoint = new IPEndPoint(IPAddress.Loopback, 0), ServerCertificate = certificate }); - var endpointId = CultMeshEndpointId.Parse("service:aetheria.daemon"); - var target = new CultMeshSessionTarget("aetheria", endpointId.Value); + var runtimeId = CultMeshRuntimeId.Parse("service:aetheria.daemon"); + var target = new CultMeshSessionTarget("aetheria", runtimeId.Value); using var sessions = CreateSessions(server, out _); var session = await sessions.ConnectRealtimeAsync(target); diff --git a/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs b/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs index bdf9928f..81ee15cc 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshBodySessionTests.cs @@ -53,6 +53,13 @@ public sealed class CultMeshBodySessionTests RuntimeId = "cultmesh-body-rudp-test", Socket = socket, MaxFragmentBytes = 1024, MaxPendingReliablePackets = 8192 }); + var endpoint = $"rudp://127.0.0.1:{server.LocalEndPoint.Port}"; + using var identityServer = new CultMeshSessionIdentityServer( + server, + "aetheria.daemon", + new[] { "aetheria" }, + new[] { CultMeshProtocols.Bodies.Value }, + new[] { "aetheria.daemon@" + endpoint }); using var bodyServer = new CultMeshBodyServer(server, store); using var pumpCancellation = new CancellationTokenSource(); var pump = Task.Run(async () => @@ -64,7 +71,6 @@ public sealed class CultMeshBodySessionTests if (progress.TransportItemsConsumed == 0) await Task.Delay(1, pumpCancellation.Token); } }); - var endpoint = $"rudp://127.0.0.1:{server.LocalEndPoint.Port}"; using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) }); using var sessions = new CultMeshSessionManager(discovery, new ICultMeshTransportConnector[] { new CultMeshSchemaTransportConnector() }); @@ -179,7 +185,7 @@ public sealed class CultMeshBodySessionTests ((Func)_handlers[typeof(T)])(message, peer); } - private sealed class LoopbackClient : ICultNetSchemaClient + private sealed class LoopbackClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient { private readonly LoopbackServer _server; private readonly Dictionary> _handlers = new(); @@ -208,6 +214,8 @@ public sealed class CultMeshBodySessionTests foreach (var value in values.ToArray()) ((Action)value)(message); } public void Dispose() { } + public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) => + verseId == "aetheria" && authorityRuntimeId == "aetheria.daemon"; } private sealed class LoopbackPeer : ICultNetSchemaServerPeer diff --git a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs index d88b5738..e603415e 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs @@ -303,7 +303,7 @@ public sealed class CultMeshClientTests result.MessageId.Should().Be("pilot-command-1"); result.Status.Should().Be("accepted"); - result.SourceRuntimeId.Should().Be("provider-daemon"); + result.SourceRuntimeId.Should().Be("aetheria-daemon"); result.Value.ReceiptId.Should().Be("receipt:pilot-command-1"); result.Value.Count.Should().Be(2); connector.Clients.Should().ContainSingle(); @@ -370,6 +370,31 @@ public sealed class CultMeshClientTests connector.Clients[0].Operations.Should().ContainSingle(); } + [Test] + public async Task InvokeOperation_RejectsReceiptFromAnotherRuntime() + { + var connector = new DocumentConnector { ResponseSourceRuntimeId = "intruder-daemon" }; + using var mesh = new CultMeshClient(new CultMeshClientOptions + { + RendezvousEndpoints = new[] { "rudp://odin:3076" }, + Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() }, + Connectors = new[] { connector } + }); + + Func invoke = async () => await mesh.InvokeAsync( + Target, + "tests.counter", + "tests.counter.increment", + "tests.counter.increment_request.v1", + "tests.counter.increment_receipt.v1", + new ClientOperationRequest { Amount = 2 }, + sourceRuntimeId: "pilot-one", + idempotencyKey: "pilot-command-wrong-source"); + + var failure = await invoke.Should().ThrowAsync(); + failure.Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority); + } + [Test] public async Task Dispose_TerminatesDocumentWaitingForInitialSnapshot() { @@ -438,13 +463,15 @@ public sealed class CultMeshClientTests } } - private sealed class ConnectedClient : ICultNetSchemaClient + private sealed class ConnectedClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient { public bool Connected => true; public void Connect(string host, int port) { } public void SendCultNet(T message) where T : ICultNetSchemaMessage { } public void OnCultNet(Action callback) where T : ICultNetSchemaMessage { } public void Dispose() { } + public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) => + verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon"; } private sealed class DocumentConnector : ICultMeshTransportConnector @@ -454,6 +481,7 @@ public sealed class CultMeshClientTests public bool SuppressFirstSubscriptionSnapshot { get; set; } public bool RejectOperations { get; set; } public bool FailFirstOperationBeforeResponse { get; set; } + public string ResponseSourceRuntimeId { get; set; } = "aetheria-daemon"; public string ConnectorId => "document-test"; public int Priority => 0; public int SubscribeCount => Clients.Sum(client => client.SubscribeCount); @@ -469,13 +497,14 @@ public sealed class CultMeshClientTests ? int.MaxValue : SuppressFirstSubscriptionSnapshot && Clients.Count == 0 ? 1 : 0, rejectOperations: RejectOperations, - failOperationBeforeResponse: FailFirstOperationBeforeResponse && Clients.Count == 0); + failOperationBeforeResponse: FailFirstOperationBeforeResponse && Clients.Count == 0, + responseSourceRuntimeId: ResponseSourceRuntimeId); Clients.Add(client); return Task.FromResult(client); } } - private sealed class DocumentClient : ICultNetSchemaClient, ICultNetSchemaClientHealth + private sealed class DocumentClient : ICultNetSchemaClient, ICultNetSchemaClientHealth, ICultMeshVerifiedSchemaClient { private readonly List> _snapshots = new(); private readonly List> _operationResponses = new(); @@ -484,17 +513,20 @@ public sealed class CultMeshClientTests private readonly TaskCompletionSource _failure = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly bool _rejectOperations; private bool _failOperationBeforeResponse; + private readonly string _responseSourceRuntimeId; private int _snapshotsToSuppress; public DocumentClient( string text, int snapshotsToSuppress = 0, bool rejectOperations = false, - bool failOperationBeforeResponse = false) + bool failOperationBeforeResponse = false, + string responseSourceRuntimeId = "aetheria-daemon") { _text = text; _snapshotsToSuppress = snapshotsToSuppress; _rejectOperations = rejectOperations; _failOperationBeforeResponse = failOperationBeforeResponse; + _responseSourceRuntimeId = responseSourceRuntimeId; var cache = CultMesh.CreateCultCacheDocumentRegistry(typeof(ClientTestDocument)); _documents = CultMesh.CreateCultNetDocumentRegistry(new[] { typeof(ClientTestDocument) }, cache); } @@ -514,6 +546,7 @@ public sealed class CultMeshClientTests } if (message is CultNetOperationRequestMessage operation) { + operation.TargetRuntimeId.Should().Be("aetheria-daemon"); Operations.Add(operation); if (_failOperationBeforeResponse) { @@ -539,7 +572,7 @@ public sealed class CultMeshClientTests failure, CultNetSchemaMessageSerialization.Options)), Diagnostics = new[] { failure.Message }, - SourceRuntimeId = "provider-daemon" + SourceRuntimeId = _responseSourceRuntimeId }; foreach (var handler in _operationResponses.ToArray()) handler(rejected); return; @@ -562,7 +595,7 @@ public sealed class CultMeshClientTests Payload = Convert.ToBase64String(MessagePackSerializer.Serialize( receipt, CultNetSchemaMessageSerialization.Options)), - SourceRuntimeId = "provider-daemon" + SourceRuntimeId = _responseSourceRuntimeId }; foreach (var handler in _operationResponses.ToArray()) handler(operationResponse); return; @@ -596,6 +629,8 @@ public sealed class CultMeshClientTests } public void Dispose() { } public void Fail(Exception error) => _failure.TrySetResult(error); + public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) => + verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon"; } private static async Task WaitUntilAsync(Func condition) diff --git a/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs b/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs index d63254d1..fa11072c 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshContentSessionTests.cs @@ -17,6 +17,7 @@ namespace GameCult.Mesh.Tests; [TestFixture] public sealed class CultMeshContentSessionTests { + private const string RouteGeneration = "content-test-route-1"; private string _directory = null!; [SetUp] @@ -41,7 +42,13 @@ public sealed class CultMeshContentSessionTests StringComparer.Ordinal); using var contentServer = new CultMeshTcpContentServer( new TcpListener(IPAddress.Loopback, 0), - hash => providerChunks.TryGetValue(hash, out var chunk) ? chunk : null); + hash => providerChunks.TryGetValue(hash, out var chunk) ? chunk : null, + new CultMeshTcpContentServerOptions + { + VerseId = "aetheria", + AuthorityRuntimeId = "aetheria.daemon", + RouteGeneration = RouteGeneration + }); var endpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{contentServer.LocalEndPoint.Port}"; var connector = new CultMeshTcpContentTransportConnector(); using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) }); @@ -73,6 +80,34 @@ public sealed class CultMeshContentSessionTests "all chunk requests borrow one identity-first content session"); } + [Test] + public async Task TcpContentSessionRejectsEndpointServingAnotherAuthority() + { + using var content = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( + typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk))); + using var contentServer = new CultMeshTcpContentServer( + new TcpListener(IPAddress.Loopback, 0), + content, + new CultMeshTcpContentServerOptions + { + VerseId = "aetheria", + AuthorityRuntimeId = "intruder.daemon", + RouteGeneration = RouteGeneration + }); + var endpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{contentServer.LocalEndPoint.Port}"; + using var discovery = new CultMeshDiscoveryService(new[] { new RouteSource(endpoint) }); + using var sessions = new CultMeshSessionManager( + discovery, + Array.Empty(), + new ICultMeshContentTransportConnector[] { new CultMeshTcpContentTransportConnector() }); + + Func connect = async () => await sessions.ConnectContentAsync( + new CultMeshSessionTarget("aetheria", "aetheria.daemon")); + + var error = await connect.Should().ThrowAsync(); + error.Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority); + } + [Test] public async Task RudpContentSessionCarriesConcurrentChunkWindowWithoutPerFragmentSchedulerDelay() { @@ -91,6 +126,12 @@ public sealed class CultMeshContentSessionTests MaxPendingReliablePackets = 8192 }); using var contentServer = new CultMeshLegacyRudpContentServer(wireServer, providerCache); + using var identityServer = new CultMeshSessionIdentityServer( + wireServer, + "aetheria.daemon", + new[] { "aetheria" }, + new[] { CultMeshProtocols.Content.Value }, + new[] { RouteGeneration }); using var pumpCancellation = new CancellationTokenSource(); var pump = Task.Run(async () => { @@ -196,8 +237,14 @@ public sealed class CultMeshContentSessionTests "Aetheria", CultMeshVerseAuthorityModel.OperatorCluster, new CultMeshVerseCompatibility("cultmesh.v1", "rules"), - new[] { _endpoint }, - new[] { "aetheria.daemon" }), + authorityRoutes: new[] + { + new CultMeshAuthorityRoute( + "aetheria.daemon", + _endpoint, + new[] { CultMeshProtocols.Content.Value }, + generation: RouteGeneration) + }), SourceId, now, now.AddMinutes(1), @@ -241,7 +288,7 @@ public sealed class CultMeshContentSessionTests ((Func)_handlers[typeof(TMessage)])(message, peer); } - private sealed class LoopbackSchemaClient : ICultNetSchemaClient + private sealed class LoopbackSchemaClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient { private readonly LoopbackSchemaServer _server; private readonly Dictionary> _handlers = new(); @@ -270,6 +317,8 @@ public sealed class CultMeshContentSessionTests handlers.Add(callback); } public void Dispose() { } + public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) => + verseId == "aetheria" && authorityRuntimeId == "aetheria.daemon"; public void Emit(T message) where T : ICultNetSchemaMessage { if (!_handlers.TryGetValue(typeof(T), out var handlers)) return; diff --git a/tests/GameCult.Mesh.Tests/CultMeshDiscoveryServiceTests.cs b/tests/GameCult.Mesh.Tests/CultMeshDiscoveryServiceTests.cs index 7d2938e4..2313a5bd 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshDiscoveryServiceTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshDiscoveryServiceTests.cs @@ -295,7 +295,7 @@ public sealed class CultMeshDiscoveryServiceTests CultMeshVerseAuthorityModel.OperatorCluster, new CultMeshVerseCompatibility("cultmesh.v0", "rules"), new[] { endpoint }, - new[] { "odin:aetheria", "odin:games", "aetheria" }), + new[] { "aetheria" }), sourceId, clock.UtcNow, clock.UtcNow + ttl, diff --git a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs index fe9ec370..46e0672f 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshSessionManagerTests.cs @@ -79,6 +79,81 @@ public sealed class CultMeshSessionManagerTests fastClient.DisposeCount.Should().Be(0); } + [Test] + public async Task Connect_NeverAttemptsRouteBoundToAnotherAuthority() + { + var clock = new ManualSessionClock(); + var descriptor = new CultMeshVerseDescriptor( + "aetheria", + "Aetheria", + CultMeshVerseAuthorityModel.OperatorCluster, + new CultMeshVerseCompatibility("cultmesh.v0", "rules"), + authorityRoutes: new[] + { + new CultMeshAuthorityRoute("decoy-daemon", "rudp://decoy:3076"), + new CultMeshAuthorityRoute("aetheria-daemon", "rudp://authority:3076") + }); + using var discovery = new CultMeshDiscoveryService( + new[] { new StaticRouteSource(clock, descriptor) }, + new CultMeshDiscoveryServiceOptions { Clock = clock }); + var attempts = new List(); + var connector = new FakeConnector((candidate, _) => + { + attempts.Add(candidate.Endpoint); + return Task.FromResult(new FakeSchemaClient()); + }); + using var manager = new CultMeshSessionManager(discovery, new[] { connector }); + + var session = await manager.ConnectAsync(Target, CultMeshProtocols.Documents); + + session.State.Path!.Endpoint.Should().Be("rudp://authority:3076"); + attempts.Should().Equal("rudp://authority:3076"); + } + + [Test] + public void Connect_RejectsRouteWhoseNativeIdentityDoesNotMatchTarget() + { + var clock = new ManualSessionClock(); + using var discovery = Discovery(clock, () => new[] { "rudp://decoy:3076" }); + var connector = new FakeConnector((_, _) => + Task.FromResult(new NativeMismatchClient())); + using var manager = new CultMeshSessionManager(discovery, new[] { connector }); + + var error = Assert.ThrowsAsync(() => + manager.ConnectAsync(Target, CultMeshProtocols.Documents)); + + error!.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority); + } + + [Test] + public void Connect_RejectsPortableHandshakeFromWrongRuntime() + { + var clock = new ManualSessionClock(); + using var discovery = Discovery(clock, () => new[] { "rudp://decoy:3076" }); + var connector = new FakeConnector((_, _) => + Task.FromResult(new HandshakeSchemaClient("decoy-daemon"))); + using var manager = new CultMeshSessionManager(discovery, new[] { connector }); + + var error = Assert.ThrowsAsync(() => + manager.ConnectAsync(Target, CultMeshProtocols.Documents)); + + error!.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority); + } + + [Test] + public void Descriptor_RejectsAmbiguousLegacyMultiAuthorityRoutes() + { + var action = () => new CultMeshVerseDescriptor( + "aetheria", + "Aetheria", + CultMeshVerseAuthorityModel.OperatorCluster, + new CultMeshVerseCompatibility("cultmesh.v0", "rules"), + discoveryEndpoints: new[] { "rudp://shared:3076" }, + authorityRuntimeIds: new[] { "authority-a", "authority-b" }); + + action.Should().Throw().WithMessage("*ambiguous*"); + } + [Test] public async Task Connect_CallerCancellationDoesNotCancelSharedAttempt() { @@ -308,6 +383,30 @@ public sealed class CultMeshSessionManagerTests CultMeshDiscoveryTrust.Signed); } + private sealed class StaticRouteSource : ICultMeshLookupSource + { + private readonly ManualSessionClock _clock; + private readonly CultMeshVerseDescriptor _descriptor; + public StaticRouteSource(ManualSessionClock clock, CultMeshVerseDescriptor descriptor) + { + _clock = clock; + _descriptor = descriptor; + } + public string SourceId => "odin"; + public Task> LookupAsync( + CultMeshDiscoveryQuery query, + CancellationToken cancellationToken = default) => + Task.FromResult>(new[] + { + new CultMeshDiscoveryObservation( + _descriptor, + SourceId, + _clock.UtcNow, + _clock.UtcNow.AddMinutes(1), + CultMeshDiscoveryTrust.Signed) + }); + } + private sealed class FakeConnector : ICultMeshTransportConnector { private readonly Func> _connect; @@ -324,7 +423,7 @@ public sealed class CultMeshSessionManagerTests } } - private sealed class FakeSchemaClient : ICultNetSchemaClient, ICultNetSchemaClientHealth + private sealed class FakeSchemaClient : ICultNetSchemaClient, ICultNetSchemaClientHealth, ICultMeshVerifiedSchemaClient { private readonly TaskCompletionSource _failure = new(TaskCreationOptions.RunContinuationsAsynchronously); private readonly List> _handlers = new(); @@ -340,6 +439,8 @@ public sealed class CultMeshSessionManagerTests _handlers.Add(message => callback((T)(object)message)); } public void Dispose() => Interlocked.Increment(ref _disposeCount); + public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) => + verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon"; public void Fail(Exception error) => _failure.TrySetResult(error); public void Emit(CultNetErrorMessage message) { foreach (var handler in _handlers.ToArray()) handler(message); } } @@ -360,7 +461,46 @@ public sealed class CultMeshSessionManagerTests public void Dispose() { } } - private sealed class RespondingSchemaClient : ICultNetSchemaClient + private sealed class NativeMismatchClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient + { + public bool Connected => true; + public void Connect(string host, int port) { } + public void SendCultNet(T message) where T : ICultNetSchemaMessage { } + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage { } + public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) => false; + public void Dispose() { } + } + + private sealed class HandshakeSchemaClient : ICultNetSchemaClient + { + private readonly string _actualRuntimeId; + private readonly List> _handlers = new(); + public HandshakeSchemaClient(string actualRuntimeId) => _actualRuntimeId = actualRuntimeId; + public bool Connected => true; + public void Connect(string host, int port) { } + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + if (message is not CultMeshSessionOpenMessage request) return; + var response = new CultMeshSessionAcceptedMessage + { + MessageId = request.MessageId, + Accepted = true, + VerseId = request.VerseId, + AuthorityRuntimeId = _actualRuntimeId, + ProtocolId = request.ProtocolId, + RouteGeneration = request.RouteGeneration + }; + foreach (var handler in _handlers.ToArray()) handler(response); + } + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + if (typeof(T) == typeof(CultMeshSessionAcceptedMessage)) + _handlers.Add(message => callback((T)(object)message)); + } + public void Dispose() { } + } + + private sealed class RespondingSchemaClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient { private readonly List> _handlers = new(); private int _disposeCount; @@ -385,9 +525,11 @@ public sealed class CultMeshSessionManagerTests _handlers.Add(message => callback((T)(object)message)); } public void Dispose() => Interlocked.Increment(ref _disposeCount); + public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) => + verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon"; } - private sealed class PeerExchangeSchemaClient : ICultNetSchemaClient + private sealed class PeerExchangeSchemaClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient { private readonly List> _handlers = new(); public bool Connected => true; @@ -413,9 +555,11 @@ public sealed class CultMeshSessionManagerTests _handlers.Add(message => callback((T)(object)message)); } public void Dispose() { } + public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) => + verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon"; } - private sealed class LeaseSchemaClient : ICultNetSchemaClient + private sealed class LeaseSchemaClient : ICultNetSchemaClient, ICultMeshVerifiedSchemaClient { private readonly List> _handlers = new(); private int _disposeCount; @@ -430,6 +574,8 @@ public sealed class CultMeshSessionManagerTests } public void Emit(CultNetErrorMessage message) { foreach (var handler in _handlers.ToArray()) handler(message); } public void Dispose() => Interlocked.Increment(ref _disposeCount); + public bool IsVerifiedFor(string verseId, string authorityRuntimeId, string protocolId, string routeGeneration) => + verseId == "aetheria" && authorityRuntimeId == "aetheria-daemon"; } private sealed class ManualSessionClock : ICultMeshClock diff --git a/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs b/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs index 5fe32fed..35b4a71c 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshTransportModularityTests.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.IO; +using System.Linq; using System.Net; using System.Net.Sockets; using System.Threading; @@ -15,10 +16,17 @@ namespace GameCult.Mesh.Tests; [TestFixture] public sealed class CultMeshTransportModularityTests { + private const string RouteGeneration = "transport-modularity-route-1"; [Test] public async Task SchemaSessionPrefersTcpTierWithoutTouchingLegacyDatagrams() { using var server = new TcpFramedCultNetSchemaServer(new TcpListener(IPAddress.Loopback, 0)); + using var identity = new CultMeshSessionIdentityServer( + server, + "aetheria.daemon", + new[] { "aetheria" }, + new[] { CultMeshProtocols.Documents.Value }, + new[] { RouteGeneration }); server.OnCultNet((message, peer) => { peer.SendCultNet(new CultNetErrorMessage { Error = "tcp:" + message.RuntimeId }); @@ -61,7 +69,15 @@ public sealed class CultMeshTransportModularityTests using var content = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( typeof(CultMeshCdnArtifactManifest), typeof(CultMeshCdnArtifactChunk))); await CultMeshCdn.PublishAsync(content, artifact); - using var server = new CultMeshTcpContentServer(new TcpListener(IPAddress.Loopback, 0), content); + using var server = new CultMeshTcpContentServer( + new TcpListener(IPAddress.Loopback, 0), + content, + new CultMeshTcpContentServerOptions + { + VerseId = "aetheria", + AuthorityRuntimeId = "aetheria.daemon", + RouteGeneration = RouteGeneration + }); var tcpEndpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{server.LocalEndPoint.Port}"; var legacy = new RejectingLegacyContentConnector(); using var discovery = new CultMeshDiscoveryService(new[] @@ -156,6 +172,26 @@ public sealed class CultMeshTransportModularityTests .WithMessage("*No realtime state connectors are configured*"); } + [Test] + public async Task RealtimeSessionRejectsConnectedTransportWithoutAuthorityProof() + { + using var discovery = new CultMeshDiscoveryService(new[] + { + new RouteSource("cultmesh-state+quic://127.0.0.1:3076") + }); + using var sessions = new CultMeshSessionManager( + discovery, + Array.Empty(), + Array.Empty(), + new ICultMeshRealtimeTransportConnector[] { new UnverifiedRealtimeConnector() }); + + Func connect = async () => await sessions.ConnectRealtimeAsync( + new CultMeshSessionTarget("aetheria", "aetheria.daemon")); + + var error = await connect.Should().ThrowAsync(); + error.Which.Failure.Reason.Should().Be(CultMeshSessionFailureReason.Authority); + } + private sealed class RouteSource : ICultMeshLookupSource { private readonly string[] _endpoints; @@ -176,8 +212,11 @@ public sealed class CultMeshTransportModularityTests "Aetheria", CultMeshVerseAuthorityModel.OperatorCluster, new CultMeshVerseCompatibility("cultmesh.v1", "rules"), - _endpoints, - new[] { "aetheria.daemon" }), + authorityRoutes: _endpoints.Select(endpoint => new CultMeshAuthorityRoute( + "aetheria.daemon", + endpoint, + protocolIds: null, + generation: RouteGeneration))), SourceId, now, now.AddMinutes(1), @@ -214,7 +253,7 @@ public sealed class CultMeshTransportModularityTests candidate.Endpoint.StartsWith("rudp://", StringComparison.OrdinalIgnoreCase); public Task ConnectAsync( CultMeshTransportCandidate candidate, - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, CancellationToken cancellationToken = default) { Interlocked.Increment(ref _connectCount); @@ -232,24 +271,70 @@ public sealed class CultMeshTransportModularityTests candidate.Endpoint.StartsWith("cultmesh-state+quic://", StringComparison.OrdinalIgnoreCase); public Task ConnectAsync( CultMeshTransportCandidate candidate, - CultMeshEndpointId endpointId, + CultMeshSessionTarget target, CancellationToken cancellationToken = default) { Interlocked.Increment(ref _connectCount); - return Task.FromResult(new LoopbackRealtimeTransport(candidate.Endpoint)); + return Task.FromResult( + new LoopbackRealtimeTransport(candidate.Endpoint, target, candidate.Generation)); } } - private sealed class LoopbackRealtimeTransport : ICultMeshRealtimeTransport + private sealed class UnverifiedRealtimeConnector : ICultMeshRealtimeTransportConnector + { + public string ConnectorId => "unverified-realtime"; + public int Priority => 0; + public bool CanConnect(CultMeshTransportCandidate candidate) => true; + public Task ConnectAsync( + CultMeshTransportCandidate candidate, + CultMeshSessionTarget target, + CancellationToken cancellationToken = default) => + Task.FromResult(new UnverifiedRealtimeTransport(candidate.Endpoint)); + } + + private sealed class UnverifiedRealtimeTransport : ICultMeshRealtimeTransport + { + public UnverifiedRealtimeTransport(string endpoint) => Endpoint = endpoint; + public string TransportId => "unverified-realtime"; + public string Endpoint { get; } + public Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default) => + Task.CompletedTask; + public Task ReceiveAsync(CancellationToken cancellationToken = default) => + Task.FromException(new InvalidOperationException()); + public void Dispose() { } + } + + private sealed class LoopbackRealtimeTransport : ICultMeshRealtimeTransport, ICultMeshVerifiedTransport { private readonly Queue _frames = new(); private readonly SemaphoreSlim _available = new(0); private bool _disposed; - public LoopbackRealtimeTransport(string endpoint) => Endpoint = endpoint; + private readonly CultMeshSessionTarget _target; + private readonly string _routeGeneration; + + public LoopbackRealtimeTransport( + string endpoint, + CultMeshSessionTarget target, + string routeGeneration) + { + Endpoint = endpoint; + _target = target; + _routeGeneration = routeGeneration; + } public string TransportId => "test-quic-state"; public string Endpoint { get; } + public bool IsVerifiedFor( + string verseId, + string authorityRuntimeId, + string protocolId, + string routeGeneration) => + string.Equals(_target.VerseId, verseId, StringComparison.Ordinal) && + string.Equals(_target.AuthorityRuntimeId, authorityRuntimeId, StringComparison.Ordinal) && + string.Equals(CultMeshProtocols.RealtimeState.Value, protocolId, StringComparison.Ordinal) && + string.Equals(_routeGeneration, routeGeneration, StringComparison.Ordinal); + public Task SendAsync(CultMeshRealtimeFrame frame, CancellationToken cancellationToken = default) { if (_disposed) throw new ObjectDisposedException(nameof(LoopbackRealtimeTransport)); diff --git a/tests/GameCult.Networking.Tests/CultNetOperationServerTests.cs b/tests/GameCult.Networking.Tests/CultNetOperationServerTests.cs index f82456fb..ca36a2f4 100644 --- a/tests/GameCult.Networking.Tests/CultNetOperationServerTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetOperationServerTests.cs @@ -77,6 +77,41 @@ namespace GameCult.Networking.Tests Is.EqualTo(4)); } + [Test] + public async Task RejectsRequestAddressedToAnotherRuntimeBeforeApplicationHandler() + { + var transport = new FakeServer(); + var peer = new FakePeer(); + var calls = 0; + using var operations = new CultNetOperationServer(transport, "sample.provider") + .Register( + "sample.counter", + "sample.counter.add", + "sample.add.v1", + "sample.add_receipt.v1", + context => + { + calls++; + return Task.FromResult(new AddReceipt { Count = context.Value.Amount }); + }); + var request = Request(new AddRequest { Amount = 3 }); + request.TargetRuntimeId = "other.provider"; + + await transport.DispatchAsync(request, peer); + + Assert.That(calls, Is.Zero); + var response = peer.Messages.OfType().Single(); + var failure = MessagePackSerializer.Deserialize( + Convert.FromBase64String(response.Payload), + CultNetSchemaMessageSerialization.Options); + Assert.Multiple(() => + { + Assert.That(response.Status, Is.EqualTo("denied")); + Assert.That(failure.Code, Is.EqualTo("target-runtime-mismatch")); + Assert.That(response.SourceRuntimeId, Is.EqualTo("sample.provider")); + }); + } + [Test] public async Task RejectsWrongSchemaBeforeApplicationHandler() { @@ -157,7 +192,8 @@ namespace GameCult.Networking.Tests Payload = Convert.ToBase64String(MessagePackSerializer.Serialize( value, CultNetSchemaMessageSerialization.Options)), - SourceRuntimeId = "sample.browser" + SourceRuntimeId = "sample.browser", + TargetRuntimeId = "sample.provider" }; [MessagePackObject] diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index dd4a63c9..03b4f76c 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -3627,8 +3627,19 @@ namespace GameCult.Networking.Tests "Aetheria", CultMeshVerseAuthorityModel.OperatorCluster, new CultMeshVerseCompatibility("cultmesh.v0", vanillaRules), - discoveryEndpoints: ["cultmesh://aetheria.example.test:3075"], - authorityRuntimeIds: ["gc-us-east", "gc-eu-west"]); + authorityRoutes: + [ + new CultMeshAuthorityRoute( + "gc-us-east", + "cultmesh://us-east.aetheria.example.test:3075", + [CultMeshProtocols.Documents.Value], + generation: "east-1"), + new CultMeshAuthorityRoute( + "gc-eu-west", + "cultmesh://eu-west.aetheria.example.test:3075", + [CultMeshProtocols.Documents.Value], + generation: "west-1") + ]); var modded = new CultMeshVerseDescriptor( "aetheria-skylands", "Aetheria: Skylands", diff --git a/tools/GameCult.TransportParity/Program.cs b/tools/GameCult.TransportParity/Program.cs index 04d7aafc..31c93e1e 100644 --- a/tools/GameCult.TransportParity/Program.cs +++ b/tools/GameCult.TransportParity/Program.cs @@ -339,6 +339,12 @@ static async Task MeasureCultMeshRudpAsync(byte[] payload) MaxPendingReliablePackets = 8192 }); using var contentServer = new CultMeshLegacyRudpContentServer(wireServer, providerCache); + using var identityServer = new CultMeshSessionIdentityServer( + wireServer, + "service:parity.provider", + ["parity"], + [CultMeshProtocols.Content.Value], + ["parity-route-1"]); using var pumpCancellation = new CancellationTokenSource(); var pump = Task.Run(async () => { @@ -359,7 +365,7 @@ static async Task MeasureCultMeshRudpAsync(byte[] payload) [new CultMeshLegacyRudpContentTransportConnector( new CultMeshLegacyRudpContentTransportOptions { ResponseTimeout = TimeSpan.FromMinutes(2) })]); var provider = new CultMeshSessionContentProvider( - "parity.provider", sessions, CultMeshEndpointId.Parse("service:parity.provider")); + "parity.provider", sessions, new CultMeshSessionTarget("parity", "service:parity.provider")); var directory = Path.Combine(Path.GetTempPath(), "cultmesh-transport-parity", Guid.NewGuid().ToString("N")); using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( typeof(CultMeshContentTransferStateDocument))); @@ -392,7 +398,14 @@ static async Task MeasureCultMeshTcpAsync(byte[] payload) await CultMeshCdn.PublishAsync(providerCache, artifact); using var contentServer = new CultMeshTcpContentServer( - new TcpListener(IPAddress.Loopback, 0), providerCache); + new TcpListener(IPAddress.Loopback, 0), + providerCache, + new CultMeshTcpContentServerOptions + { + VerseId = "parity", + AuthorityRuntimeId = "service:parity.provider", + RouteGeneration = "parity-route-1" + }); var endpoint = $"{CultMeshTcpContentTransportConnector.Scheme}://127.0.0.1:{contentServer.LocalEndPoint.Port}"; using var discovery = new CultMeshDiscoveryService([new RouteSource(endpoint)]); using var sessions = new CultMeshSessionManager( @@ -400,7 +413,7 @@ static async Task MeasureCultMeshTcpAsync(byte[] payload) Array.Empty(), [new CultMeshTcpContentTransportConnector()]); var provider = new CultMeshSessionContentProvider( - "parity.provider", sessions, CultMeshEndpointId.Parse("service:parity.provider")); + "parity.provider", sessions, new CultMeshSessionTarget("parity", "service:parity.provider")); var directory = Path.Combine(Path.GetTempPath(), "cultmesh-transport-parity", Guid.NewGuid().ToString("N")); using var transferState = new CultCache(CultMesh.CreateCultCacheDocumentRegistry( typeof(CultMeshContentTransferStateDocument))); @@ -479,8 +492,14 @@ internal sealed class RouteSource(string endpoint) : ICultMeshLookupSource new( new CultMeshVerseDescriptor( "parity", "Transport parity", CultMeshVerseAuthorityModel.OperatorCluster, - new CultMeshVerseCompatibility("cultmesh.v1", "parity"), [endpoint], - ["service:parity.provider"]), + new CultMeshVerseCompatibility("cultmesh.v1", "parity"), + authorityRoutes: + [ + new CultMeshAuthorityRoute( + "service:parity.provider", + endpoint, + generation: "parity-route-1") + ]), SourceId, now, now.AddMinutes(1), CultMeshDiscoveryTrust.Signed) ]; return Task.FromResult(observations);