Add per-IP rate limiting to prevent DoS attacks

CHANGELOG.md

@@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.2.4] - 2026-03-31
+
+### Security
+
+- Added per-IP sliding-window rate limiting to upload, processing, and retention
+  sweep endpoints to mitigate DoS attacks. Configurable via `IMG_RATE_LIMIT_UPLOAD_MAX`,
+  `IMG_RATE_LIMIT_UPLOAD_WINDOW`, `IMG_RATE_LIMIT_PROCESS_MAX`,
+  `IMG_RATE_LIMIT_PROCESS_WINDOW`, `IMG_RATE_LIMIT_READ_MAX`, and
+  `IMG_RATE_LIMIT_READ_WINDOW` environment variables. Read endpoints (list,
+  get, download) use a separate higher limit (60 req/min by default). Returns
+  HTTP 429 when the limit is exceeded.
+
 ## [1.2.3] - 2026-03-30
 
 ### Security

PROJECT_DESCRIPTION.md

@@ -84,8 +84,9 @@ Fast, reliable REST APIs are designed and developed with **FastAPI**:
 
 **API quality features:**
 - **API key authentication** on all `/api/v1/` endpoints via `X-API-Key` header (configurable, timing-safe comparison). Health endpoint stays open for Kubernetes probes.
+- **Per-IP rate limiting** on upload (10 req/min), processing (20 req/min), and read (60 req/min) endpoints using a sliding-window algorithm. Returns HTTP 429 when exceeded. All limits are configurable via environment variables.
 - Pydantic schemas with strict validation (max 20 tags, TTL range enforcement, max upload size)
-- Proper HTTP status codes (201 Created, 401 Unauthorized, 404 Not Found, 413 Payload Too Large, 415 Unsupported Media Type)
+- Proper HTTP status codes (201 Created, 401 Unauthorized, 404 Not Found, 413 Payload Too Large, 415 Unsupported Media Type, 429 Too Many Requests)
 - Content-type enforcement for upload security
 - Automatic OpenAPI/Swagger documentation at `/docs` and ReDoc at `/redoc`
 - Request logging middleware tracking method, path, status, and response time
@@ -130,6 +131,12 @@ All settings are provided via environment variables (prefix `IMG_`) using **pyda
 | `IMG_THUMBNAIL_MAX_SIZE` | `256` | Thumbnail max dimension (pixels) |
 | `IMG_RETENTION_BATCH_SIZE` | `100` | Expired images per retention sweep |
 | `IMG_API_KEY` | *(empty)* | API key for `X-API-Key` header auth (empty = disabled) |
+| `IMG_RATE_LIMIT_UPLOAD_MAX` | `10` | Max upload requests per window per IP |
+| `IMG_RATE_LIMIT_UPLOAD_WINDOW` | `60` | Upload rate limit window (seconds) |
+| `IMG_RATE_LIMIT_PROCESS_MAX` | `20` | Max process requests per window per IP |
+| `IMG_RATE_LIMIT_PROCESS_WINDOW` | `60` | Process rate limit window (seconds) |
+| `IMG_RATE_LIMIT_READ_MAX` | `60` | Max read requests per window per IP |
+| `IMG_RATE_LIMIT_READ_WINDOW` | `60` | Read rate limit window (seconds) |
 | `IMG_DEBUG` | `false` | Enable debug logging |
 
 ---

README.md

@@ -60,6 +60,7 @@ cd cpp && ./build.sh
 
 All `/api/v1/` endpoints require an `X-API-Key` header when `IMG_API_KEY` is set.
 The `/health` endpoint remains open for Kubernetes probes.
+Upload, processing, and read endpoints are rate-limited per client IP (HTTP 429 when exceeded).
 
 | Method | Path | Description |
 |--------|------|-------------|
@@ -86,6 +87,12 @@ All settings via environment variables (prefix `IMG_`), validated by [pydantic-s
 | `IMG_THUMBNAIL_MAX_SIZE` | `256` | Thumbnail max dimension (px) |
 | `IMG_RETENTION_BATCH_SIZE` | `100` | Expired images per sweep |
 | `IMG_API_KEY` | *(empty)* | API key for `X-API-Key` header auth (empty = disabled) |
+| `IMG_RATE_LIMIT_UPLOAD_MAX` | `10` | Max upload requests per window per IP |
+| `IMG_RATE_LIMIT_UPLOAD_WINDOW` | `60` | Upload rate limit window (seconds) |
+| `IMG_RATE_LIMIT_PROCESS_MAX` | `20` | Max process requests per window per IP |
+| `IMG_RATE_LIMIT_PROCESS_WINDOW` | `60` | Process rate limit window (seconds) |
+| `IMG_RATE_LIMIT_READ_MAX` | `60` | Max read requests per window per IP |
+| `IMG_RATE_LIMIT_READ_WINDOW` | `60` | Read rate limit window (seconds) |
 | `IMG_DEBUG` | `false` | Enable debug logging |
 
 ## Project Structure

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "image-processing-service"
-version = "1.2.3"
+version = "1.2.4"
 description = "High-performance image processing microservice with Clean Architecture"
 requires-python = ">=3.11"
 dependencies = [

src/config.py

@@ -29,6 +29,14 @@ class Settings(BaseSettings):
     # ── Retention ────────────────────────────────────────────────────────
     retention_batch_size: int = 100
 
+    # ── Rate Limiting ────────────────────────────────────────────────────
+    rate_limit_upload_max: int = 10
+    rate_limit_upload_window: int = 60
+    rate_limit_process_max: int = 20
+    rate_limit_process_window: int = 60
+    rate_limit_read_max: int = 60
+    rate_limit_read_window: int = 60
+
     # ── Authentication ───────────────────────────────────────────────────
     api_key: str = ""
 

src/presentation/api/dependencies.py

@@ -25,6 +25,7 @@
 from src.infrastructure.database.session import build_engine, build_session_factory
 from src.infrastructure.processing.pillow_processor import PillowImageProcessor
 from src.infrastructure.storage.local_image_storage import LocalImageStorage
+from src.presentation.api.rate_limit import RateLimiter
 
 
 @lru_cache
@@ -87,6 +88,24 @@ def _processor() -> PillowImageProcessor:
     return PillowImageProcessor(max_workers=get_settings().processing_max_workers)
 
 
+@lru_cache
+def upload_rate_limiter() -> RateLimiter:
+    settings = get_settings()
+    return RateLimiter(settings.rate_limit_upload_max, settings.rate_limit_upload_window)
+
+
+@lru_cache
+def process_rate_limiter() -> RateLimiter:
+    settings = get_settings()
+    return RateLimiter(settings.rate_limit_process_max, settings.rate_limit_process_window)
+
+
+@lru_cache
+def read_rate_limiter() -> RateLimiter:
+    settings = get_settings()
+    return RateLimiter(settings.rate_limit_read_max, settings.rate_limit_read_window)
+
+
 def get_upload_use_case() -> UploadImageUseCase:
     return UploadImageUseCase(_repository(), _storage())
 

src/presentation/api/rate_limit.py

@@ -0,0 +1,57 @@
+"""In-memory sliding-window rate limiter for FastAPI."""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections import defaultdict
+
+from fastapi import HTTPException, Request, status
+
+
+class RateLimiter:
+    """Sliding-window rate limiter keyed by client IP.
+
+    Parameters
+    ----------
+    max_requests : int
+        Maximum number of requests allowed in the time window.
+    window_seconds : int
+        Length of the sliding window in seconds.
+    """
+
+    def __init__(self, max_requests: int, window_seconds: int) -> None:
+        self._max_requests = max_requests
+        self._window_seconds = window_seconds
+        self._requests: dict[str, list[float]] = defaultdict(list)
+        self._lock = threading.Lock()
+
+    def _client_ip(self, request: Request) -> str:
+        forwarded = request.headers.get("x-forwarded-for")
+        if forwarded:
+            return forwarded.split(",")[0].strip()
+        return request.client.host if request.client else "unknown"
+
+    def _cleanup(self, key: str, now: float) -> None:
+        cutoff = now - self._window_seconds
+        timestamps = self._requests[key]
+        # Remove expired timestamps
+        while timestamps and timestamps[0] < cutoff:
+            timestamps.pop(0)
+        if not timestamps:
+            del self._requests[key]
+
+    async def __call__(self, request: Request) -> None:
+        now = time.monotonic()
+        key = self._client_ip(request)
+
+        with self._lock:
+            self._cleanup(key, now)
+            timestamps = self._requests[key]
+
+            if len(timestamps) >= self._max_requests:
+                raise HTTPException(
+                    status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+                    detail="Rate limit exceeded. Try again later.",
+                )
+            timestamps.append(now)

src/presentation/api/routes/images.py

@@ -18,7 +18,10 @@
     get_list_use_case,
     get_process_use_case,
     get_upload_use_case,
+    process_rate_limiter,
+    read_rate_limiter,
     require_api_key,
+    upload_rate_limiter,
 )
 from src.presentation.sanitize import sanitize_filename
 from src.presentation.schemas.image_schemas import (
@@ -43,6 +46,7 @@ async def upload_image(
     tags: Annotated[list[str] | None, Query()] = None,
     ttl_hours: Annotated[int | None, Query(ge=1, le=8760)] = None,
     use_case: Annotated[UploadImageUseCase, Depends(get_upload_use_case)] = None,  # type: ignore[assignment]
+    _rate: Annotated[None, Depends(upload_rate_limiter())] = None,
 ):
     if file.content_type not in ALLOWED_CONTENT_TYPES:
         raise HTTPException(
@@ -85,6 +89,7 @@ async def list_images(
     limit: Annotated[int, Query(ge=1, le=100)] = 50,
     status_filter: Annotated[str | None, Query(alias="status")] = None,
     use_case: Annotated[ListImagesUseCase, Depends(get_list_use_case)] = None,  # type: ignore[assignment]
+    _rate: Annotated[None, Depends(read_rate_limiter())] = None,
 ):
     return await use_case.execute(offset=offset, limit=limit, status=status_filter)
 
@@ -93,6 +98,7 @@ async def list_images(
 async def get_image(
     image_id: uuid.UUID,
     use_case: Annotated[GetImageUseCase, Depends(get_get_image_use_case)] = None,  # type: ignore[assignment]
+    _rate: Annotated[None, Depends(read_rate_limiter())] = None,
 ):
     result = await use_case.execute(image_id)
     if result is None:
@@ -105,6 +111,7 @@ async def download_image(
     image_id: uuid.UUID,
     thumbnail: bool = False,
     use_case: Annotated[GetImageUseCase, Depends(get_get_image_use_case)] = None,  # type: ignore[assignment]
+    _rate: Annotated[None, Depends(read_rate_limiter())] = None,
 ):
     data = await use_case.get_file(image_id, thumbnail=thumbnail)
     if data is None:
@@ -116,6 +123,7 @@ async def download_image(
 async def process_batch_images(
     body: BatchProcessRequest,
     use_case: Annotated[ProcessImageUseCase, Depends(get_process_use_case)] = None,  # type: ignore[assignment]
+    _rate: Annotated[None, Depends(process_rate_limiter())] = None,
 ):
     result = await process_batch(use_case, body.image_ids, concurrency=body.concurrency)
     return result
@@ -126,6 +134,7 @@ async def process_single_image(
     image_id: uuid.UUID,
     process_uc: Annotated[ProcessImageUseCase, Depends(get_process_use_case)] = None,  # type: ignore[assignment]
     get_uc: Annotated[GetImageUseCase, Depends(get_get_image_use_case)] = None,  # type: ignore[assignment]
+    _rate: Annotated[None, Depends(process_rate_limiter())] = None,
 ):
     ok = await process_uc.execute(image_id)
     if not ok:

src/presentation/api/routes/retention.py

@@ -8,7 +8,12 @@
 
 from src.application.use_cases.apply_retention import ApplyRetentionUseCase
 from src.config import Settings
-from src.presentation.api.dependencies import get_retention_use_case, get_settings, require_api_key
+from src.presentation.api.dependencies import (
+    get_retention_use_case,
+    get_settings,
+    process_rate_limiter,
+    require_api_key,
+)
 from src.presentation.schemas.image_schemas import RetentionResponse
 
 router = APIRouter(
@@ -22,6 +27,7 @@
 async def trigger_retention_sweep(
     use_case: Annotated[ApplyRetentionUseCase, Depends(get_retention_use_case)] = None,  # type: ignore[assignment]
     settings: Annotated[Settings, Depends(get_settings)] = None,  # type: ignore[assignment]
+    _rate: Annotated[None, Depends(process_rate_limiter())] = None,
 ):
     result = await use_case.execute(batch_size=settings.retention_batch_size)
     return RetentionResponse(deleted_count=result.deleted_count, errors=result.errors)

tests/presentation/test_rate_limit.py

@@ -0,0 +1,106 @@
+"""Tests for the rate limiter."""
+
+from __future__ import annotations
+
+import time
+from unittest.mock import patch
+
+import pytest
+from fastapi import Depends, FastAPI
+from fastapi.testclient import TestClient
+
+from src.presentation.api.rate_limit import RateLimiter
+
+
+@pytest.fixture
+def limiter() -> RateLimiter:
+    return RateLimiter(max_requests=3, window_seconds=60)
+
+
+@pytest.fixture
+def rate_app(limiter: RateLimiter) -> FastAPI:
+    """Minimal FastAPI app with a rate-limited endpoint."""
+    app = FastAPI()
+
+    @app.get("/limited")
+    async def limited(_: None = Depends(limiter)):
+        return {"ok": True}
+
+    return app
+
+
+@pytest.fixture
+def rate_client(rate_app: FastAPI) -> TestClient:
+    return TestClient(rate_app)
+
+
+class TestRateLimiter:
+    def test_allows_requests_within_limit(self, rate_client: TestClient) -> None:
+        for _ in range(3):
+            resp = rate_client.get("/limited")
+            assert resp.status_code == 200
+
+    def test_blocks_requests_over_limit(self, rate_client: TestClient) -> None:
+        for _ in range(3):
+            rate_client.get("/limited")
+
+        resp = rate_client.get("/limited")
+        assert resp.status_code == 429
+        assert "Rate limit exceeded" in resp.json()["detail"]
+
+    def test_window_expires_allows_new_requests(self) -> None:
+        limiter = RateLimiter(max_requests=2, window_seconds=1)
+        app = FastAPI()
+
+        @app.get("/limited")
+        async def limited(_: None = Depends(limiter)):
+            return {"ok": True}
+
+        client = TestClient(app)
+
+        # Use up the quota
+        for _ in range(2):
+            client.get("/limited")
+
+        assert client.get("/limited").status_code == 429
+
+        # Advance time past the window via monotonic mock
+        with patch("src.presentation.api.rate_limit.time") as mock_time:
+            mock_time.monotonic.return_value = time.monotonic() + 2
+            resp = client.get("/limited")
+            assert resp.status_code == 200
+
+    def test_separate_clients_have_separate_limits(self) -> None:
+        limiter = RateLimiter(max_requests=1, window_seconds=60)
+        app = FastAPI()
+
+        @app.get("/limited")
+        async def limited(_: None = Depends(limiter)):
+            return {"ok": True}
+
+        client = TestClient(app)
+
+        # First client at 127.0.0.1 (default for TestClient)
+        resp = client.get("/limited")
+        assert resp.status_code == 200
+        assert client.get("/limited").status_code == 429
+
+        # Different client IP via X-Forwarded-For
+        resp = client.get("/limited", headers={"X-Forwarded-For": "10.0.0.1"})
+        assert resp.status_code == 200
+
+    def test_x_forwarded_for_uses_first_ip(self) -> None:
+        limiter = RateLimiter(max_requests=1, window_seconds=60)
+        app = FastAPI()
+
+        @app.get("/limited")
+        async def limited(_: None = Depends(limiter)):
+            return {"ok": True}
+
+        client = TestClient(app)
+
+        resp = client.get("/limited", headers={"X-Forwarded-For": "192.168.1.1, 10.0.0.1"})
+        assert resp.status_code == 200
+
+        resp = client.get("/limited", headers={"X-Forwarded-For": "192.168.1.1, 10.0.0.1"})
+        assert resp.status_code == 429