diff --git a/.env.example b/.env.example index a0a5371..58c2f66 100644 --- a/.env.example +++ b/.env.example @@ -48,6 +48,14 @@ RATE_LIMIT_ENTERPRISE_PER_MINUTE=10000 RATE_LIMIT_WINDOW_MS=60000 RATE_LIMIT_BURST_MULTIPLIER=1.2 +# Distributed rate limiting (Redis-backed) — complements the per-tier limits above +# Default strategy: "token-bucket" or "sliding-window" +RATE_LIMIT_DEFAULT_STRATEGY=token-bucket +# Redis key prefix for rate-limit entries +RATE_LIMIT_REDIS_KEY_PREFIX=alian:rl: +# When false, requests are rejected (not in-memory fallback) if Redis is down +RATE_LIMIT_FALLBACK_TO_MEMORY=true + # ── Logging ──────────────────────────────────────────────────────────────── # Minimum log level emitted to all transports. # Values: verbose | debug | info | warn | error | fatal diff --git a/.env.production.example b/.env.production.example index 439956c..29976b0 100644 --- a/.env.production.example +++ b/.env.production.example @@ -69,6 +69,11 @@ RATE_LIMIT_ENTERPRISE_PER_MINUTE=10000 RATE_LIMIT_WINDOW_MS=60000 RATE_LIMIT_BURST_MULTIPLIER=1.2 +# Distributed rate limiting (Redis-backed) +RATE_LIMIT_DEFAULT_STRATEGY=token-bucket +RATE_LIMIT_REDIS_KEY_PREFIX=alian:rl: +RATE_LIMIT_FALLBACK_TO_MEMORY=false + # Security Headers HSTS_MAX_AGE=31536000 diff --git a/monitoring/grafana/dashboards/rate-limiting-dashboard.json b/monitoring/grafana/dashboards/rate-limiting-dashboard.json new file mode 100644 index 0000000..443a0c9 --- /dev/null +++ b/monitoring/grafana/dashboards/rate-limiting-dashboard.json @@ -0,0 +1,513 @@ +{ + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": { + "type": "grafana", + "uid": "-- Grafana --" + }, + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, + "description": "Rate Limiting & Abuse Protection dashboard. Shows allowed/denied request rates, storage health, per-tier and per-strategy breakdown, and the denial rate for Grant reviewers.", + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "id": null, + "links": [ + { + "asDropdown": false, + "icon": "external link", + "includeVars": false, + "keepTime": false, + "tags": ["alian-structure"], + "targetBlank": true, + "title": "Alian Structure dashboards", + "tooltip": "Other Alian Structure dashboards", + "type": "dashboards" + } + ], + "liveNow": false, + "panels": [ + { + "type": "row", + "id": 200, + "title": "Rate Limiting Overview", + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 0 }, + "panels": [] + }, + { + "type": "stat", + "id": 1, + "title": "Requests Allowed (5m rate)", + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "gridPos": { "h": 4, "w": 6, "x": 0, "y": 1 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [{ "color": "green", "value": null }] + }, + "unit": "reqps" + }, + "overrides": [] + }, + "options": { + "colorMode": "value", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "sum(rate(alian_structure_rate_limit_allowed_total{job=\"$job\"}[5m]))", + "legendFormat": "allowed", + "refId": "A" + } + ] + }, + { + "type": "stat", + "id": 2, + "title": "Requests Denied (5m rate)", + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "gridPos": { "h": 4, "w": 6, "x": 6, "y": 1 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 1 }, + { "color": "red", "value": 50 } + ] + }, + "unit": "reqps" + }, + "overrides": [] + }, + "options": { + "colorMode": "value", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "sum(rate(alian_structure_rate_limit_denied_total{job=\"$job\"}[5m]))", + "legendFormat": "denied", + "refId": "A" + } + ] + }, + { + "type": "stat", + "id": 3, + "title": "Denial Rate (%)", + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "gridPos": { "h": 4, "w": 6, "x": 12, "y": 1 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 5 }, + { "color": "red", "value": 20 } + ] + }, + "unit": "percent" + }, + "overrides": [] + }, + "options": { + "colorMode": "value", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "100 * sum(rate(alian_structure_rate_limit_denied_total{job=\"$job\"}[5m])) / clamp_min(sum(rate(alian_structure_rate_limit_allowed_total{job=\"$job\"}[5m]) + rate(alian_structure_rate_limit_denied_total{job=\"$job\"}[5m])), 0.001)", + "legendFormat": "denial rate", + "refId": "A" + } + ] + }, + { + "type": "stat", + "id": 4, + "title": "Rate-Limit Storage", + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "gridPos": { "h": 4, "w": 6, "x": 18, "y": 1 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "red", "value": null }, + { "color": "green", "value": 1 } + ] + }, + "unit": "short" + }, + "overrides": [] + }, + "options": { + "colorMode": "value", + "graphMode": "none", + "justifyMode": "auto", + "orientation": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "textMode": "auto" + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "alian_structure_rate_limit_storage_health{job=\"$job\", storage=\"redis\"}", + "legendFormat": "{{storage}}", + "refId": "A" + } + ] + }, + { + "type": "row", + "id": 201, + "title": "Rate Limiting Trends", + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 5 }, + "panels": [] + }, + { + "type": "timeseries", + "id": 5, + "title": "Allowed vs Denied Requests Over Time", + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "gridPos": { "h": 8, "w": 24, "x": 0, "y": 6 }, + "fieldConfig": { + "defaults": { + "custom": { + "drawStyle": "line", + "fillOpacity": 10, + "lineWidth": 2 + }, + "color": { "mode": "palette-classic" }, + "unit": "reqps" + }, + "overrides": [ + { + "matcher": { "id": "byName", "options": "Denied" }, + "properties": [{ "id": "color", "value": { "mode": "fixed", "fixedColor": "red" } }] + }, + { + "matcher": { "id": "byName", "options": "Allowed" }, + "properties": [{ "id": "color", "value": { "mode": "fixed", "fixedColor": "green" } }] + } + ] + }, + "options": { + "legend": { "displayMode": "list", "placement": "bottom", "showLegend": true }, + "tooltip": { "mode": "multi", "sort": "desc" } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "sum by (tier) (rate(alian_structure_rate_limit_allowed_total{job=\"$job\"}[5m]))", + "legendFormat": "Allowed ({{tier}})", + "refId": "A" + }, + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "sum by (tier) (rate(alian_structure_rate_limit_denied_total{job=\"$job\"}[5m]))", + "legendFormat": "Denied ({{tier}})", + "refId": "B" + } + ] + }, + { + "type": "row", + "id": 202, + "title": "Per-Tier & Per-Strategy Breakdown", + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 14 }, + "panels": [] + }, + { + "type": "table", + "id": 6, + "title": "Top Rate-Limited Keys by Denial Count (1h)", + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "gridPos": { "h": 10, "w": 24, "x": 0, "y": 15 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "green", "value": null }, + { "color": "yellow", "value": 10 }, + { "color": "red", "value": 100 } + ] + } + }, + "overrides": [] + }, + "options": { + "cellHeight": "sm", + "footer": { "fields": "", "reducer": ["sum"], "show": false }, + "showHeader": true + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "sum by (key, tier, strategy) (rate(alian_structure_rate_limit_denied_total{job=\"$job\"}[1h]))", + "format": "table", + "instant": true, + "legendFormat": "{{key}}", + "refId": "A" + } + ] + }, + { + "type": "piechart", + "id": 7, + "title": "Allowed Requests by Tier (1h)", + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 25 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }] } + }, + "overrides": [] + }, + "options": { + "displayMode": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "legend": { "displayMode": "list", "placement": "bottom", "showLegend": true }, + "tooltip": { "mode": "single", "sort": "none" } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "sum by (tier) (rate(alian_structure_rate_limit_allowed_total{job=\"$job\"}[1h]))", + "legendFormat": "{{tier}}", + "refId": "A" + } + ] + }, + { + "type": "piechart", + "id": 8, + "title": "Denied Requests by Tier (1h)", + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 25 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "palette-classic" }, + "mappings": [], + "thresholds": { "mode": "absolute", "steps": [{ "color": "red", "value": null }] } + }, + "overrides": [] + }, + "options": { + "displayMode": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + }, + "legend": { "displayMode": "list", "placement": "bottom", "showLegend": true }, + "tooltip": { "mode": "single", "sort": "none" } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "sum by (tier) (rate(alian_structure_rate_limit_denied_total{job=\"$job\"}[1h]))", + "legendFormat": "{{tier}}", + "refId": "A" + } + ] + }, + { + "type": "row", + "id": 203, + "title": "Errors & Storage", + "collapsed": false, + "gridPos": { "h": 1, "w": 24, "x": 0, "y": 33 }, + "panels": [] + }, + { + "type": "timeseries", + "id": 9, + "title": "Rate Limit Storage Errors", + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 34 }, + "fieldConfig": { + "defaults": { + "custom": { "drawStyle": "line", "fillOpacity": 10, "lineWidth": 2 }, + "color": { "mode": "palette-classic" }, + "unit": "short" + }, + "overrides": [ + { + "matcher": { "id": "byName", "options": "Redis Error" }, + "properties": [{ "id": "color", "value": { "mode": "fixed", "fixedColor": "red" } }] + } + ] + }, + "options": { + "legend": { "displayMode": "list", "placement": "bottom", "showLegend": true }, + "tooltip": { "mode": "multi" } + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "sum by (operation, storage) (rate(alian_structure_rate_limit_errors_total{job=\"$job\"}[5m]))", + "legendFormat": "{{operation}} ({{storage}})", + "refId": "A" + } + ] + }, + { + "type": "gauge", + "id": 10, + "title": "Storage Health", + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "gridPos": { "h": 6, "w": 12, "x": 12, "y": 34 }, + "fieldConfig": { + "defaults": { + "color": { "mode": "thresholds" }, + "thresholds": { + "mode": "absolute", + "steps": [ + { "color": "red", "value": null }, + { "color": "green", "value": 1 } + ] + }, + "min": 0, + "max": 1, + "unit": "short" + }, + "overrides": [] + }, + "options": { + "orientation": "auto", + "showThresholdLabels": false, + "showThresholdMarkers": true + }, + "targets": [ + { + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "expr": "alian_structure_rate_limit_storage_health{job=\"$job\", storage=\"redis\"}", + "legendFormat": "Redis Health", + "refId": "A" + } + ] + } + ], + "refresh": "10s", + "schemaVersion": 39, + "tags": ["alian-structure", "rate-limiting", "abuse-protection", "api-gateway"], + "templating": { + "list": [ + { + "current": { + "selected": false, + "text": "Prometheus", + "value": "Prometheus" + }, + "hide": 0, + "includeAll": false, + "label": "Datasource", + "multi": false, + "name": "DS_PROMETHEUS", + "options": [], + "query": "prometheus", + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "type": "datasource" + }, + { + "current": { + "selected": false, + "text": "All", + "value": "$__all" + }, + "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" }, + "definition": "label_values(alian_structure_rate_limit_denied_total{job=\"$job\"}, tier)", + "hide": 0, + "includeAll": true, + "label": "Rate Limit Tier", + "multi": true, + "name": "tier", + "options": [], + "query": { + "query": "label_values(alian_structure_rate_limit_denied_total{job=\"$job\"}, tier)", + "refId": "PrometheusVariableQueryEditor-VariableQuery" + }, + "refresh": 1, + "regex": "", + "skipUrlSync": false, + "sort": 1, + "type": "query" + } + ] + }, + "time": { "from": "now-1h", "to": "now" }, + "timepicker": { + "refresh_intervals": ["5s", "10s", "30s", "1m", "5m", "15m", "30m", "1h"] + }, + "timezone": "", + "title": "Alian Structure - Rate Limiting & Abuse Protection", + "uid": "alian-structure-rate-limiting", + "version": 1, + "weekStart": "" +} diff --git a/src/app.module.ts b/src/app.module.ts index 7950ee3..cfc48d9 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -53,6 +53,7 @@ import { LoggerModule } from "./logging/logger.module"; // Modules – cache import { CacheModule } from "./common/cache/cache.module"; import { BillingModule } from "./billing/billing.module"; +import { RateLimitingModule } from "./rate-limiting/rate-limiting.module"; import { ReconciliationModule } from "./reconciliation/reconciliation.module"; // Auth entities @@ -119,7 +120,7 @@ import { FileUploadModule } from "./infrastructure/file-upload/file-upload.modul // Guards import { APP_FILTER } from "@nestjs/core"; -import { QuotaGuard } from "./common/guard/quota.guard"; +import { DistributedRateLimitGuard } from "./rate-limiting/rate-limiting.guard"; import { RolesGuard } from "./common/guard/roles.guard"; import { KycGuard } from "./common/guard/kyc.guard"; import { StrategyAuthGuard } from "./core/auth/guards/strategy-auth.guard"; @@ -253,6 +254,7 @@ import { TenantModuleState } from "./modules/registry/entities/tenant-module-sta FileUploadModule, ModuleRegistryModule, CacheModule, + RateLimitingModule.forRoot(), LoggerModule.forRootAsync({ inject: [ConfigService], useFactory: (cfg: ConfigService) => ({ @@ -283,7 +285,7 @@ import { TenantModuleState } from "./modules/registry/entities/tenant-module-sta }, { provide: APP_GUARD, - useClass: QuotaGuard, + useClass: DistributedRateLimitGuard, }, { provide: APP_GUARD, diff --git a/src/common/decorators/rate-limit.decorator.ts b/src/common/decorators/rate-limit.decorator.ts index 6f12abf..ade983b 100644 --- a/src/common/decorators/rate-limit.decorator.ts +++ b/src/common/decorators/rate-limit.decorator.ts @@ -1,4 +1,5 @@ import { SetMetadata, applyDecorators } from "@nestjs/common"; +import { RateLimitStrategy } from "src/rate-limiting/interfaces"; /** * Apply a named throttle configuration to a controller or handler. @@ -17,6 +18,13 @@ export interface RateLimitOptions { limit?: number; windowMs?: number; burst?: number; + /** Rate-limiting strategy: "token-bucket" (default) or "sliding-window". */ + strategy?: RateLimitStrategy; + /** + * Custom key prefix for per-key limiting. When omitted, the key is derived + * automatically from the request tracker + scope + tier. + */ + key?: string; } const TIER_CONFIG: Record = { @@ -34,6 +42,7 @@ export function SensitiveRateLimit(tier: SensitiveTier = "default") { limit, windowMs: ttl, burst: limit, + strategy: RateLimitStrategy.TokenBucket, }), ); } diff --git a/src/config/env.validation.ts b/src/config/env.validation.ts index 83e8ab2..9e2666d 100644 --- a/src/config/env.validation.ts +++ b/src/config/env.validation.ts @@ -537,4 +537,28 @@ export class EnvironmentVariables { @IsOptional() @IsString() REDIS_PASSWORD?: string; + + // Rate Limiting & Abuse Protection (distributed, Redis-backed) + + /** Redis key prefix for rate-limit entries. Default: "alian:rl:". */ + @IsOptional() + @IsString() + RATE_LIMIT_REDIS_KEY_PREFIX?: string = "alian:rl:"; + + /** + * Default rate-limiting strategy: "token-bucket" (default) or + * "sliding-window". + */ + @IsOptional() + @IsString() + RATE_LIMIT_DEFAULT_STRATEGY?: string = "token-bucket"; + + /** + * When "false", the rate limiter will reject requests (rather than fall back + * to in-memory) if Redis is unavailable. Default: true. + */ + @IsOptional() + @IsBoolean() + @Transform(({ value }) => value !== "false") + RATE_LIMIT_FALLBACK_TO_MEMORY?: boolean = true; } diff --git a/src/rate-limiting/dto/rate-limit-dto.ts b/src/rate-limiting/dto/rate-limit-dto.ts new file mode 100644 index 0000000..8075736 --- /dev/null +++ b/src/rate-limiting/dto/rate-limit-dto.ts @@ -0,0 +1,60 @@ +import { IsEnum, IsInt, IsOptional, IsString, Max, Min } from "class-validator"; +import { RateLimitStrategy } from "../interfaces"; + +export class SetRateLimitDto { + @IsString() + key: string; + + @IsEnum(RateLimitStrategy) + strategy: RateLimitStrategy; + + @IsInt() + @Min(1) + limit: number; + + @IsInt() + @Min(1) + windowMs: number; + + @IsOptional() + @IsInt() + @Min(1) + burst?: number; + + @IsOptional() + @IsString() + scope?: string; + + @IsOptional() + @IsString() + tier?: string; +} + +export class RateLimitResetDto { + @IsString() + key: string; + + @IsOptional() + @IsString() + scope?: string; +} + +export class RateLimitStatusDto { + @IsOptional() + @IsString() + tracker?: string; + + @IsOptional() + @IsString() + scope?: string; + + @IsOptional() + @IsString() + tier?: string; + + @IsOptional() + @IsInt() + @Min(1) + @Max(500) + limit?: number = 100; +} diff --git a/src/rate-limiting/index.ts b/src/rate-limiting/index.ts new file mode 100644 index 0000000..1d670fd --- /dev/null +++ b/src/rate-limiting/index.ts @@ -0,0 +1,19 @@ +export { RateLimitingModule } from "./rate-limiting.module"; +export { RateLimiterService } from "./rate-limiter.service"; +export type { RateLimitConfig } from "./rate-limiter.service"; +export { DistributedRateLimitGuard } from "./rate-limiting.guard"; +export { RateLimitingController } from "./rate-limiting.controller"; +export { + RateLimitStrategy, + RateLimitPolicy, + RateLimitDecision, + RateLimitEntry, + RateLimitState, + RateLimitStorage, +} from "./interfaces"; +export type { + RateLimitPolicy as RateLimitPolicyType, + RateLimitDecision as RateLimitDecisionType, + RateLimitEntry as RateLimitEntryType, + RateLimitState as RateLimitStateType, +} from "./interfaces"; diff --git a/src/rate-limiting/interfaces.ts b/src/rate-limiting/interfaces.ts new file mode 100644 index 0000000..0f8b4fe --- /dev/null +++ b/src/rate-limiting/interfaces.ts @@ -0,0 +1,59 @@ +/** + * Strategies supported by the distributed rate limiter. + */ +export enum RateLimitStrategy { + TokenBucket = "token-bucket", + SlidingWindow = "sliding-window", +} + +/** + * Resolved rate-limit policy for a single request. + */ +export interface RateLimitPolicy { + limit: number; + windowMs: number; + burst: number; + strategy: RateLimitStrategy; +} + +/** + * Decision returned by the rate limiter after consuming a token. + */ +export interface RateLimitDecision { + allowed: boolean; + remaining: number; + resetAt: number; + retryAfterMs?: number; +} + +/** + * Resolved state for a single rate-limit check. + */ +export interface RateLimitState { + key: string; + policy: RateLimitPolicy; + decision: RateLimitDecision; + tracker: string; + scope: string; + tier: string; +} + +/** + * Aggregated stats for a single rate-limit key. + */ +export interface RateLimitEntry { + key: string; + tracker: string; + scope: string; + tier: string; + strategy: RateLimitStrategy; + limit: number; + windowMs: number; + remaining: number; + resetAt: number; +} + +/** + * Storage backend in use. + */ +export type RateLimitStorage = "redis" | "memory"; diff --git a/src/rate-limiting/rate-limiter.service.spec.ts b/src/rate-limiting/rate-limiter.service.spec.ts new file mode 100644 index 0000000..264edfc --- /dev/null +++ b/src/rate-limiting/rate-limiter.service.spec.ts @@ -0,0 +1,404 @@ +import { RateLimiterService, RateLimitConfig } from "./rate-limiter.service"; +import { RateLimitStrategy } from "./interfaces"; + +const DEFAULT_CONFIG: RateLimitConfig = { + keyPrefix: "alian:rl:", + defaultStrategy: RateLimitStrategy.TokenBucket, + enableFallback: true, +}; + +function makeMockRedis() { + const calls: { method: string; args: unknown[] }[] = []; + const mockRedis: any = { + eval: jest.fn(async () => { + calls.push({ method: "eval", args: [] }); + return [1, 5, 0]; + }), + ping: jest.fn(async () => "PONG"), + hgetall: jest.fn(async () => ({ tokens: "5", ts: "1000" })), + zcount: jest.fn(async () => 2), + del: jest.fn(async () => 1), + psetex: jest.fn(async () => "OK"), + expire: jest.fn(async () => 1), + }; + mockRedis._calls = calls; + return mockRedis; +} + +describe("RateLimiterService", () => { + describe("in-memory mode (no Redis)", () => { + let service: RateLimiterService; + + beforeEach(() => { + service = new RateLimiterService(null, DEFAULT_CONFIG); + }); + + describe("token bucket strategy", () => { + it("allows requests within the token budget", async () => { + const policy = { + limit: 5, + windowMs: 60_000, + burst: 5, + strategy: RateLimitStrategy.TokenBucket, + }; + + for (let i = 0; i < 5; i++) { + const decision = await service.consume( + "user:123:global:free", + policy, + "user:123", + "global", + "free", + ); + expect(decision.allowed).toBe(true); + } + }); + + it("denies requests once tokens are exhausted", async () => { + const policy = { + limit: 2, + windowMs: 60_000, + burst: 2, + strategy: RateLimitStrategy.TokenBucket, + }; + + await service.consume( + "key:global:free", + policy, + "key", + "global", + "free", + ); + await service.consume( + "key:global:free", + policy, + "key", + "global", + "free", + ); + + const decision = await service.consume( + "key:global:free", + policy, + "key", + "global", + "free", + ); + expect(decision.allowed).toBe(false); + expect(decision.remaining).toBe(0); + expect(decision.retryAfterMs).toBeGreaterThan(0); + }); + + it("sets retryAfterMs on denial", async () => { + const policy = { + limit: 1, + windowMs: 60_000, + burst: 1, + strategy: RateLimitStrategy.TokenBucket, + }; + + await service.consume("k:g:f", policy, "k", "g", "f"); + const decision = await service.consume("k:g:f", policy, "k", "g", "f"); + + expect(decision.allowed).toBe(false); + expect(decision.retryAfterMs).toBeGreaterThan(0); + }); + + it("reports positive remaining on allowed requests", async () => { + const policy = { + limit: 10, + windowMs: 60_000, + burst: 10, + strategy: RateLimitStrategy.TokenBucket, + }; + + const decision = await service.consume("k:g:f", policy, "k", "g", "f"); + expect(decision.allowed).toBe(true); + expect(decision.remaining).toBe(9); + }); + }); + + describe("sliding window strategy", () => { + it("allows requests within the limit", async () => { + const policy = { + limit: 3, + windowMs: 60_000, + burst: 3, + strategy: RateLimitStrategy.SlidingWindow, + }; + + for (let i = 0; i < 3; i++) { + const decision = await service.consume( + "sw:key:global:free", + policy, + "sw:key", + "global", + "free", + ); + expect(decision.allowed).toBe(true); + } + }); + + it("denies requests once the window limit is reached", async () => { + const policy = { + limit: 2, + windowMs: 60_000, + burst: 2, + strategy: RateLimitStrategy.SlidingWindow, + }; + + await service.consume("sw:k:g:f", policy, "sw:k", "g", "f"); + await service.consume("sw:k:g:f", policy, "sw:k", "g", "f"); + + const decision = await service.consume( + "sw:k:g:f", + policy, + "sw:k", + "g", + "f", + ); + expect(decision.allowed).toBe(false); + expect(decision.remaining).toBe(0); + expect(decision.retryAfterMs).toBeGreaterThan(0); + }); + + it("allows a new request after the window expires (memory)", async () => { + const policy = { + limit: 2, + windowMs: 60_000, + burst: 2, + strategy: RateLimitStrategy.SlidingWindow, + }; + + await service.consume("sw:exp:g:f", policy, "sw:exp", "g", "f"); + await service.consume("sw:exp:g:f", policy, "sw:exp", "g", "f"); + + // Manually manipulate the memory window to simulate time passing + const memWindows = (service as any).memoryWindows as Map< + string, + number[] + >; + const windowKey = "alian:rl:sw:exp:g:f"; + memWindows.set(windowKey, [Date.now() - 120_000]); // expired entry + + const decision = await service.consume( + "sw:exp:g:f", + policy, + "sw:exp", + "g", + "f", + ); + expect(decision.allowed).toBe(true); + }); + }); + + describe("getEntry", () => { + it("returns null for unknown key", async () => { + const entry = await service.getEntry("nonexistent"); + expect(entry).toBeNull(); + }); + + it("returns entry data after a consume call", async () => { + const policy = { + limit: 5, + windowMs: 60_000, + burst: 5, + strategy: RateLimitStrategy.TokenBucket, + }; + + await service.consume( + "test-key:g:tier", + policy, + "test-key", + "global", + "tier", + ); + const entry = await service.getEntry("test-key:g:tier"); + expect(entry).not.toBeNull(); + expect(entry!.tracker).toBe("test-key"); + expect(entry!.tier).toBe("tier"); + }); + }); + + describe("reset", () => { + it("removes the entry from the registry", async () => { + const policy = { + limit: 5, + windowMs: 60_000, + burst: 5, + strategy: RateLimitStrategy.TokenBucket, + }; + + await service.consume("reset-key:g:f", policy, "reset-key", "g", "f"); + expect(service.listEntries()).toHaveLength(1); + await service.reset("reset-key:g:f"); + expect(service.listEntries()).toHaveLength(0); + }); + }); + + describe("listEntries", () => { + it("returns all registered entries", async () => { + const policy = { + limit: 5, + windowMs: 60_000, + burst: 5, + strategy: RateLimitStrategy.TokenBucket, + }; + await service.consume("k1:g:f", policy, "k1", "g", "f"); + await service.consume("k2:g:f", policy, "k2", "g", "f"); + await service.consume("k3:g:f", policy, "k3", "g", "f"); + expect(service.listEntries()).toHaveLength(3); + }); + + it("respects the limit parameter", async () => { + const policy = { + limit: 5, + windowMs: 60_000, + burst: 5, + strategy: RateLimitStrategy.TokenBucket, + }; + await service.consume("k1:g:f", policy, "k1", "g", "f"); + await service.consume("k2:g:f", policy, "k2", "g", "f"); + await service.consume("k3:g:f", policy, "k3", "g", "f"); + expect(service.listEntries(2).length).toBeLessThanOrEqual(2); + }); + }); + + describe("getStorageHealth", () => { + it("returns memory when Redis is null", async () => { + const service = new RateLimiterService(null, DEFAULT_CONFIG); + expect(await service.getStorageHealth()).toBe("memory"); + }); + + it("returns memory when Redis is unhealthy", async () => { + const mockRedis: any = { + ping: jest.fn().mockRejectedValue(new Error("Redis is down")), + }; + const service = new RateLimiterService(mockRedis, DEFAULT_CONFIG); + expect(await service.getStorageHealth()).toBe("memory"); + }); + + it("returns redis when Redis is healthy", async () => { + const mockRedis: any = { + ping: jest.fn().mockResolvedValue("PONG"), + }; + const service = new RateLimiterService(mockRedis, DEFAULT_CONFIG); + expect(await service.getStorageHealth()).toBe("redis"); + }); + }); + }); + + describe("Redis mode", () => { + let service: RateLimiterService; + let mockRedis: any; + + beforeEach(() => { + mockRedis = makeMockRedis(); + service = new RateLimiterService(mockRedis as any, DEFAULT_CONFIG); + }); + + it("uses Redis eval for token bucket consume", async () => { + const policy = { + limit: 5, + windowMs: 60_000, + burst: 5, + strategy: RateLimitStrategy.TokenBucket, + }; + + const decision = await service.consume( + "redis-key:g:f", + policy, + null, + "global", + "f", + ); + + expect(mockRedis.eval).toHaveBeenCalled(); + expect(decision.allowed).toBe(true); + expect(decision.remaining).toBe(5); + }); + + it("uses Redis eval for sliding window consume", async () => { + const policy = { + limit: 5, + windowMs: 60_000, + burst: 5, + strategy: RateLimitStrategy.SlidingWindow, + }; + + const decision = await service.consume( + "redis-sw:g:f", + policy, + null, + "global", + "f", + ); + + expect(mockRedis.eval).toHaveBeenCalled(); + expect(decision.allowed).toBe(true); + }); + + it("falls back to memory when Redis eval throws", async () => { + mockRedis.eval = jest.fn().mockRejectedValue(new Error("Redis error")); + + const policy = { + limit: 3, + windowMs: 60_000, + burst: 3, + strategy: RateLimitStrategy.TokenBucket, + }; + + const decision = await service.consume( + "fallback-key:g:f", + policy, + null, + "global", + "f", + ); + + expect(decision.allowed).toBe(true); + }); + + it("del calls redis.del when Redis is healthy", async () => { + await service.consume( + "del-key:g:f", + { + limit: 5, + windowMs: 60_000, + burst: 5, + strategy: RateLimitStrategy.TokenBucket, + }, + "k", + "g", + "f", + ); + mockRedis.del.mockClear(); + await service.reset("del-key:g:f"); + expect(mockRedis.del).toHaveBeenCalledWith("alian:rl:del-key:g:f"); + }); + }); + + describe("default strategy", () => { + it("uses token bucket by default when no strategy specified", async () => { + const service = new RateLimiterService(null, { + keyPrefix: "alian:rl:", + defaultStrategy: RateLimitStrategy.SlidingWindow, + enableFallback: true, + }); + + const policy = { limit: 5, windowMs: 60_000, burst: 5 }; + const decision = await service.consume( + "default-strategy:g:f", + policy, + "k", + "g", + "f", + ); + expect(decision.allowed).toBe(true); + + // Sliding window stores entries as arrays, token bucket stores as buckets + const memWindows = (service as any).memoryWindows; + expect(memWindows.has("alian:rl:default-strategy:g:f")).toBe(true); + }); + }); +}); diff --git a/src/rate-limiting/rate-limiter.service.ts b/src/rate-limiting/rate-limiter.service.ts new file mode 100644 index 0000000..1f1b088 --- /dev/null +++ b/src/rate-limiting/rate-limiter.service.ts @@ -0,0 +1,447 @@ +import { Inject, Injectable, Logger, Optional } from "@nestjs/common"; +import Redis from "ioredis"; +import { CACHE_REDIS_CLIENT } from "../common/cache/cache.constants"; +import { + RateLimitDecision, + RateLimitPolicy, + RateLimitEntry, + RateLimitStorage, + RateLimitStrategy, +} from "./interfaces"; +import { + rateLimitErrorsTotal, + rateLimitStorageHealth, +} from "./rate-limiting.metrics"; + +const TOKEN_BUCKET_LUA = ` +local key = KEYS[1] +local capacity = tonumber(ARGV[1]) +local rate_per_sec = tonumber(ARGV[2]) +local now_ms = tonumber(ARGV[3]) +local cost = tonumber(ARGV[4]) + +local data = redis.call('HMGET', key, 'tokens', 'ts') +local tokens = tonumber(data[1]) +local ts = tonumber(data[2]) + +if tokens == nil then + tokens = capacity + ts = now_ms +end + +local elapsed_ms = now_ms - ts +local token_delta = elapsed_ms * rate_per_sec / 1000 +tokens = math.min(capacity, tokens + token_delta) +ts = now_ms + +if tokens >= cost then + tokens = tokens - cost + redis.call('HMSET', key, 'tokens', tostring(tokens), 'ts', tostring(ts)) + local ttl = math.ceil((capacity - tokens + 1) / rate_per_sec) + if ttl < 1 then ttl = 1 end + redis.call('EXPIRE', key, ttl) + return {1, math.floor(tokens), 0} +else + local wait_ms = math.ceil((cost - tokens) / rate_per_sec * 1000) + redis.call('HMSET', key, 'tokens', tostring(tokens), 'ts', tostring(ts)) + local ttl = math.ceil((cost - tokens + 1) / rate_per_sec) + if ttl < 1 then ttl = 1 end + redis.call('EXPIRE', key, ttl) + return {0, math.floor(tokens), wait_ms} +end +`; + +const SLIDING_WINDOW_LUA = ` +local key = KEYS[1] +local limit = tonumber(ARGV[1]) +local window_ms = tonumber(ARGV[2]) +local now_ms = tonumber(ARGV[3]) +local request_id = ARGV[4] + +local min_score = now_ms - window_ms +redis.call('ZREMRANGEBYSCORE', key, 0, min_score) + +local count = redis.call('ZCARD', key) + +if count < limit then + redis.call('ZADD', key, now_ms, request_id) + redis.call('PEXPIRE', key, window_ms) + return {1, limit - count - 1, 0} +else + local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES') + local reset_at = now_ms + if #oldest > 0 then + reset_at = tonumber(oldest[2]) + window_ms + end + local retry_after = math.max(0, reset_at - now_ms) + return {0, 0, retry_after} +end +`; + +interface MemoryTokenBucketState { + tokens: number; + ts: number; +} + +interface MemoryEntry { + tracker: string; + scope: string; + tier: string; + strategy: RateLimitStrategy; + limit: number; + windowMs: number; + burst: number; + allowed: number; + denied: number; + lastResetAt: number; + lastRemaining: number; +} + +@Injectable() +export class RateLimiterService { + private readonly logger = new Logger(RateLimiterService.name); + private readonly redis: Redis | null; + private readonly memoryBuckets = new Map(); + private readonly memoryWindows = new Map(); + private readonly registry = new Map(); + private readonly keyPrefix: string; + private readonly defaultStrategy: RateLimitStrategy; + private readonly enableFallback: boolean; + private lastRedisCheck = 0; + private redisHealthy = false; + + constructor( + @Optional() @Inject(CACHE_REDIS_CLIENT) redis: Redis | null, + @Inject("RATE_LIMIT_CONFIG") config?: RateLimitConfig, + ) { + this.redis = redis; + this.keyPrefix = config?.keyPrefix ?? "alian:rl:"; + this.defaultStrategy = + config?.defaultStrategy ?? RateLimitStrategy.TokenBucket; + this.enableFallback = config?.enableFallback ?? true; + } + + async consume( + key: string, + policy: Partial, + tracker?: string, + scope?: string, + tier?: string, + ): Promise { + const resolvedPolicy: RateLimitPolicy = { + limit: policy.limit ?? 60, + windowMs: policy.windowMs ?? 60_000, + burst: policy.burst ?? policy.limit ?? 60, + strategy: policy.strategy ?? this.defaultStrategy, + }; + + const storageKey = `${this.keyPrefix}${key}`; + + let decision: RateLimitDecision; + + if (this.redis && (await this.isRedisHealthy())) { + decision = await this.consumeRedis(storageKey, resolvedPolicy); + } else { + decision = await this.consumeMemory(storageKey, resolvedPolicy); + } + + this.updateRegistry(key, tracker, scope, tier, resolvedPolicy, decision); + + return decision; + } + + private async consumeRedis( + key: string, + policy: RateLimitPolicy, + ): Promise { + try { + const ratePerSec = policy.limit / (policy.windowMs / 1000); + + let result: [number, number, number]; + + if (policy.strategy === RateLimitStrategy.SlidingWindow) { + const requestId = `${Date.now()}-${Math.random().toString(36).slice(2)}`; + result = (await this.redis!.eval( + SLIDING_WINDOW_LUA, + 1, + key, + policy.limit, + policy.windowMs, + Date.now(), + requestId, + )) as [number, number, number]; + } else { + result = (await this.redis!.eval( + TOKEN_BUCKET_LUA, + 1, + key, + policy.burst, + ratePerSec, + Date.now(), + 1, + )) as [number, number, number]; + } + + const [allowed, remaining, retryAfterMs] = result; + + let resetAt: number; + if (policy.strategy === RateLimitStrategy.SlidingWindow) { + resetAt = + retryAfterMs > 0 + ? Date.now() + retryAfterMs + : Date.now() + policy.windowMs; + } else { + resetAt = + retryAfterMs > 0 + ? Date.now() + retryAfterMs + : Date.now() + + Math.ceil(((policy.burst - remaining) / ratePerSec) * 1000); + } + + return { + allowed: allowed === 1, + remaining: Math.max(0, remaining), + resetAt, + retryAfterMs: retryAfterMs > 0 ? retryAfterMs : undefined, + }; + } catch (err) { + rateLimitErrorsTotal.inc({ operation: "consume", storage: "redis" }); + this.logger.warn( + { error: err.message, key }, + "Redis rate-limit consume failed, falling back to memory", + ); + return this.consumeMemory(key, policy); + } + } + + private async consumeMemory( + key: string, + policy: RateLimitPolicy, + ): Promise { + const now = Date.now(); + + if (policy.strategy === RateLimitStrategy.SlidingWindow) { + return this.consumeMemorySlidingWindow(key, policy, now); + } + + return this.consumeMemoryTokenBucket(key, policy, now); + } + + private consumeMemoryTokenBucket( + key: string, + policy: RateLimitPolicy, + now: number, + ): RateLimitDecision { + const ratePerSec = policy.limit / (policy.windowMs / 1000); + const state = this.memoryBuckets.get(key); + + let tokens = policy.burst; + let ts = now; + + if (state) { + const elapsed = now - state.ts; + const refill = (elapsed * ratePerSec) / 1000; + tokens = Math.min(policy.burst, state.tokens + refill); + ts = now; + } + + const allowed = tokens >= 1; + if (allowed) { + tokens -= 1; + } + + this.memoryBuckets.set(key, { tokens, ts }); + + const remaining = Math.max(0, Math.floor(tokens)); + let retryAfterMs: number | undefined; + let resetAt = now; + + if (!allowed) { + retryAfterMs = Math.ceil(((1 - tokens) / ratePerSec) * 1000); + resetAt = now + retryAfterMs; + } else { + resetAt = now + Math.ceil(((policy.burst - tokens) / ratePerSec) * 1000); + } + + return { + allowed, + remaining, + resetAt, + retryAfterMs, + }; + } + + private consumeMemorySlidingWindow( + key: string, + policy: RateLimitPolicy, + now: number, + ): RateLimitDecision { + const windowStart = now - policy.windowMs; + let entries = this.memoryWindows.get(key); + if (!entries) { + entries = []; + this.memoryWindows.set(key, entries); + } + + entries = entries.filter((ts) => ts > windowStart); + this.memoryWindows.set(key, entries); + + const count = entries.length; + + if (count < policy.limit) { + entries.push(now); + const remaining = policy.limit - count - 1; + return { + allowed: true, + remaining: Math.max(0, remaining), + resetAt: now + policy.windowMs, + }; + } + + const oldest = entries[0]; + const retryAfterMs = Math.max(0, oldest + policy.windowMs - now); + return { + allowed: false, + remaining: 0, + resetAt: oldest + policy.windowMs, + retryAfterMs, + }; + } + + async getEntry(key: string): Promise { + const entry = this.registry.get(key); + if (!entry) return null; + + let remaining: number; + let resetAt: number; + + if (this.redis && (await this.isRedisHealthy())) { + try { + if (entry.strategy === RateLimitStrategy.SlidingWindow) { + const windowStart = Date.now() - entry.windowMs; + const count = await this.redis!.zcount( + `${this.keyPrefix}${key}`, + windowStart, + "+inf", + ); + remaining = Math.max(0, entry.limit - count); + } else { + const data = await this.redis!.hgetall(`${this.keyPrefix}${key}`); + const tokens = parseFloat(data.tokens ?? "0"); + remaining = Math.max(0, Math.floor(tokens)); + } + resetAt = entry.lastResetAt; + } catch { + remaining = entry.lastRemaining; + resetAt = entry.lastResetAt; + } + } else { + remaining = entry.lastRemaining; + resetAt = entry.lastResetAt; + } + + return { + key, + tracker: entry.tracker, + scope: entry.scope, + tier: entry.tier, + strategy: entry.strategy, + limit: entry.limit, + windowMs: entry.windowMs, + remaining, + resetAt, + }; + } + + async reset(key: string): Promise { + if (this.redis && (await this.isRedisHealthy())) { + try { + await this.redis!.del(`${this.keyPrefix}${key}`); + this.registry.delete(key); + return true; + } catch { + // fall through to memory + } + } + + this.memoryBuckets.delete(`${this.keyPrefix}${key}`); + this.memoryWindows.delete(`${this.keyPrefix}${key}`); + this.registry.delete(key); + return true; + } + + listEntries(limit = 100): RateLimitEntry[] { + return Array.from(this.registry.entries()) + .slice(0, limit) + .map(([key, entry]) => ({ + key, + tracker: entry.tracker, + scope: entry.scope, + tier: entry.tier, + strategy: entry.strategy, + limit: entry.limit, + windowMs: entry.windowMs, + remaining: entry.lastRemaining, + resetAt: entry.lastResetAt, + })); + } + + async getStorageHealth(): Promise { + if (!this.redis) return "memory"; + const healthy = await this.isRedisHealthy(); + return healthy ? "redis" : "memory"; + } + + private async isRedisHealthy(): Promise { + if (!this.redis) return false; + + const now = Date.now(); + if (now - this.lastRedisCheck < 5_000) { + return this.redisHealthy; + } + this.lastRedisCheck = now; + + try { + await this.redis.ping(); + this.redisHealthy = true; + rateLimitStorageHealth.set({ storage: "redis" }, 1); + rateLimitStorageHealth.set({ storage: "memory" }, 0); + } catch { + this.redisHealthy = false; + rateLimitStorageHealth.set({ storage: "redis" }, 0); + rateLimitStorageHealth.set({ storage: "memory" }, 1); + } + + return this.redisHealthy; + } + + private updateRegistry( + key: string, + tracker: string | undefined, + scope: string | undefined, + tier: string | undefined, + policy: RateLimitPolicy, + decision: RateLimitDecision, + ): void { + const existing = this.registry.get(key); + this.registry.set(key, { + tracker: tracker ?? existing?.tracker ?? "unknown", + scope: scope ?? existing?.scope ?? "global", + tier: tier ?? existing?.tier ?? "free", + strategy: policy.strategy, + limit: policy.limit, + windowMs: policy.windowMs, + burst: policy.burst, + allowed: (existing?.allowed ?? 0) + (decision.allowed ? 1 : 0), + denied: (existing?.denied ?? 0) + (decision.allowed ? 0 : 1), + lastResetAt: decision.resetAt, + lastRemaining: decision.remaining, + }); + } +} + +export interface RateLimitConfig { + keyPrefix?: string; + defaultStrategy?: RateLimitStrategy; + enableFallback?: boolean; +} diff --git a/src/rate-limiting/rate-limiting.constants.ts b/src/rate-limiting/rate-limiting.constants.ts new file mode 100644 index 0000000..ea14754 --- /dev/null +++ b/src/rate-limiting/rate-limiting.constants.ts @@ -0,0 +1,15 @@ +export const RATE_LIMITING_MODULE_OPTIONS = Symbol( + "RATE_LIMITING_MODULE_OPTIONS", +); + +export const RATE_LIMIT_CONFIG = "RATE_LIMIT_CONFIG"; + +export const REDIS_RATE_LIMIT_CLIENT = Symbol("REDIS_RATE_LIMIT_CLIENT"); + +export const RATE_LIMIT_KEY_PREFIX = "alian:ratelimit:"; + +export const DEFAULT_RATE_LIMIT_WINDOW_MS = 60_000; + +export const DEFAULT_RATE_LIMIT_STRATEGY = "token-bucket"; + +export const REDIS_RATE_LIMIT_TTL_SECONDS = 120; diff --git a/src/rate-limiting/rate-limiting.controller.spec.ts b/src/rate-limiting/rate-limiting.controller.spec.ts new file mode 100644 index 0000000..65e184f --- /dev/null +++ b/src/rate-limiting/rate-limiting.controller.spec.ts @@ -0,0 +1,376 @@ +import { Test, TestingModule } from "@nestjs/testing"; +import { UnauthorizedException } from "@nestjs/common"; +import { RateLimitingController } from "./rate-limiting.controller"; +import { RateLimiterService } from "./rate-limiter.service"; +import { RateLimitStorage, RateLimitStrategy } from "./interfaces"; + +function makeMockRateLimiter() { + const mock = { + getStorageHealth: jest.fn().mockResolvedValue("redis" as RateLimitStorage), + listEntries: jest.fn().mockReturnValue([]), + consume: jest.fn().mockResolvedValue({ + allowed: true, + remaining: 99, + resetAt: Date.now() + 60_000, + }), + reset: jest.fn().mockResolvedValue(true), + }; + return mock as unknown as RateLimiterService; +} + +function makeReq(opts: { authorization?: string; token?: string } = {}): any { + return { + headers: opts.authorization ? { authorization: opts.authorization } : {}, + query: opts.token ? { token: opts.token } : {}, + }; +} + +describe("RateLimitingController", () => { + let controller: RateLimitingController; + let rateLimiter: ReturnType; + + beforeEach(async () => { + rateLimiter = makeMockRateLimiter(); + + const module: TestingModule = await Test.createTestingModule({ + controllers: [RateLimitingController], + providers: [{ provide: RateLimiterService, useValue: rateLimiter }], + }).compile(); + + controller = module.get(RateLimitingController); + }); + + describe("getDashboard", () => { + it("returns aggregated metrics and entries", async () => { + rateLimiter.listEntries = jest.fn().mockReturnValue([ + { + key: "user:1:/api/test:free", + tracker: "user:1", + scope: "/api/test", + tier: "free", + strategy: RateLimitStrategy.TokenBucket, + limit: 100, + windowMs: 60_000, + remaining: 50, + resetAt: Date.now() + 30_000, + }, + ]); + + const result = await controller.getDashboard(makeReq()); + + expect(result.storage).toBe("redis"); + expect(result.totals).toHaveProperty("allowed"); + expect(result.totals).toHaveProperty("denied"); + expect(result.totals).toHaveProperty("errors"); + expect(result.totals).toHaveProperty("rate"); + expect(result.activeEntries).toBe(1); + expect(result.topEntries).toHaveLength(1); + expect(result.strategyDistribution).toHaveProperty("token-bucket"); + }); + + it("calculates denial rate from allowed and denied totals", async () => { + const { rateLimitAllowedTotal, rateLimitDeniedTotal } = + await import("./rate-limiting.metrics"); + rateLimitAllowedTotal.inc( + { + tier: "free", + scope: "global", + strategy: "token-bucket", + key: "global", + }, + 90, + ); + rateLimitDeniedTotal.inc( + { + tier: "free", + scope: "global", + strategy: "token-bucket", + key: "global", + }, + 10, + ); + + const result = await controller.getDashboard(makeReq()); + + expect(result.totals.allowed).toBe(90); + expect(result.totals.denied).toBe(10); + expect(result.totals.rate).toBe(10); + }); + }); + + describe("getStatus", () => { + it("returns all entries by default", async () => { + rateLimiter.listEntries = jest.fn().mockReturnValue([ + { + key: "user:1:global:free", + tracker: "user:1", + scope: "global", + tier: "free", + strategy: RateLimitStrategy.TokenBucket, + limit: 100, + windowMs: 60_000, + remaining: 50, + resetAt: Date.now() + 30_000, + }, + ]); + + const result = await controller.getStatus(makeReq()); + expect(result.count).toBe(1); + expect(result.entries).toHaveLength(1); + }); + + it("filters by tracker", async () => { + const entries = [ + { + key: "user:1:global:free", + tracker: "user:1", + scope: "global", + tier: "free", + strategy: RateLimitStrategy.TokenBucket, + limit: 100, + windowMs: 60000, + remaining: 50, + resetAt: 0, + }, + { + key: "user:2:global:free", + tracker: "user:2", + scope: "global", + tier: "free", + strategy: RateLimitStrategy.TokenBucket, + limit: 100, + windowMs: 60000, + remaining: 50, + resetAt: 0, + }, + ]; + rateLimiter.listEntries = jest.fn().mockReturnValue(entries); + + const result = await controller.getStatus(makeReq(), "user:1"); + expect(result.count).toBe(1); + expect(result.entries[0].tracker).toBe("user:1"); + }); + + it("filters by scope", async () => { + const entries = [ + { + key: "user:1:/api/test:free", + tracker: "user:1", + scope: "/api/test", + tier: "free", + strategy: RateLimitStrategy.TokenBucket, + limit: 100, + windowMs: 60000, + remaining: 50, + resetAt: 0, + }, + { + key: "user:1:/api/other:free", + tracker: "user:1", + scope: "/api/other", + tier: "free", + strategy: RateLimitStrategy.TokenBucket, + limit: 100, + windowMs: 60000, + remaining: 50, + resetAt: 0, + }, + ]; + rateLimiter.listEntries = jest.fn().mockReturnValue(entries); + + const result = await controller.getStatus( + makeReq(undefined), + undefined, + "/api/test", + ); + expect(result.count).toBe(1); + expect(result.entries[0].scope).toBe("/api/test"); + }); + + it("filters by tier", async () => { + const entries = [ + { + key: "user:1:g:free", + tracker: "user:1", + scope: "g", + tier: "free", + strategy: RateLimitStrategy.TokenBucket, + limit: 100, + windowMs: 60000, + remaining: 50, + resetAt: 0, + }, + { + key: "user:1:g:paid", + tracker: "user:1", + scope: "g", + tier: "paid", + strategy: RateLimitStrategy.TokenBucket, + limit: 100, + windowMs: 60000, + remaining: 50, + resetAt: 0, + }, + ]; + rateLimiter.listEntries = jest.fn().mockReturnValue(entries); + + const result = await controller.getStatus( + makeReq(undefined), + undefined, + undefined, + "paid", + ); + expect(result.count).toBe(1); + expect(result.entries[0].tier).toBe("paid"); + }); + + it("respects the limit parameter", async () => { + const entries: any[] = []; + for (let i = 0; i < 10; i++) { + entries.push({ + key: `key:${i}`, + tracker: `tracker:${i}`, + scope: "global", + tier: "free", + strategy: RateLimitStrategy.TokenBucket, + limit: 100, + windowMs: 60000, + remaining: 50, + resetAt: 0, + }); + } + rateLimiter.listEntries = jest.fn((limit?: number) => + entries.slice(0, limit ?? entries.length), + ); + + const result = await controller.getStatus( + makeReq(undefined), + undefined, + undefined, + undefined, + "5", + ); + expect(result.count).toBe(5); + }); + }); + + describe("getMetrics", () => { + it("returns Prometheus text exposition", async () => { + const result = await controller.getMetrics(makeReq()); + expect(typeof result).toBe("string"); + expect(result).toContain("alian_structure_rate_limit"); + }); + }); + + describe("setEntry", () => { + it("calls consume with the provided config", async () => { + const body = { + key: "test-key", + strategy: RateLimitStrategy.TokenBucket, + limit: 10, + windowMs: 60_000, + }; + + await controller.setEntry(makeReq(), body); + + expect(rateLimiter.consume).toHaveBeenCalledWith( + "test-key:global:free", + expect.objectContaining({ + limit: 10, + windowMs: 60_000, + strategy: RateLimitStrategy.TokenBucket, + }), + "key:test-key", + "global", + "free", + ); + }); + + it("uses custom scope and tier when provided", async () => { + const body = { + key: "test-key", + strategy: RateLimitStrategy.SlidingWindow, + limit: 20, + windowMs: 30_000, + scope: "api", + tier: "paid", + }; + + await controller.setEntry(makeReq(), body); + + expect(rateLimiter.consume).toHaveBeenCalledWith( + "test-key:api:paid", + expect.objectContaining({ + strategy: RateLimitStrategy.SlidingWindow, + }), + "key:test-key", + "api", + "paid", + ); + }); + }); + + describe("resetEntry", () => { + it("calls reset with the composed key", async () => { + await controller.resetEntry(makeReq(), "mykey", "global", "free"); + expect(rateLimiter.reset).toHaveBeenCalledWith("mykey:global:free"); + }); + + it("defaults scope and tier when not provided", async () => { + await controller.resetEntry(makeReq(), "mykey"); + expect(rateLimiter.reset).toHaveBeenCalledWith("mykey:global:free"); + }); + }); + + describe("getStorageHealth", () => { + it("returns the storage health from the service", async () => { + const result = await controller.getStorageHealth(makeReq()); + expect(result).toBe("redis"); + }); + }); + + describe("authorization", () => { + const ORIGINAL = process.env.METRICS_AUTH_TOKEN; + + afterEach(() => { + if (ORIGINAL === undefined) { + delete process.env.METRICS_AUTH_TOKEN; + } else { + process.env.METRICS_AUTH_TOKEN = ORIGINAL; + } + }); + + it("allows access when no token is configured", async () => { + delete process.env.METRICS_AUTH_TOKEN; + await expect(controller.getDashboard(makeReq())).resolves.toBeDefined(); + }); + + it("rejects when token is configured but not provided", async () => { + process.env.METRICS_AUTH_TOKEN = "secret123"; + await expect(controller.getDashboard(makeReq())).rejects.toThrow( + UnauthorizedException, + ); + }); + + it("accepts correct token via Authorization header", async () => { + process.env.METRICS_AUTH_TOKEN = "secret123"; + await expect( + controller.getDashboard(makeReq({ authorization: "Bearer secret123" })), + ).resolves.toBeDefined(); + }); + + it("accepts correct token via query param", async () => { + process.env.METRICS_AUTH_TOKEN = "secret123"; + await expect( + controller.getDashboard(makeReq({ token: "secret123" })), + ).resolves.toBeDefined(); + }); + + it("rejects wrong token", async () => { + process.env.METRICS_AUTH_TOKEN = "secret123"; + await expect( + controller.getDashboard(makeReq({ authorization: "Bearer wrong" })), + ).rejects.toThrow(UnauthorizedException); + }); + }); +}); diff --git a/src/rate-limiting/rate-limiting.controller.ts b/src/rate-limiting/rate-limiting.controller.ts new file mode 100644 index 0000000..640d069 --- /dev/null +++ b/src/rate-limiting/rate-limiting.controller.ts @@ -0,0 +1,249 @@ +import { + Body, + Controller, + Delete, + Get, + Param, + Post, + Query, + Req, + UnauthorizedException, +} from "@nestjs/common"; +import { ApiOperation, ApiResponse, ApiTags } from "@nestjs/swagger"; +import { Request } from "express"; +import { timingSafeEqual } from "crypto"; +import { Public } from "../common/decorators/public.decorator"; +import { SkipKyc } from "../common/decorators/skip-kyc.decorator"; +import { RateLimiterService } from "./rate-limiter.service"; +import { RateLimitStorage } from "./interfaces"; +import { SetRateLimitDto, RateLimitStatusDto } from "./dto/rate-limit-dto"; +import { register } from "../config/metrics"; + +const DENIED_METRIC = "alian_structure_rate_limit_denied_total"; +const ALLOWED_METRIC = "alian_structure_rate_limit_allowed_total"; +const ERRORS_METRIC = "alian_structure_rate_limit_errors_total"; + +@ApiTags("Rate Limiting") +@Controller("rate-limiting") +@Public() +@SkipKyc() +export class RateLimitingController { + constructor(private readonly rateLimiter: RateLimiterService) {} + + @Get("dashboard") + @ApiOperation({ + summary: "Rate-limiting dashboard for Grant reviewers", + description: + "Aggregated usage and impact metrics for rate-limiting — quotas, " + + "blocked requests, active strategies, and storage health.", + }) + @ApiResponse({ status: 200, description: "Dashboard snapshot" }) + async getDashboard(@Req() req: Request) { + this.assertAuthorized(req); + + const storage = await this.rateLimiter.getStorageHealth(); + const entries = this.rateLimiter.listEntries(500); + const denied = await this.getMetricValue(DENIED_METRIC); + const allowed = await this.getMetricValue(ALLOWED_METRIC); + const errors = await this.getMetricValue(ERRORS_METRIC); + + const deniedByTier: Record = {}; + for (const entry of entries) { + deniedByTier[entry.tier] = (deniedByTier[entry.tier] ?? 0) + 0; + } + + const allowedByTier: Record = {}; + for (const entry of entries) { + allowedByTier[entry.tier] = (allowedByTier[entry.tier] ?? 0) + 0; + } + + const strategyCounts: Record = {}; + for (const entry of entries) { + strategyCounts[entry.strategy] = + (strategyCounts[entry.strategy] ?? 0) + 1; + } + + return { + timestamp: new Date().toISOString(), + storage, + totals: { + allowed: allowed, + denied: denied, + errors: errors, + rate: + allowed + denied > 0 + ? Number(((denied / (allowed + denied)) * 100).toFixed(2)) + : 0, + }, + byTier: { + denied: deniedByTier, + allowed: allowedByTier, + }, + strategyDistribution: strategyCounts, + activeEntries: entries.length, + topEntries: entries.slice(0, 20), + }; + } + + @Get("status") + @ApiOperation({ + summary: "Rate-limit status for a tracker/scope", + description: + "Returns the current remaining quota and reset time for a given tracker, " + + "optionally filtered by scope and tier.", + }) + @ApiResponse({ status: 200, description: "Rate-limit status" }) + async getStatus( + @Req() req: Request, + @Query("tracker") tracker?: string, + @Query("scope") scope?: string, + @Query("tier") tier?: string, + @Query("limit") limit?: string, + ) { + this.assertAuthorized(req); + + const query = new RateLimitStatusDto(); + query.tracker = tracker; + query.scope = scope; + query.tier = tier; + query.limit = limit ? Number(limit) : 100; + + let entries = this.rateLimiter.listEntries(query.limit); + + if (query.tracker) { + entries = entries.filter((e) => e.tracker === query.tracker); + } + if (query.scope) { + entries = entries.filter((e) => e.scope === query.scope); + } + if (query.tier) { + entries = entries.filter((e) => e.tier === query.tier); + } + + return { + count: entries.length, + entries, + }; + } + + @Get("metrics") + @ApiOperation({ + summary: "Prometheus metrics for rate limiting", + description: + "Returns all rate-limiting Prometheus metrics in text-exposition format.", + }) + @ApiResponse({ status: 200, description: "Prometheus metrics" }) + async getMetrics(@Req() req: Request) { + this.assertAuthorized(req); + return register.metrics(); + } + + @Post("entries") + @ApiOperation({ + summary: "Set a custom rate-limit entry (for testing/bypass)", + description: + "Manually create or update a rate-limit configuration for a specific key. " + + "Useful for Grant reviewers to grant temporary quota adjustments.", + }) + @ApiResponse({ status: 201, description: "Entry created" }) + async setEntry(@Req() req: Request, @Body() body: SetRateLimitDto) { + this.assertAuthorized(req); + + const policy = { + limit: body.limit, + windowMs: body.windowMs, + burst: body.burst ?? body.limit, + strategy: body.strategy, + }; + + const decision = await this.rateLimiter.consume( + `${body.key}:${body.scope ?? "global"}:${body.tier ?? "free"}`, + policy, + `key:${body.key}`, + body.scope ?? "global", + body.tier ?? "free", + ); + + return { + key: body.key, + policy, + decision, + message: "Rate-limit entry configured", + }; + } + + @Delete("entries/:key") + @ApiOperation({ + summary: "Reset a rate-limit entry", + description: + "Reset (clear) the rate-limit counter for a specific tracker/scope/tier " + + "combination. Grant reviewers use this to unblock throttled keys.", + }) + @ApiResponse({ status: 200, description: "Reset result" }) + async resetEntry( + @Req() req: Request, + @Param("key") key: string, + @Query("scope") scope?: string, + @Query("tier") tier?: string, + ) { + this.assertAuthorized(req); + + const rateLimitKey = `${key}:${scope ?? "global"}:${tier ?? "free"}`; + const reset = await this.rateLimiter.reset(rateLimitKey); + + return { + key: rateLimitKey, + reset, + message: reset + ? `Rate-limit entry for "${rateLimitKey}" has been reset` + : `No rate-limit entry found for "${rateLimitKey}"`, + }; + } + + @Get("storage") + @ApiOperation({ + summary: "Rate-limit storage health", + description: + "Reports whether the rate limiter is using Redis or the in-memory fallback.", + }) + @ApiResponse({ status: 200, description: "Storage health" }) + async getStorageHealth(@Req() req: Request): Promise { + this.assertAuthorized(req); + return this.rateLimiter.getStorageHealth(); + } + + private async getMetricValue(metricName: string): Promise { + const metrics = await register.getMetricsAsJSON(); + const metric = metrics.find((m) => m.name === metricName); + if (!metric) return 0; + return metric.values.reduce((sum, v) => sum + (v.value ?? 0), 0); + } + + private assertAuthorized(req: Request): void { + const expected = process.env.METRICS_AUTH_TOKEN; + if (!expected) return; + + const header = req.headers["authorization"]; + const headerToken = + typeof header === "string" && header.startsWith("Bearer ") + ? header.slice("Bearer ".length) + : undefined; + const queryToken = + typeof req.query?.token === "string" ? req.query.token : undefined; + const provided = headerToken ?? queryToken; + + if (!provided || !this.constantTimeEquals(provided, expected)) { + throw new UnauthorizedException("Invalid or missing metrics token"); + } + } + + private constantTimeEquals(a: string, b: string): boolean { + const bufA = Buffer.from(a); + const bufB = Buffer.from(b); + if (bufA.length !== bufB.length) { + timingSafeEqual(bufA, bufA); + return false; + } + return timingSafeEqual(bufA, bufB); + } +} diff --git a/src/rate-limiting/rate-limiting.guard.spec.ts b/src/rate-limiting/rate-limiting.guard.spec.ts new file mode 100644 index 0000000..7364e48 --- /dev/null +++ b/src/rate-limiting/rate-limiting.guard.spec.ts @@ -0,0 +1,374 @@ +import { ExecutionContext, HttpException, HttpStatus } from "@nestjs/common"; +import { Reflector } from "@nestjs/core"; +import { DistributedRateLimitGuard } from "./rate-limiting.guard"; +import { RateLimiterService } from "./rate-limiter.service"; +import { RateLimitStrategy } from "./interfaces"; + +function createMockResponse() { + return { + header: jest.fn(), + setHeader: jest.fn(), + }; +} + +function createContext(overrides: Record = {}): { + context: ExecutionContext; + request: any; + response: { header: jest.Mock; setHeader: jest.Mock }; +} { + const request = { + ip: "127.0.0.1", + headers: {}, + user: undefined, + route: { path: "/api/test" }, + originalUrl: "/api/test", + url: "/api/test", + authType: undefined, + ...overrides, + }; + + const response = createMockResponse(); + + const context = { + getHandler: jest.fn(), + getClass: jest.fn(), + switchToHttp: () => ({ + getRequest: () => request, + getResponse: () => response, + }), + } as unknown as ExecutionContext; + + return { context, request, response }; +} + +function createMockRateLimiter(decision: any): RateLimiterService { + const mock = { + consume: jest.fn().mockResolvedValue(decision), + }; + return mock as any; +} + +describe("DistributedRateLimitGuard", () => { + let guard: DistributedRateLimitGuard; + let reflector: Reflector; + + beforeEach(() => { + reflector = new Reflector(); + const rateLimiter = createMockRateLimiter({ + allowed: true, + remaining: 99, + resetAt: Date.now() + 60_000, + }); + guard = new DistributedRateLimitGuard(reflector, rateLimiter); + }); + + describe("canActivate", () => { + it("allows requests within the rate limit", async () => { + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context } = createContext({ + user: { id: "user-1", role: "user" }, + }); + + await expect(guard.canActivate(context)).resolves.toBe(true); + }); + + it("emits X-RateLimit headers with tier, strategy, and remaining", async () => { + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context, response } = createContext({ + user: { id: "user-1", role: "user" }, + }); + + await guard.canActivate(context); + + expect(response.header).toHaveBeenCalledWith("X-RateLimit-Limit", 100); + expect(response.header).toHaveBeenCalledWith("X-RateLimit-Remaining", 99); + expect(response.header).toHaveBeenCalledWith("X-RateLimit-Tier", "free"); + expect(response.header).toHaveBeenCalledWith( + "X-RateLimit-Strategy", + RateLimitStrategy.TokenBucket, + ); + }); + + it("throws 429 when rate limit is exceeded", async () => { + const rateLimiter = createMockRateLimiter({ + allowed: false, + remaining: 0, + resetAt: Date.now() + 30_000, + retryAfterMs: 30_000, + }); + guard = new DistributedRateLimitGuard(reflector, rateLimiter); + + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context } = createContext({ + user: { id: "user-1", role: "user" }, + }); + + await expect(guard.canActivate(context)).rejects.toThrow(HttpException); + try { + await guard.canActivate(context); + } catch (error) { + const httpError = error as HttpException; + expect(httpError.getStatus()).toBe(HttpStatus.TOO_MANY_REQUESTS); + const body = httpError.getResponse() as any; + expect(body.statusCode).toBe(429); + expect(body.retryAfter).toBeDefined(); + expect(body.strategy).toBeDefined(); + } + }); + + it("sets Retry-After header on denial", async () => { + const rateLimiter = createMockRateLimiter({ + allowed: false, + remaining: 0, + resetAt: Date.now() + 60_000, + retryAfterMs: 60_000, + }); + guard = new DistributedRateLimitGuard(reflector, rateLimiter); + + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context, response } = createContext({ + user: { id: "user-1", role: "user" }, + }); + + try { + await guard.canActivate(context); + } catch { + // expected + } + + expect(response.header).toHaveBeenCalledWith("Retry-After", 60); + }); + + it("resolves enterprise tier for admin users", async () => { + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context, response } = createContext({ + user: { id: "user-1", role: "admin" }, + }); + + await guard.canActivate(context); + + expect(response.header).toHaveBeenCalledWith( + "X-RateLimit-Tier", + "enterprise", + ); + }); + + it("resolves paid tier for operator users", async () => { + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context, response } = createContext({ + user: { id: "user-2", role: "operator" }, + }); + + await guard.canActivate(context); + + expect(response.header).toHaveBeenCalledWith("X-RateLimit-Tier", "paid"); + }); + + it("resolves enterprise tier for api-key auth", async () => { + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context, response } = createContext({ + authType: "api-key", + }); + + await guard.canActivate(context); + + expect(response.header).toHaveBeenCalledWith( + "X-RateLimit-Tier", + "enterprise", + ); + }); + + it("uses explicit user tier when provided", async () => { + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context, response } = createContext({ + user: { id: "user-3", tier: "enterprise" }, + }); + + await guard.canActivate(context); + + expect(response.header).toHaveBeenCalledWith( + "X-RateLimit-Tier", + "enterprise", + ); + }); + + it("passes strategy from decorator options", async () => { + const rateLimiter = createMockRateLimiter({ + allowed: true, + remaining: 49, + resetAt: Date.now() + 60_000, + }); + guard = new DistributedRateLimitGuard(reflector, rateLimiter); + + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue({ + level: undefined, + limit: 50, + windowMs: 60_000, + burst: 50, + strategy: RateLimitStrategy.SlidingWindow, + }); + + const { context, response } = createContext({ + user: { id: "user-1", role: "user" }, + }); + + await guard.canActivate(context); + + const consumeCall = (rateLimiter as any).consume as jest.Mock; + expect(consumeCall).toHaveBeenCalledWith( + expect.any(String), + expect.objectContaining({ + strategy: RateLimitStrategy.SlidingWindow, + }), + "user:user-1", + "/api/test", + "free", + ); + expect(response.header).toHaveBeenCalledWith( + "X-RateLimit-Strategy", + RateLimitStrategy.SlidingWindow, + ); + }); + + it("uses custom key from decorator options as scope", async () => { + const rateLimiter = createMockRateLimiter({ + allowed: true, + remaining: 99, + resetAt: Date.now() + 60_000, + }); + guard = new DistributedRateLimitGuard(reflector, rateLimiter); + + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue({ + level: "custom", + limit: 100, + windowMs: 60_000, + burst: 120, + key: "my-custom-scope", + }); + + const { context } = createContext({ + user: { id: "user-1", role: "user" }, + }); + + await guard.canActivate(context); + + const consumeCall = (rateLimiter as any).consume as jest.Mock; + const keyArg = consumeCall.mock.calls[0][0]; + const scopeArg = consumeCall.mock.calls[0][3]; + expect(keyArg).toContain("my-custom-scope"); + expect(scopeArg).toBe("my-custom-scope"); + }); + + it("tracks request tracker from IP when no user", async () => { + const rateLimiter = createMockRateLimiter({ + allowed: true, + remaining: 99, + resetAt: Date.now() + 60_000, + }); + guard = new DistributedRateLimitGuard(reflector, rateLimiter); + + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context } = createContext({ + user: undefined, + ip: "10.0.0.1", + }); + + await guard.canActivate(context); + + const consumeCall = (rateLimiter as any).consume as jest.Mock; + const trackerArg = consumeCall.mock.calls[0][2]; + expect(trackerArg).toBe("ip:10.0.0.1"); + }); + + it("extracts first IP from X-Forwarded-For", async () => { + const rateLimiter = createMockRateLimiter({ + allowed: true, + remaining: 99, + resetAt: Date.now() + 60_000, + }); + guard = new DistributedRateLimitGuard(reflector, rateLimiter); + + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context } = createContext({ + user: undefined, + headers: { "x-forwarded-for": "203.0.113.5, 192.168.1.1" }, + }); + + await guard.canActivate(context); + + const consumeCall = (rateLimiter as any).consume as jest.Mock; + const trackerArg = consumeCall.mock.calls[0][2]; + expect(trackerArg).toBe("ip:203.0.113.5"); + }); + + it("handles missing user and IP gracefully", async () => { + const rateLimiter = createMockRateLimiter({ + allowed: true, + remaining: 99, + resetAt: Date.now() + 60_000, + }); + guard = new DistributedRateLimitGuard(reflector, rateLimiter); + + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue(undefined); + const { context } = createContext({ + user: undefined, + headers: {}, + ip: undefined, + }); + + await expect(guard.canActivate(context)).resolves.toBe(true); + + const consumeCall = (rateLimiter as any).consume as jest.Mock; + const trackerArg = consumeCall.mock.calls[0][2]; + expect(trackerArg).toBe("ip:unknown"); + }); + + it("warns when approaching the rate limit", async () => { + const rateLimiter = createMockRateLimiter({ + allowed: true, + remaining: 5, + resetAt: Date.now() + 60_000, + }); + guard = new DistributedRateLimitGuard(reflector, rateLimiter); + const warnSpy = jest.spyOn(guard["logger"], "warn").mockImplementation(); + + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue({ + level: undefined, + limit: 50, + windowMs: 60_000, + burst: 60, + }); + + const { context } = createContext({ + user: { id: "user-1", role: "user" }, + }); + + await guard.canActivate(context); + expect(warnSpy).toHaveBeenCalled(); + }); + + it("does not warn when remaining is high", async () => { + const rateLimiter = createMockRateLimiter({ + allowed: true, + remaining: 90, + resetAt: Date.now() + 60_000, + }); + guard = new DistributedRateLimitGuard(reflector, rateLimiter); + const warnSpy = jest.spyOn(guard["logger"], "warn").mockImplementation(); + + jest.spyOn(reflector, "getAllAndOverride").mockReturnValue({ + level: undefined, + limit: 100, + windowMs: 60_000, + burst: 120, + }); + + const { context } = createContext({ + user: { id: "user-1", role: "user" }, + }); + + await guard.canActivate(context); + expect(warnSpy).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/src/rate-limiting/rate-limiting.guard.ts b/src/rate-limiting/rate-limiting.guard.ts new file mode 100644 index 0000000..f1d1cf6 --- /dev/null +++ b/src/rate-limiting/rate-limiting.guard.ts @@ -0,0 +1,277 @@ +import { + CanActivate, + ExecutionContext, + HttpException, + HttpStatus, + Injectable, + Logger, +} from "@nestjs/common"; +import { Reflector } from "@nestjs/core"; +import { + RATE_LIMIT_KEY, + RateLimitOptions, +} from "../common/decorators/rate-limit.decorator"; +import { + RateLimitTier, + getRateLimitPolicyFromEnv, + normalizeRateLimitTier, + resolveRateLimitTierFromRole, +} from "../config/quota.config"; +import { RateLimitStrategy } from "./interfaces"; +import { RateLimiterService } from "./rate-limiter.service"; +import { + rateLimitAllowedTotal, + rateLimitDeniedTotal, +} from "./rate-limiting.metrics"; + +interface ResolvedPolicy { + tier: RateLimitTier; + label: string; + limit: number; + windowMs: number; + burst: number; + strategy: RateLimitStrategy; +} + +@Injectable() +export class DistributedRateLimitGuard implements CanActivate { + private readonly logger = new Logger(DistributedRateLimitGuard.name); + + constructor( + private readonly reflector: Reflector, + private readonly rateLimiter: RateLimiterService, + ) {} + + async canActivate(context: ExecutionContext): Promise { + const options = this.reflector.getAllAndOverride( + RATE_LIMIT_KEY, + [context.getHandler(), context.getClass()], + ); + + const request = context.switchToHttp().getRequest(); + const response = context.switchToHttp().getResponse(); + + const tier = this.resolveRequestTier(request); + const policy = this.resolvePolicy(options, tier); + + const tracker = this.getTrackerKey(request); + const scope = this.getScope(request, options); + const key = this.buildRateLimitKey(tracker, scope, policy.tier); + + const decision = await this.rateLimiter.consume( + key, + { + limit: policy.limit, + windowMs: policy.windowMs, + burst: policy.burst, + strategy: policy.strategy, + }, + tracker, + scope, + policy.tier, + ); + + this.applyHeaders(response, policy, decision); + this.recordMetrics(tier, scope, policy.strategy, decision); + + if (!decision.allowed) { + const retryAfter = decision.retryAfterMs + ? Math.ceil(decision.retryAfterMs / 1000) + : Math.ceil((decision.resetAt - Date.now()) / 1000); + + throw new HttpException( + { + statusCode: HttpStatus.TOO_MANY_REQUESTS, + message: "Rate limit exceeded", + limit: policy.limit, + remaining: 0, + resetAt: new Date(decision.resetAt).toISOString(), + retryAfter: retryAfter, + tier: policy.tier, + strategy: policy.strategy, + }, + HttpStatus.TOO_MANY_REQUESTS, + ); + } + + if (decision.remaining <= Math.max(1, Math.ceil(policy.limit * 0.1))) { + this.logger.warn( + `Approaching rate limit for ${tracker} (${policy.label}): ` + + `${policy.limit - decision.remaining}/${policy.limit}`, + ); + } + + return true; + } + + private resolvePolicy( + options: RateLimitOptions | undefined, + tier: RateLimitTier, + ): ResolvedPolicy { + const envPolicy = getRateLimitPolicyFromEnv( + tier, + process.env as Record, + ); + + if (!options) { + return { + tier, + label: tier, + limit: envPolicy.limit, + windowMs: envPolicy.windowMs, + burst: envPolicy.burst, + strategy: this.resolveStrategy(options?.strategy), + }; + } + + const configuredTier = options.level + ? normalizeRateLimitTier(options.level) + : tier; + const levelPolicy = getRateLimitPolicyFromEnv( + configuredTier, + process.env as Record, + ); + + return { + tier: configuredTier, + label: options.level || configuredTier, + limit: options.limit ?? levelPolicy.limit, + windowMs: options.windowMs ?? levelPolicy.windowMs, + burst: options.burst ?? levelPolicy.burst, + strategy: this.resolveStrategy(options.strategy), + }; + } + + private resolveStrategy(strategy?: RateLimitStrategy): RateLimitStrategy { + if (strategy) return strategy; + + const envStrategy = String( + process.env.RATE_LIMIT_DEFAULT_STRATEGY ?? "token-bucket", + ).toLowerCase(); + + if (envStrategy === "sliding-window") { + return RateLimitStrategy.SlidingWindow; + } + + return RateLimitStrategy.TokenBucket; + } + + private resolveRequestTier(request: { + authType?: string; + user?: { + id?: string | number; + role?: string; + tier?: string; + type?: string; + }; + }): RateLimitTier { + const explicitTier = request.user?.tier; + const authType = request.authType ?? request.user?.type; + + if (authType === "api-key") { + return normalizeRateLimitTier(explicitTier ?? "enterprise"); + } + + return resolveRateLimitTierFromRole( + request.user?.role, + authType, + explicitTier, + ); + } + + private getTrackerKey(request: { + ip?: string; + headers?: Record; + user?: { id?: string | number; sub?: string | number; address?: string }; + }): string { + const userId = request.user?.id ?? request.user?.sub; + if (userId !== undefined && userId !== null) { + return `user:${String(userId)}`; + } + + if (request.user?.address) { + return `wallet:${request.user.address.toLowerCase()}`; + } + + const xff = request.headers?.["x-forwarded-for"]; + if (typeof xff === "string" && xff.length > 0) { + return `ip:${xff.split(",")[0].trim()}`; + } + + return `ip:${request.ip ?? "unknown"}`; + } + + private getScope( + request: { + route?: { path?: string }; + originalUrl?: string; + url?: string; + }, + options: RateLimitOptions | undefined, + ): string { + if (options?.key) { + return options.key; + } + + if (!options) { + return "global"; + } + + return request.route?.path || request.originalUrl || request.url || "route"; + } + + private buildRateLimitKey( + tracker: string, + scope: string, + tier: string, + ): string { + return `${tracker}:${scope}:${tier}`; + } + + private applyHeaders( + response: any, + policy: ResolvedPolicy, + decision: { + allowed: boolean; + remaining: number; + resetAt: number; + retryAfterMs?: number; + }, + ): void { + const headers: Array<[string, string | number]> = [ + ["X-RateLimit-Limit", policy.limit], + ["X-RateLimit-Remaining", decision.remaining], + ["X-RateLimit-Reset", new Date(decision.resetAt).toISOString()], + ["X-RateLimit-Tier", policy.tier], + ["X-RateLimit-Strategy", policy.strategy], + ]; + + if (!decision.allowed && decision.retryAfterMs) { + const retryAfterSeconds = Math.ceil(decision.retryAfterMs / 1000); + headers.push(["Retry-After", retryAfterSeconds]); + } + + for (const [name, value] of headers) { + if (typeof response?.header === "function") { + response.header(name, value); + } else if (typeof response?.setHeader === "function") { + response.setHeader(name, value); + } + } + } + + private recordMetrics( + tier: RateLimitTier, + scope: string, + strategy: RateLimitStrategy, + decision: { allowed: boolean; remaining: number }, + ): void { + const label = { tier, scope, strategy, key: scope }; + + if (decision.allowed) { + rateLimitAllowedTotal.inc(label); + } else { + rateLimitDeniedTotal.inc(label); + } + } +} diff --git a/src/rate-limiting/rate-limiting.metrics.ts b/src/rate-limiting/rate-limiting.metrics.ts new file mode 100644 index 0000000..e7fd3a0 --- /dev/null +++ b/src/rate-limiting/rate-limiting.metrics.ts @@ -0,0 +1,44 @@ +import client from "prom-client"; +import { register } from "../config/metrics"; + +const PREFIX = "alian_structure_"; + +function getOrCreateCounter( + config: client.CounterConfiguration, +): client.Counter { + const existing = register.getSingleMetric(config.name); + if (existing) return existing as client.Counter; + return new client.Counter({ ...config, registers: [register] }); +} + +function getOrCreateGauge( + config: client.GaugeConfiguration, +): client.Gauge { + const existing = register.getSingleMetric(config.name); + if (existing) return existing as client.Gauge; + return new client.Gauge({ ...config, registers: [register] }); +} + +export const rateLimitAllowedTotal = getOrCreateCounter({ + name: `${PREFIX}rate_limit_allowed_total`, + help: "Total number of rate-limit requests allowed", + labelNames: ["tier", "scope", "strategy", "key"], +}); + +export const rateLimitDeniedTotal = getOrCreateCounter({ + name: `${PREFIX}rate_limit_denied_total`, + help: "Total number of rate-limit requests denied", + labelNames: ["tier", "scope", "strategy", "key"], +}); + +export const rateLimitErrorsTotal = getOrCreateCounter({ + name: `${PREFIX}rate_limit_errors_total`, + help: "Total number of rate-limiter storage errors", + labelNames: ["operation", "storage"], +}); + +export const rateLimitStorageHealth = getOrCreateGauge({ + name: `${PREFIX}rate_limit_storage_health`, + help: "Rate-limit storage health (1 = redis healthy, 0 = redis unhealthy/in-memory fallback)", + labelNames: ["storage"], +}); diff --git a/src/rate-limiting/rate-limiting.module.ts b/src/rate-limiting/rate-limiting.module.ts new file mode 100644 index 0000000..7e5e82c --- /dev/null +++ b/src/rate-limiting/rate-limiting.module.ts @@ -0,0 +1,47 @@ +import { DynamicModule, Global, Module, OnModuleInit } from "@nestjs/common"; +import { ConfigModule, ConfigService } from "@nestjs/config"; +import { RateLimiterService, RateLimitConfig } from "./rate-limiter.service"; +import { DistributedRateLimitGuard } from "./rate-limiting.guard"; +import { RateLimitingController } from "./rate-limiting.controller"; +import { RATE_LIMIT_CONFIG } from "./rate-limiting.constants"; +import { RateLimitStrategy } from "./interfaces"; + +@Global() +@Module({}) +export class RateLimitingModule implements OnModuleInit { + static forRoot(): DynamicModule { + return { + module: RateLimitingModule, + imports: [ConfigModule], + controllers: [RateLimitingController], + providers: [ + { + provide: RATE_LIMIT_CONFIG, + inject: [ConfigService], + useFactory: (configService: ConfigService): RateLimitConfig => ({ + keyPrefix: + configService.get("RATE_LIMIT_REDIS_KEY_PREFIX") ?? + "alian:rl:", + defaultStrategy: (configService.get( + "RATE_LIMIT_DEFAULT_STRATEGY", + ) ?? "token-bucket") as RateLimitStrategy, + enableFallback: + configService.get("RATE_LIMIT_FALLBACK_TO_MEMORY") !== + "false", + }), + }, + RateLimiterService, + DistributedRateLimitGuard, + ], + exports: [RateLimiterService, DistributedRateLimitGuard], + }; + } + + constructor(private readonly rateLimiter: RateLimiterService) {} + + onModuleInit(): void { + this.rateLimiter.getStorageHealth().catch(() => { + // Storage check is fire-and-forget during startup + }); + } +}