diff --git a/scripts/probe_complex_benchmark_replay_feasibility.py b/scripts/probe_complex_benchmark_replay_feasibility.py new file mode 100644 index 0000000..d9b13a4 --- /dev/null +++ b/scripts/probe_complex_benchmark_replay_feasibility.py @@ -0,0 +1,376 @@ +#!/usr/bin/env python3 +"""Probe replay feasibility for the complex synthetic benchmark fixture.""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from collections import Counter +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[1] +DEFAULT_FIXTURE_DIR = REPO_ROOT / "tests" / "fixtures" / "demo" / "complex_shared_uncertainty_iam_benchmark" +DEFAULT_OUT = Path("/tmp/iamscope-complex-replay-feasibility") + +REQUIRED_PATTERN_ORDER = [ + "passrole_lambda", + "passrole_ecs", + "assume_role_chain", + "cross_account_trust", + "permission_boundary_blocked_path", + "scp_blocked_path", + "identity_deny_suppressed_path", + "missing_iam_passrole_precondition", + "missing_target_trust_precondition", + "shared_uncertainty_grouping", +] + +SUPPORTED_CLAIMS = [ + "The probe reports replay feasibility for the local frozen complex synthetic benchmark fixture.", + "The probe identifies replay input-contract gaps without changing reasoners or fixture semantics.", + "The probe keeps unsupported and static-only rows labeled as such.", +] + +NON_CLAIMS = [ + "no broad IAMScope correctness", + "no broad PassRole correctness", + "no generic Deny correctness", + "no resource-policy Deny support", + "no SCP Deny support beyond selected synthetic fixture behavior", + "no exploitability proof", + "no downstream authorization proof", + "no Lambda invocation behavior", + "no production readiness", + "no correctness for real AWS environments", + "no correctness for other principals, roles, accounts, regions, conditions, permission boundaries, SCPs, resource policies, or findings", + "no composite benchmark score", + "no pass/fail benchmark label", +] + + +def _is_relative_to(path: Path, parent: Path) -> bool: + try: + path.relative_to(parent) + except ValueError: + return False + return True + + +def _resolve_out(output_dir: Path) -> Path: + output_abs = output_dir.expanduser().resolve() + repo_abs = REPO_ROOT.resolve() + if _is_relative_to(output_abs, repo_abs): + raise ValueError(f"refusing to write replay feasibility outputs inside repository tree: {output_abs}") + return output_abs + + +def _load_json(path: Path) -> dict[str, Any]: + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError(f"expected JSON object in {path}") + return payload + + +def _load_fixture(fixture_dir: Path) -> dict[str, dict[str, Any]]: + required = [ + "scenario.json", + "binding_metadata.json", + "findings.json", + "naive_candidates.json", + "expected_uncertainty_groups.json", + ] + return {name: _load_json(fixture_dir / name) for name in required} + + +def _scenario_contract_status(scenario: dict[str, Any]) -> tuple[str, list[str]]: + gaps: list[str] = [] + if "metadata" not in scenario: + gaps.append("scenario.json lacks emitted IAMScope metadata with canonical_hash") + if "constraints" not in scenario: + gaps.append("scenario.json lacks emitted constraints array") + for index, edge in enumerate(scenario.get("edges", [])): + if not isinstance(edge, dict): + gaps.append(f"scenario edge {index} is not an object") + continue + if not isinstance(edge.get("src"), dict) or not isinstance(edge.get("dst"), dict): + gaps.append(f"scenario edge {edge.get('edge_id', index)} uses descriptive src/dst ids instead of NodeRef objects") + break + if "features" not in edge: + gaps.append(f"scenario edge {edge.get('edge_id', index)} lacks emitted features object") + break + return ("replay_ready" if not gaps else "not_replay_ready", gaps) + + +def _binding_contract_status(binding_metadata: dict[str, Any]) -> tuple[str, list[str]]: + gaps: list[str] = [] + if not isinstance(binding_metadata, list): + gaps.append("binding_metadata.json is descriptive object metadata, not an edge-constraint binding sidecar list") + return "not_replay_ready", gaps + for index, entry in enumerate(binding_metadata): + if not isinstance(entry, dict): + gaps.append(f"binding metadata entry {index} is not an object") + continue + for key in ["edge_id", "constraint_id", "binding_metadata"]: + if key not in entry: + gaps.append(f"binding metadata entry {index} lacks {key}") + return ("replay_ready" if not gaps else "not_replay_ready", gaps) + + +def _pattern_feasibility() -> list[dict[str, str]]: + return [ + { + "pattern_category": "passrole_lambda", + "replayable_now": "partial", + "status": "blocked_by_input_contract", + "reason": "PassRoleLambdaReasoner exists, but the fixture scenario is not currently in emitted replay-ready FactGraph shape.", + "recommended_next_action": "Add a local conversion/readiness probe for replay-compatible PassRole-to-Lambda rows.", + }, + { + "pattern_category": "passrole_ecs", + "replayable_now": "partial", + "status": "blocked_by_input_contract", + "reason": "PassRoleEcsReasoner exists, but the fixture lacks replay-ready edge features and sidecar bindings.", + "recommended_next_action": "Probe ECS rows separately once input-contract compatibility is available.", + }, + { + "pattern_category": "assume_role_chain", + "replayable_now": "partial", + "status": "blocked_by_input_contract", + "reason": "AssumeRoleChainReasoner is covered by existing replay tests, but this fixture is descriptive rather than emitted scenario format.", + "recommended_next_action": "Try existing replay on a replay-ready converted subset, without changing the frozen oracle.", + }, + { + "pattern_category": "cross_account_trust", + "replayable_now": "partial", + "status": "blocked_by_input_contract_and_synthetic_cross_account_shape", + "reason": "CrossAccountTrustReasoner exists, but synthetic same-account placeholders may not encode cross-account classification.", + "recommended_next_action": "Report whether sanitized synthetic cross-account attributes can be replayed without real account IDs.", + }, + { + "pattern_category": "permission_boundary_blocked_path", + "replayable_now": "partial", + "status": "requires_replay_compatible_constraints", + "reason": "Boundary blockers require emitted Constraint and EdgeConstraint objects; current fixture has descriptive blocker rows.", + "recommended_next_action": "Keep rows static until replay-compatible boundary constraints are present.", + }, + { + "pattern_category": "scp_blocked_path", + "replayable_now": "partial", + "status": "requires_replay_compatible_constraints", + "reason": "SCP blockers require concrete constraints and binding metadata; fixture support is selected synthetic behavior only.", + "recommended_next_action": "Keep SCP rows scoped and static until sidecar constraints are proven replayable.", + }, + { + "pattern_category": "identity_deny_suppressed_path", + "replayable_now": "partial", + "status": "requires_selected_fixture_constraint_binding", + "reason": "Identity-Deny suppression must remain selected-fixture behavior and needs precise constraint binding before replay.", + "recommended_next_action": "Probe selected identity-Deny rows only; do not broaden to generic Deny correctness.", + }, + { + "pattern_category": "missing_iam_passrole_precondition", + "replayable_now": "no", + "status": "static_only_expected_absence", + "reason": "Missing iam:PassRole preconditions usually produce no selected finding; absence is not a generated finding.", + "recommended_next_action": "Compare as expected absence or non-finding reason, not replayed finding output.", + }, + { + "pattern_category": "missing_target_trust_precondition", + "replayable_now": "no", + "status": "static_only_expected_absence", + "reason": "Missing target trust preconditions usually produce no selected finding; frozen precondition-only rows are report rows.", + "recommended_next_action": "Keep as static-only unless a report-only precondition explanation layer is added.", + }, + { + "pattern_category": "shared_uncertainty_grouping", + "replayable_now": "no", + "status": "report_only_static_grouping", + "reason": "Shared uncertainty classes are report-only fixture metadata, not existing reasoner output.", + "recommended_next_action": "Compare grouping separately from generated finding replay.", + }, + ] + + +def _static_only_rows(findings: dict[str, Any], naive: dict[str, Any]) -> list[dict[str, str]]: + static_rows: list[dict[str, str]] = [] + for finding in findings["findings"]: + if finding["verdict"] == "precondition_only" or finding.get("uncertainty_class"): + static_rows.append( + { + "row_id": str(finding["finding_id"]), + "kind": "finding", + "status": "static_only", + "reason": "precondition-only or shared uncertainty row is not currently proven replayable", + } + ) + for candidate in naive["candidate_paths"]: + mapping = candidate.get("maps_to", {}) + if mapping.get("non_finding_reason"): + static_rows.append( + { + "row_id": str(candidate["candidate_id"]), + "kind": "naive_candidate_non_finding", + "status": "static_only", + "reason": str(mapping["non_finding_reason"]), + } + ) + return static_rows + + +def build_probe_result(fixture_dir: Path = DEFAULT_FIXTURE_DIR) -> dict[str, Any]: + fixture = _load_fixture(fixture_dir) + scenario = fixture["scenario.json"] + binding = fixture["binding_metadata.json"] + findings = fixture["findings.json"] + naive = fixture["naive_candidates.json"] + uncertainty = fixture["expected_uncertainty_groups.json"] + + scenario_status, scenario_gaps = _scenario_contract_status(scenario) + binding_status, binding_gaps = _binding_contract_status(binding) + input_contract_status = "replay_ready" if scenario_status == binding_status == "replay_ready" else "not_replay_ready" + safe_replay_attempted = input_contract_status == "replay_ready" + + verdict_breakdown = dict(Counter(finding["verdict"] for finding in findings["findings"])) + group_counts = {group["uncertainty_class"]: group["count"] for group in uncertainty["groups"]} + + return { + "schema_version": "complex_benchmark_replay_feasibility_probe.v1", + "timestamp": datetime.now(timezone.utc).isoformat(), + "fixture_id": findings["fixture_id"], + "fixture_path": str(fixture_dir), + "fixture_generation_mode": findings["generation_mode"], + "fixture_source_tool": findings["source_tool"], + "generated_or_replayed_by_iamscope": findings["generated_or_replayed_by_iamscope"], + "reasoners_run_in_fixture": findings["reasoners_run"], + "replay_equivalence_status": "not_proven", + "input_contract_status": input_contract_status, + "input_contract_gaps": { + "scenario_json": scenario_gaps, + "binding_metadata_json": binding_gaps, + }, + "safe_replay_attempted": safe_replay_attempted, + "reasoners_attempted": [] if not safe_replay_attempted else ["not_implemented_in_probe"], + "reasoners_skipped": { + "all": "input_contract_status_not_replay_ready" if not safe_replay_attempted else "probe_does_not_claim_equivalence" + }, + "oracle_counts": { + "naive_candidate_count": len(naive["candidate_paths"]), + "finding_count": len(findings["findings"]), + "verdict_breakdown": verdict_breakdown, + "uncertainty_group_counts": group_counts, + }, + "pattern_feasibility": _pattern_feasibility(), + "static_only_rows": _static_only_rows(findings, naive), + "unsupported_rows": [ + { + "category": "shared_uncertainty_grouping", + "reason": "report-only grouping is not existing reasoner output", + }, + { + "category": "precondition_only_rows", + "reason": "reasoner precondition failure is expected absence, not a generated finding", + }, + ], + "supported_claims": SUPPORTED_CLAIMS, + "non_claims": NON_CLAIMS, + "local_only": True, + "live_aws_used": False, + "aws_calls_made": 0, + "output_files_generated": ["replay-feasibility-summary.md", "replay-feasibility-manifest.json"], + } + + +def _render_summary(manifest: dict[str, Any], out_dir: Path) -> str: + pattern_lines = "\n".join( + f"- `{row['pattern_category']}`: {row['replayable_now']} ({row['status']})" + for row in manifest["pattern_feasibility"] + ) + gaps = [ + *manifest["input_contract_gaps"]["scenario_json"], + *manifest["input_contract_gaps"]["binding_metadata_json"], + ] + gap_lines = "\n".join(f"- {gap}" for gap in gaps) if gaps else "- none" + return f"""# Complex Benchmark Replay Feasibility Probe + +Replay-equivalence is not yet proven. + +The complex synthetic benchmark remains a frozen synthetic oracle. + +The current fixture is not generated/replayed IAMScope output. + +Unsupported or static-only rows remain labeled as such. + +No composite benchmark score or pass/fail benchmark label is produced. + +## Output + +Output directory: `{out_dir}` + +## Fixture + +- fixture id: `{manifest['fixture_id']}` +- fixture generation mode: `{manifest['fixture_generation_mode']}` +- generated/replayed by IAMScope: {str(manifest['generated_or_replayed_by_iamscope']).lower()} +- reasoners run in fixture: {manifest['reasoners_run_in_fixture']} +- replay equivalence status: `{manifest['replay_equivalence_status']}` +- input contract status: `{manifest['input_contract_status']}` +- safe replay attempted: {str(manifest['safe_replay_attempted']).lower()} + +## Input Contract Gaps + +{gap_lines} + +## Pattern Feasibility + +{pattern_lines} + +## Safety + +- local-only: true +- live AWS used: false +- AWS calls made: 0 +- no AWS credentials required +- no Terraform, AWS CLI, STS, Lambda API, or `iam:PassRole` calls +""" + + +def run_probe(output_dir: Path, *, fixture_dir: Path = DEFAULT_FIXTURE_DIR) -> dict[str, Any]: + out_abs = _resolve_out(output_dir) + out_abs.mkdir(parents=True, exist_ok=True) + manifest = build_probe_result(fixture_dir) + summary = _render_summary(manifest, out_abs) + (out_abs / "replay-feasibility-summary.md").write_text(summary, encoding="utf-8") + (out_abs / "replay-feasibility-manifest.json").write_text( + json.dumps(manifest, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + return manifest + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--out", default=str(DEFAULT_OUT), help=f"Output directory, default: {DEFAULT_OUT}") + parser.add_argument("--fixture-dir", default=str(DEFAULT_FIXTURE_DIR), help="Complex fixture directory") + args = parser.parse_args(argv) + try: + manifest = run_probe(Path(args.out), fixture_dir=Path(args.fixture_dir)) + except Exception as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + + print("IAMScope complex benchmark replay feasibility probe (local only)") + print(f"Output: {Path(args.out).expanduser().resolve()}") + print(f"Replay-equivalence status: {manifest['replay_equivalence_status']}") + print(f"Input contract status: {manifest['input_contract_status']}") + print(f"Safe replay attempted: {str(manifest['safe_replay_attempted']).lower()}") + print("Live AWS used: false") + print("AWS calls made: 0") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_complex_benchmark_replay_feasibility_probe.py b/tests/test_complex_benchmark_replay_feasibility_probe.py new file mode 100644 index 0000000..70d5760 --- /dev/null +++ b/tests/test_complex_benchmark_replay_feasibility_probe.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from scripts.probe_complex_benchmark_replay_feasibility import REPO_ROOT, build_probe_result, run_probe + +PROBE = REPO_ROOT / "scripts" / "probe_complex_benchmark_replay_feasibility.py" +FIXTURE_DIR = REPO_ROOT / "tests" / "fixtures" / "demo" / "complex_shared_uncertainty_iam_benchmark" +GENERATED_OUTPUTS = { + "replay-feasibility-summary.md", + "replay-feasibility-manifest.json", +} +REQUIRED_PATTERN_CATEGORIES = { + "passrole_lambda", + "passrole_ecs", + "assume_role_chain", + "cross_account_trust", + "permission_boundary_blocked_path", + "scp_blocked_path", + "identity_deny_suppressed_path", + "missing_iam_passrole_precondition", + "missing_target_trust_precondition", + "shared_uncertainty_grouping", +} + + +def test_probe_refuses_repo_output_paths() -> None: + with pytest.raises(ValueError, match="repository tree"): + run_probe(REPO_ROOT / "complex-replay-feasibility-output") + + +def test_probe_writes_summary_and_manifest(tmp_path: Path) -> None: + manifest = run_probe(tmp_path) + + summary_path = tmp_path / "replay-feasibility-summary.md" + manifest_path = tmp_path / "replay-feasibility-manifest.json" + assert summary_path.is_file() + assert manifest_path.is_file() + assert json.loads(manifest_path.read_text()) == manifest + + +def test_manifest_preserves_not_proven_local_only_boundary(tmp_path: Path) -> None: + manifest = run_probe(tmp_path) + + assert manifest["replay_equivalence_status"] == "not_proven" + assert manifest["fixture_generation_mode"] == "frozen_synthetic_oracle" + assert manifest["generated_or_replayed_by_iamscope"] is False + assert manifest["reasoners_run_in_fixture"] == [] + assert manifest["input_contract_status"] == "not_replay_ready" + assert manifest["safe_replay_attempted"] is False + assert manifest["local_only"] is True + assert manifest["live_aws_used"] is False + assert manifest["aws_calls_made"] == 0 + assert manifest["output_files_generated"] == [ + "replay-feasibility-summary.md", + "replay-feasibility-manifest.json", + ] + + +def test_manifest_explains_input_contract_gaps(tmp_path: Path) -> None: + manifest = run_probe(tmp_path) + scenario_gaps = manifest["input_contract_gaps"]["scenario_json"] + binding_gaps = manifest["input_contract_gaps"]["binding_metadata_json"] + + assert any("canonical_hash" in gap for gap in scenario_gaps) + assert any("constraints array" in gap for gap in scenario_gaps) + assert any("NodeRef objects" in gap for gap in scenario_gaps) + assert any("sidecar list" in gap for gap in binding_gaps) + + +def test_manifest_includes_required_pattern_feasibility_rows(tmp_path: Path) -> None: + manifest = run_probe(tmp_path) + rows = {row["pattern_category"]: row for row in manifest["pattern_feasibility"]} + + assert set(rows) == REQUIRED_PATTERN_CATEGORIES + assert rows["passrole_lambda"]["replayable_now"] == "partial" + assert rows["passrole_ecs"]["replayable_now"] == "partial" + assert rows["assume_role_chain"]["replayable_now"] == "partial" + assert rows["cross_account_trust"]["replayable_now"] == "partial" + assert rows["permission_boundary_blocked_path"]["replayable_now"] == "partial" + assert rows["scp_blocked_path"]["replayable_now"] == "partial" + assert rows["identity_deny_suppressed_path"]["replayable_now"] == "partial" + assert rows["missing_iam_passrole_precondition"]["replayable_now"] == "no" + assert rows["missing_iam_passrole_precondition"]["status"] == "static_only_expected_absence" + assert rows["missing_target_trust_precondition"]["replayable_now"] == "no" + assert rows["shared_uncertainty_grouping"]["status"] == "report_only_static_grouping" + + +def test_manifest_marks_static_only_and_unsupported_rows(tmp_path: Path) -> None: + manifest = run_probe(tmp_path) + + assert any(row["kind"] == "finding" for row in manifest["static_only_rows"]) + assert any(row["kind"] == "naive_candidate_non_finding" for row in manifest["static_only_rows"]) + assert {row["category"] for row in manifest["unsupported_rows"]} == { + "shared_uncertainty_grouping", + "precondition_only_rows", + } + + +def test_summary_states_required_boundaries(tmp_path: Path) -> None: + run_probe(tmp_path) + summary = (tmp_path / "replay-feasibility-summary.md").read_text() + + assert "Replay-equivalence is not yet proven." in summary + assert "The complex synthetic benchmark remains a frozen synthetic oracle." in summary + assert "The current fixture is not generated/replayed IAMScope output." in summary + assert "Unsupported or static-only rows remain labeled as such." in summary + assert "No composite benchmark score or pass/fail benchmark label is produced." in summary + + +def test_no_composite_score_or_pass_fail_manifest_fields(tmp_path: Path) -> None: + manifest = run_probe(tmp_path) + text = json.dumps(manifest).lower() + + assert "composite_score" not in text + assert "benchmark_passed" not in text + assert "pass/fail benchmark label" in text + + +def test_probe_does_not_require_aws_credentials(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + for key in list(os.environ): + if key.startswith("AWS_"): + monkeypatch.delenv(key, raising=False) + + manifest = run_probe(tmp_path) + + assert manifest["live_aws_used"] is False + assert manifest["aws_calls_made"] == 0 + + +def test_no_generated_outputs_are_committed() -> None: + committed = {path.name for path in FIXTURE_DIR.iterdir() if path.is_file()} + assert GENERATED_OUTPUTS.isdisjoint(committed) + + +def test_cli_writes_expected_outputs(tmp_path: Path) -> None: + result = subprocess.run( + [sys.executable, str(PROBE), "--out", str(tmp_path)], + cwd=REPO_ROOT, + text=True, + capture_output=True, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert "Replay-equivalence status: not_proven" in result.stdout + assert (tmp_path / "replay-feasibility-summary.md").is_file() + assert (tmp_path / "replay-feasibility-manifest.json").is_file() + + +def test_build_probe_result_matches_fixture_oracle_counts() -> None: + manifest = build_probe_result() + + assert manifest["fixture_id"] == "complex_shared_uncertainty_iam_benchmark_001" + assert manifest["oracle_counts"]["naive_candidate_count"] == 42 + assert manifest["oracle_counts"]["finding_count"] == 18 + assert manifest["oracle_counts"]["verdict_breakdown"] == { + "validated": 4, + "blocked": 5, + "precondition_only": 3, + "inconclusive": 6, + } + assert manifest["oracle_counts"]["uncertainty_group_counts"] == { + "shared_passrole_target_resource_scope_unknown": 3, + "shared_cross_account_trust_condition_unknown": 2, + "shared_boundary_or_session_policy_context_missing": 1, + }