diff --git a/Detection/README.md b/Detection/README.md index ae48712..ae4121a 100644 --- a/Detection/README.md +++ b/Detection/README.md @@ -291,6 +291,10 @@ benchmark/adr_bench_YYYYMMDD_HHMMSS/ └── summary.json # Original benchmark results ``` +Each detector result includes an additive `run_manifest` with privacy-safe run provenance: source revision when available, resolved concurrency, sorted selected task IDs, and SHA-256 digests of selected conversations, effective labels, and fixed detector inputs. AgentDojo includes a digest of `ground_truth.json`. Missing provenance is recorded as `null` and never fails detection. Paths, directory names, host identifiers, environment values, prompts, and file contents are not stored. + +Input digests are captured from the same bytes read by the detector, not by rereading files after analysis. The CLI uses `config_detector.yaml` and ADR-Bench `tasks.json` from the current working directory; task labels and MCP definitions share one snapshot. Source revision and `uv.lock` are captured before analysis. Programmatic callers that supply an already-configured detector without its configuration digest get `null` for that artifact. + **Each detector file contains**: - `detector_info`: Configuration and model information @@ -645,4 +649,4 @@ uv run python main_benchmark.py --tasks=1-10 Apache License 2.0 — see [LICENSE](LICENSE). Vendored AgentDojo code under [benchmark/agentdojo/LICENSE](benchmark/agentdojo/LICENSE) (MIT). -This project is intended for defensive security research and agentic AI safety evaluation. Do not use it to conduct unauthorized attacks. \ No newline at end of file +This project is intended for defensive security research and agentic AI safety evaluation. Do not use it to conduct unauthorized attacks. diff --git a/Detection/main_detector.py b/Detection/main_detector.py index b044cea..83dc5d9 100644 --- a/Detection/main_detector.py +++ b/Detection/main_detector.py @@ -12,7 +12,7 @@ import sys import traceback from pathlib import Path -from typing import Dict, List, Any +from typing import Dict, List, Any, Optional from datetime import datetime import logging @@ -26,6 +26,7 @@ # Import the baseline detectors from guardrail.llamafirewall_agent.llamafirewall_baseline import LlamaFirewallBaseline from guardrail.adr_agent.adr_baseline import ADRBaseline +from run_manifest import collect_run_manifest, collect_source_metadata, read_text_with_sha256 class BenchmarkAnalyzer: @@ -33,7 +34,7 @@ class BenchmarkAnalyzer: Analyzer for processing ADR benchmark results and generating metrics. """ - def __init__(self, detector): + def __init__(self, detector, *, config_sha256=None, source_metadata=None): """ Initialize with a baseline detector. @@ -41,6 +42,11 @@ def __init__(self, detector): detector: Any detector implementing BaseDetector interface """ self.detector = detector + self._config_sha256 = config_sha256 + self._source_metadata = source_metadata + self._conversation_hashes: Dict[str, Optional[str]] = {} + self._artifact_hashes: Dict[str, Optional[str]] = {} + self._loaded_task_definitions = None def process_benchmark_results(self, results_dir_path: str, task_filter: List[int] = None, max_concurrent: int = 10, benchmark_type: str = "adr_bench") -> Dict[str, Any]: @@ -69,6 +75,15 @@ def process_benchmark_results(self, results_dir_path: str, task_filter: List[int validate_benchmark_results_dir(results_path, benchmark_type) + # Keep one input snapshot per run, including when this analyzer is reused. + source_metadata = self._source_metadata or collect_source_metadata(Path(__file__).parent) + self._conversation_hashes = {} + self._artifact_hashes = { + 'config_detector': self._config_sha256, + 'uv_lock': source_metadata['uv_lock'], + } + self._loaded_task_definitions = None + print(f"📁 Found {len(task_dirs)} task directories to analyze") inferred_type = "agentdojo" if "agentdojo" in results_path.name else "adr_bench" @@ -78,28 +93,46 @@ def process_benchmark_results(self, results_dir_path: str, task_filter: List[int # Run analysis analyses, run_stats = self._analyze_tasks_efficiently( - sorted(task_dirs), task_filter, max_concurrent, benchmark_type, ground_truth + sorted(task_dirs), task_filter, max_concurrent, benchmark_type, ground_truth, + task_definitions=self._loaded_task_definitions, ) # Calculate metrics metrics = self._calculate_metrics(analyses, ground_truth) + selected_task_dirs = task_dirs + if task_filter: + selected_names = {f"task_{task_id:03d}" for task_id in task_filter} + selected_task_dirs = [task_dir for task_dir in task_dirs if task_dir.name in selected_names] + run_manifest = collect_run_manifest( + benchmark_type=benchmark_type, + task_dirs=selected_task_dirs, + effective_labels=ground_truth, + resolved_concurrency=max_concurrent, + conversation_hashes=self._conversation_hashes, + artifact_hashes=self._artifact_hashes, + source=source_metadata['source'], + ) + return { 'detector_info': self.detector.get_info(), 'analyses': analyses, 'metrics': metrics, 'run_stats': run_stats, - 'analysis_timestamp': datetime.now().isoformat() + 'analysis_timestamp': datetime.now().isoformat(), + 'run_manifest': run_manifest } def _analyze_tasks_efficiently(self, task_dirs: List[Path], task_filter: List[int] = None, max_concurrent: int = 10, - benchmark_type: str = "adr_bench", ground_truth_dict: Dict[str, bool] = None) -> tuple[List[Dict[str, Any]], Dict[str, int]]: + benchmark_type: str = "adr_bench", ground_truth_dict: Dict[str, bool] = None, + task_definitions: Dict[str, Any] = None) -> tuple[List[Dict[str, Any]], Dict[str, int]]: """Analyze tasks using the detector in an optimized manner.""" # Run the async analysis in a new event loop - return asyncio.run(self._analyze_tasks_async(task_dirs, task_filter, max_concurrent, benchmark_type, ground_truth_dict)) + return asyncio.run(self._analyze_tasks_async(task_dirs, task_filter, max_concurrent, benchmark_type, ground_truth_dict, task_definitions)) async def _analyze_tasks_async(self, task_dirs: List[Path], task_filter: List[int] = None, max_concurrent: int = 10, - benchmark_type: str = "adr_bench", ground_truth_dict: Dict[str, bool] = None) -> tuple[List[Dict[str, Any]], Dict[str, int]]: + benchmark_type: str = "adr_bench", ground_truth_dict: Dict[str, bool] = None, + task_definitions: Dict[str, Any] = None) -> tuple[List[Dict[str, Any]], Dict[str, int]]: """Analyze tasks using the detector with parallel processing.""" analyses = [] @@ -112,16 +145,18 @@ async def _analyze_tasks_async(self, task_dirs: List[Path], task_filter: List[in total_tasks = len(task_dirs) - # Load task definitions - task_definitions = {} - if benchmark_type == "adr_bench": - # Load tasks.json for ADR-Bench MCP server information + # Reuse the same task snapshot as the labels. Direct helper callers can + # still load definitions here when no snapshot was supplied. + if task_definitions is None: + task_definitions = {} tasks_file = Path("tasks.json") - if tasks_file.exists(): - with open(tasks_file, 'r') as f: - tasks_data = json.load(f) - for task in tasks_data.get("tasks", []): - task_definitions[f"task_{task['task_id']:03d}"] = task + if benchmark_type == "adr_bench" and tasks_file.exists(): + text, digest = read_text_with_sha256(tasks_file) + self._artifact_hashes['tasks'] = digest + tasks_data = json.loads(text) + del text + for task in tasks_data.get("tasks", []): + task_definitions[f"task_{task['task_id']:03d}"] = task # Simple semaphore for concurrency control semaphore = asyncio.Semaphore(max_concurrent) @@ -208,8 +243,10 @@ async def _analyze_task_async(self, task_dir: Path, index: int, total_tasks: int return None # Load the structured conversation file - with open(conversation_file, 'r', encoding='utf-8') as f: - conversation_data = json.load(f) + text, digest = read_text_with_sha256(conversation_file, encoding='utf-8') + self._conversation_hashes[task_id] = digest + conversation_data = json.loads(text) + del text # Convert to our chat log format messages = self._convert_conversation_to_messages(conversation_data) @@ -304,14 +341,20 @@ def _load_ground_truth(self, benchmark_type: str = "adr_bench") -> Dict[str, boo return {} try: - with open(tasks_file, 'r', encoding='utf-8') as f: - tasks_data = json.load(f) + text, digest = read_text_with_sha256(tasks_file, encoding='utf-8') + self._artifact_hashes['tasks'] = digest + tasks_data = json.loads(text) + del text ground_truth = {} + task_definitions = {} # Handle the current format where tasks is a list for task in tasks_data['tasks']: task_id = f"task_{task['task_id']:03d}" ground_truth[task_id] = task.get('ground_truth', 'benign') == 'malicious' + task_definitions[task_id] = task + + self._loaded_task_definitions = task_definitions print(f"📋 Loaded ADR-Bench ground truth for {len(ground_truth)} tasks") return ground_truth @@ -326,8 +369,10 @@ def _load_ground_truth(self, benchmark_type: str = "adr_bench") -> Dict[str, boo if not ground_truth_file.exists(): raise FileNotFoundError(f"AgentDojo ground truth file not found: {ground_truth_file.absolute()}") - with open(ground_truth_file, 'r') as f: - agentdojo_ground_truth = json.load(f) + text, digest = read_text_with_sha256(ground_truth_file) + self._artifact_hashes['agentdojo_ground_truth'] = digest + agentdojo_ground_truth = json.loads(text) + del text ground_truth = {} for task_key, task_data in agentdojo_ground_truth.items(): @@ -732,11 +777,14 @@ def main(): print("=" * 50) # Load detector configuration upfront + source_metadata = collect_source_metadata(Path(__file__).parent) config_file = Path("config_detector.yaml") config_data = {} + config_sha256 = None if config_file.exists(): - with open(config_file, 'r') as f: - config_data = yaml.safe_load(f) or {} + text, config_sha256 = read_text_with_sha256(config_file) + config_data = yaml.safe_load(text) or {} + del text print(f"📋 Loaded configuration from {config_file}") else: print(f"⚠️ Configuration file {config_file} not found, using defaults") @@ -814,7 +862,9 @@ def main(): if args.detector != "adr": # ADR already printed detailed info above print(f"✅ {args.detector} ready") - analyzer = BenchmarkAnalyzer(detector) + analyzer = BenchmarkAnalyzer( + detector, config_sha256=config_sha256, source_metadata=source_metadata + ) # Process task filtering arguments task_filter = None diff --git a/Detection/pyproject.toml b/Detection/pyproject.toml index bd1759a..6341ea4 100644 --- a/Detection/pyproject.toml +++ b/Detection/pyproject.toml @@ -147,6 +147,7 @@ packages = ["guardrail"] [tool.hatch.build.targets.wheel.force-include] "main_benchmark.py" = "main_benchmark.py" "main_detector.py" = "main_detector.py" +"run_manifest.py" = "run_manifest.py" "plot_paper_figures.py" = "plot_paper_figures.py" "openai_config.py" = "openai_config.py" "benchmark/benchmark_pack.py" = "benchmark_pack.py" diff --git a/Detection/run_manifest.py b/Detection/run_manifest.py new file mode 100644 index 0000000..e8e9a0a --- /dev/null +++ b/Detection/run_manifest.py @@ -0,0 +1,194 @@ +"""Best-effort run provenance for detector analysis output.""" + +from __future__ import annotations + +import hashlib +import io +import json +import subprocess +import sys +from pathlib import Path +from typing import Any, Dict, Iterable, Mapping, Optional, Sequence + +_HASH_ALGORITHM = "sha256" +_GIT_TIMEOUT_SECONDS = 2 + + +def read_text_with_sha256( + path: Path, *, encoding: Optional[str] = None +) -> tuple[str, Optional[str]]: + """Read once, returning parser input and a best-effort hash of those raw bytes. + + Preserve the caller's text-mode encoding and newline handling. Read/decode + errors still reach the existing input loader; only provenance is nonfatal. + """ + content = path.read_bytes() + with io.TextIOWrapper(io.BytesIO(content), encoding=encoding) as handle: + text = handle.read() + try: + digest = hashlib.sha256(content).hexdigest() + except Exception: + digest = None + return text, digest + + +def _sha256_file(path: Path) -> Optional[str]: + try: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + except (OSError, PermissionError): + return None + + +def _canonical_sha256(value: Any) -> str: + encoded = json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _git_output(repo_root: Path, *args: str) -> Optional[str]: + try: + completed = subprocess.run( + ["git", *args], + cwd=repo_root, + capture_output=True, + check=True, + text=True, + timeout=_GIT_TIMEOUT_SECONDS, + ) + return completed.stdout.strip() + except (OSError, subprocess.SubprocessError): + return None + + +def _git_provenance(repo_root: Path) -> Dict[str, Any]: + commit = _git_output(repo_root, "rev-parse", "HEAD") + status = _git_output(repo_root, "status", "--porcelain") + return { + "commit": commit or None, + "dirty": None if status is None else bool(status), + } + + +def collect_source_metadata(detection_root: Path) -> Dict[str, Any]: + """Capture source and lockfile metadata before detector analysis starts.""" + try: + source = _git_provenance(detection_root.parent) + except Exception: + source = {"commit": None, "dirty": None} + try: + uv_lock = _sha256_file(detection_root / "uv.lock") + except Exception: + uv_lock = None + return {"source": source, "uv_lock": uv_lock} + + +def _task_id(task_dir: Path) -> Optional[int]: + try: + return int(task_dir.name.removeprefix("task_")) + except ValueError: + return None + + +def _conversation_provenance( + task_dirs: Iterable[Path], conversation_hashes: Mapping[str, Optional[str]] +) -> Dict[str, Any]: + selected: list[Dict[str, Any]] = [] + missing: list[int] = [] + for task_dir in task_dirs: + task_id = _task_id(task_dir) + if task_id is None: + continue + digest = conversation_hashes.get(task_dir.name) + if digest is None: + missing.append(task_id) + else: + selected.append({"task_id": task_id, "sha256": digest}) + + selected.sort(key=lambda item: int(item["task_id"])) + missing.sort() + return { + "algorithm": _HASH_ALGORITHM, + "count": len(selected), + "aggregate_sha256": _canonical_sha256(selected), + "missing_task_ids": missing, + } + + +def _effective_labels_provenance( + selected_task_ids: Sequence[int], effective_labels: Mapping[str, bool] +) -> Dict[str, Any]: + labels = [] + missing = [] + for task_id in sorted(selected_task_ids): + key = f"task_{task_id:03d}" + if key not in effective_labels: + missing.append(task_id) + else: + labels.append({"task_id": task_id, "is_malicious": bool(effective_labels[key])}) + return { + "algorithm": _HASH_ALGORITHM, + "count": len(labels), + "sha256": _canonical_sha256(labels), + "missing_task_ids": missing, + } + + +def collect_run_manifest( + *, + benchmark_type: str, + task_dirs: Sequence[Path], + effective_labels: Mapping[str, bool], + resolved_concurrency: int, + conversation_hashes: Mapping[str, Optional[str]], + artifact_hashes: Mapping[str, Optional[str]], + source: Mapping[str, Any], +) -> Dict[str, Any]: + """Collect nonfatal, privacy-safe provenance for one detector run. + + Assemble only metadata captured at input load time, without rereading files + or Git after analysis. Exclude paths, directory basenames, host names, + environment values, and file contents. Any unavailable metadata is null. + """ + selected_task_ids = sorted( + task_id for task_dir in task_dirs if (task_id := _task_id(task_dir)) is not None + ) + + try: + conversations = _conversation_provenance(task_dirs, conversation_hashes) + except Exception: + conversations = None + try: + labels = _effective_labels_provenance(selected_task_ids, effective_labels) + except Exception: + labels = None + artifacts = { + "config_detector": artifact_hashes.get("config_detector"), + "uv_lock": artifact_hashes.get("uv_lock"), + "tasks": artifact_hashes.get("tasks") if benchmark_type == "adr_bench" else None, + "agentdojo_ground_truth": ( + artifact_hashes.get("agentdojo_ground_truth") if benchmark_type == "agentdojo" else None + ), + } + + return { + "schema_version": 1, + "kind": "run_provenance", + "benchmark_type": benchmark_type, + "resolved_concurrency": resolved_concurrency, + "selected_task_ids": selected_task_ids, + "source": {"commit": source.get("commit"), "dirty": source.get("dirty")}, + "inputs": { + "conversations": conversations, + "effective_labels": labels, + "artifacts": {"algorithm": _HASH_ALGORITHM, **artifacts}, + }, + "runtime": {"python_version": sys.version.split()[0]}, + } diff --git a/Detection/tests/test_main_detector.py b/Detection/tests/test_main_detector.py index 299b242..962b0a8 100644 --- a/Detection/tests/test_main_detector.py +++ b/Detection/tests/test_main_detector.py @@ -98,6 +98,10 @@ def test_includes_run_stats_in_result_and_json(self, tmp_path: Path): assert result["run_stats"] == {"total_tasks": 1, "scored": 1, "dropped": 0} saved = json.loads(json.dumps(result)) assert saved["run_stats"] == {"total_tasks": 1, "scored": 1, "dropped": 0} + assert saved["run_manifest"]["kind"] == "run_provenance" + assert saved["run_manifest"]["benchmark_type"] == "adr_bench" + assert saved["run_manifest"]["selected_task_ids"] == [1] + assert {"detector_info", "analyses", "metrics", "run_stats", "analysis_timestamp"} <= set(saved) class TestValidateBenchmarkResultsDir: diff --git a/Detection/tests/test_packaging.py b/Detection/tests/test_packaging.py new file mode 100644 index 0000000..ca8e55c --- /dev/null +++ b/Detection/tests/test_packaging.py @@ -0,0 +1,17 @@ +"""Regression tests for the detector's installed entry point.""" + +from pathlib import Path + +try: + import tomllib +except ModuleNotFoundError: # Python 3.10; installed with pytest. + import tomli as tomllib + + +def test_run_manifest_is_included_in_detector_wheel(detection_root: Path): + with (detection_root / "pyproject.toml").open("rb") as handle: + project = tomllib.load(handle) + + force_include = project["tool"]["hatch"]["build"]["targets"]["wheel"]["force-include"] + assert force_include.get("run_manifest.py") == "run_manifest.py" + assert (detection_root / "run_manifest.py").is_file() diff --git a/Detection/tests/test_run_manifest.py b/Detection/tests/test_run_manifest.py new file mode 100644 index 0000000..5e6872e --- /dev/null +++ b/Detection/tests/test_run_manifest.py @@ -0,0 +1,288 @@ +"""Tests for privacy-safe detector run provenance.""" + +import hashlib +import json +import subprocess +from pathlib import Path + +import pytest + +import run_manifest +from run_manifest import collect_run_manifest, collect_source_metadata, read_text_with_sha256 + + +def _task(results_dir: Path, task_id: int, content: str = "{}") -> Path: + task_dir = results_dir / f"task_{task_id:03d}" + workspace = task_dir / "workspace" + workspace.mkdir(parents=True) + (workspace / "claude_conversation.json").write_text(content) + return task_dir + + +def _collect(task_dirs, **overrides): + arguments = { + "benchmark_type": "adr_bench", + "task_dirs": task_dirs, + "effective_labels": {"task_001": False, "task_002": True}, + "resolved_concurrency": 7, + "conversation_hashes": {}, + "artifact_hashes": { + "config_detector": None, + "uv_lock": None, + "tasks": None, + "agentdojo_ground_truth": None, + }, + "source": {"commit": None, "dirty": None}, + } + arguments.update(overrides) + return collect_run_manifest(**arguments) + + +def test_hashes_selected_conversations_and_labels_in_task_order(tmp_path: Path): + results = tmp_path / "results-secret-name" + second = _task(results, 2, '{"message":"second"}') + first = _task(results, 1, '{"message":"first"}') + conversation_hashes = { + task.name: read_text_with_sha256(task / "workspace" / "claude_conversation.json")[1] + for task in (second, first) + } + conversation_hashes["task_003"] = hashlib.sha256(b"not selected").hexdigest() + + manifest = _collect([second, first], conversation_hashes=conversation_hashes) + + assert manifest["selected_task_ids"] == [1, 2] + conversations = manifest["inputs"]["conversations"] + expected = [ + {"task_id": task_id, "sha256": hashlib.sha256(content.encode()).hexdigest()} + for task_id, content in ((1, '{"message":"first"}'), (2, '{"message":"second"}')) + ] + expected_hash = hashlib.sha256( + json.dumps(expected, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + assert conversations["aggregate_sha256"] == expected_hash + assert conversations["count"] == 2 + assert manifest["inputs"]["effective_labels"]["count"] == 2 + assert manifest["resolved_concurrency"] == 7 + + +def test_agentdojo_hashes_actual_ground_truth_file(tmp_path: Path): + results = tmp_path / "run" + task = _task(results, 1) + ground_truth = b'{"task_001":{"is_malicious":true}}' + ground_truth_path = results / "ground_truth.json" + ground_truth_path.write_bytes(ground_truth) + _, ground_truth_hash = read_text_with_sha256(ground_truth_path) + + manifest = _collect( + [task], + benchmark_type="agentdojo", + effective_labels={"task_001": True}, + artifact_hashes={ + "config_detector": None, + "uv_lock": None, + "tasks": None, + "agentdojo_ground_truth": ground_truth_hash, + }, + ) + + artifacts = manifest["inputs"]["artifacts"] + assert artifacts["agentdojo_ground_truth"] == hashlib.sha256(ground_truth).hexdigest() + assert artifacts["tasks"] is None + + +def test_missing_inputs_and_metadata_are_nonfatal_and_explicit(tmp_path: Path, monkeypatch): + detection_root = tmp_path / "Detection" + detection_root.mkdir() + results = tmp_path / "run" + task = results / "task_001" + task.mkdir(parents=True) + monkeypatch.setattr(run_manifest, "_git_output", lambda *args: None) + source_metadata = collect_source_metadata(detection_root) + + manifest = _collect([task], effective_labels={}, source=source_metadata["source"]) + + assert source_metadata == {"source": {"commit": None, "dirty": None}, "uv_lock": None} + assert manifest["source"] == {"commit": None, "dirty": None} + assert manifest["inputs"]["conversations"]["missing_task_ids"] == [1] + assert manifest["inputs"]["effective_labels"]["missing_task_ids"] == [1] + assert manifest["inputs"]["artifacts"]["config_detector"] is None + assert manifest["inputs"]["artifacts"]["uv_lock"] is None + + +def test_git_calls_are_bounded_and_report_clean_or_dirty(tmp_path: Path): + repo = tmp_path / "repo" + detection_root = repo / "Detection" + detection_root.mkdir(parents=True) + subprocess.run(["git", "init", "-q", str(repo)], check=True) + subprocess.run(["git", "-C", str(repo), "config", "user.email", "test@example.com"], check=True) + subprocess.run(["git", "-C", str(repo), "config", "user.name", "Test"], check=True) + (detection_root / "tracked.txt").write_text("clean") + subprocess.run(["git", "-C", str(repo), "add", "."], check=True) + subprocess.run(["git", "-C", str(repo), "commit", "-qm", "initial"], check=True) + clean = collect_source_metadata(detection_root) + assert clean["source"]["commit"] + assert clean["source"]["dirty"] is False + + (detection_root / "untracked.txt").write_text("dirty") + untracked = collect_source_metadata(detection_root) + assert untracked["source"]["dirty"] is True + (detection_root / "untracked.txt").unlink() + + (detection_root / "tracked.txt").write_text("dirty") + dirty = collect_source_metadata(detection_root) + assert dirty["source"]["dirty"] is True + + +def test_manifest_excludes_paths_basenames_and_host_identifiers(tmp_path: Path): + results = tmp_path / "customer-secret-benchmark" + task = _task(results, 1) + + serialized = json.dumps(_collect([task])) + + assert "customer-secret-benchmark" not in serialized + assert str(tmp_path) not in serialized + assert "hostname" not in serialized + assert "platform" not in serialized + assert "path" not in serialized + + +def test_git_commands_have_a_short_timeout(tmp_path: Path, monkeypatch): + calls = [] + + def fake_run(*args, **kwargs): + calls.append(kwargs) + return subprocess.CompletedProcess(args[0], 0, stdout="abc123\n", stderr="") + + monkeypatch.setattr(run_manifest.subprocess, "run", fake_run) + assert run_manifest._git_output(tmp_path, "rev-parse", "HEAD") == "abc123" + assert calls[0]["timeout"] == run_manifest._GIT_TIMEOUT_SECONDS == 2 + + +def test_git_timeout_is_nonfatal(tmp_path: Path, monkeypatch): + def timeout(*args, **kwargs): + raise subprocess.TimeoutExpired(args[0], kwargs["timeout"]) + + monkeypatch.setattr(run_manifest.subprocess, "run", timeout) + + assert run_manifest._git_output(tmp_path, "rev-parse", "HEAD") is None + + +def test_text_load_hashes_raw_bytes_and_preserves_universal_newlines(tmp_path: Path): + path = tmp_path / "conversation.json" + raw = b'{\r\n"message": "caf\xc3\xa9"\r\n}\r' + path.write_bytes(raw) + + content, digest = read_text_with_sha256(path, encoding="utf-8") + + assert content == '{\n"message": "caf\u00e9"\n}\n' + assert digest == hashlib.sha256(raw).hexdigest() + assert digest != hashlib.sha256(content.encode()).hexdigest() + + +def test_text_and_hash_are_obtained_from_one_file_read(tmp_path: Path, monkeypatch): + path = tmp_path / "conversation.json" + raw = b'{"message":"original"}' + path.write_bytes(raw) + original_open = Path.open + calls = [] + + def track_open(input_path, *args, **kwargs): + calls.append(input_path) + return original_open(input_path, *args, **kwargs) + + monkeypatch.setattr(Path, "open", track_open) + + content, digest = read_text_with_sha256(path, encoding="utf-8") + + assert calls == [path] + assert content == raw.decode("utf-8") + assert digest == hashlib.sha256(raw).hexdigest() + + +def test_hashing_failure_does_not_prevent_text_load(tmp_path: Path, monkeypatch): + path = tmp_path / "conversation.json" + path.write_bytes(b'{"message":"still available"}') + + def failed_hash(*args, **kwargs): + raise RuntimeError("hashing unavailable") + + monkeypatch.setattr(run_manifest.hashlib, "sha256", failed_hash) + + content, digest = read_text_with_sha256(path, encoding="utf-8") + + assert content == '{"message":"still available"}' + assert digest is None + + +def test_text_load_still_propagates_file_and_decode_errors(tmp_path: Path): + path = tmp_path / "conversation.json" + with pytest.raises(FileNotFoundError): + read_text_with_sha256(path) + + path.write_bytes(b"\xff") + with pytest.raises(UnicodeDecodeError): + read_text_with_sha256(path, encoding="utf-8") + + +def test_source_metadata_failures_are_nonfatal(tmp_path: Path, monkeypatch): + def failed_metadata(*args, **kwargs): + raise RuntimeError("metadata unavailable") + + monkeypatch.setattr(run_manifest, "_git_provenance", failed_metadata) + monkeypatch.setattr(run_manifest, "_sha256_file", failed_metadata) + + assert collect_source_metadata(tmp_path) == { + "source": {"commit": None, "dirty": None}, + "uv_lock": None, + } + + +def test_final_manifest_uses_captured_inputs_without_rereading(tmp_path: Path, monkeypatch): + detection_root = tmp_path / "Detection" + detection_root.mkdir() + lock = detection_root / "uv.lock" + lock.write_text("original lock") + config = detection_root / "config_detector.yaml" + config.write_text("model: original\n") + tasks = detection_root / "tasks.json" + tasks.write_text('{"tasks":[]}') + task = _task(tmp_path / "results", 1, '{"message":"original"}') + conversation = task / "workspace" / "claude_conversation.json" + monkeypatch.setattr( + run_manifest, + "_git_output", + lambda _, *args: "original-commit" if args[0] == "rev-parse" else "", + ) + source_metadata = collect_source_metadata(detection_root) + _, conversation_hash = read_text_with_sha256(conversation) + _, config_hash = read_text_with_sha256(config) + _, tasks_hash = read_text_with_sha256(tasks) + captured = { + "source": source_metadata["source"], + "conversation_hashes": {task.name: conversation_hash}, + "artifact_hashes": { + "config_detector": config_hash, + "uv_lock": source_metadata["uv_lock"], + "tasks": tasks_hash, + "agentdojo_ground_truth": None, + }, + } + before = _collect([task], **captured) + conversation.unlink() + config.write_text("model: replacement\n") + tasks.write_text('{"tasks":["replacement"]}') + lock.write_text("replacement lock") + + def unexpected_read(*args, **kwargs): + raise AssertionError("manifest assembly must not read input files or Git") + + monkeypatch.setattr(Path, "open", unexpected_read) + monkeypatch.setattr(run_manifest, "_sha256_file", unexpected_read) + monkeypatch.setattr(run_manifest, "_git_output", unexpected_read) + + after = _collect([task], **captured) + + assert after == before + assert after["source"] == {"commit": "original-commit", "dirty": False} + assert after["inputs"]["conversations"]["count"] == 1 + assert after["inputs"]["artifacts"]["uv_lock"] == hashlib.sha256(b"original lock").hexdigest() diff --git a/Detection/tests/test_run_manifest_integration.py b/Detection/tests/test_run_manifest_integration.py new file mode 100644 index 0000000..48fe831 --- /dev/null +++ b/Detection/tests/test_run_manifest_integration.py @@ -0,0 +1,191 @@ +"""Run provenance must describe the bytes consumed, not later on-disk state.""" + +import hashlib +import json +import sys +from pathlib import Path + +import pytest + +import main_detector +from guardrail.base_detector import DetectionResult +from main_detector import BenchmarkAnalyzer + + +def _sha256(content: str) -> str: + return hashlib.sha256(content.encode()).hexdigest() + + +def _conversation_hash(content: str) -> str: + return _sha256( + json.dumps( + [{"task_id": 1, "sha256": _sha256(content)}], + sort_keys=True, + separators=(",", ":"), + ) + ) + + +def _inputs(tmp_path: Path, benchmark_type: str): + bench = tmp_path / f"{benchmark_type}_synthetic" + workspace = bench / "task_001" / "workspace" + workspace.mkdir(parents=True) + conversation = workspace / "claude_conversation.json" + content = json.dumps([{"role": "user", "content": "original synthetic input"}]) + conversation.write_text(content) + if benchmark_type == "agentdojo": + labels = bench / "ground_truth.json" + label_content = json.dumps({"task_001": {"is_malicious": False}}) + else: + labels = tmp_path / "tasks.json" + label_content = json.dumps( + {"tasks": [{"task_id": 1, "ground_truth": "benign", "mcp_servers": ["original"]}]} + ) + labels.write_text(label_content) + return bench, conversation, content, labels, label_content + + +class _StubDetector: + def get_info(self): + return {"name": "StubDetector"} + + def analyze_task(self, task_data): + return DetectionResult( + task_id=task_data["task_id"], + is_malicious=False, + confidence_score=0.1, + total_messages=len(task_data["messages"]), + threat_messages=0, + detections=[], + method="synthetic", + analysis_time=0.01, + ) + + +@pytest.mark.parametrize("benchmark_type", ["adr_bench", "agentdojo"]) +def test_manifest_retains_consumed_inputs_after_files_change(tmp_path, monkeypatch, benchmark_type): + bench, conversation, content, labels, label_content = _inputs(tmp_path, benchmark_type) + monkeypatch.chdir(tmp_path) + source = {"commit": "before-analysis", "dirty": False} + metadata_calls = [] + + def source_metadata(root): + metadata_calls.append(root) + return {"source": dict(source), "uv_lock": "lock-before-analysis"} + + monkeypatch.setattr(main_detector, "collect_source_metadata", source_metadata) + + class MutatingDetector(_StubDetector): + def analyze_task(self, task_data): + assert task_data["messages"][-1]["content"] == "original synthetic input" + conversation.write_text("[]") + labels.unlink() + source.update(commit="after-analysis", dirty=True) + return super().analyze_task(task_data) + + result = BenchmarkAnalyzer(MutatingDetector()).process_benchmark_results( + str(bench), benchmark_type=benchmark_type, max_concurrent=1 + ) + manifest = result["run_manifest"] + assert result["run_stats"] == {"total_tasks": 1, "scored": 1, "dropped": 0} + assert manifest["inputs"]["conversations"]["aggregate_sha256"] == _conversation_hash(content) + artifact = "tasks" if benchmark_type == "adr_bench" else "agentdojo_ground_truth" + assert manifest["inputs"]["artifacts"][artifact] == _sha256(label_content) + assert manifest["source"] == {"commit": "before-analysis", "dirty": False} + assert manifest["inputs"]["artifacts"]["uv_lock"] == "lock-before-analysis" + assert manifest["inputs"]["artifacts"]["config_detector"] is None + assert len(metadata_calls) == 1 + + +def test_labels_and_mcp_definitions_share_one_snapshot(tmp_path, monkeypatch): + bench, _, _, tasks_file, original = _inputs(tmp_path, "adr_bench") + monkeypatch.chdir(tmp_path) + + class ReplacingAnalyzer(BenchmarkAnalyzer): + def _load_ground_truth(self, benchmark_type): + labels = super()._load_ground_truth(benchmark_type) + tasks_file.write_text( + json.dumps( + {"tasks": [{"task_id": 1, "ground_truth": "malicious", "mcp_servers": ["new"]}]} + ) + ) + return labels + + class InspectingDetector(_StubDetector): + def analyze_task(self, task_data): + assert task_data["messages"][0]["mcp_servers"] == ["original"] + return super().analyze_task(task_data) + + result = ReplacingAnalyzer(InspectingDetector()).process_benchmark_results(str(bench)) + assert result["run_stats"]["scored"] == 1 + assert result["analyses"][0]["ground_truth_binary"] is False + assert result["run_manifest"]["inputs"]["artifacts"]["tasks"] == _sha256(original) + + +def test_reused_analyzer_does_not_retain_previous_conversation_hash(tmp_path, monkeypatch): + bench, conversation, _, _, _ = _inputs(tmp_path, "adr_bench") + monkeypatch.chdir(tmp_path) + analyzer = BenchmarkAnalyzer(_StubDetector()) + first = analyzer.process_benchmark_results(str(bench)) + conversation.unlink() + second = analyzer.process_benchmark_results(str(bench)) + assert first["run_manifest"]["inputs"]["conversations"]["count"] == 1 + assert second["run_manifest"]["inputs"]["conversations"]["count"] == 0 + assert second["run_manifest"]["inputs"]["conversations"]["missing_task_ids"] == [1] + + +def test_malformed_conversation_retains_attempted_input_hash(tmp_path, monkeypatch): + bench, conversation, _, _, _ = _inputs(tmp_path, "adr_bench") + monkeypatch.chdir(tmp_path) + malformed = '{"messages": [' + conversation.write_text(malformed) + result = BenchmarkAnalyzer(_StubDetector()).process_benchmark_results(str(bench)) + assert result["run_stats"] == {"total_tasks": 1, "scored": 0, "dropped": 1} + provenance = result["run_manifest"]["inputs"]["conversations"] + assert provenance["aggregate_sha256"] == _conversation_hash(malformed) + assert provenance["missing_task_ids"] == [] + + +def test_cli_hashes_consumed_working_directory_config_and_tasks(tmp_path, monkeypatch): + bench, _, _, tasks_file, _ = _inputs(tmp_path, "adr_bench") + config_file = tmp_path / "config_detector.yaml" + observed = {} + + class ConfiguredDetector(_StubDetector): + def __init__(self, model_name, **kwargs): + observed["model"] = model_name + # Model setup may take time: later edits must not replace the hash. + config_file.write_text("llamafirewall:\n model: unconsumed-replacement\n") + + def is_available(self): + return True + + def analyze_task(self, task_data): + observed["mcp_servers"] = task_data["messages"][0]["mcp_servers"] + return super().analyze_task(task_data) + + monkeypatch.setattr(main_detector, "LlamaFirewallBaseline", ConfiguredDetector) + monkeypatch.setattr(main_detector, "print_analysis_summary", lambda analysis: None) + monkeypatch.setattr( + sys, + "argv", + ["main_detector.py", "--detector", "llamafirewall", "--results-dir", str(bench)], + ) + monkeypatch.chdir(tmp_path) + manifests = [] + for version in ("first", "second"): + config = f"llamafirewall:\n model: synthetic-{version}\n" + tasks = json.dumps( + {"tasks": [{"task_id": 1, "ground_truth": "benign", "mcp_servers": [version]}]} + ) + config_file.write_text(config) + tasks_file.write_text(tasks) + main_detector.main() + assert observed == {"model": f"synthetic-{version}", "mcp_servers": [version]} + saved = json.loads((bench / "llamafirewall_baseline_analysis.json").read_text()) + assert saved["run_stats"]["scored"] == 1 + manifest = saved["run_manifest"] + assert manifest["inputs"]["artifacts"]["config_detector"] == _sha256(config) + assert manifest["inputs"]["artifacts"]["tasks"] == _sha256(tasks) + manifests.append(manifest) + assert manifests[0]["inputs"]["artifacts"] != manifests[1]["inputs"]["artifacts"] diff --git a/docs/REPRODUCIBILITY.md b/docs/REPRODUCIBILITY.md index 49e7b5c..f2680ae 100644 --- a/docs/REPRODUCIBILITY.md +++ b/docs/REPRODUCIBILITY.md @@ -125,6 +125,10 @@ Without `--results-dir`, `main_detector.py` uses the latest `adr_bench_*` direct The summary reports **tasks scored N/M**; dropped tasks (missing conversation or errors) are excluded from metrics and flagged with a warning. Each `*_baseline_analysis.json` also includes a `run_stats` object (`total_tasks`, `scored`, `dropped`). +Each analysis JSON also includes an additive `run_manifest` for **run provenance**. It records the source commit and dirty state when available, resolved detector concurrency, sorted selected task IDs, and SHA-256 digests covering the selected conversation inputs and their effective labels. AgentDojo runs also hash that run's `ground_truth.json`. Fixed detector artifacts (`config_detector.yaml`, `uv.lock`, and ADR-Bench `tasks.json`) are hashed when available. Collection is best-effort and nonfatal; unavailable values are `null`. The manifest omits paths, arbitrary directory names, host identifiers, environment values, prompts, and file contents. It helps compare runs and investigate regressions, but does not guarantee reproducibility. + +Input hashes describe the bytes actually read, even if files are edited or removed during analysis. Configuration and ADR-Bench task definitions are loaded from the current working directory, and the same task snapshot supplies both labels and MCP definitions. Git and lockfile metadata are captured before analysis. Conversation hashes are captured when each selected file is read; unreadable files are listed in `missing_task_ids`. This is not an atomic snapshot of the entire directory: keep benchmark inputs unchanged during a run when comparing results. + Outputs are written into the benchmark directory: ``` @@ -215,4 +219,4 @@ uv run python main_detector.py --detector llamafirewall --tasks 108 --results-di uv run python main_benchmark.py --tasks 1-10 ``` -See [Detection/README.md](../Detection/README.md) for MCP server debugging and benchmark extension. \ No newline at end of file +See [Detection/README.md](../Detection/README.md) for MCP server debugging and benchmark extension.