Merge pull request #738 from Timrossid/fix/669-contract-invocation-replay-protection

fix: add durable contract invocation replay protection

Author: MaryammAli

app/backend/src/transactions/invocation-replay.service.ts

@@ -0,0 +1,44 @@
+import { Injectable } from "@nestjs/common";
+import { SupabaseService } from "../supabase/supabase.service";
+
+export type ReplayClaim =
+  | { kind: "claimed" }
+  | { kind: "cached"; response: unknown }
+  | { kind: "conflict" }
+  | { kind: "in_flight" };
+
+@Injectable()
+export class InvocationReplayService {
+  constructor(private readonly supabase: SupabaseService) {}
+
+  async claim(
+    scope: string,
+    key: string,
+    fingerprint: string,
+  ): Promise<ReplayClaim> {
+    const { data, error } = await this.supabase.getClient().rpc("claim_transaction_invocation", {
+      p_scope: scope,
+      p_key: key,
+      p_fingerprint: fingerprint,
+    });
+    if (error) throw error;
+    return data as ReplayClaim;
+  }
+
+  async complete(scope: string, key: string, response: unknown): Promise<void> {
+    const { error } = await this.supabase.getClient().rpc("complete_transaction_invocation", {
+      p_scope: scope,
+      p_key: key,
+      p_response: response,
+    });
+    if (error) throw error;
+  }
+
+  async release(scope: string, key: string): Promise<void> {
+    const { error } = await this.supabase.getClient().rpc("release_transaction_invocation", {
+      p_scope: scope,
+      p_key: key,
+    });
+    if (error) throw error;
+  }
+}

app/backend/src/transactions/transaction.service.ts

@@ -2,6 +2,7 @@ import {
   Injectable,
   Logger,
   BadRequestException,
+  ConflictException,
   InternalServerErrorException,
 } from "@nestjs/common";
 import { createHash } from "crypto";
@@ -18,20 +19,18 @@ import { buildScVal } from "./utils/param-builder";
 import { SorobanRpcService } from "./soroban-rpc.service";
 import { mapSorobanError } from "../common/soroban-errors";
 import { SorobanErrorCode } from "../common/soroban-errors";
+import { InvocationReplayService } from "./invocation-replay.service";
 
 const STROOPS_PER_XLM = 10_000_000;
 const BASE_FEE = 100; // stroops
 
 @Injectable()
 export class TransactionsService {
   private readonly logger = new Logger(TransactionsService.name);
-  private readonly idempotencyResponses = new Map<
-    string,
-    ComposeTransactionResponse | ComposeTransactionError
-  >();
-  private readonly idempotencyFingerprints = new Map<string, string>();
-
-  constructor(private readonly sorobanRpcService: SorobanRpcService) {}
+  constructor(
+    private readonly sorobanRpcService: SorobanRpcService,
+    private readonly invocationReplayService: InvocationReplayService,
+  ) {}
 
   async composeTransaction(
     dto: ComposeTransactionDto,
@@ -40,16 +39,21 @@ export class TransactionsService {
 
     const payloadFingerprint = this.buildFingerprint(dto);
     const idempotencyKey = dto.idempotencyKey ?? payloadFingerprint;
-    const fingerprintForKey = this.idempotencyFingerprints.get(idempotencyKey);
-    if (fingerprintForKey && fingerprintForKey !== payloadFingerprint) {
-      throw new BadRequestException(
+    const replay = await this.invocationReplayService.claim(
+      `${dto.sourceAccount}:${dto.networkPassphrase ?? "__default__"}`,
+      idempotencyKey,
+      payloadFingerprint,
+    );
+    if (replay.kind === "conflict") {
+      throw new ConflictException(
         "This idempotency key was already used with a different payload.",
       );
     }
-
-    const cached = this.idempotencyResponses.get(idempotencyKey);
-    if (cached) {
-      return cached;
+    if (replay.kind === "in_flight") {
+      throw new ConflictException("This invocation is already in progress.");
+    }
+    if (replay.kind === "cached") {
+      return replay.response as ComposeTransactionResponse | ComposeTransactionError;
     }
 
     const startTime = Date.now();
@@ -64,6 +68,10 @@ export class TransactionsService {
     try {
       account = await this.sorobanRpcService.getAccount(dto.sourceAccount);
     } catch (err) {
+      await this.invocationReplayService.release(
+        `${dto.sourceAccount}:${dto.networkPassphrase ?? "__default__"}`,
+        idempotencyKey,
+      );
       return {
         success: false,
         error: err.message,
@@ -76,6 +84,10 @@ export class TransactionsService {
     try {
       scParams = dto.params.map(buildScVal);
     } catch (err) {
+      await this.invocationReplayService.release(
+        `${dto.sourceAccount}:${dto.networkPassphrase ?? "__default__"}`,
+        idempotencyKey,
+      );
       throw new BadRequestException(`Invalid parameter: ${err.message}`);
     }
 
@@ -102,6 +114,10 @@ export class TransactionsService {
       simulationResult = await this.sorobanRpcService.simulateTransaction(tx);
     } catch (err) {
       this.logger.error("RPC simulation request failed", err);
+      await this.invocationReplayService.release(
+        `${dto.sourceAccount}:${dto.networkPassphrase ?? "__default__"}`,
+        idempotencyKey,
+      );
       throw new InternalServerErrorException(
         "Failed to reach Soroban RPC provider.",
       );
@@ -119,7 +135,10 @@ export class TransactionsService {
         userMessage: mapped.message,
         details: mapped.details,
       };
-      this.rememberResponse(idempotencyKey, payloadFingerprint, failedResponse);
+      await this.invocationReplayService.release(
+        `${dto.sourceAccount}:${dto.networkPassphrase ?? "__default__"}`,
+        idempotencyKey,
+      );
       return failedResponse;
     }
 
@@ -134,7 +153,10 @@ export class TransactionsService {
           restorePreamble: simulationResult.restorePreamble,
         },
       } as ComposeTransactionError;
-      this.rememberResponse(idempotencyKey, payloadFingerprint, restoreResponse);
+      await this.invocationReplayService.release(
+        `${dto.sourceAccount}:${dto.networkPassphrase ?? "__default__"}`,
+        idempotencyKey,
+      );
       return restoreResponse;
     }
 
@@ -203,7 +225,11 @@ export class TransactionsService {
         },
       },
     };
-    this.rememberResponse(idempotencyKey, payloadFingerprint, response);
+    await this.invocationReplayService.complete(
+      `${dto.sourceAccount}:${dto.networkPassphrase ?? "__default__"}`,
+      idempotencyKey,
+      response,
+    );
     return response;
   }
 
@@ -230,12 +256,4 @@ export class TransactionsService {
     return createHash("sha256").update(normalized).digest("hex");
   }
 
-  private rememberResponse(
-    idempotencyKey: string,
-    fingerprint: string,
-    response: ComposeTransactionResponse | ComposeTransactionError,
-  ): void {
-    this.idempotencyFingerprints.set(idempotencyKey, fingerprint);
-    this.idempotencyResponses.set(idempotencyKey, response);
-  }
 }

app/backend/src/transactions/transaction.service.unit.spec.ts

@@ -76,14 +76,20 @@ jest.mock('@stellar/stellar-sdk', () => {
   };
 });
 
-import { BadRequestException } from '@nestjs/common';
+jest.mock('../supabase/supabase.service', () => ({
+  SupabaseService: class SupabaseService {},
+}));
+
+import { ConflictException } from '@nestjs/common';
 
 import { SorobanRpcService } from './soroban-rpc.service';
 import { TransactionsService } from './transaction.service';
+import { InvocationReplayService } from './invocation-replay.service';
 
 describe('TransactionsService', () => {
   let service: TransactionsService;
   let mockSorobanRpcService: jest.Mocked<Partial<SorobanRpcService>>;
+  let mockInvocationReplayService: jest.Mocked<Partial<InvocationReplayService>>;
 
   beforeEach(() => {
     mockSorobanRpcService = {
@@ -110,8 +116,16 @@ describe('TransactionsService', () => {
         },
       }),
     };
+    mockInvocationReplayService = {
+      claim: jest.fn().mockResolvedValue({ kind: 'claimed' }),
+      complete: jest.fn().mockResolvedValue(undefined),
+      release: jest.fn().mockResolvedValue(undefined),
+    };
 
-    service = new TransactionsService(mockSorobanRpcService as unknown as SorobanRpcService);
+    service = new TransactionsService(
+      mockSorobanRpcService as unknown as SorobanRpcService,
+      mockInvocationReplayService as unknown as InvocationReplayService,
+    );
   });
 
   it('returns a simulation summary and idempotency key', async () => {
@@ -130,7 +144,7 @@ describe('TransactionsService', () => {
     }
   });
 
-  it('reuses the cached response for the same idempotency key', async () => {
+  it('returns a durable cached response for the same idempotency key', async () => {
     const payload = {
       contractId: 'C123',
       method: 'health_check',
@@ -139,14 +153,20 @@ describe('TransactionsService', () => {
       idempotencyKey: 'same-key',
     };
 
+    mockInvocationReplayService.claim!.mockResolvedValueOnce({ kind: 'claimed' });
     const first = await service.composeTransaction(payload);
+    mockInvocationReplayService.claim!.mockResolvedValueOnce({
+      kind: 'cached',
+      response: first,
+    });
     const second = await service.composeTransaction(payload);
 
     expect(second).toEqual(first);
     expect(mockSorobanRpcService.simulateTransaction).toHaveBeenCalledTimes(1);
   });
 
   it('rejects reusing an idempotency key with a different payload', async () => {
+    mockInvocationReplayService.claim!.mockResolvedValueOnce({ kind: 'claimed' });
     await service.composeTransaction({
       contractId: 'C123',
       method: 'health_check',
@@ -155,6 +175,7 @@ describe('TransactionsService', () => {
       idempotencyKey: 'same-key',
     });
 
+    mockInvocationReplayService.claim!.mockResolvedValueOnce({ kind: 'conflict' });
     await expect(
       service.composeTransaction({
         contractId: 'C123',
@@ -163,6 +184,8 @@ describe('TransactionsService', () => {
         sourceAccount: 'G123',
         idempotencyKey: 'same-key',
       }),
-    ).rejects.toThrow(BadRequestException);
+    ).rejects.toThrow(ConflictException);
+
+    expect(mockInvocationReplayService.complete).toHaveBeenCalledTimes(1);
   });
 });

app/backend/src/transactions/transactions.module.ts

@@ -8,15 +8,18 @@ import { ApiKeysModule } from "../api-keys/api-keys.module";
 import { ApiKeyGuard } from "../auth/guards/api-key.guard";
 import { MetricsModule } from "../metrics/metrics.module";
 import { FeatureFlagsModule } from "../feature-flags/feature-flags.module";
+import { SupabaseModule } from "../supabase/supabase.module";
+import { InvocationReplayService } from "./invocation-replay.service";
 
 @Module({
-  imports: [AppConfigModule, ApiKeysModule, MetricsModule, FeatureFlagsModule],
+  imports: [AppConfigModule, ApiKeysModule, MetricsModule, FeatureFlagsModule, SupabaseModule],
   controllers: [TransactionsController],
   providers: [
     HorizonService,
     TransactionsService,
     SorobanRpcService,
     ApiKeyGuard,
+    InvocationReplayService,
   ],
   exports: [HorizonService, TransactionsService, SorobanRpcService],
 })

app/backend/supabase/migrations/20260828000000_transaction_invocation_replay.sql

@@ -0,0 +1,52 @@
+CREATE TABLE IF NOT EXISTS transaction_invocation_replays (
+  scope TEXT NOT NULL,
+  idempotency_key TEXT NOT NULL,
+  fingerprint TEXT NOT NULL,
+  status TEXT NOT NULL CHECK (status IN ('pending', 'completed')),
+  response JSONB,
+  claimed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  PRIMARY KEY (scope, idempotency_key)
+);
+
+CREATE OR REPLACE FUNCTION claim_transaction_invocation(
+  p_scope TEXT, p_key TEXT, p_fingerprint TEXT
+) RETURNS JSONB LANGUAGE plpgsql AS $$
+DECLARE existing transaction_invocation_replays;
+BEGIN
+  INSERT INTO transaction_invocation_replays(scope, idempotency_key, fingerprint, status)
+  VALUES (p_scope, p_key, p_fingerprint, 'pending')
+  ON CONFLICT (scope, idempotency_key) DO NOTHING;
+
+  SELECT * INTO existing FROM transaction_invocation_replays
+  WHERE scope = p_scope AND idempotency_key = p_key FOR UPDATE;
+
+  IF existing.fingerprint <> p_fingerprint THEN
+    RETURN jsonb_build_object('kind', 'conflict');
+  END IF;
+  IF existing.status = 'completed' THEN
+    RETURN jsonb_build_object('kind', 'cached', 'response', existing.response);
+  END IF;
+  IF existing.claimed_at < now() - interval '5 minutes' THEN
+    UPDATE transaction_invocation_replays SET claimed_at = now(), updated_at = now()
+    WHERE scope = p_scope AND idempotency_key = p_key;
+    RETURN jsonb_build_object('kind', 'claimed');
+  END IF;
+  RETURN jsonb_build_object('kind', 'in_flight');
+END;
+$$;
+
+CREATE OR REPLACE FUNCTION complete_transaction_invocation(
+  p_scope TEXT, p_key TEXT, p_response JSONB
+) RETURNS VOID LANGUAGE SQL AS $$
+  UPDATE transaction_invocation_replays
+  SET status = 'completed', response = p_response, updated_at = now()
+  WHERE scope = p_scope AND idempotency_key = p_key AND status = 'pending';
+$$;
+
+CREATE OR REPLACE FUNCTION release_transaction_invocation(
+  p_scope TEXT, p_key TEXT
+) RETURNS VOID LANGUAGE SQL AS $$
+  DELETE FROM transaction_invocation_replays
+  WHERE scope = p_scope AND idempotency_key = p_key AND status = 'pending';
+$$;