feat: secure IP detection and configurable reverse proxy support

- Add `UVICORN_PROXY_HEADERS` and `UVICORN_FORWARDED_ALLOW_IPS` settings.
- Configure Uvicorn and middleware to respect these settings (defaulting to restricted access).
- Refactor `get_client_ip` to rely on trusted `request.client.host` instead of manual header parsing.
- Prevent IP spoofing by removing unverified `X-Forwarded-For` trust.
- Add comprehensive tests for IP detection and middleware behavior

Author: M03ED

app/middlewares/__init__.py

@@ -3,7 +3,7 @@
 from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
 
 from app.utils.logger import get_logger
-from config import cors_settings
+from config import cors_settings, server_settings
 
 from .request_logging import RequestProcessTimeLoggingMiddleware
 
@@ -16,5 +16,6 @@ def setup_middleware(app: FastAPI):
         allow_methods=["*"],
         allow_headers=["*"],
     )
-    app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*")
+    if server_settings.proxy_headers:
+        app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=server_settings.forwarded_allow_ips)
     app.add_middleware(RequestProcessTimeLoggingMiddleware, access_logger=get_logger("uvicorn.access"))

app/routers/admin.py

@@ -12,20 +12,19 @@
     AdminListQuery,
     AdminModify,
     AdminSimpleListQuery,
-    Token,
-    AdminUsageQuery,
     AdminsResponse,
     AdminsSimpleResponse,
+    AdminUsageQuery,
     BulkAdminsActionResponse,
     BulkAdminSelection,
     RemoveAdminsResponse,
+    Token,
 )
 from app.models.stats import UserUsageStatsList
 from app.operation import OperatorType
 from app.operation.admin import AdminOperation
 from app.utils import responses
 from app.utils.jwt import create_admin_token
-from .dependencies import get_admin_list_query, get_admin_simple_list_query, get_admin_usage_query
 
 from .authentication import (
     check_sudo_admin,
@@ -34,16 +33,14 @@
     validate_admin,
     validate_mini_app_admin,
 )
+from .dependencies import get_admin_list_query, get_admin_simple_list_query, get_admin_usage_query
 
 router = APIRouter(tags=["Admin"], prefix="/api/admin", responses={401: responses._401, 403: responses._403})
 admin_operator = AdminOperation(operator_type=OperatorType.API)
 
 
 def get_client_ip(request: Request) -> str:
-    """Extract the client's IP address from the request headers or client."""
-    forwarded_for = request.headers.get("X-Forwarded-For")
-    if forwarded_for:
-        return forwarded_for.split(",")[0].strip()
+    """Extract the client's IP address from the request."""
     if request.client:
         return request.client.host
     return "Unknown"

config.py

@@ -53,6 +53,8 @@ class ServerSettings(EnvSettings):
     ssl_ca_type: str = Field(default="public", validation_alias="UVICORN_SSL_CA_TYPE")
     workers: int = Field(default=1, validation_alias="UVICORN_WORKERS")
     loop: str = Field(default="auto", validation_alias="UVICORN_LOOP")
+    proxy_headers: bool = Field(default=False, validation_alias="UVICORN_PROXY_HEADERS")
+    forwarded_allow_ips: str | list[str] = Field(default="127.0.0.1", validation_alias="UVICORN_FORWARDED_ALLOW_IPS")
 
     @field_validator("ssl_ca_type")
     @classmethod

main.py

@@ -9,9 +9,8 @@
 from cryptography.hazmat.backends import default_backend
 
 from app import create_app  # noqa: F401
-from app.utils.logger import get_logger
 from app.nats import require_nats_if_multiworker
-from app.utils.logger import LOGGING_CONFIG
+from app.utils.logger import LOGGING_CONFIG, get_logger
 from config import logging_settings, runtime_settings, server_settings
 
 logger = get_logger("uvicorn-main")
@@ -130,7 +129,7 @@ def validate_cert_and_key(cert_file_path, key_file_path, ca_type: str = "public"
 
 If you need external access, please provide the SSL files to allow the server to bind to 0.0.0.0. Alternatively, you can run the server on localhost or a Unix socket and use a reverse proxy, such as Nginx or Caddy, to handle SSL termination and provide external access.
 
-If you wish to continue without SSL, you can use SSH port forwarding to access the application from your machine. note that in this case, subscription functionality will not work. 
+If you wish to continue without SSL, you can use SSH port forwarding to access the application from your machine. note that in this case, subscription functionality will not work.
 
 Use the following command:
 
@@ -160,6 +159,8 @@ def validate_cert_and_key(cert_file_path, key_file_path, ca_type: str = "public"
             log_config=LOGGING_CONFIG,
             log_level=effective_log_level.lower(),
             loop=server_settings.loop,
+            proxy_headers=server_settings.proxy_headers,
+            forwarded_allow_ips=server_settings.forwarded_allow_ips,
         )
     except FileNotFoundError:  # to prevent error on removing unix sock
         pass

tests/api/test_ip_detection.py

@@ -0,0 +1,92 @@
+from fastapi import FastAPI, Request
+from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
+
+from app.middlewares import setup_middleware
+from app.routers.admin import get_client_ip
+from config import server_settings
+from tests.api import client
+
+
+def test_get_client_ip_no_proxy():
+    """Test that get_client_ip returns the direct client host when no proxy is involved."""
+    # Use a real Request object with a minimal ASGI scope
+    scope = {
+        "type": "http",
+        "client": ("1.1.1.1", 12345),
+        "headers": [],
+    }
+    request = Request(scope=scope)
+    assert get_client_ip(request) == "1.1.1.1"
+
+
+def test_get_client_ip_with_proxy_middleware_behavior(access_token):
+    """
+    Test the behavior of IP detection via the TestClient.
+    Since we enabled proxy_headers in conftest.py, the middleware should process X-Forwarded-For.
+    """
+
+    # We use an endpoint that returns the IP, like /api/admin/token (indirectly via notification)
+    # or we can check a subscription info endpoint which also uses request.client.host
+
+    # In tests/conftest.py we set:
+    # server_settings.proxy_headers = True
+    # server_settings.forwarded_allow_ips = "*"
+
+    # This means X-Forwarded-For should be trusted.
+    ip = "203.0.113.10"
+    # We use a known endpoint that returns IP in response for easy verification
+    # Based on grep, user_subscription_info returns IP
+
+    from tests.api.helpers import (
+        create_core,
+        create_group,
+        create_user,
+        delete_core,
+        delete_group,
+        delete_user,
+    )
+
+    core = create_core(access_token)
+    group = create_group(access_token)
+    user = create_user(
+        access_token,
+        group_ids=[group["id"]],
+        payload={"username": "iptestuser"},
+    )
+    try:
+        response = client.get(f"{user['subscription_url']}/info", headers={"X-Forwarded-For": ip})
+        assert response.status_code == 200
+        # The subscription router uses request.client.host
+        # If ProxyHeadersMiddleware is working, it will swap request.client.host with the value from X-Forwarded-For
+        assert response.json()["ip"] == ip
+    finally:
+        delete_user(access_token, "iptestuser")
+        delete_group(access_token, group["id"])
+        delete_core(access_token, core["id"])
+
+
+def test_proxy_headers_disabled_logic():
+    """
+    Verify that if proxy_headers was False, the middleware wouldn't be added.
+    This is a logic check on the middleware setup.
+    """
+
+    app = FastAPI()
+    original_val = server_settings.proxy_headers
+    try:
+        server_settings.proxy_headers = False
+        setup_middleware(app)
+
+        # Check if ProxyHeadersMiddleware is in the app.user_middleware list
+        # Note: Middleware are wrapped, but we can check the classes
+        middleware_classes = [m.cls for m in app.user_middleware]
+        assert ProxyHeadersMiddleware not in middleware_classes
+
+        # Now enable it
+        app_with_proxy = FastAPI()
+        server_settings.proxy_headers = True
+        setup_middleware(app_with_proxy)
+        middleware_classes_proxy = [m.cls for m in app_with_proxy.user_middleware]
+        assert ProxyHeadersMiddleware in middleware_classes_proxy
+    finally:
+        server_settings.proxy_headers = original_val

tests/conftest.py

@@ -10,10 +10,12 @@
 sys.path.insert(0, project_root)
 
 # Override settings for tests
-from config import auth_settings, runtime_settings  # noqa: E402
+from config import auth_settings, runtime_settings, server_settings  # noqa: E402
 
 runtime_settings.testing = True
 runtime_settings.debug = True
+server_settings.proxy_headers = True
+server_settings.forwarded_allow_ips = "*"
 auth_settings.sudoers["testadmin"] = "testadmin"