diff --git a/fable_engine/actions/fleet.py b/fable_engine/actions/fleet.py index 4f1a2f6..a333a46 100644 --- a/fable_engine/actions/fleet.py +++ b/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, diff --git a/fable_engine/schema.py b/fable_engine/schema.py index 1c5b31d..eaf55d9 100644 --- a/fable_engine/schema.py +++ b/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.", diff --git a/fable_v2/coder_fleet/red_team_swarm.py b/fable_v2/coder_fleet/red_team_swarm.py index c476882..0c33dea 100644 --- a/fable_v2/coder_fleet/red_team_swarm.py +++ b/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() diff --git a/skills/fable-mode/cortex/synaptic_matrix.json b/skills/fable-mode/cortex/synaptic_matrix.json index 1bb0af3..1be1c6f 100644 --- a/skills/fable-mode/cortex/synaptic_matrix.json +++ b/skills/fable-mode/cortex/synaptic_matrix.json @@ -1,24 +1,34 @@ { + "diagnostics": { + "red_team_swarm": 0.3693, + "system": 0.3764, + "test_harness": 0.3768 + }, "mutation": { - "property_oracle": 0.2229, - "red_team_swarm": 0.7737, + "property_oracle": 0.3687, + "red_team_swarm": 0.9249, "security": 0.6983, - "test_harness": 0.74 + "target": 0.344, + "test_harness": 0.8966 }, "property_oracle": { - "mutation": 0.2229, - "red_team_swarm": 0.218, + "mutation": 0.3687, + "red_team_swarm": 0.354, "security": 0.2648, - "test_harness": 0.2205 + "target": 0.3296, + "test_harness": 0.3615 }, "quorum": { "test_distributed_raft": 0.95 }, "red_team_swarm": { - "mutation": 0.7737, - "property_oracle": 0.218, + "diagnostics": 0.3693, + "mutation": 0.9249, + "property_oracle": 0.354, "security": 0.7082, - "test_harness": 0.7523 + "system": 0.3827, + "target": 0.3344, + "test_harness": 1.0 }, "security": { "mutation": 0.6983, @@ -26,13 +36,27 @@ "red_team_swarm": 0.7082, "test_harness": 0.6812 }, + "system": { + "diagnostics": 0.3764, + "red_team_swarm": 0.3827, + "test_harness": 0.389 + }, + "target": { + "mutation": 0.344, + "property_oracle": 0.3296, + "red_team_swarm": 0.3344, + "test_harness": 0.3392 + }, "test_distributed_raft": { "quorum": 0.95 }, "test_harness": { - "mutation": 0.74, - "property_oracle": 0.2205, - "red_team_swarm": 0.7523, - "security": 0.6812 + "diagnostics": 0.3768, + "mutation": 0.8966, + "property_oracle": 0.3615, + "red_team_swarm": 1.0, + "security": 0.6812, + "system": 0.389, + "target": 0.3392 } } \ No newline at end of file diff --git a/skills/fable-mode/cortex/system.md b/skills/fable-mode/cortex/system.md new file mode 100644 index 0000000..95ea05c --- /dev/null +++ b/skills/fable-mode/cortex/system.md @@ -0,0 +1,44 @@ +--- +{ + "name": "system", + "description": "Custom cortical lobe for system development and specialized heuristics", + "domain": "system", + "activation_count": 3, + "synaptic_weights": { + "test_harness": 0.489, + "red_team_swarm": 0.4827, + "diagnostics": 0.4764 + }, + "antibodies": [], + "specialized_heuristics": [], + "last_consolidated_at": "2026-09-10T16:23:18.184792+00:00" +} +--- + +# Cortical Lobe: `system` + +> [!NOTE] +> Custom cortical lobe for system development and specialized heuristics +> Activation count: 3. + +## Metadata & Telemetry +- **Name**: `system` +- **Description**: Custom cortical lobe for system development and specialized heuristics +- **Domain**: `system` +- **Activation Count**: `3` +- **Total Antibodies**: `0` +- **Specialized Heuristics**: `0` +- **Last Consolidated**: `2026-09-10T16:23:18.184792+00:00` + +## Specialized Domain Heuristics +- *(No domain heuristics registered yet)* + +## Synaptic Tool & Node Weights (Hebbian Association) +| Synaptic Node / Tool | Weight ($W_{ij}$) | Strength | +| :--- | :--- | :--- | +| `test_harness` | `0.4890` | 🟡 Moderate | +| `red_team_swarm` | `0.4827` | 🟡 Moderate | +| `diagnostics` | `0.4764` | 🟡 Moderate | + +## Immunological Antibodies (Red-Team Scars) +- *(Zero known fatal vulnerabilities cataloged)* diff --git a/skills/fable-mode/cortex/target.md b/skills/fable-mode/cortex/target.md new file mode 100644 index 0000000..2e7b888 --- /dev/null +++ b/skills/fable-mode/cortex/target.md @@ -0,0 +1,66 @@ +--- +{ + "name": "target", + "description": "Custom cortical lobe for target development and specialized heuristics", + "domain": "target", + "activation_count": 2, + "synaptic_weights": { + "mutation": 0.444, + "test_harness": 0.4392, + "red_team_swarm": 0.4344, + "property_oracle": 0.4296 + }, + "antibodies": [ + { + "antibody_id": "ab_target_target_chaos_01_missing_path", + "domain": "target", + "trigger_condition": "Hypothesis", + "lethal_anti_pattern": "Unchecked execution failure under adversarial pressure", + "prescribed_defense": "Enforce strict precondition verification and atomic isolation.", + "severity": "MEDIUM", + "source_task_id": "rep_prior_falsey", + "created_at": "2026-09-10T16:10:33.481288+00:00", + "verified_counterfactual": "Counterfactual validation against vector: chaos_environment" + } + ], + "specialized_heuristics": [ + "Defense against [Hypothesis]: Hardened implementation" + ], + "last_consolidated_at": "2026-09-10T16:23:18.199648+00:00" +} +--- + +# Cortical Lobe: `target` + +> [!NOTE] +> Custom cortical lobe for target development and specialized heuristics +> Activation count: 2. + +## Metadata & Telemetry +- **Name**: `target` +- **Description**: Custom cortical lobe for target development and specialized heuristics +- **Domain**: `target` +- **Activation Count**: `2` +- **Total Antibodies**: `1` +- **Specialized Heuristics**: `1` +- **Last Consolidated**: `2026-09-10T16:23:18.199648+00:00` + +## Specialized Domain Heuristics +1. Defense against [Hypothesis]: Hardened implementation + +## Synaptic Tool & Node Weights (Hebbian Association) +| Synaptic Node / Tool | Weight ($W_{ij}$) | Strength | +| :--- | :--- | :--- | +| `mutation` | `0.4440` | 🟡 Moderate | +| `test_harness` | `0.4392` | 🟡 Moderate | +| `red_team_swarm` | `0.4344` | 🟡 Moderate | +| `property_oracle` | `0.4296` | 🟡 Moderate | + +## Immunological Antibodies (Red-Team Scars) +#### Antibody `ab_target_target_chaos_01_missing_path` [MEDIUM] +- **Domain**: `target` +- **Trigger Condition**: Hypothesis +- **Lethal Anti-Pattern**: Unchecked execution failure under adversarial pressure +- **Prescribed Defense**: Enforce strict precondition verification and atomic isolation. +- **Verified Counterfactual**: `Counterfactual validation against vector: chaos_environment` +- **Source Task ID**: `rep_prior_falsey` diff --git a/tests/test_fsm_redteam_evolution.py b/tests/test_fsm_redteam_evolution.py index 527131e..12488e3 100644 --- a/tests/test_fsm_redteam_evolution.py +++ b/tests/test_fsm_redteam_evolution.py @@ -257,7 +257,9 @@ def test_closed_loop_ping_pong_remediation_and_sealing(self): } ] } - flawed_code = "def process(x):\n raise MemoryError('Still leaking')" + def flawed_code(x=None): + raise MemoryError('Still leaking') + resp_flawed = handle_fable_session({ "action": "verify_red_team_remediation", "session_name": session_name, @@ -269,7 +271,9 @@ def test_closed_loop_ping_pong_remediation_and_sealing(self): self.assertEqual(session.current_state, SessionState.REMEDIATION_REQUIRED) # 3. Subagent submits fully fixed code that survives the prior breaking probe - fixed_code = "def process(x):\n return 'clean'" + def fixed_code(x=None): + return 'clean' + resp_fixed = handle_fable_session({ "action": "verify_red_team_remediation", "session_name": session_name, diff --git a/tests/test_red_team_swarm.py b/tests/test_red_team_swarm.py index b42c9cc..33a2102 100644 --- a/tests/test_red_team_swarm.py +++ b/tests/test_red_team_swarm.py @@ -10,11 +10,13 @@ """ from __future__ import annotations +import copy import os import sys import tempfile import threading import unittest +import uuid from pathlib import Path from typing import Any @@ -109,6 +111,47 @@ def hardened_target(payload: Any = None) -> dict[str, Any]: self.assertEqual(len([f for f in report.findings if f.broken]), 0) +class TestRedTeamSwarmSecurityBoundary(unittest.TestCase): + def setUp(self) -> None: + self.swarm = RedTeamSwarm() + + def test_none_or_unexecutable_target_fails_closed(self) -> None: + report = self.swarm.execute_swarm_attack(None) + self.assertFalse(report.passed) + self.assertEqual(report.broken_count, 1) + self.assertEqual(report.findings[0].scenario_id, "target_not_executable") + self.assertFalse(report.findings[0].details.get("target_executable", True)) + self.assertIn("TypeError", report.findings[0].error_message or "") + + def test_source_string_target_fails_closed(self) -> None: + code_snippet = "def safe_fn(x=None):\n return 'ok'\n" + report = self.swarm.execute_swarm_attack(code_snippet) + self.assertFalse(report.passed) + self.assertEqual(report.broken_count, 1) + self.assertFalse(report.findings[0].details.get("target_executable", True)) + + def test_callable_object_succeeds(self) -> None: + def safe_fn(x=None): + return "ok" + + report = self.swarm.execute_swarm_attack(safe_fn) + self.assertTrue(report.passed) + self.assertEqual(report.broken_count, 0) + + def test_false_bool_callable_object_succeeds(self) -> None: + class FalseCallable: + def __bool__(self) -> bool: + return False + + def __call__(self, x: Any = None) -> str: + return "ok" + + false_callable = FalseCallable() + report = self.swarm.execute_swarm_attack(false_callable) + self.assertTrue(report.passed) + self.assertEqual(report.broken_count, 0) + + class TestReportFormattingAndSerialization(unittest.TestCase): def setUp(self) -> None: self.swarm = RedTeamSwarm() @@ -281,6 +324,190 @@ def candidate_v2(input_data: Any = None) -> str: self.assertGreater(len(lobe_reloaded.specialized_heuristics), 0) +class TestPublicActionHandlers(unittest.TestCase): + def test_handle_red_team_code_review_rejects_source_string_without_mutation(self) -> None: + from fable_engine.actions.fleet import _handle_red_team_code_review + from fable_engine.session import FableSession, ACTIVE_SESSIONS, SESSIONS_DIR + session_name = f"test_no_mutate_{uuid.uuid4().hex[:8]}" + session_file = SESSIONS_DIR / f"{session_name}.json" + session = FableSession(session_name=session_name, objective="Test no mutation", time_budget_minutes=5.0) + ACTIVE_SESSIONS[session_name] = session + snapshot = copy.deepcopy(session.to_dict()) + + try: + resp = _handle_red_team_code_review({ + "action": "red_team_code_review", + "session_name": session_name, + "target_code": "def process(): pass", + }) + self.assertIn("Error: Source-code strings cannot be evaluated in-process for security reasons", resp) + self.assertEqual(session.to_dict(), snapshot) + finally: + ACTIVE_SESSIONS.pop(session_name, None) + if session_file.exists(): + session_file.unlink() + + def test_handle_verify_red_team_remediation_rejects_source_string_without_mutation(self) -> None: + from fable_engine.actions.fleet import _handle_verify_red_team_remediation + from fable_engine.session import FableSession, ACTIVE_SESSIONS, SESSIONS_DIR + session_name = f"test_no_mutate_{uuid.uuid4().hex[:8]}" + session_file = SESSIONS_DIR / f"{session_name}.json" + session = FableSession(session_name=session_name, objective="Test no mutation", time_budget_minutes=5.0) + ACTIVE_SESSIONS[session_name] = session + snapshot = copy.deepcopy(session.to_dict()) + + try: + resp = _handle_verify_red_team_remediation({ + "action": "verify_red_team_remediation", + "session_name": session_name, + "remediated_code": "def process(): pass", + }) + self.assertIn("Error: Source-code strings cannot be evaluated in-process for security reasons", resp) + self.assertEqual(session.to_dict(), snapshot) + finally: + ACTIVE_SESSIONS.pop(session_name, None) + if session_file.exists(): + session_file.unlink() + + def test_rejected_source_string_with_nonexistent_session_name(self) -> None: + from fable_engine.actions.fleet import _handle_red_team_code_review, _handle_verify_red_team_remediation + from fable_engine.session import ACTIVE_SESSIONS, SESSIONS_DIR + + nonexistent_name = f"nonexistent_session_{uuid.uuid4().hex[:8]}" + session_file = SESSIONS_DIR / f"{nonexistent_name}.json" + + ACTIVE_SESSIONS.pop(nonexistent_name, None) + if session_file.exists(): + session_file.unlink() + + try: + self.assertNotIn(nonexistent_name, ACTIVE_SESSIONS) + + resp1 = _handle_red_team_code_review({ + "action": "red_team_code_review", + "session_name": nonexistent_name, + "target_code": "def process(): pass", + }) + self.assertIn("Error: Source-code strings cannot be evaluated in-process for security reasons", resp1) + self.assertNotIn(nonexistent_name, ACTIVE_SESSIONS) + self.assertFalse(session_file.exists()) + + resp2 = _handle_verify_red_team_remediation({ + "action": "verify_red_team_remediation", + "session_name": nonexistent_name, + "remediated_code": "def process(): pass", + }) + self.assertIn("Error: Source-code strings cannot be evaluated in-process for security reasons", resp2) + self.assertNotIn(nonexistent_name, ACTIVE_SESSIONS) + self.assertFalse(session_file.exists()) + finally: + ACTIVE_SESSIONS.pop(nonexistent_name, None) + if session_file.exists(): + session_file.unlink() + + def test_missing_session_name_does_not_create_session_or_file(self) -> None: + from fable_engine.actions.fleet import _handle_red_team_code_review, _handle_verify_red_team_remediation + from fable_engine.session import ACTIVE_SESSIONS, SESSIONS_DIR + + initial_active_keys = set(ACTIVE_SESSIONS.keys()) + + resp1 = _handle_red_team_code_review({"action": "red_team_code_review"}) + self.assertIn("Error: 'session_name' is required", resp1) + + resp2 = _handle_verify_red_team_remediation({"action": "verify_red_team_remediation"}) + self.assertIn("Error: 'session_name' is required", resp2) + + self.assertEqual(set(ACTIVE_SESSIONS.keys()), initial_active_keys) + self.assertFalse((SESSIONS_DIR / ".json").exists()) + + def test_falsey_callable_object_accepted_in_public_action_handlers(self) -> None: + from fable_engine.actions.fleet import _handle_red_team_code_review + from fable_engine.session import FableSession, ACTIVE_SESSIONS, SESSIONS_DIR, SessionState + + class FalseCallable: + def __bool__(self) -> bool: + return False + + def __call__(self, x: Any = None) -> str: + return "ok" + + session_name = f"test_falsey_callable_{uuid.uuid4().hex[:8]}" + session_file = SESSIONS_DIR / f"{session_name}.json" + + session = FableSession(session_name=session_name, objective="Test falsey callable", time_budget_minutes=5.0) + session.set_timer(5.0) + session.execution_locked = False + session.can_execute_code = True + session.transition_to(SessionState.IMPLEMENTATION, "Implemented") + session.track_file_change("sample.py", "created", "Added initial implementation") + session.transition_to(SessionState.RED_TEAM_GATE, "Ready for gate") + ACTIVE_SESSIONS[session_name] = session + + try: + resp = _handle_red_team_code_review({ + "action": "red_team_code_review", + "session_name": session_name, + "target_code": FalseCallable(), + }) + self.assertNotIn("Error: Source-code strings cannot be evaluated in-process", resp) + self.assertIn("Adversarial Red Team Resilient Attestation", resp) + finally: + ACTIVE_SESSIONS.pop(session_name, None) + if session_file.exists(): + session_file.unlink() + + def test_falsey_callable_object_accepted_in_verify_red_team_remediation(self) -> None: + from fable_engine.actions.fleet import _handle_verify_red_team_remediation + from fable_engine.session import FableSession, ACTIVE_SESSIONS, SESSIONS_DIR, SessionState + + class FalseCallable: + def __bool__(self) -> bool: + return False + + def __call__(self, x: Any = None) -> str: + return "ok" + + session_name = f"test_falsey_remediation_{uuid.uuid4().hex[:8]}" + session_file = SESSIONS_DIR / f"{session_name}.json" + + session = FableSession(session_name=session_name, objective="Test falsey remediation", time_budget_minutes=5.0) + session.set_timer(5.0) + session.execution_locked = False + session.can_execute_code = True + session.transition_to(SessionState.IMPLEMENTATION, "Implemented") + session.track_file_change("sample.py", "created", "Added initial implementation") + session.transition_to(SessionState.RED_TEAM_GATE, "Ready for gate") + ACTIVE_SESSIONS[session_name] = session + + prior_report = { + "report_id": "rep_prior_falsey", + "target_name": "target", + "broken_count": 1, + "findings": [ + { + "scenario_id": "target_chaos_01_missing_path", + "vector": "chaos_environment", + "hypothesis": "Hypothesis", + "broken": True, + } + ], + } + + try: + resp = _handle_verify_red_team_remediation({ + "action": "verify_red_team_remediation", + "session_name": session_name, + "remediated_code": FalseCallable(), + "prior_report": prior_report, + }) + self.assertNotIn("Error: Source-code strings cannot be evaluated in-process", resp) + self.assertIn("TASK COMPLETED: 0 breakages remain. Code sealed.", resp) + finally: + ACTIVE_SESSIONS.pop(session_name, None) + if session_file.exists(): + session_file.unlink() + + class TestCoderFleetDispatcherRedTeamActions(unittest.TestCase): def setUp(self) -> None: self.temp_dir = tempfile.TemporaryDirectory() @@ -314,16 +541,28 @@ def test_dispatch_red_team_generate_scenarios(self) -> None: self.assertGreater(len(scenarios), 0) def test_dispatch_red_team_full_review_cycle(self) -> None: - code_snippet = "def safe_fn(x=None):\n return 'safe'\n" + def safe_fn(x=None): + return "safe" + res = self.dispatcher.dispatch( "red_team_full_review_cycle", - {"target_callable": code_snippet, "target_name": "safe_fn"}, + {"target_callable": safe_fn, "target_name": "safe_fn"}, ) self.assertTrue(res["success"]) report = res["result"] self.assertTrue(isinstance(report, RedTeamBreakageReport)) self.assertTrue(report.passed) + # Source code strings produce fail-closed breakage reports + res_str = self.dispatcher.dispatch( + "red_team_full_review_cycle", + {"target_callable": "def safe_fn(x=None):\n return 'safe'\n", "target_name": "safe_fn"}, + ) + self.assertTrue(res_str["success"]) + report_str = res_str["result"] + self.assertFalse(report_str.passed) + self.assertEqual(report_str.broken_count, 1) + if __name__ == "__main__": unittest.main()