fix: batch rg to avoid arg-length crash (audit H4)

dsgai_scanner_tool/cli/dsgai_scan.py

@@ -200,6 +200,41 @@ def glob_match(rel, globs):
 # --------------------------------------------------------------------------- #
 # Rule execution
 # --------------------------------------------------------------------------- #
+# Keep each rg command line well under the Windows CreateProcess cap (32767);
+# other platforms allow far more, but one conservative budget is simplest.
+_ARG_BUDGET = 24000
+
+
+def _batches(files):
+    """Yield lists of files whose joined path lengths stay under _ARG_BUDGET, so
+    a single rg invocation can never exceed the OS command-line limit (audit H4)."""
+    batch, size = [], 0
+    for f in files:
+        n = len(f[0]) + 1
+        if batch and size + n > _ARG_BUDGET:
+            yield batch
+            batch, size = [], 0
+        batch.append(f)
+        size += n
+    if batch:
+        yield batch
+
+
+def _run_rg(rg, base_cmd, files, scan_root, rid):
+    """Run rg over `files` in as many batches as needed; return stdout lines."""
+    lines = []
+    for batch in _batches(files):
+        cmd = base_cmd + [f[0] for f in batch]
+        try:
+            proc = subprocess.run(cmd, capture_output=True, text=True, cwd=scan_root)
+        except OSError as exc:  # e.g. arg list too long on a pathological path
+            raise RuntimeError(f"rg invocation failed for {rid}: {exc}")
+        if proc.returncode >= 2:
+            raise RuntimeError(f"rg failed on {rid}: {proc.stderr.strip()[:200]}")
+        lines.extend(proc.stdout.splitlines())
+    return lines
+
+
 def run_rule(rg, rule, files, scan_root):
     """Return a set of (rel_path, line) matches for one rule.
 
@@ -208,22 +243,14 @@ def run_rule(rg, rule, files, scan_root):
     """
     if not files:
         return set()
-    cmd = [rg, "--pcre2", "--no-config", "--with-filename", "--line-number"]
+    base = [rg, "--pcre2", "--no-config", "--with-filename", "--line-number"]
     if rule["classification"] == "value_bearing":
         # LOCATION-ONLY: rg erases the matched text before emitting anything.
-        cmd += ["--only-matching", "--replace", ""]
-    cmd += ["-e", rule["pcre"], "--"]
-    cmd += [f[0] for f in files]
-    proc = subprocess.run(cmd, capture_output=True, text=True, cwd=scan_root)
-    if proc.returncode >= 2:
-        raise RuntimeError(f"rg failed on {rule['id']}: {proc.stderr.strip()[:200]}")
+        base += ["--only-matching", "--replace", ""]
+    base += ["-e", rule["pcre"], "--"]
     matches = set()
     abs_to_rel = {f[0]: f[1] for f in files}
-    for line in proc.stdout.splitlines():
-        # Format: <path>:<line>:<rest>. Path is an absolute path we passed in;
-        # split off the trailing :line:rest without touching a drive colon.
-        rest = line
-        # find ':<digits>:' anchor from the path we know
+    for line in _run_rg(rg, base, files, scan_root, rule.get("id", "?")):
         parsed = _parse_rg_line(line, abs_to_rel)
         if parsed:
             matches.add(parsed)
@@ -245,15 +272,13 @@ def _parse_rg_line(line, abs_to_rel):
 def detect_stack(rg, files, scan_root):
     """Return the set of detected signal categories for NOT APPLICABLE gating."""
     detected = set()
-    paths = [f[0] for f in files]
-    if not paths:
+    if not files:
         return detected
     for cat, patterns in DETECT_SIGNALS.items():
         pat = "|".join(patterns)
-        proc = subprocess.run(
-            [rg, "--pcre2", "--no-config", "-l", "-i", "-e", pat, "--"] + paths,
-            capture_output=True, text=True, cwd=scan_root)
-        if proc.returncode == 0 and proc.stdout.strip():
+        base = [rg, "--pcre2", "--no-config", "-l", "-i", "-e", pat, "--"]
+        # Batched so a large repo can't blow the OS arg limit (audit H4).
+        if _run_rg(rg, base, files, scan_root, f"detect:{cat}"):
             detected.add(cat)
     return detected
 

dsgai_scanner_tool/tests/test_runner.py

@@ -363,6 +363,18 @@ def test_gitignored_env_is_still_scanned(tmp_path):
     assert cp["controls"]["DSGAI02"] == "FAIL"
 
 
+def test_rg_arg_batching():
+    """Files are split into batches under the command-line budget so a large
+    repo can't blow the OS arg limit (audit H4)."""
+    ds = _import_cli()
+    files = [("/some/long/abs/path/module_%04d.py" % i, "m%d.py" % i) for i in range(5000)]
+    batches = list(ds._batches(files))
+    assert len(batches) > 1, "5000 files should split into multiple rg batches"
+    for b in batches:
+        assert sum(len(f[0]) + 1 for f in b) <= ds._ARG_BUDGET or len(b) == 1
+    assert sum(len(b) for b in batches) == len(files)  # no file dropped
+
+
 def test_rules_json_in_sync():
     from_yaml = yaml.safe_load(open(RULES_YAML, encoding="utf-8"))
     rebuilt = json.dumps(from_yaml, indent=2, sort_keys=True, ensure_ascii=False) + "\n"