114 lines
3.9 KiB
Python
114 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
|
|
from scripts import run_leo_direct_claim_handler_suite as suite
|
|
|
|
|
|
def safe_report() -> dict:
|
|
return {
|
|
"remote_returncode": 0,
|
|
"pass_runtime": True,
|
|
"handler": {"authorized": True},
|
|
"results": [
|
|
{
|
|
"ok": True,
|
|
"mutates_kb": False,
|
|
"database_tool_trace": {
|
|
"database_tool_call_count": 1,
|
|
"database_tool_completed_count": 1,
|
|
"database_tool_calls_read_only": True,
|
|
"calls": [
|
|
{
|
|
"database_invocations": [{"access_mode": "read_only"}],
|
|
"result": {
|
|
"content_sha256": "a" * 64,
|
|
"error_detected": False,
|
|
"nonempty": True,
|
|
},
|
|
}
|
|
],
|
|
},
|
|
}
|
|
],
|
|
"posted_to_telegram": False,
|
|
"mutates_kb_by_harness": False,
|
|
"db_counts_changed": False,
|
|
"db_fingerprint_unchanged": True,
|
|
"live_behavior_manifest_unchanged": True,
|
|
"service_before_after": {"unchanged_from_preexisting_live_readback": True},
|
|
"temp_profile_removed": True,
|
|
"model_call_trace_all_bound": True,
|
|
"execution_manifest_summary": {"all_turns_attribution_complete": True},
|
|
}
|
|
|
|
|
|
def test_redaction_preserves_json_and_removes_secret_values() -> None:
|
|
raw = json.dumps(
|
|
{
|
|
"bot_token": "123456789:abcdefghijklmnopqrstuvwxyz_ABCD",
|
|
"api_key": "sk-live-secret",
|
|
"nested": {"password": "hunter2", "access_token": "bearer-secret"},
|
|
"message": "Authorization: Bearer top-secret-value",
|
|
}
|
|
)
|
|
|
|
redacted = suite.redact(raw)
|
|
parsed = json.loads(redacted)
|
|
|
|
assert parsed["bot_token"] == "[REDACTED]"
|
|
assert parsed["api_key"] == "[REDACTED]"
|
|
assert parsed["nested"]["password"] == "[REDACTED]"
|
|
assert parsed["nested"]["access_token"] == "[REDACTED]"
|
|
assert parsed["message"] == "Authorization: Bearer [REDACTED]"
|
|
assert "hunter2" not in redacted
|
|
assert "top-secret-value" not in redacted
|
|
|
|
|
|
def test_successful_stdout_is_redacted_before_json_parsing(monkeypatch) -> None:
|
|
secret = "123456789:abcdefghijklmnopqrstuvwxyz_ABCD"
|
|
calls = iter(
|
|
[
|
|
subprocess.CompletedProcess([], 0, stdout=json.dumps({"bot_token": secret}) + "\n", stderr=""),
|
|
subprocess.CompletedProcess([], 0, stdout="", stderr=""),
|
|
]
|
|
)
|
|
monkeypatch.setattr(suite.subprocess, "run", lambda *args, **kwargs: next(calls))
|
|
|
|
result = suite.run_remote(prompts=[])
|
|
|
|
assert result["parsed"]["bot_token"] == "[REDACTED]"
|
|
assert secret not in json.dumps(result)
|
|
|
|
|
|
def test_safety_gate_passes_only_for_complete_no_send_no_write_run() -> None:
|
|
report = safe_report()
|
|
|
|
assert suite.report_passes_safety(report) is True
|
|
assert suite.build_safety_gate(report)["failed_checks"] == []
|
|
|
|
|
|
def test_safety_gate_rejects_write_capable_tool_and_incomplete_receipt() -> None:
|
|
report = safe_report()
|
|
report["results"][0]["database_tool_trace"]["database_tool_calls_read_only"] = False
|
|
report["results"][0]["database_tool_trace"]["calls"][0]["database_invocations"][0][
|
|
"access_mode"
|
|
] = "write"
|
|
report["execution_manifest_summary"]["all_turns_attribution_complete"] = False
|
|
|
|
gate = suite.build_safety_gate(report)
|
|
|
|
assert gate["status"] == "fail"
|
|
assert "all_tool_traces_read_only_and_bound" in gate["failed_checks"]
|
|
assert "all_turn_manifests_complete" in gate["failed_checks"]
|
|
|
|
|
|
def test_safety_gate_rejects_missing_no_send_declaration() -> None:
|
|
report = safe_report()
|
|
report.pop("posted_to_telegram")
|
|
|
|
gate = suite.build_safety_gate(report)
|
|
|
|
assert gate["status"] == "fail"
|
|
assert "no_telegram_post" in gate["failed_checks"]
|