From 02f05d553030d2122e609370798feaa6912c0723 Mon Sep 17 00:00:00 2001 From: Tanishq Date: Fri, 10 Jul 2026 06:40:31 -0700 Subject: [PATCH] feat: implement generic mitigation-aware rule evaluation (fixes #21) --- DOCS/research/mitigation_evaluation.md | 44 +++++++++++++ anchor/cli.py | 1 + anchor/core/engine.py | 37 ++++++++++- anchor/core/loader.py | 2 + benchmarks/bench_anchor.py | 4 ++ tests/unit/test_mitigations.py | 91 ++++++++++++++++++++++++++ 6 files changed, 177 insertions(+), 2 deletions(-) create mode 100644 DOCS/research/mitigation_evaluation.md create mode 100644 tests/unit/test_mitigations.py diff --git a/DOCS/research/mitigation_evaluation.md b/DOCS/research/mitigation_evaluation.md new file mode 100644 index 0000000..65506e5 --- /dev/null +++ b/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 ` +* `# anchor: mitigation ` +* `# anchor: mitigated ` + +Where `` 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) +``` diff --git a/anchor/cli.py b/anchor/cli.py index fa2ca29..4006548 100644 --- a/anchor/cli.py +++ b/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: diff --git a/anchor/core/engine.py b/anchor/core/engine.py index 674ef42..f60d0cf 100644 --- a/anchor/core/engine.py +++ b/anchor/core/engine.py @@ -563,7 +563,8 @@ 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: @@ -571,12 +572,44 @@ def evaluate_candidates(self, candidates: List[Dict[str, Any]], content: bytes, 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", diff --git a/anchor/core/loader.py b/anchor/core/loader.py index 1a12228..73cb095 100644 --- a/anchor/core/loader.py +++ 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 diff --git a/benchmarks/bench_anchor.py b/benchmarks/bench_anchor.py index 1a82869..e7029bd 100644 --- a/benchmarks/bench_anchor.py +++ b/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") diff --git a/tests/unit/test_mitigations.py b/tests/unit/test_mitigations.py new file mode 100644 index 0000000..24fc7f9 --- /dev/null +++ b/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 +