fix: escape id path parameter before interpolating into PostgREST filters

CHANGELOG.md

@@ -8,6 +8,7 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 ## [Unreleased]
 
 ### 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.
 
 ### Planned

backend/database.py

@@ -99,8 +99,8 @@ def db_post(table, data):
 
 
 def db_patch(table, id, data):
-    return request_supabase("PATCH", table, params=f"id=eq.{id}", data=data)
+    return request_supabase("PATCH", table, params=f"id=eq.{escape_filter_value(id)}", data=data)
 
 
 def db_delete(table, id):
-    return request_supabase("DELETE", table, params=f"id=eq.{id}")
+    return request_supabase("DELETE", table, params=f"id=eq.{escape_filter_value(id)}")

backend/main.py

@@ -217,7 +217,7 @@ def filter_tickets(
 
 @app.get("/tickets/{id}")
 def get_ticket(id: str):
-    result = db_get("tickets", f"id=eq.{id}")
+    result = db_get("tickets", f"id=eq.{escape_filter_value(id)}")
     if not result:
         raise HTTPException(status_code=404, detail="Ticket not found")
     return result[0]
@@ -234,7 +234,7 @@ def create_ticket(t: Ticket):
 @app.put("/tickets/{id}")
 def update_ticket(id: str, t: Ticket):
     data = clean_empty_strings(model_data(t), "customer_id", "device_id", "assigned_user_id")
-    old = db_get("tickets", f"id=eq.{id}&select=status")
+    old = db_get("tickets", f"id=eq.{escape_filter_value(id)}&select=status")
     result = db_patch("tickets", id, data)
     if old and old[0]["status"] != data["status"]:
         db_post(
@@ -252,7 +252,7 @@ def update_ticket(id: str, t: Ticket):
 # --- Notes ---
 @app.get("/tickets/{id}/notes")
 def get_notes(id: str):
-    return db_get("ticket_notes", f"ticket_id=eq.{id}&order=created_at.asc")
+    return db_get("ticket_notes", f"ticket_id=eq.{escape_filter_value(id)}&order=created_at.asc")
 
 
 @app.post("/notes")
@@ -263,7 +263,7 @@ def create_note(n: TicketNote):
 # --- History ---
 @app.get("/tickets/{id}/history")
 def get_history(id: str):
-    return db_get("ticket_history", f"ticket_id=eq.{id}&order=changed_at.asc")
+    return db_get("ticket_history", f"ticket_id=eq.{escape_filter_value(id)}&order=changed_at.asc")
 
 
 # --- RMAs ---

backend/test_database.py

@@ -117,3 +117,30 @@ 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\\")*"'
+
+
+# ---------------------------------------------------------------------------
+# db_patch / db_delete -- id path parameter is escaped
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("helper", ["db_patch", "db_delete"])
+def test_id_helpers_escape_injected_query_params(monkeypatch, helper):
+    monkeypatch.setattr(database, "SUPABASE_URL", "https://example.supabase.co")
+    monkeypatch.setattr(database, "SUPABASE_KEY", "secret")
+    fake_response = MagicMock()
+    fake_response.status_code = 204
+    fake_response.text = ""
+    fake_request = MagicMock(return_value=fake_response)
+    monkeypatch.setattr(database.requests, "request", fake_request)
+
+    malicious_id = "t1&status=eq.Closed"
+    if helper == "db_patch":
+        database.db_patch("tickets", malicious_id, {"title": "x"})
+    else:
+        database.db_delete("tickets", malicious_id)
+
+    url = fake_request.call_args.args[1]
+    assert url.count("&") == 0, url
+    assert "status=eq.Closed" not in url
+    assert url.endswith("/rest/v1/tickets?id=eq." + database.escape_filter_value(malicious_id))

backend/test_main.py

@@ -195,6 +195,54 @@ def fake_get(table, params=""):
     assert captured["params"] == ("assigned_user_id=eq.u2%26limit%3D1&order=created_at.desc")
 
 
+@pytest.mark.parametrize(
+    "path_template",
+    ["/tickets/{id}", "/tickets/{id}/notes", "/tickets/{id}/history"],
+)
+def test_ticket_id_routes_escape_postgrest_injection(client, monkeypatch, path_template):
+    test_client, main = client
+    captured = {}
+
+    def fake_get(table, params=""):
+        captured["params"] = params
+        return [{"id": "t1"}]
+
+    monkeypatch.setattr(main, "db_get", fake_get)
+
+    # A crafted id that tries to append an extra filter / parameter.
+    payload = quote("t1&status=eq.Closed", safe="")
+    res = test_client.get(path_template.format(id=payload))
+
+    assert res.status_code == 200
+    params = captured["params"]
+    assert "&status=eq.Closed" not in params
+    # Only the template's own "&order=" (if any) may remain.
+    assert params.count("&") <= 1, params
+
+
+def test_update_ticket_escapes_id_in_status_lookup(client, monkeypatch):
+    test_client, main = client
+    captured = {}
+
+    def fake_get(table, params=""):
+        captured["get_params"] = params
+        return [{"status": "Open"}]
+
+    monkeypatch.setattr(main, "db_get", fake_get)
+    monkeypatch.setattr(main, "db_patch", lambda table, id, data: [{"id": id, **data}])
+    monkeypatch.setattr(main, "db_post", lambda table, data: [data])
+
+    payload = quote("t1&status=eq.Closed", safe="")
+    res = test_client.put(
+        f"/tickets/{payload}",
+        json={"title": "Printer", "status": "Open", "priority": "Low"},
+    )
+
+    assert res.status_code == 200
+    assert "&status=eq.Closed" not in captured["get_params"]
+    assert captured["get_params"].endswith("&select=status")
+
+
 def test_health_returns_ok(client):
     test_client, _ = client
     res = test_client.get("/health")