Phy Rate Limit Audit

Missing rate limit detector for API codebases. Scans Express/Koa/Fastify/NestJS, FastAPI/Flask/Django, Go (gin/chi/echo), Spring Boot, and Ruby (Rack) for 10 classes of unprotected endpoints — auth routes open to brute force, payment endpoints with no throttle, email/SMS senders without limit (cost amplification), search routes with no query throttle, retry logic without backoff, admin endpoints without per-IP limiting, and more. Zero external dependencies. Works on directories or single files. CI fail-gate included.

phy-rate-limit-audit — Missing Rate Limit Detector

Scans your API source code for 10 classes of missing rate limiting that lead to brute-force attacks, cost amplification (SMS/email bombs), DoS via expensive queries, and retry storms.

Maps to OWASP API Security Top 10 2023: API4 — Unrestricted Resource Consumption.

Quick Start

bash

# Scan a directory
python rate_limit_audit.py ./src

# Single file
python rate_limit_audit.py src/routes/auth.js

# CI mode — exit 1 on CRITICAL or HIGH
python rate_limit_audit.py ./src --ci

# Only show HIGH and above
python rate_limit_audit.py ./src --only-severity HIGH

# Verbose: show which lines triggered each finding
python rate_limit_audit.py ./src --verbose

The 10 Checks

ID	Severity	Check	Attack Vector
RL001	HIGH	No rate-limit library imported in project	Global unprotected
RL002	CRITICAL	Auth endpoint (/login /register /forgot) without per-route limit	Brute force / credential stuffing
RL003	CRITICAL	Email/SMS send without rate limit guard	Cost amplification ($$$)
RL004	HIGH	Payment/billing endpoint without rate limiting	Financial DoS
RL005	HIGH	Search/list endpoint without throttle	Expensive query DoS
RL006	HIGH	Admin endpoint without per-IP rate limiting	Privilege escalation via brute force
RL007	MEDIUM	Retry loop without exponential backoff	Retry storm / self-DoS
RL008	MEDIUM	File processing endpoint without queue/semaphore	CPU/memory exhaustion
RL009	MEDIUM	External API call in request handler without circuit breaker	Cascading failure
RL010	LOW	Rate limit configured but threshold too permissive on auth routes	Slow brute force

RL001 — No Rate-Limit Library in Project

Scans all source files for imports of known rate-limiting libraries. If none are found across the entire codebase, every endpoint is unprotected.

Detected libraries:

• JS/TS: express-rate-limit, rate-limiter-flexible, koa-ratelimit, @nestjs/throttler, fastify-rate-limit, express-slow-down, bottleneck, p-throttle
• Python: slowapi, flask-limiter, django-ratelimit, limits, ratelimit
• Go: github.com/ulule/limiter, golang.org/x/time/rate, github.com/didip/tollbooth
• Ruby: rack-attack, rack/throttle
• Java/Spring: @RateLimiter, RateLimiterRegistry, resilience4j.ratelimiter

RL002 — Auth Endpoints Without Per-Route Limit (CRITICAL)

Detects route definitions for authentication paths (/login, /signin, /auth, /register, /signup, /forgot-password, /reset-password, /verify-otp, /token, /oauth) and checks whether a rate-limit guard appears within ±25 lines of the route handler.

Guard patterns recognized: rateLimit(, limiter(, @RateLimit, @limiter.limit, slowDown(, throttle(, rateLimiter, checkRateLimit, redis.incr

RL003 — Email/SMS Sends Without Rate Limit (CRITICAL)

Finds calls to email/SMS providers without a nearby guard:

• Email: sendmail(, send_email(, ses.sendEmail(, ses.send_email(, sg.send(, resend.emails.send(, nodemailer, smtplib.SMTP
• SMS: twilio.messages.create(, sns.publish(, send_sms(, nexmo.message.sendSms(

A compromised or public endpoint calling these without rate limiting = unlimited SMS/email charges.

RL004 — Payment Endpoints Without Rate Limiting (HIGH)

Finds routes matching /pay, /charge, /invoice, /subscribe, /billing, /checkout, /purchase, /order and checks for rate-limit guards nearby. Unprotected payment endpoints allow attackers to enumerate card validity at scale.

RL005 — Search/List Endpoints Without Throttle (HIGH)

Detects routes matching /search, /query, /find, /filter, /list + ?q= or ?query= patterns. Search routes often trigger full-text scans or ML inference — unprotected, a single attacker can saturate the database.

RL006 — Admin Endpoints Without Per-IP Limiting (HIGH)

Admin paths (/admin, /management, /internal, /dashboard) without a rate-limit guard allow brute-forcing admin credentials without lockout.

RL007 — Retry Without Exponential Backoff (MEDIUM)

Detects retry loops (while retries, for attempt, for i in range, retry() without exponential backoff signals (Math.pow, 2 **, backoff, sleep(2 *, jitter, exponential). Retry without backoff causes retry storms that amplify failures into outages.

RL008 — Synchronous File Processing (MEDIUM)

Finds file upload handlers (multipart, req.file, request.files, UploadedFile) that process the file in-request without a queue/semaphore (celery, bull, queue, asyncio.Semaphore, p-limit, worker). Large file processing inline blocks threads and can exhaust memory.

RL009 — External API Call Without Circuit Breaker (MEDIUM)

Detects outbound HTTP calls (fetch(, requests.get(, http.Get(, axios.get(, RestTemplate.getForObject) in request handlers without circuit-breaker patterns (circuit, breaker, hystrix, resilience4j, pybreaker, tenacity, timeout). If the external service is slow, every request hangs.

RL010 — Permissive Rate Limit on Auth Routes (LOW)

If express-rate-limit or slowapi config is found with max: > 20 or times: > 20 on paths matching auth patterns, flags as too permissive. 20 req/window is already enough for a slow brute force on a 6-digit PIN.

Sample Output

text

============================================================
  Rate Limit Audit — src/
  Files scanned: 47  |  Files flagged: 8
============================================================

── CRITICAL (2) ────────────────────────────────────────────
🔴 RL002 [CRITICAL] src/routes/auth.js:34
   Auth endpoint POST /login has no rate-limit guard within 25 lines.
   Attack: Credential stuffing / brute force — 100K attempts/hour unchecked.
   CWE: CWE-307: Improper Restriction of Excessive Authentication Attempts
   Fix: app.post('/login', rateLimit({ windowMs: 15*60*1000, max: 5 }), handler)

🔴 RL003 [CRITICAL] src/services/notifications.js:89
   Email send (ses.sendEmail) with no rate-limit guard in caller chain.
   Attack: Attackers trigger unlimited email sends → AWS SES bill amplification.
   CWE: CWE-400: Uncontrolled Resource Consumption
   Fix: Add per-user rate limit before any email-sending endpoint.

── HIGH (2) ────────────────────────────────────────────────
🟠 RL004 [HIGH] src/routes/payments.js:12
   Payment endpoint POST /checkout has no rate-limit guard.
   Attack: Card enumeration — test stolen cards at scale without detection.
   Fix: rateLimit({ windowMs: 60000, max: 3, keyGenerator: req => req.user.id })

🟠 RL005 [HIGH] src/routes/search.js:8
   Search endpoint GET /search has no throttle guard.
   Attack: Full-text search DoS — single attacker saturates Elasticsearch.
   Fix: Add rateLimit({ windowMs: 60000, max: 30 }) before search handler.

── MEDIUM (2) ──────────────────────────────────────────────
🟡 RL007 [MEDIUM] src/utils/retry.js:22
   Retry loop detected without exponential backoff (no Math.pow/backoff/jitter).
   Attack: Retry storm amplifies transient failures into sustained outage.
   Fix: Add: await sleep(Math.min(1000 * 2 ** attempt + Math.random() * 1000, 30000))

🟡 RL009 [MEDIUM] src/handlers/weather.js:45
   External HTTP call (fetch) in request handler without circuit breaker or timeout.
   Attack: Slow external service hangs all Node.js event loop threads.
   Fix: Add timeout: fetch(url, { signal: AbortSignal.timeout(3000) })

── LOW (1) ─────────────────────────────────────────────────
🔵 RL010 [LOW] src/middleware/rateLimits.js:5
   Rate limit on auth route has max: 100 — too permissive for credential endpoint.
   Fix: Use max: 5 per 15-minute window for /login, /register, /forgot-password.

────────────────────────────────────────────────────────────
  Total: 7 findings
  Critical: 2 | High: 2 | Medium: 2 | Low: 1

  ❌ CI GATE FAILED — resolve CRITICAL/HIGH findings before merging.

The Script

python

#!/usr/bin/env python3
"""
phy-rate-limit-audit — Missing Rate Limit Detector
Scans Express/FastAPI/Flask/Django/Go/Spring for unprotected API endpoints.
Zero external dependencies.
"""

import sys
import re
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional


# ─── Data Structures ─────────────────────────────────────────────────────────

@dataclass
class Finding:
    check_id: str
    severity: str      # CRITICAL / HIGH / MEDIUM / LOW
    location: str      # "file.js:34"
    message: str
    attack: str = ""
    cwe: str = ""
    fix: str = ""

    def __str__(self) -> str:
        icon = {"CRITICAL": "🔴", "HIGH": "🟠", "MEDIUM": "🟡", "LOW": "🔵"}.get(self.severity, "⚪")
        parts = [f"{icon} {self.check_id} [{self.severity}] {self.location}"]
        parts.append(f"   {self.message}")
        if self.attack:
            parts.append(f"   Attack: {self.attack}")
        if self.cwe:
            parts.append(f"   CWE: {self.cwe}")
        if self.fix:
            parts.append(f"   Fix: {self.fix}")
        return "\n".join(parts)


@dataclass
class AuditResult:
    scan_root: str
    files_scanned: int = 0
    files_flagged: int = 0
    findings: list = field(default_factory=list)

    @property
    def critical_count(self) -> int:
        return sum(1 for f in self.findings if f.severity == "CRITICAL")

    @property
    def high_count(self) -> int:
        return sum(1 for f in self.findings if f.severity == "HIGH")

    @property
    def medium_count(self) -> int:
        return sum(1 for f in self.findings if f.severity == "MEDIUM")

    @property
    def low_count(self) -> int:
        return sum(1 for f in self.findings if f.severity == "LOW")


# ─── Constants ────────────────────────────────────────────────────────────────

RATE_LIMIT_IMPORTS = re.compile(
    r"(express-rate-limit|rate-limiter-flexible|koa-ratelimit|@nestjs/throttler|"
    r"fastify-rate-limit|express-slow-down|bottleneck|p-throttle|"
    r"slowapi|flask.limiter|flask_limiter|django.ratelimit|django_ratelimit|"
    r"from limits|import ratelimit|"
    r"ulule/limiter|golang\.org/x/time/rate|didip/tollbooth|"
    r"rack.attack|rack/throttle|"
    r"RateLimiterRegistry|@RateLimiter|resilience4j\.ratelimiter)",
    re.IGNORECASE,
)

RATE_LIMIT_GUARD_RE = re.compile(
    r"(rateLimit\(|limiter\(|@RateLimit|@limiter\.limit|slowDown\(|"
    r"throttle\(|rateLimiter|checkRateLimit|redis\.incr|"
    r"Limiter\(|tollbooth|rack\.attack|RateLimiter\.create)",
    re.IGNORECASE,
)

AUTH_ROUTE_RE = re.compile(
    r"""(app|router|Route)\.(post|get|put|patch)\s*\(\s*['"`]"""
    r"""[^'"`]*(login|signin|register|signup|forgot.password|reset.password|"""
    r"""verify.?otp|verify.?email|send.?otp|token|oauth|authenticate)[^'"`]*['"`]""",
    re.IGNORECASE,
)

# Python FastAPI/Flask auth route patterns
PYTHON_AUTH_ROUTE_RE = re.compile(
    r"""@(app|router)\.(post|get|put|patch)\s*\(\s*['"`]"""
    r"""[^'"`]*(login|signin|register|signup|forgot|reset.password|"""
    r"""verify|token|oauth|authenticate)[^'"`]*['"`]""",
    re.IGNORECASE,
)

# Go auth route patterns
GO_AUTH_ROUTE_RE = re.compile(
    r"""(router|r|e|g|mux)\.(POST|GET|PUT|PATCH)\s*\(\s*"[^"]*"""
    r"""(login|signin|register|signup|forgot|reset|verify|token|oauth|auth)[^"]*""",
    re.IGNORECASE,
)

EMAIL_SEND_RE = re.compile(
    r"(sendmail|send_email|ses\.sendEmail|ses\.send_email|sg\.send|"
    r"resend\.emails\.send|nodemailer|smtplib\.SMTP|mail\.send|"
    r"sendgrid|mailgun|postmark\.send)",
    re.IGNORECASE,
)

SMS_SEND_RE = re.compile(
    r"(twilio\.messages\.create|sns\.publish|send_sms|nexmo\.message|"
    r"vonage\.sms|messagebird\.messages\.create|plivo\.send_message)",
    re.IGNORECASE,
)

PAYMENT_ROUTE_RE = re.compile(
    r"""(app|router|Route|router|r)\.(post|get|put|patch)\s*\(\s*['"`]"""
    r"""[^'"`]*(pay|charge|invoice|subscri|billing|checkout|purchase|order)[^'"`]*['"`]""",
    re.IGNORECASE,
)

SEARCH_ROUTE_RE = re.compile(
    r"""(app|router|Route|r|e|g)\.(get|post)\s*\(\s*['"`][^'"`]*(search|query|find|filter|/list)[^'"`]*['"`]""",
    re.IGNORECASE,
)

ADMIN_ROUTE_RE = re.compile(
    r"""(app|router|Route|r|e|g)\.(get|post|put|patch|delete)\s*\(\s*['"`][^'"`]*(admin|management|internal|dashboard)[^'"`]*['"`]""",
    re.IGNORECASE,
)

RETRY_LOOP_RE = re.compile(
    r"(while\s+retries|for\s+attempt|for\s+i\s+in\s+range|retry\(|attempts\s*=|max_retries)",
    re.IGNORECASE,
)

BACKOFF_RE = re.compile(
    r"(Math\.pow|2\s*\*\*\s*attempt|backoff|sleep\(2\s*\*|jitter|exponential|time\.Sleep\(.*\*\s*2)",
    re.IGNORECASE,
)

FILE_UPLOAD_RE = re.compile(
    r"(multipart|req\.file|request\.files|UploadedFile|multer|formidable|busboy)",
    re.IGNORECASE,
)

QUEUE_SEMAPHORE_RE = re.compile(
    r"(celery|bull|queue|asyncio\.Semaphore|p.limit|worker|background|defer|enqueue)",
    re.IGNORECASE,
)

OUTBOUND_HTTP_RE = re.compile(
    r"(fetch\(|requests\.get|requests\.post|http\.Get|http\.Post|"
    r"axios\.(get|post)|RestTemplate|WebClient|urllib\.request|httpx\.(get|post))",
    re.IGNORECASE,
)

CIRCUIT_BREAKER_RE = re.compile(
    r"(circuit|breaker|hystrix|resilience4j|pybreaker|tenacity|"
    r"AbortSignal\.timeout|signal:\s*AbortSignal|timeout:\s*\d)",
    re.IGNORECASE,
)

RATE_LIMIT_MAX_RE = re.compile(
    r"max:\s*(\d+)|times:\s*(\d+)|limit:\s*(\d+)",
    re.IGNORECASE,
)

SUPPORTED_EXTENSIONS = {".js", ".ts", ".jsx", ".tsx", ".py", ".go", ".rb", ".java", ".kt"}

SKIP_DIRS = {"node_modules", ".git", "venv", ".venv", "__pycache__", "dist", "build", ".next", "vendor"}


# ─── Helpers ─────────────────────────────────────────────────────────────────

def get_context_lines(lines: list, idx: int, window: int = 25) -> str:
    start = max(0, idx - window)
    end = min(len(lines), idx + window)
    return "\n".join(lines[start:end])


def has_guard_nearby(lines: list, idx: int, guard_re: re.Pattern, window: int = 25) -> bool:
    ctx = get_context_lines(lines, idx, window)
    return bool(guard_re.search(ctx))


def collect_files(path: str) -> list:
    p = Path(path)
    if p.is_file():
        return [p] if p.suffix in SUPPORTED_EXTENSIONS else []
    files = []
    for f in p.rglob("*"):
        if any(skip in f.parts for skip in SKIP_DIRS):
            continue
        if f.is_file() and f.suffix in SUPPORTED_EXTENSIONS:
            files.append(f)
    return files


# ─── Checks ──────────────────────────────────────────────────────────────────

def check_rl001_no_rate_limit_library(all_files: list, all_contents: dict) -> list:
    """RL001 — No rate-limit library imported anywhere in the project."""
    for content in all_contents.values():
        if RATE_LIMIT_IMPORTS.search(content):
            return []  # Found at least one — skip

    return [Finding(
        check_id="RL001",
        severity="HIGH",
        location="<project>",
        message="No rate-limiting library imported anywhere in the codebase. All endpoints are unprotected.",
        attack="Unlimited requests — brute force, DoS, cost amplification attacks all possible.",
        cwe="CWE-400: Uncontrolled Resource Consumption",
        fix=(
            "JS: npm install express-rate-limit  |  "
            "Python: pip install slowapi (FastAPI) or flask-limiter  |  "
            "Go: go get golang.org/x/time/rate"
        ),
    )]


def check_rl002_auth_no_limit(filepath: str, lines: list) -> list:
    """RL002 — Auth endpoints without per-route rate limiting."""
    findings = []
    for i, line in enumerate(lines):
        match = AUTH_ROUTE_RE.search(line) or PYTHON_AUTH_ROUTE_RE.search(line) or GO_AUTH_ROUTE_RE.search(line)
        if not match:
            continue
        if not has_guard_nearby(lines, i, RATE_LIMIT_GUARD_RE, window=25):
            findings.append(Finding(
                check_id="RL002",
                severity="CRITICAL",
                location=f"{filepath}:{i + 1}",
                message=f"Auth endpoint '{match.group().strip()[:80]}' has no rate-limit guard within 25 lines.",
                attack="Credential stuffing / brute force — 100K+ attempts per hour without detection.",
                cwe="CWE-307: Improper Restriction of Excessive Authentication Attempts",
                fix="rateLimit({ windowMs: 15*60*1000, max: 5, skipSuccessfulRequests: true })",
            ))
    return findings


def check_rl003_email_sms_no_limit(filepath: str, lines: list, all_content: str) -> list:
    """RL003 — Email/SMS sends without rate-limit guard."""
    # Skip if this is a service/helper file — check callers instead
    # But still flag if the send appears directly in a route handler
    findings = []
    for i, line in enumerate(lines):
        if not (EMAIL_SEND_RE.search(line) or SMS_SEND_RE.search(line)):
            continue
        if has_guard_nearby(lines, i, RATE_LIMIT_GUARD_RE, window=30):
            continue
        # Check if this is inside a route handler (heuristic: near app.post/router.post)
        ctx = get_context_lines(lines, i, 40)
        if re.search(r"(app|router)\.(post|get|put|patch)\s*\(", ctx, re.IGNORECASE):
            finding_type = "Email" if EMAIL_SEND_RE.search(line) else "SMS"
            findings.append(Finding(
                check_id="RL003",
                severity="CRITICAL",
                location=f"{filepath}:{i + 1}",
                message=f"{finding_type} send detected in route handler with no rate-limit guard.",
                attack=f"{finding_type} bombing — attacker triggers unlimited sends → cost amplification / account harassment.",
                cwe="CWE-400: Uncontrolled Resource Consumption",
                fix=f"Add per-user rate limit (max 3/{('hour' if finding_type == 'SMS' else 'minute')}) before this endpoint.",
            ))
    return findings


def check_rl004_payment_no_limit(filepath: str, lines: list) -> list:
    """RL004 — Payment/billing endpoints without rate limiting."""
    findings = []
    for i, line in enumerate(lines):
        if not PAYMENT_ROUTE_RE.search(line):
            continue
        if not has_guard_nearby(lines, i, RATE_LIMIT_GUARD_RE, window=25):
            findings.append(Finding(
                check_id="RL004",
                severity="HIGH",
                location=f"{filepath}:{i + 1}",
                message=f"Payment endpoint '{line.strip()[:80]}' has no rate-limit guard.",
                attack="Card enumeration at scale — test thousands of stolen card numbers per minute.",
                cwe="CWE-400",
                fix="rateLimit({ windowMs: 60000, max: 3, keyGenerator: req => req.user?.id || req.ip })",
            ))
    return findings


def check_rl005_search_no_throttle(filepath: str, lines: list) -> list:
    """RL005 — Search/list endpoints without rate limiting."""
    findings = []
    for i, line in enumerate(lines):
        if not SEARCH_ROUTE_RE.search(line):
            continue
        if not has_guard_nearby(lines, i, RATE_LIMIT_GUARD_RE, window=25):
            findings.append(Finding(
                check_id="RL005",
                severity="HIGH",
                location=f"{filepath}:{i + 1}",
                message=f"Search/list endpoint '{line.strip()[:80]}' has no throttle.",
                attack="Full-text search DoS — single attacker triggers parallel expensive queries saturating the database.",
                cwe="CWE-400",
                fix="rateLimit({ windowMs: 60000, max: 30 }) — search endpoints need tighter limits than read endpoints.",
            ))
    return findings


def check_rl006_admin_no_limit(filepath: str, lines: list) -> list:
    """RL006 — Admin endpoints without per-IP rate limiting."""
    findings = []
    for i, line in enumerate(lines):
        if not ADMIN_ROUTE_RE.search(line):
            continue
        if not has_guard_nearby(lines, i, RATE_LIMIT_GUARD_RE, window=25):
            findings.append(Finding(
                check_id="RL006",
                severity="HIGH",
                location=f"{filepath}:{i + 1}",
                message=f"Admin endpoint '{line.strip()[:80]}' has no per-IP rate limiting.",
                attack="Admin brute force — password/2FA code guessing without lockout on highest-privilege endpoint.",
                cwe="CWE-307",
                fix="rateLimit({ windowMs: 15*60*1000, max: 3, keyGenerator: req => req.ip })",
            ))
    return findings


def check_rl007_retry_no_backoff(filepath: str, lines: list) -> list:
    """RL007 — Retry logic without exponential backoff."""
    findings = []
    for i, line in enumerate(lines):
        if not RETRY_LOOP_RE.search(line):
            continue
        ctx = get_context_lines(lines, i, 20)
        if not BACKOFF_RE.search(ctx):
            findings.append(Finding(
                check_id="RL007",
                severity="MEDIUM",
                location=f"{filepath}:{i + 1}",
                message="Retry loop detected without exponential backoff.",
                attack="Retry storm — transient failure triggers immediate retry avalanche, converting brief outages into sustained cascades.",
                cwe="CWE-400",
                fix="await sleep(Math.min(1000 * 2 ** attempt + Math.random() * 1000, 30000))",
            ))
    return findings


def check_rl008_sync_file_processing(filepath: str, lines: list) -> list:
    """RL008 — File upload without background queue/semaphore."""
    findings = []
    for i, line in enumerate(lines):
        if not FILE_UPLOAD_RE.search(line):
            continue
        ctx = get_context_lines(lines, i, 30)
        # Check if any processing-heavy ops appear nearby (sharp, ffmpeg, pandas, pillow, resize)
        if not re.search(r"(sharp|ffmpeg|pandas|PIL|Pillow|resize|compress|transcode|convert)", ctx, re.IGNORECASE):
            continue
        if not QUEUE_SEMAPHORE_RE.search(ctx):
            findings.append(Finding(
                check_id="RL008",
                severity="MEDIUM",
                location=f"{filepath}:{i + 1}",
                message="File processing (image/video/data) handled synchronously in request without queue or semaphore.",
                attack="10 concurrent large file uploads → thread pool exhaustion → total service unavailability.",
                cwe="CWE-400",
                fix="Offload processing to a background worker (Celery/Bull/BullMQ). Use p-limit or asyncio.Semaphore for concurrency cap.",
            ))
    return findings


def check_rl009_outbound_no_circuit(filepath: str, lines: list) -> list:
    """RL009 — External API call in handler without circuit breaker or timeout."""
    findings = []
    for i, line in enumerate(lines):
        if not OUTBOUND_HTTP_RE.search(line):
            continue
        ctx = get_context_lines(lines, i, 20)
        # Only flag if inside a route handler (heuristic)
        if not re.search(r"(app|router|handler|controller|@app\.|@router\.)", ctx, re.IGNORECASE):
            continue
        if not CIRCUIT_BREAKER_RE.search(ctx):
            findings.append(Finding(
                check_id="RL009",
                severity="MEDIUM",
                location=f"{filepath}:{i + 1}",
                message="Outbound HTTP call in request handler without timeout or circuit breaker.",
                attack="External service slowdown → all in-flight requests hang → event loop saturation → total service unavailability.",
                cwe="CWE-400",
                fix="Add timeout: fetch(url, { signal: AbortSignal.timeout(3000) }) and consider circuit breaker (opossum/pybreaker).",
            ))
    return findings


def check_rl010_permissive_limit(filepath: str, lines: list) -> list:
    """RL010 — Rate limit max too high on auth-related config."""
    findings = []
    for i, line in enumerate(lines):
        # Look for rateLimit config near auth path mentions
        max_match = RATE_LIMIT_MAX_RE.search(line)
        if not max_match:
            continue
        # Extract the numeric value
        val_str = max_match.group(1) or max_match.group(2) or max_match.group(3)
        if not val_str:
            continue
        try:
            val = int(val_str)
        except ValueError:
            continue
        if val <= 20:
            continue  # Acceptable limit
        # Check if the surrounding context is for an auth endpoint
        ctx = get_context_lines(lines, i, 20)
        if re.search(r"(login|signin|register|forgot|reset.password|verify|auth|token)", ctx, re.IGNORECASE):
            findings.append(Finding(
                check_id="RL010",
                severity="LOW",
                location=f"{filepath}:{i + 1}",
                message=f"Rate limit max: {val} is too permissive for an auth-related endpoint.",
                attack=f"At {val} requests/window, slow brute force of 4-6 digit PINs completes in hours undetected.",
                cwe="CWE-307",
                fix="Use max: 5 per 15-minute window for /login and /register. Use max: 3 for /forgot-password (SMS/email send).",
            ))
    return findings


# ─── Main Audit ───────────────────────────────────────────────────────────────

def audit(path: str, verbose: bool = False) -> AuditResult:
    result = AuditResult(scan_root=path)
    files = collect_files(path)

    # Load all contents for RL001 (cross-file check)
    all_contents = {}
    for f in files:
        try:
            all_contents[str(f)] = f.read_text(errors="ignore")
        except Exception:
            pass

    result.files_scanned = len(files)

    # RL001 is a global check — run once
    rl001_findings = check_rl001_no_rate_limit_library(files, all_contents)
    result.findings.extend(rl001_findings)

    # Per-file checks
    file_finding_counts = {}
    for f in files:
        content = all_contents.get(str(f), "")
        lines = content.splitlines()
        fp = str(f)

        file_findings = []
        file_findings.extend(check_rl002_auth_no_limit(fp, lines))
        file_findings.extend(check_rl003_email_sms_no_limit(fp, lines, content))
        file_findings.extend(check_rl004_payment_no_limit(fp, lines))
        file_findings.extend(check_rl005_search_no_throttle(fp, lines))
        file_findings.extend(check_rl006_admin_no_limit(fp, lines))
        file_findings.extend(check_rl007_retry_no_backoff(fp, lines))
        file_findings.extend(check_rl008_sync_file_processing(fp, lines))
        file_findings.extend(check_rl009_outbound_no_circuit(fp, lines))
        file_findings.extend(check_rl010_permissive_limit(fp, lines))

        if file_findings:
            result.files_flagged += 1
            file_finding_counts[fp] = len(file_findings)
        result.findings.extend(file_findings)

    return result


def format_report(result: AuditResult, ci_mode: bool = False) -> str:
    lines_out = []
    lines_out.append(f"\n{'='*60}")
    lines_out.append(f"  Rate Limit Audit — {result.scan_root}")
    lines_out.append(f"  Files scanned: {result.files_scanned}  |  Files flagged: {result.files_flagged}")
    lines_out.append(f"{'='*60}")

    if not result.findings:
        lines_out.append("✅ No missing rate limits detected.")
        return "\n".join(lines_out)

    for severity in ("CRITICAL", "HIGH", "MEDIUM", "LOW"):
        sev_findings = [f for f in result.findings if f.severity == severity]
        if sev_findings:
            lines_out.append(f"\n── {severity} ({len(sev_findings)}) {'─'*40}")
            for finding in sev_findings:
                lines_out.append(str(finding))

    lines_out.append(f"\n{'─'*60}")
    lines_out.append(
        f"  Total: {len(result.findings)} findings  |  "
        f"Critical: {result.critical_count}  High: {result.high_count}  "
        f"Medium: {result.medium_count}  Low: {result.low_count}"
    )
    if ci_mode and (result.critical_count > 0 or result.high_count > 0):
        lines_out.append("\n  ❌ CI GATE FAILED — resolve CRITICAL/HIGH findings before merging.")
    return "\n".join(lines_out)


# ─── CLI Entry Point ──────────────────────────────────────────────────────────

def main():
    import argparse

    parser = argparse.ArgumentParser(
        description="phy-rate-limit-audit — Missing Rate Limit Detector",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python rate_limit_audit.py ./src
  python rate_limit_audit.py src/routes/auth.js
  python rate_limit_audit.py ./src --ci
  python rate_limit_audit.py ./src --only-severity HIGH
        """,
    )
    parser.add_argument("path", help="Directory or file to audit")
    parser.add_argument("--ci", action="store_true", help="Exit 1 on CRITICAL or HIGH findings")
    parser.add_argument(
        "--only-severity",
        choices=["CRITICAL", "HIGH", "MEDIUM", "LOW"],
        help="Filter to this severity and above",
    )
    parser.add_argument("--verbose", action="store_true", help="Show extra context for each finding")
    args = parser.parse_args()

    result = audit(args.path, verbose=args.verbose)

    sev_order = ["CRITICAL", "HIGH", "MEDIUM", "LOW"]
    if args.only_severity:
        cutoff = sev_order.index(args.only_severity)
        result.findings = [f for f in result.findings if sev_order.index(f.severity) <= cutoff]

    print(format_report(result, ci_mode=args.ci))

    if args.ci and (result.critical_count > 0 or result.high_count > 0):
        sys.exit(1)


if __name__ == "__main__":
    main()

CI Integration

yaml

# GitHub Actions
- name: Rate Limit Audit
  run: python rate_limit_audit.py ./src --ci

# GitLab CI
rate-limit-security:
  script:
    - python rate_limit_audit.py ./src --only-severity HIGH --ci
  allow_failure: false

False Positive Notes

• RL001 only fires if zero rate-limit libraries are found across the entire project
• RL002/RL004/RL005/RL006 use a ±25 line proximity window — if your middleware is registered elsewhere (e.g., in app.js), these may fire. Suppress with # phy-ignore: RL002 comment on the route line
• RL007 skips files where backoff keywords appear anywhere near the retry loop (window: 20 lines)
• RL008 only fires when file upload AND heavy processing (sharp/ffmpeg/pandas/PIL) appear together without a queue signal
• RL009 uses a heuristic to detect route handlers — may miss controller patterns in heavily abstracted frameworks (e.g., NestJS decorators)

Related Skills

• phy-async-audit — unhandled async errors (errors from rate-limit failures need proper propagation)
• phy-cors-audit — CORS misconfig can bypass IP-based rate limiting
• phy-openapi-sec-audit — checks for missing rate limit headers in OpenAPI response definitions
• phy-security-headers — Retry-After header enforcement

---

Author

Canlah AI — Run performance marketing without breaking your brand.

• GitHub: github.com/PHY041
• All Skills: clawhub.ai/PHY041