Remove in-process exec() from red_team_code_review

fable_engine/actions/fleet.py

@@ -190,17 +190,28 @@ def _handle_register_automation_pipeline(arguments: Dict[str, Any]) -> str:
 def _handle_red_team_code_review(arguments: Dict[str, Any]) -> str:
     action = arguments.get("action", "").strip().lower()
     session_name = arguments.get("session_name", "").strip()
-    session: Optional[FableSession] = None
     if not session_name:
         return "Error: 'session_name' is required for action 'red_team_code_review'."
+
     target_name = arguments.get("target_name", "system")
-    code_snippet = arguments.get("target_code") or arguments.get("code_snippet") or arguments.get("code") or ""
+    code_snippet = None
+    for k in ("target_code", "code_snippet", "code", "target_callable"):
+        if k in arguments and arguments[k] is not None:
+            code_snippet = arguments[k]
+            break
+
+    if code_snippet is not None and not callable(code_snippet):
+        return (
+            "Error: Source-code strings cannot be evaluated in-process for security reasons. "
+            "Dynamic source-code execution is disabled for public actions until an isolated sandbox executor is configured."
+        )
+
     custom_hypotheses = arguments.get("custom_hypotheses") or arguments.get("hypotheses")
     output_path = arguments.get("output_path")
 
     session = get_or_load_session(session_name)
     report = _get_swarm().run_full_review_cycle(
-        target_callable=code_snippet if code_snippet else None,
+        target_callable=code_snippet,
         target_name=target_name,
         custom_hypotheses=custom_hypotheses,
     )
@@ -309,9 +320,21 @@ def _handle_record_breakage_report(arguments: Dict[str, Any]) -> str:
 def _handle_verify_red_team_remediation(arguments: Dict[str, Any]) -> str:
     action = arguments.get("action", "").strip().lower()
     session_name = arguments.get("session_name", "").strip()
-    session: Optional[FableSession] = None
     if not session_name:
         return "Error: 'session_name' is required for action 'verify_red_team_remediation'."
+
+    remediated_code = None
+    for k in ("remediated_code", "target_code", "code_snippet", "code", "target_callable"):
+        if k in arguments and arguments[k] is not None:
+            remediated_code = arguments[k]
+            break
+
+    if remediated_code is not None and not callable(remediated_code):
+        return (
+            "Error: Source-code strings cannot be evaluated in-process for security reasons. "
+            "Dynamic source-code execution is disabled for public actions until an isolated sandbox executor is configured."
+        )
+
     session = get_or_load_session(session_name)
 
     report_id = arguments.get("report_id")
@@ -325,8 +348,6 @@ def _handle_verify_red_team_remediation(arguments: Dict[str, Any]) -> str:
 
     if not prior_report:
         return "Error: No prior breakage report found to verify. Provide 'report_id' or 'prior_report'."
-
-    remediated_code = arguments.get("remediated_code") or arguments.get("target_code") or arguments.get("code") or ""
     timeout_sec = float(arguments.get("timeout_seconds", 3.0))
     all_fixed, new_report = _get_swarm().verify_remediation(
         target_callable=remediated_code,

fable_engine/schema.py

@@ -478,11 +478,11 @@
             },
             "target_code": {
                 "type": "string",
-                "description": "Target source code under test for adversarial red-teaming."
+                "description": "Source-code string input. Dynamic target execution from source-code strings is disabled and requires a separate sandboxed executor."
             },
             "code_snippet": {
                 "type": "string",
-                "description": "Alternative alias or snippet for target source code under review."
+                "description": "Alternative alias for source-code string input. Dynamic target execution from source-code strings is disabled and requires a separate sandboxed executor."
             },
             "custom_hypotheses": {
                 "description": "List or JSON string of custom adversarial hypotheses / attack vectors.",
@@ -494,7 +494,7 @@
             },
             "remediated_code": {
                 "type": "string",
-                "description": "Remediated code submitted to verify against broken scenarios."
+                "description": "Remediated source-code string input. Dynamic target execution from source-code strings is disabled and requires a separate sandboxed executor."
             },
             "prior_report": {
                 "description": "Prior red-team breakage report dictionary or JSON string to verify remediation against.",

fable_v2/coder_fleet/red_team_swarm.py

@@ -12,7 +12,6 @@
 import inspect
 import json
 import os
-import sys
 import threading
 import time
 import traceback
@@ -219,7 +218,12 @@ def to_markdown(self) -> str:
 
 
 class RedTeamSwarm:
-    """Adversarial Red Team Swarm executing counterfactual stress probes across 5 vectors."""
+    """Adversarial Red Team Swarm executing counterfactual stress probes across 5 vectors.
+
+    RedTeamSwarm never executes source-code strings in-process. Source-code strings and
+    unexecutable targets are rejected as non-executable targets (passed=False). Dynamic source
+    execution, if required, must be provided through a separately sandboxed execution layer.
+    """
 
     def __init__(
         self,
@@ -244,36 +248,15 @@ def __init__(
                 except Exception:
                     self.plasticity_engine = None
 
-    @staticmethod
-    def _compile_code_to_callable(code: str) -> Callable[..., Any]:
-        """Safely compiles a python code string into an executable callable."""
-        scope: dict[str, Any] = {}
-        exec(code, scope, scope)
-
-        # Look for functions defined in scope
-        functions = [v for v in scope.values() if callable(v) and not isinstance(v, type)]
-        if functions:
-            return functions[-1]
-
-        # Look for classes
-        classes = [v for v in scope.values() if isinstance(v, type)]
-        if classes:
-            return classes[-1]
-
-        def _fallback_callable(*args: Any, **kwargs: Any) -> Any:
-            return scope.get("result", None)
-
-        return _fallback_callable
-
     def _resolve_callable(self, target: Any) -> Optional[Callable[..., Any]]:
-        """Resolves target callable from function, class, or code string."""
+        """Resolves target callable from function or class object.
+
+        RedTeamSwarm never executes source-code strings in-process. Source-code strings
+        are rejected as non-executable targets. Dynamic source execution, if required,
+        must be provided through a separately sandboxed execution layer.
+        """
         if callable(target):
             return target
-        if isinstance(target, str) and target.strip():
-            try:
-                return self._compile_code_to_callable(target)
-            except Exception:
-                return None
         return None
 
     def generate_break_scenarios(
@@ -290,8 +273,8 @@ def generate_break_scenarios(
 
         # Vector 1: CHAOS_ENVIRONMENT
         def _chaos_missing_path(fn: Optional[Callable[..., Any]] = None) -> Any:
-            if not fn:
-                return True
+            if fn is None:
+                raise TypeError("RedTeamSwarm target callable is missing or not executable")
             # Probe with non-existent / unlinked file path
             return fn("/nonexistent/fable_chaos_probe_file.tmp")
 
@@ -307,8 +290,8 @@ def _chaos_missing_path(fn: Optional[Callable[..., Any]] = None) -> Any:
         )
 
         def _chaos_corrupt_env(fn: Optional[Callable[..., Any]] = None) -> Any:
-            if not fn:
-                return True
+            if fn is None:
+                raise TypeError("RedTeamSwarm target callable is missing or not executable")
             return fn("")
 
         scenarios.append(
@@ -324,8 +307,8 @@ def _chaos_corrupt_env(fn: Optional[Callable[..., Any]] = None) -> Any:
 
         # Vector 2: BYZANTINE_PAYLOAD
         def _byzantine_null_bytes(fn: Optional[Callable[..., Any]] = None) -> Any:
-            if not fn:
-                return True
+            if fn is None:
+                raise TypeError("RedTeamSwarm target callable is missing or not executable")
             payload = "probe\x00hostile\x00injection\r\n\t"
             return fn(payload)
 
@@ -341,8 +324,8 @@ def _byzantine_null_bytes(fn: Optional[Callable[..., Any]] = None) -> Any:
         )
 
         def _byzantine_deep_nesting(fn: Optional[Callable[..., Any]] = None) -> Any:
-            if not fn:
-                return True
+            if fn is None:
+                raise TypeError("RedTeamSwarm target callable is missing or not executable")
             # 60 levels of nested dictionaries
             nested: dict[str, Any] = {"leaf": 42}
             for _ in range(60):
@@ -361,8 +344,8 @@ def _byzantine_deep_nesting(fn: Optional[Callable[..., Any]] = None) -> Any:
         )
 
         def _byzantine_type_confusion(fn: Optional[Callable[..., Any]] = None) -> Any:
-            if not fn:
-                return True
+            if fn is None:
+                raise TypeError("RedTeamSwarm target callable is missing or not executable")
             return fn(None)
 
         scenarios.append(
@@ -377,8 +360,8 @@ def _byzantine_type_confusion(fn: Optional[Callable[..., Any]] = None) -> Any:
         )
 
         def _byzantine_extreme_numbers(fn: Optional[Callable[..., Any]] = None) -> Any:
-            if not fn:
-                return True
+            if fn is None:
+                raise TypeError("RedTeamSwarm target callable is missing or not executable")
             # Nan, Inf, negative zero, huge int
             return fn(float("nan"))
 
@@ -395,8 +378,8 @@ def _byzantine_extreme_numbers(fn: Optional[Callable[..., Any]] = None) -> Any:
 
         # Vector 3: CONCURRENCY_RACE
         def _concurrency_multithreaded_burst(fn: Optional[Callable[..., Any]] = None) -> Any:
-            if not fn:
-                return True
+            if fn is None:
+                raise TypeError("RedTeamSwarm target callable is missing or not executable")
             errors: list[str] = []
             threads: list[threading.Thread] = []
             try:
@@ -441,8 +424,8 @@ def _worker() -> None:
 
         # Vector 4: RESOURCE_EXHAUSTION
         def _resource_massive_payload(fn: Optional[Callable[..., Any]] = None) -> Any:
-            if not fn:
-                return True
+            if fn is None:
+                raise TypeError("RedTeamSwarm target callable is missing or not executable")
             massive_str = "A" * 150_000
             return fn(massive_str)
 
@@ -458,8 +441,8 @@ def _resource_massive_payload(fn: Optional[Callable[..., Any]] = None) -> Any:
         )
 
         def _resource_rapid_churn(fn: Optional[Callable[..., Any]] = None) -> Any:
-            if not fn:
-                return True
+            if fn is None:
+                raise TypeError("RedTeamSwarm target callable is missing or not executable")
             sig = inspect.signature(fn)
             for _ in range(100):
                 if len(sig.parameters) == 0:
@@ -481,8 +464,8 @@ def _resource_rapid_churn(fn: Optional[Callable[..., Any]] = None) -> Any:
 
         # Vector 5: STATE_INVARIANT
         def _state_invariant_idempotency(fn: Optional[Callable[..., Any]] = None) -> Any:
-            if not fn:
-                return True
+            if fn is None:
+                raise TypeError("RedTeamSwarm target callable is missing or not executable")
             sig = inspect.signature(fn)
             if len(sig.parameters) == 0:
                 res1 = fn()
@@ -523,8 +506,8 @@ def _state_invariant_idempotency(fn: Optional[Callable[..., Any]] = None) -> Any
                     v = AttackVector.CHAOS_ENVIRONMENT
 
                 def _custom_attack_fn(fn: Optional[Callable[..., Any]] = None, h_text: str = hyp_clean) -> Any:
-                    if not fn:
-                        return True
+                    if fn is None:
+                        raise TypeError("RedTeamSwarm target callable is missing or not executable")
                     sig = inspect.signature(fn)
                     if len(sig.parameters) == 0:
                         return fn()
@@ -589,13 +572,37 @@ def execute_swarm_attack(
         timeout_seconds: float = 3.0,
         target_name: Optional[str] = None,
     ) -> RedTeamBreakageReport:
-        """Executes the break scenarios against target_callable with sandboxed execution.
+        """Executes the break scenarios against target_callable in-process.
 
-        Detects unhandled crashes, memory/resource leaks, invariant violations, and timeouts.
+        Executes trusted callable objects in the current process. Source-code strings are rejected
+        and require a separate sandboxed executor for dynamic execution. Detects unhandled crashes,
+        memory/resource leaks, invariant violations, and timeouts.
         """
         callable_fn = self._resolve_callable(target_callable)
         actual_name = target_name or getattr(callable_fn, "__name__", "target")
 
+        if callable_fn is None:
+            report_id = f"redteam_{int(time.time())}_{uuid.uuid4().hex[:6]}"
+            finding = BreakFinding(
+                scenario_id="target_not_executable",
+                vector=AttackVector.CHAOS_ENVIRONMENT.value,
+                hypothesis="Target must be an executable Python Callable object",
+                broken=True,
+                error_message="TypeError: Source-code strings cannot be evaluated in-process for security reasons. Provide an executable Python Callable or use an isolated sandbox executor.",
+                severity="CRITICAL",
+                details={"target_executable": False},
+            )
+            return RedTeamBreakageReport(
+                report_id=report_id,
+                target_name=actual_name,
+                total_probes=1,
+                broken_count=1,
+                passed=False,
+                findings=[finding],
+                created_at=datetime.datetime.now(datetime.timezone.utc).isoformat(),
+                remediation_directives=["Provide an executable Python callable or use an isolated sandbox executor."],
+            )
+
         if not scenarios:
             scenarios = self.generate_break_scenarios(target_name=actual_name)
 
@@ -808,8 +815,8 @@ def verify_remediation(
                     vec = AttackVector.CHAOS_ENVIRONMENT
 
                 def _repro_probe(fn: Optional[Callable[..., Any]] = None, finding: BreakFinding = f) -> Any:
-                    if not fn:
-                        return True
+                    if fn is None:
+                        raise TypeError("RedTeamSwarm target callable is missing or not executable")
                     sig = inspect.signature(fn)
                     if len(sig.parameters) == 0:
                         return fn()