From 3365d5bdc52baf098af1199e81f52d7005440325 Mon Sep 17 00:00:00 2001 From: robyroro <41092915+robyroro@users.noreply.github.com> Date: Wed, 7 Oct 2026 04:18:16 +0300 Subject: [PATCH] fix(history): reject implicit and coerced truncation targets --- routes/history/history_routes.py | 26 +++++- tests/test_history_truncate_validation.py | 97 +++++++++++++++++++++++ 2 files changed, 120 insertions(+), 3 deletions(-) create mode 100644 tests/test_history_truncate_validation.py diff --git a/routes/history/history_routes.py b/routes/history/history_routes.py index 890927c53..b822d2d95 100644 --- a/routes/history/history_routes.py +++ b/routes/history/history_routes.py @@ -391,9 +391,29 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter: async def truncate_session(request: Request, session_id: str): _verify_session_owner(request, session_id) try: - body = await request.json() - keep_count = int(body.get("keep_count", 0)) - before_msg_id = str(body.get("before_msg_id") or body.get("message_id") or "").strip() + try: + body = await request.json() + except json.JSONDecodeError: + raise HTTPException(400, "Request body must be valid JSON") + if not isinstance(body, dict): + raise HTTPException(400, "Request body must be a JSON object") + before_msg_id = body.get("before_msg_id") or body.get("message_id") or "" + if not isinstance(before_msg_id, str): + raise HTTPException(400, "Message ID must be a string") + before_msg_id = before_msg_id.strip() + if "keep_count" not in body and not before_msg_id: + raise HTTPException(400, "keep_count or before_msg_id required") + raw_count = body.get("keep_count", 0) + # Keep integer-string clients working, but do not silently convert + # booleans or fractional numbers into destructive message counts. + if type(raw_count) not in (int, str): + raise HTTPException(400, "keep_count must be a non-negative integer") + try: + keep_count = int(raw_count) + except ValueError: + raise HTTPException(400, "keep_count must be a non-negative integer") + if keep_count < 0: + raise HTTPException(400, "keep_count must be a non-negative integer") deleted_sft_pairs: list[dict[str, str]] = [] if keep_count >= 0: db = SessionLocal() diff --git a/tests/test_history_truncate_validation.py b/tests/test_history_truncate_validation.py new file mode 100644 index 000000000..be8ea8173 --- /dev/null +++ b/tests/test_history_truncate_validation.py @@ -0,0 +1,97 @@ +"""Invalid truncation requests must leave saved chat history intact.""" +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from tests.helpers.database import isolated_session_database + + +@pytest.fixture +def history_client(tmp_path, monkeypatch): + from core.models import ChatMessage + from routes.history import history_routes + from routes import session_routes + + with isolated_session_database(tmp_path) as (manager, database): + monkeypatch.setattr(history_routes, "SessionLocal", database.SessionLocal) + monkeypatch.setattr(session_routes, "SessionLocal", database.SessionLocal) + monkeypatch.setenv("AUTH_ENABLED", "true") + manager.create_session("truncate-test", "Test", "http://example.test", "test", owner="alice") + for text in ("first", "second", "third"): + manager.add_message("truncate-test", ChatMessage("user", text)) + app = FastAPI() + app.state.auth_manager = SimpleNamespace(is_configured=True) + + @app.middleware("http") + async def owner(request, call_next): + request.state.current_user = request.headers.get("X-Test-Owner", "alice") + return await call_next(request) + + app.include_router(history_routes.setup_history_routes(manager)) + with TestClient(app, raise_server_exceptions=False) as client: + yield client, manager, database + + +def saved_messages(database): + with database.SessionLocal() as db: + return [row.content for row in db.query(database.ChatMessage).filter( + database.ChatMessage.session_id == "truncate-test" + ).order_by(database.ChatMessage.timestamp).all()] + + +@pytest.mark.parametrize("payload", [ + {}, {"unrelated": 1}, {"before_msg_id": ""}, + {"keep_count": None}, {"keep_count": False}, {"keep_count": True}, + {"keep_count": 1.9}, {"keep_count": -1}, {"keep_count": "bad"}, + {"keep_count": []}, {"keep_count": {}}, + {"before_msg_id": True}, {"before_msg_id": ["message"]}, + [], None, "two", 2, False, +]) +def test_invalid_target_preserves_history(history_client, payload): + client, manager, database = history_client + import json + response = client.post("/api/session/truncate-test/truncate", content=json.dumps(payload), headers={"Content-Type": "application/json"}) + assert response.status_code == 400, response.text + assert saved_messages(database) == ["first", "second", "third"] + assert [m.content for m in manager.get_session("truncate-test").history] == ["first", "second", "third"] + + +def test_malformed_json_preserves_history(history_client): + client, _, database = history_client + response = client.post("/api/session/truncate-test/truncate", content="{", headers={"Content-Type": "application/json"}) + assert response.status_code == 400, response.text + assert saved_messages(database) == ["first", "second", "third"] + + +@pytest.mark.parametrize("count,expected", [(0, []), (2, ["first", "second"]), (9, ["first", "second", "third"]), ("2", ["first", "second"])]) +def test_explicit_count_remains_supported(history_client, count, expected): + client, _, database = history_client + response = client.post("/api/session/truncate-test/truncate", json={"keep_count": count}) + assert response.status_code == 200, response.text + assert saved_messages(database) == expected + + +@pytest.mark.parametrize("field", ["before_msg_id", "message_id"]) +def test_message_id_boundary_remains_supported(history_client, field): + client, _, database = history_client + with database.SessionLocal() as db: + message_id = db.query(database.ChatMessage).filter(database.ChatMessage.content == "second").one().id + response = client.post("/api/session/truncate-test/truncate", json={field: message_id}) + assert response.status_code == 200, response.text + assert saved_messages(database) == ["first"] + + +def test_missing_message_id_does_not_clear_history(history_client): + client, _, database = history_client + response = client.post("/api/session/truncate-test/truncate", json={"before_msg_id": "missing"}) + assert response.status_code == 404 + assert saved_messages(database) == ["first", "second", "third"] + + +def test_owner_gate_still_precedes_validation(history_client): + client, _, database = history_client + response = client.post("/api/session/truncate-test/truncate", json={}, headers={"X-Test-Owner": "bob"}) + assert response.status_code == 404 + assert saved_messages(database) == ["first", "second", "third"]