fix: escape ticket_id before interpolating into PostgREST filters in ai.py

CHANGELOG.md

@@ -10,6 +10,7 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 ### Fixed
 - `db_patch`, `db_delete`, and the `{id}`-based ticket routes (`GET /tickets/{id}`, `GET /tickets/{id}/notes`, `GET /tickets/{id}/history`, and the status lookup in `PUT /tickets/{id}`) now escape the `id` path parameter before interpolating it into PostgREST filters, so a crafted id can no longer append extra conditions or query parameters.
 - `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.
+- `POST /ai/suggest` no longer interpolates the request body's `ticket_id` unescaped into PostgREST filters. `ai._fetch_ticket` and `ai._fetch_notes` were missed by the earlier filter-escaping fixes above; they now use the same `escape_filter_value` guard as the rest of the API.
 
 ### Planned
 - Role-based auth (agent / lead / admin) with JWT bearer tokens

backend/ai.py

@@ -8,7 +8,7 @@
 from anthropic import Anthropic
 from dotenv import load_dotenv
 
-from database import db_get
+from database import db_get, escape_filter_value
 
 load_dotenv()
 
@@ -28,14 +28,16 @@ def _get_client() -> Anthropic:
 
 
 def _fetch_ticket(ticket_id: str) -> dict[str, Any]:
-    result = db_get("tickets", f"id=eq.{ticket_id}")
+    result = db_get("tickets", f"id=eq.{escape_filter_value(ticket_id)}")
     if not isinstance(result, list) or not result:
         raise ValueError(f"Ticket {ticket_id} not found")
     return result[0]
 
 
 def _fetch_notes(ticket_id: str) -> list[dict[str, Any]]:
-    notes = db_get("ticket_notes", f"ticket_id=eq.{ticket_id}&order=created_at.asc")
+    notes = db_get(
+        "ticket_notes", f"ticket_id=eq.{escape_filter_value(ticket_id)}&order=created_at.asc"
+    )
     return notes if isinstance(notes, list) else []
 
 

backend/test_ai.py

@@ -0,0 +1,44 @@
+"""Unit tests for backend/ai.py's Supabase filter construction.
+
+Mirrors the escaping coverage in test_database.py to guard against the same
+class of PostgREST filter injection via an untrusted ticket_id.
+"""
+
+import ai
+
+
+def test_fetch_ticket_escapes_injected_query_params(monkeypatch):
+    captured = {}
+
+    def fake_db_get(table, params=""):
+        captured["table"] = table
+        captured["params"] = params
+        return [{"id": "t1", "title": "Printer down"}]
+
+    monkeypatch.setattr(ai, "db_get", fake_db_get)
+
+    malicious_id = "t1&status=eq.Closed"
+    ai._fetch_ticket(malicious_id)
+
+    assert captured["table"] == "tickets"
+    assert captured["params"].count("&") == 0
+    assert "status=eq.Closed" not in captured["params"]
+
+
+def test_fetch_notes_escapes_injected_query_params(monkeypatch):
+    captured = {}
+
+    def fake_db_get(table, params=""):
+        captured["table"] = table
+        captured["params"] = params
+        return []
+
+    monkeypatch.setattr(ai, "db_get", fake_db_get)
+
+    malicious_id = "t1&status=eq.Closed"
+    ai._fetch_notes(malicious_id)
+
+    assert captured["table"] == "ticket_notes"
+    assert captured["params"].count("&") == 1  # only the trailing "&order=..." is ours
+    assert "status=eq.Closed" not in captured["params"]
+    assert captured["params"].endswith("&order=created_at.asc")