From cd0df65b7d283d8d8056b92efcd3856a7588d634 Mon Sep 17 00:00:00 2001 From: failsafesecurity Date: Wed, 14 Oct 2026 17:22:51 -0700 Subject: [PATCH] fix(permission): parse shell words before deny/destructive checks The bash permission gate matched raw substrings, so quote splicing (`r''m`), ${IFS} splitting, $'...' quoting, backslash escapes, `find -delete`, and pipe-to-shell bypassed both the deny list and the destructive-command rule and executed for real with no approval prompt. Replace the substring gate with a shared parser that tokenizes the command into shell words (normalising ${IFS}, ${VAR}, $'...' and $VAR before lexing), splits it into pipeline segments, resolves argv[0] to its basename, and denies by command name plus argv inspection. Interpreters (`bash -c`, pipe-to-shell), `find -delete`, and delegating commands (xargs/env/...) are inspected through to their payloads, redirects to sensitive paths require approval, and unparseable quoting fails closed. Applied to s03, s04, s05, s06, s07, s08, s09, s10, s11, s12, s13, s14, s15, s17. Updates the s11 background-tasks test, which previously passed only because the substring gate matched `rm -rf /` inside `rm -rf /tmp/example`. --- s03_permission/code.py | 142 ++++++++++++++++++++++++++++--- s04_hooks/code.py | 147 ++++++++++++++++++++++++++++++--- s05_todo_write/code.py | 147 ++++++++++++++++++++++++++++++--- s06_subagent/code.py | 147 ++++++++++++++++++++++++++++++--- s07_skill_loading/code.py | 147 ++++++++++++++++++++++++++++++--- s08_context_compact/code.py | 145 +++++++++++++++++++++++++++++--- s09_memory/code.py | 145 +++++++++++++++++++++++++++++--- s10_task_system/code.py | 147 ++++++++++++++++++++++++++++++--- s11_background_tasks/code.py | 147 ++++++++++++++++++++++++++++++--- s12_cron_scheduler/code.py | 147 ++++++++++++++++++++++++++++++--- s13_agent_teams/code.py | 145 +++++++++++++++++++++++++++++--- s14_mcp_plugin/code.py | 145 +++++++++++++++++++++++++++++--- s15_integrated_harness/code.py | 141 ++++++++++++++++++++++++++++++- s17_goal_loop/code.py | 145 +++++++++++++++++++++++++++++--- tests/test_background_tasks.py | 5 +- 15 files changed, 1885 insertions(+), 157 deletions(-) diff --git a/s03_permission/code.py b/s03_permission/code.py index 375a2d2be5..dd5cbd3ada 100644 --- a/s03_permission/code.py +++ b/s03_permission/code.py @@ -33,6 +33,7 @@ import os import re +import shlex import subprocess from pathlib import Path @@ -153,23 +154,139 @@ def run_glob(pattern: str) -> str: # -- New in s03: three-gate permission pipeline -- # Gate 1: Hard deny list - always forbidden -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] +DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] + +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + def check_deny_list(command: str) -> str | None: - for pattern in DENY_LIST: - if pattern in command: - return f"Blocked: '{pattern}' is on the deny list" + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" return None -# Gate 2: Rule matching - context-dependent checks -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) - - def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + PERMISSION_RULES = [ @@ -177,8 +294,7 @@ def contains_destructive_command(command: str) -> bool: "check": lambda args: not (WORKDIR / args.get("path", "")).resolve().is_relative_to(WORKDIR), "message": "Writing outside workspace"}, {"tools": ["bash"], - "check": lambda args: contains_destructive_command(args.get("command", "")) or - any(kw in args.get("command", "") for kw in ["rm ", "> /etc/", "chmod 777"]), + "check": lambda args: contains_destructive_command(args.get("command", "")), "message": "Potentially destructive command"}, ] diff --git a/s04_hooks/code.py b/s04_hooks/code.py index bf03037777..019b1f4957 100644 --- a/s04_hooks/code.py +++ b/s04_hooks/code.py @@ -22,6 +22,7 @@ import os import re +import shlex import subprocess from pathlib import Path @@ -148,28 +149,150 @@ def trigger_hooks(event: str, *args): # s03 permission check logic, now wrapped as a hook -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def permission_hook(block): """PreToolUse: s03 check_permission() logic moved here.""" if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - print(f"\n\033[31m[blocked] '{pattern}'\033[0m") - return "Permission denied by deny list" - if contains_destructive_command(command) or any( - kw in command for kw in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + print(f"\n\033[31m[blocked] {reason}\033[0m") + return "Permission denied by deny list" + if contains_destructive_command(command): print(f"\n\033[33m[permission] Potentially destructive command\033[0m") print(f" Tool: {block.name}({block.input})") choice = input(" Allow? [y/N] ").strip().lower() diff --git a/s05_todo_write/code.py b/s05_todo_write/code.py index d5105bd727..d5d72b92de 100644 --- a/s05_todo_write/code.py +++ b/s05_todo_write/code.py @@ -26,6 +26,7 @@ import json import os import re +import shlex import subprocess from pathlib import Path @@ -224,28 +225,150 @@ def trigger_hooks(event: str, *args): return result return None -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def permission_hook(block): """PreToolUse: s03 permission logic, registered as an s04 hook.""" if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - print(f"\n\033[31m[blocked] '{pattern}'\033[0m") - return "Permission denied by deny list" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + print(f"\n\033[31m[blocked] {reason}\033[0m") + return "Permission denied by deny list" + if contains_destructive_command(command): print(f"\n\033[33m[permission] Potentially destructive command\033[0m") print(f" Tool: {block.name}({block.input})") choice = input(" Allow? [y/N] ").strip().lower() diff --git a/s06_subagent/code.py b/s06_subagent/code.py index a5e0438ae3..90534690f4 100644 --- a/s06_subagent/code.py +++ b/s06_subagent/code.py @@ -20,6 +20,7 @@ import os import re +import shlex import subprocess from pathlib import Path @@ -160,28 +161,150 @@ def trigger_hooks(event: str, *args): return None -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def permission_hook(block): """PreToolUse: block denied operations and ask about risky ones.""" if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - print(f"\n\033[31m[blocked] '{pattern}'\033[0m") - return "Permission denied by deny list" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + print(f"\n\033[31m[blocked] {reason}\033[0m") + return "Permission denied by deny list" + if contains_destructive_command(command): print("\n\033[33m[permission] Potentially destructive command\033[0m") print(f" Tool: {block.name}({block.input})") choice = input(" Allow? [y/N] ").strip().lower() diff --git a/s07_skill_loading/code.py b/s07_skill_loading/code.py index 5f88ebb5cb..26426f2ac8 100644 --- a/s07_skill_loading/code.py +++ b/s07_skill_loading/code.py @@ -21,6 +21,7 @@ import os import re +import shlex import subprocess from pathlib import Path @@ -248,28 +249,150 @@ def trigger_hooks(event: str, *args): return None -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def permission_hook(block): """PreToolUse: block denied operations and ask about risky ones.""" if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - print(f"\n\033[31m[blocked] '{pattern}'\033[0m") - return "Permission denied by deny list" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + print(f"\n\033[31m[blocked] {reason}\033[0m") + return "Permission denied by deny list" + if contains_destructive_command(command): print("\n\033[33m[permission] Potentially destructive command\033[0m") print(f" Tool: {block.name}({block.input})") choice = input(" Allow? [y/N] ").strip().lower() diff --git a/s08_context_compact/code.py b/s08_context_compact/code.py index 9e9ca1c5c1..53ea258326 100644 --- a/s08_context_compact/code.py +++ b/s08_context_compact/code.py @@ -40,6 +40,7 @@ import json import os import re +import shlex import subprocess import uuid from pathlib import Path @@ -184,26 +185,148 @@ def trigger_hooks(event: str, *args): return None -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def permission_hook(block): if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - return f"Permission denied by deny list: {pattern}" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + return f"Permission denied by deny list: {reason}" + if contains_destructive_command(command): print("\n\033[33m[permission] Potentially destructive command\033[0m") print(f" Tool: {block.name}({block.input})") if input(" Allow? [y/N] ").strip().lower() not in ("y", "yes"): diff --git a/s09_memory/code.py b/s09_memory/code.py index f9baf803e5..3edfdda45e 100644 --- a/s09_memory/code.py +++ b/s09_memory/code.py @@ -12,6 +12,7 @@ import json import os import re +import shlex import subprocess from pathlib import Path @@ -639,26 +640,148 @@ def trigger_hooks(event: str, *args): return result return None -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def permission_hook(block): if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - return f"Permission denied by deny list: {pattern}" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + return f"Permission denied by deny list: {reason}" + if contains_destructive_command(command): print("\n\033[33m[permission] Potentially destructive command\033[0m") print(f" Tool: {block.name}({block.input})") if input(" Allow? [y/N] ").strip().lower() not in ("y", "yes"): diff --git a/s10_task_system/code.py b/s10_task_system/code.py index 9432896555..4ac28c9f33 100644 --- a/s10_task_system/code.py +++ b/s10_task_system/code.py @@ -25,6 +25,7 @@ import json import os import re +import shlex import secrets import subprocess from dataclasses import asdict, dataclass @@ -449,27 +450,149 @@ def trigger_hooks(event: str, *args): return None -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def permission_hook(block): if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - print(f"\n\033[31m[blocked] '{pattern}'\033[0m") - return "Permission denied by deny list" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + print(f"\n\033[31m[blocked] {reason}\033[0m") + return "Permission denied by deny list" + if contains_destructive_command(command): print("\n\033[33m[permission] Potentially destructive command\033[0m") print(f" Tool: {block.name}({block.input})") choice = input(" Allow? [y/N] ").strip().lower() diff --git a/s11_background_tasks/code.py b/s11_background_tasks/code.py index ee18f04290..d4893b50c5 100644 --- a/s11_background_tasks/code.py +++ b/s11_background_tasks/code.py @@ -15,6 +15,7 @@ import glob import os import re +import shlex import signal import subprocess import threading @@ -254,27 +255,149 @@ def trigger_hooks(event: str, *args): return None -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def permission_hook(block): if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - print(f"\n\033[31m[blocked] '{pattern}'\033[0m") - return "Permission denied by deny list" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + print(f"\n\033[31m[blocked] {reason}\033[0m") + return "Permission denied by deny list" + if contains_destructive_command(command): print("\n\033[33m[permission] Potentially destructive command\033[0m") print(f" Tool: {block.name}({block.input})") choice = input(" Allow? [y/N] ").strip().lower() diff --git a/s12_cron_scheduler/code.py b/s12_cron_scheduler/code.py index 84ca2a7fa9..bf3c8a4cac 100644 --- a/s12_cron_scheduler/code.py +++ b/s12_cron_scheduler/code.py @@ -17,6 +17,7 @@ import json import os import re +import shlex import secrets import subprocess import threading @@ -180,15 +181,139 @@ def trigger_hooks(event: str, *args): return None -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def request_permission(block, reason: str) -> str | None: @@ -206,13 +331,11 @@ def request_permission(block, reason: str) -> str | None: def permission_hook(block): if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - print(f"\n\033[31m[blocked] '{pattern}'\033[0m") - return "Permission denied by deny list" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + print(f"\n\033[31m[blocked] {reason}\033[0m") + return "Permission denied by deny list" + if contains_destructive_command(command): return request_permission(block, "Potentially destructive command") if block.name in ("read_file", "write_file", "edit_file"): diff --git a/s13_agent_teams/code.py b/s13_agent_teams/code.py index 25ae8432fa..f1e29cffa3 100644 --- a/s13_agent_teams/code.py +++ b/s13_agent_teams/code.py @@ -25,6 +25,7 @@ import os import random import re +import shlex import secrets import select import subprocess @@ -1668,15 +1669,139 @@ def run_create_worktree(name: str, task_id: str) -> str: # -- Hooks and Permission Checks -- HOOKS = {"UserPromptSubmit": [], "PreToolUse": [], "PostToolUse": [], "Stop": []} -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def register_hook(event: str, callback): @@ -1696,12 +1821,10 @@ def trigger_hooks(event: str, *args, skip_permission: bool = False): def check_permission(block, prompt_user: bool = True) -> str | None: if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - return f"Permission denied by deny list: {pattern}" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + return f"Permission denied by deny list: {reason}" + if contains_destructive_command(command): if not prompt_user: return "Permission required: ask Lead to run this command." print(f"\n[permission] {block.name}({block.input})") diff --git a/s14_mcp_plugin/code.py b/s14_mcp_plugin/code.py index 3ab12a27f9..2015b7fb7f 100644 --- a/s14_mcp_plugin/code.py +++ b/s14_mcp_plugin/code.py @@ -25,6 +25,7 @@ import glob import os import re +import shlex import subprocess from pathlib import Path @@ -375,15 +376,139 @@ def assemble_system_prompt() -> str: # -- From s04: hooks and permission checks -- HOOKS = {"UserPromptSubmit": [], "PreToolUse": [], "PostToolUse": [], "Stop": []} -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + def register_hook(event: str, callback): @@ -401,12 +526,10 @@ def trigger_hooks(event: str, *args): def permission_hook(block): if block.name == "bash": command = block.input.get("command", "") - for pattern in DENY_LIST: - if pattern in command: - return f"Permission denied by deny list: {pattern}" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + reason = check_deny_list(command) + if reason: + return f"Permission denied by deny list: {reason}" + if contains_destructive_command(command): print(f"\n[permission] {block.name}({block.input})") if input("Allow? [y/N] ").strip().lower() not in {"y", "yes"}: return "Permission denied by user" diff --git a/s15_integrated_harness/code.py b/s15_integrated_harness/code.py index e9549b2d09..cc016be953 100644 --- a/s15_integrated_harness/code.py +++ b/s15_integrated_harness/code.py @@ -25,6 +25,7 @@ import os import random import re +import shlex import secrets import signal import subprocess @@ -1790,7 +1791,139 @@ def trigger_hooks(event: str, *args): return None -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] +DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] + +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + + +def contains_destructive_command(command: str) -> bool: + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + mcp_tool_policies: dict[str, str] = {} @@ -1801,9 +1934,9 @@ def permission_hook(block): command = block.input.get("command", "") if not isinstance(command, str): return "Permission denied: shell command must be a string" - for pattern in DENY_LIST: - if pattern in command: - return f"Permission denied: '{pattern}' is on the deny list" + denied = check_deny_list(command) + if denied: + return f"Permission denied: {denied}" if threading.current_thread() is not threading.main_thread(): return ("Permission denied: interactive shell approval is unavailable " "during an asynchronous turn") diff --git a/s17_goal_loop/code.py b/s17_goal_loop/code.py index 7b05e26666..6578305e50 100644 --- a/s17_goal_loop/code.py +++ b/s17_goal_loop/code.py @@ -32,6 +32,7 @@ import json import os import re +import shlex import subprocess import sys import time @@ -51,15 +52,139 @@ ) MAX_GOAL_LENGTH = 4000 CLEAR_ALIASES = {"clear", "stop", "off", "reset", "none", "cancel"} -DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="] -DESTRUCTIVE_COMMAND_WORD = re.compile( - r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])" -) +# -- Hardened permission gate: parse into shell words before matching -- +# The previous gate matched raw substrings, so quote splicing (`r''m`), +# `${IFS}`, `$'...'`, backslash escapes, `find -delete`, and pipe-to-shell all +# slipped past the deny list and the destructive-command rule and executed for +# real. The helpers below tokenize the command into shell words and inspect the +# resolved argv segments, so those obfuscations are normalised before matching. + +DENY_LIST = ["sudo", "shutdown", "reboot", "mkfs", "dd if=", "> /dev/sda"] DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"] +_OPERATORS = {";", ";;", "&&", "||", "|", "&", "(", ")", "<", ">", ">>", "<<", "<&", ">&"} +_REDIRECT_OPERATORS = {">", ">>", ">&", "<&"} +_INLINE_FLAGS = {"-c", "--eval", "-command", "/c"} +_SHELL_INTERPRETERS = { + "sh", "bash", "dash", "zsh", "ksh", "csh", "tcsh", "fish", + "python", "python2", "python3", "perl", "ruby", "node", "nodejs", + "php", "lua", "osascript", "powershell", "pwsh", "cmd", +} +_DELEGATING_COMMANDS = {"xargs", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "setsid"} +_DESTRUCTIVE_COMMANDS = {"rm", "del", "rmdir", "shred", "unlink", "truncate", "dd", "tee"} +_SENSITIVE_PREFIXES = ("/etc/", "/dev/", "/boot/", "/sys/", "/proc/", "/usr/", "/bin/", "/sbin/", "/lib/") + + +def _normalize_shell_command(command: str) -> str: + """Collapse the shell-level obfuscations a substring gate misses.""" + text = command.replace("${IFS}", " ").replace("$IFS", " ") + text = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", r"\1", text) + text = re.sub(r"\$'", "'", text) + return re.sub(r"\$([A-Za-z_])", r"\1", text) + + +def _tokenize_shell_command(command: str) -> list: + lexer = shlex.shlex(_normalize_shell_command(command), posix=True, + punctuation_chars=";&|()<>") + lexer.whitespace_split = True + lexer.commenters = "" + return list(lexer) + + +def _command_segments(tokens: list): + segments, current, redirects, piped = [], [], [], False + index = 0 + while index < len(tokens): + token = tokens[index] + if token in _REDIRECT_OPERATORS and index + 1 < len(tokens): + redirects.append(tokens[index + 1]) + current.extend([token, tokens[index + 1]]) + index += 2 + continue + if token in _OPERATORS: + if current: + segments.append((current, piped, redirects)) + current, redirects = [], [] + piped = token == "|" + else: + current.append(token) + index += 1 + if current: + segments.append((current, piped, redirects)) + return segments + + +def _argv_basename(word: str) -> str: + return os.path.basename(word).lower() + + +def _is_sensitive_path(path: str) -> bool: + stripped = path.strip("\"'") + return stripped == "/" or stripped.startswith(_SENSITIVE_PREFIXES) + + +def _destructive_argv(argv: list, depth: int = 0) -> bool: + args = [arg for arg in argv if arg not in _OPERATORS] + if not args or depth > 6: + return False + name = _argv_basename(args[0]) + rest = args[1:] + if name in DENY_LIST or name.startswith("mkfs"): + return True + if name in _DESTRUCTIVE_COMMANDS: + return True + if name == "dd": + return any(arg.startswith(("if=", "of=")) for arg in rest) + if name == "find": + return any(arg in ("-delete", "-exec", "-execdir") for arg in rest) + if name in _SHELL_INTERPRETERS: + for position, arg in enumerate(rest): + if arg in _INLINE_FLAGS and position + 1 < len(rest): + return _destructive_argv( + _tokenize_shell_command(rest[position + 1]), depth + 1) or True + return True + if name in _DELEGATING_COMMANDS: + inner = [arg for arg in rest if not arg.startswith("-")] + return _destructive_argv(inner, depth + 1) if inner else False + return False + + +def check_deny_list(command: str) -> str | None: + if not isinstance(command, str): + return "Blocked: shell command must be a string" + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return "Blocked: unparseable shell command" + for argv, _piped, _redirects in _command_segments(tokens): + args = [arg for arg in argv if arg not in _OPERATORS] + if not args: + continue + name = _argv_basename(args[0]) + if name in DENY_LIST or name.startswith("mkfs"): + return f"Blocked: '{name}' is on the deny list" + if name in {"rm", "rmdir", "del", "shred", "truncate"} and any( + arg.strip("\"'") == "/" for arg in args[1:]): + return "Blocked: recursive delete of the filesystem root" + return None + def contains_destructive_command(command: str) -> bool: - return bool(DESTRUCTIVE_COMMAND_WORD.search(command)) + if not isinstance(command, str): + return False + try: + tokens = _tokenize_shell_command(command) + except ValueError: + return True # unparseable quoting fails closed + for argv, piped, redirects in _command_segments(tokens): + if _destructive_argv(argv): + return True + if any(_is_sensitive_path(path) for path in redirects): + return True + if piped and argv and _argv_basename(argv[0]) in _SHELL_INTERPRETERS: + return True + return False + class GoalError(Exception): @@ -609,12 +734,10 @@ def _permission_hook(self, block: Any) -> str | None: command = arguments.get("command", "") if not isinstance(command, str): return "Permission denied: shell command must be a string" - for pattern in DENY_LIST: - if pattern in command: - return f"Permission denied by deny list: {pattern}" - if contains_destructive_command(command) or any( - keyword in command for keyword in DESTRUCTIVE - ): + denied = check_deny_list(command) + if denied: + return f"Permission denied by deny list: {denied}" + if contains_destructive_command(command): print(f"\n[permission] {name}({arguments})") if input("Allow? [y/N] ").strip().lower() not in {"y", "yes"}: return "Permission denied by user" diff --git a/tests/test_background_tasks.py b/tests/test_background_tasks.py index e3b5ce5cf9..e2fdcfcc4d 100644 --- a/tests/test_background_tasks.py +++ b/tests/test_background_tasks.py @@ -92,9 +92,12 @@ def test_background_execution_requires_an_explicit_bash_flag(): ) -def test_background_bash_passes_permission_before_dispatch(): +def test_background_bash_passes_permission_before_dispatch(monkeypatch): with tempfile.TemporaryDirectory() as tmp: lesson = load_lesson(Path(tmp)) + # A scoped delete is not on the hard deny list; the parsed gate routes it + # to interactive approval, so answer the prompt with a denial. + monkeypatch.setattr("builtins.input", lambda _prompt: "n") block = types.SimpleNamespace( id="tool_denied", name="bash",