Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
182 changes: 181 additions & 1 deletion src/skillspector/nodes/analyzers/behavioral_ast.py
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,47 @@ def _constant_string(node: ast.expr) -> str | None:

_TAG = "Dangerous Code Execution"

_AST4_FIXED_ARGV_EXPLANATION = (
"Fixed argument vector: subprocess launches a named executable with literal arguments, "
"and shell expansion is disabled. This still starts an external process, but these literal "
"arguments do not create the generic shell-injection path."
)
_AST4_FIXED_ARGV_REMEDIATION = (
"Keep the explicit argument vector and shell=False. Validate the executable, its resolved path, "
"and any dynamic arguments before launch; do not switch to shell=True."
)
_AST4_SHELL_BUILD_EXPLANATION = (
"Shell command construction: subprocess runs a command string with shell=True, so "
"concatenated input and shell metacharacters can change the invocation or inject additional "
"commands."
)
_AST4_SHELL_BUILD_REMEDIATION = (
"Replace the shell command string with an explicit argument vector and shell=False, then "
"validate or allowlist every dynamic argument before launch."
)
_AST4_SHELL_STRING_EXPLANATION = (
"Shell command string: subprocess runs the command with shell expansion enabled, so shell "
"metacharacters in the command can change the invocation or inject additional commands."
)
_AST4_SHELL_STRING_REMEDIATION = _AST4_SHELL_BUILD_REMEDIATION
_AST4_UNKNOWN_EXPLANATION = (
"Unknown caller input: the executable or argument values are not statically known, while "
"shell expansion is disabled by default. If those values are attacker-controlled, they can "
"select a different program or alter that program's behavior."
)
_AST4_UNKNOWN_REMEDIATION = (
"Make the executable and argument vector explicit, avoid shell=True, resolve the executable "
"from an allowlist, and validate every dynamic argument before launch."
)
_AST4_UNKNOWN_SHELL_EXPLANATION = (
"Dynamic subprocess invocation: the effective shell mode is not statically known, so "
"command-injection risk cannot be ruled out from this call alone."
)
_AST4_UNKNOWN_SHELL_REMEDIATION = (
"Make shell mode explicit, keep it False unless a shell is required, and validate or allowlist "
"the executable and every dynamic argument before launch."
)


class _BehavioralResourceLimitError(RuntimeError):
"""Internal signal that retains findings constructed before a hard limit."""
Expand Down Expand Up @@ -331,6 +372,133 @@ def _is_true_constant(node: ast.expr) -> bool:
return isinstance(node, ast.Constant) and node.value is True


def _subprocess_shell_mode(node: ast.Call, attr: str) -> bool | None:
"""Return the literal shell mode for *node*, or None when it is dynamic."""
if attr in {"getoutput", "getstatusoutput"}:
return True
if any(keyword.arg is None for keyword in node.keywords):
return None
for keyword in reversed(node.keywords):
Comment thread
rng1995 marked this conversation as resolved.
if keyword.arg != "shell":
continue
if isinstance(keyword.value, ast.Constant) and isinstance(keyword.value.value, bool):
return keyword.value.value
return None
return False


_SHELL_INTERPRETERS = frozenset({"bash", "dash", "ksh", "sh", "zsh"})
_CMD_INTERPRETERS = frozenset({"cmd", "cmd.exe"})
_POWERSHELL_INTERPRETERS = frozenset({"powershell", "powershell.exe", "pwsh", "pwsh.exe"})


def _literal_shell_interpreter(node: ast.expr | None) -> bool:
"""Return whether literal argv[0] names a known shell interpreter."""
if not isinstance(node, (ast.List, ast.Tuple)) or not node.elts:
return False
executable = _constant_string(node.elts[0])
if executable is None:
return False
basename = executable.replace("\\", "/").rstrip("/").rsplit("/", 1)[-1].lower()
return (
basename in _SHELL_INTERPRETERS
or basename in _CMD_INTERPRETERS
or basename in _POWERSHELL_INTERPRETERS
)


def _literal_inline_shell_command(node: ast.expr | None) -> tuple[bool, ast.expr | None]:
"""Return the command argument when literal argv explicitly invokes a shell."""
if not isinstance(node, (ast.List, ast.Tuple)) or len(node.elts) < 2:
return False, None
executable = _constant_string(node.elts[0])
if executable is None:
return False, None

basename = executable.replace("\\", "/").rstrip("/").rsplit("/", 1)[-1].lower()
for index, option in enumerate(node.elts[1:], start=1):
flag = _constant_string(option)
if flag is None:
continue
normalized_flag = flag.lower()
if basename in _SHELL_INTERPRETERS:
inline = flag == "-c" or (
flag.startswith("-") and not flag.startswith("--") and "c" in flag[1:]
)
elif basename in _CMD_INTERPRETERS:
inline = normalized_flag == "/c"
elif basename in _POWERSHELL_INTERPRETERS:
inline = normalized_flag in {"-c", "-command", "/c", "/command"}
else:
inline = False
if inline:
return True, node.elts[index + 1] if index + 1 < len(node.elts) else None
return False, None


def _has_dynamic_executable(node: ast.Call) -> bool:
"""Return whether ``executable=`` is present but not a literal string."""
return any(
keyword.arg == "executable" and _constant_string(keyword.value) is None
for keyword in node.keywords
)


def _subprocess_command_arg(node: ast.Call) -> ast.expr | None:
"""Return the positional or keyword ``args`` expression for a subprocess call."""
if node.args:
return node.args[0]
for keyword in reversed(node.keywords):
if keyword.arg == "args":
return keyword.value
return None


def _is_fixed_argv(node: ast.expr | None) -> bool:
"""Return True when *node* is a literal list/tuple argument vector."""
if not isinstance(node, (ast.List, ast.Tuple)):
return False
return all(_constant_string(item) is not None for item in node.elts)


def _is_constructed_shell_command(node: ast.expr | None) -> bool:
"""Return True when *node* builds a shell command from multiple parts."""
if isinstance(node, ast.JoinedStr):
return True
if isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Add, ast.Mod)):
return True
return (
isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr in {"format", "format_map", "join"}
)


def _ast4_guidance(node: ast.Call, attr: str) -> tuple[str, str]:
"""Return contextual AST4 guidance without changing detection or scoring."""
command = _subprocess_command_arg(node)
shell_mode = _subprocess_shell_mode(node, attr)
inline_shell, inline_command = _literal_inline_shell_command(command)

if shell_mode is True:
if _is_constructed_shell_command(command):
return _AST4_SHELL_BUILD_EXPLANATION, _AST4_SHELL_BUILD_REMEDIATION
return _AST4_SHELL_STRING_EXPLANATION, _AST4_SHELL_STRING_REMEDIATION
if inline_shell:
if _is_constructed_shell_command(inline_command):
return _AST4_SHELL_BUILD_EXPLANATION, _AST4_SHELL_BUILD_REMEDIATION
return _AST4_SHELL_STRING_EXPLANATION, _AST4_SHELL_STRING_REMEDIATION
if shell_mode is None:
return _AST4_UNKNOWN_SHELL_EXPLANATION, _AST4_UNKNOWN_SHELL_REMEDIATION
if _has_dynamic_executable(node):
Comment thread
rng1995 marked this conversation as resolved.
return _AST4_UNKNOWN_EXPLANATION, _AST4_UNKNOWN_REMEDIATION
if _literal_shell_interpreter(command):
return _AST4_SHELL_STRING_EXPLANATION, _AST4_SHELL_STRING_REMEDIATION
if _is_fixed_argv(command):

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Blocker] When argv[0] is a recognized shell but no inline flag is found, this branch returns the fixed-argv text ("shell expansion is disabled … Keep the explicit argument vector"), but a shell runs code in each of these calls: subprocess.run(["bash"], input=script), subprocess.run(["sh", "-s"], input=payload), subprocess.run(["powershell", "-NoProfile", "-EncodedCommand", "SQBFAFgA…"]) (also -e/-ec) and ["cmd", "/k", …]. Rather than adding more flags, please never return fixed-argv guidance when the basename of argv[0] is in _SHELL_INTERPRETERS, _CMD_INTERPRETERS or _POWERSHELL_INTERPRETERS. _literal_inline_shell_command returns early for one-element lists, so ["bash"] needs the check outside it. Please add tests for ["bash"] with input= and for powershell -EncodedCommand. Skipping launchers such as env/sudo with the same check is optional.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 21be3c2. Literal argv[0] is now checked against _SHELL_INTERPRETERS, _CMD_INTERPRETERS, and _POWERSHELL_INTERPRETERS independently of any flag, before the fixed-argv branch. ["bash"] with input=, ["sh", "-s"], PowerShell -EncodedCommand, and ["cmd", "/k", ...] now receive shell-command guidance and never the fixed-argv text. Added regressions for all four forms; the focused file passes 146 tests and Ruff check/format pass.

return _AST4_FIXED_ARGV_EXPLANATION, _AST4_FIXED_ARGV_REMEDIATION
return _AST4_UNKNOWN_EXPLANATION, _AST4_UNKNOWN_REMEDIATION


def _kwarg_is_true(node: ast.Call, name: str) -> bool:
"""True if keyword *name* is passed as a literal ``True``."""
return any(kw.arg == name and _is_true_constant(kw.value) for kw in node.keywords)
Expand Down Expand Up @@ -441,6 +609,9 @@ def _emit(
rule_id: str,
ast_node: ast.Call,
msg_override: str | None = None,
*,
explanation: str | None = None,
remediation: str | None = None,
) -> None:
lineno = getattr(ast_node, "lineno", 1)
end_lineno = getattr(ast_node, "end_lineno", None)
Expand All @@ -467,6 +638,8 @@ def _emit(
context=context_for(lineno, start_column if start_column is not None else 0),
matched_text=complete_match[:200],
complete_match=complete_match,
explanation=explanation,
remediation=remediation,
# Evidence participates in report compaction. Keep distinct syntax
# nodes distinguishable when their source spans are incomplete.
evidence=(
Expand Down Expand Up @@ -581,7 +754,14 @@ def _emit(
elif call_name.startswith("subprocess."):
attr = call_name.split(".", 1)[1]
if attr in _SUBPROCESS_CALLS:
_emit(node_index, "AST4", ast_node)
explanation, remediation = _ast4_guidance(ast_node, attr)
_emit(
node_index,
"AST4",
ast_node,
explanation=explanation,
remediation=remediation,
)

elif call_name.startswith("os."):
attr = call_name.split(".", 1)[1]
Expand Down
4 changes: 2 additions & 2 deletions src/skillspector/nodes/analyzers/pattern_defaults.py
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,7 @@ class PatternCategory(StrEnum):
"AST1": "Direct exec() call allows arbitrary code execution. An attacker can inject code that runs with the full privileges of the process.",
"AST2": "Direct eval() call evaluates arbitrary expressions. This can be exploited to execute malicious code or exfiltrate data.",
"AST3": "Dynamic __import__() can load arbitrary modules at runtime, bypassing static analysis and potentially importing malicious code.",
"AST4": "subprocess module calls execute external commands. Without careful input validation, this enables command injection.",
"AST4": "subprocess module calls execute external commands. Command-injection risk depends on whether a shell is enabled and whether untrusted input reaches the command; process execution itself still requires review.",
"AST5": "os.system() and os exec-family calls run shell commands with the process's full privileges, enabling arbitrary command execution.",
"AST6": "compile() creates code objects from strings. When combined with exec()/eval(), it enables obfuscated code execution.",
"AST7": "Dynamic getattr() with a non-literal attribute name can access arbitrary object attributes, potentially bypassing access controls.",
Expand Down Expand Up @@ -422,7 +422,7 @@ class PatternCategory(StrEnum):
"AST1": "Replace exec() with a safe alternative. If dynamic execution is required, use a sandboxed environment or restricted eval with __builtins__ disabled.",
"AST2": "Replace eval() with ast.literal_eval() for data parsing, or use explicit parsing logic. Never evaluate untrusted strings.",
"AST3": "Use standard import statements instead of __import__(). If dynamic loading is needed, use importlib with an allowlist of permitted modules.",
"AST4": "Use subprocess.run() with shell=False and an explicit argument list. Validate all inputs and avoid passing user-controlled data to commands.",
"AST4": "Use an explicit argument vector and keep shell=False unless a shell is strictly required. Validate the executable and every dynamic argument; never pass untrusted input to a shell command.",
"AST5": "Replace os.system() with subprocess.run(shell=False). Use explicit argument lists and validate all command inputs.",
"AST6": "Avoid compile() with dynamic strings. If code generation is needed, use templates or AST manipulation with strict validation.",
"AST7": "Replace dynamic getattr() with explicit attribute access or a dictionary lookup with an allowlist of permitted attributes.",
Expand Down
Loading
Loading