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",