fischer
/
fischer-agentkit
1
0
Fork 0
Code Issues Pull Requests Actions 87 Packages Projects Releases Wiki Activity

feat(agent): Wave 4 PLAN_EXEC Hardening (U1-U5) #7

Merged
fischer merged 8 commits from feat/agent-wave4-plan-exec-hardening into main 2026-06-30 12:46:35 +08:00
Conversation 0 Commits 8 Files Changed 15 +26 -13
1 changed files with 26 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit cbbe937940 - Show all commits

39
src/agentkit/tools/shell.py
View File

@ -18,7 +18,7 @@ from typing import Any, Callable, Awaitable
from agentkit.tools.base import Tool from agentkit.tools.base import Tool
from agentkit.tools.output_parser import OutputParser, ParsedOutput from agentkit.tools.output_parser import OutputParser, ParsedOutput
from agentkit.tools.terminal_session import TerminalSession, TerminalSessionManager from agentkit.tools.terminal_session import TerminalSessionManager
from agentkit.tools.pty_session import PTYSession from agentkit.tools.pty_session import PTYSession
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -159,10 +159,22 @@ _SAFE_COMMAND_PREFIXES: tuple[str, ...] = (
# 危险命令检测 — 基于精确 token 匹配，避免子串误判 # 危险命令检测 — 基于精确 token 匹配，避免子串误判
# 总是危险的二进制命令（无论参数） # 总是危险的二进制命令（无论参数）
_DANGEROUS_BINARIES: frozenset[str] = frozenset({ _DANGEROUS_BINARIES: frozenset[str] = frozenset(
"rm", "rmdir", "mkfs", "dd", "format", "shutdown", "reboot", {
"halt", "killall", "chown", "fdisk", "parted", "rm",
}) "rmdir",
"mkfs",
"dd",
"format",
"shutdown",
"reboot",
"halt",
"killall",
"chown",
"fdisk",
"parted",
}
)
# 需要特定参数才危险的二进制命令：binary → 危险 flag/子命令集合 # 需要特定参数才危险的二进制命令：binary → 危险 flag/子命令集合
_DANGEROUS_BINARY_FLAGS: dict[str, set[str]] = { _DANGEROUS_BINARY_FLAGS: dict[str, set[str]] = {
@ -183,13 +195,16 @@ _DANGEROUS_ARG_PATTERNS: list[re.Pattern[str]] = [
re.compile(r"drop\s+database", re.IGNORECASE), re.compile(r"drop\s+database", re.IGNORECASE),
re.compile(r"truncate\s+table", re.IGNORECASE), re.compile(r"truncate\s+table", re.IGNORECASE),
# curl/wget data exfiltration: POST/PUT/upload flags # curl/wget data exfiltration: POST/PUT/upload flags
re.compile(r"\bcurl\b.*(-X\s*(POST|PUT|PATCH|DELETE)|--data|--data-binary|--data-raw|--data-urlencode|-d\s|--post\d)", re.IGNORECASE), re.compile(
r"\bcurl\b.*(-X\s*(POST|PUT|PATCH|DELETE)|--data|--data-binary|--data-raw|--data-urlencode|-d\s|--post\d)",
re.IGNORECASE,
),
re.compile(r"\bwget\b.*(--post-data|--post-file)", re.IGNORECASE), re.compile(r"\bwget\b.*(--post-data|--post-file)", re.IGNORECASE),
] ]
_SHELL_PIPE_OPERATORS = re.compile(r'\|') _SHELL_PIPE_OPERATORS = re.compile(r"\|")
_SHELL_CHAIN_OPERATORS = re.compile(r'[;&]|\|\||&&|\$\(|\$\{|`|\$<|>|<|\n') _SHELL_CHAIN_OPERATORS = re.compile(r"[;&]|\|\||&&|\$\(|\$\{|`|\$<|>|<|\n")
class ShellTool(Tool): class ShellTool(Tool):
@ -360,6 +375,7 @@ class ShellTool(Tool):
# Ensure non-empty output for successful commands (all execution modes) # Ensure non-empty output for successful commands (all execution modes)
if result.exit_code == 0 and not output.strip(): if result.exit_code == 0 and not output.strip():
from agentkit.core.fallback import SHELL_NO_OUTPUT from agentkit.core.fallback import SHELL_NO_OUTPUT
output = SHELL_NO_OUTPUT output = SHELL_NO_OUTPUT
return { return {
@ -383,7 +399,6 @@ class ShellTool(Tool):
if interactive: if interactive:
return await self._execute_with_pty(command, timeout, working_dir) return await self._execute_with_pty(command, timeout, working_dir)
start = time.monotonic()
try: try:
proc = await asyncio.create_subprocess_shell( proc = await asyncio.create_subprocess_shell(
command, command,
@ -430,9 +445,7 @@ class ShellTool(Tool):
) )
if interactive: if interactive:
return await self._execute_with_pty( return await self._execute_with_pty(command, timeout, session.cwd, session.env)
command, timeout, session.cwd, session.env
)
return await session.execute(command, timeout=timeout) return await session.execute(command, timeout=timeout)
@ -482,7 +495,7 @@ class ShellTool(Tool):
# Handle pipe commands: split and check each sub-command # Handle pipe commands: split and check each sub-command
if _SHELL_PIPE_OPERATORS.search(command_stripped): if _SHELL_PIPE_OPERATORS.search(command_stripped):
parts = command_stripped.split('|') parts = command_stripped.split("|")
for part in parts: for part in parts:
part = part.strip() part = part.strip()
if not part: if not part: