Range-validate URL query parameters at parse time in the dqlite dialect

create_connect_args coerced each URL query value through a type
converter but did not catch semantic out-of-range values: ?timeout=0,
?timeout=nan, ?timeout=-5, ?max_total_rows=-1, ?max_continuation_frames=0
all sailed through URL parsing and only surfaced as a ValueError
deep inside the DBAPI constructor. At that point the error origin is
a nested coroutine, not the URL the user actually wrote.

Extend the allowed-keys table to (converter, validator) tuples and run
the validator after coercion; an out-of-range value raises
ArgumentError at parse time with a message that points straight at the
offending URL parameter. The downstream DBAPI validation stays in
place as a second layer. Parametric regression tests cover every
numeric knob and accepted / rejected ranges.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/sqlalchemydqlite/base.py

@@ -2,6 +2,7 @@
 
 import contextlib
 import datetime
+import math
 import warnings
 from collections.abc import Callable
 from typing import Any
@@ -126,23 +127,32 @@ def import_dbapi(cls) -> Any:
 
     # Whitelist of URL query parameters we forward to the DBAPI connect
     # call. Unknown keys raise ``ArgumentError`` so typos surface.
+    # Each entry pairs a string-to-value converter with an optional
+    # predicate that runs after coercion to catch semantic out-of-range
+    # values (zero, negative, NaN, inf). The predicate may be ``None`` for
+    # bool knobs that don't admit a range check.
     # ``trust_server_heartbeat`` uses a URL-friendly bool parser because
     # bool("False") evaluates truthy (non-empty string).
-    _URL_QUERY_ALLOWED: dict[str, Callable[[str], Any]] = {
-        "timeout": float,
-        "max_total_rows": int,
-        "max_continuation_frames": int,
-        "trust_server_heartbeat": lambda s: s.strip().lower() in ("1", "true", "yes", "on"),
+    _URL_QUERY_ALLOWED: dict[str, tuple[Callable[[str], Any], Callable[[Any], bool] | None]] = {
+        "timeout": (float, lambda v: math.isfinite(v) and v > 0),
+        "max_total_rows": (int, lambda v: v > 0),
+        "max_continuation_frames": (int, lambda v: v > 0),
+        "trust_server_heartbeat": (
+            lambda s: s.strip().lower() in ("1", "true", "yes", "on"),
+            None,
+        ),
     }
 
     def create_connect_args(self, url: URL) -> tuple[list[Any], dict[str, Any]]:
         """Create connection arguments from URL.
 
         URL format: ``dqlite://host:port/database?timeout=...``
 
-        Known query parameters are typed and passed through to the
-        DBAPI. Unknown ones raise :class:`ArgumentError` so typos like
-        ``?timeoutt=5`` don't silently disappear.
+        Known query parameters are typed and range-validated at URL-parse
+        time so typos (``?timeoutt=5``), unparseable types
+        (``?timeout=abc``), and out-of-range values
+        (``?max_total_rows=-1``) all raise :class:`ArgumentError` before
+        any pool is built.
         """
         host = url.host or "localhost"
         port = url.port or 9001
@@ -158,17 +168,20 @@ def create_connect_args(self, url: URL) -> tuple[list[Any], dict[str, Any]]:
                     f"Unknown dqlite URL query parameter {key!r}. "
                     f"Allowed: {sorted(self._URL_QUERY_ALLOWED)}"
                 )
-            converter = self._URL_QUERY_ALLOWED[key]
+            converter, validator = self._URL_QUERY_ALLOWED[key]
             # URL query values can be str or tuple[str, ...] (when a key
             # appears multiple times). Take the last occurrence.
             raw_str = raw[-1] if isinstance(raw, tuple) else raw
             try:
-                kwargs[key] = converter(raw_str)
+                value = converter(raw_str)
             except (TypeError, ValueError) as e:
                 raise ArgumentError(
                     f"Cannot convert URL query {key}={raw!r} to "
                     f"{getattr(converter, '__name__', 'expected type')}: {e}"
                 ) from e
+            if validator is not None and not validator(value):
+                raise ArgumentError(f"URL query {key}={raw_str!r} is out of range")
+            kwargs[key] = value
 
         return [], kwargs
 

tests/test_dialect_dialect_config.py

@@ -127,6 +127,43 @@ def test_trust_server_heartbeat_parses_boolean(self, raw: str, expected: bool) -
         assert kwargs["trust_server_heartbeat"] is expected
 
 
+class TestURLQueryRangeValidation:
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "dqlite://host:19001/db?max_total_rows=0",
+            "dqlite://host:19001/db?max_total_rows=-1",
+            "dqlite://host:19001/db?max_continuation_frames=0",
+            "dqlite://host:19001/db?max_continuation_frames=-5",
+            "dqlite://host:19001/db?timeout=0",
+            "dqlite://host:19001/db?timeout=-1.5",
+            "dqlite://host:19001/db?timeout=nan",
+            "dqlite://host:19001/db?timeout=inf",
+            "dqlite://host:19001/db?timeout=-inf",
+        ],
+    )
+    def test_invalid_range_rejected_at_parse_time(self, url: str) -> None:
+        dialect = DqliteDialect()
+        with pytest.raises(ArgumentError, match="out of range"):
+            dialect.create_connect_args(make_url(url))
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "dqlite://host:19001/db?max_total_rows=1",
+            "dqlite://host:19001/db?max_total_rows=10000000",
+            "dqlite://host:19001/db?max_continuation_frames=1",
+            "dqlite://host:19001/db?timeout=0.001",
+            "dqlite://host:19001/db?timeout=3600",
+        ],
+    )
+    def test_valid_range_accepted(self, url: str) -> None:
+        dialect = DqliteDialect()
+        _, kwargs = dialect.create_connect_args(make_url(url))
+        # Just smoke-test that parsing completed without error.
+        assert kwargs
+
+
 class TestURLGovernorsReachAioDbapi:
     """End-to-end test: every URL governor knob must be accepted by the
     async DBAPI's connect(). The unit-level create_connect_args tests only