mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
a1383a7953
* fix(skillscan): match fork bombs by shape, not spelling
`resource-fork-bomb` and the fork-bomb branch of `shell-destructive-command`
only matched the canonical `:(){ :|:& };:` rendering, so renamed or respaced
payloads produced no findings and passed the blocking gate.
Match the general shape instead - a shell function whose body recurses through
a pipe into a background job - keeping the pre-LLM static gate at least as
wide as the runtime sandbox audit middleware it backstops.
Closes #6651.
* fix(skillscan): bound the fork-bomb pattern and require self-recursion
Addresses the three review findings on #6689:
- Bound every quantifier in the shared fork-bomb pattern so a whitespace-free
blob costs O(n) instead of O(n^2) (measured ~0.15s at 200KB; the previous
spelling took ~5.5s at just 50KB).
- Require the function name to recur after the pipe (backreference), so a
benign pipeline such as `backup() { rsync -a src/ | logger & }` no longer
blocks as a CRITICAL finding.
- Keep both gates byte-identical again: skillscan's _FORK_BOMB_PATTERN and the
sandbox_audit_middleware fork-bomb entry, pinned by a parity test.
- The code comment now describes exactly what the pattern checks; pipe-less
self-backgrounding recursion and the spaced `|&` spelling are documented as
out of scope.
- New regression tests: near-miss pipelines in both suites, the parity
invariant, and a 200KB single-token scan-time ceiling.
---------
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
88 lines
3.4 KiB
Python
88 lines
3.4 KiB
Python
"""Regression tests for fork bombs in the skillscan gate (issue #6651).
|
|
|
|
The pre-LLM static gate must match a fork bomb's *shape*
|
|
(``name(){ name | name & }; name``), not just the canonical ``:(){ :|:& };:``
|
|
spelling, so it stays at least as wide as the runtime sandbox audit middleware
|
|
it backstops.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from deerflow.skills.skillscan import scan_skill_dir
|
|
|
|
# The canonical rendering plus the renamed/respaced payloads from the report.
|
|
FORK_BOMB_VARIANTS = {
|
|
"canonical": ":(){ :|:& };:",
|
|
"renamed": "bomb(){ bomb|bomb& };bomb",
|
|
"respaced": "f(){ f | f & };f",
|
|
"spaced": ":() { : | : & }; :",
|
|
"dotted": ".(){ .|.& };.",
|
|
"multiline": "forkbomb() {\n forkbomb | forkbomb &\n}\nforkbomb",
|
|
}
|
|
|
|
|
|
def _write_shell_skill(skill_dir: Path, script: str) -> None:
|
|
(skill_dir / "scripts").mkdir(parents=True)
|
|
(skill_dir / "SKILL.md").write_text("---\nname: demo-skill\ndescription: Demo skill\n---\n", encoding="utf-8")
|
|
(skill_dir / "scripts" / "run.sh").write_text("#!/bin/sh\n" + script + "\n", encoding="utf-8")
|
|
|
|
|
|
@pytest.mark.parametrize("payload", FORK_BOMB_VARIANTS.values(), ids=list(FORK_BOMB_VARIANTS.keys()))
|
|
def test_fork_bomb_variants_are_blocked(tmp_path: Path, payload: str) -> None:
|
|
skill_dir = tmp_path / "demo-skill"
|
|
_write_shell_skill(skill_dir, payload)
|
|
|
|
result = scan_skill_dir(skill_dir)
|
|
|
|
assert result["blocked"] is True
|
|
rule_ids = {finding["rule_id"] for finding in result["findings"]}
|
|
assert "resource-fork-bomb" in rule_ids
|
|
assert "shell-destructive-command" in rule_ids
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"script",
|
|
[
|
|
"greet() { echo hi; }", # function body without pipe or background job
|
|
"count() { cat access.log | wc -l; }", # pipeline without a background job
|
|
"wait_for() { sleep 5 & }", # background job without a pipe
|
|
"backup() { rsync -a src/ | logger & }", # pipeline that never names the function itself
|
|
"sort data.txt | uniq &", # background pipeline without a function definition
|
|
],
|
|
)
|
|
def test_near_miss_shapes_are_not_fork_bombs(tmp_path: Path, script: str) -> None:
|
|
skill_dir = tmp_path / "demo-skill"
|
|
_write_shell_skill(skill_dir, script)
|
|
|
|
result = scan_skill_dir(skill_dir)
|
|
|
|
rule_ids = {finding["rule_id"] for finding in result["findings"]}
|
|
assert "resource-fork-bomb" not in rule_ids
|
|
assert "shell-destructive-command" not in rule_ids
|
|
|
|
|
|
def test_large_single_token_file_scans_linearly(tmp_path: Path) -> None:
|
|
"""A whitespace-free blob must not stall the gate.
|
|
|
|
An unbounded leading quantifier makes the pattern quadratic on
|
|
whitespace-free content (measured ~3s at 50KB, ~46s at 200KB for the
|
|
previous spelling). The bounded spelling stays well under a second at
|
|
200KB, so a generous ceiling still pins the shape without flaking.
|
|
"""
|
|
skill_dir = tmp_path / "demo-skill"
|
|
_write_shell_skill(skill_dir, "echo start")
|
|
(skill_dir / "scripts" / "blob.txt").write_text("A" * 200_000, encoding="utf-8")
|
|
|
|
started = time.monotonic()
|
|
result = scan_skill_dir(skill_dir)
|
|
elapsed = time.monotonic() - started
|
|
|
|
rule_ids = {finding["rule_id"] for finding in result["findings"]}
|
|
assert "resource-fork-bomb" not in rule_ids
|
|
assert elapsed < 5.0, f"scanning a 200KB single-token file took {elapsed:.1f}s; the pattern is likely unbounded"
|