mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-09 13:21:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

security(gateway): scope /resume and /sessions to the caller's origin (IDOR)

Browse source 
/resume resolved a persisted session id/title with no ownership check on any
adapter except Matrix, so an authorized caller could bind their gateway session
to another user's/room's transcript and read it. The titled-session listing and
numeric index were also globally enumerable on non-Matrix platforms, exposing
the ids and previews needed to target the IDOR.

Generalize the Matrix-only room guard to an adapter-agnostic ownership check
(live origin when active; DB row source + user_id for persisted-only sessions,
the only fields available), applied to the direct-id/title path and the
listing/numeric paths on every platform. An explicit admin --all override is
honored. The Matrix path is preserved unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
claudlos 2026-06-25 00:02:13 -05:00 • committed by Teknium
parent 5d613a5638
commit c4f278c021
3 changed files with 366 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

200
gateway/slash_commands.py
View file

@ -33,7 +33,11 @@ from agent.account_usage import fetch_account_usage, render_account_usage_lines
from agent.i18n import t
from gateway.config import HomeChannel, Platform, PlatformConfig
from gateway.platforms.base import EphemeralReply, MessageEvent, MessageType
from gateway.session import SessionSource, build_session_key
from gateway.session import (
SessionSource,
build_session_key,
is_shared_multi_user_session,
)
from hermes_cli.config import cfg_get, clear_model_endpoint_credentials
from utils import (
atomic_json_write,
@ -664,6 +668,152 @@ class GatewaySlashCommandsMixin:
and origin.chat_id == current.chat_id
)
def _same_origin_chat(self, current: SessionSource, origin: Optional[SessionSource]) -> bool:
"""Platform-agnostic counterpart to ``_same_matrix_room``.
True when *origin* shares *current*'s platform and chat, and the same
participant whenever the session key for this source is per-user. Group
and thread sessions that ``build_session_key`` isolates per participant
(the default ``group_sessions_per_user=True``) must also be scoped by
participant here — otherwise a co-member could resume another member's
live per-user group session (IDOR). Only an explicitly shared
group/thread (``group_sessions_per_user=False`` /
``thread_sessions_per_user``) lets co-members share, mirroring the key
contract via ``is_shared_multi_user_session``.
"""
if origin is None or current is None:
return False
if origin.platform != current.platform:
return False
if origin.chat_id != current.chat_id:
return False
# thread_id is part of the session key for every chat type when present
# (build_session_key appends it unconditionally), so a session in one
# thread is a DIFFERENT session from another thread of the same parent
# chat. is_shared_multi_user_session only decides participant sharing
# WITHIN a thread, never across threads — require thread equality before
# any sharing logic so a live origin in thread A cannot match a caller in
# thread B of the same parent chat.
if str(getattr(current, "thread_id", "") or "") != str(
getattr(origin, "thread_id", "") or ""
):
return False
chat_type = (getattr(current, "chat_type", "") or "").lower()
# DM-like chats are always per-user.
if chat_type in {"dm", "direct", "private", ""}:
if origin.user_id and current.user_id:
return origin.user_id == current.user_id
return True
# Non-DM: scope by participant whenever the session key for this source
# is per-user. is_shared_multi_user_session mirrors build_session_key's
# isolation rules exactly, so the guard stays in lock-step with the key.
shared = is_shared_multi_user_session(
current,
group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True),
thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False),
)
if shared:
return True
# Per-user key: compare the participant id the key is actually built
# from (user_id_alt or user_id — Signal/Feishu key on user_id_alt).
cur_pid = current.user_id_alt or current.user_id
org_pid = origin.user_id_alt or origin.user_id
if cur_pid and org_pid:
return cur_pid == org_pid
# Per-user key but a participant id is missing on one side: cannot prove
# the same owner — fail closed.
return False
def _resume_caller_is_admin(self, source: SessionSource) -> bool:
"""Whether *source* is an EXPLICITLY-configured admin allowed to make a
cross-origin /resume or /sessions listing.
Deliberately stricter than ``SlashAccessPolicy.is_admin()``: that returns
True for every allowed caller when slash gating is DISABLED (so commands
stay runnable by default), but cross-ORIGIN DATA ACCESS must require a
real, configured admin. Otherwise the default (no admin list) config
would treat every gateway caller as cross-origin-capable and re-open the
enumeration IDOR.
"""
try:
from gateway.slash_access import policy_for_source
policy = policy_for_source(self.config, source)
uid = getattr(source, "user_id", None)
return bool(policy.enabled and uid and policy.is_admin(uid))
except Exception:
return False
async def _resume_target_allowed(
self, source: SessionSource, target_id: str, allow_override: bool = False
) -> bool:
"""Whether *source* may resume the persisted session *target_id*.
Generalizes the Matrix-only room guard to every adapter so a caller
cannot bind their gateway session to another user's/room's persisted
session id (IDOR). Uses the live origin when the target is active;
otherwise falls back to the DB row's source + user_id (the sessions
table has no chat_id). An identity-bearing caller is allowed only when
the row PROVES the same owner; a row that lacks enough ownership data
fails closed. An explicit admin ``--all`` override bypasses scoping.
"""
if allow_override and self._resume_caller_is_admin(source):
return True
# Use the live origin only when it resolves to a real SessionSource; a
# store that can't resolve it (or an unexpected lookup error) must not
# silently allow/deny — fall through to the deterministic DB scoping.
try:
origin = self._gateway_session_origin_for_id(target_id)
except Exception:
origin = None
if isinstance(origin, SessionSource):
return self._same_origin_chat(source, origin)
# Inactive/persisted-only: best-effort scope by DB row source + user.
try:
row = await self._session_db.get_session(target_id) or {}
except Exception:
return False
caller_src = source.platform.value if source.platform else None
row_src = row.get("source")
if row_src and caller_src and str(row_src) != str(caller_src):
return False # different platform / source
caller_uid = str(getattr(source, "user_id", "") or "")
row_uid = str(row.get("user_id") or "")
if caller_uid:
# Identity-bearing caller: allow only when the row PROVES the same
# owner. A row with no/blank user_id cannot be proven to belong to
# this caller, so fail closed — an identified user must not bind to
# an unowned or other-owned persisted session by id/title. (Legacy
# NULL-owner rows are intentionally not resumable this way; use a
# live session or an explicit admin override.)
return bool(row_uid) and row_uid == caller_uid
# No caller identity (single-user / no-identity context): there is no
# cross-user boundary to enforce beyond the same-platform check above.
return True
async def _resume_row_visible(
self, source: SessionSource, row: dict, allow_all: bool
) -> bool:
"""Whether a titled-session listing *row* belongs to the caller's origin.
Prevents cross-origin enumeration of session ids/previews via the
numbered /resume list. Preserves the existing Matrix room-scoping
semantics; scopes every other platform to the caller's own sessions
unless an admin passes ``--all``.
"""
sid = str(row.get("id") or "")
if source.platform == Platform.MATRIX:
# Cross-room enumeration is cross-ORIGIN data access: gate the
# ``--all`` short-circuit behind a real configured admin, exactly
# like the non-Matrix branch below. A non-admin Matrix ``--all``
# falls back to same-room scoping rather than exposing every Matrix
# titled session.
if allow_all and self._resume_caller_is_admin(source):
return True
return self._same_matrix_room(source, self._gateway_session_origin_for_id(sid))
if allow_all and self._resume_caller_is_admin(source):
return True
return await self._resume_target_allowed(source, sid, allow_override=False)
async def _handle_agents_command(self, event: MessageEvent) -> str:
"""Handle /agents command - list active agents and running tasks."""
from gateway.run import _AGENT_PENDING_SENTINEL
@ -3162,13 +3312,10 @@ class GatewaySlashCommandsMixin:
# List recent titled sessions for this user/platform
try:
titled = await _list_titled_sessions()
if source.platform == Platform.MATRIX and not allow_all:
scoped = []
for s in titled:
origin = self._gateway_session_origin_for_id(str(s.get("id") or ""))
if self._same_matrix_room(source, origin):
scoped.append(s)
titled = scoped
titled = [
s for s in titled
if await self._resume_row_visible(source, s, allow_all)
]
if not titled:
if source.platform == Platform.MATRIX and not allow_all:
return t("gateway.resume.matrix_no_named_sessions")
@ -3193,13 +3340,10 @@ class GatewaySlashCommandsMixin:
if name.isdigit():
try:
titled = await _list_titled_sessions()
if source.platform == Platform.MATRIX and not allow_all:
scoped = []
for s in titled:
origin = self._gateway_session_origin_for_id(str(s.get("id") or ""))
if self._same_matrix_room(source, origin):
scoped.append(s)
titled = scoped
titled = [
s for s in titled
if await self._resume_row_visible(source, s, allow_all)
]
except Exception as e:
logger.debug("Failed to list titled sessions for numeric resume: %s", e)
return t("gateway.resume.list_failed", error=e)
@ -3236,6 +3380,14 @@ class GatewaySlashCommandsMixin:
room=target_origin.chat_name or target_origin.chat_id,
name=name,
)
elif not await self._resume_target_allowed(
source, target_id, allow_override=(allow_all or allow_cross_room)
):
# IDOR guard: a session id/title is a routing handle, not authority.
# Bind /resume to the caller's own platform/user/chat on every
# non-Matrix adapter so one user can't attach to another's
# persisted transcript.
return t("gateway.resume.blocked_not_owner", name=name)
# Check if already on that session
current_entry = self.session_store.get_or_create_session(source)
@ -3316,27 +3468,33 @@ class GatewaySlashCommandsMixin:
resume_event = dataclasses.replace(event, text=f"/resume {target}")
return await self._handle_resume_command(resume_event)
# A cross-origin listing (`/sessions all`) is honored only for an
# admin, mirroring the `/resume --all` override. `all` is just a parsed
# user argument, so without this gate any caller could run
# `/sessions all` and enumerate other origins' session ids / titles /
# previews / sources — the enumeration half of the /resume IDOR.
cross_origin = include_all and self._resume_caller_is_admin(source)
current_entry = self.session_store.get_or_create_session(source)
rows = await asyncio.to_thread(
query_session_listing,
getattr(self._session_db, "_db", self._session_db),
source=source.platform.value if source.platform else None,
current_session_id=current_entry.session_id,
include_all_sources=include_all,
include_all_sources=cross_origin,
include_unnamed=include_unnamed,
limit=10,
exclude_sources=["tool"],
)
if source.platform == Platform.MATRIX and not include_all:
if not cross_origin:
# Scope the listing to the caller's own origin on every adapter so
# session ids/previews from other users/rooms aren't enumerable.
rows = [
row for row in rows
if self._same_matrix_room(
source, self._gateway_session_origin_for_id(str(row.get("id") or ""))
)
if await self._resume_row_visible(source, row, allow_all=False)
]
return format_gateway_session_listing(
rows,
include_source=include_all,
include_source=cross_origin,
title="Sessions" if include_unnamed else "Named Sessions",
)

1
locales/en.yaml
View file

@ -239,6 +239,7 @@ gateway:
matrix_blocked_no_origin: "⚠️ Matrix /resume blocked: this named session has no recorded room origin, so Hermes will not resume it inside the current room by default. Use `/resume --cross-room {name}` if you intentionally want to cross room boundaries."
matrix_blocked_other_room: "⚠️ Matrix /resume blocked: that session belongs to a different Matrix room ({room}). Use `/resume --cross-room {name}` if you intentionally want to resume it here."
matrix_cross_room_success: "⚠️ Cross-room resume: resumed **{title}** inside Matrix room **{room}**.\nFuture messages in this room will use that transcript until `/reset` or another `/resume`.{msg_part}"
blocked_not_owner: "⚠️ /resume blocked: '**{name}**' belongs to a different user or chat. You can only resume sessions from this chat."
no_named_sessions: "No named sessions found.\nUse `/title My Session` to name your current session, then `/resume My Session` to return to it later."
list_header: "📋 **Named Sessions**\n"
list_item: "• **{title}**{preview_part}"

219
tests/gateway/test_resume_command.py
View file

@ -84,8 +84,8 @@ class TestHandleResumeCommand:
"""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")
db.create_session("sess_002", "telegram")
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")
@ -105,7 +105,7 @@ class TestHandleResumeCommand:
"""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") # No title
db.create_session("sess_001", "telegram", user_id="12345") # No title
event = _make_event(text="/resume")
runner = _make_runner(session_db=db, event=event)
@ -119,11 +119,11 @@ class TestHandleResumeCommand:
"""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")
db.create_session("sess_002", "telegram")
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")
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",
@ -141,9 +141,9 @@ class TestHandleResumeCommand:
"""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")
db.create_session("sess_001", "telegram", user_id="12345")
db.set_session_title("sess_001", "Research")
db.create_session("current_session_001", "telegram")
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",
@ -160,9 +160,9 @@ class TestHandleResumeCommand:
"""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")
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")
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",
@ -216,7 +216,7 @@ class TestHandleResumeCommand:
"""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")
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)
@ -229,7 +229,7 @@ class TestHandleResumeCommand:
"""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")
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")
@ -244,11 +244,11 @@ class TestHandleResumeCommand:
"""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")
db.create_session("sess_v1", "telegram", user_id="12345")
db.set_session_title("sess_v1", "My Project")
db.create_session("sess_v2", "telegram")
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")
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",
@ -267,12 +267,12 @@ class TestHandleResumeCommand:
from hermes_state import SessionDB
db = SessionDB(db_path=tmp_path / "state.db")
db.create_session("compressed_root", "telegram")
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", parent_session_id="compressed_root")
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")
db.create_session("current_session_001", "telegram", user_id="12345")
event = _make_event(text="/resume Compressed Work")
runner = _make_runner(
@ -300,9 +300,9 @@ class TestHandleResumeCommand:
"""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")
db.create_session("old_session", "telegram", user_id="12345")
db.set_session_title("old_session", "Old Work")
db.create_session("current_session_001", "telegram")
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",
@ -326,9 +326,9 @@ class TestHandleResumeCommand:
import threading
from hermes_state import SessionDB
db = SessionDB(db_path=tmp_path / "state.db")
db.create_session("old_session", "telegram")
db.create_session("old_session", "telegram", user_id="12345")
db.set_session_title("old_session", "Old Work")
db.create_session("current_session_001", "telegram")
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",
@ -353,9 +353,9 @@ class TestHandleResumeCommand:
"""
from hermes_state import SessionDB
db = SessionDB(db_path=tmp_path / "state.db")
db.create_session("abc123", "telegram")
db.create_session("abc123", "telegram", user_id="12345")
db.set_session_title("abc123", "Bracketed")
db.create_session("current_session_001", "telegram")
db.create_session("current_session_001", "telegram", user_id="12345")
for raw in ("<abc123>", "[abc123]", '"abc123"', "'abc123'"):
event = _make_event(text=f"/resume {raw}")
@ -382,9 +382,9 @@ class TestHandleResumeCommand:
"""
from hermes_state import SessionDB
db = SessionDB(db_path=tmp_path / "state.db")
db.create_session("unnamed_session_xyz", "telegram")
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")
db.create_session("current_session_001", "telegram", user_id="12345")
event = _make_event(text="/resume unnamed_session_xyz")
runner = _make_runner(
@ -409,7 +409,7 @@ class TestHandleSessionsCommand:
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")
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")
@ -426,12 +426,17 @@ class TestHandleSessionsCommand:
db.close()
@pytest.mark.asyncio
async def test_sessions_all_full_lists_cross_platform_unnamed_sessions(self, tmp_path):
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")
db.create_session("tg_named", "telegram", user_id="12345")
db.set_session_title("tg_named", "Telegram Work")
db.create_session("discord_unnamed", "discord")
db.create_session("discord_unnamed", "discord") # other origin
db.append_message("discord_unnamed", "user", "discord first prompt")
event = _make_event(text="/sessions all full")
@ -439,16 +444,40 @@ class TestHandleSessionsCommand:
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" in result
assert "discord" 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_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")
db.create_session("tg_session", "telegram", user_id="12345")
db.set_session_title("tg_session", "Telegram Work")
event = _make_event(text="/sessions")
@ -460,3 +489,127 @@ class TestHandleSessionsCommand:
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