Fides Webserver API Rate Limiting Vulnerability in Proxied Environments

@@ -64,6 +64,7 @@ Changes can also be flagged with a GitHub label for tracking purposes. The URL o
 
 ### Security
 - Added stricter rate limiting to authentication endpoints to mitigate against brute force attacks. [CVE-2025-57815](https://github.com/ethyca/fides/security/advisories/GHSA-7q62-r88r-j5gw)
+- Adds Redis-driven rate limiting across all endpoints [CVE-2025-57816](https://github.com/ethyca/fides/security/advisories/GHSA-fq34-xw6c-fphf)
 
 ## [2.68.0](https://github.com/ethyca/fides/compare/2.67.2...2.68.0)
 

@@ -221,6 +221,100 @@ services:
       - CELERY_BROKER_URL=redis://:redispassword@redis:6379/0
       - CELERY_RESULT_BACKEND=redis://:redispassword@redis:6379/0
 
+  # Cluster services for nginx load balancing
+  fides-1:
+    container_name: fides-1
+    image: ethyca/fides:local
+    command: uvicorn --host 0.0.0.0 --port 8080 --reload --reload-dir src --reload-dir data --reload-include='*.yml' fides.api.main:app
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://0.0.0.0:8080/health"]
+      interval: 20s
+      timeout: 5s
+      retries: 10
+    ports:
+      - "8081:8080"
+    depends_on:
+      fides-db:
+        condition: service_healthy
+      redis:
+        condition: service_started
+    expose:
+      - 8080
+    env_file:
+      - .env
+    environment:
+      FIDES__CONFIG_PATH: ${FIDES__CONFIG_PATH:-/fides/.fides/fides.toml}
+      FIDES__CLI__ANALYTICS_ID: ${FIDES__CLI__ANALYTICS_ID-}
+      FIDES__CLI__SERVER_HOST: "fides"
+      FIDES__CLI__SERVER_PORT: "8080"
+      FIDES__DATABASE__SERVER: "fides-db"
+      FIDES__DEV_MODE: "True"
+      FIDES__LOGGING__COLORIZE: "True"
+      FIDES__USER__ANALYTICS_OPT_OUT: "True"
+      FIDES__SECURITY__BASTION_SERVER_HOST: ${FIDES__SECURITY__BASTION_SERVER_HOST-}
+      FIDES__SECURITY__BASTION_SERVER_SSH_USERNAME: ${FIDES__SECURITY__BASTION_SERVER_SSH_USERNAME-}
+      FIDES__SECURITY__BASTION_SERVER_SSH_PRIVATE_KEY: ${FIDES__SECURITY__BASTION_SERVER_SSH_PRIVATE_KEY-}
+      SAAS_OP_SERVICE_ACCOUNT_TOKEN: ${SAAS_OP_SERVICE_ACCOUNT_TOKEN-}
+      SAAS_SECRETS_OP_VAULT_ID: ${SAAS_SECRETS_OP_VAULT_ID-}
+    volumes:
+      - type: bind
+        source: .
+        target: /fides
+        read_only: False
+
+  fides-2:
+    container_name: fides-2
+    image: ethyca/fides:local
+    command: uvicorn --host 0.0.0.0 --port 8080 --reload --reload-dir src --reload-dir data --reload-include='*.yml' fides.api.main:app
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://0.0.0.0:8080/health"]
+      interval: 20s
+      timeout: 5s
+      retries: 10
+    ports:
+      - "8082:8080"
+    depends_on:
+      fides-db:
+        condition: service_healthy
+      redis:
+        condition: service_started
+    expose:
+      - 8080
+    env_file:
+      - .env
+    environment:
+      FIDES__CONFIG_PATH: ${FIDES__CONFIG_PATH:-/fides/.fides/fides.toml}
+      FIDES__CLI__ANALYTICS_ID: ${FIDES__CLI__ANALYTICS_ID-}
+      FIDES__CLI__SERVER_HOST: "fides"
+      FIDES__CLI__SERVER_PORT: "8080"
+      FIDES__DATABASE__SERVER: "fides-db"
+      FIDES__DEV_MODE: "True"
+      FIDES__LOGGING__COLORIZE: "True"
+      FIDES__USER__ANALYTICS_OPT_OUT: "True"
+      FIDES__SECURITY__BASTION_SERVER_HOST: ${FIDES__SECURITY__BASTION_SERVER_HOST-}
+      FIDES__SECURITY__BASTION_SERVER_SSH_USERNAME: ${FIDES__SECURITY__BASTION_SERVER_SSH_USERNAME-}
+      FIDES__SECURITY__BASTION_SERVER_SSH_PRIVATE_KEY: ${FIDES__SECURITY__BASTION_SERVER_SSH_PRIVATE_KEY-}
+      SAAS_OP_SERVICE_ACCOUNT_TOKEN: ${SAAS_OP_SERVICE_ACCOUNT_TOKEN-}
+      SAAS_SECRETS_OP_VAULT_ID: ${SAAS_SECRETS_OP_VAULT_ID-}
+    volumes:
+      - type: bind
+        source: .
+        target: /fides
+        read_only: False
+
+  fides-proxy:
+    container_name: fides-proxy
+    image: nginx:latest
+    ports:
+      - "8083:8080"
+    volumes:
+      - ./docker/nginx/nginx.conf:/etc/nginx/nginx.conf:ro
+    depends_on:
+      fides-1:
+        condition: service_healthy
+      fides-2:
+        condition: service_healthy
+
 volumes:
   postgres: null
 

@@ -0,0 +1,24 @@
+events {}
+
+http {
+
+  log_format upstream_log '$remote_addr - $remote_user [$time_local] '
+                        '"$request" $status $body_bytes_sent '
+                        '"$http_referer" "$http_user_agent" '
+                        'upstream_addr="$upstream_addr" '
+                        'upstream_status="$upstream_status"';
+
+  upstream fides {
+    server fides-1:8080;
+    server fides-2:8080;
+  }
+
+  server {
+    listen 8080;
+    access_log /dev/stdout upstream_log;
+
+    location / {
+      proxy_pass http://fides;
+    }
+  }
+}

@@ -54,6 +54,7 @@ def dev(session: Session) -> None:
         - workers-all = Run all available Fides workers (see below)
         - flower = Run Flower monitoring dashboard for Celery
         - child = Run a Fides child node
+        - nginx = Run two Fides webservers with nginx load balancer proxy
         - <datastore(s)> = Run a test datastore (e.g. 'mssql', 'mongodb')
 
     To run specific workers only, use any of the following posargs:
@@ -111,7 +112,20 @@ def dev(session: Session) -> None:
 
     open_shell = "shell" in session.posargs
     remote_debug = "remote_debug" in session.posargs
-    if not datastores:
+    use_nginx = "nginx" in session.posargs
+
+    if use_nginx:
+        # Run two Fides webservers with nginx load balancer proxy
+        session.run(
+            "docker",
+            "compose",
+            "up",
+            "fides-1",
+            "fides-2",
+            "fides-proxy",
+            external=True,
+        )
+    elif not datastores:
         if open_shell:
             session.run(*START_APP, external=True)
             session.log("~~Remember to login with `fides user login`!~~")

@@ -21,7 +21,7 @@
 from fides.api.schemas.storage.storage import StorageType
 from fides.api.service.storage.streaming.s3 import S3StorageClient
 from fides.api.util.api_router import APIRouter
-from fides.api.util.endpoint_utils import fides_limiter
+from fides.api.util.rate_limit import fides_limiter
 from fides.common.api.v1.urn_registry import PRIVACY_CENTER_DSR_PACKAGE, V1_URL_PREFIX
 from fides.config import CONFIG
 
@@ -62,7 +62,7 @@ def raise_error(status_code: int, detail: str) -> None:
     PRIVACY_CENTER_DSR_PACKAGE,
     status_code=HTTP_302_FOUND,
 )
-@fides_limiter.limit(CONFIG.security.public_request_rate_limit)
+@fides_limiter.limit(CONFIG.security.request_rate_limit)
 def get_access_results_urls(
     privacy_request_id: str,
     token: str,

@@ -37,7 +37,7 @@
 )
 from fides.api.util.api_router import APIRouter
 from fides.api.util.connection_util import connection_status
-from fides.api.util.endpoint_utils import fides_limiter
+from fides.api.util.rate_limit import fides_limiter
 from fides.common.api.scope_registry import (
     CLIENT_CREATE,
     CLIENT_DELETE,

@@ -59,7 +59,7 @@
 )
 from fides.api.service.deps import get_user_service
 from fides.api.util.api_router import APIRouter
-from fides.api.util.endpoint_utils import fides_limiter
+from fides.api.util.rate_limit import fides_limiter
 from fides.common.api.scope_registry import (
     SCOPE_REGISTRY,
     SYSTEM_MANAGER_DELETE,

@@ -48,9 +48,13 @@
 from fides.api.util.api_router import APIRouter
 from fides.api.util.cache import get_cache
 from fides.api.util.consent_util import create_default_tcf_purpose_overrides_on_startup
-from fides.api.util.endpoint_utils import fides_limiter
 from fides.api.util.errors import FidesError
 from fides.api.util.logger import setup as setup_logging
+from fides.api.util.rate_limit import (
+    RateLimitIPValidationMiddleware,
+    fides_limiter,
+    is_rate_limit_enabled,
+)
 from fides.config import CONFIG
 from fides.config.config_proxy import ConfigProxy
 
@@ -88,7 +92,17 @@ def create_fides_app(
     for handler in ExceptionHandlers.get_handlers():
         # Starlette bug causing this to fail mypy
         fastapi_app.add_exception_handler(RedisNotConfigured, handler)  # type: ignore
-    fastapi_app.add_middleware(SlowAPIMiddleware)
+
+    if is_rate_limit_enabled:
+        # Validate header before SlowAPI processes the request
+        fastapi_app.add_middleware(RateLimitIPValidationMiddleware)
+        # Required for default rate limiting to work
+        fastapi_app.add_middleware(SlowAPIMiddleware)
+    else:
+        logger.warning(
+            "Rate limiting client IPs is disabled because the FIDES__SECURITY__RATE_LIMIT_CLIENT_IP_HEADER env var is not configured."
+        )
+
     fastapi_app.add_middleware(
         GZipMiddleware, minimum_size=1000, compresslevel=5
     )  # minimum_size is in bytes

@@ -18,6 +18,8 @@
 from fideslog.sdk.python.event import AnalyticsEvent
 from loguru import logger
 from pyinstrument import Profiler
+from slowapi import _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
 from starlette.background import BackgroundTask
 from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY
 from uvicorn import Config, Server
@@ -60,6 +62,7 @@
 )
 from fides.api.util.endpoint_utils import API_PREFIX
 from fides.api.util.logger import _log_exception
+from fides.api.util.rate_limit import safe_rate_limit_key
 from fides.cli.utils import FIDES_ASCII_ART
 from fides.config import CONFIG, check_required_webserver_config_values
 
@@ -388,3 +391,22 @@ async def request_validation_exception_handler(
             "detail": jsonable_encoder(exc.errors(), exclude={"input", "url", "ctx"})
         },
     )
+
+
+@app.exception_handler(RateLimitExceeded)
+async def rate_limit_handler(request: Request, exc: RateLimitExceeded) -> Response:
+    """Log rate limit violations and delegate to default handler."""
+    client_ip = safe_rate_limit_key(
+        request
+    )  # non exception-raising, falls back to source IP
+
+    # Log the rate limit event
+    logger.warning(
+        "Rate limit exceeded - IP: %s, Path: %s, Method: %s",
+        client_ip,
+        request.url.path,
+        request.method,
+    )
+
+    # Use the default handler to generate the proper response
+    return _rate_limit_exceeded_handler(request, exc)

@@ -5,8 +5,6 @@
 
 from fastapi import HTTPException
 from fideslang import FidesModelType
-from slowapi import Limiter
-from slowapi.util import get_remote_address  # type: ignore
 from sqlalchemy.ext.asyncio import AsyncSession
 from starlette.status import HTTP_400_BAD_REQUEST
 
@@ -23,7 +21,6 @@
     ORGANIZATION,
     SYSTEM,
 )
-from fides.config import CONFIG
 
 from fides.api.models.sql_models import (  # type: ignore[attr-defined] # isort: skip
     ModelWithDefaultField,
@@ -44,16 +41,6 @@
     "system": SYSTEM,
 }
 
-# Used for rate limiting with Slow API
-# Decorate individual routes to deviate from the default rate limits
-fides_limiter = Limiter(
-    default_limits=[CONFIG.security.request_rate_limit],
-    headers_enabled=True,
-    key_prefix=CONFIG.security.rate_limit_prefix,
-    key_func=get_remote_address,
-    retry_after="http-date",
-)
-
 
 async def forbid_if_editing_is_default(
     sql_model: Base,

@@ -0,0 +1,194 @@
+from __future__ import annotations
+
+from ipaddress import ip_address
+from typing import Optional
+
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from loguru import logger
+from slowapi import Limiter
+from slowapi.util import get_remote_address  # type: ignore
+from starlette.middleware.base import BaseHTTPMiddleware
+
+from fides.config import CONFIG
+
+
+class InvalidClientIPError(Exception):
+    def __init__(self, detail: str, header_value: str, header_name: str):
+        self.detail = detail
+        self.header_value = header_value
+        self.header_name = header_name
+        super().__init__(detail)
+
+
+def validate_client_ip(ip: Optional[str]) -> bool:
+    """
+    Returns true if the provided ip is valid and not from a reserved range.
+    Returns false otherwise.
+    """
+    if not ip:
+        return False
+    try:
+        ip_obj = ip_address(ip)
+        if (
+            ip_obj.is_loopback
+            or ip_obj.is_link_local
+            or ip_obj.is_reserved
+            or ip_obj.is_multicast
+            or ip_obj.is_private
+        ):
+            return False
+        return True
+    except ValueError:
+        return False
+
+
+def _extract_hostname_from_ip(ip: str) -> Optional[str]:
+    """
+    Extract hostname/IP address from header value, stripping port if present.
+
+    Simple string-based approach following the reference implementation pattern.
+    Does not validate whether the result is a valid IP address.
+
+    Examples:
+        # IPv4 cases
+        _extract_hostname_from_ip("192.168.1.1") -> "192.168.1.1"
+        _extract_hostname_from_ip("192.168.1.1:8080") -> "192.168.1.1"
+
+        # IPv6 cases
+        _extract_hostname_from_ip("2001:db8::1") -> "2001:db8::1"
+        _extract_hostname_from_ip("[2001:db8::1]:8080") -> "2001:db8::1"
+
+        # Edge cases (alidation will later reject)
+        _extract_hostname_from_ip("192.168.1.1, 192.168.1.2") -> "192.168.1.1, 192.168.1.2"
+        _extract_hostname_from_ip("not-an-ip:8080") -> "not-an-ip"
+
+        # Error
+        _extract_hostname_from_ip("") -> raises ValueError
+
+    Raises:
+        ValueError: If no hostname can be extracted from the input
+    """
+
+    clean_ip = ip.strip()
+
+    if not clean_ip:
+        raise ValueError("Could not parse IP from header value")
+
+    # Handle IPv6 with port: [IPv6]:port
+    if "]:" in clean_ip:
+        return clean_ip.split("]:")[0].replace("[", "").strip()
+
+    # Handle IPv4 with port: IPv4:port
+    if ":" in clean_ip and "::" not in clean_ip:
+        return clean_ip.split(":")[0].strip()
+
+    # Return as-is (IPv6 without port, IPv4 without port, or other values)
+    return clean_ip
+
+
+def _resolve_client_ip_from_header(request: Request, strict: bool) -> str:
+    """Shared resolver for client IP from the configured header.
+
+    - When strict=True: raise InvalidClientIPError on invalid/malformed header values.
+    - When strict=False: never raise; fall back to the connection source IP.
+    """
+    header_name = CONFIG.security.rate_limit_client_ip_header
+    if not header_name:
+        # This line should never be reached when rate limiting is enabled
+        logger.warning(
+            "Rate limit client IP header not configured. Falling back to source IP.",
+            header_name,
+        )
+        return get_remote_address(request)
+
+    ip_address_from_header = request.headers.get(header_name)
+    if not ip_address_from_header:
+        logger.debug(
+            "Rate limit header '{}' not found. Falling back to source IP.",
+            header_name,
+        )
+        return get_remote_address(request)
+
+    # Extract and validate IP
+    try:
+        extracted_ip = _extract_hostname_from_ip(ip_address_from_header)
+        if extracted_ip and validate_client_ip(extracted_ip):
+            return extracted_ip
+        raise ValueError("IP failed validation")
+    except ValueError:
+        if strict:
+            logger.error(
+                "Invalid IP '{}' in header '{}'. Rejecting request.",
+                ip_address_from_header,
+                header_name,
+            )
+            raise InvalidClientIPError(
+                detail="Invalid IP address format",
+                header_value=ip_address_from_header,
+                header_name=header_name,
+            )
+        # Non-strict path: fall back silently to source IP
+        return get_remote_address(request)
+
+
+def get_client_ip_from_header(request: Request) -> str:
+    """
+    Extracts the client IP from the configured CDN header.
+
+    If the header is not configured or is missing, it falls back to the
+    source IP on the request.
+
+    Raises InvalidClientIPError if header contains invalid IP format.
+    """
+    return _resolve_client_ip_from_header(request, strict=True)
+
+
+def safe_rate_limit_key(request: Request) -> str:
+    """
+    Safe key function for SlowAPI limiter.
+
+    Must never raise. If the configured header is missing or malformed,
+    fall back to the connection source IP for rate limiting purposes.
+    """
+    return _resolve_client_ip_from_header(request, strict=False)
+
+
+class RateLimitIPValidationMiddleware(BaseHTTPMiddleware):
+    """
+    Pre-validate the configured client IP header when rate limiting is enabled.
+
+    If the header is present but invalid, short-circuit the request with 422.
+    This keeps SlowAPI's middleware path free of exceptions from the key function.
+    """
+
+    async def dispatch(self, request: Request, call_next):  # type: ignore
+        if is_rate_limit_enabled:
+            try:
+                # Triggers parsing/validation; raises on invalid header
+                get_client_ip_from_header(request)
+            except InvalidClientIPError:
+                return JSONResponse(
+                    status_code=422, content={"detail": "Invalid client IP header"}
+                )
+        return await call_next(request)
+
+
+# Used for rate limiting with Slow API
+# Decorate individual routes to deviate from the default rate limits
+is_rate_limit_enabled = (
+    CONFIG.security.rate_limit_client_ip_header is not None
+    and CONFIG.security.rate_limit_client_ip_header != ""
+)
+fides_limiter = Limiter(
+    storage_uri=CONFIG.redis.connection_url_unencoded,
+    application_limits=[
+        CONFIG.security.request_rate_limit
+    ],  # Creates ONE shared bucket for all endpoints
+    headers_enabled=True,
+    key_prefix=CONFIG.security.rate_limit_prefix,
+    key_func=safe_rate_limit_key,
+    retry_after="http-date",
+    in_memory_fallback_enabled=False,  # Fall back to no rate limiting if Redis unavailable
+    enabled=is_rate_limit_enabled,
+)

@@ -181,8 +181,24 @@ def resolve_read_only_ssl_ca_certs(
         description="A full connection URL to the read-only Redis cache. If not specified, this URL is automatically assembled from the read_only_host, read_only_port, read_only_password and read_only_db_index specified above.",
         exclude=True,
     )
+    connection_url_unencoded: Optional[str] = Field(
+        default=None,
+        description="A full connection URL to the Redis cache with the password unencoded. If not specified, this URL is automatically assembled from the host, port, password and db_index specified above.",
+        exclude=True,
+    )
+    read_only_connection_url_unencoded: Optional[str] = Field(
+        default=None,
+        description="A full connection URL to the read-only Redis cache with the password unencoded. If not specified, this URL is automatically assembled from the read_only_host, read_only_port, read_only_password and read_only_db_index specified above.",
+        exclude=True,
+    )
 
-    @field_validator("connection_url", "read_only_connection_url", mode="before")
+    @field_validator(
+        "connection_url",
+        "read_only_connection_url",
+        "connection_url_unencoded",
+        "read_only_connection_url_unencoded",
+        mode="before",
+    )
     @classmethod
     def assemble_connection_url(
         cls,
@@ -195,7 +211,14 @@ def assemble_connection_url(
             return v
 
         # Determine which set of settings to use based on field name
-        is_read_only = info.field_name == "read_only_connection_url"
+        is_read_only = info.field_name in (
+            "read_only_connection_url",
+            "read_only_connection_url_unencoded",
+        )
+        is_unencoded = info.field_name in (
+            "connection_url_unencoded",
+            "read_only_connection_url_unencoded",
+        )
 
         # Extract settings - fallbacks already resolved by field validators for read-only fields
         user = (
@@ -244,7 +267,8 @@ def assemble_connection_url(
         # redis://<user>:<password>@<host>
         auth_prefix = ""
         if password or user:
-            auth_prefix = f"{quote_plus(user)}:{quote_plus(password)}@"
+            encoded_password = password if is_unencoded else quote_plus(password)
+            auth_prefix = f"{quote_plus(user)}:{encoded_password}@"
 
         # For host, we don't have a fallback - read replica should be a different host
         host = (

@@ -89,16 +89,16 @@ class SecuritySettings(FidesSettings):
         default=None,
         description="When using a parent/child Fides deployment, this username will be used by the child server to access the parent server.",
     )
-    public_request_rate_limit: str = Field(
-        default="2000/minute",
-        description="The number of requests from a single IP address allowed to hit a public endpoint within the specified time period",
-    )
     rate_limit_prefix: str = Field(
-        default="fides-",
+        default="rate-limit",
         description="The prefix given to keys in the Redis cache used by the rate limiter.",
     )
+    rate_limit_client_ip_header: Optional[str] = Field(
+        default=None,
+        description="The header used to determine the client IP address for rate limiting. If not set or set to empty string, rate limiting will be disabled.",
+    )
     request_rate_limit: str = Field(
-        default="1000/minute",
+        default="2000/minute",
         description="The number of requests from a single IP address allowed to hit an endpoint within a rolling 60 second period.",
     )
     auth_rate_limit: str = Field(
@@ -217,6 +217,21 @@ def assemble_root_access_token(
         oauth_root_client_secret_hash = (hashed_client_id, salt.encode(encoding))  # type: ignore
         return oauth_root_client_secret_hash
 
+    @field_validator("rate_limit_client_ip_header")
+    @classmethod
+    def validate_rate_limit_client_ip_header(
+        cls,
+        v: str,
+    ) -> str:
+        """Validate supported `rate_limit_client_ip_header`"""
+        insecure_headers = ["x-forwarded-for"]
+
+        if v.lower() in insecure_headers:
+            raise ValueError(
+                "The rate_limit_client_ip_header cannot be set to a header that is not secure."
+            )
+        return v
+
     @field_validator("request_rate_limit", "auth_rate_limit")
     @classmethod
     def validate_rate_limits(

@@ -0,0 +1,281 @@
+from unittest import mock
+
+import pytest
+from fastapi import Request
+
+from fides.api.util.rate_limit import (
+    InvalidClientIPError,
+    _extract_hostname_from_ip,
+    get_client_ip_from_header,
+    validate_client_ip,
+)
+from fides.config import CONFIG
+
+
+@pytest.fixture(scope="function")
+def set_rate_limit_client_ip_header_none(db):
+    """Set the rate limit client IP header to None for the duration of a test."""
+    original_value = CONFIG.security.rate_limit_client_ip_header
+    CONFIG.security.rate_limit_client_ip_header = None
+    yield
+    CONFIG.security.rate_limit_client_ip_header = original_value
+
+
+@pytest.fixture(scope="function")
+def set_rate_limit_client_ip_header_x_real_ip(db):
+    """Set the rate limit client IP header to X-Real-IP for the duration of a test."""
+    original_value = CONFIG.security.rate_limit_client_ip_header
+    CONFIG.security.rate_limit_client_ip_header = "X-Real-IP"
+    yield
+    CONFIG.security.rate_limit_client_ip_header = original_value
+
+
+@pytest.fixture(scope="function")
+def set_rate_limit_client_ip_header_cloudfront_viewer_address(db):
+    """Set the rate limit client IP header to CloudFront-Viewer-Address for the duration of a test."""
+    original_value = CONFIG.security.rate_limit_client_ip_header
+    CONFIG.security.rate_limit_client_ip_header = "CloudFront-Viewer-Address"
+    yield
+    CONFIG.security.rate_limit_client_ip_header = original_value
+
+
+class TestValidateClientIp:
+    @pytest.mark.parametrize(
+        "ip",
+        [
+            # Public IPv4 addresses
+            "150.51.100.10",
+            "8.8.8.8",
+            # Public IPv6 addresses
+            "2001:4860:4860::8888",
+            "2606:4700:4700::1111",
+        ],
+    )
+    def test_validate_client_ip_valid(self, ip):
+        assert validate_client_ip(ip) is True
+
+    @pytest.mark.parametrize(
+        "ip",
+        [
+            # Reserved/private/special addresses
+            "198.51.100.10",  # Reserved
+            "2001:db8::1",  # Reserved IPv6
+            "127.0.0.1",  # Loopback
+            "172.16.0.1",  # Private
+            "192.168.1.100",  # Private
+            "10.0.0.5",  # Private
+            "169.254.1.1",  # Link-local
+            "::1",  # Loopback IPv6
+            "fe80::1",  # Link-local IPv6
+            "fc00::1",  # Unique Local IPv6
+            "224.0.0.1",  # Multicast
+            # Invalid formats
+            "not an ip",
+            "",
+            # Multiple IPs
+            "192.168.1.1, 192.168.1.2",
+            # IP with port
+            "192.168.1.1:8080",
+        ],
+    )
+    def test_validate_client_ip_invalid(self, ip):
+        assert validate_client_ip(ip) is False
+
+
+class TestExtractIpFromValue:
+    @pytest.mark.parametrize(
+        "header_value, expected",
+        [
+            # IPv4 without port
+            ("192.168.1.1", "192.168.1.1"),
+            ("150.51.100.10", "150.51.100.10"),
+            # IPv4 with port
+            ("198.51.100.10:46532", "198.51.100.10"),
+            ("192.168.1.1:8080", "192.168.1.1"),
+            ("10.0.0.1:3000", "10.0.0.1"),
+            # IPv6 without port
+            ("2001:db8::1", "2001:db8::1"),
+            ("2001:4860:4860::8888", "2001:4860:4860::8888"),
+            ("::1", "::1"),
+            ("fe80::1", "fe80::1"),
+            # IPv6 with port (bracket notation)
+            ("[2001:db8::1]:8080", "2001:db8::1"),
+            ("[2001:4860:4860::8888]:443", "2001:4860:4860::8888"),
+            ("[::1]:3000", "::1"),
+            ("[fe80::1]:80", "fe80::1"),
+            # Edge cases with whitespace
+            ("  192.168.1.1:8080  ", "192.168.1.1"),
+            ("  [2001:db8::1]:8080  ", "2001:db8::1"),
+            ("  2001:db8::1  ", "2001:db8::1"),
+            # Edge case port without IP
+            (":8080", ""),
+            # Cases that urlparse extracts but may be invalid (validation happens elsewhere)
+            ("not-an-ip:8080", "not-an-ip"),  # Invalid IP with port
+            ("not-an-ip", "not-an-ip"),  # Invalid IP without port
+            (
+                "192.168.1.1:8080:extra",
+                "192.168.1.1",
+            ),  # Extra colon - urlparse handles gracefully
+            # Multiple values - urlparse takes first part before delimiter
+            (
+                "192.168.1.1, 192.168.1.2",
+                "192.168.1.1, 192.168.1.2",
+            ),  # Comma treated as part of hostname
+            (
+                "192.168.1.1 192.168.1.2",
+                "192.168.1.1 192.168.1.2",
+            ),  # Space treated as part of hostname
+        ],
+    )
+    def test_extract_ip_from_value_success(self, header_value, expected):
+        assert _extract_hostname_from_ip(header_value) == expected
+
+    @pytest.mark.parametrize(
+        "header_value",
+        [
+            # Cases where no hostname can be extracted
+            "",  # Empty string
+            " ",  # Just whitespace
+        ],
+    )
+    def test_extract_ip_from_value_failure(self, header_value):
+        with pytest.raises(ValueError, match="Could not parse IP from header value"):
+            _extract_hostname_from_ip(header_value)
+
+
+class TestGetClientIpFromHeader:
+    @pytest.mark.usefixtures("set_rate_limit_client_ip_header_x_real_ip")
+    def test_get_client_ip_from_header_configured_and_present(self, config):
+        mock_request = mock.Mock(spec=Request)
+        mock_request.headers = {"X-Real-IP": "150.51.100.10"}
+        mock_request.client.host = "127.0.0.1"
+        try:
+            result = get_client_ip_from_header(mock_request)
+        except ValueError as err:  # explicit guard: should never raise ValueError
+            pytest.fail(f"Unexpected ValueError: {err}")
+        assert result == "150.51.100.10"
+
+    @pytest.mark.usefixtures("set_rate_limit_client_ip_header_x_real_ip")
+    def test_get_client_ip_from_header_with_port(self, config):
+        mock_request = mock.Mock(spec=Request)
+        mock_request.headers = {"X-Real-IP": "150.51.100.10:46532"}
+        mock_request.client.host = "127.0.0.1"
+        try:
+            result = get_client_ip_from_header(mock_request)
+        except ValueError as err:
+            pytest.fail(f"Unexpected ValueError: {err}")
+        assert result == "150.51.100.10"
+
+    @pytest.mark.usefixtures("set_rate_limit_client_ip_header_x_real_ip")
+    def test_get_client_ip_from_header_ipv6_with_port(self, config):
+        mock_request = mock.Mock(spec=Request)
+        mock_request.headers = {"X-Real-IP": "[2001:4860:4860::8888]:443"}
+        mock_request.client.host = "127.0.0.1"
+        try:
+            result = get_client_ip_from_header(mock_request)
+        except ValueError as err:
+            pytest.fail(f"Unexpected ValueError: {err}")
+        assert result == "2001:4860:4860::8888"
+
+    @pytest.mark.usefixtures(
+        "set_rate_limit_client_ip_header_cloudfront_viewer_address"
+    )
+    def test_get_client_ip_from_header_missing(self, config):
+        mock_request = mock.Mock(spec=Request)
+        mock_request.headers = {}
+        mock_request.client.host = "192.0.2.1"
+        try:
+            result = get_client_ip_from_header(mock_request)
+        except ValueError as err:
+            pytest.fail(f"Unexpected ValueError: {err}")
+        assert result == "192.0.2.1"
+
+    @pytest.mark.usefixtures(
+        "set_rate_limit_client_ip_header_cloudfront_viewer_address"
+    )
+    def test_get_client_ip_from_header_mismatch(self, config):
+        mock_request = mock.Mock(spec=Request)
+        mock_request.headers = {"X-Real-IP": "150.51.100.10:46532"}
+        mock_request.client.host = "192.0.2.1"
+        try:
+            result = get_client_ip_from_header(mock_request)
+        except ValueError as err:
+            pytest.fail(f"Unexpected ValueError: {err}")
+        assert result == "192.0.2.1"
+
+    @pytest.mark.usefixtures("set_rate_limit_client_ip_header_none")
+    def test_get_client_ip_header_not_configured(self, config):
+        mock_request = mock.Mock(spec=Request)
+        mock_request.headers = {"CloudFront-Viewer-Address": "150.51.100.10"}
+        mock_request.client.host = "192.0.2.1"
+        try:
+            result = get_client_ip_from_header(mock_request)
+        except ValueError as err:
+            pytest.fail(f"Unexpected ValueError: {err}")
+        assert result == "192.0.2.1"
+
+    @pytest.mark.usefixtures(
+        "set_rate_limit_client_ip_header_cloudfront_viewer_address"
+    )
+    def test_get_client_ip_from_header_invalid_ip_raises_exception(self, config):
+        mock_request = mock.Mock(spec=Request)
+        mock_request.headers = {
+            "CloudFront-Viewer-Address": "127.0.0.1"
+        }  # Loopback - invalid
+        mock_request.client.host = "192.0.2.1"
+
+        with pytest.raises(InvalidClientIPError) as exc_info:
+            get_client_ip_from_header(mock_request)
+
+        assert exc_info.value.detail == "Invalid IP address format"
+        assert exc_info.value.header_value == "127.0.0.1"
+        assert exc_info.value.header_name == "CloudFront-Viewer-Address"
+
+    @pytest.mark.usefixtures(
+        "set_rate_limit_client_ip_header_cloudfront_viewer_address"
+    )
+    def test_get_client_ip_from_header_multiple_ips_raises_exception(self, config):
+        mock_request = mock.Mock(spec=Request)
+        mock_request.headers = {
+            "CloudFront-Viewer-Address": "203.0.113.195, 70.41.3.18"
+        }
+        mock_request.client.host = "192.0.2.1"
+
+        with pytest.raises(InvalidClientIPError) as exc_info:
+            get_client_ip_from_header(mock_request)
+
+        assert exc_info.value.detail == "Invalid IP address format"
+        assert exc_info.value.header_value == "203.0.113.195, 70.41.3.18"
+        assert exc_info.value.header_name == "CloudFront-Viewer-Address"
+
+    @pytest.mark.usefixtures(
+        "set_rate_limit_client_ip_header_cloudfront_viewer_address"
+    )
+    def test_get_client_ip_from_header_malformed_ip_raises_exception(self, config):
+        mock_request = mock.Mock(spec=Request)
+        mock_request.headers = {"CloudFront-Viewer-Address": "not-an-ip"}
+        mock_request.client.host = "192.0.2.1"
+
+        with pytest.raises(InvalidClientIPError) as exc_info:
+            get_client_ip_from_header(mock_request)
+
+        assert exc_info.value.detail == "Invalid IP address format"
+        assert exc_info.value.header_value == "not-an-ip"
+        assert exc_info.value.header_name == "CloudFront-Viewer-Address"
+
+    @pytest.mark.usefixtures(
+        "set_rate_limit_client_ip_header_cloudfront_viewer_address"
+    )
+    def test_get_client_ip_from_header_private_ip_raises_exception(self, config):
+        mock_request = mock.Mock(spec=Request)
+        mock_request.headers = {
+            "CloudFront-Viewer-Address": "192.168.1.1"
+        }  # Private IP
+        mock_request.client.host = "192.0.2.1"
+
+        with pytest.raises(InvalidClientIPError) as exc_info:
+            get_client_ip_from_header(mock_request)
+
+        assert exc_info.value.detail == "Invalid IP address format"
+        assert exc_info.value.header_value == "192.168.1.1"
+        assert exc_info.value.header_name == "CloudFront-Viewer-Address"

@@ -99,6 +99,10 @@ def test_validate_auth_rate_limit_valid_format(self, monkeypatch):
         settings = SecuritySettings(auth_rate_limit="5 per hour")
         assert settings.auth_rate_limit == "5 per hour"
 
+    def test_validate_rate_limit_client_ip_header_invalid(self):
+        with pytest.raises(ValueError):
+            SecuritySettings(request_rate_limit="X-Forwarded-For")
+
     def test_security_settings_env_default_to_prod(self):
         settings = SecuritySettings()
         assert settings.env == "prod"