feat: implement directions validation to converters

Signed-off-by: Adria Montoto <75563346+adriamontoto@users.noreply.github.com>

Author: adriamontoto

criteria_pattern/converters/criteria_to_mysql_converter.py

@@ -6,7 +6,7 @@
 from typing import Any, assert_never
 
 from criteria_pattern import Criteria, Direction, Operator
-from criteria_pattern.errors import InvalidColumnError, InvalidOperatorError, InvalidTableError
+from criteria_pattern.errors import InvalidColumnError, InvalidDirectionError, InvalidOperatorError, InvalidTableError
 from criteria_pattern.models.criteria import AndCriteria, NotCriteria, OrCriteria
 
 
@@ -42,9 +42,11 @@ def convert(
         check_column_injection: bool = False,
         check_criteria_injection: bool = False,
         check_operator_injection: bool = False,
+        check_direction_injection: bool = False,
         valid_tables: Sequence[str] | None = None,
         valid_columns: Sequence[str] | None = None,
         valid_operators: Sequence[Operator] | None = None,
+        valid_directions: Sequence[Direction] | None = None,
     ) -> tuple[str, list[Any]]:
         """
         Convert the Criteria object to a MySQL query.
@@ -62,14 +64,18 @@ def convert(
             Default to False.
             check_operator_injection (bool, optional): Raise an error if the operator is not in the list of valid
             operators. Default to False.
+            check_direction_injection (bool, optional): Raise an error if the direction is not in the list of valid
+            directions. Default to False.
             valid_tables (Sequence[str], optional): List of valid tables to query. Default to empty list.
             valid_columns (Sequence[str], optional): List of valid columns to select. Default to empty list.
             valid_operators (Sequence[Operator], optional): List of valid operators to use. Default to empty list.
+            valid_directions (Sequence[Direction], optional): List of valid directions to use. Default to empty list.
 
         Raises:
             InvalidTableError: If the table is not in the list of valid tables (only if check_table_injection=True).
             InvalidColumnError: If the column is not in the list of valid columns (only if check_column_injection=True).
             InvalidOperatorError: If the operator is not in the list of valid operators (only if check_operator_injection=True).
+            InvalidDirectionError: If the direction is not in the list of valid directions (only if check_direction_injection=True).
 
         Returns:
             tuple[str, list[Any]]: The MySQL query string and the query parameters as a list.
@@ -95,6 +101,7 @@ def convert(
         valid_tables = valid_tables or []
         valid_columns = valid_columns or []
         valid_operators = valid_operators or []
+        valid_directions = valid_directions or []
 
         if check_table_injection:
             cls._validate_table(table=table, valid_tables=valid_tables)
@@ -108,6 +115,9 @@ def convert(
         if check_operator_injection:
             cls._validate_operators(criteria=criteria, valid_operators=valid_operators)
 
+        if check_direction_injection:
+            cls._validate_directions(criteria=criteria, valid_directions=valid_directions)
+
         query = f'SELECT {", ".join(columns)} FROM {table}'  # noqa: S608  # nosec
         parameters: list[Any] = []
 
@@ -205,6 +215,25 @@ def _validate_operators(cls, *, criteria: Criteria, valid_operators: Sequence[Op
             if filter.operator not in valid_operators:
                 raise InvalidOperatorError(operator=Operator(value=filter.operator), valid_operators=valid_operators)
 
+    @classmethod
+    def _validate_directions(cls, *, criteria: Criteria, valid_directions: Sequence[Direction]) -> None:
+        """
+        Validate the Criteria object directions to prevent SQL injection.
+
+        Args:
+            criteria (Criteria): Criteria to validate.
+            valid_directions (Sequence[Direction]): List of valid directions to use.
+
+        Raises:
+            InvalidDirectionError: If the direction is not in the list of valid directions.
+        """
+        for order in criteria.orders:
+            if order.direction not in valid_directions:
+                raise InvalidDirectionError(
+                    direction=Direction(value=order.direction),
+                    valid_directions=valid_directions,
+                )
+
     @classmethod
     def _process_filters(cls, *, criteria: Criteria, columns_mapping: Mapping[str, str]) -> tuple[str, list[Any]]:
         """

criteria_pattern/converters/criteria_to_postgresql_converter.py

@@ -6,7 +6,7 @@
 from typing import Any, assert_never
 
 from criteria_pattern import Criteria, Direction, Operator
-from criteria_pattern.errors import InvalidColumnError, InvalidOperatorError, InvalidTableError
+from criteria_pattern.errors import InvalidColumnError, InvalidDirectionError, InvalidOperatorError, InvalidTableError
 from criteria_pattern.models.criteria import AndCriteria, NotCriteria, OrCriteria
 
 
@@ -42,9 +42,11 @@ def convert(
         check_column_injection: bool = False,
         check_criteria_injection: bool = False,
         check_operator_injection: bool = False,
+        check_direction_injection: bool = False,
         valid_tables: Sequence[str] | None = None,
         valid_columns: Sequence[str] | None = None,
         valid_operators: Sequence[Operator] | None = None,
+        valid_directions: Sequence[Direction] | None = None,
     ) -> tuple[str, dict[str, Any]]:
         """
         Convert the Criteria object to a Postgresql query.
@@ -62,14 +64,18 @@ def convert(
             Default to False.
             check_operator_injection (bool, optional): Raise an error if the operator is not in the list of valid operators.
             Default to False.
+            check_direction_injection (bool, optional): Raise an error if the direction is not in the list of valid
+            directions. Default to False.
             valid_tables (Sequence[str], optional): List of valid tables to query. Default to empty list.
             valid_columns (Sequence[str], optional): List of valid columns to select. Default to empty list.
             valid_operators (Sequence[Operator], optional): List of valid operators to use. Default to empty list.
+            valid_directions (Sequence[Direction], optional): List of valid directions to use. Default to empty list.
 
         Raises:
             InvalidTableError: If the table is not in the list of valid tables (only if check_table_injection=True).
             InvalidColumnError: If the column is not in the list of valid columns (only if check_column_injection=True).
             InvalidOperatorError: If the operator is not in the list of valid operators (only if check_operator_injection=True).
+            InvalidDirectionError: If the direction is not in the list of valid directions (only if check_direction_injection=True).
 
         Returns:
             tuple[str, dict[str, Any]]: The Postgresql query string and the query parameters.
@@ -95,6 +101,7 @@ def convert(
         valid_tables = valid_tables or []
         valid_columns = valid_columns or []
         valid_operators = valid_operators or []
+        valid_directions = valid_directions or []
 
         if check_table_injection:
             cls._validate_table(table=table, valid_tables=valid_tables)
@@ -108,6 +115,9 @@ def convert(
         if check_operator_injection:
             cls._validate_operators(criteria=criteria, valid_operators=valid_operators)
 
+        if check_direction_injection:
+            cls._validate_directions(criteria=criteria, valid_directions=valid_directions)
+
         quoted_columns = ['*' if column == '*' else f'"{column}"' for column in columns]
         quoted_table = '.'.join(f'"{part}"' for part in table.split('.'))
         query = f'SELECT {", ".join(quoted_columns)} FROM {quoted_table}'  # noqa: S608  # nosec
@@ -207,6 +217,25 @@ def _validate_operators(cls, *, criteria: Criteria, valid_operators: Sequence[Op
             if filter.operator not in valid_operators:
                 raise InvalidOperatorError(operator=Operator(value=filter.operator), valid_operators=valid_operators)
 
+    @classmethod
+    def _validate_directions(cls, *, criteria: Criteria, valid_directions: Sequence[Direction]) -> None:
+        """
+        Validate the Criteria object directions to prevent SQL injection.
+
+        Args:
+            criteria (Criteria): Criteria to validate.
+            valid_directions (Sequence[Direction]): List of valid directions to use.
+
+        Raises:
+            InvalidDirectionError: If the direction is not in the list of valid directions.
+        """
+        for order in criteria.orders:
+            if order.direction not in valid_directions:
+                raise InvalidDirectionError(
+                    direction=Direction(value=order.direction),
+                    valid_directions=valid_directions,
+                )
+
     @classmethod
     def _process_filters(cls, *, criteria: Criteria, columns_mapping: Mapping[str, str]) -> tuple[str, dict[str, Any]]:
         """

criteria_pattern/converters/criteria_to_sqlite_converter.py

@@ -6,7 +6,7 @@
 from typing import Any, assert_never
 
 from criteria_pattern import Criteria, Direction, Operator
-from criteria_pattern.errors import InvalidColumnError, InvalidOperatorError, InvalidTableError
+from criteria_pattern.errors import InvalidColumnError, InvalidDirectionError, InvalidOperatorError, InvalidTableError
 from criteria_pattern.models.criteria import AndCriteria, NotCriteria, OrCriteria
 
 
@@ -42,9 +42,11 @@ def convert(
         check_column_injection: bool = False,
         check_criteria_injection: bool = False,
         check_operator_injection: bool = False,
+        check_direction_injection: bool = False,
         valid_tables: Sequence[str] | None = None,
         valid_columns: Sequence[str] | None = None,
         valid_operators: Sequence[Operator] | None = None,
+        valid_directions: Sequence[Direction] | None = None,
     ) -> tuple[str, dict[str, Any]]:
         """
         Convert the Criteria object to a SQLite query.
@@ -62,14 +64,18 @@ def convert(
             Default to False.
             check_operator_injection (bool, optional): Raise an error if the operator is not in the list of valid
             operators. Default to False.
+            check_direction_injection (bool, optional): Raise an error if the direction is not in the list of valid
+            directions. Default to False.
             valid_tables (Sequence[str], optional): List of valid tables to query. Default to empty list.
             valid_columns (Sequence[str], optional): List of valid columns to select. Default to empty list.
             valid_operators (Sequence[Operator], optional): List of valid operators to use. Default to empty list.
+            valid_directions (Sequence[Direction], optional): List of valid directions to use. Default to empty list.
 
         Raises:
             InvalidTableError: If the table is not in the list of valid tables (only if check_table_injection=True).
             InvalidColumnError: If the column is not in the list of valid columns (only if check_column_injection=True).
             InvalidOperatorError: If the operator is not in the list of valid operators (only if check_operator_injection=True).
+            InvalidDirectionError: If the direction is not in the list of valid directions (only if check_direction_injection=True).
 
         Returns:
             tuple[str, dict[str, Any]]: The SQLite query string and the query parameters.
@@ -95,6 +101,7 @@ def convert(
         valid_tables = valid_tables or []
         valid_columns = valid_columns or []
         valid_operators = valid_operators or []
+        valid_directions = valid_directions or []
 
         if check_table_injection:
             cls._validate_table(table=table, valid_tables=valid_tables)
@@ -108,6 +115,9 @@ def convert(
         if check_operator_injection:
             cls._validate_operators(criteria=criteria, valid_operators=valid_operators)
 
+        if check_direction_injection:
+            cls._validate_directions(criteria=criteria, valid_directions=valid_directions)
+
         quoted_columns = ['*' if column == '*' else f'"{column}"' for column in columns]
         quoted_table = '.'.join(f'"{part}"' for part in table.split('.'))
         query = f'SELECT {", ".join(quoted_columns)} FROM {quoted_table}'  # noqa: S608  # nosec
@@ -207,6 +217,25 @@ def _validate_operators(cls, *, criteria: Criteria, valid_operators: Sequence[Op
             if filter.operator not in valid_operators:
                 raise InvalidOperatorError(operator=Operator(value=filter.operator), valid_operators=valid_operators)
 
+    @classmethod
+    def _validate_directions(cls, *, criteria: Criteria, valid_directions: Sequence[Direction]) -> None:
+        """
+        Validate the Criteria object directions to prevent SQL injection.
+
+        Args:
+            criteria (Criteria): Criteria to validate.
+            valid_directions (Sequence[Direction]): List of valid directions to use.
+
+        Raises:
+            InvalidDirectionError: If the direction is not in the list of valid directions.
+        """
+        for order in criteria.orders:
+            if order.direction not in valid_directions:
+                raise InvalidDirectionError(
+                    direction=Direction(value=order.direction),
+                    valid_directions=valid_directions,
+                )
+
     @classmethod
     def _process_filters(cls, *, criteria: Criteria, columns_mapping: Mapping[str, str]) -> tuple[str, dict[str, Any]]:
         """

criteria_pattern/converters/url_to_criteria_converter.py

@@ -8,7 +8,7 @@
 from urllib.parse import parse_qs, unquote_plus, urlparse
 
 from criteria_pattern import Criteria, Direction, Filter, Operator, Order
-from criteria_pattern.errors import InvalidColumnError, InvalidOperatorError
+from criteria_pattern.errors import InvalidColumnError, InvalidDirectionError, InvalidOperatorError
 
 
 class UrlToCriteriaConverter:
@@ -66,8 +66,10 @@ def convert(
         fields_mapping: Mapping[str, str] | None = None,
         check_field_injection: bool = False,
         check_operator_injection: bool = False,
+        check_direction_injection: bool = False,
         valid_fields: Sequence[str] | None = None,
         valid_operators: Sequence[Operator] | None = None,
+        valid_directions: Sequence[Direction] | None = None,
     ) -> Criteria:
         """
         Converts an URL query string into a Criteria object.
@@ -77,8 +79,10 @@ def convert(
             fields_mapping (Mapping[str, str], optional): Mapping of field names to aliases. Default to empty dict.
             check_field_injection (bool, optional): Whether to check for field injection.
             check_operator_injection (bool, optional): Whether to check for operator injection.
+            check_direction_injection (bool, optional): Whether to check for direction injection.
             valid_fields (Sequence[str], optional): A list of valid field names. Default to empty list.
             valid_operators (Sequence[Operator], optional): A list of valid operators. Default to empty list.
+            valid_directions (Sequence[Direction], optional): A list of valid directions. Default to empty list.
 
         Raises:
             TypeError: If the filter index is not an integer.
@@ -93,6 +97,7 @@ def convert(
             InvalidColumnError: If an invalid field name is found in filters.
             InvalidColumnError: If an invalid field name is found in orders.
             InvalidOperatorError: If an invalid operator is found in filters.
+            InvalidDirectionError: If an invalid direction is found in orders.
 
         Example:
         ```python
@@ -107,6 +112,7 @@ def convert(
         valid_fields = valid_fields or []
         fields_mapping = fields_mapping or {}
         valid_operators = valid_operators or []
+        valid_directions = valid_directions or []
 
         query_params = parse_qs(qs=urlparse(url=url).query, keep_blank_values=True)
 
@@ -128,6 +134,9 @@ def convert(
         if check_operator_injection:
             cls._validate_operators(criteria=criteria, valid_operators=valid_operators)
 
+        if check_direction_injection:
+            cls._validate_directions(criteria=criteria, valid_directions=valid_directions)
+
         return criteria
 
     @classmethod
@@ -417,3 +426,22 @@ def _validate_operators(cls, *, criteria: Criteria, valid_operators: Sequence[Op
         for filter in criteria.filters:
             if filter.operator not in valid_operators:
                 raise InvalidOperatorError(operator=Operator(value=filter.operator), valid_operators=valid_operators)
+
+    @classmethod
+    def _validate_directions(cls, *, criteria: Criteria, valid_directions: Sequence[Direction]) -> None:
+        """
+        Validate the Criteria object directions to prevent injection.
+
+        Args:
+            criteria (Criteria): Criteria to validate.
+            valid_directions (Sequence[Direction]): List of valid directions to use.
+
+        Raises:
+            InvalidDirectionError: If the direction is not in the list of valid directions.
+        """
+        for order in criteria.orders:
+            if order.direction not in valid_directions:
+                raise InvalidDirectionError(
+                    direction=Direction(value=order.direction),
+                    valid_directions=valid_directions,
+                )

criteria_pattern/errors/__init__.py

@@ -1,10 +1,12 @@
 from .invalid_column_error import InvalidColumnError
+from .invalid_direction_error import InvalidDirectionError
 from .invalid_operator_error import InvalidOperatorError
 from .invalid_table_error import InvalidTableError
 from .sql_converter_error import SqlConverterError
 
 __all__ = (
     'InvalidColumnError',
+    'InvalidDirectionError',
     'InvalidOperatorError',
     'InvalidTableError',
     'SqlConverterError',