diff --git a/app/config.py b/app/config.py index 7546b86..d29eead 100644 --- a/app/config.py +++ b/app/config.py @@ -87,7 +87,7 @@ class Settings(BaseSettings): hnsw_ef_search: int = 40 vector_index_type: Literal["hnsw", "ivfflat"] = "hnsw" - # Rate Limiting + # Rate Limiting — legacy per-endpoint settings (kept for backward compat) rate_limit_login_max: int = 5 rate_limit_login_window: int = 900 # 15 min rate_limit_reset_max: int = 3 @@ -97,6 +97,16 @@ class Settings(BaseSettings): rate_limit_general_max: int = 300 rate_limit_general_window: int = 60 # 1 min + # Rate Limiting — unified policies for abuse/cost-sensitive endpoints + rate_limit_auth_max: int = 5 # login, password-reset + rate_limit_auth_window: int = 300 # 5 minutes + rate_limit_ai_max: int = 20 # AI/LLM calls + rate_limit_ai_window: int = 60 # 1 minute + rate_limit_upload_max: int = 30 # file uploads + rate_limit_upload_window: int = 60 # 1 minute + rate_limit_webhook_max: int = 100 # incoming webhooks + rate_limit_webhook_window: int = 60 # 1 minute + @property def cors_origin_list(self) -> list[str]: """Parse comma-separated CORS origins into a list.""" diff --git a/app/core/rate_limit.py b/app/core/rate_limit.py index b57c811..2e2eab3 100644 --- a/app/core/rate_limit.py +++ b/app/core/rate_limit.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +from enum import Enum from fastapi import HTTPException, Request, status from starlette.middleware.base import BaseHTTPMiddleware @@ -13,6 +14,47 @@ from app.core.auth import get_redis logger = logging.getLogger(__name__) +class RateLimitPolicy(Enum): + """Named rate-limit policies for abuse- and cost-sensitive endpoints. + + Each policy maps to a pair of Settings fields: + ``rate_limit__max`` and ``rate_limit__window``. + """ + + AUTH = "auth" + AI = "ai" + UPLOAD = "upload" + WEBHOOK = "webhook" + + @property + def _max_field(self) -> str: + return f"rate_limit_{self.value}_max" + + @property + def _window_field(self) -> str: + return f"rate_limit_{self.value}_window" + + def limits(self) -> tuple[int, int]: + """Return ``(max_attempts, window_seconds)`` from current settings.""" + from app.config import get_settings + + s = get_settings() + return getattr(s, self._max_field), getattr(s, self._window_field) + + +async def check_rate_limit_policy( + redis_key: str, + policy: RateLimitPolicy, +) -> None: + """Check rate limit using a named :class:`RateLimitPolicy`. + + Reads ``max_attempts`` and ``window_seconds`` from application settings + and delegates to :func:`check_rate_limit`. + """ + max_attempts, window_seconds = policy.limits() + await check_rate_limit(redis_key, max_attempts, window_seconds) + + async def check_rate_limit( redis_key: str, max_attempts: int, diff --git a/app/plugins/builtins/ai_assistant/routes.py b/app/plugins/builtins/ai_assistant/routes.py index d793921..076ecb0 100644 --- a/app/plugins/builtins/ai_assistant/routes.py +++ b/app/plugins/builtins/ai_assistant/routes.py @@ -7,7 +7,7 @@ from __future__ import annotations import uuid import aiofiles -from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile +from fastapi import APIRouter, Depends, HTTPException, Query, Request, UploadFile from fastapi.responses import FileResponse, StreamingResponse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -556,11 +556,20 @@ async def list_messages( async def chat_stream( session_id: str, data: ChatSendRequest, + request: Request, current_user: dict = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) + + # Rate limit — AI policy (cost-sensitive LLM call) + from app.core.rate_limit import RateLimitPolicy, check_rate_limit_policy + await check_rate_limit_policy( + f"rate:ai:chat:{tenant_id}:{user_id}", + RateLimitPolicy.AI, + ) + await set_tenant_context(db, tenant_id) session = await get_session_by_id(db, uuid.UUID(session_id), user_id, tenant_id) if not session: diff --git a/app/plugins/builtins/ai_proactive/services.py b/app/plugins/builtins/ai_proactive/services.py index dce8195..47f84e7 100644 --- a/app/plugins/builtins/ai_proactive/services.py +++ b/app/plugins/builtins/ai_proactive/services.py @@ -75,19 +75,21 @@ async def push_suggestion(user_id: str, suggestion: dict[str, Any]) -> None: async def is_rate_limited( tenant_id: uuid.UUID, user_id: uuid.UUID, rate_limit_seconds: int ) -> bool: - """Check if user is rate-limited using Redis. + """Check if user is rate-limited via central check_rate_limit(). - Returns True if rate-limited (key exists), False otherwise. - Sets a key with TTL = rate_limit_seconds on first call. + Delegates to ``app.core.rate_limit.check_rate_limit`` with ``max_attempts=1`` + and ``window_seconds=rate_limit_seconds``. + Returns ``True`` if rate-limited, ``False`` if allowed. """ + from app.core.rate_limit import check_rate_limit + from fastapi import HTTPException + + redis_key = f"rate:ai_proactive:{tenant_id}:{user_id}" try: - r = get_cache() - key = f"ai_proactive:rate:{user_id}" - existing = await r.get(key) - if existing is not None: - return True - await r.setex(key, rate_limit_seconds, "1") + await check_rate_limit(redis_key, max_attempts=1, window_seconds=rate_limit_seconds) return False + except HTTPException: + return True except Exception: logger.exception("Rate-limit check failed, allowing request") return False diff --git a/app/plugins/builtins/automation/agent_runner.py b/app/plugins/builtins/automation/agent_runner.py index f3eba19..2182012 100644 --- a/app/plugins/builtins/automation/agent_runner.py +++ b/app/plugins/builtins/automation/agent_runner.py @@ -61,6 +61,10 @@ async def run_agent( return {"error": "Agent is inactive", "status": "skipped"} # ── Safety Check 1: Rate Limit ── + # Uses DB-based counting (AgentRun rows in last hour) rather than + # check_rate_limit() because this counts actual executions per agent, + # not just attempts. This is more accurate for per-agent execution caps + # and respects the agent-specific max_executions_per_hour setting. if agent.max_executions_per_hour: async with factory() as db: one_hour_ago = datetime.now(UTC) diff --git a/app/plugins/builtins/dms/routes.py b/app/plugins/builtins/dms/routes.py index 214e6a5..cabf739 100644 --- a/app/plugins/builtins/dms/routes.py +++ b/app/plugins/builtins/dms/routes.py @@ -494,6 +494,13 @@ async def upload_file( user_id = uuid.UUID(current_user["user_id"]) fid = _parse_uuid(folder_id, "folder_id") if folder_id else None + # Rate limit — UPLOAD policy + from app.core.rate_limit import RateLimitPolicy, check_rate_limit_policy + await check_rate_limit_policy( + f"rate:upload:dms:{tenant_id}:{user_id}", + RateLimitPolicy.UPLOAD, + ) + # Check for blocked file types if _is_blocked_filetype(file.filename or ""): raise HTTPException( diff --git a/app/plugins/builtins/forgejo_error_reporter/service.py b/app/plugins/builtins/forgejo_error_reporter/service.py index 74e7eba..e690a9c 100644 --- a/app/plugins/builtins/forgejo_error_reporter/service.py +++ b/app/plugins/builtins/forgejo_error_reporter/service.py @@ -21,10 +21,9 @@ _DEDUP_MAX_SIZE = 100 _DEDUP_TTL_SECONDS = 3600 # 1 hour # ── Rate limiting ──────────────────────────────────────────────────────────── -# Max 10 issues per hour +# Max 10 reports per 5 minutes — enforced via central check_rate_limit() _RATE_LIMIT_MAX = 10 -_RATE_LIMIT_WINDOW = 3600 # 1 hour in seconds -_rate_limit_timestamps: list[float] = [] +_RATE_LIMIT_WINDOW = 300 # 5 minutes # ── Lock for thread safety ────────────────────────────────────────────────── _lock = asyncio.Lock() @@ -79,30 +78,6 @@ async def _is_duplicate(dedup_key: str) -> bool: return False -async def _check_rate_limit() -> bool: - """Check if we're within the rate limit. - - Returns True if request is allowed, False if rate limited. - """ - async with _lock: - now = time.monotonic() - window_start = now - _RATE_LIMIT_WINDOW - - # Remove timestamps outside the window - while _rate_limit_timestamps and _rate_limit_timestamps[0] < window_start: - _rate_limit_timestamps.pop(0) - - if len(_rate_limit_timestamps) >= _RATE_LIMIT_MAX: - logger.warning( - "Forgejo Error Reporter rate limited: %d issues in the last hour", - len(_rate_limit_timestamps), - ) - return False - - _rate_limit_timestamps.append(now) - return True - - async def _ensure_labels_exist(client: httpx.AsyncClient, settings: dict[str, str]) -> list[int]: """Ensure required labels exist in the Forgejo repository. @@ -178,8 +153,22 @@ async def report_error_to_forgejo(entry: dict[str, Any]) -> bool: if await _is_duplicate(dedup_key): return False - # Rate limit check - if not await _check_rate_limit(): + # Rate limit check — uses central check_rate_limit() with Redis + in-memory fallback + from app.core.rate_limit import check_rate_limit + from fastapi import HTTPException + + try: + await check_rate_limit( + "rate:forgejo_report:global", + _RATE_LIMIT_MAX, + _RATE_LIMIT_WINDOW, + ) + except HTTPException: + logger.warning( + "Forgejo Error Reporter rate limited: max %d reports per %d seconds", + _RATE_LIMIT_MAX, + _RATE_LIMIT_WINDOW, + ) return False # Build issue body diff --git a/app/plugins/builtins/kommunikation/routes.py b/app/plugins/builtins/kommunikation/routes.py index 0c1b9d5..796e1bd 100644 --- a/app/plugins/builtins/kommunikation/routes.py +++ b/app/plugins/builtins/kommunikation/routes.py @@ -350,6 +350,13 @@ async def upload_attachment( tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) msg_id = _parse_uuid(message_id, "message_id") + + # Rate limit — UPLOAD policy + from app.core.rate_limit import RateLimitPolicy, check_rate_limit_policy + await check_rate_limit_policy( + f"rate:upload:comm:{tenant_id}:{user_id}", + RateLimitPolicy.UPLOAD, + ) # Get conversation_id from message from sqlalchemy import select from app.plugins.builtins.kommunikation.models import CommMessage diff --git a/app/plugins/builtins/mail/routes.py b/app/plugins/builtins/mail/routes.py index 6fe0367..aa0622c 100644 --- a/app/plugins/builtins/mail/routes.py +++ b/app/plugins/builtins/mail/routes.py @@ -748,6 +748,13 @@ async def upload_attachment( user_id = uuid.UUID(current_user["user_id"]) is_system_admin = current_user.get("is_system_admin", False) + # Rate limit — UPLOAD policy + from app.core.rate_limit import RateLimitPolicy, check_rate_limit_policy + await check_rate_limit_policy( + f"rate:upload:mail:{tenant_id}:{user_id}", + RateLimitPolicy.UPLOAD, + ) + # Read file content and check size content = await file.read() if len(content) > MAX_ATTACHMENT_SIZE: diff --git a/app/routes/ai_copilot.py b/app/routes/ai_copilot.py index 3a3fa89..b713d73 100644 --- a/app/routes/ai_copilot.py +++ b/app/routes/ai_copilot.py @@ -4,10 +4,11 @@ from __future__ import annotations import uuid -from fastapi import APIRouter, Depends, HTTPException, Query, status +from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from sqlalchemy.ext.asyncio import AsyncSession from app.core.db import get_db +from app.core.rate_limit import RateLimitPolicy, check_rate_limit_policy, get_client_ip from app.deps import get_current_user, require_permission from app.schemas.ai_copilot import ( CopilotExecuteRequest, @@ -21,6 +22,7 @@ router = APIRouter(prefix="/api/v1/ai/copilot", tags=["ai-copilot"]) @router.post("/query") async def copilot_query( body: CopilotQueryRequest, + request: Request, db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("ai:write")), ): @@ -32,6 +34,12 @@ async def copilot_query( tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) + # Rate limit — AI policy (cost-sensitive LLM call) + await check_rate_limit_policy( + f"rate:ai:copilot:{tenant_id}:{user_id}", + RateLimitPolicy.AI, + ) + result = await ai_copilot_service.process_query( db, tenant_id, @@ -53,6 +61,7 @@ async def copilot_query( @router.post("/execute") async def copilot_execute( body: CopilotExecuteRequest, + request: Request, db: AsyncSession = Depends(get_db), current_user: dict = Depends(require_permission("ai:write")), ): @@ -63,6 +72,12 @@ async def copilot_execute( """ tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) + + # Rate limit — AI policy (cost-sensitive LLM call) + await check_rate_limit_policy( + f"rate:ai:copilot:{tenant_id}:{user_id}", + RateLimitPolicy.AI, + ) resolved = { "permissions": current_user.get("permissions", []), "denied": current_user.get("denied_permissions", []), diff --git a/app/routes/attachments.py b/app/routes/attachments.py index 0625a18..45001e8 100644 --- a/app/routes/attachments.py +++ b/app/routes/attachments.py @@ -5,11 +5,12 @@ from __future__ import annotations import os import uuid -from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status +from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile, status from fastapi.responses import FileResponse from sqlalchemy.ext.asyncio import AsyncSession from app.core.db import get_db +from app.core.rate_limit import RateLimitPolicy, check_rate_limit_policy from app.deps import require_permission from app.services import attachment_service @@ -18,6 +19,7 @@ router = APIRouter(prefix="/api/v1/attachments", tags=["attachments"]) @router.post("", status_code=status.HTTP_201_CREATED) async def upload_attachment( + request: Request, file: UploadFile = File(...), entity_type: str = Form(...), entity_id: str = Form(...), @@ -29,6 +31,12 @@ async def upload_attachment( user_id = uuid.UUID(current_user["user_id"]) is_admin = current_user.get("is_system_admin", False) + # Rate limit — UPLOAD policy + await check_rate_limit_policy( + f"rate:upload:attachments:{tenant_id}:{user_id}", + RateLimitPolicy.UPLOAD, + ) + try: eid = uuid.UUID(entity_id) except ValueError: diff --git a/app/routes/auth.py b/app/routes/auth.py index 335257f..beacfa9 100644 --- a/app/routes/auth.py +++ b/app/routes/auth.py @@ -10,7 +10,12 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.config import get_settings from app.core.auth import get_redis from app.core.db import get_auth_db -from app.core.rate_limit import check_rate_limit, get_client_ip, reset_rate_limit +from app.core.rate_limit import ( + RateLimitPolicy, + check_rate_limit_policy, + get_client_ip, + reset_rate_limit, +) from app.schemas.auth import ( AuthResponse, LoginRequest, @@ -36,11 +41,10 @@ async def login( ip = get_client_ip(request) redis = get_redis() - # Rate limit - await check_rate_limit( - f"auth:login:{ip}:{body.email}", - settings.rate_limit_login_max, - settings.rate_limit_login_window, + # Rate limit — unified AUTH policy + await check_rate_limit_policy( + f"rate:auth:login:{ip}:{body.email}", + RateLimitPolicy.AUTH, ) result = await auth_service.login(db, redis, body.email, body.password, body.tenant_slug) @@ -53,7 +57,7 @@ async def login( session_id, csrf_token, user, tenant, role = result # Reset rate limit on success - await reset_rate_limit(f"auth:login:{ip}:{body.email}") + await reset_rate_limit(f"rate:auth:login:{ip}:{body.email}") response = Response(status_code=status.HTTP_200_OK) response.set_cookie( @@ -217,10 +221,9 @@ async def password_reset_request( ip = get_client_ip(request) redis = get_redis() # noqa: F841 - await check_rate_limit( - f"auth:reset:{ip}", - settings.rate_limit_reset_max, - settings.rate_limit_reset_window, + await check_rate_limit_policy( + f"rate:auth:reset:{ip}", + RateLimitPolicy.AUTH, ) await auth_service.request_password_reset(db, body.email) @@ -237,10 +240,9 @@ async def password_reset_confirm( ip = get_client_ip(request) redis = get_redis() # noqa: F841 - await check_rate_limit( - f"auth:reset_confirm:{ip}", - settings.rate_limit_reset_confirm_max, - settings.rate_limit_reset_confirm_window, + await check_rate_limit_policy( + f"rate:auth:reset_confirm:{ip}", + RateLimitPolicy.AUTH, ) success = await auth_service.confirm_password_reset(db, body.token, body.new_password) diff --git a/app/routes/errors.py b/app/routes/errors.py index c35497f..4e662b3 100644 --- a/app/routes/errors.py +++ b/app/routes/errors.py @@ -1,29 +1,26 @@ """Error logging endpoint — accepts frontend errors and logs them. No auth required so errors can be logged even during logout. -Rate-limited to 10 requests per minute per IP (simple in-memory implementation). +Rate-limited to 10 requests per minute per IP via central check_rate_limit(). Context data is sanitized to prevent leaking sensitive information. """ from __future__ import annotations -import time import logging import re -from collections import defaultdict, deque from typing import Any -from fastapi import APIRouter, Request, Response, status +from fastapi import APIRouter, HTTPException, Request, Response, status from pydantic import BaseModel, Field logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/errors", tags=["errors"]) -# -- Simple in-memory rate limiter (10 req/min per IP) -- -RATE_LIMIT = 10 # max requests -RATE_WINDOW = 60 # seconds -_ip_requests: dict[str, deque[float]] = defaultdict(deque) +# -- Rate limit constants (10 req/min per IP) -- +_RATE_LIMIT_MAX = 10 +_RATE_LIMIT_WINDOW = 60 # seconds # -- Sensitive key patterns to strip from context -- _SENSITIVE_PATTERNS = re.compile( @@ -33,22 +30,6 @@ _SENSITIVE_PATTERNS = re.compile( ) -def _is_rate_limited(client_ip: str) -> bool: - """Return True if the IP has exceeded the rate limit.""" - now = time.monotonic() - dq = _ip_requests[client_ip] - - # Remove timestamps outside the window - while dq and now - dq[0] > RATE_WINDOW: - dq.popleft() - - if len(dq) >= RATE_LIMIT: - return True - - dq.append(now) - return False - - def _sanitize_context(context: Any, max_depth: int = 3, _depth: int = 0) -> Any: """Recursively remove sensitive keys and limit depth/size of context data.""" if _depth > max_depth: @@ -82,11 +63,17 @@ class ErrorReport(BaseModel): @router.post("", status_code=status.HTTP_204_NO_CONTENT) async def report_error(error: ErrorReport, request: Request) -> Response: """Log a frontend error. No auth required. Rate-limited per IP.""" - from app.core.rate_limit import get_client_ip + from app.core.rate_limit import check_rate_limit, get_client_ip client_ip = get_client_ip(request) - if _is_rate_limited(client_ip): + try: + await check_rate_limit( + f"rate:errors:{client_ip}", + _RATE_LIMIT_MAX, + _RATE_LIMIT_WINDOW, + ) + except HTTPException: return Response(status_code=status.HTTP_429_TOO_MANY_REQUESTS) # Sanitize context to prevent leaking sensitive data diff --git a/app/routes/webhooks.py b/app/routes/webhooks.py index a9a5121..c876a54 100644 --- a/app/routes/webhooks.py +++ b/app/routes/webhooks.py @@ -4,10 +4,11 @@ from __future__ import annotations import uuid -from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi import APIRouter, Depends, HTTPException, Query, Request from sqlalchemy.ext.asyncio import AsyncSession from app.core.db import get_db +from app.core.rate_limit import RateLimitPolicy, check_rate_limit_policy from app.deps import get_current_user, require_permission from app.schemas.webhook import ( WebhookCreate, @@ -166,6 +167,7 @@ async def delete_webhook( ) async def test_webhook( webhook_id: str, + request: Request, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): @@ -174,6 +176,12 @@ async def test_webhook( user_id = uuid.UUID(current_user["user_id"]) is_admin = current_user.get("is_system_admin", False) + # Rate limit — WEBHOOK policy + await check_rate_limit_policy( + f"rate:webhook:test:{tenant_id}:{user_id}", + RateLimitPolicy.WEBHOOK, + ) + try: wh_id = uuid.UUID(webhook_id) except (ValueError, TypeError): diff --git a/tests/test_rate_limit_policies.py b/tests/test_rate_limit_policies.py new file mode 100644 index 0000000..718e377 --- /dev/null +++ b/tests/test_rate_limit_policies.py @@ -0,0 +1,335 @@ +"""Tests for unified rate-limit policies and central check_rate_limit(). + +Covers: +- check_rate_limit() with various policies (AUTH, AI, UPLOAD, WEBHOOK) +- Rate limit triggers on exceeding max attempts +- Reset after window timeout (in-memory fallback) +- get_client_ip() with and without trusted proxy +- Forgejo error reporter uses central check_rate_limit() +- AI proactive is_rate_limited() delegates to central check_rate_limit() +""" + +from __future__ import annotations + +import asyncio +import uuid +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import HTTPException, Request +from starlette.datastructures import Headers + +from app.core.rate_limit import ( + RateLimitPolicy, + check_rate_limit, + check_rate_limit_policy, + get_client_ip, + reset_rate_limit, +) + + +# ─── Unit Tests: RateLimitPolicy ─── + +@pytest.mark.asyncio +async def test_rate_limit_policy_auth_limits(): + """RateLimitPolicy.AUTH returns (rate_limit_auth_max, rate_limit_auth_window).""" + max_attempts, window = RateLimitPolicy.AUTH.limits() + assert max_attempts == 5 + assert window == 300 + + +@pytest.mark.asyncio +async def test_rate_limit_policy_ai_limits(): + """RateLimitPolicy.AI returns (rate_limit_ai_max, rate_limit_ai_window).""" + max_attempts, window = RateLimitPolicy.AI.limits() + assert max_attempts == 20 + assert window == 60 + + +@pytest.mark.asyncio +async def test_rate_limit_policy_upload_limits(): + """RateLimitPolicy.UPLOAD returns (rate_limit_upload_max, rate_limit_upload_window).""" + max_attempts, window = RateLimitPolicy.UPLOAD.limits() + assert max_attempts == 30 + assert window == 60 + + +@pytest.mark.asyncio +async def test_rate_limit_policy_webhook_limits(): + """RateLimitPolicy.WEBHOOK returns (rate_limit_webhook_max, rate_limit_webhook_window).""" + max_attempts, window = RateLimitPolicy.WEBHOOK.limits() + assert max_attempts == 100 + assert window == 60 + + +# ─── Unit Tests: check_rate_limit via in-memory fallback ─── + +def _force_inmemory_fallback(): + """Return a context manager that patches the circuit breaker to force in-memory fallback.""" + circuit = MagicMock() + circuit.can_proceed = AsyncMock(return_value=False) + circuit.record_success = AsyncMock() + circuit.record_failure = AsyncMock() + return patch("app.core.resilience.get_circuit", return_value=circuit) + + +@pytest.mark.asyncio +async def test_check_rate_limit_allows_under_max(): + """check_rate_limit allows requests up to max_attempts.""" + key = f"test:allow:{uuid.uuid4()}" + with _force_inmemory_fallback(): + for _ in range(3): + await check_rate_limit(key, max_attempts=3, window_seconds=60) + + +@pytest.mark.asyncio +async def test_check_rate_limit_blocks_on_exceed(): + """check_rate_limit raises 429 when max_attempts is exceeded.""" + key = f"test:block:{uuid.uuid4()}" + with _force_inmemory_fallback(): + for _ in range(2): + await check_rate_limit(key, max_attempts=2, window_seconds=60) + with pytest.raises(HTTPException) as exc_info: + await check_rate_limit(key, max_attempts=2, window_seconds=60) + assert exc_info.value.status_code == 429 + + +@pytest.mark.asyncio +async def test_check_rate_limit_reset_after_window(): + """Rate limit resets after the window timeout (in-memory fallback).""" + key = f"test:reset:{uuid.uuid4()}" + with _force_inmemory_fallback(): + # Use up the limit with a very short window + await check_rate_limit(key, max_attempts=1, window_seconds=1) + with pytest.raises(HTTPException): + await check_rate_limit(key, max_attempts=1, window_seconds=1) + # Wait for window to expire + await asyncio.sleep(1.1) + # Should be allowed again + await check_rate_limit(key, max_attempts=1, window_seconds=1) + + +@pytest.mark.asyncio +async def test_reset_rate_limit_clears_counter(): + """reset_rate_limit clears the counter so subsequent calls succeed.""" + key = f"test:reset2:{uuid.uuid4()}" + with _force_inmemory_fallback(): + await check_rate_limit(key, max_attempts=1, window_seconds=60) + with pytest.raises(HTTPException): + await check_rate_limit(key, max_attempts=1, window_seconds=60) + await reset_rate_limit(key) + await check_rate_limit(key, max_attempts=1, window_seconds=60) + + +# ─── Unit Tests: check_rate_limit_policy ─── + +@pytest.mark.asyncio +async def test_check_rate_limit_policy_auth(): + """check_rate_limit_policy with AUTH policy enforces auth limits.""" + key = f"test:policy:auth:{uuid.uuid4()}" + with _force_inmemory_fallback(): + max_attempts, _ = RateLimitPolicy.AUTH.limits() + for _ in range(max_attempts): + await check_rate_limit_policy(key, RateLimitPolicy.AUTH) + with pytest.raises(HTTPException) as exc_info: + await check_rate_limit_policy(key, RateLimitPolicy.AUTH) + assert exc_info.value.status_code == 429 + + +@pytest.mark.asyncio +async def test_check_rate_limit_policy_ai(): + """check_rate_limit_policy with AI policy enforces AI limits.""" + key = f"test:policy:ai:{uuid.uuid4()}" + with _force_inmemory_fallback(): + max_attempts, _ = RateLimitPolicy.AI.limits() + for _ in range(max_attempts): + await check_rate_limit_policy(key, RateLimitPolicy.AI) + with pytest.raises(HTTPException) as exc_info: + await check_rate_limit_policy(key, RateLimitPolicy.AI) + assert exc_info.value.status_code == 429 + + +# ─── Unit Tests: get_client_ip ─── + +def _make_request(client_host: str = "127.0.0.1", headers: dict | None = None) -> Request: + """Build a mock Request for get_client_ip testing.""" + scope = { + "type": "http", + "method": "GET", + "path": "/", + "headers": Headers(headers or {}).raw, + "client": (client_host, 8000), + } + return Request(scope) + + +def test_get_client_ip_no_trusted_proxy(): + """get_client_ip returns direct IP when no trusted proxies configured.""" + with patch("app.config.get_settings") as mock_settings: + settings = MagicMock() + settings.trusted_proxy_cidrs = "" + mock_settings.return_value = settings + req = _make_request("192.168.1.100", {"x-forwarded-for": "10.0.0.1"}) + ip = get_client_ip(req) + assert ip == "192.168.1.100" + + +def test_get_client_ip_with_trusted_proxy(): + """get_client_ip uses X-Forwarded-For when direct client is a trusted proxy.""" + with patch("app.config.get_settings") as mock_settings: + settings = MagicMock() + settings.trusted_proxy_cidrs = "10.0.0.0/8" + mock_settings.return_value = settings + req = _make_request("10.0.0.1", {"x-forwarded-for": "203.0.113.50"}) + ip = get_client_ip(req) + assert ip == "203.0.113.50" + + +def test_get_client_ip_trusted_proxy_uses_x_real_ip(): + """get_client_ip falls back to X-Real-IP when X-Forwarded-For is absent.""" + with patch("app.config.get_settings") as mock_settings: + settings = MagicMock() + settings.trusted_proxy_cidrs = "10.0.0.0/8" + mock_settings.return_value = settings + req = _make_request("10.0.0.1", {"x-real-ip": "203.0.113.99"}) + ip = get_client_ip(req) + assert ip == "203.0.113.99" + + +def test_get_client_ip_untrusted_proxy_ignored(): + """get_client_ip ignores X-Forwarded-For from untrusted proxy.""" + with patch("app.config.get_settings") as mock_settings: + settings = MagicMock() + settings.trusted_proxy_cidrs = "10.0.0.0/8" + mock_settings.return_value = settings + req = _make_request("192.168.1.1", {"x-forwarded-for": "203.0.113.50"}) + ip = get_client_ip(req) + assert ip == "192.168.1.1" + + +def test_get_client_ip_multiple_forwarded(): + """get_client_ip uses leftmost IP from comma-separated X-Forwarded-For.""" + with patch("app.config.get_settings") as mock_settings: + settings = MagicMock() + settings.trusted_proxy_cidrs = "10.0.0.0/8" + mock_settings.return_value = settings + req = _make_request("10.0.0.1", {"x-forwarded-for": "203.0.113.50, 10.0.0.2, 10.0.0.3"}) + ip = get_client_ip(req) + assert ip == "203.0.113.50" + + +# ─── Integration Tests: Forgejo reporter uses central rate limit ─── + +@pytest.mark.asyncio +async def test_forgejo_reporter_uses_central_rate_limit(): + """Forgejo error reporter delegates to check_rate_limit() and respects limits.""" + from app.plugins.builtins.forgejo_error_reporter.service import ( + _RATE_LIMIT_MAX, + _RATE_LIMIT_WINDOW, + report_error_to_forgejo, + ) + + assert _RATE_LIMIT_MAX == 10 + assert _RATE_LIMIT_WINDOW == 300 + + call_count = 0 + + async def mock_check_rate_limit(redis_key, max_attempts, window_seconds): + nonlocal call_count + call_count += 1 + assert redis_key == "rate:forgejo_report:global" + assert max_attempts == 10 + assert window_seconds == 300 + if call_count > _RATE_LIMIT_MAX: + raise HTTPException(status_code=429, detail="Rate limit exceeded") + + with patch( + "app.plugins.builtins.forgejo_error_reporter.service._is_duplicate", + new_callable=AsyncMock, + return_value=False, + ), patch( + "app.plugins.builtins.forgejo_error_reporter.service._get_settings", + return_value={ + "url": "https://forgejo.example.com", + "token": "test-token", + "owner": "test", + "repo": "test", + }, + ), patch( + "app.core.rate_limit.check_rate_limit", + side_effect=mock_check_rate_limit, + ): + entry = {"message": "Test error", "stack": "trace"} + for i in range(_RATE_LIMIT_MAX): + await report_error_to_forgejo(entry) + assert call_count == i + 1 + + result = await report_error_to_forgejo(entry) + assert result is False + assert call_count == _RATE_LIMIT_MAX + 1 + + +# ─── Integration Tests: AI proactive uses central rate limit ─── + +@pytest.mark.asyncio +async def test_ai_proactive_uses_central_rate_limit(): + """AI proactive is_rate_limited() delegates to check_rate_limit().""" + from app.plugins.builtins.ai_proactive.services import is_rate_limited + + tenant_id = uuid.uuid4() + user_id = uuid.uuid4() + rate_limit_seconds = 5 + + # First call should not be rate-limited + result = await is_rate_limited(tenant_id, user_id, rate_limit_seconds) + assert result is False + + # Second call (within window) should be rate-limited + result = await is_rate_limited(tenant_id, user_id, rate_limit_seconds) + assert result is True + + +@pytest.mark.asyncio +async def test_ai_proactive_rate_limit_key_format(): + """AI proactive uses the correct Redis key format: rate:ai_proactive:{tenant_id}:{user_id}.""" + tenant_id = uuid.uuid4() + user_id = uuid.uuid4() + expected_key = f"rate:ai_proactive:{tenant_id}:{user_id}" + + captured_key = None + + async def mock_check_rate_limit(redis_key, max_attempts, window_seconds): + nonlocal captured_key + captured_key = redis_key + assert max_attempts == 1 + assert window_seconds == 10 + + with patch("app.core.rate_limit.check_rate_limit", side_effect=mock_check_rate_limit): + from app.plugins.builtins.ai_proactive.services import is_rate_limited + await is_rate_limited(tenant_id, user_id, rate_limit_seconds=10) + assert captured_key == expected_key + + +@pytest.mark.asyncio +async def test_ai_proactive_rate_limit_exception_allows(): + """AI proactive allows request when rate limit check raises non-HTTP exception.""" + with patch("app.core.rate_limit.check_rate_limit", side_effect=RuntimeError("Redis down")): + from app.plugins.builtins.ai_proactive.services import is_rate_limited + result = await is_rate_limited(uuid.uuid4(), uuid.uuid4(), rate_limit_seconds=10) + assert result is False + + +# ─── Integration Tests: Independent keys per user/tenant ─── + +@pytest.mark.asyncio +async def test_independent_rate_limit_keys(): + """Different keys have independent rate limits.""" + key1 = f"test:indep:1:{uuid.uuid4()}" + key2 = f"test:indep:2:{uuid.uuid4()}" + + with _force_inmemory_fallback(): + await check_rate_limit(key1, max_attempts=1, window_seconds=60) + with pytest.raises(HTTPException): + await check_rate_limit(key1, max_attempts=1, window_seconds=60) + # key2 should still be allowed + await check_rate_limit(key2, max_attempts=1, window_seconds=60)