From f0f477285d354612bfb9a37cd90466f30c67d2c8 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sat, 15 Aug 2026 20:52:35 +0800 Subject: [PATCH 1/3] feat: migrate shell safety to plugin api v3 --- .github/workflows/plugin-api-v2.yml | 28 ---- .github/workflows/plugin-api-v3.yml | 51 ++++++++ plugin.py | 195 +++++++++++++++------------- tests/test_plugin.py | 167 +++++++++++++++++++++++- 4 files changed, 321 insertions(+), 120 deletions(-) delete mode 100644 .github/workflows/plugin-api-v2.yml create mode 100644 .github/workflows/plugin-api-v3.yml diff --git a/.github/workflows/plugin-api-v2.yml b/.github/workflows/plugin-api-v2.yml deleted file mode 100644 index 7c1876b..0000000 --- a/.github/workflows/plugin-api-v2.yml +++ /dev/null @@ -1,28 +0,0 @@ -name: plugin-api-v2 - -on: - pull_request: - push: - branches: - - main - -permissions: - contents: read - -jobs: - contract: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - uses: actions/checkout@v4 - with: - repository: akashic-plugins/plugin-contracts - ref: 24543445c7b99ca63fcd90b5828f754a148b184c - path: .plugin-contracts - - uses: actions/setup-python@v5 - with: - python-version: "3.13" - - name: Check Plugin API v2 - env: - PYTHONPATH: .plugin-contracts - run: python -m akashic_plugin_contracts check plugin.py diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml new file mode 100644 index 0000000..3fe4ca6 --- /dev/null +++ b/.github/workflows/plugin-api-v3.yml @@ -0,0 +1,51 @@ +name: plugin-api-v3 + +on: + pull_request: + push: + branches: + - main + +permissions: + contents: read + +jobs: + contract: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/checkout@v4 + with: + repository: akashic-plugins/plugin-contracts + ref: 4dd69dd621e029e51e99aa428443fa3a4ec1f6cf + path: .plugin-contracts + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - name: Check Plugin API v3 + env: + PYTHONPATH: .plugin-contracts + run: python -m akashic_plugin_contracts check plugin.py + + composition-parity: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v4 + - uses: actions/checkout@v4 + with: + repository: kachofugetsu09/akashic-agent + ref: b60a7ed1bcbed1f772e041336d5c858b4b9fac90 + path: .akashic-core + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + cache-dependency-path: .akashic-core/requirements.txt + - name: Install pinned Core dependencies + run: python -m pip install -r .akashic-core/requirements.txt pytest pytest-asyncio + - name: Verify Shell Safety v3 behavior + env: + AKASHIC_AGENT_ROOT: .akashic-core + PYTHONPATH: .akashic-core + run: python -m pytest -q tests/ diff --git a/plugin.py b/plugin.py index 85b99cc..7db9aa1 100644 --- a/plugin.py +++ b/plugin.py @@ -3,9 +3,8 @@ import shlex from pathlib import Path -from agent.lifecycle.types import PreToolCtx -from agent.plugins import Plugin, on_tool_pre -from agent.tool_hooks import HookOutcome +from agent.plugin_composition import Bail, Context +from agent.tools.events import TOOL_EXECUTION_AUTHORIZE, ToolInput INTERACTIVE_COMMANDS = { "vi", @@ -24,96 +23,116 @@ "--sysupgrade", } +api_version = 3 +name = "shell_safety" +version = "2.0.0" +desc = "阻止 shell 工具执行容易卡住的交互式命令" +author = "Akashic" +inject: tuple[()] = () -class ShellSafety(Plugin): - api_version = 2 - name = "shell_safety" - version = "1.0.0" - desc = "阻止 shell 工具执行容易卡住的交互式命令" - @on_tool_pre(tool_name="shell") - async def block_interactive_shell(self, event: PreToolCtx) -> HookOutcome | None: - command = str(event.arguments.get("command") or "").strip() - if not command: +async def apply(ctx: Context, config: object) -> None: + """Register final-argument shell authorization without owning execution.""" + + _ = config + + def authorize(tool_input: ToolInput) -> Bail[str] | None: + if tool_input.tool_name != "shell": return None - reason = self._deny_reason(command) - if not reason: + command = str(tool_input.arguments.get("command") or "").strip() + if not command: return None - return HookOutcome(decision="deny", reason=reason) - - def _deny_reason(self, command: str) -> str: - try: - tokens = shlex.split(command, posix=True) - except ValueError: - return "" - if not tokens: - return "" - editor = self._find_interactive_command(tokens) - if editor: - return f"shell_safety 拦截：{editor} 会打开交互式界面，请改用非交互命令。" - if self._sudo_needs_password(tokens): - return "shell_safety 拦截：sudo 可能等待密码，请改用 sudo -n，让它在没有缓存时立即失败。" - package_manager = self._find_interactive_package_command(tokens) - if package_manager: - return f"shell_safety 拦截：{package_manager} 写操作需要加 --noconfirm，避免卡在确认提示。" - if self._opens_system_editor(tokens): - return "shell_safety 拦截：该命令会打开系统编辑器，请改用写文件或非交互参数。" - return "" + reason = deny_reason(command) + return Bail(reason) if reason else None + + _ = await ctx.on(TOOL_EXECUTION_AUTHORIZE, authorize) - def _find_interactive_command(self, tokens: list[str]) -> str: - for token in tokens: - name = Path(token).name - if name in INTERACTIVE_COMMANDS: - return name - return "" - def _sudo_needs_password(self, tokens: list[str]) -> bool: - for index, token in enumerate(tokens): - if Path(token).name != "sudo": - continue - if not self._sudo_has_non_interactive_option(tokens[index + 1 :]): - return True - return False - - def _sudo_has_non_interactive_option(self, tokens: list[str]) -> bool: - index = 0 - while index < len(tokens): - token = tokens[index] - if token == "--": - return False - if not token.startswith("-") or token == "-": - return False - if token == "-n" or (token.startswith("-") and not token.startswith("--") and "n" in token[1:]): - return True - if token in {"-u", "-g", "-p", "-C", "-D", "-R", "-T", "-h"}: - index += 2 - continue - index += 1 - return False - - def _find_interactive_package_command(self, tokens: list[str]) -> str: - for index, token in enumerate(tokens): - name = Path(token).name - if name not in PACKAGE_MANAGERS: - continue - args = tokens[index + 1 :] - if self._has_package_write_option(args) and "--noconfirm" not in args: - return name +def deny_reason(command: str) -> str: + try: + tokens = shlex.split(command, posix=True) + except ValueError: + return "" + if not tokens: return "" + editor = _find_interactive_command(tokens) + if editor: + return f"shell_safety 拦截：{editor} 会打开交互式界面，请改用非交互命令。" + if _sudo_needs_password(tokens): + return "shell_safety 拦截：sudo 可能等待密码，请改用 sudo -n，让它在没有缓存时立即失败。" + package_manager = _find_interactive_package_command(tokens) + if package_manager: + return f"shell_safety 拦截：{package_manager} 写操作需要加 --noconfirm，避免卡在确认提示。" + if _opens_system_editor(tokens): + return "shell_safety 拦截：该命令会打开系统编辑器，请改用写文件或非交互参数。" + return "" + + +def _find_interactive_command(tokens: list[str]) -> str: + for token in tokens: + candidate = Path(token).name + if candidate in INTERACTIVE_COMMANDS: + return candidate + return "" + + +def _sudo_needs_password(tokens: list[str]) -> bool: + for index, token in enumerate(tokens): + if Path(token).name != "sudo": + continue + if not _sudo_has_non_interactive_option(tokens[index + 1 :]): + return True + return False + + +def _sudo_has_non_interactive_option(tokens: list[str]) -> bool: + index = 0 + while index < len(tokens): + token = tokens[index] + if token == "--": + return False + if not token.startswith("-") or token == "-": + return False + if token == "-n" or ( + token.startswith("-") + and not token.startswith("--") + and "n" in token[1:] + ): + return True + if token in {"-u", "-g", "-p", "-C", "-D", "-R", "-T", "-h"}: + index += 2 + continue + index += 1 + return False + + +def _find_interactive_package_command(tokens: list[str]) -> str: + for index, token in enumerate(tokens): + candidate = Path(token).name + if candidate not in PACKAGE_MANAGERS: + continue + arguments = tokens[index + 1 :] + if _has_package_write_option(arguments) and "--noconfirm" not in arguments: + return candidate + return "" + + +def _has_package_write_option(arguments: list[str]) -> bool: + for argument in arguments: + if argument in PACKAGE_WRITE_OPTIONS: + return True + if argument.startswith("-S") or argument.startswith("-R"): + return True + if argument.startswith("-U"): + return True + return False + - def _has_package_write_option(self, args: list[str]) -> bool: - for arg in args: - if arg in PACKAGE_WRITE_OPTIONS: - return True - if arg.startswith("-S") or arg.startswith("-R") or arg.startswith("-U"): - return True - return False - - def _opens_system_editor(self, tokens: list[str]) -> bool: - for index, token in enumerate(tokens[:-1]): - name = Path(token).name - if name == "systemctl" and tokens[index + 1] == "edit": - return True - if name == "crontab" and tokens[index + 1] == "-e": - return True - return False +def _opens_system_editor(tokens: list[str]) -> bool: + for index, token in enumerate(tokens[:-1]): + candidate = Path(token).name + if candidate == "systemctl" and tokens[index + 1] == "edit": + return True + if candidate == "crontab" and tokens[index + 1] == "-e": + return True + return False diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 1b38c81..3a412df 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -1,18 +1,177 @@ from __future__ import annotations -from plugin import ShellSafety +import shutil +from pathlib import Path +from typing import Any + +import pytest + +import plugin as shell_safety +from agent.plugin_composition import CompositionRoot, PluginRuntime +from agent.plugins.composable import ComposablePlugin +from agent.plugins.manager import PluginManager +from agent.plugins.snapshot import ( + RuntimeSnapshotCompiler, + RuntimeSnapshotStore, + bind_runtime_snapshot, + reset_runtime_snapshot, +) +from agent.tool_hooks.executor import ToolExecutor +from agent.tool_hooks.types import ToolExecutionRequest +from bus.event_bus import EventBus + + +def test_v3_namespace_is_loadable() -> None: + loaded = ComposablePlugin.from_module(shell_safety) + + assert loaded.name == "shell_safety" + assert loaded.version == "2.0.0" + assert loaded.inject == () def test_blocks_sudo_without_non_interactive() -> None: - reason = ShellSafety()._deny_reason("sudo pacman -Syu --noconfirm") + reason = shell_safety.deny_reason("sudo pacman -Syu --noconfirm") assert "sudo -n" in reason def test_blocks_interactive_editor() -> None: - reason = ShellSafety()._deny_reason("sudo -n vim /etc/example.service") + reason = shell_safety.deny_reason("sudo -n vim /etc/example.service") assert "vim" in reason +def test_blocks_package_write_without_noconfirm() -> None: + reason = shell_safety.deny_reason("pacman -Syu package") + assert "--noconfirm" in reason + + +def test_blocks_system_editor() -> None: + assert "系统编辑器" in shell_safety.deny_reason("systemctl edit sshd") + assert "系统编辑器" in shell_safety.deny_reason("crontab -e") + + def test_allows_non_interactive_write() -> None: - reason = ShellSafety()._deny_reason("sudo -n pacman -Syu --noconfirm") + reason = shell_safety.deny_reason("sudo -n pacman -Syu --noconfirm") assert reason == "" + + +def test_malformed_shell_is_left_to_shell_boundary() -> None: + assert shell_safety.deny_reason("sudo '") == "" + + +@pytest.mark.asyncio +async def test_authorizer_denies_without_invoking(tmp_path: Path) -> None: + root = CompositionRoot("shell-safety-direct") + _ = await root.mount( + lambda ctx: shell_safety.apply(ctx, {}), + name="shell_safety", + runtime=PluginRuntime( + plugin_id="shell_safety", + plugin_dir=tmp_path / "plugin", + data_dir=tmp_path / "plugin-data" / "shell_safety", + workspace=tmp_path / "workspace", + config={}, + ), + ) + invoked: list[str] = [] + + async def invoke(tool_name: str, _: dict[str, Any]) -> str: + invoked.append(tool_name) + return "ok" + + store = RuntimeSnapshotStore() + store.install(RuntimeSnapshotCompiler().compile({}, composition_root=root)) + lease = store.lease() + token = bind_runtime_snapshot(lease) + try: + result = await ToolExecutor().execute( + ToolExecutionRequest( + call_id="direct-deny", + tool_name="shell", + arguments={"command": "sudo pacman -Syu --noconfirm"}, + source="passive", + ), + invoke, + ) + unrelated = await ToolExecutor().execute( + ToolExecutionRequest( + call_id="direct-unrelated", + tool_name="dummy", + arguments={"command": "sudo pacman -Syu"}, + source="passive", + ), + invoke, + ) + finally: + reset_runtime_snapshot(token) + await lease.release() + await store.close() + + assert result.status == "denied" + assert "sudo -n" in str(result.output) + assert unrelated.status == "success" + assert invoked == ["dummy"] + await root.dispose() + assert root.topology_view().listeners == () + + +@pytest.mark.asyncio +async def test_manager_snapshot_authorizes_final_arguments(tmp_path: Path) -> None: + plugin_home = tmp_path / "plugins" + plugin_home.mkdir() + _ = shutil.copytree( + Path(__file__).parents[1], + plugin_home / "shell_safety", + ignore=shutil.ignore_patterns( + ".git", + ".pytest_cache", + "__pycache__", + ), + ) + manager = PluginManager( + plugin_dirs=[plugin_home], + event_bus=EventBus(), + tool_registry=None, + workspace=tmp_path / "workspace", + installed_cache_root=tmp_path / "plugin-home" / "cache", + ) + await manager.load_all() + generation = manager.generation("shell_safety") + snapshot = manager.current_snapshot + assert generation is not None and snapshot is not None + assert isinstance(generation.instance, ComposablePlugin) + assert snapshot.composition_topology is not None + assert snapshot.composition_topology.listeners == ( + "serial:tool.execution.authorize" + "[bail=akashic.tool-deny-reason.v1]:shell_safety", + ) + root = snapshot.composition_root + assert root is not None + invoked = False + + async def invoke(_: str, __: dict[str, Any]) -> str: + nonlocal invoked + invoked = True + return "unreachable" + + lease = manager._snapshot_store.lease() + token = bind_runtime_snapshot(lease) + try: + result = await ToolExecutor().execute( + ToolExecutionRequest( + call_id="manager-deny", + tool_name="shell", + arguments={"command": "pacman -Syu package"}, + source="passive", + ), + invoke, + ) + finally: + reset_runtime_snapshot(token) + await lease.release() + + assert result.status == "denied" + assert "--noconfirm" in str(result.output) + assert invoked is False + await manager.terminate_all() + assert root.topology_view().listeners == () + assert root.receipt().effects == () From 68fc1385ed2ace64cd772445b4178cc5969c9ed0 Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sat, 15 Aug 2026 21:08:50 +0800 Subject: [PATCH 2/3] fix: reject sudo mode bypasses --- plugin.py | 98 ++++++++++++++++++++++++++++++++++++-------- tests/test_plugin.py | 23 +++++++++++ 2 files changed, 104 insertions(+), 17 deletions(-) diff --git a/plugin.py b/plugin.py index 7db9aa1..b90bc89 100644 --- a/plugin.py +++ b/plugin.py @@ -22,6 +22,33 @@ "--upgrade", "--sysupgrade", } +_SUDO_SHORT_OPTIONS_WITH_VALUE = frozenset( + {"u", "g", "p", "C", "D", "R", "T", "h"} +) +_SUDO_LONG_OPTIONS_WITH_VALUE = frozenset( + { + "--user", + "--group", + "--prompt", + "--close-from", + "--chdir", + "--chroot", + "--command-timeout", + "--host", + } +) +_SUDO_MODE_SHORT_FLAGS = frozenset({"e", "l", "s", "i", "v", "h", "V"}) +_SUDO_MODE_LONG_FLAGS = frozenset( + { + "--edit", + "--list", + "--shell", + "--login", + "--validate", + "--help", + "--version", + } +) api_version = 3 name = "shell_safety" @@ -58,8 +85,9 @@ def deny_reason(command: str) -> str: editor = _find_interactive_command(tokens) if editor: return f"shell_safety 拦截：{editor} 会打开交互式界面，请改用非交互命令。" - if _sudo_needs_password(tokens): - return "shell_safety 拦截：sudo 可能等待密码，请改用 sudo -n，让它在没有缓存时立即失败。" + sudo_issue = _sudo_issue(tokens) + if sudo_issue: + return sudo_issue package_manager = _find_interactive_package_command(tokens) if package_manager: return f"shell_safety 拦截：{package_manager} 写操作需要加 --noconfirm，避免卡在确认提示。" @@ -76,34 +104,55 @@ def _find_interactive_command(tokens: list[str]) -> str: return "" -def _sudo_needs_password(tokens: list[str]) -> bool: +def _sudo_issue(tokens: list[str]) -> str: for index, token in enumerate(tokens): if Path(token).name != "sudo": continue - if not _sudo_has_non_interactive_option(tokens[index + 1 :]): - return True - return False + non_interactive, mode_flag = _parse_sudo_options(tokens[index + 1 :]) + if mode_flag: + return ( + "shell_safety 拦截：sudo 的交互、编辑或状态模式不作为普通命令执行，" + "请改用明确的非交互命令。" + ) + if not non_interactive: + return ( + "shell_safety 拦截：sudo 可能等待密码，请改用 sudo -n，" + "让它在没有缓存时立即失败。" + ) + return "" -def _sudo_has_non_interactive_option(tokens: list[str]) -> bool: +def _parse_sudo_options(tokens: list[str]) -> tuple[bool, str]: + non_interactive = False index = 0 while index < len(tokens): token = tokens[index] if token == "--": - return False + break if not token.startswith("-") or token == "-": - return False - if token == "-n" or ( - token.startswith("-") - and not token.startswith("--") - and "n" in token[1:] - ): - return True - if token in {"-u", "-g", "-p", "-C", "-D", "-R", "-T", "-h"}: + break + if token == "--non-interactive": + non_interactive = True + index += 1 + continue + if token.startswith("--"): + option = token.split("=", 1)[0] + if option in _SUDO_MODE_LONG_FLAGS: + return non_interactive, option + if token in _SUDO_LONG_OPTIONS_WITH_VALUE: + index += 2 + continue + index += 1 + continue + has_non_interactive, consumes_next, mode_flag = _short_sudo_options(token) + non_interactive = non_interactive or has_non_interactive + if mode_flag: + return non_interactive, mode_flag + if consumes_next: index += 2 continue index += 1 - return False + return non_interactive, "" def _find_interactive_package_command(tokens: list[str]) -> str: @@ -136,3 +185,18 @@ def _opens_system_editor(tokens: list[str]) -> bool: if candidate == "crontab" and tokens[index + 1] == "-e": return True return False + + +def _short_sudo_options(token: str) -> tuple[bool, bool, str]: + has_non_interactive = False + cluster = token[1:] + for offset, option in enumerate(cluster): + if option in _SUDO_MODE_SHORT_FLAGS: + return has_non_interactive, False, option + if option == "n": + has_non_interactive = True + continue + if option not in _SUDO_SHORT_OPTIONS_WITH_VALUE: + continue + return has_non_interactive, offset + 1 == len(cluster), "" + return has_non_interactive, False, "" diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 3a412df..e712b9e 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -52,6 +52,29 @@ def test_blocks_system_editor() -> None: def test_allows_non_interactive_write() -> None: reason = shell_safety.deny_reason("sudo -n pacman -Syu --noconfirm") assert reason == "" + assert shell_safety.deny_reason("sudo -nE pacman -Syu --noconfirm") == "" + assert ( + shell_safety.deny_reason("sudo -nuroot pacman -Syu --noconfirm") == "" + ) + assert ( + shell_safety.deny_reason( + "sudo -n --preserve-env=HOME pacman -Syu --noconfirm" + ) + == "" + ) + + +def test_sudo_option_value_containing_n_is_not_non_interactive() -> None: + reason = shell_safety.deny_reason("sudo -unroot pacman -Syu --noconfirm") + assert "sudo -n" in reason + + +@pytest.mark.parametrize("mode_flag", ["-e", "-l", "-s", "-i", "-v", "-h"]) +def test_sudo_mode_flag_is_denied_after_non_interactive( + mode_flag: str, +) -> None: + reason = shell_safety.deny_reason(f"sudo -n {mode_flag} rm /tmp/a.txt") + assert "不作为普通命令执行" in reason def test_malformed_shell_is_left_to_shell_boundary() -> None: From a3a1ed8aceed03e62530a384343a0769ac10b52e Mon Sep 17 00:00:00 2001 From: huashen <2494946808@qq.com> Date: Sun, 16 Aug 2026 03:02:18 +0800 Subject: [PATCH 3/3] test(plugin): isolate Core checkout from fixture copy --- tests/test_plugin.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_plugin.py b/tests/test_plugin.py index e712b9e..9e20d89 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -146,6 +146,7 @@ async def test_manager_snapshot_authorizes_final_arguments(tmp_path: Path) -> No plugin_home / "shell_safety", ignore=shutil.ignore_patterns( ".git", + ".akashic-core", ".pytest_cache", "__pycache__", ),