import "server-only"; import { randomUUID } from "node:crypto"; import { env } from "@/env"; import { raiseCrowdsecAlert } from "@/lib/crowdsec-alerts"; import { reportCrowdsecSignal } from "@/lib/crowdsec-report"; import { bumpCrowdsecBreakdownStat, bumpCrowdsecStat, } from "@/lib/crowdsec-stats"; import { logger } from "@/lib/logger"; import { redis } from "@/lib/redis"; import { UNKNOWN_CLIENT_IP } from "./client-ip"; /** * CrowdSec CTI (community threat intelligence) integration for the anti-DDoS * gate — the reputation side of the auto-block pipeline. * * When an IP trips a rate bucket, the gate consults CrowdSec's community * reputation for that IP (`GET /smoke/{ip}`, the freemium Enrichment API, * `x-api-key` auth) and hard-blocks known-bad repeat offenders immediately * instead of waiting for the local `maxViolations` threshold. The block lives * only in the gate's own shared Redis key (`antiddos:block:{ip}`) so every * existing consumer — the proxy check, the admin block list, the admin unban — * keeps working unchanged. CrowdSec never talks to Cloudflare and never * creates edge rules; if a Cloudflare mirror is wanted it is the gate's own * escalation logic that decides, never this module. * * Quota safety: lookups only run for IPs that already tripped a bucket (never * on the plain hot path), verdicts are cached in Redis for an hour (so a * flood from one IP costs at most one API call), concurrent lookups for the * same IP are deduped across instances with a Redis NX lock, and a 403/429 * response trips a module-wide backoff instead of hammering the API. * * Credentials come from env only (`CROWDSEC_API_KEY`) and are never written * into the admin-visible config — same contract as the Cloudflare token. * * Quota guard: every enrichment call counts against a per-day Redis counter so * a spread DDoS (many distinct IPs tripping buckets) can exhaust the day's * freemium quota only until the configured ceiling, after which lookups pause * until tomorrow instead of hammering a 429 wall. * * Sharing detections back: after a block is created this module fires the * signal push in `@/lib/crowdsec-report` (Central API watcher login + POST * /signals), strictly opt-in via CROWDSEC_REPORT_ENABLED and always * fire-and-forget. */ export class CrowdsecApiError extends Error {} export interface CrowdsecApiConfig { baseUrl: string; apiKey: string | null; } export type CrowdsecReputation = | "malicious" | "suspicious" | "known" | "safe" | "benign" | "unknown"; export interface CrowdsecVerdict { ip: string; /** Raw CTI reputation enum; null when the IP is unknown to the community. */ reputation: CrowdsecReputation | null; /** `scores.overall.total` — CrowdSec malevolence score, 0-5. */ score: number; /** `scores.overall.aggressiveness` — 0-5. */ aggressiveness: number; confidence: string | null; /** Reported attack categories, e.g. ["http:scan", "ssh:bruteforce"]. */ behaviors: string[]; /** CrowdSec tags IPs carrying false-positive classifications as safe. */ falsePositive: boolean; checkedAt: number; } export interface CrowdsecConnectionStatus { ok: boolean; message?: string; at: number; } /** Why a CrowdSec-sourced block exists — persisted next to the block key. */ export interface CrowdsecBlockMeta { source: typeof CROWDSEC_BLOCK_SOURCE | "gate"; category: string; reputation: CrowdsecReputation | null; score: number; behaviors: string[]; ttlSeconds: number; blockedAt: number; } /** Daily CTI usage counter as shown in the admin panel. */ export interface CrowdsecQuotaUsage { /** UTC calendar day the counter belongs to (YYYY-MM-DD). */ date: string; used: number; /** 0 = unlimited. */ quota: number; exhausted: boolean; } /** Value written into the shared block key so the admin UI can label the source. */ export const CROWDSEC_BLOCK_SOURCE = "crowdsec"; const API_TIMEOUT_MS = 10_000; const VERDICT_CACHE_TTL_SECONDS = 3600; const VERDICT_CACHE_TTL_MS = VERDICT_CACHE_TTL_SECONDS * 1000; const LOOKUP_LOCK_TTL_SECONDS = 60; const RATE_LIMIT_BACKOFF_MS = 60_000; const AUTH_BACKOFF_MS = 300_000; const VERDICT_PREFIX = "crowdsec:cti:"; const LOOKUP_LOCK_PREFIX = "crowdsec:lock:"; const LAST_VERIFY_KEY = "crowdsec:last-verify"; const BLOCK_META_PREFIX = "antiddos:block:meta:"; const QUOTA_PREFIX = "crowdsec:usage:"; const QUOTA_KEY_TTL_SECONDS = 48 * 3_600; /** Shared 403/429 pause marker, so every instance respects the backoff. */ const BACKOFF_KEY = "crowdsec:backoff-until"; /** Short-window block burst counter: crowdsec:burst:recent (ZSET of timestamps). */ const BURST_PREFIX = "crowdsec:burst:"; const BURST_WINDOW_SECONDS = 300; /** In-process verdict cache cap so a flood of distinct IPs cannot grow it forever. */ const MEMORY_VERDICT_CACHE_MAX = 2_000; /** Warn at this fraction of the daily quota, once per day. */ const QUOTA_WARN_RATIO = 0.8; /** Well-known, community-safe address used by the admin "verify" button. */ const PROBE_IP = "1.1.1.1"; export function getCrowdsecApiConfig(): CrowdsecApiConfig { return { baseUrl: env.CROWDSEC_CTI_BASE_URL || "https://cti.api.crowdsec.net/v2", apiKey: env.CROWDSEC_API_KEY?.trim() || null, }; } /** True when a CTI API key is present so the gate may call the API. */ export function crowdsecEnabled(): boolean { return Boolean(getCrowdsecApiConfig().apiKey); } interface CrowdsecScore { aggressiveness?: number; threat?: number; trust?: number; anomaly?: number; total?: number; } interface CrowdsecSmokeItem { ip?: string; reputation?: string; confidence?: string; scores?: { overall?: CrowdsecScore }; classifications?: { false_positives?: unknown[] }; behaviors?: { name?: string }[]; } async function crowdsecRequest( path: string, init: { method?: "GET" | "POST"; body?: unknown } = {}, ): Promise { const config = getCrowdsecApiConfig(); if (!config.apiKey) { throw new CrowdsecApiError("CROWDSEC_API_KEY is not configured"); } const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), API_TIMEOUT_MS); try { return await fetch(`${config.baseUrl}${path}`, { method: init.method ?? "GET", headers: { "x-api-key": config.apiKey, Accept: "application/json", "Content-Type": "application/json", }, body: init.body === undefined ? undefined : JSON.stringify(init.body), signal: controller.signal, cache: "no-store", }); } finally { clearTimeout(timer); } } async function errorDetail(response: Response): Promise { try { const body = (await response.json()) as { message?: string }; return body.message ?? `HTTP ${response.status}`; } catch { return `HTTP ${response.status}`; } } function toNumber(value: unknown): number { const n = Number(value); return Number.isFinite(n) ? n : 0; } function parseVerdict(ip: string, item: CrowdsecSmokeItem): CrowdsecVerdict { const overall = item.scores?.overall; const falsePositives = item.classifications?.false_positives ?? []; return { ip, reputation: (item.reputation as CrowdsecReputation | undefined) ?? null, score: toNumber(overall?.total), aggressiveness: toNumber(overall?.aggressiveness), confidence: item.confidence ?? null, behaviors: (item.behaviors ?? []) .map((behavior) => behavior?.name) .filter((name): name is string => Boolean(name)), // CrowdSec: "Any IP with false_positives tags shouldn't be considered // as malicious" — this veto always wins over reputation and score. falsePositive: falsePositives.length > 0, checkedAt: Date.now(), }; } /** * Resolve a cached CTI verdict into a block/no-block decision against the * admin-configurable score threshold. `malicious` is always blocked; `safe` * and `benign` never are; everything else follows the 0-5 score threshold * (with score 0 = "unknown" never blocking, even at threshold 0). */ export function verdictIsMalicious( verdict: CrowdsecVerdict, threshold: number, ): boolean { if (verdict.falsePositive) return false; if (verdict.reputation === "malicious") return true; if (verdict.reputation === "safe" || verdict.reputation === "benign") { return false; } const effective = Math.min(5, Math.max(0, threshold)); return verdict.score >= effective && verdict.score >= 1; } // --- Verdict cache (Redis backed, in-process fallback) --- const memoryVerdicts = new Map(); /** * Insert/refresh an in-process verdict while keeping the cache bounded: Map * iteration order is insertion order, so the oldest (leftmost) entry is * dropped first and re-inserted entries are refreshed to the back. */ function rememberVerdict(verdict: CrowdsecVerdict): void { memoryVerdicts.delete(verdict.ip); memoryVerdicts.set(verdict.ip, verdict); while (memoryVerdicts.size > MEMORY_VERDICT_CACHE_MAX) { const oldest = memoryVerdicts.keys().next(); if (oldest.done) break; memoryVerdicts.delete(oldest.value); } } /** Test hook only — reports the bounded in-process cache size. */ export function getMemoryVerdictCacheSize(): number { return memoryVerdicts.size; } function verdictKey(ip: string): string { return `${VERDICT_PREFIX}${ip}`; } async function readVerdictCache(ip: string): Promise { const cached = memoryVerdicts.get(ip); if (cached && Date.now() - cached.checkedAt < VERDICT_CACHE_TTL_MS) { return cached; } if (redis) { try { const raw = await redis.get(verdictKey(ip)); if (raw) { const parsed = JSON.parse(raw) as CrowdsecVerdict; rememberVerdict(parsed); return parsed; } } catch { // fall through to a cache miss — the API call below is the fallback. } } return null; } async function writeVerdictCache(verdict: CrowdsecVerdict): Promise { rememberVerdict(verdict); if (redis) { try { await redis.set( verdictKey(verdict.ip), JSON.stringify(verdict), "EX", VERDICT_CACHE_TTL_SECONDS, ); } catch { // cache is best-effort — a miss only costs one extra API call later. } } } /** Cross-instance dedupe so a cold-cache flood costs one lookup, not N. */ async function acquireLookupLock(ip: string): Promise { if (!redis) return true; try { const acquired = await redis.set( `${LOOKUP_LOCK_PREFIX}${ip}`, "1", "EX", LOOKUP_LOCK_TTL_SECONDS, "NX", ); return acquired === "OK"; } catch { // Redis hiccup — allow the lookup; the verdict cache still dedupes. return true; } } let backoffUntil = 0; let quotaWarnedDate: string | null = null; let quotaExhaustedDate: string | null = null; /** * Next moment (epoch ms) the CTI API may be called again — the max of the * in-process view and the shared Redis marker so every instance respects a * backoff discovered by any of them. Redis is only read when the local view is * not already active, keeping the hot path cheap. */ async function getBackoffUntil(): Promise { if (Date.now() < backoffUntil) return backoffUntil; if (redis) { try { const raw = await redis.get(BACKOFF_KEY); const shared = Number(raw ?? 0); if (Number.isFinite(shared) && shared > backoffUntil) { backoffUntil = shared; } } catch { // Redis hiccup — local view is enough } } return backoffUntil; } async function setBackoff(ms: number): Promise { const until = Date.now() + ms; backoffUntil = until; if (redis) { try { // EX rounds up so the marker outlives the wait it encodes, plus a // second of slack for the read path. await redis.set( BACKOFF_KEY, String(until), "EX", Math.ceil(ms / 1000) + 1, ); } catch { // local view still protects this instance } } } function quotaDate(): string { return new Date().toISOString().slice(0, 10); } function quotaKey(date: string): string { return `${QUOTA_PREFIX}${date}`; } /** * Today's configured ceiling. Coerced because tests run with * SKIP_ENV_VALIDATION (raw process.env strings, no zod defaults) while * production gets a parsed number. 0 (or unset) = unlimited. */ function dailyQuota(): number { const raw = Number(env.CROWDSEC_CTI_DAILY_QUOTA ?? 0); return Number.isFinite(raw) && raw > 0 ? raw : 0; } /** * Today's CTI usage against the configured daily ceiling. Counted in Redis so * every instance shares one budget. */ export async function getCrowdsecQuotaUsage(): Promise { const date = quotaDate(); const quota = dailyQuota(); let used = 0; if (redis) { try { used = Number((await redis.get(quotaKey(date))) ?? 0); if (!Number.isFinite(used)) used = 0; } catch { // counter unavailable — report zero rather than blocking the admin } } return { date, used, quota, exhausted: quota > 0 && used >= quota }; } /** * Reserve one API call against today's quota. Atomic: the counter is INCR'd * BEFORE the call and compared to the ceiling, so concurrent instances can * never slip calls past the budget; a reserve that overshoots rolls itself * back. Returns false once the budget is spent (and raises an ops alert). */ async function reserveQuota(): Promise { const quota = dailyQuota(); if (quota <= 0) return true; if (!redis) return true; // no shared counter → unlimited best-effort const date = quotaDate(); const key = quotaKey(date); try { const used = await redis.incr(key); await redis.expire(key, QUOTA_KEY_TTL_SECONDS); if (used > quota) { // Concurrent reserves nudged us past the ceiling — give the slot // back and refuse: the budget would be spent the very next call // anyway, so stopping here is both safe and quota-exact. await redis.decr(key); quotaExhaustedDate = date; logger.warn( "[crowdsec-api] CTI daily quota exhausted — pausing lookups until tomorrow", { quota }, ); void raiseCrowdsecAlert("quota", { type: "ddos", severity: "warning", message: `CrowdSec reputation quota exhausted for today (${used} of ${quota} enrichment calls) — lookups are paused until tomorrow.`, context: { used, quota, date }, }); return false; } if (used >= quota * QUOTA_WARN_RATIO && quotaWarnedDate !== date) { quotaWarnedDate = date; logger.warn("[crowdsec-api] CTI daily quota nearing its limit", { used, quota, }); } if (quotaExhaustedDate) { // A reserve just succeeded after an exhaustion day (counter was // reset or the calendar rolled over) — say so, once per cooldown. void raiseCrowdsecAlert("quota-restored", { type: "ddos", severity: "info", message: `CrowdSec reputation quota is available again (${used} of ${quota} used today) — lookups resumed.`, context: { used, quota, date }, }); quotaExhaustedDate = null; } return true; } catch { // Redis hiccup at a moment we could not count — allow the call rather // than break the gate; the verdict cache still limits frequency. return true; } } /** Block burst threshold from env, defensively coerced (falls back to 10). */ function dailyBlockBurstThreshold(): number { const raw = Number(env.CROWDSEC_ALERT_BLOCK_BURST ?? 10); return Number.isFinite(raw) && raw > 0 ? Math.floor(raw) : 10; } /** * A burst of new blocks is usually an automated attack wave. Track block * timestamps in a rolling window (Redis sorted set, 5 minutes) so a burst that * straddles a bucket boundary is still counted together, and alert once per * cooldown window when the count crosses CROWDSEC_ALERT_BLOCK_BURST. * Fire-and-forget. */ async function trackBlockBurst(): Promise { if (!redis) return; const now = Date.now(); const key = `${BURST_PREFIX}recent`; const threshold = dailyBlockBurstThreshold(); try { await redis.zadd(key, now, randomUUID()); await redis.zremrangebyscore(key, 0, now - BURST_WINDOW_SECONDS * 1000); const count = await redis.zcard(key); await redis.expire(key, BURST_WINDOW_SECONDS * 2); if (count >= threshold) { void raiseCrowdsecAlert("block-burst", { type: "ddos", severity: "warning", message: `Anti-DDoS auto-block created ${count} blocks in the last ${BURST_WINDOW_SECONDS / 60} minutes — likely an automated attack wave.`, context: { blocks: count, windowSeconds: BURST_WINDOW_SECONDS, threshold, }, }); } } catch { // alert is best-effort — never break the block path } } /** * Community reputation verdict for an IP, from cache when possible. Returns * null when the API is not configured, the lookup failed, the API is in * backoff, or today's quota is spent — never throws, so it is safe on the * gate's hot path. */ export async function lookupCrowdsecVerdict( ip: string, ): Promise { if (!crowdsecEnabled()) return null; if (!ip || ip === UNKNOWN_CLIENT_IP) return null; if (Date.now() < (await getBackoffUntil())) return null; const cached = await readVerdictCache(ip); if (cached) return cached; if (!(await acquireLookupLock(ip))) { // Another instance is mid-lookup for this IP; skip rather than // double-spend API quota on the same address. return null; } try { // Re-read after claiming the lock — a concurrent instance may have // filled the cache while we were acquiring it. const raced = await readVerdictCache(ip); if (raced) return raced; // Cache miss costs a paid call — reserve against today's quota first. if (!(await reserveQuota())) { logger.warn( "[crowdsec-api] CTI daily quota exhausted — pausing lookups until tomorrow", { quota: dailyQuota() }, ); return null; } const response = await crowdsecRequest(`/smoke/${encodeURIComponent(ip)}`); // The enrichment call happened — count it for the daily histogram, // regardless of whether the verdict was positive, negative, or n/a. void bumpCrowdsecStat("lookups"); if (response.status === 404) { // Unknown to the community — cache the negative result so a clean // repeat offender never costs another API call this hour. const verdict = parseVerdict(ip, {}); await writeVerdictCache(verdict); return verdict; } if (response.status === 403) { const detail = await errorDetail(response); await setBackoff(AUTH_BACKOFF_MS); // A rejected key paralyses the whole reputation pipeline — surface // it once (cooldown-gated) so rotating the key is an ops priority. void raiseCrowdsecAlert("cti-auth", { type: "ddos", severity: "critical", message: `CrowdSec CTI API key rejected (HTTP 403): ${detail} — reputation lookups are paused for ${Math.round(AUTH_BACKOFF_MS / 60_000)} minutes. Rotate CROWDSEC_API_KEY.`, context: { status: 403, detail, backoffMs: AUTH_BACKOFF_MS }, }); throw new CrowdsecApiError( `CrowdSec API key rejected (HTTP 403): ${detail}`, ); } if (response.status === 429) { await setBackoff(RATE_LIMIT_BACKOFF_MS); logger.warn("[crowdsec-api] CTI API rate limit hit — backing off", { ip, backoffMs: RATE_LIMIT_BACKOFF_MS, }); void raiseCrowdsecAlert("cti-ratelimit", { type: "ddos", severity: "warning", message: `CrowdSec CTI API rate limited — all instances backed off for ${Math.round(RATE_LIMIT_BACKOFF_MS / 1000)}s.`, context: { status: 429, backoffMs: RATE_LIMIT_BACKOFF_MS }, }); return null; } if (!response.ok) { throw new CrowdsecApiError( `CrowdSec CTI API error (HTTP ${response.status}): ${await errorDetail(response)}`, ); } const item = (await response.json()) as CrowdsecSmokeItem; const verdict = parseVerdict(ip, item); await writeVerdictCache(verdict); return verdict; } catch (error) { logger.error("[crowdsec-api] CTI lookup failed", { ip, err: error }); return null; } } /** * Consult the CrowdSec community reputation of an IP that just tripped a rate * bucket and hard-block it when the community flags it as known-bad. Safe to * call fire-and-forget from the hot path: it is never awaited by the caller, * does nothing when the API is not configured or the runtime toggle is off, * never shortens an already-active block, and never lets an API failure * surface to the request. */ export async function maybeAutoBlockCrowdsec(input: { ip: string; category: string; ttlSeconds: number; scoreThreshold: number; enabled: boolean; }): Promise { const { ip, category, ttlSeconds, scoreThreshold, enabled } = input; if (!enabled) return; if (!crowdsecEnabled()) return; if (!ip || ip === UNKNOWN_CLIENT_IP) return; // The gate only ever reads its block key through shared Redis — without it // there is nowhere durable to record the block. if (!redis) return; if (Date.now() < (await getBackoffUntil())) return; try { const verdict = await lookupCrowdsecVerdict(ip); if (!verdict || !verdictIsMalicious(verdict, scoreThreshold)) return; const blockKey = `antiddos:block:${ip}`; const existingTtl = await redis.pttl(blockKey); // -2 = no key, -1 = no expiry; both fall through and get overwritten // with the CrowdSec TTL. An equal or longer block is left untouched. if (existingTtl >= ttlSeconds * 1000) return; await redis.set(blockKey, CROWDSEC_BLOCK_SOURCE, "EX", ttlSeconds); // Record why this block exists so the admin panel can surface the // community reasoning (reputation, score, behaviors) for the IP. const meta: CrowdsecBlockMeta = { source: CROWDSEC_BLOCK_SOURCE, category, reputation: verdict.reputation, score: verdict.score, behaviors: verdict.behaviors, ttlSeconds, blockedAt: Date.now(), }; try { await redis.set( `${BLOCK_META_PREFIX}${ip}`, JSON.stringify(meta), "EX", ttlSeconds, ); } catch { // metadata is display sugar only — the block itself is already set. } // CrowdSec only records the block in the gate's own key. It never // creates Cloudflare edge rules — the gate's own escalation logic is // the only place that may mirror a host-level block to the edge. logger.info( "[crowdsec-api] Automatic IP block created from community reputation", { ip, category, ttlSeconds, reputation: verdict.reputation, score: verdict.score, behaviors: verdict.behaviors, }, ); // Opt-in community signal push (CAPI), fire-and-forget: never awaited, // never throws, and internally deduped per IP. void reportCrowdsecSignal({ ip, category, ttlSeconds, verdict, meta }); // Daily histogram + burst detection (cooldown-gated ops alert). void bumpCrowdsecStat("blocks"); void bumpCrowdsecBreakdownStat("category", category); if (verdict.reputation) { void bumpCrowdsecBreakdownStat("reputation", verdict.reputation); } void trackBlockBurst(); } catch (error) { logger.error("[crowdsec-api] Automatic IP block failed", { ip, err: error, }); } } let lastVerifyMemory: CrowdsecConnectionStatus | null = null; /** Validate that the configured key can query the CTI (Enrichment) API. */ export async function verifyCrowdsecConnection(): Promise { const config = getCrowdsecApiConfig(); if (!config.apiKey) { return { ok: false, message: "CROWDSEC_API_KEY is not configured", at: Date.now(), }; } try { const response = await crowdsecRequest(`/smoke/${PROBE_IP}`); if (response.ok) { const item = (await response .json() .catch(() => null)) as CrowdsecSmokeItem | null; const reputation = item?.reputation ? ` (reputation ${item.reputation})` : ""; return { ok: true, message: `CTI key accepted — probed ${PROBE_IP}${reputation}`, at: Date.now(), }; } if (response.status === 403) { return { ok: false, message: `API key rejected: ${await errorDetail(response)}`, at: Date.now(), }; } if (response.status === 429) { return { ok: false, message: "CTI API rate limit reached — try again shortly", at: Date.now(), }; } return { ok: false, message: `CrowdSec CTI API error (HTTP ${response.status}): ${await errorDetail(response)}`, at: Date.now(), }; } catch (error) { return { ok: false, message: error instanceof Error ? error.message : "CrowdSec API unreachable", at: Date.now(), }; } } export async function getLastCrowdsecVerify(): Promise { if (redis) { try { const raw = await redis.get(LAST_VERIFY_KEY); if (raw) return JSON.parse(raw) as CrowdsecConnectionStatus; } catch { // fall back to the in-process view } } return lastVerifyMemory; } export async function setLastCrowdsecVerify( status: CrowdsecConnectionStatus, ): Promise { lastVerifyMemory = status; if (redis) { try { await redis.set(LAST_VERIFY_KEY, JSON.stringify(status)); } catch { // redis unavailable — in-process view is enough } } } /** Why an IP is blocked by CrowdSec, when known. */ export async function getCrowdsecBlockMeta( ip: string, ): Promise { if (!redis) return null; try { const raw = await redis.get(`${BLOCK_META_PREFIX}${ip}`); if (!raw) return null; return JSON.parse(raw) as CrowdsecBlockMeta; } catch { return null; } } /** Test hook only — drop in-memory state between unit runs. */ export function resetCrowdsecCache(): void { memoryVerdicts.clear(); backoffUntil = 0; quotaWarnedDate = null; quotaExhaustedDate = null; lastVerifyMemory = null; }