Feature/red team remediation

.anchor/mitigation.anchor

@@ -53,24 +53,73 @@ mitigations:
   # --- SEC-007: Shell Injection (os-level) ---
   - id: "MIT-014-A"
     rule_id: "SEC-007"
-    name: "Shell Command Execution"
+    name: "Shell Command Execution (os module)"
     match:
-      type: "regex"
-      # Simplified to match os.system/popen anywhere in the line, ignoring preceding code
-      pattern: >-
-        \bos\.(system|popen|spawn)\s*\(
-    message: "Potential shell injection via os.system detects. Use subprocess with list arguments instead."
+      type: "function_call"
+      # AST: matches os.system(...), os.popen(...), os.spawn*(...) call nodes only.
+      # Regex previously matched inside string literals and comments.
+      name: "os.system"
+    message: "Potential shell injection via os.system. Use subprocess with list arguments instead."
+    severity: "blocker"
+
+  # SEC-007 supplementary — os.popen
+  - id: "MIT-014-A2"
+    rule_id: "SEC-007"
+    name: "Shell Command Execution (os.popen)"
+    match:
+      type: "function_call"
+      name: "os.popen"
+    message: "Potential shell injection via os.popen. Use subprocess with list arguments instead."
+    severity: "blocker"
+
+  # SEC-007 supplementary — os.spawn*
+  - id: "MIT-014-A3"
+    rule_id: "SEC-007"
+    name: "Shell Command Execution (os.spawn)"
+    match:
+      type: "function_call"
+      name: "os.spawnl"
+    message: "Potential shell injection via os.spawn family. Use subprocess with list arguments instead."
     severity: "blocker"
 
   # --- SEC-007: Shell Injection (subprocess-level) ---
   - id: "MIT-014-B"
     rule_id: "SEC-007"
-    name: "Unsandboxed Subprocess in Agent"
+    name: "Unsandboxed Subprocess (subprocess.run)"
     match:
-      type: "regex"
-      # Excludes occurrences inside string literals or comments
-      pattern: >-
-        ^(?:[^"\'#]|(["\'])(?:(?!\1).|\\\1)*\1)*\bsubprocess\.(run|call|Popen|check_output)\s*\(
+      type: "function_call"
+      # AST: matches only real subprocess.run() call nodes, not string mentions.
+      name: "subprocess.run"
+    message: "Native subprocess execution detected. Use Diamond Cage (WASM) sandboxing for agent tools."
+    severity: "blocker"
+
+  # SEC-007 supplementary — subprocess.call
+  - id: "MIT-014-B2"
+    rule_id: "SEC-007"
+    name: "Unsandboxed Subprocess (subprocess.call)"
+    match:
+      type: "function_call"
+      name: "subprocess.call"
+    message: "Native subprocess execution detected. Use Diamond Cage (WASM) sandboxing for agent tools."
+    severity: "blocker"
+
+  # SEC-007 supplementary — subprocess.Popen
+  - id: "MIT-014-B3"
+    rule_id: "SEC-007"
+    name: "Unsandboxed Subprocess (subprocess.Popen)"
+    match:
+      type: "function_call"
+      name: "subprocess.Popen"
+    message: "Native subprocess execution detected. Use Diamond Cage (WASM) sandboxing for agent tools."
+    severity: "blocker"
+
+  # SEC-007 supplementary — subprocess.check_output
+  - id: "MIT-014-B4"
+    rule_id: "SEC-007"
+    name: "Unsandboxed Subprocess (subprocess.check_output)"
+    match:
+      type: "function_call"
+      name: "subprocess.check_output"
     message: "Native subprocess execution detected. Use Diamond Cage (WASM) sandboxing for agent tools."
     severity: "blocker"
 
@@ -82,7 +131,7 @@ mitigations:
       type: "regex"
       # Only fire on bulk access or sensitive key names
       pattern: >-
-        ^(?:[^"\'#]|(["\'])(?:(?!\1).|\\\1)*\1)*\bos\.(environ\.(copy|items)\(\)|\benviron\b\s*\[.*(?i)(TOKEN|KEY|SECRET|PASSWORD|CREDENTIAL|API).*\]|\{\*\*os\.environ)
+        (?i)^(?:[^"\'#]|(["'])(?:(?!\1).|\\\\1)*\1)*\bos\.(environ\.(copy|items)\(\)|\benviron\b\s*\[.*(TOKEN|KEY|SECRET|PASSWORD|CREDENTIAL|API).*\]|\{\*\*os\.environ)
     message: "Broad environment variable access detected. Agents may harvest secrets from env."
     severity: "error"
   # --- SEC-012: Governance Configuration Modification ---
@@ -96,3 +145,55 @@ mitigations:
         \.(write|save|update)\s*\(.*\.anchor["']
     message: "Unauthorized attempt to modify Anchor governance configuration detected. Governance state must be immutable at runtime."
     severity: "blocker"
+
+  # --- SEC-004: Hardcoded Secrets (Mitigation Catalog Extension) ---
+  - id: "MIT-004-B"
+    rule_id: "SEC-004"
+    name: "Hardcoded API Key / Secret"
+    match:
+      type: "regex"
+      pattern: >-
+        (?i)(?:[a-zA-Z0-9_]*key|[a-zA-Z0-9_]*secret|[a-zA-Z0-9_]*token|password|credential)\s*=\s*['"][a-zA-Z0-9_\-\.\\:\/]{16,}['"]
+    message: "Hardcoded API key, secret, token or credential detected."
+    severity: "error"
+
+  # --- SEC-010: Insecure Python Execution (AST-based) ---
+  - id: "MIT-010-A"
+    rule_id: "SEC-010"
+    name: "Insecure Dynamic Execution (eval)"
+    match:
+      type: "function_call"
+      # AST: matches only real eval() call nodes, not mentions in strings/comments.
+      name: "eval"
+    message: "Dynamic python execution sink (eval) detected. Use restricted sandboxes instead."
+    severity: "error"
+
+  # SEC-010 supplementary — exec
+  - id: "MIT-010-A2"
+    rule_id: "SEC-010"
+    name: "Insecure Dynamic Execution (exec)"
+    match:
+      type: "function_call"
+      name: "exec"
+    message: "Dynamic python execution sink (exec) detected. Use restricted sandboxes instead."
+    severity: "error"
+
+  # SEC-010 supplementary — pickle.loads / pickle.load
+  - id: "MIT-010-A3"
+    rule_id: "SEC-010"
+    name: "Insecure Deserialization (pickle.loads)"
+    match:
+      type: "function_call"
+      name: "pickle.loads"
+    message: "Unsafe deserialization via pickle.loads detected. Never deserialize untrusted data with pickle."
+    severity: "error"
+
+  # SEC-010 supplementary — marshal.loads
+  - id: "MIT-010-A4"
+    rule_id: "SEC-010"
+    name: "Insecure Deserialization (marshal.loads)"
+    match:
+      type: "function_call"
+      name: "marshal.loads"
+    message: "Unsafe deserialization via marshal.loads detected."
+    severity: "error"

anchor/cli.py

@@ -986,13 +986,35 @@ def check(ctx, policy, paths, dir, model, metadata, context, server_mode, genera
                             and existing_match.get("pattern")
                             and match_data.get("pattern")
                         ):
-                            combined = (
-                                f"(?:{existing_match['pattern']})"
-                                f"|"
-                                f"(?:{match_data['pattern']})"
-                            )
+                            p1 = existing_match['pattern']
+                            p2 = match_data['pattern']
+                            case_insensitive = False
+                            if p1.startswith("(?i)"):
+                                p1 = p1[4:]
+                                case_insensitive = True
+                            if p2.startswith("(?i)"):
+                                p2 = p2[4:]
+                                case_insensitive = True
+
+                            combined = f"(?:{p1})|(?:{p2})"
+                            if case_insensitive:
+                                combined = "(?i)" + combined
+
                             rule_dict[resolved_id]["match"] = {"type": "regex", "pattern": combined}
                             rule_dict[resolved_id]["pattern"] = combined
+                        elif (
+                            existing_match
+                            and existing_match.get("type") == "function_call"
+                            and match_data.get("type") == "function_call"
+                        ):
+                            existing_names = existing_match.get("name", [])
+                            if isinstance(existing_names, str):
+                                existing_names = [existing_names]
+                            new_names = match_data.get("name", [])
+                            if isinstance(new_names, str):
+                                new_names = [new_names]
+                            combined_names = sorted(list(set(existing_names + new_names)))
+                            rule_dict[resolved_id]["match"] = {"type": "function_call", "name": combined_names}
                         else:
                             rule_dict[resolved_id]["match"] = match_data
                             # If the match block has a top-level pattern (V3 style) or internal

anchor/core/constitution.py

@@ -21,7 +21,7 @@
 
 # SHA-256 of the official legacy files (optional in V3).
 CONSTITUTION_SHA256 = "46539F9D069801C3ACA161696E9AF264299F8B578A510FEA3A28318A58ADC136"
-MITIGATION_SHA256 = "80489726A05A31933EA1448FA1E4AFB6E27A8B628E40EFD73B99535E1283DA32"
+MITIGATION_SHA256 = "EB519BD8B63DC9E033E8314049E46A2A590FD475513C89702664E140621EB75F"
 
 
 # =============================================================================

anchor/core/engine.py

@@ -364,7 +364,10 @@ def scan_file(self, content: bytes, file_path: str, adapter: LanguageAdapter) ->
 
                     # Map High-Level Intent to Adapter Implementation
                     if rule_type == "function_call":
-                        s_expr = adapter.build_dangerous_call_query([match_config.get("name")])
+                        names = match_config.get("name", [])
+                        if isinstance(names, str):
+                            names = [names]
+                        s_expr = adapter.build_dangerous_call_query(names)
                     elif rule_type == "import":
                         s_expr = adapter.build_import_query([match_config.get("module")])
                     elif rule_type == "inheritance":
@@ -413,7 +416,7 @@ def scan_file(self, content: bytes, file_path: str, adapter: LanguageAdapter) ->
 
                         matches = self._execute_query(tree.root_node, adapter, s_expr)
                         if self.verbose:
-                            click.secho(f"    🔢 Raw Matches: {len(matches)}", fg="white", dim=True)
+                            click.secho(f"    [AST] Raw Matches: {len(matches)}", fg="white", dim=True)
 
                         for match_data in matches:
                             # 1. Selection Layer: Find the violation node
@@ -443,7 +446,9 @@ def scan_file(self, content: bytes, file_path: str, adapter: LanguageAdapter) ->
                                 if hasattr(f_text, "decode"): f_text = f_text.decode('utf-8', errors='ignore')
 
                                 # Check against the exact name or list of names
-                                expected_names = [match_config.get("name")] if "name" in match_config else []
+                                expected_names = match_config.get("name", [])
+                                if isinstance(expected_names, str):
+                                    expected_names = [expected_names]
                                 if f_text not in expected_names:
                                     is_valid = False
 
@@ -457,8 +462,8 @@ def scan_file(self, content: bytes, file_path: str, adapter: LanguageAdapter) ->
                                     is_valid = False
 
                             if not is_valid:
-                                if self.verbose: 
-                                    click.echo(f"    ⏩ [FILTERED] False positive match for {rule['id']}")
+                                if self.verbose:
+                                    click.echo(f"    >> [FILTERED] False positive match for {rule['id']}")
                                 continue
 
                             line_num = v_node.start_point[0] + 1
@@ -480,7 +485,7 @@ def scan_file(self, content: bytes, file_path: str, adapter: LanguageAdapter) ->
                                             author = self._get_suppression_author(file_path, line_num)
                                             if self.verbose: 
                                                 import click
-                                                click.echo(f"    🙈 [SUPPRESSED] Rule {rule['id']} at line {line_num} (Author: {author})")
+                                                click.echo(f"    [SUPPRESSED] Rule {rule['id']} at line {line_num} (Author: {author})")
 
                                             suppressed.append({
                                                 "id": rule["id"],
@@ -524,7 +529,7 @@ def scan_file(self, content: bytes, file_path: str, adapter: LanguageAdapter) ->
                 except Exception as e:
                     if self.verbose:
                         import click
-                        click.secho(f"    [!]️  Rule Error ({rule.get('id', 'unknown')}): {e}", fg="yellow", dim=True)
+                        click.secho(f"    [!]  Rule Error ({rule.get('id', 'unknown')}): {e}", fg="yellow", dim=True)
                     pass
 
             # --- MODE B: Regex (Fallback) ---
@@ -604,7 +609,7 @@ def evaluate_candidates(self, candidates: List[Dict[str, Any]], content: bytes,
                 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)
+                        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 ---
@@ -632,7 +637,7 @@ def evaluate_candidates(self, candidates: List[Dict[str, Any]], content: bytes,
                 if has_validation:
                     if self.verbose:
                         import click
-                        click.secho(f"    🛡️ [EVALUATOR] ALN-001 candidate at line {candidate['line']} discarded (validation markers found).", fg="green", dim=True)
+                        click.secho(f"    [EVALUATOR] ALN-001 candidate at line {candidate['line']} discarded (validation markers found).", fg="green", dim=True)
                     continue
 
             # Specialized evaluator for Process Execution (SEC-007, FINOS-014, OWASP-002, RBI-018)
@@ -710,12 +715,25 @@ def evaluate_candidates(self, candidates: List[Dict[str, Any]], content: bytes,
                 has_ai_vars = any(re.search(rf"\b{kw}\w*\b", content_str, re.IGNORECASE) for kw in ai_keywords)
                 has_ai_influence = has_ai_import or has_ai_vars
 
-                # 3. Escalate severity only if dynamic AND AI influence exists
+                # 3. Clamp severity: only downgrade if static or no AI influence exists,
+                # but NEVER below the rule's declared min_severity floor.
+                # SEC-007 (and friends) declare min_severity: "error", so a static
+                # subprocess call must floor at "error", not silently drop to "warning".
+                _SEV_ORDER = {"blocker": 3, "error": 2, "warning": 1, "ignore": 0}
                 if not (not is_static and has_ai_influence):
-                    candidate["severity"] = "warning"
+                    target_sev = "warning"
+                    # Enforce min_severity floor from each matching rule config
+                    for cid in candidate_ids:
+                        for r in getattr(self, "all_rules", self.rules):
+                            if r.get("id") == cid:
+                                floor = r.get("min_severity", "warning")
+                                if _SEV_ORDER.get(floor, 0) > _SEV_ORDER.get(target_sev, 0):
+                                    target_sev = floor
+                                break
+                    candidate["severity"] = target_sev
                     if self.verbose:
                         import click
-                        click.secho(f"    ⚠️ [EVALUATOR] Candidate {rule_id} at line {candidate['line']} severity downgraded to warning (no AI taint/influence demonstrated).", fg="yellow", dim=True)
+                        click.secho(f"    [EVALUATOR] Candidate {rule_id} at line {candidate['line']} severity clamped to {target_sev} (no AI taint/influence demonstrated).", fg="yellow", dim=True)
 
             violations.append(candidate)
         return violations
@@ -736,7 +754,7 @@ def _execute_query(self, root_node, adapter: LanguageAdapter, s_expr: str) -> Li
             except:
                 query_obj = Query(language, s_expr)
         except Exception as e:
-            if self.verbose: print(f"    ❗ Query Syntax Error: {e}")
+            if self.verbose: print(f"    [!] Query Syntax Error: {e}")
             return []
 
         results = []
@@ -813,7 +831,7 @@ def _execute_query(self, root_node, adapter: LanguageAdapter, s_expr: str) -> Li
                     results.append(match_data)
         except Exception as e:
             if self.verbose: 
-                print(f"    ❗ Execution Error: {e}")
+                print(f"    [!] Execution Error: {e}")
                 import traceback
                 traceback.print_exc()