fix(streams): enforce sender ownership on createStream (#809)

createStream never read req.user, so any authenticated wallet could POST
an arbitrary sender or flip another owner's cancelled stream back to active
via the upsert update branch (keyed only on client-supplied streamId).

- 401 when unauthenticated; 403 when JWT subject != body sender
- validate sender/recipient/tokenAddress before any DB write
- reject (403) upsert against a streamId already owned by another wallet,
  so the update branch can never reactivate someone else's stream
- add controller tests covering non-owner reactivation and field validation

Author: jadonamite

backend/src/controllers/stream.controller.ts

@@ -66,8 +66,34 @@ function sumStringI128(values: string[]): string {
  */
 export const createStream = async (req: Request, res: Response) => {
   try {
+    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);
     const parsedRatePerSecond = BigInt(ratePerSecond);
@@ -91,6 +117,18 @@ export const createStream = async (req: Request, res: Response) => {
 
     const endTime = parsedStartTime + Number(parsedDepositedAmount / parsedRatePerSecond);
 
+    // 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: {

backend/tests/stream.controller.test.ts

@@ -62,6 +62,8 @@ describe('Stream Controller', () => {
       query: {},
       params: {},
     };
+    // Authenticated caller matches body.sender by default (Issue #809).
+    (req as any).user = { publicKey: 'GSENDER' };
     res = {
       status: vi.fn().mockReturnThis(),
       json: vi.fn().mockReturnThis(),
@@ -70,6 +72,7 @@ describe('Stream Controller', () => {
 
   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);
@@ -78,6 +81,52 @@ describe('Stream Controller', () => {
       expect(prisma.stream.upsert).toHaveBeenCalled();
     });
 
+    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(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 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 invalid streamId', async () => {
       req.body.streamId = 'abc';
       await createStream(req as Request, res as Response);

backend/tests/stream.test.ts

@@ -59,7 +59,7 @@ describe('POST /v1/streams', () => {
     const mockStream = {
       id: 'uuid-123',
       streamId: 1,
-      sender: 'GABC123XYZ456DEF789GHI012JKL345MNO678PQR901STU234VWX567YZA',
+      sender: 'GTEST_USER_PUBLIC_KEY',
       recipient: 'GDEF456ABC789GHI012JKL345MNO678PQR901STU234VWX567YZA123BCD',
       tokenAddress: 'CBCD789EFG012HIJ345KLM678NOP901QRS234TUV567WXY890ZAB123CDE',
       ratePerSecond: '100',
@@ -80,7 +80,7 @@ describe('POST /v1/streams', () => {
 
     const validData = {
       streamId: '1',
-      sender: 'GABC123XYZ456DEF789GHI012JKL345MNO678PQR901STU234VWX567YZA',
+      sender: 'GTEST_USER_PUBLIC_KEY',
       recipient: 'GDEF456ABC789GHI012JKL345MNO678PQR901STU234VWX567YZA123BCD',
       tokenAddress: 'CBCD789EFG012HIJ345KLM678NOP901QRS234TUV567WXY890ZAB123CDE',
       ratePerSecond: '100',
@@ -108,7 +108,7 @@ describe('POST /v1/streams', () => {
 
     const validData = {
       streamId: '2',
-      sender: 'GABC123XYZ456DEF789GHI012JKL345MNO678PQR901STU234VWX567YZA',
+      sender: 'GTEST_USER_PUBLIC_KEY',
       recipient: 'GDEF456ABC789GHI012JKL345MNO678PQR901STU234VWX567YZA123BCD',
       tokenAddress: 'CBCD789EFG012HIJ345KLM678NOP901QRS234TUV567WXY890ZAB123CDE',
       ratePerSecond: '100',