amirelrefai commited on
Commit
efdcc61
·
1 Parent(s): a506cc3

feat(ratelimit): implement dual-key in-memory sliding window rate limiter

Browse files
app/core/dependencies.py CHANGED
@@ -54,3 +54,12 @@ def get_memory_service() -> 'MemoryService':
54
  from services.memory_service import MemoryService
55
  return MemoryService(get_llm_service())
56
 
 
 
 
 
 
 
 
 
 
 
54
  from services.memory_service import MemoryService
55
  return MemoryService(get_llm_service())
56
 
57
+
58
+ from core.rate_limiter import DualKeyRateLimitChecker
59
+
60
+ rate_limit_ask = DualKeyRateLimitChecker("ask", default_limit=20, env_var_name="RATE_LIMIT_ASK_PER_MIN")
61
+ rate_limit_upload = DualKeyRateLimitChecker("upload", default_limit=5, env_var_name="RATE_LIMIT_UPLOAD_PER_MIN")
62
+ rate_limit_quiz = DualKeyRateLimitChecker("quiz", default_limit=5, env_var_name="RATE_LIMIT_QUIZ_PER_MIN")
63
+ rate_limit_tools = DualKeyRateLimitChecker("tools", default_limit=20, env_var_name="RATE_LIMIT_TOOLS_PER_MIN")
64
+
65
+
app/core/rate_limiter.py ADDED
@@ -0,0 +1,142 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """Rate Limiter Module for Atlas API.
2
+
3
+ Provides a thread-safe, in-memory sliding-window rate limiter.
4
+ Dual-keyed on client_id AND client IP to prevent:
5
+ 1. Limit bypass via rotating client UUIDs (guarded by IP bucket).
6
+ 2. Shared bucket starvation behind proxies (guarded by client_id bucket).
7
+
8
+ Behind Hugging Face Spaces / reverse proxies, request.client.host points to the
9
+ internal proxy infrastructure. The originating client IP is extracted from the
10
+ first hop of the X-Forwarded-For header.
11
+ """
12
+
13
+ import os
14
+ import threading
15
+ import time
16
+ from typing import Dict, List, Tuple, Optional
17
+ from fastapi import Request, HTTPException, status
18
+ from utils.logger import app_logger
19
+
20
+
21
+ def get_client_ip(request: Request) -> str:
22
+ """
23
+ Extracts the originating client IP address.
24
+ Behind Hugging Face / reverse proxies, request.client.host is the internal proxy IP.
25
+ The first hop in X-Forwarded-For contains the true client IP.
26
+ """
27
+ xff = request.headers.get("x-forwarded-for")
28
+ if xff:
29
+ # First hop is the original client IP before traversing upstream proxies
30
+ return xff.split(",")[0].strip()
31
+ if request.client and request.client.host:
32
+ return request.client.host
33
+ return "127.0.0.1"
34
+
35
+
36
+ class SlidingWindowRateLimiter:
37
+ """Thread-safe sliding-window rate limiter with sub-second precision."""
38
+
39
+ def __init__(self, window_seconds: float = 60.0):
40
+ self.window_seconds = window_seconds
41
+ self._lock = threading.Lock()
42
+ self._hits: Dict[str, List[float]] = {}
43
+
44
+ def check_and_record(
45
+ self,
46
+ key: str,
47
+ limit: int,
48
+ ) -> Tuple[bool, int]:
49
+ """
50
+ Checks if key is within limit. If allowed, records the hit.
51
+ Returns (is_allowed, retry_after_seconds).
52
+ """
53
+ now = time.monotonic()
54
+ cutoff = now - self.window_seconds
55
+
56
+ with self._lock:
57
+ timestamps = self._hits.get(key, [])
58
+ # Evict timestamps outside sliding window
59
+ timestamps = [t for t in timestamps if t > cutoff]
60
+
61
+ if len(timestamps) >= limit:
62
+ # Oldest timestamp in window determines remaining cooldown
63
+ oldest_hit = timestamps[0]
64
+ retry_after = max(1, int(self.window_seconds - (now - oldest_hit) + 0.99))
65
+ self._hits[key] = timestamps
66
+ return False, retry_after
67
+
68
+ timestamps.append(now)
69
+ self._hits[key] = timestamps
70
+ return True, 0
71
+
72
+ def reset(self) -> None:
73
+ """Clear all recorded hits (useful for test isolation)."""
74
+ with self._lock:
75
+ self._hits.clear()
76
+
77
+
78
+ # Global rate limiter instance
79
+ _global_limiter = SlidingWindowRateLimiter(window_seconds=60.0)
80
+
81
+
82
+ def get_rate_limiter() -> SlidingWindowRateLimiter:
83
+ return _global_limiter
84
+
85
+
86
+ class DualKeyRateLimitChecker:
87
+ """
88
+ FastAPI dependency callable that checks rate limits against both
89
+ client_id and originating IP address.
90
+ """
91
+
92
+ def __init__(self, action_name: str, default_limit: int, env_var_name: str):
93
+ self.action_name = action_name
94
+ self.default_limit = default_limit
95
+ self.env_var_name = env_var_name
96
+
97
+ def _get_limit(self) -> int:
98
+ raw = os.getenv(self.env_var_name)
99
+ if raw:
100
+ try:
101
+ val = int(raw.strip())
102
+ if val > 0:
103
+ return val
104
+ except ValueError:
105
+ pass
106
+ return self.default_limit
107
+
108
+ async def __call__(self, request: Request) -> None:
109
+ client_id = request.headers.get("x-client-id", "").strip().lower()
110
+ ip_addr = get_client_ip(request)
111
+ limit = self._get_limit()
112
+
113
+ limiter = get_rate_limiter()
114
+
115
+ # Check client_id bucket if present
116
+ if client_id:
117
+ client_key = f"client:{self.action_name}:{client_id}"
118
+ allowed, retry_after = limiter.check_and_record(client_key, limit)
119
+ if not allowed:
120
+ app_logger.warning(
121
+ f"[RateLimit] Client rate limit exceeded for {client_id} on {self.action_name} "
122
+ f"({limit}/min). IP: {ip_addr}"
123
+ )
124
+ raise HTTPException(
125
+ status_code=status.HTTP_429_TOO_MANY_REQUESTS,
126
+ detail=f"Rate limit exceeded for {self.action_name}. Maximum {limit} requests per minute.",
127
+ headers={"Retry-After": str(retry_after)},
128
+ )
129
+
130
+ # Check IP bucket
131
+ ip_key = f"ip:{self.action_name}:{ip_addr}"
132
+ allowed_ip, retry_after_ip = limiter.check_and_record(ip_key, limit)
133
+ if not allowed_ip:
134
+ app_logger.warning(
135
+ f"[RateLimit] IP rate limit exceeded for {ip_addr} on {self.action_name} "
136
+ f"({limit}/min). Client ID: {client_id}"
137
+ )
138
+ raise HTTPException(
139
+ status_code=status.HTTP_429_TOO_MANY_REQUESTS,
140
+ detail=f"Rate limit exceeded for {self.action_name}. Maximum {limit} requests per minute.",
141
+ headers={"Retry-After": str(retry_after_ip)},
142
+ )
app/routers/chat.py CHANGED
@@ -19,6 +19,7 @@ from core.dependencies import (
19
  get_storage,
20
  get_memory_service,
21
  get_client_id,
 
22
  )
23
  from core.rag_pipeline import VectorlessRAG
24
  from services.llm_service import LLMService
@@ -138,6 +139,7 @@ async def ask_question(
138
  rag_pipeline: VectorlessRAG = Depends(get_rag_pipeline),
139
  memory_service: MemoryService = Depends(get_memory_service),
140
  client_id: str = Depends(get_client_id),
 
141
  ):
142
  # Ensure chat session exists in persistent storage and is owned by client_id
143
  if storage.chat_exists(request.chat_id):
 
19
  get_storage,
20
  get_memory_service,
21
  get_client_id,
22
+ rate_limit_ask,
23
  )
24
  from core.rag_pipeline import VectorlessRAG
25
  from services.llm_service import LLMService
 
139
  rag_pipeline: VectorlessRAG = Depends(get_rag_pipeline),
140
  memory_service: MemoryService = Depends(get_memory_service),
141
  client_id: str = Depends(get_client_id),
142
+ _rl: None = Depends(rate_limit_ask),
143
  ):
144
  # Ensure chat session exists in persistent storage and is owned by client_id
145
  if storage.chat_exists(request.chat_id):
app/routers/documents.py CHANGED
@@ -8,7 +8,13 @@ from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Depends, R
8
  from services.document_service import DocumentService
9
  from services.llm_service import LLMService
10
  from core.storage import MemoryStorage
11
- from core.dependencies import get_document_service, get_llm_service, get_storage, get_client_id
 
 
 
 
 
 
12
  from schemas.schemas import UploadResponse, DocumentStatusResponse
13
  from utils.logger import app_logger
14
  from core.persistent_db import DATA_DIR
@@ -129,6 +135,7 @@ async def upload_pdf(
129
  llm_service: LLMService = Depends(get_llm_service),
130
  storage: MemoryStorage = Depends(get_storage),
131
  client_id: str = Depends(get_client_id),
 
132
  ):
133
  """
134
  Asynchronous PDF upload and indexing endpoint.
 
8
  from services.document_service import DocumentService
9
  from services.llm_service import LLMService
10
  from core.storage import MemoryStorage
11
+ from core.dependencies import (
12
+ get_document_service,
13
+ get_llm_service,
14
+ get_storage,
15
+ get_client_id,
16
+ rate_limit_upload,
17
+ )
18
  from schemas.schemas import UploadResponse, DocumentStatusResponse
19
  from utils.logger import app_logger
20
  from core.persistent_db import DATA_DIR
 
135
  llm_service: LLMService = Depends(get_llm_service),
136
  storage: MemoryStorage = Depends(get_storage),
137
  client_id: str = Depends(get_client_id),
138
+ _rl: None = Depends(rate_limit_upload),
139
  ):
140
  """
141
  Asynchronous PDF upload and indexing endpoint.
app/routers/studio.py CHANGED
@@ -4,7 +4,14 @@ from typing import List, Dict, Any, Optional
4
  from fastapi import APIRouter, Depends, HTTPException, status
5
  from pydantic import BaseModel
6
 
7
- from core.dependencies import get_storage, get_document_service, get_llm_service, get_memory_service, get_client_id
 
 
 
 
 
 
 
8
  from core.storage import MemoryStorage
9
  from services.document_service import DocumentService
10
  from services.llm_service import LLMService
@@ -170,6 +177,7 @@ async def generate_document_quiz(
170
  storage: MemoryStorage = Depends(get_storage),
171
  doc_service: DocumentService = Depends(get_document_service),
172
  llm_service: LLMService = Depends(get_llm_service),
 
173
  ):
174
  """
175
  Generates 5 Multiple Choice Questions and 5 Study Flashcards for a document
 
4
  from fastapi import APIRouter, Depends, HTTPException, status
5
  from pydantic import BaseModel
6
 
7
+ from core.dependencies import (
8
+ get_storage,
9
+ get_document_service,
10
+ get_llm_service,
11
+ get_memory_service,
12
+ get_client_id,
13
+ rate_limit_quiz,
14
+ )
15
  from core.storage import MemoryStorage
16
  from services.document_service import DocumentService
17
  from services.llm_service import LLMService
 
177
  storage: MemoryStorage = Depends(get_storage),
178
  doc_service: DocumentService = Depends(get_document_service),
179
  llm_service: LLMService = Depends(get_llm_service),
180
+ _rl: None = Depends(rate_limit_quiz),
181
  ):
182
  """
183
  Generates 5 Multiple Choice Questions and 5 Study Flashcards for a document
app/routers/tools.py CHANGED
@@ -12,7 +12,7 @@ from typing import Any, Dict, Optional
12
  from fastapi import APIRouter, HTTPException, Depends
13
  from pydantic import BaseModel, Field
14
 
15
- from core.dependencies import get_client_id
16
  from mcp_server.web_search_mcp import execute_web_search
17
  from mcp_server.research_tools import (
18
  fetch_webpage_markdown,
@@ -21,7 +21,11 @@ from mcp_server.research_tools import (
21
  evaluate_math_expression_safe,
22
  )
23
 
24
- router = APIRouter(prefix="/tools", tags=["Tools & MCP"], dependencies=[Depends(get_client_id)])
 
 
 
 
25
 
26
 
27
  class WebSearchRequest(BaseModel):
 
12
  from fastapi import APIRouter, HTTPException, Depends
13
  from pydantic import BaseModel, Field
14
 
15
+ from core.dependencies import get_client_id, rate_limit_tools
16
  from mcp_server.web_search_mcp import execute_web_search
17
  from mcp_server.research_tools import (
18
  fetch_webpage_markdown,
 
21
  evaluate_math_expression_safe,
22
  )
23
 
24
+ router = APIRouter(
25
+ prefix="/tools",
26
+ tags=["Tools & MCP"],
27
+ dependencies=[Depends(get_client_id), Depends(rate_limit_tools)],
28
+ )
29
 
30
 
31
  class WebSearchRequest(BaseModel):
tests/test_rate_limiter.py ADDED
@@ -0,0 +1,110 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import uuid
2
+ import pytest
3
+ from fastapi.testclient import TestClient
4
+
5
+ from app import app
6
+ from core.rate_limiter import get_rate_limiter, get_client_ip
7
+
8
+
9
+ @pytest.fixture(autouse=True)
10
+ def reset_limiter():
11
+ """Reset in-memory rate limiter buckets before and after each test."""
12
+ limiter = get_rate_limiter()
13
+ limiter.reset()
14
+ yield
15
+ limiter.reset()
16
+
17
+
18
+ @pytest.fixture
19
+ def client():
20
+ return TestClient(app)
21
+
22
+
23
+ def test_first_hop_ip_extraction():
24
+ """Verifies that only the first hop of X-Forwarded-For is extracted (ignoring proxy hops)."""
25
+ from starlette.requests import Request
26
+
27
+ scope = {
28
+ "type": "http",
29
+ "headers": [
30
+ (b"x-forwarded-for", b"198.51.100.42, 10.0.0.1, 172.16.0.2"),
31
+ ],
32
+ "client": ("10.0.0.1", 54321),
33
+ }
34
+ req = Request(scope)
35
+ assert get_client_ip(req) == "198.51.100.42"
36
+
37
+
38
+ def test_rate_limiter_blocks_after_limit_and_sets_retry_after(client, monkeypatch):
39
+ """Enforces rate limit and checks HTTP 429 and Retry-After header."""
40
+ monkeypatch.setenv("RATE_LIMIT_TOOLS_PER_MIN", "3")
41
+
42
+ client_id = str(uuid.uuid4())
43
+ headers = {
44
+ "X-Client-ID": client_id,
45
+ "X-Forwarded-For": "203.0.113.10",
46
+ }
47
+
48
+ # 3 allowed requests
49
+ for i in range(3):
50
+ r = client.post("/tools/calculate-math", headers=headers, json={"expression": f"{i} + 1"})
51
+ assert r.status_code == 200
52
+
53
+ # 4th request must be rejected with 429
54
+ r4 = client.post("/tools/calculate-math", headers=headers, json={"expression": "99 + 1"})
55
+ assert r4.status_code == 429
56
+ assert "Rate limit exceeded" in r4.json()["detail"]
57
+ assert "Retry-After" in r4.headers
58
+ retry_after = int(r4.headers["Retry-After"])
59
+ assert retry_after > 0
60
+
61
+
62
+ def test_rotating_uuids_cannot_bypass_rate_limit(client, monkeypatch):
63
+ """
64
+ Rotating client_id UUIDs from the SAME IP must NOT bypass rate limiting.
65
+ The IP-level bucket must catch and throttle the attacker.
66
+ """
67
+ monkeypatch.setenv("RATE_LIMIT_TOOLS_PER_MIN", "3")
68
+ attacker_ip = "198.51.100.99"
69
+
70
+ # Make 3 requests, each with a different UUID but same IP
71
+ for i in range(3):
72
+ headers = {
73
+ "X-Client-ID": str(uuid.uuid4()),
74
+ "X-Forwarded-For": f"{attacker_ip}, 10.0.0.1",
75
+ }
76
+ r = client.post("/tools/calculate-math", headers=headers, json={"expression": f"{i} + 10"})
77
+ assert r.status_code == 200
78
+
79
+ # 4th request with yet another brand-new UUID from same IP must be rejected
80
+ headers_new_uuid = {
81
+ "X-Client-ID": str(uuid.uuid4()),
82
+ "X-Forwarded-For": f"{attacker_ip}, 10.0.0.1",
83
+ }
84
+ r4 = client.post("/tools/calculate-math", headers=headers_new_uuid, json={"expression": "100 + 1"})
85
+ assert r4.status_code == 429
86
+ assert "Rate limit exceeded" in r4.json()["detail"]
87
+
88
+
89
+ def test_distinct_clients_on_distinct_ips_have_independent_limits(client, monkeypatch):
90
+ """Client A hitting limit does not block Client B on a different IP."""
91
+ monkeypatch.setenv("RATE_LIMIT_TOOLS_PER_MIN", "2")
92
+
93
+ headers_a = {
94
+ "X-Client-ID": str(uuid.uuid4()),
95
+ "X-Forwarded-For": "192.0.2.1",
96
+ }
97
+ headers_b = {
98
+ "X-Client-ID": str(uuid.uuid4()),
99
+ "X-Forwarded-For": "192.0.2.2",
100
+ }
101
+
102
+ # Exhaust Client A
103
+ client.post("/tools/calculate-math", headers=headers_a, json={"expression": "1 + 1"})
104
+ client.post("/tools/calculate-math", headers=headers_a, json={"expression": "2 + 2"})
105
+ r_a_blocked = client.post("/tools/calculate-math", headers=headers_a, json={"expression": "3 + 3"})
106
+ assert r_a_blocked.status_code == 429
107
+
108
+ # Client B should still succeed
109
+ r_b = client.post("/tools/calculate-math", headers=headers_b, json={"expression": "4 + 4"})
110
+ assert r_b.status_code == 200