feat: migrate shell safety to plugin api v3

.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/

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",
@@ -23,97 +22,181 @@
     "--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"
+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
+
+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} 会打开交互式界面，请改用非交互命令。"
+    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，避免卡在确认提示。"
+    if _opens_system_editor(tokens):
+        return "shell_safety 拦截：该命令会打开系统编辑器，请改用写文件或非交互参数。"
+    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"}:
+
+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_issue(tokens: list[str]) -> str:
+    for index, token in enumerate(tokens):
+        if Path(token).name != "sudo":
+            continue
+        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 _parse_sudo_options(tokens: list[str]) -> tuple[bool, str]:
+    non_interactive = False
+    index = 0
+    while index < len(tokens):
+        token = tokens[index]
+        if token == "--":
+            break
+        if not token.startswith("-") or token == "-":
+            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
-        return False
+            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 non_interactive, ""
+
+
+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 _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
 
-    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
-        return ""
 
-    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 _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, ""