Harden OData string filter serialization

bcb/odata/framework.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+import math
 import threading
 from io import BytesIO
 from typing import Any, Optional, Union
@@ -51,6 +52,52 @@ def _load_xml_document(content: bytes, *, context: str) -> Any:
         raise ODataError(f"{context} returned invalid XML: {ex}") from ex
 
 
+def _format_odata_string_literal(value: Any) -> str:
+    if value is None:
+        raise ODataError("Edm.String filter values cannot be None")
+    escaped = str(value).replace("'", "''")
+    return f"'{escaped}'"
+
+
+def _format_odata_literal(edm_type: Optional[str], value: Any) -> str:
+    if value is None:
+        raise ODataError(f"{edm_type or 'Unknown'} filter values cannot be None")
+
+    if edm_type == "Edm.Decimal":
+        try:
+            decimal_value = float(value)
+        except (TypeError, ValueError) as ex:
+            raise ODataError(f"Invalid Edm.Decimal filter value: {value!r}") from ex
+        if not math.isfinite(decimal_value):
+            raise ODataError(f"Invalid Edm.Decimal filter value: {value!r}")
+        return f"{decimal_value}"
+
+    if edm_type in ("Edm.Int16", "Edm.Int32", "Edm.Int64"):
+        try:
+            return f"{int(value)}"
+        except (TypeError, ValueError) as ex:
+            raise ODataError(f"Invalid {edm_type} filter value: {value!r}") from ex
+
+    if edm_type == "Edm.String":
+        return _format_odata_string_literal(value)
+
+    if edm_type == "Edm.Date":
+        try:
+            formatted = value.strftime("%Y-%m-%d")
+        except AttributeError as ex:
+            raise ODataError(f"Invalid Edm.Date filter value: {value!r}") from ex
+        if not isinstance(formatted, str):
+            raise ODataError(f"Invalid Edm.Date filter value: {value!r}")
+        return formatted
+
+    if edm_type == "Edm.Boolean":
+        if not isinstance(value, bool):
+            raise ODataError(f"Invalid Edm.Boolean filter value: {value!r}")
+        return str(value).lower()
+
+    raise ODataError(f"Unsupported OData filter literal type: {edm_type or 'Unknown'}")
+
+
 # Edm.Boolean
 # Edm.Byte
 # Edm.Date
@@ -212,16 +259,8 @@ def __init__(self, obj: "ODataProperty", oth: Any, operator: str) -> None:
         self.operator = operator
 
     def statement(self) -> str:
-        if self.obj.type == "Edm.Decimal":
-            return f"{self.obj.name} {self.operator} {float(self.other)}"
-        elif self.obj.type == "Edm.Int32":
-            return f"{self.obj.name} {self.operator} {int(self.other)}"
-        elif self.obj.type == "Edm.String":
-            return f"{self.obj.name} {self.operator} '{str(self.other)}'"
-        elif self.obj.type == "Edm.Date":
-            return f"{self.obj.name} {self.operator} {self.other.strftime('%Y-%m-%d')}"
-        else:
-            return f"{self.obj.name} {self.operator} '{self.other}'"
+        literal = _format_odata_literal(self.obj.type, self.other)
+        return f"{self.obj.name} {self.operator} {literal}"
 
     def __str__(self) -> str:
         return self.statement()

tests/test_odata.py

@@ -1,12 +1,12 @@
 import re
-from datetime import datetime
+from datetime import date, datetime
 
 import httpx
 import pandas as pd
 import pytest
 
 from bcb.odata.api import Expectativas
-from bcb.odata.framework import ODataPropertyFilter, ODataPropertyOrderBy
+from bcb.odata.framework import ODataProperty, ODataPropertyFilter, ODataPropertyOrderBy
 from bcb.exceptions import ODataError
 from tests.conftest import (
     ODATA_SERVICE_ROOT_JSON,
@@ -215,6 +215,11 @@ def test_string_property_equality_filter(httpx_mock):
     assert str(f) == "Indicador eq 'IPCA'"
 
 
+def test_string_property_filter_escapes_apostrophes():
+    indicador = ODataProperty(Name="Indicador", Type="Edm.String")
+    assert str(indicador == "Focus's IPCA") == "Indicador eq 'Focus''s IPCA'"
+
+
 def test_decimal_property_comparison_filters(httpx_mock):
     add_service_mocks(httpx_mock)
     api = Expectativas()
@@ -226,6 +231,34 @@ def test_decimal_property_comparison_filters(httpx_mock):
     assert str(mediana <= 4.0) == "Mediana le 4.0"
 
 
+def test_date_property_filter_formats_dates():
+    data = ODataProperty(Name="Data", Type="Edm.Date")
+    assert str(data == date(2024, 1, 31)) == "Data eq 2024-01-31"
+
+
+def test_int_property_filter_formats_ints():
+    prazo = ODataProperty(Name="Prazo", Type="Edm.Int32")
+    assert str(prazo == "12") == "Prazo eq 12"
+
+
+@pytest.mark.parametrize(
+    ("prop", "value", "message"),
+    [
+        (ODataProperty(Name="Indicador", Type="Edm.String"), None, "Edm.String"),
+        (
+            ODataProperty(Name="Mediana", Type="Edm.Decimal"),
+            "not-a-number",
+            "Edm.Decimal",
+        ),
+        (ODataProperty(Name="Data", Type="Edm.Date"), "2024-01-31", "Edm.Date"),
+        (ODataProperty(Name="Codigo", Type="Edm.Guid"), "abc", "Unsupported"),
+    ],
+)
+def test_property_filter_invalid_values_raise_odata_error(prop, value, message):
+    with pytest.raises(ODataError, match=message):
+        str(ODataPropertyFilter(prop, value, "eq"))
+
+
 def test_property_orderby(httpx_mock):
     add_service_mocks(httpx_mock)
     api = Expectativas()