feat: implement generic mitigation-aware rule evaluation (fixes #21)

DOCS/research/mitigation_evaluation.md

@@ -0,0 +1,44 @@
+# Mitigation-Aware Rule Evaluation
+
+Anchor's rule engine supports evaluating mitigation status before generating findings. This enables rules to define a required number of mitigations that must be satisfied in the implementation.
+
+Mitigation requirements are declared at the rule level:
+```yaml
+rules:
+  - id: "SEC-007"
+    name: "Shell Command Execution"
+    min_mitigations: 1
+```
+
+## Declaring Mitigations in Source Code
+
+Developers can declare implemented mitigations using inline comments matching one of the following formats:
+* `# anchor: mitigate <id>`
+* `# anchor: mitigation <id>`
+* `# anchor: mitigated <id>`
+
+Where `<id>` is the rule ID (e.g. `SEC-007`) or the mitigation ID (e.g. `MIT-014-A`).
+
+### Example
+
+```python
+import subprocess
+
+# anchor: mitigate SEC-007
+def run_command(args):
+    # This call is safe because args are not shell-evaluated
+    subprocess.run(args, shell=False)
+```
+
+## Engine Evaluation Flow
+
+```
+Evidence Collection (Pattern Matches)
+         ↓
+Parse inline mitigations in target file
+         ↓
+Verify if count of mitigations for rule ID meets min_mitigations
+         ↓
+         ├─ [YES] ── Discard Candidate (Compliant)
+         └─ [NO]  ── Generate Violation (Non-Compliant)
+```

anchor/cli.py

@@ -876,6 +876,7 @@ def check(ctx, policy, paths, dir, model, metadata, context, server_mode, genera
                 "pattern":     rule.pattern,
                 "message":     rule.message,
                 "mitigation":  rule.mitigation,
+                "min_mitigations": rule.min_mitigations,
             }
 
         if verbose or not rule_dict:

anchor/core/engine.py

@@ -563,20 +563,53 @@ def scan_file(self, content: bytes, file_path: str, adapter: LanguageAdapter) ->
     def evaluate_candidates(self, candidates: List[Dict[str, Any]], content: bytes, file_path: str) -> List[Dict[str, Any]]:
         """
         Runs candidate findings through an evidence-based rule evaluation pipeline.
-        Promotes candidates to final violations only if conditions are met.
+        Promotes candidates to final violations only if conditions are met, including
+        mitigation-aware checks and min_mitigations enforcement.
         """
         violations = []
         try:
             content_str = content.decode("utf-8", errors="ignore")
         except Exception:
             content_str = ""
 
+        # Parse active inline mitigations in the file
+        import re
+        mitigation_pattern = r"#\s*anchor:\s*(?:mitigate|mitigation|mitigated)\s+([a-zA-Z0-9_-]+)"
+        active_mitigations = re.findall(mitigation_pattern, content_str)
+
         for candidate in candidates:
             rule_id = candidate.get("id", "")
+            candidate_ids = [cid.strip() for cid in rule_id.split(",") if cid.strip()]
 
+            # --- 1. Generic Mitigation-Aware Evaluation ---
+            # Lookup min_mitigations requirement for the candidate's rule(s)
+            min_mit = 0
+            for cid in candidate_ids:
+                rule_config = None
+                for r in getattr(self, "all_rules", self.rules):
+                    if r.get("id") == cid:
+                        rule_config = r
+                        break
+                if rule_config:
+                    m_val = rule_config.get("min_mitigations")
+                    if m_val is not None:
+                        try:
+                            min_mit = max(min_mit, int(m_val))
+                        except (ValueError, TypeError):
+                            pass
+
+            if min_mit > 0:
+                # Count matching mitigations defined in the file
+                mit_count = sum(1 for m_id in active_mitigations if m_id in candidate_ids)
+                if mit_count >= min_mit:
+                    if self.verbose:
+                        import click
+                        click.secho(f"    🛡️ [EVALUATOR] Candidate {rule_id} at line {candidate['line']} discarded ({mit_count}/{min_mit} mitigations satisfied).", fg="green", dim=True)
+                    continue
+
+            # --- 2. Specialized Rule Evaluators ---
             # Specialized evaluator for ALN-001 / MIT-003-A (LLM Output Without Validation)
             if "ALN-001" in rule_id or "MIT-003-A" in rule_id:
-                import re
                 # Check for common validation frameworks, helper functions, and custom schemas
                 validation_patterns = [
                     r"\bimport\s+(pydantic|instructor|guardrails|marshmallow|jsonschema)\b",

anchor/core/loader.py

@@ -49,6 +49,7 @@ class Rule:
     message: Optional[str] = None
     mitigation: Optional[str] = None
     primitives: Optional[Primitives] = None
+    min_mitigations: Optional[int] = None
 
 
 @dataclass
@@ -260,6 +261,7 @@ def load_domain_file(
             message=rule_data.get("message"),
             mitigation=rule_data.get("mitigation"),
             primitives=primitives,
+            min_mitigations=rule_data.get("min_mitigations"),
         )
         rules[rule_id] = rule
 

benchmarks/bench_anchor.py

@@ -429,6 +429,10 @@ def main() -> None:
     )
     args = parser.parse_args()
 
+    # Disable remote Ledger and Relay URLs to avoid blocking HTTP/socket timeouts in benchmarks
+    os.environ.pop("ANCHOR_LEDGER_URL", None)
+    os.environ.pop("ANCHOR_RELAY_URL", None)
+
     if not args.json:
         print(DIV)
         print("  Anchor Runtime Benchmark")

tests/unit/test_mitigations.py

@@ -0,0 +1,91 @@
+import pytest
+from anchor.core.engine import PolicyEngine
+from anchor.adapters.python import PythonAdapter
+
+def test_min_mitigations_not_satisfied():
+    # File has the violation but NO mitigation comments
+    content = b"""
+def run_process():
+    dangerous_call()
+"""
+    rules = [
+        {
+            "id": "SEC-007",
+            "name": "Shell Command Execution",
+            "match": {
+                "type": "regex",
+                "pattern": r"dangerous_call\("
+            },
+            "min_mitigations": 1,
+            "message": "Shell command execution detected.",
+            "severity": "error"
+        }
+    ]
+    engine = PolicyEngine(config={"rules": rules})
+    adapter = PythonAdapter()
+
+    results = engine.scan_file(content, "test.py", adapter)
+    assert len(results["violations"]) == 1
+
+def test_min_mitigations_satisfied_single():
+    # File has the violation and ONE mitigation comment
+    content = b"""
+# anchor: mitigate SEC-007
+def run_process():
+    dangerous_call()
+"""
+    rules = [
+        {
+            "id": "SEC-007",
+            "name": "Shell Command Execution",
+            "match": {
+                "type": "regex",
+                "pattern": r"dangerous_call\("
+            },
+            "min_mitigations": 1,
+            "message": "Shell command execution detected.",
+            "severity": "error"
+        }
+    ]
+    engine = PolicyEngine(config={"rules": rules})
+    adapter = PythonAdapter()
+
+    results = engine.scan_file(content, "test.py", adapter)
+    assert len(results["violations"]) == 0
+
+def test_min_mitigations_multiple_required():
+    # File has one mitigation comment but needs two
+    content = b"""
+# anchor: mitigate SEC-007
+def run_process():
+    dangerous_call()
+"""
+    rules = [
+        {
+            "id": "SEC-007",
+            "name": "Shell Command Execution",
+            "match": {
+                "type": "regex",
+                "pattern": r"dangerous_call\("
+            },
+            "min_mitigations": 2,
+            "message": "Shell command execution detected.",
+            "severity": "error"
+        }
+    ]
+    engine = PolicyEngine(config={"rules": rules})
+    adapter = PythonAdapter()
+
+    results = engine.scan_file(content, "test.py", adapter)
+    assert len(results["violations"]) == 1
+
+    # Now add a second mitigation comment and verify it passes
+    content_two = b"""
+# anchor: mitigate SEC-007
+# anchor: mitigation SEC-007: Checked arguments
+def run_process():
+    dangerous_call()
+"""
+    results_two = engine.scan_file(content_two, "test.py", adapter)
+    assert len(results_two["violations"]) == 0
+