fix(handoff): flag injection payloads in content-guard lint (#477)

Extend handoff lint --content-guard with line-numbered injection heuristics
alongside the leak scan, add a fixture corpus, and document the broader scope.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: solomonneas

docs/security.md

@@ -10,7 +10,7 @@ Content Guard is Brigade's publish and memory-safety scanner. Brigade shells out
 
 Use it in three places:
 
-- `brigade handoff lint --content-guard --guard-policy personal` checks pending handoffs before memory ingest.
+- `brigade handoff lint --content-guard --guard-policy personal` checks pending handoffs before memory ingest. The flag runs content-guard for secret and identity leaks and Brigade injection heuristics for instruction-shaped payloads in handoff bodies (for example override phrases, fake system blocks, and base64-decode chains). Injection hits are reported as line-numbered warnings; benign discussion of prompt injection may appear as info-level notes.
 - `brigade handoff draft --guard --guard-policy personal ...` writes a draft and returns failure if Content Guard blocks it.
 - `brigade work import content-guard --policy public-repo` runs a scan and turns blocking findings into reviewable work imports.
 

src/brigade/cli/handoff.py

@@ -45,7 +45,9 @@ def register(sub: argparse._SubParsersAction) -> None:
     )
     p_handoff_lint.add_argument("--target", "-t", type=Path, default=Path("."), help="Repo or workspace to inspect.")
     p_handoff_lint.add_argument(
-        "--content-guard", action="store_true", help="Also scan handoff files with content-guard."
+        "--content-guard",
+        action="store_true",
+        help="Run content-guard leak scan plus handoff injection heuristics (secrets/PII and instruction-shaped payloads).",
     )
     p_handoff_lint.add_argument(
         "--guard-policy", default="personal", help="Content Guard policy name or path for --content-guard."

src/brigade/handoff_cmd/linting.py

@@ -24,6 +24,30 @@
 globals().update({name: value for name, value in vars(_family_base).items() if not name.startswith("__")})
 
 
+def _injection_hit_dict(hit: Any) -> dict[str, Any]:
+    return {
+        "line": hit.line,
+        "severity": hit.severity,
+        "rule": hit.rule,
+        "excerpt": hit.excerpt,
+    }
+
+
+def _injection_messages(hits: tuple[Any, ...]) -> tuple[str, ...]:
+    messages: list[str] = []
+    for hit in hits:
+        prefix = "info" if hit.severity == "info" else "warning"
+        messages.append(f"line {hit.line}: {prefix}: [{hit.rule}] {hit.excerpt}")
+    return tuple(messages)
+
+
+def _read_handoff_text(path: Path) -> str | None:
+    try:
+        return path.read_text(errors="replace")
+    except OSError:
+        return None
+
+
 def lint(
     *,
     target: Path,
@@ -36,29 +60,55 @@ def lint(
     if not target.is_dir():
         print(f"error: --target is not a directory: {target}", file=sys.stderr)
         return 2
-    from ..untrusted import scan_untrusted
+    from ..untrusted import scan_handoff_injection_heuristics, scan_untrusted
 
     results = lint_targets(target, paths=paths)
-    guard_results = (
-        [_guard_handoff_path(path, target=target, policy=guard_policy) for path in [result.path for result in results]]
-        if content_guard
-        else []
-    )
+    guard_results: list[dict[str, Any]] = []
+    if content_guard:
+        for result in results:
+            guard_item = _guard_handoff_path(result.path, target=target, policy=guard_policy)
+            text = _read_handoff_text(result.path)
+            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_results.append(guard_item)
     guard_ok = all(item.get("exit_code") == 0 for item in guard_results)
-    # Content-guard checks egress (secrets/PII), not instructions. Surface the
-    # injection signal here too so a poisoned note never reads as fully clean.
     injection_counts: dict[str, int] = {}
+    injection_hits_by_path: dict[str, tuple[Any, ...]] = {}
+    enriched_results: list[HandoffLintResult] = []
     for result in results:
-        try:
-            signal = scan_untrusted(result.path.read_text(errors="replace"))
-        except OSError:
+        text = _read_handoff_text(result.path)
+        if text is None:
+            enriched_results.append(result)
             continue
+        hits = scan_handoff_injection_heuristics(text)
+        injection_hits_by_path[str(result.path)] = hits
+        signal = scan_untrusted(text)
         if signal.flagged:
             injection_counts[str(result.path)] = signal.count
+        injection_messages = _injection_messages(hits) if content_guard or hits else ()
+        if not injection_messages and signal.flagged:
+            injection_messages = (
+                f"line ?: warning: [{signal.count} prompt-injection signal(s); see `brigade security scan`]",
+            )
+        enriched_results.append(
+            HandoffLintResult(
+                path=result.path,
+                action=result.action,
+                valid=result.valid,
+                errors=result.errors,
+                warnings=result.warnings + injection_messages,
+                hints=result.hints,
+            )
+        )
+    results = tuple(enriched_results)
     result_dicts = []
     for result in results:
         row = result.as_dict()
-        row["injection_signals"] = injection_counts.get(str(result.path), 0)
+        path_key = str(result.path)
+        row["injection_signals"] = injection_counts.get(path_key, 0)
+        hits = injection_hits_by_path.get(path_key, ())
+        row["injection_heuristics"] = [_injection_hit_dict(hit) for hit in hits]
         result_dicts.append(row)
     payload = {
         "target": str(target),
@@ -82,17 +132,23 @@ def lint(
         for hint in result.hints:
             print(f"  hint: {hint}")
         for warning in result.warnings:
-            print(f"  warning: {warning}")
-        signals = injection_counts.get(str(result.path), 0)
-        if signals:
-            print(
-                f"  warning: {signals} prompt-injection signal(s); content-guard does not check this, see `brigade security scan`"
-            )
+            if warning.startswith("line "):
+                print(f"  {warning}")
+            else:
+                print(f"  warning: {warning}")
     if content_guard:
-        print(f"content_guard_policy: {guard_policy}")
+        print(f"content_guard_policy: {guard_policy} (leak scan + injection heuristics)")
         for item in guard_results:
-            status = OK if item.get("exit_code") == 0 else FAIL
-            print(f"[{status}] content_guard: {item.get('path')} {item.get('detail')}")
+            leak_status = OK if item.get("exit_code") == 0 else FAIL
+            print(f"[{leak_status}] content_guard leaks: {item.get('path')} {item.get('detail')}")
+            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 []:
+                if hit.get("severity") == "info":
+                    print(f"  line {hit['line']}: info: [{hit['rule']}] {hit['excerpt']}")
+                elif hit.get("severity") == "warning":
+                    print(f"  line {hit['line']}: warning: [{hit['rule']}] {hit['excerpt']}")
     return 0 if payload["valid"] else 1
 
 

src/brigade/templates/policies/personal.json

@@ -2,7 +2,7 @@
   "_comment": [
     "Personal policy: for local working notes and memory handoffs before ingest.",
     "It blocks secrets and attribution trailers, while warning on personal or infrastructure-like context.",
-    "Use via: brigade handoff lint --content-guard --guard-policy personal"
+    "Use via: brigade handoff lint --content-guard --guard-policy personal (leak scan + injection heuristics)"
   ],
   "_brigade_version": "0.25.1",
   "categories": {

src/brigade/untrusted.py

@@ -14,7 +14,7 @@
 import hashlib
 import re
 from dataclasses import dataclass
-from typing import List, Optional
+from typing import Iterable, List, Optional
 
 _INJECTION_PATTERNS = (
     "ig" + "nore (all )?(previous|prior) instructions",
@@ -82,27 +82,150 @@ def wrap_untrusted(
     return "\n".join(parts)
 
 
+@dataclass(frozen=True)
+class InjectionHit:
+    line: int
+    severity: str
+    rule: str
+    excerpt: str
+
+
 @dataclass
 class InjectionSignal:
     flagged: bool
     count: int
     markers: List[str]
 
 
+_LINE_RULES: tuple[tuple[str, re.Pattern[str]], ...] = (
+    ("classic-injection", PROMPT_INJECTION_RE),
+    ("disregard-system-prompt", re.compile(r"(?i)disregard (your |the )?system prompt")),
+    (
+        "ignore-instructions",
+        re.compile(r"(?i)ignore (all |any )?(previous|prior|above) (instructions|directives)"),
+    ),
+    ("fake-system-block", re.compile(r"(?i)(<\s*/?\s*system\b|\[INST\]|\[/INST\]|<<\s*SYS\s*>>)")),
+    ("role-override", re.compile(r"(?i)\byou are now\b")),
+    (
+        "assistant-directive",
+        re.compile(
+            r"(?i)(?:^(?:assistant|agent|ai)\s*[,:-]\s*(?:you )?(?:must|should|need to)\b"
+            r"|(?:dear|hey) (?:assistant|agent|ai)\b.*\b(?:ignore|disregard|override)\b)"
+        ),
+    ),
+)
+
+_BASE64_BLOB = re.compile(r"[A-Za-z0-9+/]{80,}={0,2}")
+_DECODE_INSTRUCTION = re.compile(r"(?i)\b(?:base64|atob|decode|decrypt)\b")
+
+_BENIGN_LINE_MARKERS = (
+    re.compile(r"(?i)\bprompt[- ]injection\b"),
+    re.compile(r"(?i)\binjection (?:heuristic|signal|scan|detection|mitigation|fixture)\b"),
+    re.compile(r"(?i)\b(?:example|documented|detected|benign|fixture|quoted|pattern|mitigation)\b"),
+    re.compile(r"(?i)\b(?:scans?|checks?) for\b"),
+)
+
+
+def _excerpt(line: str) -> str:
+    return line.strip()[:_MARKER_MAX]
+
+
+def _benign_injection_discussion(line: str, *, text: str) -> bool:
+    if any(marker.search(line) for marker in _BENIGN_LINE_MARKERS):
+        return True
+    if "`" in line and any(token in line.lower() for token in ("ignore", "disregard", "<system", "[inst]")):
+        return True
+    if re.search(r'(?i)["\'].*(?:ignore|disregard).*(?:instructions|system prompt).*["\']', line):
+        return True
+    if "prompt injection" in text.lower() and re.search(r"(?i)\b(?:issue|handoff|heuristic|#)\b", line):
+        return True
+    return False
+
+
+def _line_hits(line: str, line_number: int, *, text: str) -> list[InjectionHit]:
+    hits: list[InjectionHit] = []
+    seen_rules: set[str] = set()
+    for rule_id, pattern in _LINE_RULES:
+        if not pattern.search(line):
+            continue
+        severity = "info" if _benign_injection_discussion(line, text=text) else "warning"
+        if rule_id in seen_rules:
+            continue
+        seen_rules.add(rule_id)
+        hits.append(InjectionHit(line=line_number, severity=severity, rule=rule_id, excerpt=_excerpt(line)))
+    return hits
+
+
+def _cross_line_hits(text: str) -> list[InjectionHit]:
+    normalized = re.sub(r"\s+", " ", text)
+    if not PROMPT_INJECTION_RE.search(normalized):
+        return []
+    for _line_number, line in enumerate(text.splitlines(), start=1):
+        if PROMPT_INJECTION_RE.search(line):
+            return []
+    start = PROMPT_INJECTION_RE.search(normalized)
+    if not start:
+        return []
+    excerpt = normalized[start.start() :].strip()[:_MARKER_MAX]
+    severity = "info" if _benign_injection_discussion(excerpt, text=text) else "warning"
+    return [InjectionHit(line=1, severity=severity, rule="classic-injection", excerpt=excerpt)]
+
+
+def _base64_decode_hits(lines: list[str]) -> list[InjectionHit]:
+    hits: list[InjectionHit] = []
+    for index, line in enumerate(lines):
+        if not _BASE64_BLOB.search(line):
+            continue
+        window = "\n".join(lines[index : index + 4])
+        if not _DECODE_INSTRUCTION.search(window):
+            continue
+        line_number = index + 1
+        severity = "info" if _benign_injection_discussion(line, text=window) else "warning"
+        hits.append(
+            InjectionHit(
+                line=line_number,
+                severity=severity,
+                rule="base64-decode-chain",
+                excerpt=_excerpt(line),
+            )
+        )
+    return hits
+
+
+def scan_handoff_injection_heuristics(content: str) -> tuple[InjectionHit, ...]:
+    """Scan handoff bodies for instruction-shaped injection payloads."""
+    text = content if isinstance(content, str) else ""
+    lines = text.splitlines()
+    hits: list[InjectionHit] = []
+    seen: set[tuple[int, str]] = set()
+    for line_number, line in enumerate(lines, start=1):
+        for hit in _line_hits(line, line_number, text=text):
+            key = (hit.line, hit.rule)
+            if key in seen:
+                continue
+            seen.add(key)
+            hits.append(hit)
+    for hit in _cross_line_hits(text):
+        key = (hit.line, hit.rule)
+        if key not in seen:
+            seen.add(key)
+            hits.append(hit)
+    for hit in _base64_decode_hits(lines):
+        key = (hit.line, hit.rule)
+        if key in seen:
+            continue
+        seen.add(key)
+        hits.append(hit)
+    return tuple(hits)
+
+
+def _warning_hits(hits: Iterable[InjectionHit]) -> list[InjectionHit]:
+    return [hit for hit in hits if hit.severity == "warning"]
+
+
 def scan_untrusted(content: str) -> InjectionSignal:
     """Report whether `content` carries injection-style instructions."""
-    text = content if isinstance(content, str) else ""
-    markers: List[str] = []
-    for line in text.splitlines():
-        if PROMPT_INJECTION_RE.search(line):
-            markers.append(line.strip()[:_MARKER_MAX])
-    # Per-line matching alone is evadable by splitting a phrase across newlines
-    # ("ignore all\nprevious instructions"). Scan a whitespace-normalized copy
-    # too so a cross-line phrase is still caught; only add a marker if the
-    # per-line pass missed it, to avoid double-counting single-line hits.
-    if not markers:
-        normalized = re.sub(r"\s+", " ", text)
-        m = PROMPT_INJECTION_RE.search(normalized)
-        if m:
-            markers.append(normalized[m.start() :].strip()[:_MARKER_MAX])
-    return InjectionSignal(flagged=bool(markers), count=len(markers), markers=markers)
+    hits = scan_handoff_injection_heuristics(content)
+    warnings = _warning_hits(hits)
+    markers = [hit.excerpt for hit in warnings]
+    return InjectionSignal(flagged=bool(warnings), count=len(warnings), markers=markers)

tests/fixtures/handoff_lint/injection/benign-about-injection.md

@@ -0,0 +1,21 @@
+# Memory Handoff
+
+## Type
+decision
+
+## Title
+Add handoff injection heuristics for issue 477
+
+## Summary
+Document the decision to scan handoffs for prompt-injection payloads during content-guard lint.
+
+## Recommended memory action
+no-card
+
+## Target document
+.learnings/LEARNINGS.md
+
+## Suggested document content
+### Add handoff injection heuristics for issue 477
+
+We added injection heuristics to `brigade handoff lint --content-guard` so instruction-shaped payloads are flagged with line numbers before ingest.

tests/fixtures/handoff_lint/injection/benign-instruction-docs.md

@@ -0,0 +1,21 @@
+# Memory Handoff
+
+## Type
+workflow
+
+## Title
+Normal user instruction routing
+
+## Summary
+Describe how explicit user instructions should flow through ingest without injection heuristics firing.
+
+## Recommended memory action
+no-card
+
+## Target document
+.learnings/LEARNINGS.md
+
+## Suggested document content
+### Normal user instruction routing
+
+Operators should write explicit user instructions in the handoff summary and evidence sections. The ingest path promotes only lint-valid routes and never auto-executes suggested document content.