Merge ac937294d4 into 4b5514a259

CHANGELOG.md

@@ -6,6 +6,7 @@ Docs: https://docs.molt.bot
 Status: beta.
 
 ### Changes
+- Gateway: add configurable rate limiting with token bucket algorithm for abuse prevention. (#3951)
 - Rebrand: rename the npm package/CLI to `moltbot`, add a `moltbot` compatibility shim, and move extensions to the `@moltbot/*` scope.
 - Commands: group /help and /commands output with Telegram paging. (#2504) Thanks @hougangdev.
 - macOS: limit project-local `node_modules/.bin` PATH preference to debug builds (reduce PATH hijacking risk).

src/config/types.gateway.ts

@@ -191,6 +191,29 @@ export type GatewayHttpConfig = {
   endpoints?: GatewayHttpEndpointsConfig;
 };
 
+/**
+ * Rate limiting configuration for the gateway.
+ * Uses token bucket algorithm for smooth limiting with burst support.
+ */
+export type GatewayRateLimitConfig = {
+  /** Enable rate limiting (default: true). */
+  enabled?: boolean;
+  /** Max requests/min for unauthenticated clients (default: 60, 0 = unlimited). */
+  unauthenticated?: number;
+  /** Max requests/min for authenticated clients (default: 0 = unlimited). */
+  authenticated?: number;
+  /** Max messages/min per channel (default: 200, 0 = unlimited). */
+  channelMessages?: number;
+  /** Burst multiplier for token bucket (default: 2). */
+  burstMultiplier?: number;
+  /** Auth failures before exponential backoff starts (default: 5). */
+  authFailuresBeforeBackoff?: number;
+  /** Base delay in ms for auth backoff (default: 1000). */
+  authBackoffBaseMs?: number;
+  /** Max delay in ms for auth backoff (default: 60000). */
+  authBackoffMaxMs?: number;
+};
+
 export type GatewayNodesConfig = {
   /** Browser routing policy for node-hosted browser proxies. */
   browser?: {
@@ -233,6 +256,8 @@ export type GatewayConfig = {
   tls?: GatewayTlsConfig;
   http?: GatewayHttpConfig;
   nodes?: GatewayNodesConfig;
+  /** Rate limiting configuration for abuse prevention. */
+  rateLimit?: GatewayRateLimitConfig;
   /**
    * IPs of trusted reverse proxies (e.g. Traefik, nginx). When a connection
    * arrives from one of these IPs, the Gateway trusts `x-forwarded-for` (or

src/gateway/rate-limit.test.ts

@@ -0,0 +1,265 @@
+import { describe, expect, it, afterEach, vi } from "vitest";
+import {
+  GatewayRateLimiter,
+  resolveRateLimitConfig,
+  DEFAULT_RATE_LIMIT_CONFIG,
+} from "./rate-limit.js";
+
+describe("GatewayRateLimiter", () => {
+  let limiter: GatewayRateLimiter;
+
+  afterEach(() => {
+    limiter?.close();
+  });
+
+  describe("checkRequest", () => {
+    it("allows requests when disabled", () => {
+      limiter = new GatewayRateLimiter({ enabled: false });
+      const result = limiter.checkRequest("client-1", false);
+      expect(result.allowed).toBe(true);
+      expect(result.remaining).toBe(Infinity);
+    });
+
+    it("allows unlimited authenticated requests by default", () => {
+      limiter = new GatewayRateLimiter({ authenticated: 0 });
+      for (let i = 0; i < 100; i++) {
+        const result = limiter.checkRequest("client-1", true);
+        expect(result.allowed).toBe(true);
+      }
+    });
+
+    it("rate limits unauthenticated requests", () => {
+      limiter = new GatewayRateLimiter({
+        unauthenticated: 5,
+        burstMultiplier: 1, // No burst for simpler testing
+      });
+
+      // First 5 should be allowed (burst = 5 * 1)
+      for (let i = 0; i < 5; i++) {
+        const result = limiter.checkRequest("client-1", false);
+        expect(result.allowed).toBe(true);
+      }
+
+      // 6th should be rate limited
+      const result = limiter.checkRequest("client-1", false);
+      expect(result.allowed).toBe(false);
+      expect(result.reason).toBe("rate_limit");
+      expect(result.retryAfterMs).toBeGreaterThan(0);
+    });
+
+    it("allows burst traffic with burstMultiplier", () => {
+      limiter = new GatewayRateLimiter({
+        unauthenticated: 5,
+        burstMultiplier: 2, // Allow 10 requests in burst
+      });
+
+      // First 10 should be allowed (5 * 2 = 10)
+      for (let i = 0; i < 10; i++) {
+        const result = limiter.checkRequest("client-1", false);
+        expect(result.allowed).toBe(true);
+      }
+
+      // 11th should be rate limited
+      const result = limiter.checkRequest("client-1", false);
+      expect(result.allowed).toBe(false);
+    });
+
+    it("tracks separate buckets per client", () => {
+      limiter = new GatewayRateLimiter({
+        unauthenticated: 2,
+        burstMultiplier: 1,
+      });
+
+      // Exhaust client-1's quota
+      limiter.checkRequest("client-1", false);
+      limiter.checkRequest("client-1", false);
+      expect(limiter.checkRequest("client-1", false).allowed).toBe(false);
+
+      // client-2 should still have quota
+      expect(limiter.checkRequest("client-2", false).allowed).toBe(true);
+    });
+
+    it("refills tokens over time", async () => {
+      vi.useFakeTimers();
+
+      limiter = new GatewayRateLimiter({
+        unauthenticated: 60, // 1 per second
+        burstMultiplier: 1,
+      });
+
+      // Exhaust quota
+      for (let i = 0; i < 60; i++) {
+        limiter.checkRequest("client-1", false);
+      }
+      expect(limiter.checkRequest("client-1", false).allowed).toBe(false);
+
+      // Advance time by 2 seconds (should refill 2 tokens)
+      vi.advanceTimersByTime(2000);
+
+      expect(limiter.checkRequest("client-1", false).allowed).toBe(true);
+      expect(limiter.checkRequest("client-1", false).allowed).toBe(true);
+      expect(limiter.checkRequest("client-1", false).allowed).toBe(false);
+
+      vi.useRealTimers();
+    });
+  });
+
+  describe("checkChannelMessage", () => {
+    it("rate limits channel messages", () => {
+      limiter = new GatewayRateLimiter({
+        channelMessages: 3,
+        burstMultiplier: 1,
+      });
+
+      // First 3 should be allowed
+      for (let i = 0; i < 3; i++) {
+        expect(limiter.checkChannelMessage("telegram:123").allowed).toBe(true);
+      }
+
+      // 4th should be rate limited
+      expect(limiter.checkChannelMessage("telegram:123").allowed).toBe(false);
+    });
+
+    it("tracks separate channels independently", () => {
+      limiter = new GatewayRateLimiter({
+        channelMessages: 2,
+        burstMultiplier: 1,
+      });
+
+      // Exhaust telegram channel
+      limiter.checkChannelMessage("telegram:123");
+      limiter.checkChannelMessage("telegram:123");
+      expect(limiter.checkChannelMessage("telegram:123").allowed).toBe(false);
+
+      // Discord channel should still work
+      expect(limiter.checkChannelMessage("discord:456").allowed).toBe(true);
+    });
+
+    it("allows unlimited when channelMessages is 0", () => {
+      limiter = new GatewayRateLimiter({ channelMessages: 0 });
+      for (let i = 0; i < 1000; i++) {
+        expect(limiter.checkChannelMessage("telegram:123").allowed).toBe(true);
+      }
+    });
+  });
+
+  describe("auth failure backoff", () => {
+    it("does not apply backoff before threshold", () => {
+      limiter = new GatewayRateLimiter({
+        authFailuresBeforeBackoff: 5,
+      });
+
+      // Record 4 failures (below threshold)
+      for (let i = 0; i < 4; i++) {
+        limiter.recordAuthFailure("client-1");
+      }
+
+      // Should still be allowed
+      const result = limiter.checkRequest("client-1", false);
+      expect(result.allowed).toBe(true);
+    });
+
+    it("applies backoff after threshold", () => {
+      limiter = new GatewayRateLimiter({
+        authFailuresBeforeBackoff: 3,
+        authBackoffBaseMs: 1000,
+      });
+
+      // Record 3 failures (at threshold)
+      for (let i = 0; i < 3; i++) {
+        limiter.recordAuthFailure("client-1");
+      }
+
+      // Should be blocked
+      const result = limiter.checkRequest("client-1", false);
+      expect(result.allowed).toBe(false);
+      expect(result.reason).toBe("auth_backoff");
+    });
+
+    it("applies exponential backoff", () => {
+      limiter = new GatewayRateLimiter({
+        authFailuresBeforeBackoff: 1,
+        authBackoffBaseMs: 1000,
+        authBackoffMaxMs: 60000,
+      });
+
+      // First failure - 1000ms backoff
+      limiter.recordAuthFailure("client-1");
+      let state = limiter.getAuthBackoffState("client-1");
+      expect(state?.failures).toBe(1);
+
+      // Second failure - 2000ms backoff (1000 * 2^1)
+      limiter.recordAuthFailure("client-1");
+      state = limiter.getAuthBackoffState("client-1");
+      expect(state?.failures).toBe(2);
+
+      // Third failure - 4000ms backoff (1000 * 2^2)
+      limiter.recordAuthFailure("client-1");
+      state = limiter.getAuthBackoffState("client-1");
+      expect(state?.failures).toBe(3);
+    });
+
+    it("clears backoff after successful auth", () => {
+      limiter = new GatewayRateLimiter({
+        authFailuresBeforeBackoff: 1,
+      });
+
+      limiter.recordAuthFailure("client-1");
+      expect(limiter.getAuthBackoffState("client-1")).not.toBeNull();
+
+      limiter.clearAuthFailure("client-1");
+      expect(limiter.getAuthBackoffState("client-1")).toBeNull();
+    });
+
+    it("resets failure count after 10 minutes of inactivity", () => {
+      vi.useFakeTimers();
+
+      limiter = new GatewayRateLimiter({
+        authFailuresBeforeBackoff: 5,
+      });
+
+      // Record 4 failures
+      for (let i = 0; i < 4; i++) {
+        limiter.recordAuthFailure("client-1");
+      }
+
+      // Advance 11 minutes
+      vi.advanceTimersByTime(11 * 60 * 1000);
+
+      // Record 1 more failure - should be treated as first failure
+      limiter.recordAuthFailure("client-1");
+
+      // Should not be in backoff (only 1 failure after reset)
+      expect(limiter.getAuthBackoffState("client-1")).toBeNull();
+
+      vi.useRealTimers();
+    });
+  });
+
+  describe("updateConfig", () => {
+    it("updates configuration at runtime", () => {
+      limiter = new GatewayRateLimiter({ enabled: true });
+      expect(limiter.getConfig().enabled).toBe(true);
+
+      limiter.updateConfig({ enabled: false });
+      expect(limiter.getConfig().enabled).toBe(false);
+    });
+  });
+});
+
+describe("resolveRateLimitConfig", () => {
+  it("returns defaults when no config provided", () => {
+    const config = resolveRateLimitConfig();
+    expect(config).toEqual(DEFAULT_RATE_LIMIT_CONFIG);
+  });
+
+  it("merges partial config with defaults", () => {
+    const config = resolveRateLimitConfig({
+      enabled: false,
+      unauthenticated: 100,
+    });
+    expect(config.enabled).toBe(false);
+    expect(config.unauthenticated).toBe(100);
+    expect(config.authenticated).toBe(DEFAULT_RATE_LIMIT_CONFIG.authenticated);
+  });
+});

src/gateway/rate-limit.ts

@@ -0,0 +1,345 @@
+/**
+ * Gateway Rate Limiting
+ *
+ * Implements configurable rate limiting for the gateway server:
+ * - Per-client request rate limiting (token bucket algorithm)
+ * - Exponential backoff after repeated auth failures
+ * - Separate limits for authenticated vs unauthenticated requests
+ * - Channel message rate limiting
+ *
+ * Defaults are generous for power users while preventing abuse:
+ * - Unauthenticated: 60 req/min (prevents brute-force)
+ * - Authenticated: unlimited by default
+ * - Channel messages: 200/min per channel
+ */
+
+import type { GatewayRateLimitConfig } from "../config/types.gateway.js";
+import { createSubsystemLogger } from "../logging/subsystem.js";
+
+const log = createSubsystemLogger("rate-limit");
+
+export type RateLimitResult = {
+  allowed: boolean;
+  remaining: number;
+  resetMs: number;
+  retryAfterMs?: number;
+  reason?: "rate_limit" | "auth_backoff";
+};
+
+export type ResolvedRateLimitConfig = Required<GatewayRateLimitConfig>;
+
+/** Default rate limit configuration (generous for power users) */
+export const DEFAULT_RATE_LIMIT_CONFIG: ResolvedRateLimitConfig = {
+  enabled: true,
+  unauthenticated: 60, // 60 req/min for unauthenticated (prevents brute-force)
+  authenticated: 0, // 0 = unlimited for authenticated users
+  channelMessages: 200, // 200 msg/min per channel
+  burstMultiplier: 2, // Allow 2x burst for short spikes
+  authFailuresBeforeBackoff: 5, // Start backoff after 5 failures
+  authBackoffBaseMs: 1000, // 1 second base delay
+  authBackoffMaxMs: 60000, // Max 1 minute delay
+};
+
+/** Internal state for a rate limit bucket */
+type RateBucket = {
+  tokens: number;
+  lastRefillMs: number;
+  maxTokens: number;
+  refillRatePerMs: number;
+};
+
+/** Auth failure tracking per client */
+type AuthFailureState = {
+  failures: number;
+  lastFailureMs: number;
+  backoffUntilMs: number;
+};
+
+/**
+ * Gateway rate limiter with per-client tracking.
+ * Uses token bucket algorithm for smooth rate limiting with burst support.
+ */
+export class GatewayRateLimiter {
+  private config: ResolvedRateLimitConfig;
+  private buckets: Map<string, RateBucket> = new Map();
+  private authFailures: Map<string, AuthFailureState> = new Map();
+  private cleanupInterval: NodeJS.Timeout | null = null;
+
+  constructor(config?: Partial<GatewayRateLimitConfig>) {
+    this.config = resolveRateLimitConfig(config);
+
+    // Periodic cleanup of stale entries (every 5 minutes)
+    this.cleanupInterval = setInterval(() => this.cleanup(), 5 * 60 * 1000);
+  }
+
+  /**
+   * Check if a request should be allowed based on rate limits.
+   *
+   * @param clientId - Unique client identifier (IP address, session ID, etc.)
+   * @param authenticated - Whether the client is authenticated
+   * @returns Rate limit result with allowed status and metadata
+   */
+  checkRequest(clientId: string, authenticated: boolean): RateLimitResult {
+    if (!this.config.enabled) {
+      return { allowed: true, remaining: Infinity, resetMs: 0 };
+    }
+
+    const now = Date.now();
+
+    // Check auth backoff first (applies to all requests from blocked clients)
+    const authState = this.authFailures.get(clientId);
+    if (authState && authState.backoffUntilMs > now) {
+      const retryAfterMs = authState.backoffUntilMs - now;
+      log.debug("Request blocked by auth backoff", {
+        clientId: clientId.slice(0, 16),
+        retryAfterMs,
+      });
+      return {
+        allowed: false,
+        remaining: 0,
+        resetMs: retryAfterMs,
+        retryAfterMs,
+        reason: "auth_backoff",
+      };
+    }
+
+    // Determine rate limit based on auth status
+    const ratePerMinute = authenticated ? this.config.authenticated : this.config.unauthenticated;
+
+    // 0 = unlimited
+    if (ratePerMinute === 0) {
+      return { allowed: true, remaining: Infinity, resetMs: 0 };
+    }
+
+    const bucketKey = `${authenticated ? "auth" : "unauth"}:${clientId}`;
+    const bucket = this.getOrCreateBucket(bucketKey, ratePerMinute);
+
+    // Refill tokens based on elapsed time
+    this.refillBucket(bucket, now);
+
+    if (bucket.tokens >= 1) {
+      bucket.tokens -= 1;
+      return {
+        allowed: true,
+        remaining: Math.floor(bucket.tokens),
+        resetMs: Math.ceil((bucket.maxTokens - bucket.tokens) / bucket.refillRatePerMs),
+      };
+    }
+
+    // Rate limited
+    const resetMs = Math.ceil((1 - bucket.tokens) / bucket.refillRatePerMs);
+    log.debug("Request rate limited", {
+      clientId: clientId.slice(0, 16),
+      authenticated,
+      resetMs,
+    });
+
+    return {
+      allowed: false,
+      remaining: 0,
+      resetMs,
+      retryAfterMs: resetMs,
+      reason: "rate_limit",
+    };
+  }
+
+  /**
+   * Check if a channel message should be allowed.
+   *
+   * @param channelKey - Unique channel identifier (e.g., "telegram:123")
+   * @returns Rate limit result
+   */
+  checkChannelMessage(channelKey: string): RateLimitResult {
+    if (!this.config.enabled || this.config.channelMessages === 0) {
+      return { allowed: true, remaining: Infinity, resetMs: 0 };
+    }
+
+    const now = Date.now();
+    const bucketKey = `channel:${channelKey}`;
+    const bucket = this.getOrCreateBucket(bucketKey, this.config.channelMessages);
+
+    this.refillBucket(bucket, now);
+
+    if (bucket.tokens >= 1) {
+      bucket.tokens -= 1;
+      return {
+        allowed: true,
+        remaining: Math.floor(bucket.tokens),
+        resetMs: Math.ceil((bucket.maxTokens - bucket.tokens) / bucket.refillRatePerMs),
+      };
+    }
+
+    const resetMs = Math.ceil((1 - bucket.tokens) / bucket.refillRatePerMs);
+    log.warn("Channel message rate limited", { channelKey, resetMs });
+
+    return {
+      allowed: false,
+      remaining: 0,
+      resetMs,
+      retryAfterMs: resetMs,
+      reason: "rate_limit",
+    };
+  }
+
+  /**
+   * Record an authentication failure for exponential backoff.
+   *
+   * @param clientId - Unique client identifier
+   */
+  recordAuthFailure(clientId: string): void {
+    const now = Date.now();
+    const state = this.authFailures.get(clientId) ?? {
+      failures: 0,
+      lastFailureMs: 0,
+      backoffUntilMs: 0,
+    };
+
+    // Reset counter if last failure was more than 10 minutes ago
+    if (now - state.lastFailureMs > 10 * 60 * 1000) {
+      state.failures = 0;
+    }
+
+    state.failures += 1;
+    state.lastFailureMs = now;
+
+    // Apply exponential backoff after threshold
+    if (state.failures >= this.config.authFailuresBeforeBackoff) {
+      const exponent = state.failures - this.config.authFailuresBeforeBackoff;
+      const backoffMs = Math.min(
+        this.config.authBackoffBaseMs * Math.pow(2, exponent),
+        this.config.authBackoffMaxMs,
+      );
+      state.backoffUntilMs = now + backoffMs;
+
+      log.warn("Auth failure backoff applied", {
+        clientId: clientId.slice(0, 16),
+        failures: state.failures,
+        backoffMs,
+      });
+    }
+
+    this.authFailures.set(clientId, state);
+  }
+
+  /**
+   * Clear auth failure state after successful authentication.
+   *
+   * @param clientId - Unique client identifier
+   */
+  clearAuthFailure(clientId: string): void {
+    this.authFailures.delete(clientId);
+  }
+
+  /**
+   * Get current auth backoff state for a client.
+   *
+   * @param clientId - Unique client identifier
+   * @returns Backoff state or null if not in backoff
+   */
+  getAuthBackoffState(clientId: string): { failures: number; backoffUntilMs: number } | null {
+    const state = this.authFailures.get(clientId);
+    if (!state || state.backoffUntilMs <= Date.now()) {
+      return null;
+    }
+    return { failures: state.failures, backoffUntilMs: state.backoffUntilMs };
+  }
+
+  /**
+   * Update configuration at runtime.
+   */
+  updateConfig(config: Partial<GatewayRateLimitConfig>): void {
+    this.config = resolveRateLimitConfig(config);
+    log.info("Rate limit config updated", { enabled: this.config.enabled });
+  }
+
+  /**
+   * Get current configuration.
+   */
+  getConfig(): ResolvedRateLimitConfig {
+    return { ...this.config };
+  }
+
+  /**
+   * Clean up resources.
+   */
+  close(): void {
+    if (this.cleanupInterval) {
+      clearInterval(this.cleanupInterval);
+      this.cleanupInterval = null;
+    }
+    this.buckets.clear();
+    this.authFailures.clear();
+  }
+
+  private getOrCreateBucket(key: string, ratePerMinute: number): RateBucket {
+    let bucket = this.buckets.get(key);
+    if (!bucket) {
+      const maxTokens = ratePerMinute * this.config.burstMultiplier;
+      bucket = {
+        tokens: maxTokens, // Start full
+        lastRefillMs: Date.now(),
+        maxTokens,
+        refillRatePerMs: ratePerMinute / (60 * 1000), // tokens per ms
+      };
+      this.buckets.set(key, bucket);
+    }
+    return bucket;
+  }
+
+  private refillBucket(bucket: RateBucket, now: number): void {
+    const elapsedMs = now - bucket.lastRefillMs;
+    if (elapsedMs <= 0) return;
+
+    const tokensToAdd = elapsedMs * bucket.refillRatePerMs;
+    bucket.tokens = Math.min(bucket.tokens + tokensToAdd, bucket.maxTokens);
+    bucket.lastRefillMs = now;
+  }
+
+  private cleanup(): void {
+    const now = Date.now();
+    const staleThresholdMs = 10 * 60 * 1000; // 10 minutes
+
+    // Clean up stale buckets
+    for (const [key, bucket] of this.buckets) {
+      if (now - bucket.lastRefillMs > staleThresholdMs) {
+        this.buckets.delete(key);
+      }
+    }
+
+    // Clean up stale auth failures
+    for (const [key, state] of this.authFailures) {
+      if (now - state.lastFailureMs > staleThresholdMs && state.backoffUntilMs < now) {
+        this.authFailures.delete(key);
+      }
+    }
+  }
+}
+
+/**
+ * Resolve rate limit config with defaults.
+ */
+export function resolveRateLimitConfig(
+  config?: Partial<GatewayRateLimitConfig>,
+): ResolvedRateLimitConfig {
+  return {
+    enabled: config?.enabled ?? DEFAULT_RATE_LIMIT_CONFIG.enabled,
+    unauthenticated: config?.unauthenticated ?? DEFAULT_RATE_LIMIT_CONFIG.unauthenticated,
+    authenticated: config?.authenticated ?? DEFAULT_RATE_LIMIT_CONFIG.authenticated,
+    channelMessages: config?.channelMessages ?? DEFAULT_RATE_LIMIT_CONFIG.channelMessages,
+    burstMultiplier: config?.burstMultiplier ?? DEFAULT_RATE_LIMIT_CONFIG.burstMultiplier,
+    authFailuresBeforeBackoff:
+      config?.authFailuresBeforeBackoff ?? DEFAULT_RATE_LIMIT_CONFIG.authFailuresBeforeBackoff,
+    authBackoffBaseMs: config?.authBackoffBaseMs ?? DEFAULT_RATE_LIMIT_CONFIG.authBackoffBaseMs,
+    authBackoffMaxMs: config?.authBackoffMaxMs ?? DEFAULT_RATE_LIMIT_CONFIG.authBackoffMaxMs,
+  };
+}
+
+/**
+ * Create a global rate limiter instance.
+ * This is typically called once during gateway startup.
+ */
+export function createGatewayRateLimiter(
+  config?: Partial<GatewayRateLimitConfig>,
+): GatewayRateLimiter {
+  return new GatewayRateLimiter(config);
+}