mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-20 15:33:54 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(sessions): validate imported session payloads

Browse source 
Reject metadata that would make session queries fail, bound import work, and detach cyclic lineage links. Guard lineage traversal against pre-existing corrupt cycles.
This commit is contained in:
kshitijk4poor 2026-07-13 13:21:35 +05:30 • committed by kshitij
parent b51d365ef0
commit ac705b52c9
4 changed files with 400 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

40
hermes_cli/web_server.py
View file

@ -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)

219
hermes_state.py
View file

@ -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 {

59
tests/hermes_cli/test_web_server.py
View file

@ -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

110
tests/test_hermes_state.py
View file

@ -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