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` (`ai.suggest_for_ticket`) escapes `ticket_id` before interpolating it into the `tickets` and `ticket_notes` PostgREST filters, closing the same injection gap already fixed for the `{id}`-based ticket routes.
 
 ### 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,75 @@
+from unittest.mock import MagicMock
+from urllib.parse import parse_qs, urlsplit
+
+import pytest
+
+import ai
+import database
+
+
+def test_fetch_ticket_escapes_id_before_interpolating(monkeypatch):
+    fake_db_get = MagicMock(return_value=[{"id": "t1"}])
+    monkeypatch.setattr(ai, "db_get", fake_db_get)
+
+    ai._fetch_ticket("t1&status=eq.Closed")
+
+    fake_db_get.assert_called_once_with("tickets", "id=eq.t1%26status%3Deq.Closed")
+
+
+def test_fetch_ticket_raises_when_not_found(monkeypatch):
+    monkeypatch.setattr(ai, "db_get", MagicMock(return_value=[]))
+
+    with pytest.raises(ValueError, match="Ticket t1 not found"):
+        ai._fetch_ticket("t1")
+
+
+@pytest.mark.parametrize(
+    ("ticket_id", "expected_filter"),
+    [
+        ("t1),status.eq.Closed", "t1%29%2Cstatus.eq.Closed"),
+        ("t1&select=*", "t1%26select%3D%2A"),
+        ("t1%26select%3D*", "t1%2526select%253D%2A"),
+    ],
+)
+def test_fetch_notes_escapes_ticket_id_before_interpolating(
+    monkeypatch, ticket_id, expected_filter
+):
+    fake_db_get = MagicMock(return_value=[])
+    monkeypatch.setattr(ai, "db_get", fake_db_get)
+
+    ai._fetch_notes(ticket_id)
+
+    fake_db_get.assert_called_once_with(
+        "ticket_notes",
+        f"ticket_id=eq.{expected_filter}&order=created_at.asc",
+    )
+
+
+@pytest.mark.parametrize(
+    ("fetch", "table", "expected_query"),
+    [
+        (ai._fetch_ticket, "tickets", {"id": ["eq.t1&select=*"]}),
+        (
+            ai._fetch_notes,
+            "ticket_notes",
+            {"ticket_id": ["eq.t1&select=*"], "order": ["created_at.asc"]},
+        ),
+    ],
+)
+def test_fetch_preserves_one_id_filter_at_http_boundary(monkeypatch, fetch, table, expected_query):
+    monkeypatch.setattr(database, "SUPABASE_URL", "https://example.invalid")
+    monkeypatch.setattr(database, "SUPABASE_KEY", "test-key")
+    response = MagicMock(status_code=200, text='[{"id": "t1"}]')
+    response.json.return_value = [{"id": "t1"}]
+    request = MagicMock(return_value=response)
+    monkeypatch.setattr(database.requests, "request", request)
+
+    fetch("t1&select=*")
+
+    request.assert_called_once()
+    method, url = request.call_args.args
+    prepared_url = database.requests.Request(method, url).prepare().url
+    parsed = urlsplit(prepared_url)
+    assert method == "GET"
+    assert parsed.path == f"/rest/v1/{table}"
+    assert parse_qs(parsed.query) == expected_query