fix: revoke local shell sessions when permissions change

This commit is contained in:
RuochenPan
2026-09-08 15:44:02 +08:00
parent 390c213fe3
commit 721cd89695
6 changed files with 315 additions and 145 deletions
+176 -130
View File
@@ -10,6 +10,7 @@ import subprocess
import sys
import time
import uuid
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, cast
@@ -144,6 +145,7 @@ class _LocalShellSession:
output_event: asyncio.Event
reader_task: asyncio.Task[None]
wait_task: asyncio.Task[int]
permission_check: Callable[[], bool] | None = None
timeout_task: asyncio.Task[None] | None = None
cursor: int = 0
timed_out: bool = False
@@ -263,6 +265,7 @@ class LocalShellComponent(ShellComponent):
creator_id: str,
creator_is_admin: bool,
sandboxed: bool,
permission_check: Callable[[], bool],
allow_network: bool = False,
filesystem_scope: str = "workspace",
readable_roots: tuple[Path, ...] = (),
@@ -281,6 +284,7 @@ class LocalShellComponent(ShellComponent):
creator_id: Sender ID that created the session.
creator_is_admin: Whether the creator was an administrator.
sandboxed: Whether the process is isolated from the host.
permission_check: Check that the creation permissions still apply.
allow_network: Whether an isolated process may access the network.
filesystem_scope: Filesystem scope applied to an isolated process.
readable_roots: Additional directories readable by an isolated process.
@@ -295,7 +299,7 @@ class LocalShellComponent(ShellComponent):
Process result with output, status, and session metadata.
Raises:
PermissionError: If the command matches a blocked pattern.
PermissionError: If the command is blocked or its permissions changed.
RuntimeError: If the requested platform sandbox is unavailable.
ValueError: If a timing or output limit is invalid.
"""
@@ -316,132 +320,143 @@ class LocalShellComponent(ShellComponent):
output_path = output_dir / f"{session_id}.log"
output_path.touch()
try:
if sandboxed:
process = await create_process_sandbox().spawn_shell(
command,
SandboxSpec(
workspace=working_dir,
allow_network=allow_network,
filesystem_scope=filesystem_scope,
readable_roots=readable_roots,
writable_roots=writable_roots,
),
env={str(k): str(v) for k, v in (env or {}).items()},
# Configuration invalidation must also see processes still being spawned.
async with self._sessions_lock:
if not permission_check():
output_path.unlink(missing_ok=True)
raise PermissionError(
"Local shell permissions changed; retry the command."
)
else:
run_env = os.environ.copy()
if env:
run_env.update({str(k): str(v) for k, v in env.items()})
process_kwargs: dict[str, Any] = {}
if sys.platform == "win32":
# Keep managed-session Python output UTF-8.
run_env.setdefault("PYTHONIOENCODING", "utf-8")
process_factory = asyncio.create_subprocess_exec
shell_executable = resolve_windows_shell()
process_args = (
shell_executable,
"-NoLogo",
"-NoProfile",
"-NonInteractive",
"-Command",
try:
if sandboxed:
process = await create_process_sandbox().spawn_shell(
command,
)
process_kwargs["creationflags"] = getattr(
subprocess,
"CREATE_NEW_PROCESS_GROUP",
0,
SandboxSpec(
workspace=working_dir,
allow_network=allow_network,
filesystem_scope=filesystem_scope,
readable_roots=readable_roots,
writable_roots=writable_roots,
),
env={str(k): str(v) for k, v in (env or {}).items()},
)
else:
process_factory = asyncio.create_subprocess_shell
process_args = (command,)
process_kwargs["start_new_session"] = True
process = await process_factory(
*process_args,
cwd=working_dir,
env=run_env,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
**process_kwargs,
)
except Exception:
output_path.unlink(missing_ok=True)
raise
run_env = os.environ.copy()
if env:
run_env.update({str(k): str(v) for k, v in env.items()})
process_kwargs: dict[str, Any] = {}
if sys.platform == "win32":
# Keep managed-session Python output UTF-8.
run_env.setdefault("PYTHONIOENCODING", "utf-8")
process_factory = asyncio.create_subprocess_exec
shell_executable = resolve_windows_shell()
process_args = (
shell_executable,
"-NoLogo",
"-NoProfile",
"-NonInteractive",
"-Command",
command,
)
process_kwargs["creationflags"] = getattr(
subprocess,
"CREATE_NEW_PROCESS_GROUP",
0,
)
else:
process_factory = asyncio.create_subprocess_shell
process_args = (command,)
process_kwargs["start_new_session"] = True
process = await process_factory(
*process_args,
cwd=working_dir,
env=run_env,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
**process_kwargs,
)
except Exception:
output_path.unlink(missing_ok=True)
raise
output_event = asyncio.Event()
output_event = asyncio.Event()
async def _capture_output() -> None:
if process.stdout is None:
return
output_size = 0
with output_path.open("ab") as output_file:
while chunk := await process.stdout.read(8192):
if sandboxed:
remaining = _LOCAL_SANDBOX_MAX_OUTPUT_BYTES - output_size
if remaining <= 0:
session.output_limited = True
async def _capture_output() -> None:
if process.stdout is None:
return
output_size = 0
with output_path.open("ab") as output_file:
while chunk := await process.stdout.read(8192):
if sandboxed:
remaining = _LOCAL_SANDBOX_MAX_OUTPUT_BYTES - output_size
if remaining <= 0:
session.output_limited = True
process.terminate()
return
if len(chunk) > remaining:
chunk = chunk[:remaining]
session.output_limited = True
output_file.write(chunk)
output_file.flush()
output_size += len(chunk)
output_event.set()
if session.output_limited:
process.terminate()
return
if len(chunk) > remaining:
chunk = chunk[:remaining]
session.output_limited = True
output_file.write(chunk)
output_file.flush()
output_size += len(chunk)
output_event.set()
if session.output_limited:
process.terminate()
return
reader_task = asyncio.create_task(
_capture_output(),
name=f"local_shell_output_{session_id}",
)
wait_task = asyncio.create_task(
process.wait(),
name=f"local_shell_wait_{session_id}",
)
wait_task.add_done_callback(lambda _: output_event.set())
session = _LocalShellSession(
session_id=session_id,
owner_id=owner_id,
creator_id=creator_id,
creator_is_admin=creator_is_admin,
sandboxed=sandboxed,
process=process,
output_path=output_path,
started_at=time.time(),
output_event=output_event,
reader_task=reader_task,
wait_task=wait_task,
)
if timeout is not None:
async def _enforce_timeout() -> None:
try:
await asyncio.wait_for(
asyncio.shield(wait_task),
timeout=timeout,
)
except asyncio.TimeoutError:
session.timed_out = True
logger.warning(
"Managed local shell session timed out: session_id=%s pid=%s",
session_id,
process.pid,
)
await self._terminate_process(session)
session.timeout_task = asyncio.create_task(
_enforce_timeout(),
name=f"local_shell_timeout_{session_id}",
reader_task = asyncio.create_task(
_capture_output(),
name=f"local_shell_output_{session_id}",
)
wait_task = asyncio.create_task(
process.wait(),
name=f"local_shell_wait_{session_id}",
)
wait_task.add_done_callback(lambda _: output_event.set())
session = _LocalShellSession(
session_id=session_id,
owner_id=owner_id,
creator_id=creator_id,
creator_is_admin=creator_is_admin,
sandboxed=sandboxed,
process=process,
output_path=output_path,
started_at=time.time(),
output_event=output_event,
reader_task=reader_task,
wait_task=wait_task,
permission_check=permission_check,
)
async with self._sessions_lock:
if timeout is not None:
async def _enforce_timeout() -> None:
try:
await asyncio.wait_for(
asyncio.shield(wait_task),
timeout=timeout,
)
except asyncio.TimeoutError:
session.timed_out = True
logger.warning(
"Managed local shell session timed out: session_id=%s pid=%s",
session_id,
process.pid,
)
await self._terminate_process(session)
session.timeout_task = asyncio.create_task(
_enforce_timeout(),
name=f"local_shell_timeout_{session_id}",
)
self._sessions[session_id] = session
if not permission_check():
await self.shutdown_sessions(invalid_only=True)
raise PermissionError("Local shell permissions changed; retry the command.")
if yield_time_ms > 0:
try:
await asyncio.wait_for(
@@ -688,7 +703,11 @@ class LocalShellComponent(ShellComponent):
requester_is_admin,
session_id,
)
if session.process.returncode is not None or session.process.stdin is None:
if (
session.terminated
or session.process.returncode is not None
or session.process.stdin is None
):
raise ValueError(f"Shell session {session_id} is not accepting input.")
session.process.stdin.write(chars.encode("utf-8"))
await session.process.stdin.drain()
@@ -787,12 +806,22 @@ class LocalShellComponent(ShellComponent):
max_output_chars=max_output_chars,
)
async def shutdown_sessions(self) -> None:
"""Terminate and remove every managed local shell session."""
async def shutdown_sessions(self, *, invalid_only: bool = False) -> None:
"""Terminate and remove managed local shell sessions.
Args:
invalid_only: Keep sessions whose creation permissions still apply.
"""
async with self._sessions_lock:
sessions = list(self._sessions.values())
for session in sessions:
session.terminated = True
sessions = [
session
for session in self._sessions.values()
if not invalid_only
or getattr(session, "permission_check", None) is None
or not session.permission_check()
]
for session in sessions:
session.terminated = True
termination_results = await asyncio.gather(
*(self._terminate_process(session) for session in sessions),
return_exceptions=True,
@@ -830,7 +859,7 @@ class LocalShellComponent(ShellComponent):
Matching managed shell session.
Raises:
ValueError: If the session does not exist for this owner.
ValueError: If the session is unavailable or its permissions changed.
"""
async with self._sessions_lock:
session = self._sessions.get(session_id)
@@ -842,7 +871,19 @@ class LocalShellComponent(ShellComponent):
and (session.creator_is_admin or session.creator_id != requester_id)
)
):
raise ValueError(f"Shell session {session_id} was not found.")
raise ValueError(
f"Shell session {session_id} was not found or has expired. "
"Start a new shell session."
)
if (
getattr(session, "permission_check", None) is None
or not session.permission_check()
):
await self.shutdown_sessions(invalid_only=True)
raise ValueError(
f"Shell session {session_id} expired after a permission change. "
"Start a new shell session."
)
return session
async def _terminate_process(self, session: _LocalShellSession) -> None:
@@ -851,7 +892,7 @@ class LocalShellComponent(ShellComponent):
Args:
session: Managed shell session to terminate.
"""
if session.process.returncode is not None:
if os.name == "nt" and session.process.returncode is not None:
return
if session.sandboxed:
session.process.terminate()
@@ -881,14 +922,19 @@ class LocalShellComponent(ShellComponent):
timeout=5,
)
except asyncio.TimeoutError:
if session.sandboxed or os.name == "nt":
pass
# The leader may have exited while children remain in its process group.
if session.sandboxed:
session.process.kill()
elif os.name == "nt":
if session.process.returncode is None:
session.process.kill()
else:
try:
os.killpg(session.process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
await session.wait_task
else:
try:
os.killpg(session.process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
await session.wait_task
async def _remove_session(self, session: _LocalShellSession) -> None:
"""Remove a completed session and its temporary output file.
+9 -7
View File
@@ -12,13 +12,14 @@ from astrbot.core.agent.run_context import ContextWrapper
from astrbot.core.agent.tool import ToolExecResult
from astrbot.core.astr_agent_context import AstrAgentContext
from astrbot.core.computer.booters.local import LocalShellComponent
from astrbot.core.computer.computer_client import get_booter
from astrbot.core.computer.computer_client import get_booter, get_local_booter
from astrbot.core.utils.astrbot_path import get_astrbot_system_tmp_path
from ..registry import builtin_tool
from .fs import _read_allowed_roots, _write_allowed_roots
from .util import (
check_local_execution_permission,
get_local_permission_policy,
is_local_runtime,
workspace_root_for_context,
)
@@ -150,6 +151,10 @@ class ExecuteShellTool(FunctionTool):
creator_id=creator_id,
creator_is_admin=context.context.event.role == "admin",
sandboxed=sandboxed,
permission_check=lambda: (
is_local_runtime(context)
and get_local_permission_policy(context) == local_policy
),
allow_network=(
local_policy.allow_network if local_policy else True
),
@@ -368,16 +373,13 @@ class ShellSessionTool(FunctionTool):
context,
"Shell session management",
)
if permission_error:
if permission_error and action != "terminate":
return permission_error
if not is_local_runtime(context):
if not is_local_runtime(context) and action != "terminate":
return "Error managing shell session: only local runtime is supported."
try:
sb = await get_booter(
context.context.context,
context.context.event.unified_msg_origin,
)
sb = get_local_booter()
if not isinstance(sb.shell, LocalShellComponent):
return "Error managing shell session: local shell component is unavailable."
@@ -11,6 +11,8 @@ from typing import Any
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from astrbot.core import file_token_service, logger
from astrbot.core.computer import computer_client
from astrbot.core.computer.booters.local import LocalShellComponent
from astrbot.core.config.agent_runner import normalize_agent_runner
from astrbot.core.config.astrbot_config import AstrBotConfig
from astrbot.core.config.default import (
@@ -802,6 +804,9 @@ class ConfigProfileService:
save_config(
config, self.acm.confs[config_id], is_core=True, runtime=self.runtime
)
booter = computer_client.local_booter
if booter is not None and isinstance(booter.shell, LocalShellComponent):
await booter.shell.shutdown_sessions(invalid_only=True)
if protected_2fa_changed and self.db is not None:
await revoke_user_trusted_devices(self.db)
await self.core_lifecycle.reload_pipeline_scheduler(config_id)
+77
View File
@@ -1720,6 +1720,83 @@ async def test_v1_system_config_update_preserves_independent_bot_provider_sectio
assert fake_core_lifecycle.reloaded_config_ids == ["default"]
@pytest.mark.asyncio
async def test_config_update_revokes_only_affected_shell_sessions(
asgi_app, fake_core_lifecycle, monkeypatch, tmp_path
):
from astrbot.core.computer import computer_client
from astrbot.core.computer.booters.local import LocalBooter
from astrbot.core.tools.computer_tools import shell as shell_tools
config = fake_core_lifecycle.astrbot_config
config["agent_runner"] = {"runner_type": "local"}
config["provider_settings"] = {
"computer_use_runtime": "local",
"computer_use_local_permissions": {
role: {
"allow_execution": True,
"allow_network": True,
"filesystem_scope": "host",
}
for role in ("admin", "member")
},
}
other_config = copy.deepcopy(config)
booter = LocalBooter()
monkeypatch.setattr(computer_client, "local_booter", booter)
monkeypatch.setattr(
shell_tools, "workspace_root_for_context", AsyncMock(return_value=tmp_path)
)
sessions = []
try:
for role, profile in (
("member", config),
("admin", config),
("member", other_config),
):
context = SimpleNamespace(
context=SimpleNamespace(
context=SimpleNamespace(get_config=lambda umo, p=profile: p),
event=SimpleNamespace(
role=role,
unified_msg_origin="test:friend:revocation",
get_sender_id=lambda: "creator",
),
)
)
result = json.loads(
await shell_tools.LocalExecuteShellTool().call(
context, command='python -u -c "input()"', yield_time_ms=0
)
)
sessions.append(booter.shell._sessions[result["session_id"]])
service = asgi_app.state.services.config_profiles
payload = copy.deepcopy(config)
payload["wake_prefix"] = ["changed"]
await service.update_profile("default", payload)
assert all(session.process.returncode is None for session in sessions)
payload = copy.deepcopy(config)
payload["provider_settings"]["computer_use_local_permissions"]["member"][
"allow_execution"
] = False
await service.update_profile("default", payload)
assert sessions[0].process.returncode is not None
assert all(session.process.returncode is None for session in sessions[1:])
assert sessions[0].session_id not in booter.shell._sessions
# Pre-upgrade sessions must not acquire a grant from the current config.
sessions[2].permission_check = None
payload = copy.deepcopy(config)
payload["provider_settings"]["computer_use_runtime"] = "none"
await service.update_profile("default", payload)
assert all(session.process.returncode is not None for session in sessions)
assert not booter.shell._sessions
finally:
await booter.shell.shutdown_sessions()
@pytest.mark.asyncio
@pytest.mark.parametrize("scope", ["workspace", "host"])
@pytest.mark.parametrize(
+44
View File
@@ -245,6 +245,7 @@ async def test_managed_shell_uses_windows_powershell(monkeypatch, tmp_path):
creator_id="user-a",
creator_is_admin=False,
sandboxed=False,
permission_check=lambda: True,
cwd=str(tmp_path),
yield_time_ms=5_000,
)
@@ -844,6 +845,7 @@ async def test_managed_shell_uses_platform_sandbox(
cwd=str(tmp_path),
yield_time_ms=5_000,
sandboxed=True,
permission_check=lambda: True,
)
assert result["status"] == "completed"
@@ -896,6 +898,7 @@ async def test_sandboxed_managed_shell_delegates_launch(monkeypatch, tmp_path):
cwd=str(tmp_path),
yield_time_ms=5_000,
sandboxed=True,
permission_check=lambda: True,
)
assert result["status"] == "completed"
@@ -954,6 +957,7 @@ async def test_sandboxed_managed_shell_stops_at_output_limit(monkeypatch, tmp_pa
cwd=str(tmp_path),
yield_time_ms=5_000,
sandboxed=True,
permission_check=lambda: True,
)
assert result["status"] == "output_limited"
@@ -1103,6 +1107,7 @@ async def test_managed_shell_prefers_pwsh_when_available(monkeypatch, tmp_path):
creator_id="user-a",
creator_is_admin=False,
sandboxed=False,
permission_check=lambda: True,
cwd=str(tmp_path),
yield_time_ms=5_000,
)
@@ -1235,6 +1240,39 @@ def test_local_shell_component_falls_back_when_windows_taskkill_fails(monkeypatc
assert proc.wait_timeout == 5
@pytest.mark.asyncio
@pytest.mark.skipif(os.name == "nt", reason="Unix process-group cleanup.")
async def test_managed_shell_cleans_children_after_leader_exits():
shell = LocalShellComponent()
result = await shell.exec_managed(
"(trap '' TERM; printf 'ready\\n'; sleep 30) & read line",
owner_id="owner",
creator_id="user",
creator_is_admin=False,
sandboxed=False,
permission_check=lambda: True,
yield_time_ms=0,
)
session = shell._sessions[result["session_id"]]
try:
await asyncio.wait_for(session.output_event.wait(), 5)
session.process.stdin.write(b"\n")
await session.process.stdin.drain()
for _ in range(500):
if session.process.returncode is not None:
break
await asyncio.sleep(0.01)
assert session.process.returncode is not None
# The child keeps stdout open and ignores SIGTERM after its parent exits.
assert not session.reader_task.done()
session.permission_check = None
await asyncio.wait_for(shell.shutdown_sessions(invalid_only=True), 8)
assert session.reader_task.done()
assert not shell._sessions
finally:
await shell.shutdown_sessions()
@pytest.mark.asyncio
async def test_managed_shell_returns_completed_output_without_open_session():
shell = LocalShellComponent()
@@ -1245,6 +1283,7 @@ async def test_managed_shell_returns_completed_output_without_open_session():
creator_id="user-a",
creator_is_admin=False,
sandboxed=False,
permission_check=lambda: True,
yield_time_ms=5_000,
)
@@ -1268,6 +1307,7 @@ async def test_managed_shell_allows_creator_and_conversation_admin():
creator_id="user-a",
creator_is_admin=False,
sandboxed=False,
permission_check=lambda: True,
# Cold-starting PowerShell + Python on CI runners exceeds 200ms.
yield_time_ms=5_000,
)
@@ -1329,6 +1369,7 @@ async def test_managed_shell_rejects_cross_user_session_access():
creator_id="admin-user",
creator_is_admin=True,
sandboxed=False,
permission_check=lambda: True,
yield_time_ms=100,
)
@@ -1418,6 +1459,7 @@ async def test_managed_shell_accepts_stdin_and_polls_incremental_output():
creator_id="user-a",
creator_is_admin=False,
sandboxed=False,
permission_check=lambda: True,
yield_time_ms=100,
)
@@ -1464,6 +1506,7 @@ async def test_managed_shell_hard_timeout_terminates_session():
creator_id="user-a",
creator_is_admin=False,
sandboxed=False,
permission_check=lambda: True,
timeout=1,
yield_time_ms=0,
)
@@ -1493,6 +1536,7 @@ async def test_managed_shell_keeps_completed_session_until_output_is_drained():
creator_id="user-a",
creator_is_admin=False,
sandboxed=False,
permission_check=lambda: True,
yield_time_ms=5_000,
max_output_chars=10_000,
)
+4 -8
View File
@@ -183,6 +183,7 @@ async def test_local_execute_shell_manages_running_and_closed_results(
creator_id="admin-user",
creator_is_admin=True,
sandboxed=True,
permission_check=ANY,
allow_network=True,
filesystem_scope="workspace",
readable_roots=ANY,
@@ -357,6 +358,7 @@ async def test_local_member_shell_uses_sandbox_backend(
timeout=300,
yield_time_ms=250,
sandboxed=True,
permission_check=ANY,
allow_network=False,
filesystem_scope="workspace",
readable_roots=ANY,
@@ -441,10 +443,7 @@ async def test_shell_session_tool_lists_sessions_for_current_owner(monkeypatch):
class FakeWrapper:
context = FakeAstrContext()
async def fake_get_booter(context, session_id):
return booter
monkeypatch.setattr(shell_tools, "get_booter", fake_get_booter)
monkeypatch.setattr(shell_tools, "get_local_booter", lambda: booter)
result = await ShellSessionTool().call(FakeWrapper(), action="list")
@@ -509,10 +508,7 @@ async def test_shell_session_tool_passes_member_identity_to_session_actions(
class FakeWrapper:
context = FakeAstrContext()
async def fake_get_booter(context, session_id):
return booter
monkeypatch.setattr(shell_tools, "get_booter", fake_get_booter)
monkeypatch.setattr(shell_tools, "get_local_booter", lambda: booter)
result = await ShellSessionTool().call(
FakeWrapper(),