from __future__ import annotations import json import tempfile from datetime import datetime, timezone from pathlib import Path from secrets import token_hex from typing import Any from uuid import uuid4 import gradio as gr from verifiable_tool_invocation_flow.guarded_tool_call import guarded_tool_call from verifiable_tool_invocation_flow.models import ExecutionRequest, PolicySnapshot, ToolManifest from verifiable_tool_invocation_flow.resources import load_example_json from verifiable_tool_invocation_flow.signer import ReceiptSigner from verifiable_tool_invocation_flow.tools.demo_metadata_lookup_tool import demo_metadata_lookup_tool from verifiable_tool_invocation_flow.validator import validate_receipt DEFAULT_AUDIENCE = "demo-validator" CHECK_FIELDS = [ "schema_valid", "input_hash_match", "policy_hash_match", "tool_manifest_hash_match", "tool_input_hash_match", "tool_output_hash_match", "result_hash_match", "pre_execution_commitment_match", "policy_decision_valid", "signature_valid", "time_window_valid", "replay_check_performed", "replay_detected", "audience_match", "request_binding_match", ] def load_json_upload(file_path: str | None) -> dict[str, Any]: if not file_path: raise gr.Error("Missing JSON file upload.") try: payload = json.loads(Path(file_path).read_text(encoding="utf-8")) except json.JSONDecodeError as exc: raise gr.Error(f"Invalid JSON upload: {exc.msg}") from exc if not isinstance(payload, dict): raise gr.Error("Uploaded JSON must be an object.") return payload def load_text_upload(file_path: str | None) -> bytes: if not file_path: raise gr.Error("Missing text file upload.") return Path(file_path).read_bytes() def summarize_report(report: dict[str, Any]) -> list[list[str]]: rows: list[list[str]] = [] for field in CHECK_FIELDS: value = report.get(field) if isinstance(value, bool): status = "pass" if value else "fail" else: status = "n/a" rows.append([field, status]) return rows def validate_uploaded_receipt( receipt_file: str | None, evidence_file: str | None, public_key_file: str | None, audience: str | None, ) -> tuple[str, list[list[str]], list[str], list[str], dict[str, Any], str]: receipt = load_json_upload(receipt_file) evidence = load_json_upload(evidence_file) public_key = load_text_upload(public_key_file) report = validate_receipt( receipt=receipt, evidence_bundle=evidence, public_key_pem=public_key, audience=(audience or DEFAULT_AUDIENCE).strip() or DEFAULT_AUDIENCE, replay_cache_path=None, update_replay_cache=False, ) report_path = _write_temp_json("verification_report.json", report) return ( str(report["verdict"]), summarize_report(report), report.get("errors", []), report.get("warnings", []), report, report_path, ) def run_builtin_demo() -> tuple[str, list[list[str]], dict[str, Any], str, str, str, str]: request = _build_demo_request() policy = PolicySnapshot.model_validate(load_example_json("policy_snapshot.json")) tool_manifest = ToolManifest.model_validate(load_example_json("tool_manifest.json")) tool_input = load_example_json("tool_input.json") signer = ReceiptSigner.generate_demo() result = guarded_tool_call( request, policy, tool_manifest, tool_input, demo_metadata_lookup_tool, signer, audience=DEFAULT_AUDIENCE, replay_cache_path=None, update_replay_cache=False, ) receipt_path = _write_temp_json("execution_receipt.json", result.receipt) evidence_path = _write_temp_json("evidence_bundle.json", result.evidence_bundle) report_path = _write_temp_json("verification_report.json", result.verification_report) public_key_path = _write_temp_bytes("demo_public_key.pem", signer.public_key_pem()) return ( str(result.verification_report["verdict"]), summarize_report(result.verification_report), result.verification_report, receipt_path, evidence_path, public_key_path, report_path, ) def _build_demo_request() -> ExecutionRequest: payload = load_example_json("input_request.json") now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") payload.update( { "request_id": f"req-{uuid4()}", "execution_id": f"exec-{uuid4()}", "nonce": token_hex(16), "requested_at": now, } ) return ExecutionRequest.model_validate(payload) def _write_temp_json(filename: str, payload: dict[str, Any]) -> str: temp_dir = Path(tempfile.mkdtemp(prefix="agent-receipt-validator-")) target = temp_dir / filename target.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") return str(target) def _write_temp_bytes(filename: str, payload: bytes) -> str: temp_dir = Path(tempfile.mkdtemp(prefix="agent-receipt-validator-")) target = temp_dir / filename target.write_bytes(payload) return str(target) with gr.Blocks() as demo: gr.Markdown( """ # Agent Receipt Validator Validate signed agent execution receipts against evidence bundles. This is a demo for verifiable execution evidence, not a compliance certification system. """ ) with gr.Tab("Validate uploaded receipt"): uploaded_receipt = gr.File(label="execution_receipt.json", type="filepath") uploaded_evidence = gr.File(label="evidence_bundle.json", type="filepath") uploaded_public_key = gr.File(label="public key PEM", type="filepath") uploaded_audience = gr.Textbox(label="Expected audience", value=DEFAULT_AUDIENCE) uploaded_button = gr.Button("Validate") uploaded_verdict = gr.Label(label="Verdict") uploaded_checks = gr.Dataframe(label="Validation checks", headers=["check", "status"], interactive=False) uploaded_errors = gr.JSON(label="Errors") uploaded_warnings = gr.JSON(label="Warnings") uploaded_report = gr.JSON(label="Verification report") uploaded_report_file = gr.File(label="Download verification_report.json") uploaded_button.click( validate_uploaded_receipt, inputs=[uploaded_receipt, uploaded_evidence, uploaded_public_key, uploaded_audience], outputs=[ uploaded_verdict, uploaded_checks, uploaded_errors, uploaded_warnings, uploaded_report, uploaded_report_file, ], ) with gr.Tab("Run built-in demo"): demo_button = gr.Button("Generate and validate fresh demo artifacts") demo_verdict = gr.Label(label="Verdict") demo_checks = gr.Dataframe(label="Validation checks", headers=["check", "status"], interactive=False) demo_report = gr.JSON(label="Verification report") demo_receipt_file = gr.File(label="Download execution_receipt.json") demo_evidence_file = gr.File(label="Download evidence_bundle.json") demo_public_key_file = gr.File(label="Download demo_public_key.pem") demo_report_file = gr.File(label="Download verification_report.json") demo_button.click( run_builtin_demo, inputs=[], outputs=[ demo_verdict, demo_checks, demo_report, demo_receipt_file, demo_evidence_file, demo_public_key_file, demo_report_file, ], ) gr.Markdown( "This Space does not store uploaded files permanently. Avoid uploading confidential production receipts, " "private keys, API tokens, or sensitive evidence bundles." ) if __name__ == "__main__": demo.launch()