fix: escape user input before interpolating into PostgREST filters

CHANGELOG.md

@@ -7,6 +7,9 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Fixed
+- `GET /tickets/search` and `GET /tickets/filter` no longer interpolate raw user input into PostgREST filter strings. Reserved logical-filter values are now quoted with embedded backslashes and quotes escaped, and all values are percent-encoded, so a crafted `q`, `status`, `priority`, or `assigned_user_id` can no longer break out of the intended filter or inject extra query parameters.
+
 ### Planned
 - Role-based auth (agent / lead / admin) with JWT bearer tokens
 - SLA timers with breach alerts

backend/database.py

@@ -1,10 +1,30 @@
 import os
+from urllib.parse import quote
 
 import requests
 from dotenv import load_dotenv
 
 load_dotenv()
 
+
+def escape_filter_value(value: str, *, wildcard: bool = False) -> str:
+    """Make a user-supplied value safe to interpolate into a PostgREST filter string.
+
+    The value is percent-encoded so query-string delimiters cannot create new
+    parameters.
+
+    ``wildcard=True`` additionally quotes the complete ``*value*`` pattern for use
+    inside a PostgREST logical expression. Embedded backslashes and double quotes
+    are escaped within that quoted value, so commas and parentheses cannot terminate
+    an ``or=(...)`` expression. Operators and other template structure remain the
+    caller's responsibility.
+    """
+    if wildcard:
+        escaped = value.replace("\\", "\\\\").replace('"', '\\"')
+        return quote(f'"*{escaped}*"', safe="*")
+    return quote(value, safe="")
+
+
 SUPABASE_URL = os.getenv("SUPABASE_URL")
 SUPABASE_KEY = os.getenv("SUPABASE_KEY")
 DATABASE_UNAVAILABLE_DETAIL = (

backend/main.py

@@ -5,7 +5,15 @@
 
 import ai
 import database
-from database import DatabaseConfigError, DatabaseRequestError, db_delete, db_get, db_patch, db_post
+from database import (
+    DatabaseConfigError,
+    DatabaseRequestError,
+    db_delete,
+    db_get,
+    db_patch,
+    db_post,
+    escape_filter_value,
+)
 
 app = FastAPI(title="SupportOps API")
 
@@ -172,10 +180,10 @@ def search_tickets(q: str = Query(..., min_length=1, description="Search term"))
     Returns tickets where the term appears in title OR description, sorted by
     most recently created first.
     """
-    term = q.strip()
+    term = escape_filter_value(q.strip(), wildcard=True)
     results = db_get(
         "tickets",
-        f"or=(title.ilike.*{term}*,description.ilike.*{term}*)&order=created_at.desc",
+        f"or=(title.ilike.{term},description.ilike.{term})&order=created_at.desc",
     )
     return results if isinstance(results, list) else []
 
@@ -196,11 +204,11 @@ def filter_tickets(
     """
     parts: list[str] = []
     if status:
-        parts.append(f"status=eq.{status}")
+        parts.append(f"status=eq.{escape_filter_value(status)}")
     if priority:
-        parts.append(f"priority=eq.{priority}")
+        parts.append(f"priority=eq.{escape_filter_value(priority)}")
     if assigned_user_id:
-        parts.append(f"assigned_user_id=eq.{assigned_user_id}")
+        parts.append(f"assigned_user_id=eq.{escape_filter_value(assigned_user_id)}")
     parts.append("order=created_at.desc")
     query = "&".join(parts)
     results = db_get("tickets", query)

backend/test_database.py

@@ -1,4 +1,5 @@
 from unittest.mock import MagicMock
+from urllib.parse import unquote
 
 import pytest
 import requests
@@ -90,3 +91,29 @@ def test_supabase_network_error_raises_database_error(monkeypatch, error):
 
     with pytest.raises(database.DatabaseRequestError, match="temporarily unavailable"):
         database.db_get("tickets")
+
+
+# ---------------------------------------------------------------------------
+# escape_filter_value -- PostgREST filter injection guard
+# ---------------------------------------------------------------------------
+
+
+def test_escape_filter_value_passes_plain_text_through():
+    assert database.escape_filter_value("printer") == "printer"
+    assert database.escape_filter_value("agent-7") == "agent-7"
+    assert (
+        database.escape_filter_value("123e4567-e89b-12d3-a456-426614174000")
+        == "123e4567-e89b-12d3-a456-426614174000"
+    )
+
+
+def test_escape_filter_value_percent_encodes_query_delimiters():
+    escaped = database.escape_filter_value("x,status.like.Closed&select=*&path=/tickets")
+
+    assert escaped == "x%2Cstatus.like.Closed%26select%3D%2A%26path%3D%2Ftickets"
+
+
+def test_escape_filter_value_quotes_and_escapes_wildcard_pattern():
+    escaped = database.escape_filter_value('a,b("c\\d")', wildcard=True)
+
+    assert unquote(escaped) == '"*a,b(\\"c\\\\d\\")*"'

backend/test_main.py

@@ -6,6 +6,7 @@
 import sys
 import types
 from unittest.mock import MagicMock
+from urllib.parse import quote, unquote
 
 import pytest
 from fastapi.testclient import TestClient
@@ -51,7 +52,9 @@ def fake_get(table, params=""):
     assert res.status_code == 200
     assert res.json() == rows
     assert captured["table"] == "tickets"
-    assert "printer" in captured["params"].lower()
+    assert captured["params"] == (
+        "or=(title.ilike.%22*printer*%22,description.ilike.%22*printer*%22)&order=created_at.desc"
+    )
 
 
 def test_search_tickets_returns_empty_list_when_no_matches(client, monkeypatch):
@@ -150,6 +153,48 @@ def test_filter_tickets_returns_empty_list_when_no_matches(client, monkeypatch):
     assert res.json() == []
 
 
+def test_search_tickets_escapes_postgrest_injection(client, monkeypatch):
+    test_client, main = client
+    captured = {}
+
+    def fake_get(table, params=""):
+        captured["params"] = params
+        return []
+
+    monkeypatch.setattr(main, "db_get", fake_get)
+
+    # Backslashes do not protect logical delimiters unless the value is quoted.
+    payload = quote("x),status.like.Closed&limit=1", safe="")
+    res = test_client.get(f"/tickets/search?q={payload}")
+
+    assert res.status_code == 200
+    params = captured["params"]
+    assert params.count("&") == 1, params
+    assert params.endswith("&order=created_at.desc")
+    logical_filter = unquote(params.removesuffix("&order=created_at.desc"))
+    assert logical_filter == (
+        'or=(title.ilike."*x),status.like.Closed&limit=1*",'
+        'description.ilike."*x),status.like.Closed&limit=1*")'
+    )
+
+
+def test_filter_tickets_escapes_postgrest_injection(client, monkeypatch):
+    test_client, main = client
+    captured = {}
+
+    def fake_get(table, params=""):
+        captured["params"] = params
+        return []
+
+    monkeypatch.setattr(main, "db_get", fake_get)
+
+    payload = quote("u2&limit=1", safe="")
+    res = test_client.get(f"/tickets/filter?assigned_user_id={payload}")
+
+    assert res.status_code == 200
+    assert captured["params"] == ("assigned_user_id=eq.u2%26limit%3D1&order=created_at.desc")
+
+
 def test_health_returns_ok(client):
     test_client, _ = client
     res = test_client.get("/health")