Merge pull request #559 from escoffier-labs/fix/content-guard-injection-crossref

fix(handoff): fail content guard lint on injection warnings

Author: solomonneas

src/brigade/handoff_cmd/linting.py

@@ -41,6 +41,21 @@ def _injection_messages(hits: tuple[Any, ...]) -> tuple[str, ...]:
     return tuple(messages)
 
 
+def _egress_verdict(exit_code: Any) -> str:
+    return OK if exit_code == 0 else FAIL
+
+
+def _injection_verdict(hits: tuple[Any, ...]) -> str:
+    return FAIL if any(hit.severity == "warning" for hit in hits) else OK
+
+
+def _injection_detail(*, warning_count: int) -> str:
+    if warning_count:
+        label = "warning" if warning_count == 1 else "warnings"
+        return f"{warning_count} injection {label}"
+    return "clean"
+
+
 def _read_handoff_text(path: Path) -> str | None:
     try:
         return path.read_text(errors="replace")
@@ -71,8 +86,10 @@ def lint(
             hits = scan_handoff_injection_heuristics(text or "") if text is not None else ()
             guard_item["injection_heuristics"] = [_injection_hit_dict(hit) for hit in hits]
             guard_item["injection_warning_count"] = len([hit for hit in hits if hit.severity == "warning"])
+            guard_item["egress_verdict"] = _egress_verdict(guard_item.get("exit_code"))
+            guard_item["injection_verdict"] = _injection_verdict(hits)
             guard_results.append(guard_item)
-    guard_ok = all(item.get("exit_code") == 0 for item in guard_results)
+    guard_ok = all(item.get("egress_verdict") == OK and item.get("injection_verdict") == OK for item in guard_results)
     injection_counts: dict[str, int] = {}
     injection_hits_by_path: dict[str, tuple[Any, ...]] = {}
     enriched_results: list[HandoffLintResult] = []
@@ -110,14 +127,21 @@ def lint(
         hits = injection_hits_by_path.get(path_key, ())
         row["injection_heuristics"] = [_injection_hit_dict(hit) for hit in hits]
         result_dicts.append(row)
-    payload = {
+    payload: dict[str, Any] = {
         "target": str(target),
         "count": len(results),
         "valid": all(result.valid for result in results) and guard_ok,
         "injection_flagged_count": len(injection_counts),
         "results": result_dicts,
         "content_guard": guard_results,
     }
+    if content_guard:
+        payload["content_guard_egress"] = (
+            OK if all(item.get("egress_verdict") == OK for item in guard_results) else FAIL
+        )
+        payload["content_guard_injection"] = (
+            OK if all(item.get("injection_verdict") == OK for item in guard_results) else FAIL
+        )
     if json_output:
         print(json.dumps(payload, indent=2, sort_keys=True))
         return 0 if payload["valid"] else 1
@@ -137,14 +161,18 @@ def lint(
             else:
                 print(f"  warning: {warning}")
     if content_guard:
-        print(f"content_guard_policy: {guard_policy} (leak scan + injection heuristics)")
+        print(f"content_guard_policy: {guard_policy} (egress leak scan + injection heuristics)")
         for item in guard_results:
-            leak_status = OK if item.get("exit_code") == 0 else FAIL
-            print(f"[{leak_status}] content_guard leaks: {item.get('path')} {item.get('detail')}")
+            hits = item.get("injection_heuristics") or []
+            egress_status = item.get("egress_verdict", OK)
+            print(f"[{egress_status}] content_guard egress: {item.get('path')} {item.get('detail')}")
+            injection_status = item.get("injection_verdict", OK)
             warning_count = int(item.get("injection_warning_count") or 0)
-            if warning_count:
-                print(f"  warning: {warning_count} injection heuristic hit(s)")
-            for hit in item.get("injection_heuristics") or []:
+            print(
+                f"[{injection_status}] content_guard injection: {item.get('path')} "
+                f"{_injection_detail(warning_count=warning_count)}"
+            )
+            for hit in hits:
                 if hit.get("severity") == "info":
                     print(f"  line {hit['line']}: info: [{hit['rule']}] {hit['excerpt']}")
                 elif hit.get("severity") == "warning":

tests/test_handoff_injection_lint.py

@@ -14,6 +14,28 @@
 BENIGN_FIXTURES = sorted(FIXTURE_DIR.glob("benign-*.md"))
 
 
+def _clean_egress_scan(*args, **kwargs):
+    return {
+        "available": True,
+        "status": "ok",
+        "exit_code": 0,
+        "detail": "clean",
+        "stdout": "",
+        "stderr": "",
+    }
+
+
+def _leaky_egress_scan(*args, **kwargs):
+    return {
+        "available": True,
+        "status": "blocked",
+        "exit_code": 1,
+        "detail": "content-guard reported findings",
+        "stdout": "finding in handoff",
+        "stderr": "",
+    }
+
+
 @pytest.mark.parametrize("fixture_path", EVIL_FIXTURES, ids=lambda path: path.name)
 def test_injection_fixture_flags_warning(fixture_path: Path):
     text = fixture_path.read_text()
@@ -31,51 +53,95 @@ def test_benign_injection_fixture_has_no_warnings(fixture_path: Path):
     assert not warnings, f"benign fixture should not warn: {warnings}"
 
 
-def test_handoff_lint_content_guard_reports_injection_with_line_numbers(tmp_path, capsys, monkeypatch):
+def test_handoff_lint_content_guard_fails_on_injection_laden_egress_clean(tmp_path, capsys, monkeypatch):
     evil = FIXTURE_DIR / "evil-ignore-previous.md"
     path = tmp_path / "evil.md"
     path.write_text(evil.read_text())
 
-    def fake_run_scan(scan_target, *, repo_target=None, policy="public-repo"):
-        return {
-            "available": True,
-            "status": "ok",
-            "exit_code": 0,
-            "detail": "clean",
-            "stdout": "",
-            "stderr": "",
-        }
-
-    monkeypatch.setattr("brigade.scrub.run_scan", fake_run_scan)
-    assert handoff_cmd.lint(target=tmp_path, paths=[path], content_guard=True, json_output=True) == 0
+    monkeypatch.setattr("brigade.scrub.run_scan", _clean_egress_scan)
+    assert handoff_cmd.lint(target=tmp_path, paths=[path], content_guard=True, json_output=True) == 1
     payload = json.loads(capsys.readouterr().out)
+    assert payload["valid"] is False
+    assert payload["content_guard_egress"] == "ok"
+    assert payload["content_guard_injection"] == "fail"
     heuristics = payload["results"][0]["injection_heuristics"]
     assert heuristics
     assert any(item["severity"] == "warning" for item in heuristics)
     assert all("line" in item and item["line"] >= 1 for item in heuristics)
     guard = payload["content_guard"][0]
+    assert guard["egress_verdict"] == "ok"
+    assert guard["injection_verdict"] == "fail"
     assert guard["injection_warning_count"] >= 1
     assert guard["injection_heuristics"]
 
 
+def test_handoff_lint_content_guard_fails_on_egress_leaky_injection_free(tmp_path, capsys, monkeypatch):
+    benign = FIXTURE_DIR / "benign-about-injection.md"
+    path = tmp_path / "benign.md"
+    path.write_text(benign.read_text())
+
+    monkeypatch.setattr("brigade.scrub.run_scan", _leaky_egress_scan)
+    assert handoff_cmd.lint(target=tmp_path, paths=[path], content_guard=True, json_output=True) == 1
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["valid"] is False
+    assert payload["content_guard_egress"] == "fail"
+    assert payload["content_guard_injection"] == "ok"
+    guard = payload["content_guard"][0]
+    assert guard["egress_verdict"] == "fail"
+    assert guard["injection_verdict"] == "ok"
+    assert guard["exit_code"] == 1
+    assert guard["injection_warning_count"] == 0
+
+
+def test_handoff_lint_content_guard_clean_passes_with_both_axes_named(tmp_path, capsys, monkeypatch):
+    benign = FIXTURE_DIR / "benign-about-injection.md"
+    path = tmp_path / "benign.md"
+    path.write_text(benign.read_text())
+
+    monkeypatch.setattr("brigade.scrub.run_scan", _clean_egress_scan)
+    assert handoff_cmd.lint(target=tmp_path, paths=[path], content_guard=True, json_output=True) == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["valid"] is True
+    assert payload["content_guard_egress"] == "ok"
+    assert payload["content_guard_injection"] == "ok"
+    guard = payload["content_guard"][0]
+    assert guard["egress_verdict"] == "ok"
+    assert guard["injection_verdict"] == "ok"
+
+    handoff_cmd.lint(target=tmp_path, paths=[path], content_guard=True)
+    out = capsys.readouterr().out
+    assert "content_guard egress:" in out
+    assert "content_guard injection:" in out
+    assert "[ok] content_guard egress:" in out
+    assert "[ok] content_guard injection:" in out
+
+
+def test_handoff_lint_content_guard_info_only_injection_does_not_fail(tmp_path, capsys, monkeypatch):
+    benign = FIXTURE_DIR / "benign-quoted-payload.md"
+    path = tmp_path / "benign.md"
+    path.write_text(benign.read_text())
+
+    monkeypatch.setattr("brigade.scrub.run_scan", _clean_egress_scan)
+    assert handoff_cmd.lint(target=tmp_path, paths=[path], content_guard=True, json_output=True) == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["valid"] is True
+    assert payload["content_guard_injection"] == "ok"
+    guard = payload["content_guard"][0]
+    assert guard["injection_verdict"] == "ok"
+    assert guard["injection_warning_count"] == 0
+    assert any(hit["severity"] == "info" for hit in guard["injection_heuristics"])
+
+
 def test_handoff_lint_content_guard_prints_injection_scope(tmp_path, capsys, monkeypatch):
     evil = FIXTURE_DIR / "evil-disregard-system.md"
     path = tmp_path / "evil.md"
     path.write_text(evil.read_text())
 
-    monkeypatch.setattr(
-        "brigade.scrub.run_scan",
-        lambda *args, **kwargs: {
-            "available": True,
-            "status": "ok",
-            "exit_code": 0,
-            "detail": "clean",
-            "stdout": "",
-            "stderr": "",
-        },
-    )
+    monkeypatch.setattr("brigade.scrub.run_scan", _clean_egress_scan)
     handoff_cmd.lint(target=tmp_path, paths=[path], content_guard=True)
     out = capsys.readouterr().out
-    assert "leak scan + injection heuristics" in out
+    assert "egress leak scan + injection heuristics" in out
+    assert "[ok] content_guard egress:" in out
+    assert "[fail] content_guard injection:" in out
     assert "line " in out
     assert "disregard-system-prompt" in out or "classic-injection" in out