Validate page and pageSize query parameters (#752)

Co-authored-by: Amin Alaee <[email protected]>

Author: BhuwanPandey

sqladmin/application.py

@@ -440,6 +440,14 @@ async def list(self, request: Request) -> Response:
         pagination = await model_view.list(request)
         pagination.add_pagination_urls(request.url)
 
+        if (
+            pagination.page * pagination.page_size
+            > pagination.count + pagination.page_size
+        ):
+            raise HTTPException(
+                status_code=400, detail="Invalid page or pageSize parameter"
+            )
+
         context = {"model_view": model_view, "pagination": pagination}
         return await self.templates.TemplateResponse(
             request, model_view.list_template, context

sqladmin/models.py

@@ -25,6 +25,7 @@
 from sqlalchemy.sql.elements import ClauseElement
 from sqlalchemy.sql.expression import Select, select
 from starlette.datastructures import URL
+from starlette.exceptions import HTTPException
 from starlette.requests import Request
 from starlette.responses import StreamingResponse
 from wtforms import Field, Form
@@ -746,15 +747,26 @@ def _default_formatter(self, value: Any) -> Any:
 
         return value
 
+    def validate_page_number(self, number: Union[str, None], default: int) -> int:
+        if not number:
+            return default
+
+        try:
+            return int(number)
+        except ValueError:
+            raise HTTPException(
+                status_code=400, detail="Invalid page or pageSize parameter"
+            )
+
     async def count(self, request: Request, stmt: Optional[Select] = None) -> int:
         if stmt is None:
             stmt = self.count_query(request)
         rows = await self._run_query(stmt)
         return rows[0]
 
     async def list(self, request: Request) -> Pagination:
-        page = int(request.query_params.get("page", 1))
-        page_size = int(request.query_params.get("pageSize", 0))
+        page = self.validate_page_number(request.query_params.get("page"), 1)
+        page_size = self.validate_page_number(request.query_params.get("pageSize"), 0)
         page_size = min(page_size or self.page_size, max(self.page_size_options))
         search = request.query_params.get("search", None)
 

tests/test_application.py

@@ -1,3 +1,6 @@
+from typing import Generator
+
+import pytest
 from sqlalchemy import Column, Integer, String
 from sqlalchemy.orm import declarative_base
 from starlette.applications import Starlette
@@ -27,6 +30,13 @@ class User(Base):
     name = Column(String(32), default="SQLAdmin")
 
 
+@pytest.fixture(autouse=True)
+def prepare_database() -> Generator[None, None, None]:
+    Base.metadata.create_all(engine)
+    yield
+    Base.metadata.drop_all(engine)
+
+
 def test_application_title() -> None:
     app = Starlette()
     Admin(app=app, engine=engine)
@@ -153,3 +163,21 @@ class DataModelAdmin(ModelView, model=DataModel):
     assert admin._denormalize_wtform_data({"data_": "abcdef"}, datamodel) == {
         "data": "abcdef"
     }
+
+
+def test_validate_page_and_page_size():
+    app = Starlette()
+    admin = Admin(app=app, engine=engine)
+
+    class UserAdmin(ModelView, model=User):
+        ...
+
+    admin.add_view(UserAdmin)
+
+    client = TestClient(app)
+
+    response = client.get("/admin/user/list?page=10000")
+    assert response.status_code == 400
+
+    response = client.get("/admin/user/list?page=aaaa")
+    assert response.status_code == 400