fix(gateway): scope /yolo to the active session

This commit is contained in:
tars
2026-04-10 16:55:51 +09:00
committed by Teknium
parent 68528068ec
commit 9a0dfb5a6d
4 changed files with 185 additions and 10 deletions

View File

@@ -0,0 +1,62 @@
"""Tests for gateway /yolo session scoping."""
import os
import pytest
import gateway.run as gateway_run
from gateway.config import Platform
from gateway.platforms.base import MessageEvent
from gateway.session import SessionSource
from tools.approval import clear_session, is_session_yolo_enabled
@pytest.fixture(autouse=True)
def _clean_yolo_state(monkeypatch):
monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
clear_session("agent:main:telegram:dm:chat-a")
clear_session("agent:main:telegram:dm:chat-b")
yield
monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
clear_session("agent:main:telegram:dm:chat-a")
clear_session("agent:main:telegram:dm:chat-b")
def _make_runner():
runner = object.__new__(gateway_run.GatewayRunner)
runner.session_store = None
runner.config = None
return runner
def _make_event(chat_id: str) -> MessageEvent:
source = SessionSource(
platform=Platform.TELEGRAM,
user_id=f"user-{chat_id}",
chat_id=chat_id,
user_name="tester",
chat_type="dm",
)
return MessageEvent(text="/yolo", source=source)
@pytest.mark.asyncio
async def test_yolo_command_toggles_only_current_session(monkeypatch):
runner = _make_runner()
event_a = _make_event("chat-a")
session_a = runner._session_key_for_source(event_a.source)
session_b = runner._session_key_for_source(_make_event("chat-b").source)
result_on = await runner._handle_yolo_command(event_a)
assert "ON" in result_on
assert is_session_yolo_enabled(session_a) is True
assert is_session_yolo_enabled(session_b) is False
assert os.environ.get("HERMES_YOLO_MODE") is None
result_off = await runner._handle_yolo_command(event_a)
assert "OFF" in result_off
assert is_session_yolo_enabled(session_a) is False
assert os.environ.get("HERMES_YOLO_MODE") is None

View File

@@ -10,6 +10,11 @@ from tools.approval import (
check_all_command_guards,
check_dangerous_command,
detect_dangerous_command,
disable_session_yolo,
enable_session_yolo,
is_session_yolo_enabled,
reset_current_session_key,
set_current_session_key,
)
@@ -18,10 +23,14 @@ def _clear_approval_state():
approval_module._permanent_approved.clear()
approval_module.clear_session("default")
approval_module.clear_session("test-session")
approval_module.clear_session("session-a")
approval_module.clear_session("session-b")
yield
approval_module._permanent_approved.clear()
approval_module.clear_session("default")
approval_module.clear_session("test-session")
approval_module.clear_session("session-a")
approval_module.clear_session("session-b")
class TestYoloMode:
@@ -108,3 +117,67 @@ class TestYoloMode:
result = check_dangerous_command("rm -rf /", "local",
approval_callback=lambda *a: "deny")
assert not result["approved"]
def test_session_scoped_yolo_only_bypasses_current_session(self, monkeypatch):
"""Gateway /yolo should only bypass approvals for the active session."""
monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
enable_session_yolo("session-a")
assert is_session_yolo_enabled("session-a") is True
assert is_session_yolo_enabled("session-b") is False
token_a = set_current_session_key("session-a")
try:
approved = check_dangerous_command("rm -rf /", "local")
assert approved["approved"] is True
finally:
reset_current_session_key(token_a)
token_b = set_current_session_key("session-b")
try:
blocked = check_dangerous_command(
"rm -rf /",
"local",
approval_callback=lambda *a: "deny",
)
assert blocked["approved"] is False
finally:
reset_current_session_key(token_b)
disable_session_yolo("session-a")
assert is_session_yolo_enabled("session-a") is False
def test_session_scoped_yolo_bypasses_combined_guard_only_for_current_session(self, monkeypatch):
"""Combined guard should honor session-scoped YOLO without affecting others."""
monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
enable_session_yolo("session-a")
token_a = set_current_session_key("session-a")
try:
approved = check_all_command_guards("rm -rf /", "local")
assert approved["approved"] is True
finally:
reset_current_session_key(token_a)
token_b = set_current_session_key("session-b")
try:
blocked = check_all_command_guards(
"rm -rf /",
"local",
approval_callback=lambda *a: "deny",
)
assert blocked["approved"] is False
finally:
reset_current_session_key(token_b)
def test_clear_session_removes_session_yolo_state(self):
"""Session cleanup must remove YOLO bypass state."""
enable_session_yolo("session-a")
assert is_session_yolo_enabled("session-a") is True
approval_module.clear_session("session-a")
assert is_session_yolo_enabled("session-a") is False