feat(e2e): detect legacy ciphertext rewrites in migration Runs

.github/workflows/tests.yml

@@ -503,6 +503,13 @@ jobs:
         env:
           DATABASE_URL: ${{ steps.db.outputs.database-url }}
 
+      # Deterministic Fake Migrators only: no model credentials or token spend.
+      - name: Run the credentialed Migration Harness self-test
+        if: ${{ !cancelled() }}
+        run: pnpm exec turbo run harness:selftest --filter @cipherstash/e2e --env-mode=loose
+        env:
+          HARNESS_LIVE: '1'
+
       # Guarded on the project being set: if the stack never came up there is
       # nothing to tear down, and an unguarded `-p ""` would mask the real error.
       - name: Stop postgres

e2e/migration-harness/__tests__/no-rewrite-live.test.ts

@@ -0,0 +1,77 @@
+import 'dotenv/config'
+import assert from 'node:assert/strict'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { resolve } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { test } from 'vitest'
+import { createIsolation } from '../isolation.mjs'
+import { createRealDriver } from '../run/real-driver.mjs'
+import { run } from '../run/run.mjs'
+
+const keys = [
+  'CS_WORKSPACE_CRN',
+  'CS_CLIENT_ID',
+  'CS_CLIENT_KEY',
+  'CS_CLIENT_ACCESS_KEY',
+]
+const enabled =
+  process.env.HARNESS_LIVE === '1' && keys.every((key) => process.env[key])
+for (const rewrite of [false, true]) {
+  test.skipIf(!enabled)(
+    `real Run ${rewrite ? 'detects a read rewrite' : 'keeps Legacy Data unchanged through mutations'}`,
+    async () => {
+      const output = await mkdtemp(resolve(tmpdir(), 'mh-no-rewrite-'))
+      const isolation = createIsolation()
+      const manifest = { stack: '^1.2.1', guideHash: null }
+      const replacement =
+        'async get(pk) { if (pk === "legacy") { const customer = await decrypt(await stores.get(pk)); await this.put(customer); }'
+      const script = `import {readFile,writeFile} from "node:fs/promises"; const p="src/app.mjs"; let s=await readFile(p,"utf8"); s=s.replace("async get(pk) {", ${JSON.stringify(replacement)}); await writeFile(p,s); await writeFile("friction.json","[]")`
+      const tool = rewrite
+        ? `node --input-type=module -e '${script}'`
+        : `printf '[]' > friction.json`
+      const adapter = resolve(output, 'adapter.mjs')
+      await (await import('node:fs/promises')).writeFile(
+        adapter,
+        `
+        import {createInterface} from 'node:readline';
+        const lines=createInterface({input:process.stdin});
+        console.log(JSON.stringify({id:'fake',method:'exec',command:['sh','-c',${JSON.stringify(tool)}]}));
+        for await(const line of lines){ const response=JSON.parse(line); process.exitCode=response.result.exitCode; break; }
+        lines.close();
+      `,
+      )
+      const command = `node '${adapter}'`
+      const driver = createRealDriver({
+        id: `live-${Date.now()}-${rewrite}`,
+        root: fileURLToPath(new URL('../../..', import.meta.url)),
+        output,
+        command,
+        isolation,
+        credentials: Object.fromEntries(
+          keys.map((key) => [key, process.env[key] ?? '']),
+        ),
+        manifest,
+      })
+      try {
+        const report = await run({
+          id: 'live-no-rewrite',
+          manifest,
+          driver,
+          supervision: isolation,
+        })
+        const check = report.evidence.find(
+          (e) => e.checkId === 'no-rewrite.legacy-bytes',
+        )
+        assert.ok(check, JSON.stringify(report))
+        assert.equal(Object(check.payload).passed, !rewrite)
+        if (rewrite)
+          assert.match(JSON.stringify(check.payload), /legacy.*__source/)
+      } finally {
+        await driver.close()
+        await rm(output, { recursive: true, force: true })
+      }
+    },
+    600_000,
+  )
+}

e2e/migration-harness/__tests__/no-rewrite.test.ts

@@ -0,0 +1,102 @@
+import assert from 'node:assert/strict'
+import { test } from 'vitest'
+import { requiredOracleChecks } from '../run/migration-checks.mjs'
+import { noRewrite } from '../run/no-rewrite.mjs'
+import { run } from '../run/run.mjs'
+
+const baseline = {
+  dynamo: {
+    Item: {
+      pk: { S: 'legacy' },
+      email__source: { S: 'cipher' },
+      email__hmac: { S: 'term' },
+    },
+  },
+  postgres: [
+    {
+      pk: 'legacy',
+      email: { c: 'pg-cipher', hm: 'pg-term' },
+      notes: { c: 'note' },
+    },
+  ],
+}
+for (const changed of [false, true]) {
+  test(`Run records ${changed ? 'rewritten' : 'unchanged'} Legacy Data`, async () => {
+    const current = structuredClone(baseline)
+    if (changed) current.dynamo.Item.email__source.S = 'rewritten'
+    const report = await run({
+      id: 'rewrite',
+      manifest: { guideHash: 'guide' },
+      driver: {
+        provision: async () => {},
+        installUnmigrated: async () => {},
+        initializeLegacyData: async () => {},
+        snapshot: async () => {},
+        migrate: async () => ({ exitCode: 0 }),
+        friction: async () => [],
+        installAppHost: async () => {},
+        behaviour: async () => [
+          noRewrite(baseline, current),
+          ...requiredOracleChecks
+            .filter((id) => id !== 'no-rewrite.legacy-bytes')
+            .map((id) => ({ id, passed: true })),
+        ],
+      },
+    })
+    assert.equal(report.outcome, changed ? 'behaviour' : 'pass')
+    const check = report.evidence.find(
+      (e) => e.checkId === 'no-rewrite.legacy-bytes',
+    )
+    assert.ok(check)
+    if (changed)
+      assert.match(
+        JSON.stringify(check.payload),
+        /dynamo.*legacy.*email__source/,
+      )
+  })
+}
+
+test('Run rejects a rewritten duplicate Legacy Data row even when the original is last', async () => {
+  const current = structuredClone(baseline)
+  const rewritten = structuredClone(current.postgres[0])
+  rewritten.email.c = 'rewritten'
+  current.postgres.unshift(rewritten)
+  const report = await run({
+    id: 'duplicate-legacy',
+    manifest: { guideHash: 'guide' },
+    driver: {
+      provision: async () => {},
+      installUnmigrated: async () => {},
+      initializeLegacyData: async () => {},
+      snapshot: async () => {},
+      migrate: async () => ({ exitCode: 0 }),
+      friction: async () => [],
+      installAppHost: async () => {},
+      behaviour: async () => [
+        noRewrite(baseline, current),
+        ...requiredOracleChecks
+          .filter((id) => id !== 'no-rewrite.legacy-bytes')
+          .map((id) => ({ id, passed: true })),
+      ],
+    },
+  })
+  assert.equal(report.outcome, 'behaviour')
+  assert.equal(report.editorEligible, true)
+  const check = report.evidence.find(
+    (e) => e.checkId === 'no-rewrite.legacy-bytes',
+  )
+  assert.ok(check, 'a malformed current snapshot still reports the check')
+  assert.equal(Object(check.payload).passed, false)
+})
+
+test('a deleted Legacy Data row is a failed check, while a malformed baseline still throws', () => {
+  const current = structuredClone(baseline)
+  current.postgres = []
+  const check = noRewrite(baseline, current)
+  assert.equal(check.id, 'no-rewrite.legacy-bytes')
+  assert.equal(check.passed, false)
+  assert.throws(
+    () => noRewrite(current, baseline),
+    /exactly one Legacy Data Postgres row/,
+  )
+})

e2e/migration-harness/__tests__/review-driver.test.ts

@@ -9,14 +9,41 @@ import { legacy } from '../run/behaviour.mjs'
 import { migrationChecks, noRewriteCheck } from '../run/migration-checks.mjs'
 import { createRealDriver } from '../run/real-driver.mjs'
 import { run } from '../run/run.mjs'
+import { dynamoSnapshotError } from '../run/snapshot-errors.mjs'
 import type { IsolationOptions } from '../types.js'
 
 const observed = vi.hoisted(() => ({
   dispose: vi.fn(async () => {}),
-  snapshot: vi.fn(async () => ({
-    dynamo: { Item: { email__source: { S: 'original' } } },
-    postgres: [{ pk: 'legacy', email: { c: 'original', hm: 'term' } }],
-  })),
+  snapshot: vi.fn(
+    async (): Promise<unknown> => ({
+      dynamo: {
+        Item: {
+          pk: { S: 'legacy' },
+          ...Object.fromEntries(
+            [
+              'email',
+              'emailLower',
+              'age',
+              'balance',
+              'active',
+              'notes',
+            ].flatMap((key) => [
+              [`${key}__source`, { S: 'original' }],
+              [`${key}__hmac`, { S: 'term' }],
+            ]),
+          ),
+        },
+      },
+      postgres: [
+        {
+          pk: 'legacy',
+          email: { c: 'original', hm: 'term' },
+          age: { c: 'original', hm: 'term' },
+          notes: { c: 'original' },
+        },
+      ],
+    }),
+  ),
 }))
 
 vi.mock('../run/real-stores.mjs', () => ({
@@ -273,6 +300,103 @@ test('F5 checks legacy ciphertext after reads and before mutations', async () =>
   expect(mutationBeforeSnapshot).toBe(false)
 })
 
+test.each([
+  [
+    'dropped table',
+    Object.assign(Error('missing table'), { code: '42P01' }),
+    'behaviour',
+  ],
+  [
+    'renamed column',
+    Object.assign(Error('missing column'), { code: '42703' }),
+    'behaviour',
+  ],
+  [
+    'invalid payload cast',
+    Object.assign(Error('invalid cast'), { code: '42846' }),
+    'behaviour',
+  ],
+  ['deleted DynamoDB table', dynamoSnapshotError(400), 'behaviour'],
+  ['DynamoDB service unavailable', dynamoSnapshotError(503), 'infrastructure'],
+  ['DynamoDB throttling', dynamoSnapshotError(429), 'infrastructure'],
+  [
+    'unknown snapshot failure',
+    Error('unrecognized snapshot failure'),
+    'behaviour',
+  ],
+  [
+    'refused connection',
+    Object.assign(Error('connection refused'), { code: 'ECONNREFUSED' }),
+    'infrastructure',
+  ],
+  [
+    'reset connection',
+    Object.assign(Error('connection reset'), { code: 'ECONNRESET' }),
+    'infrastructure',
+  ],
+  [
+    'connection timeout',
+    Object.assign(Error('connection timed out'), { code: 'ETIMEDOUT' }),
+    'infrastructure',
+  ],
+  [
+    'broken socket pipe',
+    Object.assign(Error('broken pipe'), { code: 'EPIPE' }),
+    'infrastructure',
+  ],
+  [
+    'Postgres.js closed connection',
+    Object.assign(Error('connection closed'), { code: 'CONNECTION_CLOSED' }),
+    'infrastructure',
+  ],
+  [
+    'Postgres connection loss',
+    Object.assign(Error('connection failure'), { code: '08006' }),
+    'infrastructure',
+  ],
+  [
+    'Postgres shutdown',
+    Object.assign(Error('admin shutdown'), { code: '57P01' }),
+    'infrastructure',
+  ],
+  [
+    'fetch transport failure',
+    new TypeError('fetch failed', {
+      cause: Object.assign(Error('socket closed'), { code: 'UND_ERR_SOCKET' }),
+    }),
+    'infrastructure',
+  ],
+] as const)(
+  'post-read snapshot classifies %s',
+  async (_scenario, error, outcome) => {
+    const { driver, calls } = await setup()
+    const baseline = await observed.snapshot()
+    observed.snapshot
+      .mockResolvedValueOnce(baseline)
+      .mockRejectedValueOnce(error)
+    try {
+      const report = await run({
+        id: 'post-read-snapshot',
+        manifest: { guideHash: 'guide' },
+        driver: {
+          ...driver,
+          provision: async () => {},
+          migrate: async () => ({ exitCode: 0 }),
+          friction: async () => [],
+        },
+      })
+      expect(report.outcome).toBe(outcome)
+      expect(report.editorEligible).toBe(outcome === 'behaviour')
+      expect(calls).toContain('findPostgresAge')
+      expect(
+        report.stages.find((stage) => stage.name === 'behaviour')?.status,
+      ).toBe('failed')
+    } finally {
+      await driver.close()
+    }
+  },
+)
+
 test('F12 disposes stores even when the isolation supervisor fails to stop', async () => {
   const { driver } = await setup({ stopFails: true })
   await expect(driver.close()).rejects.toThrow('supervisor stop failed')