Fixes #695: Filter validation (#697)

m0nkeyc0de · web-flow · commit 97f1e8b0c0e7 · 2025-06-23T07:57:32.000-05:00
* Validating filters on filter Endpoint

* Refactor validation. Tests and docs are incomplete

* Tests

* Tests do pass

* Linting with black

* Fix error message

* Corrections after review, add infos in README.md

* Typo

* Remove unused boilerplate

* Remove unused import

---------

Co-authored-by: m0nkey c0de &lt;m0nkey@overfl0w.net&gt;
diff --git a/README.md b/README.md
@@ -73,6 +73,30 @@ nb = pynetbox.api(
     threading=True,
 )
 ```
+### Filters validation
+
+NetBox doesn't validate filters passed to the GET API endpoints, which are accessed with `.get()` and `.filter()`. If a filter is incorrect, NetBox silently returns the entire database table content. Pynetbox allows to check provided parameters against NetBox OpenAPI specification before doing the call, and raise an exception if a parameter is incorrect.
+
+This can be enabled globally by setting `strict_filters=True` in the API object initialization:
+
+```python
+nb = pynetbox.api(
+    'http://localhost:8000',
+    strict_filters=True,
+)
+```
+
+This can also be enabled and disabled on a per-request basis:
+
+```python
+# Disable for one request when enabled globally.
+# Will not raise an exception and return the entire Device table.
+nb.dcim.devices.filter(non_existing_filter="aaaa", strict_filters=False)
+
+# Enable for one request when not enabled globally.
+# Will raise an exception.
+nb.dcim.devices.filter(non_existing_filter="aaaa", strict_filters=True)
+```
 
 ## Running Tests
 
diff --git a/pynetbox/__init__.py b/pynetbox/__init__.py
@@ -1,4 +1,9 @@
 from pynetbox.core.api import Api as api
-from pynetbox.core.query import AllocationError, ContentError, RequestError
+from pynetbox.core.query import (
+    AllocationError,
+    ContentError,
+    RequestError,
+    ParameterValidationError,
+)
 
 __version__ = "7.5.0"
diff --git a/pynetbox/core/api.py b/pynetbox/core/api.py
@@ -79,19 +79,22 @@ def __init__(
         url,
         token=None,
         threading=False,
+        strict_filters=False,
     ):
         """Initialize the API client.
 
         Args:
             url (str): The base URL to the instance of NetBox you wish to connect to.
             token (str, optional): Your NetBox API token. If not provided, authentication will be required for each request.
             threading (bool, optional): Set to True to use threading in `.all()` and `.filter()` requests, defaults to False.
+            strict_filters (bool, optional): Set to True to check GET call filters against OpenAPI specifications (intentionally not done in NetBox API), defaults to False.
         """
         base_url = "{}/api".format(url if url[-1] != "/" else url[:-1])
         self.token = token
         self.base_url = base_url
         self.http_session = requests.Session()
         self.threading = threading
+        self.strict_filters = strict_filters
 
         # Initialize NetBox apps
         self.circuits = App(self, "circuits")
@@ -139,6 +142,7 @@ def openapi(self):
         """Returns the OpenAPI spec.
 
         Quick helper function to pull down the entire OpenAPI spec.
+        It is stored in memory to avoid repeated calls on NetBox API.
 
         ## Returns
         dict: The OpenAPI specification as a dictionary.
@@ -155,10 +159,13 @@ def openapi(self):
         # {...}
         ```
         """
-        return Request(
-            base=self.base_url,
-            http_session=self.http_session,
-        ).get_openapi()
+        if not (openapi := getattr(self, "_openapi", None)):
+            openapi = self._openapi = Request(
+                base=self.base_url,
+                http_session=self.http_session,
+            ).get_openapi()
+
+        return openapi
 
     def status(self):
         """Gets the status information from NetBox.
diff --git a/pynetbox/core/endpoint.py b/pynetbox/core/endpoint.py
@@ -14,7 +14,7 @@
 limitations under the License.
 """
 
-from pynetbox.core.query import Request, RequestError
+from pynetbox.core.query import Request, RequestError, ParameterValidationError
 from pynetbox.core.response import Record, RecordSet
 
 RESERVED_KWARGS = ()
@@ -46,6 +46,7 @@ def __init__(self, api, app, name, model=None):
         self.return_obj = self._lookup_ret_obj(name, model)
         self.name = name.replace("_", "-")
         self.api = api
+        self.app = app
         self.base_url = api.base_url
         self.token = api.token
         self.url = "{base_url}/{app}/{endpoint}".format(
@@ -76,6 +77,58 @@ def _lookup_ret_obj(self, name, model):
             ret = Record
         return ret
 
+    def _validate_openapi_parameters(self, method: str, parameters: dict) -> None:
+        """Validate GET request parameters against OpenAPI specification
+
+        This method raises a **ParameterValidationError** if parameters passed to NetBox API
+        do not match the OpenAPI specification or validation fails.
+
+        ## Parameters
+
+        * **method** : Only "get" is supported as for other methods NetBox already does proper validation
+        * **parameters** : kwargs passed to filter() method
+
+        ## Returns
+        None
+        """
+        if method.lower() != "get":
+            raise RuntimeError(f"Unsupported method '{method}'.")
+
+        openapi_definition_path = "/api/{app}/{endpoint}/".format(
+            app=self.app.name,
+            endpoint=self.name,
+        )
+
+        # Parse NetBox OpenAPI definition
+        try:
+            openapi_definition = self.api.openapi()["paths"].get(
+                openapi_definition_path
+            )
+
+            if not openapi_definition:
+                raise ParameterValidationError(
+                    f"Path '{openapi_definition_path}' does not exist in NetBox OpenAPI specification."
+                )
+
+            openapi_parameters = openapi_definition[method]["parameters"]
+            allowed_parameters = [p["name"] for p in openapi_parameters]
+
+        except KeyError as exc:
+            raise ParameterValidationError(
+                f"Error while parsing Netbox OpenAPI specification: {exc}"
+            )
+
+        # Validate all parameters
+        validation_errors = []
+        for p in parameters:
+            if p not in allowed_parameters:
+                validation_errors.append(
+                    f"'{p}' is not allowed as parameter on path '{openapi_definition_path}'."
+                )
+
+        if len(validation_errors) > 0:
+            raise ParameterValidationError(validation_errors)
+
     def all(self, limit=0, offset=None):
         """Queries the 'ListView' of a given endpoint.
 
@@ -134,6 +187,8 @@ def get(self, *args, **kwargs):
         * **key** (int, optional): id for the item to be retrieved.
         * **kwargs**: Accepts the same keyword args as filter(). Any search argument the endpoint accepts can
             be added as a keyword arg.
+        * **strict_filters** (bool, optional): Overrides the global filter
+            validation per-request basis. Handled by the filter() method.
 
         ## Returns
         A single Record object or None
@@ -164,7 +219,6 @@ def get(self, *args, **kwargs):
         # Row 1
         ```
         """
-
         try:
             key = args[0]
         except IndexError:
@@ -217,6 +271,8 @@ def filter(self, *args, **kwargs):
             be returned with each query to the Netbox server. The queries
             will be made as you iterate through the result set.
         * **offset** (int, optional): Overrides the offset on paginated returns.
+        * **strict_filters** (bool, optional): Overrides the global filter
+            validation per-request basis.
 
         ## Returns
         A RecordSet object.
@@ -272,9 +328,20 @@ def filter(self, *args, **kwargs):
             )
         limit = kwargs.pop("limit") if "limit" in kwargs else 0
         offset = kwargs.pop("offset") if "offset" in kwargs else None
+        strict_filters = (
+            # kwargs value takes precedence on globally set value
+            kwargs.pop("strict_filters")
+            if "strict_filters" in kwargs
+            else self.api.strict_filters
+        )
+
         if limit == 0 and offset is not None:
             raise ValueError("offset requires a positive limit value")
         filters = {x: y if y is not None else "null" for x, y in kwargs.items()}
+
+        if strict_filters:
+            self._validate_openapi_parameters("get", filters)
+
         req = Request(
             filters=filters,
             base=self.url,
diff --git a/pynetbox/core/query.py b/pynetbox/core/query.py
@@ -109,6 +109,29 @@ def __str__(self):
         return self.error
 
 
+class ParameterValidationError(Exception):
+    """API parameter validation Exception.
+
+    Raised when filter parameters do not match Netbox OpenAPI specification.
+
+    ## Examples
+
+    ```python
+    try:
+        nb.dcim.devices.filter(field_which_does_not_exist="destined-for-failure")
+    except pynetbox.ParameterValidationError as e:
+        print(e.error)
+    ```
+    """
+
+    def __init__(self, errors):
+        super().__init__(errors)
+        self.error = f"The request parameter validation returned an error: {errors}"
+
+    def __str__(self):
+        return self.error
+
+
 class Request:
     """Creates requests to the Netbox API.
 
diff --git a/tests/unit/test_endpoint.py b/tests/unit/test_endpoint.py
@@ -9,7 +9,7 @@ def test_filter(self):
         with patch(
             "pynetbox.core.query.Request._make_call", return_value=Mock()
         ) as mock:
-            api = Mock(base_url="http://localhost:8000/api")
+            api = Mock(base_url="http://localhost:8000/api", strict_filters=False)
             app = Mock(name="test")
             mock.return_value = [{"id": 123}, {"id": 321}]
             test_obj = Endpoint(api, app, "test")
@@ -24,7 +24,7 @@ def test_filter_invalid_pagination_args(self):
             test_obj.filter(offset=1)
 
     def test_filter_replace_none_with_null(self):
-        api = Mock(base_url="http://localhost:8000/api")
+        api = Mock(base_url="http://localhost:8000/api", strict_filters=False)
         app = Mock(name="test")
         test_obj = Endpoint(api, app, "test")
         test = test_obj.filter(name=None, id=0)
@@ -118,7 +118,7 @@ def test_get_with_filter(self):
             "pynetbox.core.query.Request._make_call", return_value=Mock()
         ) as mock:
             mock.return_value = [{"id": 123}]
-            api = Mock(base_url="http://localhost:8000/api")
+            api = Mock(base_url="http://localhost:8000/api", strict_filters=False)
             app = Mock(name="test")
             test_obj = Endpoint(api, app, "test")
             test = test_obj.get(name="test")
@@ -181,7 +181,7 @@ def test_get_greater_than_one(self):
             "pynetbox.core.query.Request._make_call", return_value=Mock()
         ) as mock:
             mock.return_value = [{"id": 123}, {"id": 321}]
-            api = Mock(base_url="http://localhost:8000/api")
+            api = Mock(base_url="http://localhost:8000/api", strict_filters=False)
             app = Mock(name="test")
             test_obj = Endpoint(api, app, "test")
             with self.assertRaises(ValueError) as _:
@@ -192,7 +192,7 @@ def test_get_no_results(self):
             "pynetbox.core.query.Request._make_call", return_value=Mock()
         ) as mock:
             mock.return_value = []
-            api = Mock(base_url="http://localhost:8000/api")
+            api = Mock(base_url="http://localhost:8000/api", strict_filters=False)
             app = Mock(name="test")
             test_obj = Endpoint(api, app, "test")
             test = test_obj.get(name="test")
diff --git a/tests/unit/test_endpoint_strict_filter.py b/tests/unit/test_endpoint_strict_filter.py
@@ -0,0 +1,57 @@
+import unittest
+from unittest.mock import Mock
+from tests.util import openapi_mock
+
+from pynetbox.core.endpoint import Endpoint
+from pynetbox import ParameterValidationError
+
+
+class StrictFilterTestCase(unittest.TestCase):
+
+    def test_filter_strict_valid_kwargs(self):
+        api = Mock(
+            base_url="http://localhost:8000/api",
+            strict_filters=True,
+            openapi=openapi_mock,
+        )
+        app = Mock(name="test")
+        app.name = "test"
+        test_obj = Endpoint(api, app, "test")
+        test_obj.filter(test="test")
+
+    def test_filter_strict_invalid_kwarg(self):
+        api = Mock(
+            base_url="http://localhost:8000/api",
+            strict_filters=True,
+            openapi=openapi_mock,
+        )
+        app = Mock(name="test")
+        app.name = "test"
+        test_obj = Endpoint(api, app, "test")
+        with self.assertRaises(ParameterValidationError):
+            test_obj.filter(very_invalid_kwarg="test")
+
+    def test_filter_strict_per_request_disable_invalid_kwarg(self):
+        api = Mock(
+            base_url="http://localhost:8000/api",
+            strict_filters=True,  # Enable globally
+            openapi=openapi_mock,
+        )
+        app = Mock(name="test")
+        app.name = "test"
+        test_obj = Endpoint(api, app, "test")
+        # Disable strict_filters only for this specific request
+        test_obj.filter(very_invalid_kwarg="test", strict_filters=False)
+
+    def test_filter_strict_per_request_enable_invalid_kwarg(self):
+        api = Mock(
+            base_url="http://localhost:8000/api",
+            strict_filters=False,  # Disable globally
+            openapi=openapi_mock,
+        )
+        app = Mock(name="test")
+        app.name = "test"
+        test_obj = Endpoint(api, app, "test")
+        with self.assertRaises(ParameterValidationError):
+            # Enable strict_filters only for this specific request
+            test_obj.filter(very_invalid_kwarg="test", strict_filters=True)
diff --git a/tests/util.py b/tests/util.py
@@ -15,3 +15,19 @@ def load_fixture(self, path):
 
     def json(self):
         return json.loads(self.content)
+
+
+def openapi_mock():
+    """Mock function to simulate Api.openapi()"""
+    return {
+        "paths": {
+            "/api/test/test/": {
+                "get": {
+                    "parameters": [
+                        {"name": "test"},
+                        {"name": "name"},
+                    ]
+                },
+            },
+        },
+    }
