From ac705b52c90e114342370c3637e49c8d78b5afe6 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Mon, 13 Jul 2026 13:21:35 +0530 Subject: [PATCH] fix(sessions): validate imported session payloads Reject metadata that would make session queries fail, bound import work, and detach cyclic lineage links. Guard lineage traversal against pre-existing corrupt cycles. --- hermes_cli/web_server.py | 40 ++++- hermes_state.py | 219 +++++++++++++++++++++++++--- tests/hermes_cli/test_web_server.py | 59 ++++++++ tests/test_hermes_state.py | 110 ++++++++++++++ 4 files changed, 400 insertions(+), 28 deletions(-) diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index e7c18eae8def..9666d53db886 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -9574,6 +9574,29 @@ class SessionImport(BaseModel): profile: Optional[str] = None +# Keep the dashboard import endpoint stream-safe: FastAPI otherwise parses and +# buffers an arbitrarily large JSON body before SessionDB can enforce its own +# per-session and transaction-work limits. +_SESSION_IMPORT_MAX_BYTES = 25 * 1024 * 1024 + + +async def _read_session_import_body(request: Request) -> bytes: + body = bytearray() + async for chunk in request.stream(): + if len(body) + len(chunk) > _SESSION_IMPORT_MAX_BYTES: + raise HTTPException(status_code=413, detail="Session import payload is too large") + body.extend(chunk) + return bytes(body) + + +def _import_sessions_for_profile(profile: Optional[str], sessions: List[Dict[str, Any]]) -> Dict[str, Any]: + db = _open_session_db_for_profile(profile) + try: + return db.import_sessions(sessions) + finally: + db.close() + + @app.post("/api/sessions/bulk-delete") async def bulk_delete_sessions_endpoint(body: BulkDeleteSessions): """Delete every session in ``body.ids`` in a single DB transaction. @@ -9625,20 +9648,25 @@ async def bulk_delete_sessions_endpoint(body: BulkDeleteSessions): @app.post("/api/sessions/import") -async def import_sessions_endpoint(body: SessionImport): +async def import_sessions_endpoint(request: Request): """Import one or more sessions exported from the dashboard or CLI. This is intentionally separate from ``/api/ops/import``: that endpoint restores a whole Hermes backup archive, while this endpoint is scoped to session rows/messages and is safe to use from the Sessions page. """ - db = _open_session_db_for_profile(body.profile) try: - result = db.import_sessions(body.sessions) + raw_body = await _read_session_import_body(request) + body = SessionImport.model_validate_json(raw_body) + except HTTPException: + raise except ValueError as exc: - raise HTTPException(status_code=400, detail=str(exc)) - finally: - db.close() + raise HTTPException(status_code=400, detail="Invalid session import payload") from exc + + try: + result = await asyncio.to_thread(_import_sessions_for_profile, body.profile, body.sessions) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc if not result.get("ok", False): raise HTTPException(status_code=400, detail=result) diff --git a/hermes_state.py b/hermes_state.py index 0d3e87f735b5..da2e484f7cf1 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -942,6 +942,15 @@ class SessionDB: # pays almost nothing; the cadence is deliberately coarse so the one-off # merge cost is amortised far below the checkpoint cadence. _OPTIMIZE_EVERY_N_WRITES = 1000 + # Session imports intentionally use a lower cap than exports: import holds + # one BEGIN IMMEDIATE transaction, so bounded batches avoid starving live + # gateway/CLI writers. The dashboard accepts one exported JSON/JSONL file + # at a time, so these still cover normal history restores. + _IMPORT_MAX_SESSIONS = 500 + _IMPORT_MAX_MESSAGES_PER_SESSION = 10_000 + _IMPORT_MAX_TOTAL_MESSAGES = 50_000 + _IMPORT_MAX_SESSION_BYTES = 5 * 1024 * 1024 + _IMPORT_MAX_TOTAL_BYTES = 25 * 1024 * 1024 def __init__(self, db_path: Path = None, read_only: bool = False): self.db_path = db_path or DEFAULT_DB_PATH @@ -5278,13 +5287,16 @@ class SessionDB: return [session_id] if session else [] root = session + ancestors = {root["id"]} while self._is_compression_child_row(root): parent = self.get_session(root["parent_session_id"]) - if not parent: + if not parent or parent["id"] in ancestors: break root = parent + ancestors.add(root["id"]) lineage = [root["id"]] + seen = {root["id"]} current = root while current.get("end_reason") == "compression": with self._lock: @@ -5302,9 +5314,10 @@ class SessionDB: if not self._is_branch_child_row(candidate): next_child = candidate break - if not next_child: + if not next_child or next_child["id"] in seen: break lineage.append(next_child["id"]) + seen.add(next_child["id"]) current = next_child if current["id"] == session_id: # Continue to include later compression tips only when the @@ -5353,15 +5366,31 @@ class SessionDB: return results @staticmethod - def _json_text_or_none(value: Any) -> Optional[str]: + def _import_text_or_none(value: Any, field: str) -> Optional[str]: if value is None: return None if isinstance(value, str): return value + raise ValueError(f"{field} must be a string") + + @staticmethod + def _import_json_object_or_none(value: Any, field: str) -> Optional[str]: + if value is None: + return None + if isinstance(value, str): + try: + parsed = json.loads(value) + except json.JSONDecodeError as exc: + raise ValueError(f"{field} must be valid JSON") from exc + if not isinstance(parsed, dict): + raise ValueError(f"{field} must be a JSON object") + return value + if not isinstance(value, dict): + raise ValueError(f"{field} must be a JSON object") try: return json.dumps(value) - except (TypeError, ValueError): - return str(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{field} must be JSON serializable") from exc @staticmethod def _float_or_none(value: Any) -> Optional[float]: @@ -5372,6 +5401,15 @@ class SessionDB: except (TypeError, ValueError): return None + @staticmethod + def _import_int_or_none(value: Any, field: str) -> Optional[int]: + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{field} must be an integer") from exc + @staticmethod def _int_or_default(value: Any, default: int = 0) -> int: if value is None: @@ -5390,6 +5428,13 @@ class SessionDB: except (json.JSONDecodeError, TypeError): return value + @staticmethod + def _import_error(index: int, session_id: str, error: str) -> Dict[str, Any]: + item: Dict[str, Any] = {"index": index, "error": error} + if session_id: + item["session_id"] = session_id + return item + def import_sessions(self, sessions: List[Dict[str, Any]]) -> Dict[str, Any]: """Import sessions exported by :meth:`export_session` or ``export_all``. @@ -5402,43 +5447,148 @@ class SessionDB: """ if not isinstance(sessions, list): raise ValueError("sessions must be a list") - if len(sessions) > 500: - raise ValueError("sessions must contain at most 500 entries") + if len(sessions) > self._IMPORT_MAX_SESSIONS: + raise ValueError( + f"sessions must contain at most {self._IMPORT_MAX_SESSIONS} entries" + ) normalized: List[Dict[str, Any]] = [] errors: List[Dict[str, Any]] = [] seen_ids: set[str] = set() + total_messages = 0 + total_bytes = 0 + session_text_fields = ( + "source", + "user_id", + "model", + "system_prompt", + "end_reason", + "cwd", + "git_branch", + "git_repo_root", + "billing_provider", + "billing_base_url", + "billing_mode", + "cost_status", + "cost_source", + "pricing_version", + "title", + ) + message_text_fields = ( + "role", + "tool_call_id", + "tool_name", + "effect_disposition", + "finish_reason", + "reasoning", + "reasoning_content", + "platform_message_id", + "message_id", + ) for index, raw in enumerate(sessions): if not isinstance(raw, dict): - errors.append({"index": index, "error": "session must be an object"}) + errors.append(self._import_error(index, "", "session must be an object")) continue session_id = str(raw.get("id") or "").strip() if not session_id: - errors.append({"index": index, "error": "session id is required"}) + errors.append(self._import_error(index, "", "session id is required")) continue if session_id in seen_ids: - errors.append( - {"index": index, "session_id": session_id, "error": "duplicate session id"} - ) + errors.append(self._import_error(index, session_id, "duplicate session id")) continue messages = raw.get("messages") or [] if not isinstance(messages, list): + errors.append(self._import_error(index, session_id, "messages must be a list")) + continue + if len(messages) > self._IMPORT_MAX_MESSAGES_PER_SESSION: errors.append( - {"index": index, "session_id": session_id, "error": "messages must be a list"} + self._import_error( + index, + session_id, + "messages exceeds the per-session import limit", + ) ) continue if any(not isinstance(msg, dict) for msg in messages): errors.append( - { - "index": index, - "session_id": session_id, - "error": "messages must contain only objects", - } + self._import_error( + index, + session_id, + "messages must contain only objects", + ) + ) + continue + + try: + session_bytes = len( + json.dumps(raw, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + ) + except (TypeError, ValueError): + errors.append( + self._import_error(index, session_id, "session must be JSON serializable") + ) + continue + if session_bytes > self._IMPORT_MAX_SESSION_BYTES: + errors.append( + self._import_error(index, session_id, "session exceeds the import size limit") + ) + continue + total_bytes += session_bytes + if total_bytes > self._IMPORT_MAX_TOTAL_BYTES: + errors.append( + self._import_error(index, session_id, "import exceeds the total size limit") + ) + continue + + try: + clean_session = dict(raw) + clean_session["id"] = session_id + clean_session["model_config"] = self._import_json_object_or_none( + clean_session.get("model_config"), "model_config" + ) + clean_session["parent_session_id"] = self._import_text_or_none( + clean_session.get("parent_session_id"), "parent_session_id" + ) + for field in session_text_fields: + clean_session[field] = self._import_text_or_none( + clean_session.get(field), field + ) + + clean_messages: List[Dict[str, Any]] = [] + for message_index, message in enumerate(messages): + clean_message = dict(message) + role = clean_message.get("role") + if not isinstance(role, str) or not role: + raise ValueError(f"messages[{message_index}].role must be a non-empty string") + for field in message_text_fields: + if field == "role": + continue + clean_message[field] = self._import_text_or_none( + clean_message.get(field), field + ) + clean_message["token_count"] = self._import_int_or_none( + clean_message.get("token_count"), "token_count" + ) + clean_messages.append(clean_message) + except ValueError as exc: + errors.append(self._import_error(index, session_id, str(exc))) + continue + + total_messages += len(clean_messages) + if total_messages > self._IMPORT_MAX_TOTAL_MESSAGES: + errors.append( + self._import_error( + index, + session_id, + "messages exceeds the total import limit", + ) ) continue seen_ids.add(session_id) - normalized.append({"index": index, "session": raw, "messages": messages}) + normalized.append( + {"index": index, "session": clean_session, "messages": clean_messages} + ) if errors: return { @@ -5499,7 +5649,7 @@ class SessionDB: "source": str(raw.get("source") or "import"), "user_id": raw.get("user_id"), "model": raw.get("model"), - "model_config": self._json_text_or_none(raw.get("model_config")), + "model_config": raw.get("model_config"), "system_prompt": raw.get("system_prompt"), "started_at": started_at, "ended_at": self._float_or_none(raw.get("ended_at")), @@ -5558,21 +5708,46 @@ class SessionDB: ) parent_id = str(raw.get("parent_session_id") or "").strip() - if parent_id and parent_id != session_id: + if parent_id: parent_updates.append((session_id, parent_id)) imported_ids.append(session_id) + parent_by_child = dict(parent_updates) + + def _would_create_cycle(session_id: str, parent_id: str) -> bool: + seen = {session_id} + current = parent_id + while current: + if current in seen: + return True + seen.add(current) + if current in parent_by_child: + current = parent_by_child[current] + continue + row = conn.execute( + "SELECT parent_session_id FROM sessions WHERE id = ? LIMIT 1", + (current,), + ).fetchone() + if row is None: + return False + current = row["parent_session_id"] + return False + for session_id, parent_id in parent_updates: parent_exists = conn.execute( "SELECT 1 FROM sessions WHERE id = ? LIMIT 1", (parent_id,), ).fetchone() - if parent_exists: + if parent_exists and not _would_create_cycle(session_id, parent_id): conn.execute( "UPDATE sessions SET parent_session_id = ? WHERE id = ?", (parent_id, session_id), ) else: + # Drop only the closing edge. Later entries can still attach + # to this now-root session, preserving the acyclic portion + # of a malformed imported lineage. + parent_by_child.pop(session_id, None) detached += 1 return { diff --git a/tests/hermes_cli/test_web_server.py b/tests/hermes_cli/test_web_server.py index 16b1962726ed..f88d60c56c57 100644 --- a/tests/hermes_cli/test_web_server.py +++ b/tests/hermes_cli/test_web_server.py @@ -1236,6 +1236,65 @@ class TestWebServerEndpoints: {"index": 0, "error": "session id is required"} ] + def test_import_sessions_endpoint_rejects_oversized_stream(self): + import hermes_cli.web_server as web_server + + payload = b'{"sessions":[]}' + b" " * web_server._SESSION_IMPORT_MAX_BYTES + response = self.client.post( + "/api/sessions/import", + content=payload, + headers={"content-type": "application/json"}, + ) + + assert response.status_code == 413 + assert response.json() == {"detail": "Session import payload is too large"} + + def test_import_sessions_endpoint_rejects_metadata_that_would_break_session_list(self): + invalid = self.client.post( + "/api/sessions/import", + json={ + "sessions": [ + { + "id": "bad-model-config", + "source": "cli", + "model_config": "{not-json", + "messages": [], + } + ] + }, + ) + + assert invalid.status_code == 400 + assert invalid.json()["detail"]["errors"] == [ + { + "index": 0, + "session_id": "bad-model-config", + "error": "model_config must be valid JSON", + } + ] + listed = self.client.get("/api/sessions") + assert listed.status_code == 200 + + @pytest.mark.parametrize( + "message", + [{"content": "missing role"}, {"role": None, "content": "null role"}], + ) + def test_import_sessions_endpoint_rejects_missing_or_null_message_role(self, message): + response = self.client.post( + "/api/sessions/import", + json={"sessions": [{"id": "bad-message-role", "messages": [message]}]}, + ) + + assert response.status_code == 400 + assert response.json()["detail"]["errors"] == [ + { + "index": 0, + "session_id": "bad-message-role", + "error": "messages[0].role must be a non-empty string", + } + ] + assert self.client.get("/api/sessions").status_code == 200 + def test_archive_session_via_patch(self): """PATCH archived=true soft-hides a session; archived=false restores it.""" from hermes_state import SessionDB diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index cbc2286cb6ed..be21f71859fb 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -2273,6 +2273,116 @@ class TestDeleteAndExport: ] assert db.get_session("valid") is None + def test_import_sessions_detaches_cycle_and_lineage_still_terminates(self, db): + result = db.import_sessions( + [ + { + "id": "a", + "source": "cli", + "parent_session_id": "b", + "end_reason": "compression", + "messages": [], + }, + { + "id": "b", + "source": "cli", + "parent_session_id": "a", + "end_reason": "compression", + "messages": [], + }, + ] + ) + + assert result["ok"] is True + assert result["detached"] == 1 + assert db.get_session("a")["parent_session_id"] is None + assert db.get_session("b")["parent_session_id"] == "a" + assert db.get_compression_lineage("a") == ["a", "b"] + + def test_import_sessions_detaches_self_parent(self, db): + result = db.import_sessions( + [ + { + "id": "self", + "source": "cli", + "parent_session_id": "self", + "end_reason": "compression", + "messages": [], + } + ] + ) + + assert result["ok"] is True + assert result["detached"] == 1 + assert db.get_session("self")["parent_session_id"] is None + + def test_compression_lineage_terminates_for_preexisting_cycle(self, db): + db.create_session("a", "cli") + db.end_session("a", "compression") + db.create_session("b", "cli", parent_session_id="a") + db.end_session("b", "compression") + db._conn.execute("UPDATE sessions SET parent_session_id = ? WHERE id = ?", ("b", "a")) + db._conn.commit() + + lineage = db.get_compression_lineage("a") + assert set(lineage) == {"a", "b"} + assert len(lineage) == 2 + assert set(db.export_session_lineage("a")["lineage_session_ids"]) == {"a", "b"} + + @pytest.mark.parametrize( + ("payload", "error"), + [ + ( + {"id": "bad-json", "model_config": "{not-json", "messages": []}, + "model_config must be valid JSON", + ), + ( + {"id": "bad-text", "user_id": {"not": "text"}, "messages": []}, + "user_id must be a string", + ), + ( + {"id": "missing-role", "messages": [{"content": "x"}]}, + "messages[0].role must be a non-empty string", + ), + ( + {"id": "null-role", "messages": [{"role": None, "content": "x"}]}, + "messages[0].role must be a non-empty string", + ), + ], + ) + def test_import_sessions_rejects_invalid_metadata(self, db, payload, error): + result = db.import_sessions([payload]) + + assert result["ok"] is False + assert result["errors"] == [{"index": 0, "session_id": payload["id"], "error": error}] + assert db.get_session(payload["id"]) is None + + def test_import_sessions_rejects_oversized_payloads_atomically(self, db): + oversized = "x" * (SessionDB._IMPORT_MAX_SESSION_BYTES + 1) + result = db.import_sessions( + [{"id": "oversized", "messages": [{"role": "user", "content": oversized}]}] + ) + + assert result["ok"] is False + assert result["errors"][0]["error"] == "session exceeds the import size limit" + assert db.get_session("oversized") is None + + result = db.import_sessions( + [ + { + "id": "too-many-messages", + "messages": [ + {"role": "user", "content": "x"} + ] + * (SessionDB._IMPORT_MAX_MESSAGES_PER_SESSION + 1), + } + ] + ) + + assert result["ok"] is False + assert result["errors"][0]["error"] == "messages exceeds the per-session import limit" + assert db.get_session("too-many-messages") is None + # ========================================================================= # Prune