refactor: simplify destructive command matching

s12_cron_scheduler/code.py

@@ -17,7 +17,6 @@ import glob
 import json
 import os
 import re
-import shlex
 import secrets
 import subprocess
 import threading
@@ -175,191 +174,14 @@ def trigger_hooks(event: str, *args):
 
 
 DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="]
-SHELL_SEPARATORS = ";&|\n"
-DESTRUCTIVE_COMMANDS = {"rm", "del"}
-SHELL_WRAPPERS = {"sh", "bash", "zsh", "dash", "cmd", "cmd.exe"}
-COMMAND_PREFIXES = {"command", "call"}
-CONTROL_PREFIXES = {"then", "do", "else", "!", "{"}
-COMPARE_OPERATORS = {"equ", "neq", "lss", "leq", "gtr", "geq"}
-MAX_COMMAND_NESTING = 16
-DESTRUCTIVE_SUBCOMMAND = re.compile(
-    r"(?i)(?:\$\(|[<>]\(|\x60)\s*(?:rm|del)"
-    r"(?=\s|$|[;&|()])"
+DESTRUCTIVE_COMMAND_WORD = re.compile(
+    r"(?i)(?:^|[;&|()\n])\s*(?:rm|del)(?=\s|$|[;&|()])"
 )
-DESTRUCTIVE = ["> /etc/", "chmod 777"]
+DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"]
 
 
-def shell_tokens(command: str) -> list[str]:
-    lexer = shlex.shlex(
-        command, posix=False, punctuation_chars=SHELL_SEPARATORS
-    )
-    lexer.whitespace = " \t\r"
-    lexer.whitespace_split = True
-    lexer.commenters = ""
-    return list(lexer)
-
-
-def shell_syntax_outside_single_quotes(command: str) -> str:
-    visible = []
-    single_quoted = double_quoted = escaped = False
-    for char in command:
-        if escaped:
-            visible.append(" ")
-            escaped = False
-        elif char == "\\" and not single_quoted:
-            visible.append(" ")
-            escaped = True
-        elif char == '"' and not single_quoted:
-            double_quoted = not double_quoted
-            visible.append(char)
-        elif char == "'" and not double_quoted:
-            single_quoted = not single_quoted
-            visible.append(" ")
-        else:
-            visible.append(" " if single_quoted else char)
-    return "".join(visible)
-
-
-def unquote_shell_token(token: str) -> str:
-    if len(token) >= 2 and token[0] in "'\"" and token[-1] == token[0]:
-        return token[1:-1]
-    return token
-
-
-def command_name(token: str) -> str:
-    value = unquote_shell_token(token).lstrip("@").strip("()").casefold()
-    if value.startswith("del/"):
-        return "del"
-    return value.replace("\\", "/").rsplit("/", 1)[-1]
-
-
-def is_shell_separator(token: str) -> bool:
-    return bool(token) and all(char in SHELL_SEPARATORS for char in token)
-
-
-def is_shell_assignment(token: str) -> bool:
-    name, separator, _ = unquote_shell_token(token).partition("=")
-    return bool(
-        separator
-        and name
-        and not name[0].isdigit()
-        and name.replace("_", "a").isalnum()
-    )
-
-
-def segment_has_destructive_command(
-    tokens: list[str], depth: int = 0
-) -> bool:
-    if depth >= MAX_COMMAND_NESTING:
-        return True
-
-    index = 0
-    while index < len(tokens) and is_shell_assignment(tokens[index]):
-        index += 1
-    if index >= len(tokens):
-        return False
-
-    name = command_name(tokens[index])
-    if name in DESTRUCTIVE_COMMANDS:
-        return True
-    if name in CONTROL_PREFIXES:
-        return segment_has_destructive_command(tokens[index + 1:], depth + 1)
-    if name == "env":
-        index += 1
-        while index < len(tokens) and (
-            unquote_shell_token(tokens[index]).startswith("-")
-            or is_shell_assignment(tokens[index])
-        ):
-            index += 1
-        return segment_has_destructive_command(tokens[index:], depth + 1)
-    if name in COMMAND_PREFIXES:
-        index += 1
-        options = []
-        while (
-            index < len(tokens)
-            and unquote_shell_token(tokens[index]).startswith("-")
-        ):
-            options.append(unquote_shell_token(tokens[index]))
-            index += 1
-        if name == "command" and any(
-            "v" in option.lstrip("-").casefold() for option in options
-        ):
-            return False
-        return segment_has_destructive_command(tokens[index:], depth + 1)
-    if name in SHELL_WRAPPERS:
-        for flag_index in range(index + 1, len(tokens)):
-            flag = unquote_shell_token(tokens[flag_index]).casefold()
-            is_command_flag = (
-                flag in {"/c", "/k"}
-                if name.startswith("cmd")
-                else flag.startswith("-")
-                and not flag.startswith("--")
-                and "c" in flag[1:]
-            )
-            if is_command_flag:
-                nested = " ".join(
-                    unquote_shell_token(token)
-                    for token in tokens[flag_index + 1:]
-                )
-                return contains_destructive_command(nested, depth + 1)
-        return False
-    if name == "if":
-        index += 1
-        while (
-            index < len(tokens)
-            and command_name(tokens[index]) in {"/i", "not"}
-        ):
-            index += 1
-        if index >= len(tokens):
-            return False
-        condition = command_name(tokens[index])
-        if condition in {"exist", "defined", "errorlevel", "cmdextversion"}:
-            return segment_has_destructive_command(
-                tokens[index + 2:], depth + 1
-            )
-        if "==" in unquote_shell_token(tokens[index]):
-            return segment_has_destructive_command(
-                tokens[index + 1:], depth + 1
-            )
-        if (
-            index + 2 < len(tokens)
-            and command_name(tokens[index + 1]) in COMPARE_OPERATORS
-        ):
-            return segment_has_destructive_command(
-                tokens[index + 3:], depth + 1
-            )
-        return False
-    if name == "for":
-        for do_index, token in enumerate(tokens[index + 1:], index + 1):
-            if command_name(token) == "do":
-                return segment_has_destructive_command(
-                    tokens[do_index + 1:], depth + 1
-                )
-    return False
-
-
-def contains_destructive_command(command: str, depth: int = 0) -> bool:
-    if depth >= MAX_COMMAND_NESTING:
-        return True
-
-    try:
-        tokens = shell_tokens(command)
-    except ValueError:
-        return True
-    if DESTRUCTIVE_SUBCOMMAND.search(
-        shell_syntax_outside_single_quotes(command)
-    ):
-        return True
-
-    segment = []
-    for token in tokens:
-        if is_shell_separator(token):
-            if segment_has_destructive_command(segment, depth):
-                return True
-            segment = []
-        else:
-            segment.append(token)
-    return segment_has_destructive_command(segment, depth)
+def contains_destructive_command(command: str) -> bool:
+    return bool(DESTRUCTIVE_COMMAND_WORD.search(command))
 
 
 def request_permission(block, reason: str) -> str | None: