diff --git a/README.md b/README.md index 2e6bc5e..9d22efc 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,10 @@ A **murmuration** is one of nature's most extraordinary phenomena — thousands - **Message streaming — complete.** Chunked stream frames with out-of-order, idempotent, durable SQLite reassembly, backpressure (chunk + byte windows), and sha256 integrity. - **Auth/authz enforcement.** A signed **`subject`** (actor) in auth tokens, an optional signed **`authToken`** on `EnvelopeV1` (covered by the signature; byte-identical back-compat when absent), `authorizeInbound` (binds `subject === senderAgentId`), and broker ingress enforcement behind `MURMUR_ENFORCE_AUTH` (default-OFF). *Daemon end-to-end wiring is the remaining step.* - **Conformance + versioned protocol spec — all wire types.** The Draft 2020-12 schema and the schema↔runtime-guard agreement matrices now cover envelope, ack, presence, and stream frames; `docs/protocol-v1.md` + `docs/protocol-compatibility.md` document them. +- **Signed, peer-bound acknowledgements.** Daemons emit Ed25519-signed ACKs bound to the original + message digest, conversation, sender, recipient, timestamp, and nonce. After every peer is + upgraded, set `ackSecurity.requireSigned: true` (or `MURMUR_REQUIRE_SIGNED_ACKS=1`) to reject + unsigned, mismatched, stale, or replayed ACKs without logging message bodies. - **Validated: real cross-host A2A.** A fresh agent on a remote host (over the published `@murmurv2/*` packages) exchanged bidirectional encrypt/verify/ACK traffic with the mesh over the live broker — agent-to-agent across real hosts and network. - **Single canonical signing payload.** `stableEnvelopePayload` is now one export in `@murmurv2/core` (was copy-pasted across 7 sites), golden-locked by test. diff --git a/docs/protocol-v1.md b/docs/protocol-v1.md index e4b8622..1655add 100644 --- a/docs/protocol-v1.md +++ b/docs/protocol-v1.md @@ -12,7 +12,8 @@ the guards cannot drift. Versioning and forward-compatibility rules live in | Type | Purpose | Schema `$def` | Runtime guard | |------|---------|---------------|---------------| | `EnvelopeV1` | encrypted inbound message | document root (`#/$defs/EnvelopeV1`) | `isEnvelopeV1` | -| `AckV1` | delivery acknowledgement | `#/$defs/AckV1` | — | +| `AckV1` | legacy unsigned delivery acknowledgement | `#/$defs/AckV1` | — | +| `SignedAckV1` | signed, peer/message-bound delivery acknowledgement | `#/$defs/SignedAckV1` | `isSignedAckV1` | | `PresenceFrameV1` | discovery announcement (public metadata) | `#/$defs/PresenceFrameV1` | `isPresenceFrameV1` | | `SignedPresenceFrameV1` | Ed25519-signed presence | `#/$defs/SignedPresenceFrameV1` | `isSignedPresenceFrameV1` | | `StreamStart` / `StreamChunk` / `StreamEnd` | chunked payload streaming | `#/$defs/Stream*` (+ `StreamFrame` union) | `isStreamStart` / `isStreamChunk` / `isStreamEnd` / `isStreamFrame` | @@ -28,8 +29,21 @@ Envelope message payloads are encrypted on the wire; presence frames are intenti 3. Publish to subject `msg.` 4. Consumer validates schema+signature 5. Consumer processes idempotently using `msgId` -6. Consumer emits ACK or NACK -7. Retry policy moves failed messages; terminal failures go to DLQ +6. Consumer emits a signed ACK or NACK bound to the message digest, conversation, sender, + recipient, timestamp, and nonce. +7. Sender verifies the signature against the expected peer key and applies an atomic transition + only while the outbox row is in flight. Replays and mismatched bindings are rejected. +8. Retry policy moves failed messages; terminal failures go to DLQ. + +### Signed ACK migration + +The daemon emits `SignedAckV1` by default. During a rolling upgrade, +`ackSecurity.requireSigned` (or `MURMUR_REQUIRE_SIGNED_ACKS=1`) remains disabled until every peer +emits signed ACKs; old consumers ignore the additional signed fields. Once peers are upgraded, +enable strict mode on every endpoint. Strict correlation rejects legacy ACKs, stale/future +timestamps, wrong peers, wrong conversations or recipients, digest mismatches, invalid signatures, +and repeated/non-in-flight transitions. Rejections increment reason-tagged counters and emit +metadata-only security events; ACK bodies and message contents are never logged. An optional `authToken` (bearer `MURMUR-AUTH:…`) authorizes the sender. When present it is part of the signed payload (cannot be stripped/swapped) and can be verified with diff --git a/package-lock.json b/package-lock.json index dbe18c0..9a64d2e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1842,19 +1842,10 @@ "version": "0.2.0", "license": "MIT", "dependencies": { - "@murmurv2/core": "^0.2.0", + "@murmurv2/core": "^0.3.1", "nats": "^2.28.2" } }, - "packages/broker-nats/node_modules/@murmurv2/core": { - "version": "0.2.0", - "resolved": "https://registry.npmjs.org/@murmurv2/core/-/core-0.2.0.tgz", - "integrity": "sha512-ldczX36kucZCajlo4kYfR8MJAlixpITAA629iXbMSNRp5ylaTCdIXm+xoM5vAEBnV3+bE4Eu+sMCydOJWgo8kw==", - "license": "MIT", - "dependencies": { - "pg": "^8.16.3" - } - }, "packages/broker-ws": { "name": "@murmurv2/broker-ws", "version": "0.1.0", diff --git a/packages/broker-nats/package.json b/packages/broker-nats/package.json index 42de7e4..f6ac9b9 100644 --- a/packages/broker-nats/package.json +++ b/packages/broker-nats/package.json @@ -9,7 +9,7 @@ "prepack": "npm run build" }, "dependencies": { - "@murmurv2/core": "^0.2.0", + "@murmurv2/core": "^0.3.1", "nats": "^2.28.2" }, "license": "MIT", diff --git a/packages/broker-nats/src/index.ts b/packages/broker-nats/src/index.ts index f27c38f..ac1a4cf 100644 --- a/packages/broker-nats/src/index.ts +++ b/packages/broker-nats/src/index.ts @@ -13,15 +13,20 @@ import { applyJitter, computeBackoffMs, createAck, + createBoundAck, + envelopeDigest, estimateBase64DecodedBytes, type EnvelopeV1, + isSignedAckV1, isEnvelopeV1, isSignedPresenceFrameV1, type SignedPresenceFrameV1, type DedupeStore, type OutboxStore, type AckV1, + type SignedAckV1, type SecurityPolicy, + type UnsignedAckV1, streamBackpressureAllowsSend, validateEnvelopePolicy, } from "@murmurv2/core"; @@ -70,6 +75,15 @@ export interface AckWindowConfig { maxInFlightBytes: number; } +export type AckSigner = (ack: UnsignedAckV1) => Promise; +export type AckVerifier = (ack: SignedAckV1) => Promise; + +export interface InvalidAckEvent { + reason: string; + msgId?: string; + senderAgentId?: string; +} + interface JetStreamConsumerAdvisory { type?: string; stream?: string; @@ -95,6 +109,8 @@ export class NatsBroker { private jsm?: JetStreamManager; private readonly sc = StringCodec(); private readonly failedDeliveries = new Map(); + private readonly seenAckNonces = new Set(); + private readonly invalidAckCounts = new Map(); private reconnects = 0; private statusLoop?: Promise; @@ -138,6 +154,10 @@ export class NatsBroker { return this.reconnects; } + getAckSecurityMetrics(): Readonly> { + return Object.fromEntries(this.invalidAckCounts.entries()); + } + private jetStreamEnabled(): boolean { return this.config.jetstream === true || !!this.config.stream; } @@ -239,18 +259,31 @@ export class NatsBroker { this.nc!.publish(subject, payload); } - async publishAck(subject: string, envelope: ReturnType): Promise { + async publishAck(subject: string, envelope: AckV1 | SignedAckV1): Promise { await this.connect(); const payload = this.sc.encode(JSON.stringify(envelope)); + const ackSender = "senderAgentId" in envelope ? envelope.senderAgentId : envelope.consumerId; + const nonce = "nonce" in envelope ? `:${envelope.nonce}` : ""; if (this.js) { await this.js.publish(subject, payload, { - msgID: `ack:${envelope.msgId}:${envelope.consumerId}:${envelope.status}`, + msgID: `ack:${envelope.msgId}:${ackSender}:${envelope.status}${nonce}`, }); return; } this.nc!.publish(subject, payload); } + private async createDeliveryAck( + envelope: EnvelopeV1, + consumerId: string, + status: AckV1["status"], + reason: string | undefined, + signAck: AckSigner | undefined, + ): Promise { + if (!signAck) return createAck(envelope.msgId, consumerId, status, reason); + return signAck(createBoundAck(envelope, consumerId, status, reason)); + } + private async processEnvelopeFrame( data: Uint8Array, params: { @@ -259,10 +292,12 @@ export class NatsBroker { onMessage: MessageHandler; maxPoisonAttempts?: number; authorize?: InboundAuthorizer; + signAck?: AckSigner; }, ): Promise<"ack" | "retry"> { let msgId = "unknown"; let ackSubject = `ack.${params.consumerId}`; + let decodedEnvelope: EnvelopeV1 | undefined; try { const decoded = JSON.parse(this.sc.decode(data)); if (!isEnvelopeV1(decoded)) { @@ -270,11 +305,15 @@ export class NatsBroker { return "ack"; } + decodedEnvelope = decoded; msgId = decoded.msgId; ackSubject = `ack.${decoded.senderAgentId}`; const isDup = await params.dedupe.seen(decoded.msgId, params.consumerId); if (isDup) { - await this.publishAck(ackSubject, createAck(decoded.msgId, params.consumerId, "ack", "duplicate-ignored")); + await this.publishAck( + ackSubject, + await this.createDeliveryAck(decoded, params.consumerId, "ack", "duplicate-ignored", params.signAck), + ); return "ack"; } @@ -286,7 +325,13 @@ export class NatsBroker { if (!authz.accepted) { await this.publishAck( ackSubject, - createAck(decoded.msgId, params.consumerId, "nack", `auth-rejected:${authz.reason ?? "denied"}`), + await this.createDeliveryAck( + decoded, + params.consumerId, + "nack", + `auth-rejected:${authz.reason ?? "denied"}`, + params.signAck, + ), ); return "ack"; } @@ -295,7 +340,10 @@ export class NatsBroker { await params.onMessage(decoded); await params.dedupe.markSeen(decoded.msgId, params.consumerId); this.failedDeliveries.delete(`${params.consumerId}:${decoded.msgId}`); - await this.publishAck(ackSubject, createAck(decoded.msgId, params.consumerId, "ack")); + await this.publishAck( + ackSubject, + await this.createDeliveryAck(decoded, params.consumerId, "ack", undefined, params.signAck), + ); return "ack"; } catch (err) { const reason = err instanceof Error ? err.message : "handler-failed"; @@ -306,10 +354,22 @@ export class NatsBroker { if (msgId !== "unknown" && failures >= maxPoisonAttempts) { await params.dedupe.markSeen(msgId, params.consumerId); this.failedDeliveries.delete(key); - await this.publishAck(ackSubject, createAck(msgId, params.consumerId, "nack", `poison-message:${reason}`)); + const ack = decodedEnvelope + ? await this.createDeliveryAck( + decodedEnvelope, + params.consumerId, + "nack", + `poison-message:${reason}`, + params.signAck, + ) + : createAck(msgId, params.consumerId, "nack", `poison-message:${reason}`); + await this.publishAck(ackSubject, ack); return "ack"; } - await this.publishAck(ackSubject, createAck(msgId, params.consumerId, "nack", reason)); + const ack = decodedEnvelope + ? await this.createDeliveryAck(decodedEnvelope, params.consumerId, "nack", reason, params.signAck) + : createAck(msgId, params.consumerId, "nack", reason); + await this.publishAck(ackSubject, ack); return "retry"; } } @@ -390,6 +450,8 @@ export class NatsBroker { /** Optional ingress authorizer; when set, envelopes are authorized before delivery * (wire @murmurv2/federation authorizeInbound here behind MURMUR_ENFORCE_AUTH). */ authorize?: InboundAuthorizer; + /** Signs ACKs with the receiving agent's long-term signing key. */ + signAck?: AckSigner; }): Promise { await this.connect(); @@ -496,13 +558,18 @@ export class NatsBroker { outbox: OutboxStore; ackSubject: string; consumerId?: string; + verifyAck?: AckVerifier; + requireSignedAcks?: boolean; + maxAckAgeMs?: number; + maxFutureSkewMs?: number; + onInvalidAck?: (event: InvalidAckEvent) => void; }): Promise { await this.connect(); if (this.js) { const consumerId = params.consumerId ?? `${params.ackSubject.replaceAll(".", "-")}-consumer`; return this.consumeJetStream(params.ackSubject, consumerId, async (data) => { - await this.processAckFrame(data, params.outbox); + await this.processAckFrame(data, params); }); } @@ -510,7 +577,7 @@ export class NatsBroker { (async () => { for await (const m of sub) { - await this.processAckFrame(m.data, params.outbox); + await this.processAckFrame(m.data, params); } })().catch((err) => { const e = err instanceof Error ? err : new Error(String(err)); @@ -520,30 +587,126 @@ export class NatsBroker { return sub; } - private async processAckFrame(data: Uint8Array, outbox: OutboxStore): Promise { + private invalidAck( + params: { + onInvalidAck?: (event: InvalidAckEvent) => void; + }, + reason: string, + candidate?: Partial, + ): void { + this.invalidAckCounts.set(reason, (this.invalidAckCounts.get(reason) ?? 0) + 1); + const event = { + reason, + ...(typeof candidate?.msgId === "string" ? { msgId: candidate.msgId } : {}), + ...(typeof candidate?.senderAgentId === "string" ? { senderAgentId: candidate.senderAgentId } : {}), + }; + console.warn("[NatsBroker.security] invalid ACK rejected", event); + params.onInvalidAck?.(event); + } + + private rememberAckNonce(nonce: string): boolean { + if (this.seenAckNonces.has(nonce)) return false; + this.seenAckNonces.add(nonce); + if (this.seenAckNonces.size > 10_000) { + const oldest = this.seenAckNonces.values().next(); + if (!oldest.done) this.seenAckNonces.delete(oldest.value); + } + return true; + } + + private async processAckFrame( + data: Uint8Array, + params: { + outbox: OutboxStore; + verifyAck?: AckVerifier; + requireSignedAcks?: boolean; + maxAckAgeMs?: number; + maxFutureSkewMs?: number; + onInvalidAck?: (event: InvalidAckEvent) => void; + }, + ): Promise { try { - const decoded = JSON.parse(this.sc.decode(data)) as AckV1; - if (typeof decoded.msgId !== "string" || decoded.msgId.length === 0) return; + const decoded = JSON.parse(this.sc.decode(data)) as unknown; - if (decoded.status === "ack") { - await outbox.markAcked(decoded.msgId); + if (!isSignedAckV1(decoded)) { + const legacy = decoded as Partial; + if (params.requireSignedAcks === true) { + this.invalidAck(params, "unsigned-or-malformed", { + msgId: typeof legacy?.msgId === "string" ? legacy.msgId : undefined, + }); + return; + } + if (typeof legacy?.msgId !== "string" || legacy.msgId.length === 0) return; + if (legacy.status === "ack") { + await params.outbox.markAcked(legacy.msgId); + } else if (legacy.status === "nack") { + await params.outbox.markFailed( + legacy.msgId, + legacy.reason ?? "nack", + new Date().toISOString(), + ); + } return; } - if (decoded.status === "nack") { - await outbox.markFailed( - decoded.msgId, - decoded.reason ?? "nack", - new Date().toISOString(), - ); + const record = await params.outbox.getOutboxRecord(decoded.msgId); + if (!record) { + this.invalidAck(params, "unknown-message", decoded); + return; } - } catch (err) { - const e = err instanceof Error ? err : new Error(String(err)); - console.error("[NatsBroker.startAckCorrelation] malformed ack frame", { - message: e.message, - stack: e.stack, - raw: this.sc.decode(data), - }); + if (record.status !== "sent") { + this.invalidAck(params, "message-not-in-flight", decoded); + return; + } + if (decoded.messageDigest !== envelopeDigest(record.envelope)) { + this.invalidAck(params, "message-digest-mismatch", decoded); + return; + } + if (decoded.conversationId !== record.envelope.conversationId) { + this.invalidAck(params, "conversation-mismatch", decoded); + return; + } + if (decoded.recipientAgentId !== record.envelope.senderAgentId) { + this.invalidAck(params, "recipient-mismatch", decoded); + return; + } + if (!record.envelope.recipients.includes(decoded.senderAgentId)) { + this.invalidAck(params, "unexpected-peer", decoded); + return; + } + + const atMs = Date.parse(decoded.at); + const now = Date.now(); + const maxAckAgeMs = params.maxAckAgeMs ?? 5 * 60_000; + const maxFutureSkewMs = params.maxFutureSkewMs ?? 30_000; + if (atMs < now - maxAckAgeMs || atMs > now + maxFutureSkewMs) { + this.invalidAck(params, "timestamp-out-of-window", decoded); + return; + } + if (!params.verifyAck || !(await params.verifyAck(decoded))) { + this.invalidAck(params, "signature-invalid", decoded); + return; + } + if (!this.rememberAckNonce(`${decoded.senderAgentId}:${decoded.nonce}`)) { + this.invalidAck(params, "nonce-replay", decoded); + return; + } + + if (decoded.status === "ack") { + const result = await params.outbox.applyAckTransition(decoded.msgId, "ack"); + if (result !== "applied") this.invalidAck(params, `transition-${result}`, decoded); + return; + } + + const result = await params.outbox.applyAckTransition( + decoded.msgId, + "nack", + decoded.reason ?? "nack", + new Date().toISOString(), + ); + if (result !== "applied") this.invalidAck(params, `transition-${result}`, decoded); + } catch { + this.invalidAck(params, "processing-error"); } } diff --git a/packages/core/schema/protocol-v1.schema.json b/packages/core/schema/protocol-v1.schema.json index d552088..30f9b0f 100644 --- a/packages/core/schema/protocol-v1.schema.json +++ b/packages/core/schema/protocol-v1.schema.json @@ -50,6 +50,35 @@ "at": { "type": "string", "format": "date-time" } } }, + "SignedAckV1": { + "description": "Ed25519-signed delivery acknowledgement bound to the original envelope and expected peer.", + "type": "object", + "required": [ + "ackVersion", + "msgId", + "messageDigest", + "conversationId", + "senderAgentId", + "recipientAgentId", + "status", + "at", + "nonce", + "signature" + ], + "properties": { + "ackVersion": { "const": "1.0" }, + "msgId": { "type": "string", "minLength": 1 }, + "messageDigest": { "type": "string", "pattern": "^sha256:[a-f0-9]{64}$" }, + "conversationId": { "type": "string", "minLength": 1 }, + "senderAgentId": { "type": "string", "minLength": 1 }, + "recipientAgentId": { "type": "string", "minLength": 1 }, + "status": { "enum": ["ack", "nack"] }, + "reason": { "type": "string" }, + "at": { "type": "string", "format": "date-time" }, + "nonce": { "type": "string", "minLength": 1 }, + "signature": { "type": "string", "minLength": 1 } + } + }, "PresenceFrameV1": { "description": "Discovery presence announcement. PUBLIC metadata only; carries no secret. A signed wrapper (SignedPresenceFrameV1) proves integrity, NOT that agentId is who it claims — trust is established out-of-band at operator promotion.", "type": "object", diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index af54641..5fb6408 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -45,6 +45,57 @@ export interface AckV1 { at: string; } +export interface SignedAckV1 { + ackVersion: "1.0"; + msgId: string; + messageDigest: string; + conversationId: string; + senderAgentId: string; + recipientAgentId: string; + status: "ack" | "nack"; + reason?: string; + at: string; + nonce: string; + signature: string; +} + +export type UnsignedAckV1 = Omit; + +export const envelopeDigest = (envelope: EnvelopeV1): string => + `sha256:${createHash("sha256").update(stableEnvelopePayload(envelope)).digest("hex")}`; + +export const stableAckPayload = (ack: UnsignedAckV1 | SignedAckV1): string => + JSON.stringify({ + ackVersion: ack.ackVersion, + msgId: ack.msgId, + messageDigest: ack.messageDigest, + conversationId: ack.conversationId, + senderAgentId: ack.senderAgentId, + recipientAgentId: ack.recipientAgentId, + status: ack.status, + ...(ack.reason !== undefined ? { reason: ack.reason } : {}), + at: ack.at, + nonce: ack.nonce, + }); + +export const isSignedAckV1 = (value: unknown): value is SignedAckV1 => { + if (!value || typeof value !== "object") return false; + const ack = value as Record; + return ( + ack.ackVersion === "1.0" && + typeof ack.msgId === "string" && ack.msgId.length > 0 && + typeof ack.messageDigest === "string" && /^sha256:[a-f0-9]{64}$/.test(ack.messageDigest) && + typeof ack.conversationId === "string" && ack.conversationId.length > 0 && + typeof ack.senderAgentId === "string" && ack.senderAgentId.length > 0 && + typeof ack.recipientAgentId === "string" && ack.recipientAgentId.length > 0 && + (ack.status === "ack" || ack.status === "nack") && + (ack.reason === undefined || typeof ack.reason === "string") && + typeof ack.at === "string" && !Number.isNaN(Date.parse(ack.at)) && + typeof ack.nonce === "string" && ack.nonce.length > 0 && + typeof ack.signature === "string" && ack.signature.length > 0 + ); +}; + /** * Canonical signing payload for an EnvelopeV1 — the SINGLE SOURCE OF TRUTH every * signer and verifier MUST use, so signatures interoperate across the whole mesh @@ -174,10 +225,17 @@ export interface OutboxStore { enqueue(subject: string, envelope: EnvelopeV1): Promise; claimDue(limit?: number): Promise; listInFlight?(): Promise; + getOutboxRecord(msgId: string): Promise; markSent(msgId: string): Promise; markAcked(msgId: string): Promise; markFailed(msgId: string, error: string, nextAttemptAt: string): Promise; markDlq(msgId: string, error: string): Promise; + applyAckTransition( + msgId: string, + status: AckV1["status"], + error?: string, + nextAttemptAt?: string, + ): Promise<"applied" | "not-found" | "not-in-flight">; requeueStaleSent?(ackTimeoutMs: number, reason?: string): Promise; } @@ -237,6 +295,11 @@ export class JsonFileOutboxStore implements OutboxStore { return state.records.filter((r) => r.status === "sent"); } + async getOutboxRecord(msgId: string): Promise { + const state = await this.load(); + return state.records.find((record) => record.msgId === msgId); + } + async markSent(msgId: string): Promise { const state = await this.load(); const row = state.records.find((r) => r.msgId === msgId); @@ -281,6 +344,31 @@ export class JsonFileOutboxStore implements OutboxStore { await this.save(state); } + async applyAckTransition( + msgId: string, + status: AckV1["status"], + error = "nack", + nextAttemptAt = new Date().toISOString(), + ): Promise<"applied" | "not-found" | "not-in-flight"> { + const state = await this.load(); + const row = state.records.find((record) => record.msgId === msgId); + if (!row) return "not-found"; + if (row.status !== "sent") return "not-in-flight"; + + const now = new Date().toISOString(); + if (status === "ack") { + row.status = "acked"; + } else { + row.status = "failed"; + row.lastError = error; + row.nextAttemptAt = nextAttemptAt; + } + row.updatedAt = now; + row.version = (row.version ?? 0) + 1; + await this.save(state); + return "applied"; + } + async requeueStaleSent(ackTimeoutMs: number, reason = "ack-timeout"): Promise { const state = await this.load(); const now = Date.now(); @@ -384,6 +472,10 @@ export class SQLiteDedupeOutboxStore implements DedupeStore, OutboxStore { return rows.map((row) => this.toOutboxRecord(row)); } + async getOutboxRecord(msgId: string): Promise { + return this.getOutboxRow(msgId); + } + async markSent(msgId: string): Promise { await this.updateOutboxOptimistic(msgId, (row) => ({ status: "sent", @@ -416,6 +508,26 @@ export class SQLiteDedupeOutboxStore implements DedupeStore, OutboxStore { })); } + async applyAckTransition( + msgId: string, + status: AckV1["status"], + error = "nack", + nextAttemptAt = new Date().toISOString(), + ): Promise<"applied" | "not-found" | "not-in-flight"> { + const now = new Date().toISOString(); + const nextStatus = status === "ack" ? "acked" : "failed"; + const nextError = status === "ack" ? null : error; + const changed = this.db + .prepare( + `UPDATE outbox + SET status = ?, last_error = ?, next_attempt_at = ?, updated_at = ?, version = version + 1 + WHERE msg_id = ? AND status = 'sent'`, + ) + .run(nextStatus, nextError, nextAttemptAt, now, msgId); + if (changed.changes > 0) return "applied"; + return this.getOutboxRow(msgId) ? "not-in-flight" : "not-found"; + } + async requeueStaleSent(ackTimeoutMs: number, reason = "ack-timeout"): Promise { const threshold = new Date(Date.now() - ackTimeoutMs).toISOString(); const res = this.db @@ -1228,6 +1340,24 @@ export const createAck = ( at: new Date().toISOString(), }); +export const createBoundAck = ( + envelope: EnvelopeV1, + consumerId: string, + status: SignedAckV1["status"], + reason?: string, +): UnsignedAckV1 => ({ + ackVersion: "1.0", + msgId: envelope.msgId, + messageDigest: envelopeDigest(envelope), + conversationId: envelope.conversationId, + senderAgentId: consumerId, + recipientAgentId: envelope.senderAgentId, + status, + ...(reason !== undefined ? { reason } : {}), + at: new Date().toISOString(), + nonce: randomUUID(), +}); + export const computeBackoffMs = (attempt: number, baseMs = 500, maxMs = 60_000): number => { const raw = baseMs * Math.pow(2, Math.max(0, attempt - 1)); return Math.min(maxMs, raw); diff --git a/packages/core/test/conformance.test.mjs b/packages/core/test/conformance.test.mjs index cc01b6d..b790219 100644 --- a/packages/core/test/conformance.test.mjs +++ b/packages/core/test/conformance.test.mjs @@ -2,10 +2,11 @@ // Validates fixtures against the machine-readable JSON Schema (schema/protocol-v1.schema.json) // AND asserts the schema and the runtime guards AGREE on every structural case — so the spec and // the implementation can't silently drift, and a cross-language implementer can trust either as -// the contract. Covered wire types: EnvelopeV1, AckV1, PresenceFrameV1, SignedPresenceFrameV1, +// the contract. Covered wire types: EnvelopeV1, AckV1, SignedAckV1, PresenceFrameV1, SignedPresenceFrameV1, // StreamStart/StreamChunk/StreamEnd (+ the discriminated StreamFrame union). No external validator // dep: a tiny subset validator interprets the flat protocol $defs -// (type incl. boolean / required / const / enum / minLength / minItems / items / exclusiveMinimum / oneOf). +// (type incl. boolean / required / const / enum / minLength / pattern / minItems / items / +// exclusiveMinimum / oneOf). // The only runtime-only check is `format: date-time` validity, which Draft 2020-12 treats as an // advisory annotation (not an assertion); the guards enforce it via Date.parse and it sits outside // the agreement matrices (documented per-type). Everything else — including ttlMs > 0 and non-empty @@ -17,6 +18,7 @@ import path from "node:path"; import { fileURLToPath } from "node:url"; import { isEnvelopeV1, + isSignedAckV1, isPresenceFrameV1, isSignedPresenceFrameV1, isStreamStart, @@ -60,6 +62,7 @@ function validate(def, value, p = "$") { case "string": if (typeof value !== "string") errs.push(`${p}: string`); else if (def.minLength != null && value.length < def.minLength) errs.push(`${p}: minLength`); + else if (def.pattern != null && !(new RegExp(def.pattern)).test(value)) errs.push(`${p}: pattern`); break; case "number": if (typeof value !== "number") errs.push(`${p}: number`); @@ -80,6 +83,7 @@ function validate(def, value, p = "$") { } const envelopeOk = (v) => validate(schema.$defs.EnvelopeV1, v).length === 0; const ackOk = (v) => validate(schema.$defs.AckV1, v).length === 0; +const signedAckOk = (v) => validate(schema.$defs.SignedAckV1, v).length === 0; const presenceOk = (v) => validate(schema.$defs.PresenceFrameV1, v).length === 0; const signedPresenceOk = (v) => validate(schema.$defs.SignedPresenceFrameV1, v).length === 0; const streamStartOk = (v) => validate(schema.$defs.StreamStart, v).length === 0; @@ -105,7 +109,7 @@ const without = (key) => { const e = { ...GOOD }; delete e[key]; return e; }; test("schema bundle is a valid Draft 2020-12 $defs registry", () => { assert.equal(schema.$schema, "https://json-schema.org/draft/2020-12/schema"); - assert.ok(schema.$defs?.EnvelopeV1 && schema.$defs?.AckV1); + assert.ok(schema.$defs?.EnvelopeV1 && schema.$defs?.AckV1 && schema.$defs?.SignedAckV1); assert.equal(schema.$defs.EnvelopeV1.properties.schemaVersion.const, "1.0"); }); @@ -158,6 +162,44 @@ test("AckV1: required fields + status enum", () => { assert.equal(ackOk({ msgId: "m", status: "ack", at: "2026-06-21T00:00:00Z" }), false); }); +const GOOD_SIGNED_ACK = Object.freeze({ + ackVersion: "1.0", + msgId: "m", + messageDigest: `sha256:${"a".repeat(64)}`, + conversationId: "conversation", + senderAgentId: "agent-b", + recipientAgentId: "agent-a", + status: "ack", + at: "2026-06-21T00:00:00Z", + nonce: "nonce", + signature: "signature", +}); + +test("a conformant SignedAckV1 passes BOTH the schema and isSignedAckV1", () => { + assert.equal(signedAckOk(GOOD_SIGNED_ACK), true); + assert.equal(isSignedAckV1(GOOD_SIGNED_ACK), true); + assert.equal(signedAckOk({ ...GOOD_SIGNED_ACK, reason: "duplicate-ignored", futureFieldV2: "x" }), true); + assert.equal(isSignedAckV1({ ...GOOD_SIGNED_ACK, reason: "duplicate-ignored", futureFieldV2: "x" }), true); +}); + +test("schema and isSignedAckV1 agree on every structural violation", () => { + const bad = [ + ["ackVersion const", { ...GOOD_SIGNED_ACK, ackVersion: "2.0" }], + ["missing msgId", omit(GOOD_SIGNED_ACK, "msgId")], + ["bad digest", { ...GOOD_SIGNED_ACK, messageDigest: "sha256:nope" }], + ["empty conversation", { ...GOOD_SIGNED_ACK, conversationId: "" }], + ["empty sender", { ...GOOD_SIGNED_ACK, senderAgentId: "" }], + ["empty recipient", { ...GOOD_SIGNED_ACK, recipientAgentId: "" }], + ["bad status", { ...GOOD_SIGNED_ACK, status: "maybe" }], + ["empty nonce", { ...GOOD_SIGNED_ACK, nonce: "" }], + ["empty signature", { ...GOOD_SIGNED_ACK, signature: "" }], + ]; + for (const [label, ack] of bad) { + assert.equal(signedAckOk(ack), false, `schema must reject: ${label}`); + assert.equal(isSignedAckV1(ack), false, `isSignedAckV1 must reject: ${label}`); + } +}); + // Note: `createdAt` carries JSON-Schema `format: date-time` (advisory) and is enforced at // runtime by isEnvelopeV1 via Date.parse — that one check lives in the runtime guard, not in // this minimal validator, so date validity is intentionally not part of the agreement matrix. diff --git a/packages/core/test/stable-envelope-payload.test.mjs b/packages/core/test/stable-envelope-payload.test.mjs index a9b3c11..a0788b0 100644 --- a/packages/core/test/stable-envelope-payload.test.mjs +++ b/packages/core/test/stable-envelope-payload.test.mjs @@ -4,7 +4,7 @@ // signature interop. If this test fails, EVERY signer must change together (wire-breaking). import test from "node:test"; import assert from "node:assert/strict"; -import { stableEnvelopePayload } from "../dist/src/index.js"; +import { stableAckPayload, stableEnvelopePayload } from "../dist/src/index.js"; const ENV = Object.freeze({ schemaVersion: "1.0", @@ -65,3 +65,25 @@ test("stableEnvelopePayload copies recipients (no shared mutable reference)", () assert.ok(out.includes('"recipients":["agent-b","agent-c"]')); assert.ok(!out.includes("agent-d")); }); + +test("stableAckPayload emits an exact canonical string and excludes the signature", () => { + const ack = { + ackVersion: "1.0", + msgId: "m1", + messageDigest: `sha256:${"a".repeat(64)}`, + conversationId: "c1", + senderAgentId: "agent-b", + recipientAgentId: "agent-a", + status: "nack", + reason: "retry", + at: "2026-06-22T00:00:01.000Z", + nonce: "nonce-1", + signature: "SIG-SHOULD-BE-EXCLUDED", + }; + + assert.equal( + stableAckPayload(ack), + `{"ackVersion":"1.0","msgId":"m1","messageDigest":"sha256:${"a".repeat(64)}","conversationId":"c1","senderAgentId":"agent-b","recipientAgentId":"agent-a","status":"nack","reason":"retry","at":"2026-06-22T00:00:01.000Z","nonce":"nonce-1"}`, + ); + assert.ok(!stableAckPayload(ack).includes("signature")); +}); diff --git a/scripts/agent-config-init.mjs b/scripts/agent-config-init.mjs index 75f78a2..b20f4e3 100644 --- a/scripts/agent-config-init.mjs +++ b/scripts/agent-config-init.mjs @@ -54,6 +54,11 @@ const run = async () => { dataDir, cryptoProvider: getCryptoProvider().name, keys: { encryption, signing }, + ackSecurity: { + emitSigned: true, + requireSigned: false, + maxAgeMs: 300000, + }, peers: {}, }; diff --git a/scripts/murmur-daemon.mjs b/scripts/murmur-daemon.mjs index 6b2c455..94ef9de 100644 --- a/scripts/murmur-daemon.mjs +++ b/scripts/murmur-daemon.mjs @@ -7,8 +7,14 @@ import { DatabaseSync } from "node:sqlite"; import path from "node:path"; import { setTimeout as sleep } from "node:timers/promises"; import { NatsBroker } from "@murmurv2/broker-nats"; -import { ChannelRosterStore, SQLiteDedupeOutboxStore, SQLiteMessageStore, stableEnvelopePayload } from "@murmurv2/core"; -import { decryptPayload, verifyEnvelopeSignature } from "@murmurv2/security"; +import { + ChannelRosterStore, + SQLiteDedupeOutboxStore, + SQLiteMessageStore, + stableAckPayload, + stableEnvelopePayload, +} from "@murmurv2/core"; +import { decryptPayload, signEnvelope, verifyEnvelopeSignature } from "@murmurv2/security"; import { NotifyQueue, flushNotifyQueue, normalizeNotifyTargets } from "./notify-router.mjs"; import { createChannelThreadStartBindingResolver, createCodexAppServerInjector } from "./codex-app-server-wake.mjs"; import { startJetStreamAdvisoryDlqIfEnabled } from "./murmur-jetstream-advisory.mjs"; @@ -61,6 +67,17 @@ const ackTimeoutMs = optionalPositiveInteger( "ack-timeout-ms", firstDefined(streamingConfig.ackTimeoutMs, process.env.MURMUR_ACK_TIMEOUT_MS), ) ?? 15_000; +const ackSecurityConfig = config.ackSecurity || {}; +const emitSignedAcks = process.env.MURMUR_EMIT_SIGNED_ACKS !== undefined + ? process.env.MURMUR_EMIT_SIGNED_ACKS !== "0" + : ackSecurityConfig.emitSigned ?? true; +const requireSignedAcks = process.env.MURMUR_REQUIRE_SIGNED_ACKS !== undefined + ? process.env.MURMUR_REQUIRE_SIGNED_ACKS === "1" + : ackSecurityConfig.requireSigned ?? false; +const maxAckAgeMs = optionalPositiveInteger( + "ack-max-age-ms", + firstDefined(ackSecurityConfig.maxAgeMs, process.env.MURMUR_ACK_MAX_AGE_MS), +) ?? 5 * 60_000; const ackWindowEnabled = ackWindowConfig.enabled ?? process.env.MURMUR_STREAM_ACK_WINDOW === "1"; const ackWindow = ackWindowEnabled ? { @@ -98,6 +115,11 @@ log("info", "Daemon starting", { jetstreamMaxDeliver, jetstreamAckWaitMs, ackTimeoutMs, + ackSecurity: { + emitSigned: emitSignedAcks, + requireSigned: requireSignedAcks, + maxAgeMs: maxAckAgeMs, + }, ackWindow, notifyTargets: effectiveNotifyTargets.map((t) => `${t.type}:${t.channel}`), notifyFallbackFromEnv: envTelegramFallback.length > 0, @@ -136,6 +158,17 @@ const broker = new NatsBroker({ jetstreamAckWaitMs, }); +const signAck = async (unsignedAck) => ({ + ...unsignedAck, + signature: await signEnvelope(stableAckPayload(unsignedAck), keys.signing.privateKey), +}); + +const verifyAck = async (ack) => { + const peer = peers[ack.senderAgentId]; + if (!peer?.signing?.publicKey) return false; + return verifyEnvelopeSignature(stableAckPayload(ack), ack.signature, peer.signing.publicKey); +}; + const durableSafe = (value) => value.replace(/[^A-Za-z0-9_-]/g, "-"); const inboundCursor = () => { @@ -312,7 +345,13 @@ try { await broker.connect(); log("info", "NATS connected", { url: natsUrl }); - await broker.subscribeWithAck({ subject, consumerId: agentId, dedupe: store, onMessage }); + await broker.subscribeWithAck({ + subject, + consumerId: agentId, + dedupe: store, + onMessage, + ...(emitSignedAcks ? { signAck } : {}), + }); log("info", "Subscribed", { subject }); // Also subscribe to proxy subjects (agents without their own daemon) @@ -334,8 +373,20 @@ try { log("info", "Subscribed (proxy)", { subject: ps }); } - await broker.startAckCorrelation({ outbox: store, ackSubject: `ack.${agentId}`, consumerId: `${agentId}-ack` }); - log("info", "ACK correlation started", { ackSubject: `ack.${agentId}` }); + await broker.startAckCorrelation({ + outbox: store, + ackSubject: `ack.${agentId}`, + consumerId: `${agentId}-ack`, + verifyAck, + requireSignedAcks, + maxAckAgeMs, + onInvalidAck: (event) => log("warn", "Invalid ACK rejected", event), + }); + log("info", "ACK correlation started", { + ackSubject: `ack.${agentId}`, + emitSignedAcks, + requireSignedAcks, + }); await startJetStreamAdvisoryDlqIfEnabled({ broker, outbox: store, diff --git a/tests/broker-signed-ack.test.mjs b/tests/broker-signed-ack.test.mjs new file mode 100644 index 0000000..9c32d38 --- /dev/null +++ b/tests/broker-signed-ack.test.mjs @@ -0,0 +1,190 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { StringCodec } from "nats"; +import { NatsBroker } from "../packages/broker-nats/dist/src/index.js"; +import { + JsonFileOutboxStore, + createAck, + createBoundAck, + isSignedAckV1, + stableAckPayload, +} from "../packages/core/dist/src/index.js"; +import { + createSigningKeyPair, + signEnvelope, + verifyEnvelopeSignature, +} from "../packages/security/dist/src/index.js"; + +const sc = StringCodec(); + +const envelope = { + schemaVersion: "1.0", + msgId: "msg-signed-ack", + conversationId: "conv-signed-ack", + senderAgentId: "agent-sender", + recipients: ["agent-receiver"], + createdAt: new Date().toISOString(), + payloadCiphertext: Buffer.from("encrypted-message").toString("base64"), + payloadNonce: "nonce", + signature: "envelope-signature", +}; + +const createSentOutbox = async () => { + const dir = mkdtempSync(join(tmpdir(), "murmur-signed-ack-")); + const outbox = new JsonFileOutboxStore(join(dir, "outbox.json")); + await outbox.enqueue("msg.agent-receiver", envelope); + await outbox.markSent(envelope.msgId); + return outbox; +}; + +const signAck = async (unsignedAck, privateKey) => ({ + ...unsignedAck, + signature: await signEnvelope(stableAckPayload(unsignedAck), privateKey), +}); + +const processAck = async (broker, outbox, ack, verifyAck, events = []) => { + await broker.processAckFrame(sc.encode(JSON.stringify(ack)), { + outbox, + requireSignedAcks: true, + verifyAck, + onInvalidAck: (event) => events.push(event), + }); + return events; +}; + +test("subscribeWithAck emits a signed message-bound ACK when a signer is supplied", async () => { + const signing = await createSigningKeyPair(); + const published = []; + const fakeSub = { + async *[Symbol.asyncIterator]() { + yield { data: sc.encode(JSON.stringify(envelope)) }; + }, + }; + const broker = new NatsBroker({ url: "nats://example.invalid" }); + broker.nc = { + subscribe() { return fakeSub; }, + publish(subject, data) { published.push({ subject, ack: JSON.parse(sc.decode(data)) }); }, + async drain() {}, + }; + + await broker.subscribeWithAck({ + subject: "msg.agent-receiver", + consumerId: "agent-receiver", + dedupe: { + async seen() { return false; }, + async markSeen() {}, + }, + onMessage: async () => {}, + signAck: (unsigned) => signAck(unsigned, signing.privateKey), + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + + assert.equal(published[0].subject, "ack.agent-sender"); + assert.equal(isSignedAckV1(published[0].ack), true); + assert.equal(published[0].ack.messageDigest, createBoundAck(envelope, "agent-receiver", "ack").messageDigest); + assert.equal( + await verifyEnvelopeSignature( + stableAckPayload(published[0].ack), + published[0].ack.signature, + signing.publicKey, + ), + true, + ); +}); + +test("strict ACK correlation rejects unsigned ACKs without changing the outbox", async () => { + const outbox = await createSentOutbox(); + const broker = new NatsBroker({ url: "nats://example.invalid" }); + const events = await processAck( + broker, + outbox, + createAck(envelope.msgId, "agent-receiver", "ack"), + async () => true, + ); + + assert.equal((await outbox.getOutboxRecord(envelope.msgId)).status, "sent"); + assert.equal(events[0].reason, "unsigned-or-malformed"); + assert.equal(broker.getAckSecurityMetrics()["unsigned-or-malformed"], 1); +}); + +test("a valid signed ACK is bound to the pending message and expected peer", async () => { + const outbox = await createSentOutbox(); + const signing = await createSigningKeyPair(); + const unsigned = createBoundAck(envelope, "agent-receiver", "ack"); + const ack = await signAck(unsigned, signing.privateKey); + const broker = new NatsBroker({ url: "nats://example.invalid" }); + + await processAck( + broker, + outbox, + ack, + (candidate) => verifyEnvelopeSignature( + stableAckPayload(candidate), + candidate.signature, + signing.publicKey, + ), + ); + + assert.equal((await outbox.getOutboxRecord(envelope.msgId)).status, "acked"); +}); + +test("wrong peer, conversation, recipient, digest, time, and signature cannot change outbox state", async () => { + const signing = await createSigningKeyPair(); + const wrongSigning = await createSigningKeyPair(); + const cases = [ + { name: "unexpected-peer", patch: { senderAgentId: "agent-attacker" }, signer: signing }, + { name: "conversation-mismatch", patch: { conversationId: "other-conversation" }, signer: signing }, + { name: "recipient-mismatch", patch: { recipientAgentId: "other-recipient" }, signer: signing }, + { name: "message-digest-mismatch", patch: { messageDigest: `sha256:${"0".repeat(64)}` }, signer: signing }, + { name: "timestamp-out-of-window", patch: { at: "2020-01-01T00:00:00.000Z" }, signer: signing }, + { name: "signature-invalid", patch: {}, signer: wrongSigning }, + ]; + + for (const item of cases) { + const outbox = await createSentOutbox(); + const broker = new NatsBroker({ url: "nats://example.invalid" }); + const unsigned = { ...createBoundAck(envelope, "agent-receiver", "ack"), ...item.patch }; + const ack = await signAck(unsigned, item.signer.privateKey); + const events = await processAck( + broker, + outbox, + ack, + (candidate) => verifyEnvelopeSignature( + stableAckPayload(candidate), + candidate.signature, + signing.publicKey, + ), + ); + + assert.equal((await outbox.getOutboxRecord(envelope.msgId)).status, "sent", item.name); + assert.equal(events[0].reason, item.name); + } +}); + +test("replaying a signed ACK cannot create another outbox transition", async () => { + const outbox = await createSentOutbox(); + const signing = await createSigningKeyPair(); + const ack = await signAck( + createBoundAck(envelope, "agent-receiver", "ack"), + signing.privateKey, + ); + const broker = new NatsBroker({ url: "nats://example.invalid" }); + const verifyAck = (candidate) => verifyEnvelopeSignature( + stableAckPayload(candidate), + candidate.signature, + signing.publicKey, + ); + + await processAck(broker, outbox, ack, verifyAck); + const afterFirst = await outbox.getOutboxRecord(envelope.msgId); + const events = await processAck(broker, outbox, ack, verifyAck); + const afterReplay = await outbox.getOutboxRecord(envelope.msgId); + + assert.equal(afterFirst.status, "acked"); + assert.equal(afterReplay.status, "acked"); + assert.equal(afterReplay.version, afterFirst.version); + assert.equal(events[0].reason, "message-not-in-flight"); +}); diff --git a/tests/core-dedupe-store.test.mjs b/tests/core-dedupe-store.test.mjs index 6c4e5ad..54b9dfc 100644 --- a/tests/core-dedupe-store.test.mjs +++ b/tests/core-dedupe-store.test.mjs @@ -41,3 +41,36 @@ test("SQLiteDedupeOutboxStore markSeen/seen roundtrip", async () => { rmSync(dir, { recursive: true, force: true }); } }); + +test("SQLiteDedupeOutboxStore applies an ACK transition exactly once while sent", async () => { + const dir = mkdtempSync(join(tmpdir(), "murmur-ack-transition-")); + const dbPath = join(dir, "murmur.db"); + const envelope = { + schemaVersion: "1.0", + msgId: "m-ack-once", + conversationId: "conversation", + senderAgentId: "sender", + recipients: ["receiver"], + createdAt: new Date().toISOString(), + payloadCiphertext: "ciphertext", + payloadNonce: "nonce", + signature: "signature", + }; + + try { + const store = new SQLiteDedupeOutboxStore(dbPath); + await store.enqueue("msg.receiver", envelope); + await store.markSent(envelope.msgId); + + assert.equal(await store.applyAckTransition(envelope.msgId, "ack"), "applied"); + const afterFirst = await store.getOutboxRecord(envelope.msgId); + assert.equal(afterFirst.status, "acked"); + + assert.equal(await store.applyAckTransition(envelope.msgId, "ack"), "not-in-flight"); + const afterReplay = await store.getOutboxRecord(envelope.msgId); + assert.equal(afterReplay.status, "acked"); + assert.equal(afterReplay.version, afterFirst.version); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +});