fix(backend): i128 cap, event-history auth, SSE client ids, cancel signing docs (#1532, #1533, #1534, #1547)

backend/docs/ARCHITECTURE.md

@@ -141,6 +141,7 @@ for the transaction.
 | Action | Signer | Mechanism |
 |--------|--------|-----------|
 | **Top-up** | Server (custodial) | Backend submits the transaction using `KEEPER_SECRET_KEY`. The frontend sends only the stream ID and amount. |
+| **Cancel** | Sender wallet (non-custodial) | `src/controllers/stream/cancel.ts` submits the cancel server-side, but signs with the caller-supplied `senderSecret` from the request body. It does not use `KEEPER_SECRET_KEY`. |
 | **Withdraw** | Wallet (non-custodial) | Frontend builds and signs the transaction via the connected wallet (Freighter). The backend simulate endpoint exists for fee estimation only. |
 | **Pause / Resume** | Wallet (non-custodial) | Same as withdraw — frontend-signed. Backend simulate endpoints exist for fee estimation but do not submit. |
 | **Create stream** | Wallet (non-custodial) | Frontend signs via wallet and submits directly to the Soroban RPC. |
@@ -155,15 +156,16 @@ for the transaction.
   Soroban RPC.
 - If `KEEPER_SECRET_KEY` is not configured, `topUpStream` throws
   `'KEEPER_SECRET_KEY not configured'` and the request returns HTTP 500.
-- The cancel endpoint (`src/controllers/stream/cancel.ts`) also reads
-  `KEEPER_SECRET_KEY` but only to check whether the server wallet is configured;
-  the actual cancel transaction is wallet-signed by the sender.
+- The cancel endpoint (`src/controllers/stream/cancel.ts`) does **not** read
+  `KEEPER_SECRET_KEY`. It requires `senderSecret` in the request body and signs
+  the cancel transaction with the sender's own key.
 
 ### Security Boundary
 
 > **Do not wire pause/resume/withdraw to a server-side submit path.** Only
-> `top-up` is intentionally custodial. All other mutating actions must be
-> wallet-signed by the user to preserve the non-custodial security model.
+> `top-up` is intentionally custodial. All other mutating actions — including
+> cancel — must be authorized by the user's own key to preserve the
+> non-custodial security model.
 
 The keeper key is a server-side secret and is never exposed to the frontend.
 It lives exclusively in the backend's environment configuration.

backend/src/controllers/sse.controller.ts

@@ -1,4 +1,5 @@
 import type { Request, Response } from 'express';
+import { randomUUID } from 'crypto';
 import { sseService } from '../services/sse.service.js';
 import { prisma } from '../lib/prisma.js';
 import { requestContext } from '../logger.js';
@@ -89,7 +90,9 @@ export const subscribe = async (req: Request, res: Response) => {
     }
     subscriptions.push(...userSubscriptions);
 
-    const clientId = `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`;
+    // Issue #1534: client ids must come from a CSPRNG so concurrently
+    // connecting clients cannot collide and clobber each other's registration.
+    const clientId = randomUUID();
 
     res.writeHead(200, {
       'Content-Type': 'text/event-stream',

backend/src/controllers/user.controller.ts

@@ -132,6 +132,10 @@ export const getUser = async (
  * Get user events (history) - paginated list of stream events where the
  * given wallet was either the sender or recipient.
  *
+ * Issue #1533: wallet history is private. The endpoint requires auth and the
+ * `publicKey` path param must match the authenticated caller, mirroring
+ * GET /v1/events.
+ *
  * Query params:
  *   - type: optional comma-separated list of event types to filter by
  *     (e.g. "PAUSED,RESUMED"); unknown values are ignored, and a filter
@@ -147,6 +151,11 @@ export const getUserEvents = async (
   next: NextFunction,
 ) => {
   try {
+    const callerPublicKey = (req as AuthenticatedRequest).user?.publicKey;
+    if (!callerPublicKey) {
+      return sendApiError(res, 401, "UNAUTHORIZED", "Authentication required");
+    }
+
     const { publicKey } = req.params;
     if (typeof publicKey !== "string") {
       return sendApiError(res, 400, "INVALID_PUBLIC_KEY", "Invalid publicKey parameter");
@@ -155,6 +164,17 @@ export const getUserEvents = async (
       return sendApiError(res, 400, "INVALID_PUBLIC_KEY", "Invalid Stellar public key format");
     }
 
+    // Issue #1533: prevent unauthenticated/cross-wallet enumeration of a
+    // wallet's stream history (same rule as GET /v1/events).
+    if (publicKey !== callerPublicKey) {
+      return sendApiError(
+        res,
+        403,
+        "FORBIDDEN",
+        "You can only view your own event history",
+      );
+    }
+
     const { requested, types } = parseEventTypeFilter(req.query["type"]);
     if (requested.length > 0 && types.length === 0) {
       return res

backend/src/routes/v1/user.routes.ts

@@ -171,7 +171,9 @@ router.get("/:publicKey", getUser);
  *     tags:
  *       - Users
  *     summary: Fetch user activity history
- *     description: Returns a paginated chronological history of all stream events associated with the user.
+ *     description: Returns a paginated chronological history of all stream events associated with the authenticated wallet. The `publicKey` path parameter must match the authenticated caller.
+ *     security:
+ *       - BearerAuth: []
  *     parameters:
  *       - in: path
  *         name: publicKey
@@ -205,6 +207,18 @@ router.get("/:publicKey", getUser);
  *           application/json:
  *             schema:
  *               $ref: '#/components/schemas/Error'
+ *       401:
+ *         description: Unauthorized - missing or invalid authentication token
+ *         content:
+ *           application/json:
+ *             schema:
+ *               $ref: '#/components/schemas/Error'
+ *       403:
+ *         description: Forbidden - `publicKey` does not match the authenticated wallet
+ *         content:
+ *           application/json:
+ *             schema:
+ *               $ref: '#/components/schemas/Error'
  *       404:
  *         description: User not found
  *         content:
@@ -218,7 +232,7 @@ router.get("/:publicKey", getUser);
  *             schema:
  *               $ref: '#/components/schemas/Error'
  */
-router.get("/:publicKey/events", getUserEvents);
+router.get("/:publicKey/events", requireAuth, getUserEvents);
 
 export default router;
 

backend/src/validators/stream.validator.ts

@@ -1,7 +1,18 @@
 import { z } from 'zod';
 import { parseStreamId } from '../lib/stream-id.js';
 
-const MAX_I128 = BigInt('170141183460469231731687303715884105727');
+/** Soroban `i128` upper bound (2^127 - 1). Values above this overflow the
+ * on-chain type and the Postgres i128 columns, so they must never pass
+ * request validation. */
+export const MAX_I128 = BigInt('170141183460469231731687303715884105727');
+
+function isAtMostI128(val: string): boolean {
+  try {
+    return BigInt(val) <= MAX_I128;
+  } catch {
+    return true;
+  }
+}
 
 export const createStreamSchema = z.object({
   streamId: z
@@ -23,17 +34,11 @@ export const createStreamSchema = z.object({
   ratePerSecond: z
     .string()
     .regex(/^\d+$/, 'Rate must be a positive integer as string')
-    .refine(
-      (val) => {
-        try {
-          return BigInt(val) <= MAX_I128;
-        } catch {
-          return true;
-        }
-      },
-      { message: 'Rate exceeds maximum allowed value' },
-    ),
-  depositedAmount: z.string().regex(/^\d+$/, 'Amount must be a positive integer as string'),
+    .refine(isAtMostI128, { message: 'Rate exceeds maximum allowed value' }),
+  depositedAmount: z
+    .string()
+    .regex(/^\d+$/, 'Amount must be a positive integer as string')
+    .refine(isAtMostI128, { message: 'Amount exceeds maximum allowed value' }),
   startTime: z.union([z.number().int().nonnegative(), z.string().regex(/^\d+$/).transform(v => parseInt(v))]),
 });
 

backend/swagger/flowfi.openapi.json

@@ -1544,7 +1544,12 @@
           "Users"
         ],
         "summary": "Fetch user activity history",
-        "description": "Returns a paginated chronological history of all stream events associated with the user.",
+        "description": "Returns a paginated chronological history of all stream events associated with the authenticated wallet. The `publicKey` path parameter must match the authenticated caller.",
+        "security": [
+          {
+            "BearerAuth": []
+          }
+        ],
         "parameters": [
           {
             "in": "path",
@@ -1596,6 +1601,26 @@
               }
             }
           },
+          "401": {
+            "description": "Unauthorized - missing or invalid authentication token",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/Error"
+                }
+              }
+            }
+          },
+          "403": {
+            "description": "Forbidden - `publicKey` does not match the authenticated wallet",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/Error"
+                }
+              }
+            }
+          },
           "404": {
             "description": "User not found",
             "content": {

backend/tests/integration/events-list.test.ts

@@ -212,9 +212,9 @@ describe('GET /v1/events', () => {
  * both sit on top of the shared listEventsForWallet helper
  * (backend/src/repositories/streamEvent.repository.ts) and must therefore
  * apply identical sender/recipient and type-filtering logic, even though
- * they differ in auth model (unauthenticated + :publicKey param vs.
- * requireAuth + session address) and response envelope field name
- * (`data` vs `events`).
+ * both require auth and are scoped to the authenticated wallet; they differ
+ * in the wallet selector (`:publicKey` path param vs. `address` query param)
+ * and response envelope field name (`data` vs `events`).
  */
 describe('GET /v1/users/:publicKey/events and GET /v1/events filtering parity', () => {
   // GET /v1/users/:publicKey/events validates the Stellar public key format
@@ -234,6 +234,31 @@ describe('GET /v1/users/:publicKey/events and GET /v1/events filtering parity',
     vi.clearAllMocks();
   });
 
+  it('rejects unauthenticated requests (#1533)', async () => {
+    const res = await request(app).get(`/v1/users/${VALID_ADDR}/events`);
+
+    expect(res.status).toBe(401);
+    expect(mocks.prisma.streamEvent.findMany).not.toHaveBeenCalled();
+  });
+
+  it('rejects requests for a different wallet\'s history (#1533)', async () => {
+    const otherAddr = 'GCNO64M4RVXL4QA43XCMQHA4S2MYUWLFEAYS6OQUXNA4EFC7XUMXBBEY';
+    const otherToken = signJwt({
+      sub: otherAddr,
+      iat: validAddrNow,
+      exp: validAddrNow + 3600,
+      iss: 'flowfi-api',
+      aud: 'flowfi-api',
+    });
+
+    const res = await request(app)
+      .get(`/v1/users/${VALID_ADDR}/events`)
+      .set('Authorization', `Bearer ${otherToken}`);
+
+    expect(res.status).toBe(403);
+    expect(mocks.prisma.streamEvent.findMany).not.toHaveBeenCalled();
+  });
+
   it('apply the same sender/recipient OR where-clause for a given wallet', async () => {
     mocks.prisma.streamEvent.findMany.mockResolvedValue([]);
     mocks.prisma.streamEvent.count.mockResolvedValue(0);
@@ -247,7 +272,9 @@ describe('GET /v1/users/:publicKey/events and GET /v1/events filtering parity',
     mocks.prisma.streamEvent.findMany.mockResolvedValue([]);
     mocks.prisma.streamEvent.count.mockResolvedValue(0);
 
-    await request(app).get(`/v1/users/${VALID_ADDR}/events`);
+    await request(app)
+      .get(`/v1/users/${VALID_ADDR}/events`)
+      .set('Authorization', `Bearer ${validAddrToken}`);
     const userEventsWhere = mocks.prisma.streamEvent.findMany.mock.calls[0]![0].where;
 
     expect(userEventsWhere.stream).toEqual(eventsWhere.stream);
@@ -266,7 +293,9 @@ describe('GET /v1/users/:publicKey/events and GET /v1/events filtering parity',
     mocks.prisma.streamEvent.findMany.mockResolvedValue([]);
     mocks.prisma.streamEvent.count.mockResolvedValue(0);
 
-    await request(app).get(`/v1/users/${VALID_ADDR}/events?type=PAUSED,RESUMED`);
+    await request(app)
+      .get(`/v1/users/${VALID_ADDR}/events?type=PAUSED,RESUMED`)
+      .set('Authorization', `Bearer ${validAddrToken}`);
     const userEventsWhere = mocks.prisma.streamEvent.findMany.mock.calls[0]![0].where;
 
     expect(userEventsWhere.eventType).toEqual(eventsWhere.eventType);
@@ -277,7 +306,9 @@ describe('GET /v1/users/:publicKey/events and GET /v1/events filtering parity',
     const eventsRes = await request(app)
       .get(`/v1/events?address=${VALID_ADDR}&type=BOGUS`)
       .set('Authorization', `Bearer ${validAddrToken}`);
-    const userEventsRes = await request(app).get(`/v1/users/${VALID_ADDR}/events?type=BOGUS`);
+    const userEventsRes = await request(app)
+      .get(`/v1/users/${VALID_ADDR}/events?type=BOGUS`)
+      .set('Authorization', `Bearer ${validAddrToken}`);
 
     expect(eventsRes.status).toBe(400);
     expect(userEventsRes.status).toBe(400);

backend/tests/sse.controller.test.ts

@@ -71,6 +71,25 @@ describe('SSE Controller', () => {
     expect(sseService.addClient).toHaveBeenCalled();
   });
 
+  it('generates clientId with a CSPRNG instead of Math.random (#1534)', async () => {
+    (sseService.isShuttingDown as any).mockReturnValue(false);
+    (sseService.checkCapacity as any).mockReturnValue({ allowed: true });
+    (req as any).user = { publicKey: 'GUSER1' };
+    (prisma.stream.findMany as any).mockResolvedValue([]);
+
+    await subscribe(req as Request, res as Response);
+
+    const clientId = (sseService.addClient as any).mock.calls[0][0] as string;
+    expect(clientId).toMatch(
+      /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/,
+    );
+
+    const connectedPayload = JSON.parse(
+      ((res.write as any).mock.calls[0][0] as string).replace(/^data: /, '').trim(),
+    );
+    expect(connectedPayload.clientId).toBe(clientId);
+  });
+
   it('should include user subscriptions when users query params are provided', async () => {
     (sseService.isShuttingDown as any).mockReturnValue(false);
     (sseService.checkCapacity as any).mockReturnValue({ allowed: true });

backend/tests/stream.controller.test.ts

@@ -210,6 +210,20 @@ describe("Stream Controller", () => {
       );
     });
 
+    it('should return 400 when depositedAmount exceeds i128 max (#1532)', async () => {
+      req.body.depositedAmount = '170141183460469231731687303715884105728';
+      await createStream(req as Request, res as Response);
+      expect(res.status).toHaveBeenCalledWith(400);
+      expect(res.status).not.toHaveBeenCalledWith(500);
+      expect(prisma.stream.findUnique).not.toHaveBeenCalled();
+      expect(prisma.stream.upsert).not.toHaveBeenCalled();
+      expect(res.json).toHaveBeenCalledWith(
+        expect.objectContaining({
+          error: expect.objectContaining({ message: expect.stringContaining('depositedAmount') }),
+        })
+      );
+    });
+
     it('should return 400, not 500, when depositedAmount is missing', async () => {
       delete req.body.depositedAmount;
       await createStream(req as Request, res as Response);

backend/tests/stream.test.ts

@@ -138,6 +138,27 @@ describe('POST /v1/streams', () => {
     expect(prisma.stream.upsert).not.toHaveBeenCalled();
   });
 
+  it('should return 400 when depositedAmount exceeds i128 max (#1532)', async () => {
+    const invalidData = {
+      streamId: '2',
+      sender: 'GTEST_USER_PUBLIC_KEY',
+      recipient: 'GDEF456ABC789GHI012JKL345MNO678PQR901STU234VWX567YZA123BCD',
+      tokenAddress: 'CBCD789EFG012HIJ345KLM678NOP901QRS234TUV567WXY890ZAB123CDE',
+      ratePerSecond: '100',
+      depositedAmount: '170141183460469231731687303715884105728',
+      startTime: '1700000000',
+    };
+
+    const response = await request(app)
+      .post('/v1/streams')
+      .send(invalidData)
+      .set('Accept', 'application/json');
+
+    expect(response.status).toBe(400);
+    expect(JSON.stringify(response.body)).toContain('depositedAmount');
+    expect(prisma.stream.upsert).not.toHaveBeenCalled();
+  });
+
   it('should return 400 when a required numeric field is missing', async () => {
     const invalidData = {
       streamId: '2',

backend/tests/stream.validator.test.ts

@@ -64,6 +64,35 @@ describe('Stream Validator', () => {
     expect(result.error!.issues[0]!.message).toBe('Rate exceeds maximum allowed value');
   });
 
+  it('should reject depositedAmount exceeding i128 max (#1532)', () => {
+    const data = {
+      streamId: '123',
+      sender: 'GSENDER',
+      recipient: 'GRECIPIENT',
+      tokenAddress: 'TABC',
+      ratePerSecond: '100',
+      depositedAmount: '170141183460469231731687303715884105728',
+      startTime: 1622505600,
+    };
+    const result = createStreamSchema.safeParse(data);
+    expect(result.success).toBe(false);
+    expect(result.error!.issues[0]!.message).toBe('Amount exceeds maximum allowed value');
+  });
+
+  it('should accept depositedAmount at i128 max (#1532)', () => {
+    const data = {
+      streamId: '123',
+      sender: 'GSENDER',
+      recipient: 'GRECIPIENT',
+      tokenAddress: 'TABC',
+      ratePerSecond: '100',
+      depositedAmount: '170141183460469231731687303715884105727',
+      startTime: 1622505600,
+    };
+    const result = createStreamSchema.safeParse(data);
+    expect(result.success).toBe(true);
+  });
+
   it('should accept ratePerSecond at i128 max', () => {
     const data = {
       streamId: '123',