mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-25 07:28:07 +00:00
fix(skills): close AST literal-only shell=True bypass in SkillScan (#4057)
* fix(skills): close AST literal-only shell=True bypass in SkillScan SkillScan's Python analyzer only classified a subprocess call as the CRITICAL, hard-blocked python-shell-exec rule when its shell= keyword was the literal AST constant True. Any non-literal value with the same runtime effect - a variable (shell=shell_flag) or an expression (shell=bool(1)) - fell through to the HIGH, non-blocking python-subprocess classification instead, silently bypassing enforce_static_scan's deterministic CRITICAL gate despite behaving identically to shell=True at runtime. _call_has_shell_true is renamed to _call_shell_may_be_true and now fails closed on ambiguity: any shell= value that is not a literal, statically-provable False is treated as CRITICAL, matching the literal shell=True case. A call with no shell= keyword at all is unaffected (subprocess already defaults to shell=False). Adds regression tests for the variable and expression bypass shapes, plus a boundary test locking in that literal shell=False remains a non-blocking warning. * fix(skills): fail closed on **-unpacked shell= in SkillScan _call_shell_may_be_true only checked keyword.arg == "shell", so a subprocess.* call that supplies shell via **-unpacking (a keyword node with arg is None) fell through to the non-blocking python-subprocess classification instead of the CRITICAL python-shell-exec path. Treat any **-unpacked keyword as shell-ambiguous and fail closed, same as the existing shell=variable/shell=expression handling. This intentionally over-blocks a **-unpack that carries no shell key, since a mapping's contents are not knowable by static analysis; that tradeoff is documented inline and covered by a dedicated test.
This commit is contained in:
parent
09e25b8a32
commit
6544d96cc4
@ -379,7 +379,7 @@ def _scan_python(rel_path: str, text: str) -> list[SecurityFinding]:
|
||||
call_name = _python_call_name(node, aliases)
|
||||
if call_name in {"eval", "exec"} or (call_name == "compile" and _compile_mode_is_exec(node)):
|
||||
findings.append(_finding_for_node("python-dynamic-exec", rel_path, node, call_name))
|
||||
elif call_name in {"os.system", "os.popen"} or (call_name.startswith("subprocess.") and _call_has_shell_true(node)):
|
||||
elif call_name in {"os.system", "os.popen"} or (call_name.startswith("subprocess.") and _call_shell_may_be_true(node)):
|
||||
findings.append(_finding_for_node("python-shell-exec", rel_path, node, call_name))
|
||||
elif call_name.startswith("subprocess."):
|
||||
findings.append(_finding_for_node("python-subprocess", rel_path, node, call_name))
|
||||
@ -691,8 +691,27 @@ def _compile_mode_is_exec(node: ast.Call) -> bool:
|
||||
return any(keyword.arg == "mode" and isinstance(keyword.value, ast.Constant) and keyword.value.value == "exec" for keyword in node.keywords)
|
||||
|
||||
|
||||
def _call_has_shell_true(node: ast.Call) -> bool:
|
||||
return any(keyword.arg == "shell" and isinstance(keyword.value, ast.Constant) and keyword.value.value is True for keyword in node.keywords)
|
||||
def _call_shell_may_be_true(node: ast.Call) -> bool:
|
||||
# Fail closed on ambiguity: a ``shell=`` keyword is only safe when it is a
|
||||
# literal, statically-provable ``False``. Any other value under that keyword,
|
||||
# the literal ``True``, any other literal, or a non-literal expression such as
|
||||
# a variable or a function call (``shell=shell_flag``, ``shell=bool(1)``),
|
||||
# cannot be proven safe by static analysis, so it is treated the same as an
|
||||
# explicit ``shell=True`` rather than silently falling through to the
|
||||
# non-blocking ``python-subprocess`` classification.
|
||||
for keyword in node.keywords:
|
||||
if keyword.arg is None:
|
||||
# ``**mapping`` / ``**kwargs`` unpacking: represented in the AST as a
|
||||
# keyword with ``arg is None``. The mapping's contents (and whether it
|
||||
# even carries a ``shell`` key) are not knowable by static analysis, so
|
||||
# this fails closed the same as a variable/expression shell= value.
|
||||
# Deliberate, documented over-block: a harmless kwargs-unpack with no
|
||||
# ``shell`` key also blocks, since the alternative (inspecting the
|
||||
# unpacked mapping's contents) is not generally possible statically.
|
||||
return True
|
||||
if keyword.arg == "shell":
|
||||
return not (isinstance(keyword.value, ast.Constant) and keyword.value.value is False)
|
||||
return False
|
||||
|
||||
|
||||
def _call_is_network_sink(call_name: str) -> bool:
|
||||
|
||||
@ -198,6 +198,112 @@ def test_python_subprocess_without_shell_warns(tmp_path: Path) -> None:
|
||||
assert not [item for item in findings if item["severity"] == "CRITICAL"]
|
||||
|
||||
|
||||
def test_python_subprocess_shell_false_literal_warns_not_block(tmp_path: Path) -> None:
|
||||
skill_dir = tmp_path / "demo-skill"
|
||||
_write_skill(skill_dir)
|
||||
scripts_dir = skill_dir / "scripts"
|
||||
scripts_dir.mkdir()
|
||||
(scripts_dir / "run.py").write_text("import subprocess\nsubprocess.run(['whoami'], shell=False)\n", encoding="utf-8")
|
||||
|
||||
findings = scan_skill_dir(skill_dir)["findings"]
|
||||
|
||||
finding = _finding_by_rule(findings, "python-subprocess")
|
||||
assert finding["severity"] == "HIGH"
|
||||
assert not [item for item in findings if item["severity"] == "CRITICAL"]
|
||||
|
||||
|
||||
def test_python_subprocess_shell_via_variable_blocks(tmp_path: Path) -> None:
|
||||
# A non-literal shell= value (a variable) is statically indistinguishable
|
||||
# from shell=True in its effect at runtime, so it must be classified and
|
||||
# blocked the same way, not silently downgraded to the non-blocking
|
||||
# python-subprocess warning.
|
||||
skill_dir = tmp_path / "demo-skill"
|
||||
_write_skill(skill_dir)
|
||||
scripts_dir = skill_dir / "scripts"
|
||||
scripts_dir.mkdir()
|
||||
(scripts_dir / "run.py").write_text(
|
||||
"import subprocess\nshell_flag = True\nsubprocess.run(['whoami'], shell=shell_flag)\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
findings = scan_skill_dir(skill_dir)["findings"]
|
||||
|
||||
assert _finding_by_rule(findings, "python-shell-exec")["severity"] == "CRITICAL"
|
||||
assert not [item for item in findings if item["rule_id"] == "python-subprocess"]
|
||||
|
||||
with pytest.raises(StaticScanBlockedError) as excinfo:
|
||||
enforce_static_scan(skill_dir, skill_name="demo-skill")
|
||||
assert _finding_by_rule(excinfo.value.findings, "python-shell-exec")["severity"] == "CRITICAL"
|
||||
|
||||
|
||||
def test_python_subprocess_shell_via_expression_blocks(tmp_path: Path) -> None:
|
||||
# Same bypass shape as the variable case above, but via a call expression
|
||||
# (shell=bool(1)) instead of a bare name.
|
||||
skill_dir = tmp_path / "demo-skill"
|
||||
_write_skill(skill_dir)
|
||||
scripts_dir = skill_dir / "scripts"
|
||||
scripts_dir.mkdir()
|
||||
(scripts_dir / "run.py").write_text("import subprocess\nsubprocess.run(['whoami'], shell=bool(1))\n", encoding="utf-8")
|
||||
|
||||
findings = scan_skill_dir(skill_dir)["findings"]
|
||||
|
||||
assert _finding_by_rule(findings, "python-shell-exec")["severity"] == "CRITICAL"
|
||||
assert not [item for item in findings if item["rule_id"] == "python-subprocess"]
|
||||
|
||||
with pytest.raises(StaticScanBlockedError) as excinfo:
|
||||
enforce_static_scan(skill_dir, skill_name="demo-skill")
|
||||
assert _finding_by_rule(excinfo.value.findings, "python-shell-exec")["severity"] == "CRITICAL"
|
||||
|
||||
|
||||
def test_python_subprocess_shell_via_kwargs_unpacking_blocks(tmp_path: Path) -> None:
|
||||
# A ``**``-unpacked mapping can carry a ``shell=True`` key that is invisible
|
||||
# to a plain ``keyword.arg == "shell"`` scan: in the AST, a ``**mapping``
|
||||
# argument is represented as a keyword with ``arg is None``. Its effect is
|
||||
# statically indistinguishable from a literal ``shell=True``, so it must be
|
||||
# classified and blocked the same way, not silently downgraded to the
|
||||
# non-blocking python-subprocess warning.
|
||||
skill_dir = tmp_path / "demo-skill"
|
||||
_write_skill(skill_dir)
|
||||
scripts_dir = skill_dir / "scripts"
|
||||
scripts_dir.mkdir()
|
||||
(scripts_dir / "run.py").write_text(
|
||||
"import subprocess\nopts = {'shell': True}\nsubprocess.run(['whoami'], **opts)\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
findings = scan_skill_dir(skill_dir)["findings"]
|
||||
|
||||
assert _finding_by_rule(findings, "python-shell-exec")["severity"] == "CRITICAL"
|
||||
assert not [item for item in findings if item["rule_id"] == "python-subprocess"]
|
||||
|
||||
with pytest.raises(StaticScanBlockedError) as excinfo:
|
||||
enforce_static_scan(skill_dir, skill_name="demo-skill")
|
||||
assert _finding_by_rule(excinfo.value.findings, "python-shell-exec")["severity"] == "CRITICAL"
|
||||
|
||||
|
||||
def test_python_subprocess_kwargs_unpacking_without_shell_key_still_blocks(tmp_path: Path) -> None:
|
||||
# Known, deliberate over-block, documented here rather than in a code
|
||||
# comment alone: a ``**``-unpacked mapping that provably carries no
|
||||
# ``shell`` key (only ``check`` below) is still treated as shell-ambiguous
|
||||
# and blocked, because what a ``**``-unpacked mapping contains is not
|
||||
# knowable by static analysis in general. Failing closed on every
|
||||
# ``**``-unpack is the conservative, defensible choice over trying to
|
||||
# inspect the unpacked mapping's contents.
|
||||
skill_dir = tmp_path / "demo-skill"
|
||||
_write_skill(skill_dir)
|
||||
scripts_dir = skill_dir / "scripts"
|
||||
scripts_dir.mkdir()
|
||||
(scripts_dir / "run.py").write_text(
|
||||
"import subprocess\nopts = {'check': True}\nsubprocess.run(['whoami'], **opts)\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
findings = scan_skill_dir(skill_dir)["findings"]
|
||||
|
||||
assert _finding_by_rule(findings, "python-shell-exec")["severity"] == "CRITICAL"
|
||||
assert not [item for item in findings if item["rule_id"] == "python-subprocess"]
|
||||
|
||||
|
||||
def test_cloud_metadata_access_is_reported_by_one_rule(tmp_path: Path) -> None:
|
||||
skill_dir = tmp_path / "demo-skill"
|
||||
_write_skill(skill_dir)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user