fix(#256): handle nul character in string parameters

Author: captaincoordinates

tests/test_collections.py

@@ -692,3 +692,12 @@ def test_collections_cql_filter(filter_expr, filter_lang, app):
     assert len(resp) == 1
     assert list(resp[0]) == ["id", "bbox", "assets", "collection"]
     assert resp[0]["id"] == "20200307aC0853000w361030"
+
+
+def test_nul_byte_handling(app) -> None:
+    """Ensure nul bytes do not generate 500 errors"""
+    response = app.get("/collections/missing_collection%00id/tiles")
+    assert response.status_code == 404
+
+    response = app.get("/collections/missing_collection_id/tiles?param1=bad%00value")
+    assert response.status_code == 404

titiler/pgstac/main.py

@@ -50,6 +50,7 @@
     add_search_list_route,
     add_search_register_route,
 )
+from titiler.pgstac.middleware import StripNulMiddleware
 from titiler.pgstac.reader import PgSTACReader
 from titiler.pgstac.settings import ApiSettings
 
@@ -122,6 +123,7 @@ async def lifespan(app: FastAPI):
     cachecontrol=settings.cachecontrol,
     exclude_path=settings.cachecontrol_exclude_paths,
 )
+app.add_middleware(StripNulMiddleware)
 
 optional_headers = []
 if settings.debug:

titiler/pgstac/middleware.py

@@ -0,0 +1,42 @@
+"""titiler-pgstac Middleware."""
+
+import re
+from typing import Final, cast
+from urllib.parse import quote
+
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+_encoding: Final[str] = "utf-8"
+_nul: Final[str] = "\x00"
+
+
+class StripNulMiddleware:
+    r"""
+    Postgres cannot handle NUL bytes. If left unhandled \x00 will first cause problems
+    when JSON is passed to SQL, as JSON escapes \x00 to \u0000 but does not escape the
+    \ escape character. If this character _is_ escaped then Postgres will report
+    "PostgreSQL text fields cannot contain NUL (0x00) bytes".
+    This class presumes that no legitimate API client will intentionally submit NUL bytes
+    and simply strips them from all parameters.
+    """
+
+    def __init__(
+        self,
+        app: ASGIApp,
+    ) -> None:
+        """Strip NUL Middleware."""
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send):
+        """Remove NUL bytes from path and query string."""
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        scope["path"] = re.sub(_nul, "", scope["path"])
+        for property in ["query_string", "raw_path"]:
+            scope[property] = re.sub(
+                quote(_nul), "", cast(bytes, scope[property]).decode(_encoding)
+            ).encode(_encoding)
+
+        await self.app(scope, receive, send)