Skip to content

Commit 039765f

Browse files
Benedict315Benedict315
authored
Merge branch 'main' into fix/jwt-signature-verification
2 parents d251d4d + ca944dc commit 039765f

5 files changed

Lines changed: 283 additions & 48 deletions

File tree

‎backend/src/controllers/stream.controller.ts‎

Lines changed: 45 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,35 @@ function sumStringI128(values: string[]): string {
6161
return total.toString();
6262
}
6363

64+
/**
65+
* Thrown when a request body field fails presence/format validation. Kept
66+
* distinct from generic errors so createStream can reliably map it to a 400
67+
* response instead of falling through to the catch-all 500.
68+
*/
69+
class StreamValidationError extends Error {
70+
constructor(message: string) {
71+
super(message);
72+
this.name = 'StreamValidationError';
73+
}
74+
}
75+
76+
/**
77+
* Validate presence and integer format of a required i128-style field, then
78+
* coerce it to a BigInt. Any missing value or conversion failure (SyntaxError
79+
* from a non-numeric string, TypeError from undefined/null/objects, etc.) is
80+
* normalized into a StreamValidationError so the caller can map it to 400.
81+
*/
82+
function parseRequiredBigIntField(fieldName: string, value: unknown): bigint {
83+
if (value === undefined || value === null || value === '') {
84+
throw new StreamValidationError(`Missing required field: ${fieldName}`);
85+
}
86+
try {
87+
return BigInt(value as bigint | number | string | boolean);
88+
} catch {
89+
throw new StreamValidationError(`Invalid ${fieldName}: must be a valid integer`);
90+
}
91+
}
92+
6493
/**
6594
* Create a new stream (stub for on-chain indexing)
6695
*/
@@ -70,8 +99,6 @@ export const createStream = async (req: Request, res: Response) => {
7099

71100
const parsedStreamId = Number.parseInt(streamId, 10);
72101
const parsedStartTime = Number.parseInt(startTime, 10);
73-
const parsedRatePerSecond = BigInt(ratePerSecond);
74-
const parsedDepositedAmount = BigInt(depositedAmount);
75102

76103
if (!Number.isFinite(parsedStreamId)) {
77104
return res.status(400).json({ error: 'Invalid streamId: must be a valid integer' });
@@ -81,6 +108,21 @@ export const createStream = async (req: Request, res: Response) => {
81108
return res.status(400).json({ error: 'Invalid startTime: must be a non-negative integer' });
82109
}
83110

111+
// Presence/format validation happens here, before any BigInt coercion,
112+
// so a malformed or missing numeric field always yields 400 rather than
113+
// an uncaught SyntaxError/TypeError falling through to 500.
114+
let parsedRatePerSecond: bigint;
115+
let parsedDepositedAmount: bigint;
116+
try {
117+
parsedRatePerSecond = parseRequiredBigIntField('ratePerSecond', ratePerSecond);
118+
parsedDepositedAmount = parseRequiredBigIntField('depositedAmount', depositedAmount);
119+
} catch (validationError) {
120+
if (validationError instanceof StreamValidationError) {
121+
return res.status(400).json({ error: validationError.message });
122+
}
123+
throw validationError;
124+
}
125+
84126
if (parsedRatePerSecond <= 0n) {
85127
return res.status(400).json({ error: 'Invalid ratePerSecond: must be greater than zero' });
86128
}
@@ -774,4 +816,4 @@ export const resumeStream = async (req: Request, res: Response) => {
774816
logger.error('Error resuming stream:', error);
775817
return res.status(500).json({ error: 'Internal server error' });
776818
}
777-
};
819+
};

‎backend/src/middleware/auth.ts‎

Lines changed: 19 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -99,19 +99,23 @@ export function verifyJwt(token: string): { publicKey: string } | null {
9999
const [header, body, sig] = token.split('.');
100100
if (!header || !body || !sig) return null;
101101

102-
// Verify signature
102+
// Compute the expected signature and re-encode it as base64url so we can
103+
// compare strings directly, rather than decoding the provided signature
104+
// to bytes first. Decoding base64url to bytes silently discards the
105+
// unused trailing bits of the final character (a 32-byte digest only
106+
// uses 4 of the 6 bits in its last base64 character), which means a
107+
// tampered last character can decode to the *same* bytes as the
108+
// original and slip past a byte-level comparison undetected.
103109
const expected = crypto
104110
.createHmac('sha256', JWT_SECRET)
105111
.update(`${header}.${body}`)
106112
.digest();
113+
const expectedSig = b64url(expected);
107114

108-
let providedSig: Buffer;
109-
try {
110-
providedSig = Buffer.from(sig, 'base64url');
111-
} catch {
112-
return null;
113-
}
115+
const providedSigBuf = Buffer.from(sig);
116+
const expectedSigBuf = Buffer.from(expectedSig);
114117

118+
<<<<<<< fix/jwt-signature-verification
115119
// Use timingSafeEqual to prevent timing attacks
116120
// This will throw if lengths differ, or return false if content differs
117121
try {
@@ -121,6 +125,14 @@ export function verifyJwt(token: string): { publicKey: string } | null {
121125
return null;
122126
}
123127
} catch {
128+
=======
129+
// Use timingSafeEqual to prevent timing attacks. Lengths are checked
130+
// first since timingSafeEqual throws on mismatched buffer lengths.
131+
if (
132+
providedSigBuf.length !== expectedSigBuf.length ||
133+
!crypto.timingSafeEqual(providedSigBuf, expectedSigBuf)
134+
) {
135+
>>>>>>> main
124136
return null;
125137
}
126138

‎backend/src/workers/soroban-event-worker.ts‎

Lines changed: 44 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -553,6 +553,17 @@ export class SorobanEventWorker {
553553
const timestamp = Math.floor(Date.now() / 1000);
554554

555555
await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
556+
// Check for a duplicate BEFORE mutating any Stream fields so that a
557+
// replayed event never re-applies the top-up.
558+
const existingEvent = await tx.streamEvent.findUnique({
559+
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'TOPPED_UP' } },
560+
select: { id: true },
561+
});
562+
if (existingEvent) {
563+
logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=TOPPED_UP`);
564+
return;
565+
}
566+
556567
const stream = await tx.stream.findUniqueOrThrow({
557568
where: { streamId },
558569
select: { ratePerSecond: true, startTime: true, totalPausedDuration: true }
@@ -575,27 +586,19 @@ export class SorobanEventWorker {
575586
},
576587
});
577588

578-
const existingEvent = await tx.streamEvent.findUnique({
589+
await tx.streamEvent.upsert({
579590
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'TOPPED_UP' } },
580-
select: { id: true },
591+
create: {
592+
streamId,
593+
eventType: 'TOPPED_UP',
594+
amount,
595+
transactionHash: event.txHash,
596+
ledgerSequence: event.ledger,
597+
timestamp,
598+
metadata: JSON.stringify({ newDepositedAmount }),
599+
},
600+
update: {},
581601
});
582-
if (existingEvent) {
583-
logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=TOPPED_UP`);
584-
} else {
585-
await tx.streamEvent.upsert({
586-
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'TOPPED_UP' } },
587-
create: {
588-
streamId,
589-
eventType: 'TOPPED_UP',
590-
amount,
591-
transactionHash: event.txHash,
592-
ledgerSequence: event.ledger,
593-
timestamp,
594-
metadata: JSON.stringify({ newDepositedAmount }),
595-
},
596-
update: {},
597-
});
598-
}
599602
});
600603

601604
sseService.broadcastToStream(String(streamId), 'stream.topped_up', {
@@ -624,6 +627,17 @@ export class SorobanEventWorker {
624627
const timestamp = Number(decodeU64(body['timestamp']));
625628

626629
await prisma.$transaction(async (tx: Prisma.TransactionClient) => {
630+
// Check for a duplicate BEFORE mutating any Stream fields so that a
631+
// replayed event never double-increments withdrawnAmount.
632+
const existingEvent = await tx.streamEvent.findUnique({
633+
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
634+
select: { id: true },
635+
});
636+
if (existingEvent) {
637+
logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=WITHDRAWN`);
638+
return;
639+
}
640+
627641
const stream = await tx.stream.findUniqueOrThrow({
628642
where: { streamId },
629643
select: { withdrawnAmount: true },
@@ -641,27 +655,19 @@ export class SorobanEventWorker {
641655
},
642656
});
643657

644-
const existingEvent = await tx.streamEvent.findUnique({
658+
await tx.streamEvent.upsert({
645659
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
646-
select: { id: true },
660+
create: {
661+
streamId,
662+
eventType: 'WITHDRAWN',
663+
amount,
664+
transactionHash: event.txHash,
665+
ledgerSequence: event.ledger,
666+
timestamp,
667+
metadata: JSON.stringify({ recipient }),
668+
},
669+
update: {},
647670
});
648-
if (existingEvent) {
649-
logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=WITHDRAWN`);
650-
} else {
651-
await tx.streamEvent.upsert({
652-
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
653-
create: {
654-
streamId,
655-
eventType: 'WITHDRAWN',
656-
amount,
657-
transactionHash: event.txHash,
658-
ledgerSequence: event.ledger,
659-
timestamp,
660-
metadata: JSON.stringify({ recipient }),
661-
},
662-
update: {},
663-
});
664-
}
665671
});
666672

667673
sseService.broadcastToStream(String(streamId), 'stream.withdrawn', {

‎backend/tests/soroban-event-worker.test.ts‎

Lines changed: 135 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -443,6 +443,141 @@ describe('SorobanEventWorker', () => {
443443
expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('Duplicate StreamEvent skipped'));
444444
});
445445

446+
it('should not double-increment withdrawnAmount when a tokens_withdrawn event is re-processed', async () => {
447+
const txHash = 'withdraw-tx-hash';
448+
const streamId = 21;
449+
450+
const mockEvent: rpc.Api.EventResponse = {
451+
id: 'withdraw-event-1',
452+
type: 'contract',
453+
ledger: 4000,
454+
ledgerClosedAt: '2024-01-01T00:00:00Z',
455+
txHash,
456+
transactionIndex: 0,
457+
operationIndex: 0,
458+
inSuccessfulContractCall: true,
459+
topic: [
460+
{ switch: () => ({ value: 0 }), sym: () => 'tokens_withdrawn' } as any,
461+
{ switch: () => ({ value: 1 }), u64: () => ({ toString: () => streamId.toString() }) } as any,
462+
],
463+
value: {
464+
switch: () => ({ value: 4 }),
465+
map: () => [
466+
{ key: () => ({ sym: () => 'recipient' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) },
467+
{ key: () => ({ sym: () => 'amount' }), val: () => ({ i128: () => ({ hi: () => ({ toString: () => '0' }), lo: () => ({ toString: () => '500' }) }) }) },
468+
{ key: () => ({ sym: () => 'timestamp' }), val: () => ({ u64: () => ({ toString: () => '1700002000' }) }) },
469+
] as any,
470+
} as any,
471+
};
472+
473+
// withdrawnAmount starts at '1000'; a single successful withdrawal of
474+
// 500 should bring it to '1500' and stay there under replay.
475+
const mockTx = {
476+
stream: {
477+
findUniqueOrThrow: vi.fn().mockResolvedValue({ withdrawnAmount: '1000' }),
478+
update: vi.fn().mockResolvedValue({}),
479+
},
480+
streamEvent: {
481+
findUnique: vi.fn(),
482+
upsert: vi.fn().mockResolvedValue({ id: 'withdraw-event-row' }),
483+
},
484+
};
485+
486+
(prisma.$transaction as ReturnType<typeof vi.fn>).mockImplementation((cb) => cb(mockTx));
487+
488+
// First processing: no existing event → withdrawnAmount is updated once.
489+
mockTx.streamEvent.findUnique.mockResolvedValueOnce(null);
490+
await expect((worker as any).handleTokensWithdrawn(mockEvent, mockEvent.topic![1])).resolves.not.toThrow();
491+
expect(mockTx.stream.update).toHaveBeenCalledTimes(1);
492+
expect(mockTx.stream.update).toHaveBeenCalledWith({
493+
where: { streamId },
494+
data: { withdrawnAmount: '1500', lastUpdateTime: 1700002000 },
495+
});
496+
expect(mockTx.streamEvent.upsert).toHaveBeenCalledTimes(1);
497+
expect(logger.warn).not.toHaveBeenCalled();
498+
499+
vi.clearAllMocks();
500+
(prisma.$transaction as ReturnType<typeof vi.fn>).mockImplementation((cb) => cb(mockTx));
501+
502+
// Second processing (replay of same txHash): the event now exists, so
503+
// withdrawnAmount must NOT be touched a second time.
504+
mockTx.streamEvent.findUnique.mockResolvedValueOnce({ id: 'withdraw-event-row' });
505+
await expect((worker as any).handleTokensWithdrawn(mockEvent, mockEvent.topic![1])).resolves.not.toThrow();
506+
expect(mockTx.stream.update).not.toHaveBeenCalled();
507+
expect(mockTx.streamEvent.upsert).not.toHaveBeenCalled();
508+
expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('Duplicate StreamEvent skipped'));
509+
});
510+
511+
it('should not double-apply depositedAmount/endTime when a stream_topped_up event is re-processed', async () => {
512+
const txHash = 'topup-tx-hash';
513+
const streamId = 22;
514+
515+
const mockEvent: rpc.Api.EventResponse = {
516+
id: 'topup-event-1',
517+
type: 'contract',
518+
ledger: 4001,
519+
ledgerClosedAt: '2024-01-01T00:00:00Z',
520+
txHash,
521+
transactionIndex: 0,
522+
operationIndex: 0,
523+
inSuccessfulContractCall: true,
524+
topic: [
525+
{ switch: () => ({ value: 0 }), sym: () => 'stream_topped_up' } as any,
526+
{ switch: () => ({ value: 1 }), u64: () => ({ toString: () => streamId.toString() }) } as any,
527+
],
528+
value: {
529+
switch: () => ({ value: 4 }),
530+
map: () => [
531+
{ key: () => ({ sym: () => 'amount' }), val: () => ({ i128: () => ({ hi: () => ({ toString: () => '0' }), lo: () => ({ toString: () => '200' }) }) }) },
532+
{ key: () => ({ sym: () => 'new_deposited_amount' }), val: () => ({ i128: () => ({ hi: () => ({ toString: () => '0' }), lo: () => ({ toString: () => '1200' }) }) }) },
533+
] as any,
534+
} as any,
535+
};
536+
537+
const mockTx = {
538+
stream: {
539+
findUniqueOrThrow: vi.fn().mockResolvedValue({
540+
ratePerSecond: '10',
541+
startTime: 1700000000,
542+
totalPausedDuration: 0,
543+
}),
544+
update: vi.fn().mockResolvedValue({}),
545+
},
546+
streamEvent: {
547+
findUnique: vi.fn(),
548+
upsert: vi.fn().mockResolvedValue({ id: 'topup-event-row' }),
549+
},
550+
};
551+
552+
(prisma.$transaction as ReturnType<typeof vi.fn>).mockImplementation((cb) => cb(mockTx));
553+
554+
// First processing: no existing event → depositedAmount/endTime are set once.
555+
mockTx.streamEvent.findUnique.mockResolvedValueOnce(null);
556+
await expect((worker as any).handleStreamToppedUp(mockEvent, mockEvent.topic![1])).resolves.not.toThrow();
557+
expect(mockTx.stream.update).toHaveBeenCalledTimes(1);
558+
const firstUpdateArgs = mockTx.stream.update.mock.calls[0]![0];
559+
expect(firstUpdateArgs.data.depositedAmount).toBe('1200');
560+
const expectedEndTime = firstUpdateArgs.data.endTime;
561+
expect(mockTx.streamEvent.upsert).toHaveBeenCalledTimes(1);
562+
expect(logger.warn).not.toHaveBeenCalled();
563+
564+
vi.clearAllMocks();
565+
(prisma.$transaction as ReturnType<typeof vi.fn>).mockImplementation((cb) => cb(mockTx));
566+
567+
// Second processing (replay of same txHash): the event now exists, so
568+
// depositedAmount/endTime must NOT be re-applied.
569+
mockTx.streamEvent.findUnique.mockResolvedValueOnce({ id: 'topup-event-row' });
570+
await expect((worker as any).handleStreamToppedUp(mockEvent, mockEvent.topic![1])).resolves.not.toThrow();
571+
expect(mockTx.stream.update).not.toHaveBeenCalled();
572+
expect(mockTx.streamEvent.upsert).not.toHaveBeenCalled();
573+
expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('Duplicate StreamEvent skipped'));
574+
575+
// Sanity check: depositedAmount/endTime from the (only) applied update
576+
// match what a single application should produce.
577+
expect(firstUpdateArgs.data.depositedAmount).toBe('1200');
578+
expect(expectedEndTime).toBe(1700000000 + Math.floor(1200 / 10) + 0);
579+
});
580+
446581
it('should process admin_transferred events successfully', async () => {
447582
const txHash = 'admin-transferred-tx-hash';
448583

0 commit comments

Comments
 (0)