Merge branch 'main' into test/815-stream-post-negative-cases

Author: Emelie-Dev

backend/.dockerignore

@@ -1,6 +1,10 @@
 node_modules
 dist
 coverage
+
+# Omit secrets and logs
 .env*
 *.log
-src/generated
+
+# Omit code generated
+src/generated

backend/src/controllers/stream.controller.ts

@@ -100,16 +100,33 @@ function parseRequiredBigIntField(fieldName: string, value: unknown): bigint {
  */
 export const createStream = async (req: Request, res: Response) => {
   try {
-    const {
-      streamId,
-      sender,
-      recipient,
-      tokenAddress,
-      ratePerSecond,
-      depositedAmount,
-      startTime,
-    } = req.body;
-    const callerAddress = (req as AuthenticatedRequest).user?.publicKey;
+    const callerPublicKey = (req as AuthenticatedRequest).user?.publicKey;
+    if (!callerPublicKey) {
+      return res.status(401).json({ error: 'Unauthorized', message: 'Authentication required' });
+    }
+
+    const { streamId, sender, recipient, tokenAddress, ratePerSecond, depositedAmount, startTime } = req.body;
+
+    // Issue #809: validate identity fields before any DB write.
+    if (typeof sender !== 'string' || sender.length === 0) {
+      return res.status(400).json({ error: 'Invalid sender: must be a non-empty string' });
+    }
+    if (typeof recipient !== 'string' || recipient.length === 0) {
+      return res.status(400).json({ error: 'Invalid recipient: must be a non-empty string' });
+    }
+    if (typeof tokenAddress !== 'string' || tokenAddress.length === 0) {
+      return res.status(400).json({ error: 'Invalid tokenAddress: must be a non-empty string' });
+    }
+
+    // Issue #809: the authenticated wallet may only create/modify streams it owns.
+    // Without this, any logged-in wallet could POST an arbitrary `sender` and have
+    // it persisted, or flip another owner's cancelled stream back to active.
+    if (sender !== callerPublicKey) {
+      return res.status(403).json({
+        error: 'Forbidden',
+        message: 'sender must match the authenticated wallet',
+      });
+    }
 
     const parsedStreamId = Number.parseInt(streamId, 10);
     const parsedStartTime = Number.parseInt(startTime, 10);
@@ -162,9 +179,18 @@ export const createStream = async (req: Request, res: Response) => {
     const endTime =
       parsedStartTime + Number(parsedDepositedAmount / parsedRatePerSecond);
 
-    if (callerAddress && sender && callerAddress !== sender) {
-      return res.status(403).json({ error: "Forbidden" });
+    // Issue #809: never let the upsert update branch touch a stream owned by a
+    // different wallet. The caller is already proven to equal `sender` above, so
+    // reject any existing row whose sender differs — this blocks reactivating or
+    // overwriting someone else's (e.g. cancelled) stream.
+    const existing = await prisma.stream.findUnique({ where: { streamId: parsedStreamId } });
+    if (existing && existing.sender !== callerPublicKey) {
+      return res.status(403).json({
+        error: 'Forbidden',
+        message: 'Cannot modify a stream owned by another wallet',
+      });
     }
+
     const stream = await prisma.stream.upsert({
       where: { streamId: parsedStreamId },
       update: {
@@ -699,6 +725,13 @@ export const topUpStreamHandler = async (req: Request, res: Response) => {
         .json({ error: "Only the stream sender may top up this stream" });
     }
 
+    if (!stream.isActive) {
+      return res.status(409).json({ error: 'Conflict', message: 'Cannot top up an inactive stream' });
+    }
+    if (stream.isPaused) {
+      return res.status(409).json({ error: 'Conflict', message: 'Cannot top up a paused stream' });
+    }
+
     const txHash = await topUpStream(streamId, amount, callerAddress);
 
     const newDeposited = (BigInt(stream.depositedAmount) + amount).toString();

backend/src/routes/v1/stream.routes.ts

@@ -91,8 +91,8 @@ router.get('/:streamId', getStream);
  *           type: integer
  *           default: 50
  *           minimum: 1
- *           maximum: 500
- *         description: "Number of events to return per page (default: 50, max: 500)"
+ *           maximum: 200
+ *         description: "Number of events to return per page (default: 50, max: 200)"
  *       - in: query
  *         name: offset
  *         schema:

backend/tests/integration/stream-lifecycle.test.ts

@@ -557,6 +557,72 @@ describe("Stream Lifecycle Integration Tests", () => {
     });
   });
 
+  describe("StreamEvent @@unique([transactionHash, eventType]) deduplication", () => {
+    it("rejects a duplicate (transactionHash, eventType) insert at the database level", async () => {
+      const streamId = 11;
+      const txHash = "unique-constraint-tx-hash";
+
+      await testPrisma.stream.create({
+        data: {
+          streamId,
+          sender: SENDER,
+          recipient: RECIPIENT,
+          tokenAddress: TOKEN,
+          ratePerSecond: "10",
+          depositedAmount: "86400",
+          withdrawnAmount: "0",
+          startTime: 1700000000,
+          endTime: 1700000000 + 8640,
+          lastUpdateTime: 1700000000,
+          isActive: true,
+          isPaused: false,
+        },
+      });
+
+      const eventData = {
+        streamId,
+        eventType: "CREATED",
+        transactionHash: txHash,
+        ledgerSequence: 12345,
+        timestamp: 1700000000,
+      };
+
+      await testPrisma.streamEvent.create({ data: eventData });
+
+      await expect(
+        testPrisma.streamEvent.create({
+          data: {
+            ...eventData,
+            ledgerSequence: 12346,
+            timestamp: 1700000001,
+          },
+        }),
+      ).rejects.toMatchObject({ code: "P2002" });
+
+      const count = await testPrisma.streamEvent.count({
+        where: { transactionHash: txHash, eventType: "CREATED" },
+      });
+      expect(count).toBe(1);
+    });
+
+    it("dedupes worker replay of the same event without creating a second row", async () => {
+      const streamId = 12;
+      const txHash = "worker-replay-dedup-tx-hash";
+      const event = { ...createStreamCreatedEvent(streamId), txHash };
+
+      await worker.processEvent(event);
+      await expect(worker.processEvent(event)).resolves.toBeUndefined();
+
+      const count = await testPrisma.streamEvent.count({
+        where: {
+          transactionHash: event.txHash,
+          eventType: "CREATED",
+        },
+      });
+      expect(count).toBe(1);
+    });
+  });
+
   describe("SSE client receives broadcast for each stream event", () => {
     let eventSource: EventSource;
 

backend/tests/integration/top-up.test.ts

@@ -161,4 +161,28 @@ describe('POST /v1/streams/:streamId/top-up', () => {
       }),
     );
   });
+
+  it('returns 409 when stream is inactive', async () => {
+    vi.mocked(mockPrisma.stream.findUnique).mockResolvedValue({ ...mockStream, isActive: false } as any);
+
+    const res = await request(app)
+      .post('/v1/streams/42/top-up')
+      .set('Authorization', 'Bearer dummy')
+      .send({ amount: '1000' });
+
+    expect(res.status).toBe(409);
+    expect(res.body.message).toMatch(/inactive stream/);
+  });
+
+  it('returns 409 when stream is paused', async () => {
+    vi.mocked(mockPrisma.stream.findUnique).mockResolvedValue({ ...mockStream, isPaused: true } as any);
+
+    const res = await request(app)
+      .post('/v1/streams/42/top-up')
+      .set('Authorization', 'Bearer dummy')
+      .send({ amount: '1000' });
+
+    expect(res.status).toBe(409);
+    expect(res.body.message).toMatch(/paused stream/);
+  });
 });

backend/tests/stream.controller.test.ts

@@ -78,14 +78,17 @@ describe("Stream Controller", () => {
         publicKey: "GSENDER",
       },
     };
+    // Authenticated caller matches body.sender by default (Issue #809).
+    (req as any).user = { publicKey: 'GSENDER' };
     res = {
       status: vi.fn().mockReturnThis(),
       json: vi.fn().mockReturnThis(),
     };
   });
 
-  describe("createStream", () => {
-    it("should create a stream successfully", async () => {
+  describe('createStream', () => {
+    it('should create a stream successfully', async () => {
+      (prisma.stream.findUnique as any).mockResolvedValue(null);
       (prisma.stream.upsert as any).mockResolvedValue({ streamId: 123 });
 
       await createStream(req as Request, res as Response);
@@ -94,20 +97,54 @@ describe("Stream Controller", () => {
       expect(prisma.stream.upsert).toHaveBeenCalled();
     });
 
-    it("should return 400 for invalid streamId", async () => {
-      req.body.streamId = "abc";
+    it('should return 401 when the request is unauthenticated', async () => {
+      (req as any).user = undefined;
+
       await createStream(req as Request, res as Response);
-      expect(res.status).toHaveBeenCalledWith(400);
+
+      expect(res.status).toHaveBeenCalledWith(401);
+      expect(prisma.stream.upsert).not.toHaveBeenCalled();
+    });
+
+    it('should return 403 when the caller is not the body sender (Issue #809)', async () => {
+      (req as any).user = { publicKey: 'GATTACKER' };
+
+      await createStream(req as Request, res as Response);
+
+      expect(res.status).toHaveBeenCalledWith(403);
+      expect(prisma.stream.upsert).not.toHaveBeenCalled();
     });
 
-    it("should return 400 for non-positive ratePerSecond", async () => {
-      req.body.ratePerSecond = "0";
+    it('should reject 400 when sender is missing (Issue #809)', async () => {
+      delete req.body.sender;
+      (req as any).user = { publicKey: 'GSENDER' };
+
       await createStream(req as Request, res as Response);
+
       expect(res.status).toHaveBeenCalledWith(400);
+      expect(prisma.stream.upsert).not.toHaveBeenCalled();
+    });
+
+    it('should return 403 and not reactivate a cancelled stream owned by another wallet (Issue #809)', async () => {
+      // A victim previously created (and cancelled) this stream.
+      (prisma.stream.findUnique as any).mockResolvedValue({
+        streamId: 123,
+        sender: 'GVICTIM',
+        isActive: false,
+      });
+      // Attacker authenticates as themselves and sets sender to their own key,
+      // trying to hijack the victim's streamId.
+      req.body.sender = 'GATTACKER';
+      (req as any).user = { publicKey: 'GATTACKER' };
+
+      await createStream(req as Request, res as Response);
+
+      expect(res.status).toHaveBeenCalledWith(403);
+      expect(prisma.stream.upsert).not.toHaveBeenCalled();
     });
 
-    it("should return 400 for malformed numeric fields", async () => {
-      req.body.ratePerSecond = "abc";
+    it('should return 400 for invalid streamId', async () => {
+      req.body.streamId = 'abc';
       await createStream(req as Request, res as Response);
       expect(res.status).toHaveBeenCalledWith(400);
     });

backend/tests/stream.repository.test.ts

@@ -23,6 +23,14 @@ describe('Stream Repository', () => {
     });
   });
 
+  it('should update isActive to false for COMPLETED', async () => {
+    await updateStatus(123, 'COMPLETED');
+    expect(prisma.stream.update).toHaveBeenCalledWith({
+      where: { streamId: 123 },
+      data: { isActive: false },
+    });
+  });
+
   it('should update isActive to true for ACTIVE', async () => {
     await updateStatus(123, 'ACTIVE');
     expect(prisma.stream.update).toHaveBeenCalledWith({

contracts/stream_contract/Cargo.toml

@@ -2,6 +2,7 @@
 name = "stream_contract"
 version = "0.1.0"
 edition = "2021"
+description = "Soroban payment-streaming contract with protocol fees"
 
 [lib]
 crate-type = ["cdylib"]