feat: Streaming payment improvements

backend/STREAMING_SECURITY.md

@@ -0,0 +1,87 @@
+# Streaming Payment Security
+
+## Per-Stream Secret Implementation
+
+As an interim improvement before full delegated signing, the streaming worker now uses per-stream secrets stored encrypted at rest in the database.
+
+### Architecture
+
+- Each `PaymentStream` has an encrypted `senderSecret` field
+- Secrets are encrypted using `STREAM_SECRET_ENCRYPTION_KEY` environment variable
+- The streaming worker decrypts secrets only when processing payments
+- Decrypted secrets are never persisted to disk or logs
+
+### Security Trade-offs
+
+#### Current Implementation (Per-Stream Secrets)
+
+**Advantages:**
+- Each stream can use a different sender's secret
+- Enables multi-user streaming scenarios
+- Secrets are encrypted at rest in the database
+- Reduces blast radius if one stream's secret is compromised
+
+**Disadvantages:**
+- Secrets must be decrypted in application memory during payment processing
+- If the application is compromised, all decrypted secrets in memory are at risk
+- Encryption key (`STREAM_SECRET_ENCRYPTION_KEY`) is a single point of failure
+- Requires secure key management and rotation procedures
+
+#### Previous Implementation (Single Global Secret)
+
+**Disadvantages:**
+- Single `STREAM_WORKER_SECRET` used for all streams
+- Compromise of one stream affects all streams
+- No per-user isolation
+
+### Recommended Future Improvements
+
+1. **Delegated Signing**: Use Stellar's multi-sig or a dedicated signing service
+   - Secrets never stored in application database
+   - Signing requests sent to secure signing service
+   - Reduces application attack surface
+
+2. **Hardware Security Module (HSM)**
+   - Store secrets in HSM instead of database
+   - HSM handles encryption/decryption
+   - Application never sees plaintext secrets
+
+3. **Key Rotation**
+   - Implement automatic key rotation for `STREAM_SECRET_ENCRYPTION_KEY`
+   - Re-encrypt all stored secrets with new key
+   - Maintain backward compatibility during rotation
+
+### Environment Configuration
+
+Required environment variables:
+
+```bash
+# Encryption key for per-stream secrets (must be 32 bytes for AES-256)
+STREAM_SECRET_ENCRYPTION_KEY=<32-byte-hex-string>
+
+# Optional: Key rotation
+STREAM_SECRET_ENCRYPTION_KEY_OLD=<previous-key-for-decryption-only>
+```
+
+### Usage
+
+When creating a stream, provide the sender's secret:
+
+```javascript
+const stream = await createStream({
+  senderPublicKey: 'G...',
+  senderSecret: 'S...',  // Sender's Stellar secret key
+  recipientPublicKey: 'G...',
+  rateAmount: 10,
+  intervalSeconds: 60,
+});
+```
+
+The secret is encrypted and stored. During payment processing, it's decrypted only when needed.
+
+### Audit & Monitoring
+
+- All stream creation/update events are logged
+- Payment processing failures are logged with stream ID (not secret)
+- Monitor for unusual decryption patterns or errors
+- Implement alerts for repeated decryption failures

backend/src/routes/streaming.js

@@ -76,31 +76,43 @@ router.post('/', streamRules.create, validate, async (req, res) => {
  * @swagger
  * /api/streaming:
  *   get:
- *     summary: List streaming payments
+ *     summary: List streaming payments for authenticated user
  *     tags: [Streaming]
  *     parameters:
  *       - in: query
  *         name: senderPublicKey
  *         schema: { type: string }
- *         description: Filter by sender public key
+ *         required: true
+ *         description: Sender's public key to filter streams
  *     responses:
  *       200:
- *         description: List of streams
+ *         description: List of streams with status, totalStreamed, and nextPaymentAt
+ *       400:
+ *         description: Missing senderPublicKey parameter
  *       500:
  *         description: Server error
  */
 router.get('/', async (req, res) => {
   try {
     const { senderPublicKey } = req.query;
-    const where = senderPublicKey
-      ? { sender: { publicKey: senderPublicKey } }
-      : {};
+    if (!senderPublicKey) {
+      return res.status(400).json({ error: 'senderPublicKey query parameter is required' });
+    }
+
     const streams = await StreamingService.prisma.paymentStream.findMany({
-      where,
+      where: { sender: { publicKey: senderPublicKey } },
       include: { sender: true, recipient: true },
       orderBy: { startTime: 'desc' },
     });
-    res.json(streams);
+
+    const enriched = streams.map(stream => ({
+      ...stream,
+      nextPaymentAt: stream.status === 'ACTIVE' 
+        ? new Date(new Date(stream.lastProcessedAt).getTime() + stream.intervalSeconds * 1000)
+        : null,
+    }));
+
+    res.json(enriched);
   } catch (error) {
     res.status(500).json({ error: error.message });
   }
@@ -238,4 +250,49 @@ router.post('/:id/cancel', streamRules.idParam, validate, async (req, res) => {
   }
 });
 
+/**
+ * @swagger
+ * /api/streaming/{id}:
+ *   patch:
+ *     summary: Update a streaming payment (rate, interval, or endTime)
+ *     tags: [Streaming]
+ *     parameters:
+ *       - in: path
+ *         name: id
+ *         required: true
+ *         schema: { type: string, format: uuid }
+ *     requestBody:
+ *       required: true
+ *       content:
+ *         application/json:
+ *           schema:
+ *             type: object
+ *             properties:
+ *               rateAmount: { type: number, description: New amount per interval }
+ *               intervalSeconds: { type: integer, minimum: 10, description: New interval in seconds }
+ *               endTime: { type: string, format: date-time, description: New end time }
+ *     responses:
+ *       200:
+ *         description: Stream updated
+ *       400:
+ *         description: Validation error or invalid stream status
+ *       404:
+ *         description: Stream not found
+ *       500:
+ *         description: Server error
+ */
+router.patch('/:id', streamRules.idParam, [
+  body('rateAmount').optional().isFloat({ gt: 0 }).withMessage('rateAmount must be a positive number'),
+  body('intervalSeconds').optional().isInt({ min: 10 }).withMessage('intervalSeconds must be at least 10'),
+  body('endTime').optional().isISO8601().withMessage('endTime must be a valid ISO8601 date'),
+], validate, async (req, res) => {
+  try {
+    const stream = await StreamingService.updateStream(req.params.id, req.body);
+    res.json(stream);
+  } catch (error) {
+    const statusCode = error.message.includes('not found') ? 404 : 400;
+    res.status(statusCode).json({ error: error.message });
+  }
+});
+
 export default router;

backend/src/services/streaming.js

@@ -5,6 +5,17 @@ import { eventMonitor } from '../eventSourcing/index.js';
 import logger from '../config/logger.js';
 import { encryptToEnvValue, decryptFromEnvValue } from '../config/secrets.js';
 
+/**
+ * Per-stream secret encryption/decryption
+ * 
+ * SECURITY MODEL:
+ * - Each PaymentStream stores an encrypted senderSecret
+ * - Secrets are encrypted at rest using STREAM_SECRET_ENCRYPTION_KEY
+ * - Decryption happens only during payment processing
+ * - This is an interim solution before full delegated signing
+ * 
+ * See STREAMING_SECURITY.md for detailed security trade-offs and future improvements
+ */
 function getStreamEncryptionKey() {
   const key = process.env.STREAM_SECRET_ENCRYPTION_KEY;
   if (!key) throw new Error('STREAM_SECRET_ENCRYPTION_KEY is not set');
@@ -99,6 +110,37 @@ export async function cancelStream(id) {
   return stream;
 }
 
+export async function updateStream(id, updates) {
+  const stream = await prisma.paymentStream.findUnique({
+    where: { id },
+    include: { sender: true },
+  });
+
+  if (!stream) throw new Error('Stream not found');
+  if (!['ACTIVE', 'PAUSED'].includes(stream.status)) {
+    throw new Error(`Cannot update stream with status ${stream.status}`);
+  }
+
+  const updateData = {};
+  if (updates.rateAmount !== undefined) updateData.rateAmount = updates.rateAmount;
+  if (updates.intervalSeconds !== undefined) updateData.intervalSeconds = updates.intervalSeconds;
+  if (updates.endTime !== undefined) updateData.endTime = updates.endTime ? new Date(updates.endTime) : null;
+
+  const updated = await prisma.paymentStream.update({
+    where: { id },
+    data: updateData,
+    include: { sender: true },
+  });
+
+  await eventMonitor.publishEvent(stream.sender.publicKey, {
+    type: 'StreamUpdated',
+    data: { streamId: id, updates: updateData },
+    version: 1,
+  });
+
+  return updated;
+}
+
 export async function getStreamAnalytics() {
   const [statusCounts, totalVolumeResult, assets] = await Promise.all([
     prisma.paymentStream.groupBy({
@@ -108,9 +150,11 @@ export async function getStreamAnalytics() {
     prisma.paymentStream.aggregate({
       _sum: { totalStreamed: true },
     }),
-    prisma.paymentStream.findMany({
-      select: { assetCode: true },
-      distinct: ['assetCode'],
+    prisma.paymentStream.groupBy({
+      by: ['assetCode'],
+      _count: true,
+      orderBy: { _count: { assetCode: 'desc' } },
+      take: 10,
     }),
   ]);
 
@@ -119,15 +163,15 @@ export async function getStreamAnalytics() {
     return acc;
   }, {});
 
-  const totalVolume = statusMap.totalStreamed || 0;
-
   return {
     totalVolume: (totalVolumeResult._sum.totalStreamed || 0).toFixed(7),
     activeStreams: statusMap.ACTIVE || 0,
     pausedStreams: statusMap.PAUSED || 0,
     failedStreams: statusMap.FAILED || 0,
+    completedStreams: statusMap.COMPLETED || 0,
+    cancelledStreams: statusMap.CANCELLED || 0,
     totalStreams: Object.values(statusMap).reduce((a, b) => a + b, 0),
-    topAssets: assets.map(a => a.assetCode),
+    topAssets: assets.map(a => ({ assetCode: a.assetCode, count: a._count })),
   };
 }