diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c98462..17cf542 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/backend/database.py b/backend/database.py index 731d138..7da6c31 100644 --- a/backend/database.py +++ b/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)}") diff --git a/backend/main.py b/backend/main.py index 328bc5a..a434e29 100644 --- a/backend/main.py +++ b/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 --- diff --git a/backend/test_database.py b/backend/test_database.py index 7cf4b0f..965f52a 100644 --- a/backend/test_database.py +++ b/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)) diff --git a/backend/test_main.py b/backend/test_main.py index 879319f..ebd5e80 100644 --- a/backend/test_main.py +++ b/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")