diff --git a/agentrace/checks.py b/agentrace/checks.py index d8eba15..fb80ee7 100644 --- a/agentrace/checks.py +++ b/agentrace/checks.py @@ -81,6 +81,40 @@ def check_refused_or_gave_up(run: AgentRun) -> list[Finding]: return [] +def check_destructive_command(run: AgentRun) -> list[Finding]: + """Runnable destructive shell/database commands without a nearby safety warning. + + A subagent can emit commands that an orchestrator or human may copy-paste. Destructive commands + should be paired with a dry-run, backup, or explicit verification warning before execution. + """ + patterns = [ + r"\brm\s+-[^\n]*r[^\n]*f[^\n]*\s+\S+", + r"\bDROP\s+TABLE\b", + r"\bkubectl\s+delete\b", + r":\s*\(\s*\)\s*\{\s*:\s*\|\s*:\s*&\s*\}\s*;\s*:", + ] + caution = re.compile(r"\b(?:dry[- ]?run|backup|verify first|check first|confirm first)\b|--dry-run", re.I) + head = run.result[:1500] + + for pattern in patterns: + m = re.search(pattern, head, re.I) + if not m: + continue + start = max(0, m.start() - 140) + end = min(len(head), m.end() + 140) + if caution.search(head[start:end]): + continue + return [ + Finding( + "destructive_command", + "high", + "Runnable destructive command without a nearby dry-run, backup, or verification warning.", + _context(head, m.start()), + ) + ] + return [] + + def check_unverified_claim(run: AgentRun) -> list[Finding]: """Hedged language presented as a finding. @@ -264,6 +298,7 @@ def check_runaway(run: AgentRun, slow_s: float = 900.0) -> list[Finding]: CHECKS: list[Callable[[AgentRun], list[Finding]]] = [ check_empty_result, check_refused_or_gave_up, + check_destructive_command, check_unverified_claim, check_absence_as_evidence, check_url_without_verification, diff --git a/tests/test_destructive_command.py b/tests/test_destructive_command.py new file mode 100644 index 0000000..451b624 --- /dev/null +++ b/tests/test_destructive_command.py @@ -0,0 +1,45 @@ +from datetime import datetime, timezone + +from agentrace.checks import analyse +from agentrace.parse import AgentRun + + +def _run(result: str) -> AgentRun: + return AgentRun( + tool_use_id="toolu_destructive_test", + description="test run", + prompt="Review the operation and report what should be run.", + result=result, + started_at=datetime(2026, 7, 16, 12, 0, tzinfo=timezone.utc), + ended_at=datetime(2026, 7, 16, 12, 1, tzinfo=timezone.utc), + ) + + +def _findings(result: str): + return [f for f in analyse(_run(result)) if f.check == "destructive_command"] + + +def test_destructive_command_flags_rm_rf_without_warning(): + findings = _findings("Run: rm -rf /data/cache") + assert findings + assert findings[0].severity == "high" + + +def test_destructive_command_allows_rm_rf_with_dry_run_warning(): + assert not _findings("Verify with a dry-run first, then run: rm -rf /data/cache") + + +def test_destructive_command_flags_drop_table_without_warning(): + assert _findings("Execute: DROP TABLE users;") + + +def test_destructive_command_flags_kubectl_delete_without_warning(): + assert _findings("Run kubectl delete namespace production") + + +def test_destructive_command_allows_command_with_backup_warning(): + assert not _findings("Take a backup before running: DROP TABLE users;") + + +def test_destructive_command_ignores_unrelated_safe_output(): + assert not _findings("Run pytest -q and review the failures before making changes.")