feat(security): add payer-aware rate limiting (#670)

backend/package.json

@@ -28,6 +28,7 @@
     "express": "^4.19.2",
     "express-rate-limit": "^7.5.0",
     "helmet": "^8.2.0",
+    "ioredis": "^6.0.0",
     "jsonwebtoken": "^9.0.2",
     "mqtt": "^5.7.0",
     "pino": "^10.3.1",

backend/src/config/rateLimits.ts

@@ -9,16 +9,16 @@
  * in one place.
  */
 
-/** Window duration in milliseconds (default: 60 s). */
+/** Anonymous request window in milliseconds (default: 15 minutes). */
 export const RATE_LIMIT_WINDOW_MS = Number(
-  process.env.RATE_LIMIT_WINDOW_MS ?? 60 * 1000,
+  process.env.RATE_LIMIT_WINDOW_MS ?? 15 * 60 * 1000,
 );
 
 /**
- * Maximum requests per window for the general / global limiter (default: 60).
+ * Maximum requests per window for the general / global limiter (default: 100).
  * This is also used as the ceiling for read-heavy routes.
  */
-export const RATE_LIMIT_MAX = Number(process.env.RATE_LIMIT_MAX ?? 60);
+export const RATE_LIMIT_MAX = Number(process.env.RATE_LIMIT_MAX ?? 100);
 
 /**
  * Maximum write requests per window for the strict write limiter (default:
@@ -31,12 +31,15 @@ export const WRITE_RATE_LIMIT_MAX = Number(
 );
 
 /**
- * Maximum payment requests per window (default: 10).  Payments hit the
- * Stellar network and cost gas, so they get a tighter budget than generic
- * writes.
+ * Maximum payer-scoped payment and meter-management requests per hour
+ * (default: 50). Payments hit the Stellar network and cost gas, so they get a
+ * tighter budget than generic anonymous reads.
  */
+export const PAYMENTS_RATE_LIMIT_WINDOW_MS = Number(
+  process.env.PAYMENTS_RATE_LIMIT_WINDOW_MS ?? 60 * 60 * 1000,
+);
 export const PAYMENTS_RATE_LIMIT_MAX = Number(
-  process.env.PAYMENTS_RATE_LIMIT_MAX ?? 10,
+  process.env.PAYMENTS_RATE_LIMIT_MAX ?? 50,
 );
 
 /** Human-readable message returned to clients that exceed any limiter. */

backend/src/index.ts

@@ -32,6 +32,7 @@ import { startIoTBridge } from "./iot/bridge.js";
 import { logger } from "./lib/logger.js";
 import { register } from "./lib/metrics.js";
 import { writeLimiter, paymentsLimiter } from "./middleware/rateLimit.js";
+import { payerRateLimiter } from "./middleware/payerRateLimit.js";
 import { sanitiseBody } from "./middleware/sanitise.js";
 import requestLoggerMiddleware from "./middleware/requestLogger.js";
 import {
@@ -217,8 +218,10 @@ try {
 }
 
 app.use("/api/admin/login", writeLimiter, adminLoginRouter);
-app.use("/api/meters", createMeterRouter(stellarService));
-app.use("/api/payments", writeLimiter, paymentsRouter);
+// Body parsing above makes payer/owner available before this limiter runs.
+// Missing payer identities remain governed by the global IP limiter.
+app.use("/api/meters", payerRateLimiter, createMeterRouter(stellarService));
+app.use("/api/payments", payerRateLimiter, writeLimiter, paymentsRouter);
 app.use("/api/webhooks", writeLimiter, webhookRouter);
 app.use("/api/allowlist", writeLimiter, allowlistRouter);
 app.use("/api/collaborators", collaboratorRouter);

backend/src/middleware/payerRateLimit.ts

@@ -0,0 +1,200 @@
+import Redis from "ioredis";
+import type { NextFunction, Request, Response, RequestHandler } from "express";
+import {
+  PAYMENTS_RATE_LIMIT_MAX,
+  PAYMENTS_RATE_LIMIT_WINDOW_MS,
+  RATE_LIMIT_MESSAGE,
+} from "../config/rateLimits.js";
+
+export type PayerRateLimitResult = {
+  allowed: boolean;
+  count: number;
+  remaining: number;
+  resetAt: number;
+  limit: number;
+};
+
+export interface PayerRateLimitStore {
+  check(key: string, now: number): Promise<PayerRateLimitResult>;
+  clear?(): void | Promise<void>;
+}
+
+function resultFromCount(count: number, oldest: number, now: number): PayerRateLimitResult {
+  const resetAt = oldest > 0 ? oldest + PAYMENTS_RATE_LIMIT_WINDOW_MS : now + PAYMENTS_RATE_LIMIT_WINDOW_MS;
+  return {
+    allowed: count <= PAYMENTS_RATE_LIMIT_MAX,
+    count,
+    remaining: Math.max(0, PAYMENTS_RATE_LIMIT_MAX - count),
+    resetAt,
+    limit: PAYMENTS_RATE_LIMIT_MAX,
+  };
+}
+
+/** Precise sliding-window store for single-instance deployments and fallback. */
+export class MemoryPayerRateLimitStore implements PayerRateLimitStore {
+  private readonly buckets = new Map<string, number[]>();
+
+  async check(key: string, now = Date.now()): Promise<PayerRateLimitResult> {
+    const cutoff = now - PAYMENTS_RATE_LIMIT_WINDOW_MS;
+    const active = (this.buckets.get(key) ?? []).filter((timestamp) => timestamp > cutoff);
+    active.push(now);
+    this.buckets.set(key, active);
+    return resultFromCount(active.length, active[0] ?? 0, now);
+  }
+
+  clear() {
+    this.buckets.clear();
+  }
+}
+
+const REDIS_WINDOW_SCRIPT = `
+local key = KEYS[1]
+local now = tonumber(ARGV[1])
+local window_ms = tonumber(ARGV[2])
+local limit = tonumber(ARGV[3])
+local ttl_seconds = tonumber(ARGV[4])
+local cutoff = now - window_ms
+
+redis.call('ZREMRANGEBYSCORE', key, '-inf', cutoff)
+local sequence = redis.call('INCR', key .. ':sequence')
+redis.call('ZADD', key, now, tostring(now) .. ':' .. tostring(sequence))
+redis.call('EXPIRE', key, ttl_seconds)
+redis.call('EXPIRE', key .. ':sequence', ttl_seconds)
+
+local count = redis.call('ZCARD', key)
+local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
+local oldest_score = 0
+if #oldest > 0 then
+  oldest_score = tonumber(oldest[2])
+end
+return { count, oldest_score }
+`;
+
+/** Redis-backed sliding window for multi-instance deployments. */
+export class RedisPayerRateLimitStore implements PayerRateLimitStore {
+  private readonly ttlSeconds = Math.ceil(PAYMENTS_RATE_LIMIT_WINDOW_MS / 1000) + 1;
+
+  constructor(private readonly redis: Redis) {}
+
+  async check(key: string, now = Date.now()): Promise<PayerRateLimitResult> {
+    const [count, oldest] = (await this.redis.eval(
+      REDIS_WINDOW_SCRIPT,
+      1,
+      key,
+      String(now),
+      String(PAYMENTS_RATE_LIMIT_WINDOW_MS),
+      String(PAYMENTS_RATE_LIMIT_MAX),
+      String(this.ttlSeconds),
+    )) as [number, number];
+    return resultFromCount(Number(count), Number(oldest), now);
+  }
+}
+
+let redisClient: Redis | null | undefined;
+let defaultStore: PayerRateLimitStore | undefined;
+const fallbackStore = new MemoryPayerRateLimitStore();
+
+function getRedisClient(): Redis | null {
+  if (redisClient !== undefined) return redisClient;
+  const url = process.env.REDIS_URL;
+  if (!url) {
+    redisClient = null;
+    return redisClient;
+  }
+
+  try {
+    const client = new Redis(url, {
+      connectTimeout: Number(process.env.REDIS_CONNECT_TIMEOUT ?? 2_000),
+      commandTimeout: Number(process.env.REDIS_COMMAND_TIMEOUT ?? 500),
+      maxRetriesPerRequest: 1,
+      enableOfflineQueue: false,
+      lazyConnect: false,
+    });
+    client.on("error", (error) => {
+      if (process.env.NODE_ENV !== "test") {
+        console.error("[payer-rate-limit] Redis error:", error.message);
+      }
+    });
+    redisClient = client;
+  } catch (error) {
+    console.error("[payer-rate-limit] Redis setup failed; using memory:", error);
+    redisClient = null;
+  }
+  return redisClient;
+}
+
+function getDefaultStore(): PayerRateLimitStore {
+  if (!defaultStore) {
+    const redis = getRedisClient();
+    defaultStore = redis ? new RedisPayerRateLimitStore(redis) : fallbackStore;
+  }
+  return defaultStore;
+}
+
+/**
+ * Extract the payer identity after express.json has parsed the request body.
+ * Registration uses `owner`; payment uses `payer`; callers may also provide
+ * X-Payer-Address for meter-management operations whose body lacks an owner.
+ */
+export function extractPayerAddress(req: Request): string | null {
+  const header = req.header("x-payer-address");
+  const body = req.body as Record<string, unknown> | undefined;
+  const candidate = body?.payer ?? body?.owner ?? body?.new_owner ?? header;
+  if (typeof candidate !== "string") return null;
+  const value = candidate.trim();
+  return value.length > 0 ? value : null;
+}
+
+function addRateLimitHeaders(res: Response, result: PayerRateLimitResult) {
+  const retryAfter = Math.max(1, Math.ceil((result.resetAt - Date.now()) / 1000));
+  res.setHeader("RateLimit-Limit", String(result.limit));
+  res.setHeader("RateLimit-Remaining", String(result.remaining));
+  res.setHeader("RateLimit-Reset", String(retryAfter));
+  res.setHeader("X-RateLimit-Limit", String(result.limit));
+  res.setHeader("X-RateLimit-Remaining", String(result.remaining));
+  res.setHeader("X-RateLimit-Reset", String(retryAfter));
+  return retryAfter;
+}
+
+/** Create an injectable payer limiter for route tests and deployments. */
+export function createPayerRateLimiter(store: PayerRateLimitStore = getDefaultStore()): RequestHandler {
+  return async (req: Request, res: Response, next: NextFunction) => {
+    const payer = extractPayerAddress(req);
+    if (!payer) {
+      // Anonymous requests remain governed by the global IP limiter.
+      return next();
+    }
+
+    const key = `rl:payer:${payer}`;
+    let result: PayerRateLimitResult;
+    try {
+      result = await store.check(key, Date.now());
+    } catch (error) {
+      // Redis is an optimization for sharing state; an outage must not make the
+      // API unavailable. Fall back to this instance’s bounded memory store.
+      if (process.env.NODE_ENV !== "test") {
+        console.error("[payer-rate-limit] store error; using memory:", error);
+      }
+      result = await fallbackStore.check(key, Date.now());
+    }
+
+    const retryAfter = addRateLimitHeaders(res, result);
+    if (!result.allowed) {
+      res.setHeader("Retry-After", String(retryAfter));
+      return res.status(429).json({
+        error: RATE_LIMIT_MESSAGE,
+        code: "RATE_LIMITED",
+        retryAfter,
+      });
+    }
+    return next();
+  };
+}
+
+export const payerRateLimiter = createPayerRateLimiter();
+
+/** Test-only reset hook; it does not affect Redis data. */
+export async function resetPayerRateLimiterState() {
+  fallbackStore.clear();
+  if (defaultStore?.clear) await defaultStore.clear();
+}

backend/src/middleware/rateLimit.ts

@@ -11,6 +11,8 @@
 import rateLimit from "express-rate-limit";
 import {
   RATE_LIMIT_WINDOW_MS,
+  PAYMENTS_RATE_LIMIT_WINDOW_MS,
+  PAYMENTS_RATE_LIMIT_MAX,
   WRITE_RATE_LIMIT_MAX,
   RATE_LIMIT_MESSAGE,
 } from "../config/rateLimits.js";
@@ -52,10 +54,21 @@ export const readLimiter = rateLimit({
   },
 });
 
+/**
+ * Legacy IP-based payment limiter retained for callers that need a standalone
+ * limiter. The main payment route uses payerRateLimiter first so authenticated
+ * requests are bucketed by payer address.
+ */
 export const paymentsLimiter = rateLimit({
-  windowMs,
-  max: parseInt(process.env.PAYMENTS_RATE_LIMIT_MAX ?? '10', 10),
+  windowMs: PAYMENTS_RATE_LIMIT_WINDOW_MS,
+  max: PAYMENTS_RATE_LIMIT_MAX,
   standardHeaders: true,
   legacyHeaders: false,
-  message: { error: 'Too many payment requests', code: 'RATE_LIMITED' },
+  handler: (_req, res) => {
+    res.setHeader(
+      "Retry-After",
+      String(Math.ceil(PAYMENTS_RATE_LIMIT_WINDOW_MS / 1000)),
+    );
+    res.status(429).json({ error: RATE_LIMIT_MESSAGE, code: "RATE_LIMITED" });
+  },
 });

backend/tests/rate-limit.integration.test.ts

@@ -0,0 +1,87 @@
+import { describe, expect, it, vi } from "vitest";
+import type { NextFunction, Request, Response } from "express";
+import {
+  createPayerRateLimiter,
+  extractPayerAddress,
+  MemoryPayerRateLimitStore,
+} from "../src/middleware/payerRateLimit.js";
+
+function request(body: Record<string, unknown> = {}, headers: Record<string, string> = {}) {
+  return {
+    body,
+    header: (name: string) => headers[name.toLowerCase()],
+  } as unknown as Request;
+}
+
+function response() {
+  const res = {
+    setHeader: vi.fn(),
+    status: vi.fn(),
+    json: vi.fn(),
+  } as unknown as Response;
+  vi.mocked(res.status).mockReturnValue(res);
+  return res;
+}
+
+async function invoke(
+  middleware: ReturnType<typeof createPayerRateLimiter>,
+  req: Request,
+  res: Response,
+) {
+  const next = vi.fn() as unknown as NextFunction;
+  await middleware(req, res, next);
+  return next as unknown as ReturnType<typeof vi.fn>;
+}
+
+describe("payer-aware rate limiting", () => {
+  it("allows 50 requests per payer and rejects the 51st with retry headers", async () => {
+    const middleware = createPayerRateLimiter(new MemoryPayerRateLimitStore());
+    const payer = "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF";
+    const req = request({ payer });
+
+    for (let i = 0; i < 50; i += 1) {
+      const res = response();
+      const next = await invoke(middleware, req, res);
+      expect(next).toHaveBeenCalledOnce();
+      expect(res.status).not.toHaveBeenCalled();
+    }
+
+    const limitedResponse = response();
+    const next = await invoke(middleware, req, limitedResponse);
+    expect(next).not.toHaveBeenCalled();
+    expect(limitedResponse.status).toHaveBeenCalledWith(429);
+    expect(limitedResponse.json).toHaveBeenCalledWith(
+      expect.objectContaining({ code: "RATE_LIMITED", retryAfter: expect.any(Number) }),
+    );
+    expect(limitedResponse.setHeader).toHaveBeenCalledWith("RateLimit-Limit", "50");
+    expect(limitedResponse.setHeader).toHaveBeenCalledWith("RateLimit-Remaining", "0");
+    expect(limitedResponse.setHeader).toHaveBeenCalledWith("Retry-After", expect.any(String));
+  });
+
+  it("keeps payer buckets independent and bypasses payer counting for anonymous requests", async () => {
+    const middleware = createPayerRateLimiter(new MemoryPayerRateLimitStore());
+    const payerA = "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF";
+    const payerB = "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWIG";
+
+    for (let i = 0; i < 50; i += 1) {
+      await invoke(middleware, request({ payer: payerA }), response());
+    }
+
+    const payerBResponse = response();
+    const payerBNext = await invoke(middleware, request({ payer: payerB }), payerBResponse);
+    expect(payerBNext).toHaveBeenCalledOnce();
+
+    const anonymousResponse = response();
+    const anonymousNext = await invoke(middleware, request(), anonymousResponse);
+    expect(anonymousNext).toHaveBeenCalledOnce();
+    expect(anonymousResponse.status).not.toHaveBeenCalled();
+  });
+
+  it("extracts payer, owner, and explicit header identities after JSON parsing", () => {
+    const payer = "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF";
+    expect(extractPayerAddress(request({ payer }))).toBe(payer);
+    expect(extractPayerAddress(request({ owner: payer }))).toBe(payer);
+    expect(extractPayerAddress(request({}, { "x-payer-address": payer }))).toBe(payer);
+    expect(extractPayerAddress(request({ payer: "   " }))).toBeNull();
+  });
+});