diff --git a/backend/packages/harness/deerflow/skills/skillscan/orchestrator.py b/backend/packages/harness/deerflow/skills/skillscan/orchestrator.py index 723db9d7a..62ee64c99 100644 --- a/backend/packages/harness/deerflow/skills/skillscan/orchestrator.py +++ b/backend/packages/harness/deerflow/skills/skillscan/orchestrator.py @@ -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: diff --git a/backend/tests/test_skillscan_native.py b/backend/tests/test_skillscan_native.py index b671f3e4a..d743bb2b5 100644 --- a/backend/tests/test_skillscan_native.py +++ b/backend/tests/test_skillscan_native.py @@ -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)