fix: scope file manager sessions to active profile
This commit is contained in:
1 parent
2d9da824fe
commit
beb39f1ef2
2 files changed
+96
-6
No files matched your search
+14
-5
@@ -3392,21 +3392,30 @@ class _ExternalSessionView:
|
||||
|
||||
|
||||
def get_session_for_file_ops(sid: str):
|
||||
"""Return a session-like object for file-manager handlers.
|
||||
"""Return a profile-authorized session-like object for file-manager handlers.
|
||||
|
||||
Tries ``get_session`` first (preserves all existing behavior for WebUI
|
||||
sessions). If that raises ``KeyError``, checks state.db; when the session
|
||||
exists there, returns an ``_ExternalSessionView`` whose ``workspace`` is
|
||||
the active WebUI workspace. If neither has the session, re-raises
|
||||
sessions) and only returns that session when its stored profile belongs to
|
||||
the active request profile. If that lookup fails, checks state.db; when the
|
||||
session exists there, returns an ``_ExternalSessionView`` whose ``workspace``
|
||||
is the active WebUI workspace. If neither has the session, re-raises
|
||||
``KeyError`` so callers continue to return their existing 404.
|
||||
"""
|
||||
try:
|
||||
return get_session(sid, metadata_only=True)
|
||||
session = get_session(sid, metadata_only=True)
|
||||
except KeyError:
|
||||
if state_db_has_session(sid):
|
||||
return _ExternalSessionView(str(sid), str(get_last_workspace()))
|
||||
raise
|
||||
|
||||
from api.profiles import _profiles_match, get_active_profile_name
|
||||
|
||||
if not _profiles_match(
|
||||
getattr(session, 'profile', None), get_active_profile_name()
|
||||
):
|
||||
raise KeyError(sid)
|
||||
return session
|
||||
|
||||
|
||||
def _active_state_db_path() -> Path:
|
||||
"""Return state.db for the active Hermes profile, degrading to HERMES_HOME."""
|
||||
|
||||
@@ -11,9 +11,12 @@ Covers:
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import re
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -107,7 +110,8 @@ def models_module():
|
||||
|
||||
def test_get_session_for_file_ops_webui_passthrough(models_module, monkeypatch):
|
||||
"""(a) WebUI session — delegates to get_session, no state.db consulted."""
|
||||
sentinel = object()
|
||||
profiles_module = pytest.importorskip("api.profiles")
|
||||
sentinel = SimpleNamespace(profile=None)
|
||||
called = {"get_session": 0, "state_db": 0}
|
||||
|
||||
def fake_get_session(sid, metadata_only=False):
|
||||
@@ -120,11 +124,88 @@ def test_get_session_for_file_ops_webui_passthrough(models_module, monkeypatch):
|
||||
|
||||
monkeypatch.setattr(models_module, "get_session", fake_get_session)
|
||||
monkeypatch.setattr(models_module, "state_db_has_session", fake_has)
|
||||
monkeypatch.setattr(profiles_module, "get_active_profile_name", lambda: "default")
|
||||
result = models_module.get_session_for_file_ops("webui-sid")
|
||||
assert result is sentinel
|
||||
assert called == {"get_session": 1, "state_db": 0}
|
||||
|
||||
|
||||
def test_get_session_for_file_ops_rejects_foreign_profile(
|
||||
models_module, monkeypatch, tmp_path
|
||||
):
|
||||
"""WebUI sessions must belong to the active profile before file access."""
|
||||
profiles_module = pytest.importorskip("api.profiles")
|
||||
foreign_session = SimpleNamespace(profile="research", workspace=str(tmp_path))
|
||||
called = {"get_session": 0, "state_db": 0}
|
||||
|
||||
def fake_get_session(sid, metadata_only=False):
|
||||
called["get_session"] += 1
|
||||
return foreign_session
|
||||
|
||||
def fake_has(_sid):
|
||||
called["state_db"] += 1
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(models_module, "get_session", fake_get_session)
|
||||
monkeypatch.setattr(models_module, "state_db_has_session", fake_has)
|
||||
monkeypatch.setattr(profiles_module, "get_active_profile_name", lambda: "default")
|
||||
|
||||
with pytest.raises(KeyError):
|
||||
models_module.get_session_for_file_ops("foreign-webui-sid")
|
||||
# A found-but-foreign WebUI sidecar is an authorization failure, not a
|
||||
# missing-session condition that can fall through to the state.db fallback.
|
||||
assert called == {"get_session": 1, "state_db": 0}
|
||||
|
||||
|
||||
def test_file_read_rejects_foreign_profile_session(
|
||||
models_module, monkeypatch, tmp_path
|
||||
):
|
||||
"""A default-profile file route cannot read a named-profile workspace."""
|
||||
profiles_module = pytest.importorskip("api.profiles")
|
||||
routes_module = pytest.importorskip("api.routes")
|
||||
workspace = tmp_path / "named-workspace"
|
||||
workspace.mkdir()
|
||||
(workspace / "marker.txt").write_text("foreign profile marker")
|
||||
session = models_module.Session(
|
||||
session_id="foreign-profile-file-read",
|
||||
workspace=str(workspace),
|
||||
profile="research",
|
||||
)
|
||||
models_module.SESSIONS[session.session_id] = session
|
||||
|
||||
class Handler:
|
||||
command = "GET"
|
||||
headers = {}
|
||||
|
||||
def __init__(self):
|
||||
self.status = None
|
||||
self.headers_sent = []
|
||||
self.wfile = io.BytesIO()
|
||||
|
||||
def send_response(self, code):
|
||||
self.status = code
|
||||
|
||||
def send_header(self, key, value):
|
||||
self.headers_sent.append((key, value))
|
||||
|
||||
def end_headers(self):
|
||||
pass
|
||||
|
||||
monkeypatch.setattr(profiles_module, "get_active_profile_name", lambda: "default")
|
||||
try:
|
||||
handler = Handler()
|
||||
routes_module._handle_file_read(
|
||||
handler,
|
||||
urlparse(
|
||||
"/api/file?session_id=foreign-profile-file-read&path=marker.txt"
|
||||
),
|
||||
)
|
||||
assert handler.status == 404
|
||||
assert b"foreign profile marker" not in handler.wfile.getvalue()
|
||||
finally:
|
||||
models_module.SESSIONS.pop(session.session_id, None)
|
||||
|
||||
|
||||
def test_get_session_for_file_ops_state_db_fallback(
|
||||
models_module, monkeypatch, tmp_path
|
||||
):
|
||||
|
||||
Reference in new issue
Block a user