Skip to content

Commit 37e173f

Browse files
authored
Merge pull request #286 from shakurJJ/feat/indexer-auth-admin-rpc-fallback
2 parents 4ab5e4a + 0d199f1 commit 37e173f

10 files changed

Lines changed: 539 additions & 10 deletions

File tree

backend/.env.example

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,12 @@ INDEXER_POLL_INTERVAL_MS=5000
3131
# Ledger sequence to start indexing from on first run (0 = latest)
3232
INDEXER_START_LEDGER=0
3333

34+
# ─── Auth ─────────────────────────────────────────────────────────────────────
35+
# Secret used to sign JWTs (generate with: openssl rand -hex 32)
36+
JWT_SECRET=
37+
38+
# Stellar public key of the admin user (for /v1/admin/* endpoints)
39+
ADMIN_PUBLIC_KEY=
3440
# ─── Redis (optional) ────────────────────────────────────────────────────────
3541
# When set, enables horizontal SSE scaling via Redis pub/sub so events
3642
# emitted by any backend instance are broadcast to clients on all instances.

backend/src/controllers/sse.controller.ts

Lines changed: 22 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,11 @@
11
import type { Request, Response } from 'express';
22
import { sseService } from '../services/sse.service.js';
3+
import { prisma } from '../lib/prisma.js';
4+
import type { AuthenticatedRequest } from '../types/auth.types.js';
35
import { z } from 'zod';
46

57
const subscribeSchema = z.object({
68
streams: z.array(z.string()).optional().default([]),
7-
users: z.array(z.string()).optional().default([]),
89
all: z.boolean().optional().default(false),
910
});
1011

@@ -14,17 +15,30 @@ export const subscribe = (req: Request, res: Response) => {
1415
}
1516

1617
try {
17-
const { streams, users, all } = subscribeSchema.parse(req.query);
18-
19-
const subscriptions: string[] = [];
20-
18+
const { publicKey } = (req as AuthenticatedRequest).user;
19+
const { streams, all } = subscribeSchema.parse(req.query);
20+
21+
// Scope: only streams where the authenticated user is sender or recipient
22+
const ownedStreams = await prisma.stream.findMany({
23+
where: { OR: [{ sender: publicKey }, { recipient: publicKey }] },
24+
select: { streamId: true },
25+
});
26+
const ownedIds = new Set(ownedStreams.map((s) => String(s.streamId)));
27+
28+
let subscriptions: string[];
2129
if (all) {
22-
subscriptions.push('*');
30+
// "all" still scoped to the user's own streams
31+
subscriptions = [...ownedIds];
32+
} else if (streams.length > 0) {
33+
// Only allow subscribing to streams the user owns
34+
subscriptions = streams.filter((id) => ownedIds.has(id));
2335
} else {
24-
subscriptions.push(...streams);
25-
subscriptions.push(...users.map(u => `user:${u}`));
36+
subscriptions = [...ownedIds];
2637
}
2738

39+
// Always add user-scoped subscription key
40+
subscriptions.push(`user:${publicKey}`);
41+
2842
const clientId = `${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
2943

3044
res.writeHead(200, {

backend/src/controllers/stream.controller.ts

Lines changed: 42 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import type { Request, Response } from 'express';
22
import { prisma } from '../lib/prisma.js';
33
import logger from '../logger.js';
44
import { claimableAmountService } from '../services/claimable.service.js';
5+
import { getStreamFromChain, getClaimableFromChain, isStale } from '../services/sorobanService.js';
56

67
/**
78
* Create a new stream (stub for on-chain indexing)
@@ -92,7 +93,20 @@ export const getStream = async (req: Request, res: Response) => {
9293
});
9394

9495
if (!stream) {
95-
return res.status(404).json({ error: 'Stream not found' });
96+
// Fallback: try live RPC
97+
const chainStream = await getStreamFromChain(parsedStreamId);
98+
if (!chainStream) {
99+
return res.status(404).json({ error: 'Stream not found' });
100+
}
101+
return res.status(200).json({ ...chainStream, source: 'chain' });
102+
}
103+
104+
// If DB data is stale, attempt live RPC fallback
105+
if (isStale(stream.updatedAt)) {
106+
const chainStream = await getStreamFromChain(parsedStreamId);
107+
if (chainStream) {
108+
return res.status(200).json({ ...stream, ...chainStream, source: 'chain' });
109+
}
96110
}
97111

98112
return res.status(200).json(stream);
@@ -192,9 +206,36 @@ export const getStreamClaimableAmount = async (req: Request, res: Response) => {
192206
});
193207

194208
if (!stream) {
209+
// Fallback: try live RPC for claimable amount
210+
const chainClaimable = await getClaimableFromChain(parsedStreamId);
211+
if (chainClaimable !== null) {
212+
return res.status(200).json({
213+
streamId: parsedStreamId,
214+
claimableAmount: chainClaimable,
215+
actionable: BigInt(chainClaimable) > 0n,
216+
calculatedAt: Math.floor(Date.now() / 1000),
217+
cached: false,
218+
source: 'chain',
219+
});
220+
}
195221
return res.status(404).json({ error: 'Stream not found' });
196222
}
197223

224+
// If DB data is stale, use live RPC
225+
if (isStale(stream.updatedAt)) {
226+
const chainClaimable = await getClaimableFromChain(parsedStreamId);
227+
if (chainClaimable !== null) {
228+
return res.status(200).json({
229+
streamId: parsedStreamId,
230+
claimableAmount: chainClaimable,
231+
actionable: BigInt(chainClaimable) > 0n,
232+
calculatedAt: Math.floor(Date.now() / 1000),
233+
cached: false,
234+
source: 'chain',
235+
});
236+
}
237+
}
238+
198239
const result = claimableAmountService.getClaimableAmount(stream, requestedAt);
199240

200241
return res.status(200).json(result);

backend/src/middleware/auth.ts

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,149 @@
1+
import type { Request, Response, NextFunction } from 'express';
2+
import * as crypto from 'crypto';
3+
import * as StellarSdk from '@stellar/stellar-sdk';
4+
import type { AuthenticatedRequest } from '../types/auth.types.js';
5+
import logger from '../logger.js';
6+
7+
const JWT_SECRET = process.env.JWT_SECRET ?? crypto.randomBytes(32).toString('hex');
8+
const JWT_EXPIRY_SECONDS = 3600; // 1 hour max per spec
9+
10+
const STELLAR_NETWORK =
11+
process.env.STELLAR_NETWORK === 'mainnet'
12+
? StellarSdk.Networks.PUBLIC
13+
: StellarSdk.Networks.TESTNET;
14+
15+
// In-memory challenge store: publicKey -> { nonce, expiresAt }
16+
const challenges = new Map<string, { nonce: string; expiresAt: number }>();
17+
18+
// ─── Minimal JWT (no external dep) ──────────────────────────────────────────
19+
20+
function b64url(buf: Buffer | string): string {
21+
const b = typeof buf === 'string' ? Buffer.from(buf) : buf;
22+
return b.toString('base64url');
23+
}
24+
25+
function signJwt(payload: object): string {
26+
const header = b64url(JSON.stringify({ alg: 'HS256', typ: 'JWT' }));
27+
const body = b64url(JSON.stringify(payload));
28+
const sig = crypto
29+
.createHmac('sha256', JWT_SECRET)
30+
.update(`${header}.${body}`)
31+
.digest();
32+
return `${header}.${body}.${b64url(sig)}`;
33+
}
34+
35+
function verifyJwt(token: string): { publicKey: string } | null {
36+
try {
37+
const [header, body, sig] = token.split('.');
38+
if (!header || !body || !sig) return null;
39+
const expected = crypto
40+
.createHmac('sha256', JWT_SECRET)
41+
.update(`${header}.${body}`)
42+
.digest();
43+
if (!crypto.timingSafeEqual(Buffer.from(sig, 'base64url'), expected)) return null;
44+
const payload = JSON.parse(Buffer.from(body, 'base64url').toString());
45+
if (payload.exp < Math.floor(Date.now() / 1000)) return null;
46+
return { publicKey: payload.sub };
47+
} catch {
48+
return null;
49+
}
50+
}
51+
52+
// ─── Challenge / Verify handlers ────────────────────────────────────────────
53+
54+
export function issueChallenge(req: Request, res: Response): void {
55+
const { publicKey } = req.body as { publicKey?: string };
56+
if (!publicKey || !StellarSdk.StrKey.isValidEd25519PublicKey(publicKey)) {
57+
res.status(400).json({ error: 'Invalid publicKey' });
58+
return;
59+
}
60+
const nonce = crypto.randomBytes(32).toString('hex');
61+
challenges.set(publicKey, { nonce, expiresAt: Date.now() + 60_000 }); // 60s to sign
62+
res.json({ nonce, expiresAt: Date.now() + 60_000 });
63+
}
64+
65+
export function verifyChallenge(req: Request, res: Response): void {
66+
const { publicKey, signedTransaction } = req.body as {
67+
publicKey?: string;
68+
signedTransaction?: string;
69+
};
70+
71+
if (!publicKey || !signedTransaction) {
72+
res.status(400).json({ error: 'publicKey and signedTransaction required' });
73+
return;
74+
}
75+
76+
const challenge = challenges.get(publicKey);
77+
if (!challenge || challenge.expiresAt < Date.now()) {
78+
res.status(401).json({ error: 'Challenge expired or not found' });
79+
return;
80+
}
81+
82+
try {
83+
const tx = StellarSdk.TransactionBuilder.fromXDR(
84+
signedTransaction,
85+
STELLAR_NETWORK,
86+
) as StellarSdk.Transaction;
87+
88+
if (tx.source !== publicKey) {
89+
res.status(401).json({ error: 'Transaction source does not match publicKey' });
90+
return;
91+
}
92+
93+
// Verify the manage_data op contains our nonce
94+
const op = tx.operations[0] as StellarSdk.Operation.ManageData | undefined;
95+
if (!op || op.type !== 'manageData' || op.value?.toString('hex') !== challenge.nonce) {
96+
res.status(401).json({ error: 'Invalid challenge nonce in transaction' });
97+
return;
98+
}
99+
100+
const keypair = StellarSdk.Keypair.fromPublicKey(publicKey);
101+
const txHash = tx.hash();
102+
const valid = tx.signatures.some((s) => {
103+
try { return keypair.verify(txHash, s.signature()); } catch { return false; }
104+
});
105+
106+
if (!valid) {
107+
res.status(401).json({ error: 'Invalid signature' });
108+
return;
109+
}
110+
111+
challenges.delete(publicKey);
112+
113+
const now = Math.floor(Date.now() / 1000);
114+
const token = signJwt({ sub: publicKey, iat: now, exp: now + JWT_EXPIRY_SECONDS });
115+
res.json({ token, expiresIn: JWT_EXPIRY_SECONDS });
116+
} catch (err) {
117+
logger.error('[Auth] verifyChallenge error:', err);
118+
res.status(401).json({ error: 'Invalid signed transaction' });
119+
}
120+
}
121+
122+
// ─── Auth middleware ─────────────────────────────────────────────────────────
123+
124+
export function requireAuth(req: Request, res: Response, next: NextFunction): void {
125+
const header = req.headers.authorization;
126+
if (!header?.startsWith('Bearer ')) {
127+
res.status(401).json({ error: 'Unauthorized', message: 'Missing Bearer token' });
128+
return;
129+
}
130+
const payload = verifyJwt(header.slice(7));
131+
if (!payload) {
132+
res.status(401).json({ error: 'Unauthorized', message: 'Invalid or expired token' });
133+
return;
134+
}
135+
(req as AuthenticatedRequest).user = { publicKey: payload.publicKey };
136+
next();
137+
}
138+
139+
export function requireAdmin(req: Request, res: Response, next: NextFunction): void {
140+
requireAuth(req, res, () => {
141+
const user = (req as AuthenticatedRequest).user;
142+
const adminKey = process.env.ADMIN_PUBLIC_KEY;
143+
if (!adminKey || user.publicKey !== adminKey) {
144+
res.status(403).json({ error: 'Forbidden', message: 'Admin access required' });
145+
return;
146+
}
147+
next();
148+
});
149+
}
Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
import { Router } from 'express';
2+
import type { Request, Response } from 'express';
3+
import { requireAdmin } from '../../middleware/auth.js';
4+
import {
5+
getIndexerStatus,
6+
resetIndexer,
7+
replayFromLedger,
8+
} from '../../services/indexerService.js';
9+
10+
const router = Router();
11+
12+
// All admin routes require admin JWT
13+
router.use(requireAdmin);
14+
15+
/**
16+
* @openapi
17+
* /v1/admin/indexer/status:
18+
* get:
19+
* tags: [Admin]
20+
* summary: Get indexer status
21+
* security: [{ bearerAuth: [] }]
22+
* responses:
23+
* 200:
24+
* description: Indexer status
25+
*/
26+
router.get('/indexer/status', async (req: Request, res: Response) => {
27+
try {
28+
const status = await getIndexerStatus();
29+
res.json(status);
30+
} catch (err) {
31+
res.status(500).json({ error: 'Failed to fetch indexer status' });
32+
}
33+
});
34+
35+
/**
36+
* @openapi
37+
* /v1/admin/indexer/reset:
38+
* post:
39+
* tags: [Admin]
40+
* summary: Reset indexer lastProcessedLedger
41+
* security: [{ bearerAuth: [] }]
42+
* requestBody:
43+
* required: true
44+
* content:
45+
* application/json:
46+
* schema:
47+
* type: object
48+
* required: [ledger]
49+
* properties:
50+
* ledger:
51+
* type: integer
52+
* responses:
53+
* 200:
54+
* description: Reset successful
55+
*/
56+
router.post('/indexer/reset', async (req: Request, res: Response) => {
57+
const ledger = Number(req.body?.ledger);
58+
if (!Number.isInteger(ledger) || ledger < 0) {
59+
res.status(400).json({ error: 'ledger must be a non-negative integer' });
60+
return;
61+
}
62+
try {
63+
await resetIndexer(ledger);
64+
res.json({ ok: true, lastLedger: ledger });
65+
} catch (err) {
66+
res.status(500).json({ error: 'Reset failed' });
67+
}
68+
});
69+
70+
/**
71+
* @openapi
72+
* /v1/admin/indexer/replay:
73+
* post:
74+
* tags: [Admin]
75+
* summary: Replay events from a given ledger (idempotent)
76+
* security: [{ bearerAuth: [] }]
77+
* parameters:
78+
* - in: query
79+
* name: from_ledger
80+
* required: true
81+
* schema:
82+
* type: integer
83+
* responses:
84+
* 202:
85+
* description: Replay started
86+
*/
87+
router.post('/indexer/replay', async (req: Request, res: Response) => {
88+
const fromLedger = Number(req.query.from_ledger);
89+
if (!Number.isInteger(fromLedger) || fromLedger < 0) {
90+
res.status(400).json({ error: 'from_ledger must be a non-negative integer' });
91+
return;
92+
}
93+
try {
94+
await replayFromLedger(fromLedger);
95+
res.status(202).json({ ok: true, replayingFrom: fromLedger });
96+
} catch (err) {
97+
res.status(500).json({ error: 'Replay failed' });
98+
}
99+
});
100+
101+
export default router;

0 commit comments

Comments
 (0)