From c23b1be3f8b2c30064ec133664d8f9eb6bb1b748 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Robert=20Vind-Gardo=C8=99?= <41092915+robyroro@users.noreply.github.com> Date: Wed, 14 Oct 2026 10:41:38 +0300 Subject: [PATCH] fix(history): reject implicit and coerced truncation targets (#6565) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(history): reject implicit and coerced truncation targets * fix(history): reject falsy non-string message IDs; drop redundant keep_count guard The or-chain used to resolve before_msg_id / message_id evaluated falsy non-strings (False, 0) to '' before the isinstance(…, str) check was reached, so they silently bypassed type validation. Replace the chain with explicit presence-aware logic that falls through on None or empty string (preserving alias semantics) but rejects any other non-string value with HTTP 400. Also remove the now-tautological 'if keep_count >= 0:' guard, which is always True after the negativity check introduced in the parent commit. The DB block is unconditional and indented accordingly. Tests: add {'before_msg_id': False}, {'before_msg_id': 0}, {'message_id': 0} to the parametrised rejection suite. 36 passed, 0 failed (Python 3.14.4 / pytest 9.1.1). --------- Co-authored-by: Alexandre Teixeira <111787685+alteixeira20@users.noreply.github.com> --- routes/history/history_routes.py | 65 ++++++++++----- tests/test_history_truncate_validation.py | 98 +++++++++++++++++++++++ 2 files changed, 144 insertions(+), 19 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 0900a0931..a14be5908 100644 --- a/routes/history/history_routes.py +++ b/routes/history/history_routes.py @@ -403,26 +403,53 @@ 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") + # Resolve before_msg_id / message_id alias with explicit presence checks. + # None (JSON null) and empty string are both treated as "not provided here" + # and fall through to the alias, preserving the original OR-chain semantics. + # Any other non-string value (False, 0, True, 123, list, …) is rejected + # immediately so falsy non-strings cannot bypass type validation. + raw_msg_id = body.get("before_msg_id") if "before_msg_id" in body else None + if raw_msg_id is None or raw_msg_id == "": + raw_msg_id = body.get("message_id") + if raw_msg_id is not None and not isinstance(raw_msg_id, str): + raise HTTPException(400, "Message ID must be a string") + before_msg_id = (raw_msg_id or "").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() - try: - all_db_messages = db.query(DbChatMessage).filter( - DbChatMessage.session_id == session_id - ).order_by(DbChatMessage.timestamp).all() - if before_msg_id: - resolved_keep_count = _keep_count_before_message(all_db_messages, before_msg_id) - if resolved_keep_count is None: - raise HTTPException(404, "Message not found") - keep_count = resolved_keep_count - for pos, row in enumerate(all_db_messages): - row._will_delete_for_sft = pos >= keep_count - deleted_sft_pairs = _deleted_sft_pairs_from_db_rows(all_db_messages) - finally: - db.close() + # keep_count is always >= 0 here; the block is unconditional. + db = SessionLocal() + try: + all_db_messages = db.query(DbChatMessage).filter( + DbChatMessage.session_id == session_id + ).order_by(DbChatMessage.timestamp).all() + if before_msg_id: + resolved_keep_count = _keep_count_before_message(all_db_messages, before_msg_id) + if resolved_keep_count is None: + raise HTTPException(404, "Message not found") + keep_count = resolved_keep_count + for pos, row in enumerate(all_db_messages): + row._will_delete_for_sft = pos >= keep_count + deleted_sft_pairs = _deleted_sft_pairs_from_db_rows(all_db_messages) + finally: + db.close() result = session_manager.truncate_messages(session_id, keep_count) _remove_deleted_sft_trace_rows( owner=effective_user(request), diff --git a/tests/test_history_truncate_validation.py b/tests/test_history_truncate_validation.py new file mode 100644 index 000000000..9772037db --- /dev/null +++ b/tests/test_history_truncate_validation.py @@ -0,0 +1,98 @@ +"""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"]}, + {"before_msg_id": False}, {"before_msg_id": 0}, {"message_id": 0}, + [], 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"]