fix(tess): secure checkpoint idempotency

2026-07-13 00:00:03 -05:00
parent 444988d23b
commit cbdc38af95
6 changed files with 4401 additions and 9 deletions
--- a/apps/gateway/src/agent/tess-durable-session.repository.test.ts
+++ b/apps/gateway/src/agent/tess-durable-session.repository.test.ts
@@ -1,7 +1,8 @@
 import { mkdtempSync, rmSync } from 'node:fs';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
-import { eq, sql, interactionInbox } from '@mosaicstack/db';
+import { createHash } from 'node:crypto';
+import { eq, sql, interactionCheckpoints, interactionInbox } from '@mosaicstack/db';
 import { DurableSessionCoordinator, type DurableSessionIdentity } from '@mosaicstack/agent';
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
 import { createPgliteDb, runPgliteMigrations, type DbHandle } from '@mosaicstack/db';
@@ -155,6 +156,104 @@ describe('TessDurableSessionRepository', () => {
    expect(persisted?.content).not.toContain('[REDACTED]');
  }, 30_000);

+  it('fails closed when the configured idempotency secret is unavailable', async () => {
+    const coordinator = new DurableSessionCoordinator(new TessDurableSessionRepository(handle.db));
+    await coordinator.create(IDENTITY);
+    const secret = process.env['BETTER_AUTH_SECRET'];
+    delete process.env['BETTER_AUTH_SECRET'];
+    try {
+      await expect(
+        coordinator.receive({
+          sessionId: IDENTITY.sessionId,
+          idempotencyKey: 'requires-idempotency-secret',
+          correlationId: 'correlation-secret',
+          content: 'sensitive payload',
+        }),
+      ).rejects.toThrow(/required for durable idempotency digests/);
+    } finally {
+      if (secret === undefined) delete process.env['BETTER_AUTH_SECRET'];
+      else process.env['BETTER_AUTH_SECRET'] = secret;
+    }
+  }, 30_000);
+
+  it('uses keyed pre-redaction digests to reject distinct sensitive checkpoint payloads', async () => {
+    const coordinator = new DurableSessionCoordinator(new TessDurableSessionRepository(handle.db));
+    await coordinator.create(IDENTITY);
+    const input = {
+      sessionId: IDENTITY.sessionId,
+      checkpointId: 'checkpoint-secret-conflict',
+      cursor: 'api_key=secret-one',
+      summary: 'bearer secret-one',
+      compactionEpoch: 1,
+    };
+    await coordinator.checkpoint(input);
+
+    await expect(
+      coordinator.checkpoint({
+        ...input,
+        cursor: 'api_key=secret-two',
+        summary: 'bearer secret-two',
+      }),
+    ).rejects.toThrow(/checkpoint identity conflict/);
+
+    const [persisted] = await handle.db
+      .select({
+        digest: interactionCheckpoints.contentDigest,
+        cursor: interactionCheckpoints.cursor,
+      })
+      .from(interactionCheckpoints)
+      .where(eq(interactionCheckpoints.checkpointId, input.checkpointId));
+    expect(persisted?.cursor).not.toContain('secret-one');
+    expect(persisted?.digest).not.toBe(
+      createHash('sha256')
+        .update(JSON.stringify([input.cursor, input.summary]))
+        .digest('hex'),
+    );
+
+    await coordinator.checkpoint({
+      ...input,
+      checkpointId: 'checkpoint-delimiter-conflict',
+      cursor: 'a\u0000b',
+      summary: 'c',
+    });
+    await expect(
+      coordinator.checkpoint({
+        ...input,
+        checkpointId: 'checkpoint-delimiter-conflict',
+        cursor: 'a',
+        summary: 'b\u0000c',
+      }),
+    ).rejects.toThrow(/checkpoint identity conflict/);
+  }, 30_000);
+
+  it('rejects distinct sensitive inbox and outbox payloads under reused idempotency keys', async () => {
+    const coordinator = new DurableSessionCoordinator(new TessDurableSessionRepository(handle.db));
+    await coordinator.create(IDENTITY);
+    const inbox = {
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'inbox-secret-conflict',
+      correlationId: 'correlation-inbox-secret',
+      content: 'api_key=secret-one',
+    };
+    const outbox = {
+      sessionId: IDENTITY.sessionId,
+      idempotencyKey: 'outbox-secret-conflict',
+      correlationId: 'correlation-outbox-secret',
+      channelId: 'cli',
+      kind: 'provider.send',
+      content: 'api_key=secret-one',
+    };
+    await coordinator.receive(inbox);
+    await coordinator.enqueueOutbox(outbox);
+
+    await expect(coordinator.receive({ ...inbox, content: 'api_key=secret-two' })).rejects.toThrow(
+      /idempotency conflict/,
+    );
+    await expect(
+      coordinator.enqueueOutbox({ ...outbox, content: 'api_key=secret-two' }),
+    ).rejects.toThrow(/idempotency conflict/);
+  }, 30_000);
+
  it('rejects database inbox and outbox idempotency-key conflicts', async () => {
    const coordinator = new DurableSessionCoordinator(new TessDurableSessionRepository(handle.db));
    await coordinator.create(IDENTITY);
--- a/apps/gateway/src/agent/tess-durable-session.repository.ts
+++ b/apps/gateway/src/agent/tess-durable-session.repository.ts
@@ -1,4 +1,4 @@
-import { createHash } from 'node:crypto';
+import { createHash, createHmac } from 'node:crypto';
 import { Inject, Injectable } from '@nestjs/common';
 import {
  and,
@@ -126,7 +126,10 @@ export class TessDurableSessionRepository implements DurableSessionStore {
      .limit(1);
    if (!existing[0]) throw new Error(`Durable Tess inbox enqueue failed: ${input.idempotencyKey}`);
    const entry = toInbox(existing[0]);
-    if (!sameInbox(entry, record) || existing[0].contentDigest !== digest) {
+    if (
+      !sameInbox(entry, record) ||
+      !matchesContentDigest(existing[0].contentDigest, input.content)
+    ) {
      throw new Error(`Durable Tess inbox idempotency conflict: ${input.idempotencyKey}`);
    }
    return { accepted: false, status: entry.status };
@@ -213,7 +216,10 @@ export class TessDurableSessionRepository implements DurableSessionStore {
    if (!existing[0])
      throw new Error(`Durable Tess outbox enqueue failed: ${input.idempotencyKey}`);
    const entry = toOutbox(existing[0]);
-    if (!sameOutbox(entry, record) || existing[0].contentDigest !== digest) {
+    if (
+      !sameOutbox(entry, record) ||
+      !matchesContentDigest(existing[0].contentDigest, input.content)
+    ) {
      throw new Error(`Durable Tess outbox idempotency conflict: ${input.idempotencyKey}`);
    }
    return { accepted: false, status: entry.status };
@@ -286,6 +292,9 @@ export class TessDurableSessionRepository implements DurableSessionStore {
  }

  async checkpoint(input: DurableCheckpointInput): Promise<void> {
+    // Compute identity before redaction.  The persisted digest is keyed so a database
+    // reader cannot use it as an offline oracle for sensitive cursor/summary values.
+    const digest = contentDigest(JSON.stringify([input.cursor, input.summary]));
    const checkpoint: DurableCheckpointInput = {
      ...input,
      cursor: redactSensitiveContent(input.cursor).content,
@@ -295,6 +304,7 @@ export class TessDurableSessionRepository implements DurableSessionStore {
      .insert(interactionCheckpoints)
      .values({
        ...checkpoint,
+        contentDigest: digest,
        cursor: seal(checkpoint.cursor),
        summary: seal(checkpoint.summary),
      })
@@ -302,8 +312,21 @@ export class TessDurableSessionRepository implements DurableSessionStore {
      .returning({ checkpointId: interactionCheckpoints.checkpointId });
    if (inserted[0]) return;

-    const existing = await this.findCheckpoint(input.sessionId, input.checkpointId);
-    if (!existing || !sameCheckpoint(existing, checkpoint)) {
+    const existing = await this.db
+      .select()
+      .from(interactionCheckpoints)
+      .where(
+        and(
+          eq(interactionCheckpoints.sessionId, input.sessionId),
+          eq(interactionCheckpoints.checkpointId, input.checkpointId),
+        ),
+      )
+      .limit(1);
+    if (
+      !existing[0] ||
+      !sameCheckpoint(toCheckpoint(existing[0]), checkpoint) ||
+      !matchesCheckpointDigest(existing[0].contentDigest, digest)
+    ) {
      throw new Error(`Durable Tess checkpoint identity conflict: ${input.checkpointId}`);
    }
  }
@@ -384,7 +407,24 @@ export class TessDurableSessionRepository implements DurableSessionStore {
 }

 function contentDigest(content: string): string {
-  return createHash('sha256').update(content).digest('hex');
+  const secret = process.env['BETTER_AUTH_SECRET'];
+  if (!secret) {
+    throw new Error('BETTER_AUTH_SECRET is required for durable idempotency digests');
+  }
+  return `hmac:v1:${createHmac('sha256', secret).update(content).digest('hex')}`;
+}
+
+function matchesContentDigest(stored: string, content: string): boolean {
+  return (
+    stored === contentDigest(content) ||
+    stored === createHash('sha256').update(content).digest('hex')
+  );
+}
+
+function matchesCheckpointDigest(stored: string, digest: string): boolean {
+  // Legacy rows predate any pre-redaction identity and cannot safely prove equality.
+  // Reject rather than let redaction collapse distinct sensitive checkpoint payloads.
+  return stored === digest;
 }

 function sameIdentity(left: DurableSessionIdentity, right: DurableSessionIdentity): boolean {
@@ -422,8 +462,6 @@ function sameCheckpoint(left: DurableCheckpoint, right: DurableCheckpointInput):
  return (
    left.sessionId === right.sessionId &&
    left.checkpointId === right.checkpointId &&
-    left.cursor === right.cursor &&
-    left.summary === right.summary &&
    left.compactionEpoch === right.compactionEpoch
  );
 }