mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-17 14:42:06 +00:00
33a5090bf6
Addresses egilewski (Codex/CodeRabbit) follow-up on PR #52355: the no-identity branch of _resume_target_allowed() returned True after only checking that the row's source didn't mismatch the caller platform. The sessions table has no chat_id, so same-platform alone is not ownership proof — a Telegram group caller in chat-a with user_id=None could resume (and /sessions could list) a persisted row owned by another chat/user (e.g. victim_chat_b_uid, source=telegram, user_id=victim). Fail closed: an identity-less caller can no longer bind to or enumerate a persisted session by id/title. A legitimate same-chat resume of an ACTIVE session still works via the live-origin branch (which compares chat_id), and an operator can use the admin --all override. The listing path inherits the fix because _resume_row_visible() routes non-Matrix rows through the same helper. Adds an end-to-end no-identity probe (resume blocked) and a unit-level persisted-fallback assertion. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
729 lines
33 KiB
Python
729 lines
33 KiB
Python
"""Tests for /resume gateway slash command.
|
|
|
|
Tests the _handle_resume_command handler (switch to a previously-named session)
|
|
across gateway messenger platforms.
|
|
"""
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from gateway.config import Platform
|
|
from gateway.platforms.base import MessageEvent
|
|
from gateway.session import SessionSource, build_session_key
|
|
|
|
|
|
def _make_event(text="/resume", platform=Platform.TELEGRAM,
|
|
user_id="12345", chat_id="67890"):
|
|
"""Build a MessageEvent for testing."""
|
|
source = SessionSource(
|
|
platform=platform,
|
|
user_id=user_id,
|
|
chat_id=chat_id,
|
|
user_name="testuser",
|
|
)
|
|
return MessageEvent(text=text, source=source)
|
|
|
|
|
|
def _session_key_for_event(event):
|
|
"""Get the session key that build_session_key produces for an event."""
|
|
return build_session_key(event.source)
|
|
|
|
|
|
def _make_runner(session_db=None, current_session_id="current_session_001",
|
|
event=None):
|
|
"""Create a bare GatewayRunner with a mock session_store and optional session_db."""
|
|
from gateway.run import GatewayRunner
|
|
runner = object.__new__(GatewayRunner)
|
|
runner.adapters = {}
|
|
runner.config = SimpleNamespace(platforms={})
|
|
runner._voice_mode = {}
|
|
# Gateway holds the async facade; the slash handlers await it.
|
|
if session_db is not None:
|
|
from hermes_state import AsyncSessionDB
|
|
session_db = AsyncSessionDB(session_db)
|
|
runner._session_db = session_db
|
|
runner._running_agents = {}
|
|
runner._is_user_authorized = lambda _source: True
|
|
|
|
# Compute the real session key if an event is provided
|
|
session_key = build_session_key(event.source) if event else "agent:main:telegram:dm"
|
|
|
|
# Mock session_store that returns a session entry with a known session_id
|
|
mock_session_entry = MagicMock()
|
|
mock_session_entry.session_id = current_session_id
|
|
mock_session_entry.session_key = session_key
|
|
mock_store = MagicMock()
|
|
mock_store.get_or_create_session.return_value = mock_session_entry
|
|
mock_store.load_transcript.return_value = []
|
|
mock_store.switch_session.return_value = mock_session_entry
|
|
runner.session_store = mock_store
|
|
|
|
return runner
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _handle_resume_command
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHandleResumeCommand:
|
|
"""Tests for GatewayRunner._handle_resume_command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_session_db(self):
|
|
"""Returns error when session database is unavailable."""
|
|
runner = _make_runner(session_db=None)
|
|
event = _make_event(text="/resume My Project")
|
|
result = await runner._handle_resume_command(event)
|
|
assert "not available" in result.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_named_sessions_when_no_arg(self, tmp_path):
|
|
"""With no argument, lists recently titled sessions."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("sess_001", "telegram", user_id="12345")
|
|
db.create_session("sess_002", "telegram", user_id="12345")
|
|
db.set_session_title("sess_001", "Research")
|
|
db.set_session_title("sess_002", "Coding")
|
|
|
|
event = _make_event(text="/resume")
|
|
runner = _make_runner(session_db=db, event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
assert "Research" in result
|
|
assert "Coding" in result
|
|
assert "Named Sessions" in result
|
|
assert "1." in result
|
|
assert "2." in result
|
|
assert "/resume 1" in result
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_shows_usage_when_no_titled(self, tmp_path):
|
|
"""With no arg and no titled sessions, shows instructions."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("sess_001", "telegram", user_id="12345") # No title
|
|
|
|
event = _make_event(text="/resume")
|
|
runner = _make_runner(session_db=db, event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
assert "No named sessions" in result
|
|
assert "/title" in result
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_by_index(self, tmp_path):
|
|
"""Numeric argument resumes the indexed titled session from the list."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("sess_001", "telegram", user_id="12345")
|
|
db.create_session("sess_002", "telegram", user_id="12345")
|
|
db.set_session_title("sess_001", "Research")
|
|
db.set_session_title("sess_002", "Coding")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
event = _make_event(text="/resume 2")
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
|
|
assert "Resumed" in result
|
|
runner.session_store.switch_session.assert_called_once()
|
|
call_args = runner.session_store.switch_session.call_args
|
|
assert call_args[0][1] == "sess_001"
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_index_out_of_range(self, tmp_path):
|
|
"""Out-of-range numeric arguments show a helpful error."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("sess_001", "telegram", user_id="12345")
|
|
db.set_session_title("sess_001", "Research")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
event = _make_event(text="/resume 9")
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
|
|
assert "out of range" in result.lower()
|
|
assert "/resume" in result
|
|
runner.session_store.switch_session.assert_not_called()
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_by_name(self, tmp_path):
|
|
"""Resolves a title and switches to that session."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("old_session_abc", "telegram", user_id="12345")
|
|
db.set_session_title("old_session_abc", "My Project")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
event = _make_event(text="/resume My Project")
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
|
|
assert "Resumed" in result
|
|
assert "My Project" in result
|
|
# Verify switch_session was called with the old session ID
|
|
runner.session_store.switch_session.assert_called_once()
|
|
call_args = runner.session_store.switch_session.call_args
|
|
assert call_args[0][1] == "old_session_abc"
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_clears_session_model_overrides(self, tmp_path):
|
|
"""Resume must not carry a previous session's /model override into the
|
|
restored conversation, while leaving other chats' overrides intact (#10702)."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("old_session_abc", "telegram")
|
|
db.set_session_title("old_session_abc", "My Project")
|
|
db.create_session("current_session_001", "telegram")
|
|
|
|
event = _make_event(text="/resume My Project")
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
key = _session_key_for_event(event)
|
|
runner._session_model_overrides = {
|
|
key: {"model": "gpt-5", "provider": "openai"},
|
|
"agent:main:telegram:dm:other": {"model": "keep-me"},
|
|
}
|
|
runner._pending_model_notes = {
|
|
key: "[Note: switched to gpt-5]",
|
|
"agent:main:telegram:dm:other": "[Note: keep-me]",
|
|
}
|
|
|
|
result = await runner._handle_resume_command(event)
|
|
|
|
assert "Resumed" in result
|
|
# The resumed chat's override + pending note are cleared...
|
|
assert key not in runner._session_model_overrides
|
|
assert key not in runner._pending_model_notes
|
|
# ...but an unrelated chat's state is untouched.
|
|
assert runner._session_model_overrides["agent:main:telegram:dm:other"] == {"model": "keep-me"}
|
|
assert runner._pending_model_notes["agent:main:telegram:dm:other"] == "[Note: keep-me]"
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_nonexistent_name(self, tmp_path):
|
|
"""Returns error for unknown session name."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
event = _make_event(text="/resume Nonexistent Session")
|
|
runner = _make_runner(session_db=db, event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
assert "No session found" in result
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_already_on_session(self, tmp_path):
|
|
"""Returns friendly message when already on the requested session."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
db.set_session_title("current_session_001", "Active Project")
|
|
|
|
event = _make_event(text="/resume Active Project")
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
assert "Already on session" in result
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_auto_lineage(self, tmp_path):
|
|
"""Asking for 'My Project' when 'My Project #2' exists gets the latest."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("sess_v1", "telegram", user_id="12345")
|
|
db.set_session_title("sess_v1", "My Project")
|
|
db.create_session("sess_v2", "telegram", user_id="12345")
|
|
db.set_session_title("sess_v2", "My Project #2")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
event = _make_event(text="/resume My Project")
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
|
|
assert "Resumed" in result
|
|
# Should resolve to #2 (latest in lineage)
|
|
call_args = runner.session_store.switch_session.call_args
|
|
assert call_args[0][1] == "sess_v2"
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_follows_compression_continuation(self, tmp_path):
|
|
"""Gateway /resume should reopen the live descendant after compression."""
|
|
from hermes_state import SessionDB
|
|
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("compressed_root", "telegram", user_id="12345")
|
|
db.set_session_title("compressed_root", "Compressed Work")
|
|
db.end_session("compressed_root", "compression")
|
|
db.create_session("compressed_child", "telegram", user_id="12345", parent_session_id="compressed_root")
|
|
db.append_message("compressed_child", "user", "hello from continuation")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
event = _make_event(text="/resume Compressed Work")
|
|
runner = _make_runner(
|
|
session_db=db,
|
|
current_session_id="current_session_001",
|
|
event=event,
|
|
)
|
|
runner.session_store.load_transcript.side_effect = (
|
|
lambda session_id: [{"role": "user", "content": "hello from continuation"}]
|
|
if session_id == "compressed_child"
|
|
else []
|
|
)
|
|
|
|
result = await runner._handle_resume_command(event)
|
|
|
|
assert "Resumed session" in result
|
|
assert "(1 message)" in result
|
|
call_args = runner.session_store.switch_session.call_args
|
|
assert call_args[0][1] == "compressed_child"
|
|
runner.session_store.load_transcript.assert_called_with("compressed_child")
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_clears_running_agent(self, tmp_path):
|
|
"""Switching sessions clears any cached running agent."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("old_session", "telegram", user_id="12345")
|
|
db.set_session_title("old_session", "Old Work")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
event = _make_event(text="/resume Old Work")
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
# Simulate a running agent using the real session key
|
|
real_key = _session_key_for_event(event)
|
|
runner._running_agents[real_key] = MagicMock()
|
|
|
|
await runner._handle_resume_command(event)
|
|
|
|
assert real_key not in runner._running_agents
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_evicts_cached_agent(self, tmp_path):
|
|
"""Gateway /resume evicts the cached AIAgent so the next message
|
|
rebuilds with the correct session_id end-to-end — mirrors /branch
|
|
and /reset. Without this, the cached agent's memory provider keeps
|
|
writing into the wrong session. See #6672.
|
|
"""
|
|
import threading
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("old_session", "telegram", user_id="12345")
|
|
db.set_session_title("old_session", "Old Work")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
event = _make_event(text="/resume Old Work")
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
# Seed the cache with a fake agent
|
|
real_key = _session_key_for_event(event)
|
|
runner._agent_cache = {real_key: (MagicMock(), object())}
|
|
runner._agent_cache_lock = threading.RLock()
|
|
|
|
await runner._handle_resume_command(event)
|
|
|
|
assert real_key not in runner._agent_cache
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_strips_outer_brackets(self, tmp_path):
|
|
"""Users may copy `<session_id>` from the usage hint literally.
|
|
|
|
The gateway should strip outer ``<>``, ``[]``, ``""``, and ``''``
|
|
before lookup so ``/resume <abc123>`` works the same as
|
|
``/resume abc123``.
|
|
"""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("abc123", "telegram", user_id="12345")
|
|
db.set_session_title("abc123", "Bracketed")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
for raw in ("<abc123>", "[abc123]", '"abc123"', "'abc123'"):
|
|
event = _make_event(text=f"/resume {raw}")
|
|
runner = _make_runner(
|
|
session_db=db,
|
|
current_session_id="current_session_001",
|
|
event=event,
|
|
)
|
|
result = await runner._handle_resume_command(event)
|
|
# Either the session was resumed (and we get a "Resumed" / "Already on" reply)
|
|
# or it was found-then-redirected. Failure mode = "No session found matching '<abc123>'".
|
|
assert "abc123" not in str(result) or "not found" not in str(result).lower(), (
|
|
f"bracket stripping failed for {raw!r}: gateway returned {result!r}"
|
|
)
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_resolves_by_session_id(self, tmp_path):
|
|
"""The gateway should accept a bare session ID, not just a title.
|
|
|
|
Before this fix, /resume in the gateway only called
|
|
``resolve_session_by_title``, so ``/resume <session_id>`` always
|
|
returned "Session not found" even for valid IDs.
|
|
"""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("unnamed_session_xyz", "telegram", user_id="12345")
|
|
# Deliberately no title set — this session can ONLY be resolved by ID.
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
event = _make_event(text="/resume unnamed_session_xyz")
|
|
runner = _make_runner(
|
|
session_db=db,
|
|
current_session_id="current_session_001",
|
|
event=event,
|
|
)
|
|
result = await runner._handle_resume_command(event)
|
|
|
|
# Should NOT be the not-found error.
|
|
assert "not found" not in str(result).lower(), (
|
|
f"session-id lookup failed: {result!r}"
|
|
)
|
|
db.close()
|
|
|
|
|
|
|
|
class TestHandleSessionsCommand:
|
|
"""Tests for GatewayRunner._handle_sessions_command."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sessions_command_lists_current_platform_sessions(self, tmp_path):
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("tg_session", "telegram", user_id="12345")
|
|
db.set_session_title("tg_session", "Telegram Work")
|
|
db.create_session("discord_session", "discord")
|
|
db.set_session_title("discord_session", "Discord Work")
|
|
|
|
event = _make_event(text="/sessions")
|
|
runner = _make_runner(session_db=db, event=event)
|
|
|
|
result = await runner._handle_sessions_command(event)
|
|
|
|
assert "Sessions" in result
|
|
assert "Telegram Work" in result
|
|
assert "tg_session" in result
|
|
assert "Discord Work" not in result
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sessions_all_does_not_leak_cross_origin_for_non_admin(self, tmp_path):
|
|
"""`/sessions all` from a non-admin caller must stay scoped to the
|
|
caller's own origin — it must NOT enumerate other origins' sessions
|
|
(the enumeration half of the /resume IDOR). Cross-origin listing is
|
|
gated behind an explicitly-configured admin, which the default test
|
|
config is not."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("tg_named", "telegram", user_id="12345")
|
|
db.set_session_title("tg_named", "Telegram Work")
|
|
db.create_session("discord_unnamed", "discord") # other origin
|
|
db.append_message("discord_unnamed", "user", "discord first prompt")
|
|
|
|
event = _make_event(text="/sessions all full")
|
|
runner = _make_runner(session_db=db, event=event)
|
|
|
|
result = await runner._handle_sessions_command(event)
|
|
|
|
# Caller's own (telegram) session is shown; the cross-origin (discord)
|
|
# session is NOT leaked even with `all`.
|
|
assert "Telegram Work" in result
|
|
assert "discord_unnamed" not in result
|
|
assert "Discord" not in result
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_blocks_cross_user_and_unowned_rows(self, tmp_path):
|
|
"""An identity-bearing caller cannot resume a session it can't prove it
|
|
owns: a row owned by a different user, or a same-platform row with no
|
|
recorded owner (NULL user_id) must both be denied (IDOR)."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("victim_other_uid", "telegram", user_id="99999")
|
|
db.set_session_title("victim_other_uid", "Other User")
|
|
db.create_session("victim_missing_uid", "telegram") # NULL owner
|
|
db.set_session_title("victim_missing_uid", "Unowned")
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
for name in ("Other User", "victim_other_uid", "Unowned", "victim_missing_uid"):
|
|
event = _make_event(text=f"/resume {name}")
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
runner.session_store.switch_session.assert_not_called()
|
|
assert "Resumed" not in result, name
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_blocks_blank_source_same_uid_row(self, tmp_path):
|
|
"""A persisted row whose `source` is blank/legacy cannot prove it shares
|
|
the caller's platform, so user_id equality alone must NOT authorize a
|
|
resume — the blank source fails closed exactly like a missing user_id
|
|
(IDOR regression: an identified caller could otherwise bind to an
|
|
unproven-origin transcript)."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("blank_source_same_uid", "telegram", user_id="12345")
|
|
db.set_session_title("blank_source_same_uid", "Blank Source Same UID")
|
|
# Simulate a malformed/legacy row that does not record its origin.
|
|
db._conn.execute(
|
|
"UPDATE sessions SET source = '' WHERE id = ?", ("blank_source_same_uid",)
|
|
)
|
|
db._conn.commit()
|
|
db.create_session("current_session_001", "telegram", user_id="12345")
|
|
|
|
for name in ("Blank Source Same UID", "blank_source_same_uid"):
|
|
event = _make_event(text=f"/resume {name}")
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
runner.session_store.switch_session.assert_not_called()
|
|
assert "Resumed" not in result, name
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_blocks_no_identity_caller_on_persisted_row(self, tmp_path):
|
|
"""A caller with no user_id must not resume a persisted row on
|
|
same-platform alone: the row has no chat_id to prove ownership, so a
|
|
Telegram group caller in chat-a (user_id=None) cannot bind to a row
|
|
owned by another chat/user (IDOR regression for the no-identity branch)."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("victim_chat_b_uid", "telegram", user_id="victim")
|
|
db.set_session_title("victim_chat_b_uid", "Victim Chat B")
|
|
db.create_session("current_session_001", "telegram")
|
|
|
|
for name in ("Victim Chat B", "victim_chat_b_uid"):
|
|
event = _make_event(text=f"/resume {name}", user_id=None,
|
|
chat_id="chat-a")
|
|
event.source.chat_type = "group"
|
|
runner = _make_runner(session_db=db, current_session_id="current_session_001",
|
|
event=event)
|
|
result = await runner._handle_resume_command(event)
|
|
runner.session_store.switch_session.assert_not_called()
|
|
assert "Resumed" not in result, name
|
|
db.close()
|
|
|
|
def test_resume_target_allowed_blocks_no_identity_persisted(self, tmp_path):
|
|
"""Unit-level: the persisted-row fallback fails closed for an
|
|
identity-less caller (no live origin resolvable)."""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("victim_chat_b_uid", "telegram", user_id="victim")
|
|
runner = _make_runner(session_db=db)
|
|
runner._gateway_session_origin_for_id = lambda sid: None # inactive/persisted-only
|
|
caller = SessionSource(platform=Platform.TELEGRAM, chat_id="chat-a",
|
|
chat_type="group", user_id=None)
|
|
assert runner._resume_target_allowed(caller, "victim_chat_b_uid",
|
|
allow_override=False) is False
|
|
db.close()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_dispatches_sessions_command(self, tmp_path):
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=tmp_path / "state.db")
|
|
db.create_session("tg_session", "telegram", user_id="12345")
|
|
db.set_session_title("tg_session", "Telegram Work")
|
|
|
|
event = _make_event(text="/sessions")
|
|
runner = _make_runner(session_db=db, event=event)
|
|
runner._handle_sessions_command = AsyncMock(return_value="sessions output")
|
|
|
|
result = await runner._handle_message(event)
|
|
|
|
assert result == "sessions output"
|
|
runner._handle_sessions_command.assert_awaited_once_with(event)
|
|
db.close()
|
|
|
|
|
|
class TestSameOriginChatGroupScoping:
|
|
"""Live group sessions are per-user by default (group_sessions_per_user=True),
|
|
so a co-member must not be able to resume another member's live group session
|
|
via the live-origin branch of _resume_target_allowed (IDOR)."""
|
|
|
|
@staticmethod
|
|
def _src(user_id, *, chat_type="group", chat_id="guild-123",
|
|
platform=Platform.DISCORD, user_id_alt=None, thread_id=None):
|
|
return SessionSource(platform=platform, chat_id=chat_id,
|
|
chat_type=chat_type, user_id=user_id,
|
|
user_id_alt=user_id_alt, thread_id=thread_id)
|
|
|
|
def test_blocks_cross_user_live_group_by_default(self):
|
|
runner = _make_runner()
|
|
assert runner._same_origin_chat(self._src("alice"), self._src("bob")) is False
|
|
|
|
def test_allows_same_user_live_group(self):
|
|
runner = _make_runner()
|
|
assert runner._same_origin_chat(self._src("alice"), self._src("alice")) is True
|
|
|
|
def test_allows_cross_user_when_group_explicitly_shared(self):
|
|
runner = _make_runner()
|
|
runner.config.group_sessions_per_user = False
|
|
assert runner._same_origin_chat(self._src("alice"), self._src("bob")) is True
|
|
|
|
def test_dm_cross_user_still_blocked(self):
|
|
runner = _make_runner()
|
|
a = self._src("alice", chat_type="dm", chat_id="dm-1")
|
|
b = self._src("bob", chat_type="dm", chat_id="dm-1")
|
|
assert runner._same_origin_chat(a, b) is False
|
|
|
|
def test_resume_target_allowed_blocks_cross_user_live_group(self):
|
|
"""End-to-end via the live-origin branch: Alice cannot resume Bob's
|
|
active group session in the same chat."""
|
|
runner = _make_runner()
|
|
bob = self._src("bob")
|
|
runner._gateway_session_origin_for_id = lambda sid: bob
|
|
assert runner._resume_target_allowed(
|
|
self._src("alice"), "bobs_live_sid", allow_override=False
|
|
) is False
|
|
|
|
# --- thread scoping: thread_id is part of the session key, so a session in
|
|
# one thread must never match a caller in another thread of the same chat,
|
|
# even when threads are shared among participants by default. ---
|
|
|
|
def test_blocks_cross_thread_same_user_same_chat(self):
|
|
"""Same user, same parent chat, different thread → different session."""
|
|
runner = _make_runner()
|
|
a = self._src("alice", thread_id="thread-A")
|
|
b = self._src("alice", thread_id="thread-B")
|
|
assert runner._same_origin_chat(a, b) is False
|
|
|
|
def test_allows_same_thread_shared_participants(self):
|
|
"""Threads are shared by default (thread_sessions_per_user=False), so
|
|
co-members in the SAME thread share the session."""
|
|
runner = _make_runner()
|
|
a = self._src("alice", thread_id="thread-A")
|
|
b = self._src("bob", thread_id="thread-A")
|
|
assert runner._same_origin_chat(a, b) is True
|
|
|
|
def test_blocks_cross_thread_even_when_shared(self):
|
|
"""Cross-thread is blocked regardless of thread-sharing: sharing only
|
|
applies WITHIN a thread, never across threads."""
|
|
runner = _make_runner()
|
|
a = self._src("alice", thread_id="thread-A")
|
|
b = self._src("bob", thread_id="thread-B")
|
|
assert runner._same_origin_chat(a, b) is False
|
|
|
|
def test_blocks_thread_vs_no_thread(self):
|
|
"""A threaded origin must not match a non-threaded caller in the same
|
|
parent chat (and vice versa)."""
|
|
runner = _make_runner()
|
|
threaded = self._src("alice", thread_id="thread-A")
|
|
parent = self._src("alice", thread_id=None)
|
|
assert runner._same_origin_chat(parent, threaded) is False
|
|
assert runner._same_origin_chat(threaded, parent) is False
|
|
|
|
|
|
class TestResumeRowVisibleMatrixAllScoping:
|
|
"""Non-admin Matrix `/resume --all` must NOT enumerate every Matrix titled
|
|
session: the cross-room listing short-circuit is admin-only, mirroring the
|
|
non-Matrix branch. A non-admin `--all` falls back to same-room scoping."""
|
|
|
|
@staticmethod
|
|
def _matrix_src(chat_id="!room-a:hs", user_id="@alice:hs"):
|
|
return SessionSource(platform=Platform.MATRIX, chat_id=chat_id,
|
|
chat_type="group", user_id=user_id)
|
|
|
|
def test_non_admin_all_does_not_expose_other_room(self):
|
|
runner = _make_runner()
|
|
runner._resume_caller_is_admin = lambda src: False
|
|
# Titled row whose live origin is a DIFFERENT Matrix room.
|
|
other_room = SessionSource(platform=Platform.MATRIX, chat_id="!room-b:hs",
|
|
chat_type="group", user_id="@bob:hs")
|
|
runner._gateway_session_origin_for_id = lambda sid: other_room
|
|
row = {"id": "sid_other_room"}
|
|
assert runner._resume_row_visible(self._matrix_src(), row, allow_all=True) is False
|
|
|
|
def test_non_admin_all_still_shows_same_room(self):
|
|
runner = _make_runner()
|
|
runner._resume_caller_is_admin = lambda src: False
|
|
same_room = SessionSource(platform=Platform.MATRIX, chat_id="!room-a:hs",
|
|
chat_type="group", user_id="@bob:hs")
|
|
runner._gateway_session_origin_for_id = lambda sid: same_room
|
|
row = {"id": "sid_same_room"}
|
|
assert runner._resume_row_visible(self._matrix_src(), row, allow_all=True) is True
|
|
|
|
def test_admin_all_exposes_cross_room(self):
|
|
runner = _make_runner()
|
|
runner._resume_caller_is_admin = lambda src: True
|
|
other_room = SessionSource(platform=Platform.MATRIX, chat_id="!room-b:hs",
|
|
chat_type="group", user_id="@bob:hs")
|
|
runner._gateway_session_origin_for_id = lambda sid: other_room
|
|
row = {"id": "sid_other_room"}
|
|
assert runner._resume_row_visible(self._matrix_src(), row, allow_all=True) is True
|
|
|
|
def test_non_admin_all_fails_closed_on_unknown_origin(self):
|
|
runner = _make_runner()
|
|
runner._resume_caller_is_admin = lambda src: False
|
|
runner._gateway_session_origin_for_id = lambda sid: None
|
|
row = {"id": "sid_unknown"}
|
|
assert runner._resume_row_visible(self._matrix_src(), row, allow_all=True) is False
|
|
|
|
|
|
class TestSameMatrixRoomThreadScoping:
|
|
"""Matrix `/resume` (direct and listing) scopes by room AND thread: a live
|
|
session in another thread of the same room is a different session
|
|
(build_session_key appends thread_id), so a caller in thread A must not
|
|
resume/enumerate a target whose origin is in thread B. Non-threaded rooms
|
|
keep room-level sharing unchanged."""
|
|
|
|
@staticmethod
|
|
def _msrc(chat_id="!room-a:hs", user_id="@alice:hs", thread_id=None):
|
|
return SessionSource(platform=Platform.MATRIX, chat_id=chat_id,
|
|
chat_type="group", user_id=user_id, thread_id=thread_id)
|
|
|
|
def test_same_room_no_thread_still_shared(self):
|
|
runner = _make_runner()
|
|
a = self._msrc(user_id="@alice:hs")
|
|
b = self._msrc(user_id="@bob:hs")
|
|
assert runner._same_matrix_room(a, b) is True
|
|
|
|
def test_same_room_same_thread_shared(self):
|
|
runner = _make_runner()
|
|
a = self._msrc(user_id="@alice:hs", thread_id="thr-1")
|
|
b = self._msrc(user_id="@bob:hs", thread_id="thr-1")
|
|
assert runner._same_matrix_room(a, b) is True
|
|
|
|
def test_cross_thread_same_room_blocked(self):
|
|
"""The reviewer's probe: caller in thread-a, target origin in thread-b
|
|
of the same room → must not match."""
|
|
runner = _make_runner()
|
|
caller = self._msrc(thread_id="thread-a")
|
|
victim_origin = self._msrc(thread_id="thread-b")
|
|
assert runner._same_matrix_room(caller, victim_origin) is False
|
|
|
|
def test_thread_vs_no_thread_blocked(self):
|
|
runner = _make_runner()
|
|
threaded = self._msrc(thread_id="thread-a")
|
|
room_level = self._msrc(thread_id=None)
|
|
assert runner._same_matrix_room(threaded, room_level) is False
|
|
assert runner._same_matrix_room(room_level, threaded) is False
|
|
|
|
def test_resume_row_visible_blocks_cross_thread(self):
|
|
"""End-to-end through the Matrix listing guard."""
|
|
runner = _make_runner()
|
|
runner._resume_caller_is_admin = lambda src: False
|
|
origin_thread_b = self._msrc(thread_id="thread-b")
|
|
runner._gateway_session_origin_for_id = lambda sid: origin_thread_b
|
|
row = {"id": "sid_thread_b"}
|
|
caller_thread_a = self._msrc(thread_id="thread-a")
|
|
assert runner._resume_row_visible(caller_thread_a, row, allow_all=False) is False
|