import argparse import hashlib import json import re import subprocess import sys import tempfile from dataclasses import dataclass from pathlib import Path, PurePosixPath ACCEPTANCE_RUNNER_VERSION = "2.1.0" CATALOG_VERSION = "2.0.0" TOKEN_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}") CATALOG_REQUIRED_FIELDS = ( "validator_id", "delivery_type", "schema_version", "validator_path", "validator_sha256", "policy_path", "policy_sha256", ) class ContractError(ValueError): pass @dataclass(frozen=True) class RuntimeRoots: rules_root: Path = Path("/delivery-rules") input_root: Path = Path("/delivery-input") report_root: Path = Path("/delivery-reports") def read_bytes(path, label): try: return path.read_bytes() except OSError as error: raise ContractError(f"{label} is not readable: {error}") from error def sha256_bytes(value): return hashlib.sha256(value).hexdigest() def parse_json_object(value, label): try: parsed = json.loads(value.decode("utf-8")) except (UnicodeError, json.JSONDecodeError) as error: raise ContractError(f"{label} is not readable JSON: {error}") from error if not isinstance(parsed, dict): raise ContractError(f"{label} root must be an object") return parsed def validate_token(value, label): if not isinstance(value, str) or not TOKEN_PATTERN.fullmatch(value) or value in {".", ".."}: raise ContractError(f"{label} contains unsupported characters") def require_nonempty_string(value, label): if not isinstance(value, str) or not value or value.strip() != value: raise ContractError(f"{label} must be a non-empty string without surrounding whitespace") return value def require_normalized_relative_path(value, label): value = require_nonempty_string(value, label) if "\\" in value: raise ContractError(f"{label} must use a normalized POSIX relative path") relative = PurePosixPath(value) if relative.is_absolute() or relative.as_posix() != value or any(part in {"", ".", ".."} for part in relative.parts): raise ContractError(f"{label} must use a normalized POSIX relative path") return value def resolve_exact_child(root, name, label, expect_directory=False): validate_token(name, label) resolved_root = root.resolve() candidate = (resolved_root / name).resolve() if candidate.parent != resolved_root: raise ContractError(f"{label} is outside the allowed root") if expect_directory and not candidate.is_dir(): raise ContractError(f"{label} directory is unavailable") return candidate def resolve_contained_file(root, relative_path, label): relative_path = require_normalized_relative_path(relative_path, label) resolved_root = root.resolve() candidate = (resolved_root / Path(relative_path)).resolve() try: candidate.relative_to(resolved_root) except ValueError as error: raise ContractError(f"{label} is outside the allowed root") from error if not candidate.is_file(): raise ContractError(f"{label} is unavailable") return candidate def require_sha256(value, label): if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value): raise ContractError(f"{label} must be a lowercase SHA-256 value") return value def validate_catalog(catalog_path, rules_root): catalog_bytes = read_bytes(catalog_path, "validator catalog") catalog = parse_json_object(catalog_bytes, "validator catalog") if catalog.get("version") != CATALOG_VERSION: raise ContractError(f"validator catalog version must be exactly {CATALOG_VERSION}") validators = catalog.get("validators") if not isinstance(validators, list) or not validators: raise ContractError("validator catalog validators must be a non-empty array") validated = [] seen_validator_ids = set() seen_type_versions = set() for index, raw_entry in enumerate(validators): label = f"validator catalog entry {index}" if not isinstance(raw_entry, dict): raise ContractError(f"{label} must be an object") entry = { field: require_nonempty_string(raw_entry.get(field), f"{label} {field}") for field in CATALOG_REQUIRED_FIELDS } validate_token(entry["validator_id"], f"{label} validator_id") if entry["validator_id"] in seen_validator_ids: raise ContractError("validator catalog validator_id values must be unique") seen_validator_ids.add(entry["validator_id"]) type_version = (entry["delivery_type"], entry["schema_version"]) if type_version in seen_type_versions: raise ContractError("validator catalog delivery_type and schema_version pairs must be unique") seen_type_versions.add(type_version) entry["validator_path"] = require_normalized_relative_path( entry["validator_path"], f"{label} validator_path", ) entry["policy_path"] = require_normalized_relative_path( entry["policy_path"], f"{label} policy_path", ) entry["validator_sha256"] = require_sha256( entry["validator_sha256"], f"{label} validator_sha256", ) entry["policy_sha256"] = require_sha256( entry["policy_sha256"], f"{label} policy_sha256", ) validator_path = resolve_contained_file( rules_root, entry["validator_path"], f"{label} validator_path", ) policy_path = resolve_contained_file( rules_root, entry["policy_path"], f"{label} policy_path", ) validator_bytes = read_bytes(validator_path, f"{label} validator") policy_bytes = read_bytes(policy_path, f"{label} policy") if sha256_bytes(validator_bytes) != entry["validator_sha256"]: raise ContractError(f"{label} validator SHA-256 does not match the catalog") if sha256_bytes(policy_bytes) != entry["policy_sha256"]: raise ContractError(f"{label} policy SHA-256 does not match the catalog") policy = parse_json_object(policy_bytes, f"{label} policy") for field in ("validator_id", "delivery_type", "schema_version"): if policy.get(field) != entry[field]: raise ContractError(f"{label} policy {field} does not match the catalog") entry["validator_path_resolved"] = validator_path entry["policy_path_resolved"] = policy_path entry["policy"] = policy validated.append(entry) return catalog_bytes, validated def plan_basis(context): return { "contract_version": "2.0", "delivery_id": context["delivery_id"], "delivery_type": context["delivery_type"], "schema_version": context["schema_version"], "report_key": context["report_key"], "validator_id": context["validator_id"], "selected_paths": { "dispatcher": "rules/accept_delivery.py", "catalog": "rules/validator-catalog.json", "manifest": f"input/{context['delivery_id']}/manifest.json", "data": f"input/{context['delivery_id']}/{context['manifest']['file']}", "validator": f"rules/{context['validator_relative_path']}", "policy": f"rules/{context['policy_relative_path']}", "report_object": context["report_path"].name, }, "provenance_sha256": { "dispatcher": context["provenance"]["dispatcher_sha256"], "catalog": context["provenance"]["catalog_sha256"], "manifest": context["provenance"]["manifest_sha256"], "data": context["provenance"]["data_sha256"], "validator": context["provenance"]["validator_sha256"], "policy": context["provenance"]["policy_sha256"], }, } def canonical_sha256(value): encoded = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def inspect_delivery(delivery_id, report_key, roots, allow_no_match=False): validate_token(delivery_id, "delivery-id") validate_token(report_key, "report-key") rules_root = roots.rules_root.resolve() input_root = roots.input_root.resolve() report_root = roots.report_root.resolve() if not rules_root.is_dir(): raise ContractError("rules root is unavailable") if not input_root.is_dir(): raise ContractError("input root is unavailable") if not report_root.is_dir(): raise ContractError("report root is unavailable") dispatcher_path = resolve_contained_file(rules_root, "accept_delivery.py", "dispatcher path") if dispatcher_path != Path(__file__).resolve(): raise ContractError("running dispatcher does not match the rules-root dispatcher") dispatcher_bytes = read_bytes(dispatcher_path, "dispatcher") dispatcher_sha256 = sha256_bytes(dispatcher_bytes) delivery_dir = resolve_exact_child(input_root, delivery_id, "delivery-id", expect_directory=True) manifest_path = resolve_contained_file(delivery_dir, "manifest.json", "manifest") manifest_bytes = read_bytes(manifest_path, "manifest") manifest = parse_json_object(manifest_bytes, "manifest") if manifest.get("delivery_id") != delivery_id: raise ContractError("manifest delivery_id does not match --delivery-id") delivery_type = require_nonempty_string(manifest.get("delivery_type"), "manifest delivery_type") schema_version = require_nonempty_string(manifest.get("schema_version"), "manifest schema_version") declared_file = require_normalized_relative_path(manifest.get("file"), "manifest file") data_path = resolve_contained_file(delivery_dir, declared_file, "manifest file") if data_path.parent != delivery_dir or data_path.name != declared_file: raise ContractError("manifest file must be a direct child of the delivery directory") data_bytes = read_bytes(data_path, "data file") expected_data_sha256 = require_sha256(manifest.get("sha256"), "manifest sha256") actual_data_sha256 = sha256_bytes(data_bytes) if actual_data_sha256 != expected_data_sha256: raise ContractError("data-file SHA-256 does not match the manifest") if not isinstance(manifest.get("rows"), int) or manifest["rows"] < 0: raise ContractError("manifest rows must be a non-negative integer") report_path = resolve_exact_child(report_root, f"{report_key}.json", "report object") if report_path.exists(): raise ContractError("report object already exists; choose a new report-key") catalog_path = resolve_contained_file(rules_root, "validator-catalog.json", "validator catalog") catalog_bytes, validators = validate_catalog(catalog_path, rules_root) matches = [ entry for entry in validators if entry["delivery_type"] == delivery_type and entry["schema_version"] == schema_version ] shared_provenance = { "dispatcher_path": str(dispatcher_path), "dispatcher_sha256": dispatcher_sha256, "catalog_path": str(catalog_path), "catalog_sha256": sha256_bytes(catalog_bytes), "manifest_path": str(manifest_path), "manifest_sha256": sha256_bytes(manifest_bytes), "data_path": str(data_path), "data_sha256": actual_data_sha256, } if not matches and allow_no_match: return { "delivery_id": delivery_id, "delivery_type": delivery_type, "schema_version": schema_version, "report_key": report_key, "report_path": report_path, "manifest_path": manifest_path, "manifest": manifest, "data_path": data_path, "catalog_path": catalog_path, "dispatcher_path": dispatcher_path, "validator_id": None, "provenance": shared_provenance, } if len(matches) != 1: raise ContractError("validator catalog must contain exactly one matching delivery_type and schema_version") selected = matches[0] if selected["policy"].get("required_file") != declared_file: raise ContractError("manifest file does not match the selected policy") validator_path = selected["validator_path_resolved"] policy_path = selected["policy_path_resolved"] provenance = { **shared_provenance, "validator_path": str(validator_path), "validator_sha256": selected["validator_sha256"], "policy_path": str(policy_path), "policy_sha256": selected["policy_sha256"], } return { "delivery_id": delivery_id, "delivery_type": delivery_type, "schema_version": schema_version, "report_key": report_key, "report_path": report_path, "dispatcher_path": dispatcher_path, "catalog_path": catalog_path, "manifest_path": manifest_path, "manifest": manifest, "data_path": data_path, "validator_id": selected["validator_id"], "validator_path": validator_path, "validator_relative_path": selected["validator_path"], "policy_path": policy_path, "policy_relative_path": selected["policy_path"], "provenance": provenance, } def build_plan(context): if context["validator_id"] is None: return { "phase": "plan", "status": "NOT_READY", "reason_code": "NO_APPROVED_VALIDATOR", "delivery_id": context["delivery_id"], "delivery_type": context["delivery_type"], "schema_version": context["schema_version"], "selected_validator_id": None, "dispatcher_sha256": context["provenance"]["dispatcher_sha256"], "provenance": context["provenance"], "proposed_run": None, "report_object_name": context["report_path"].name, "mount_output_path": str(context["report_path"]), "failed_checks": ["approved_validator_match"], "next_action": { "code": "STOP_AND_ADD_APPROVED_VALIDATOR", "summary": "Do not run this delivery. Ask the rule owner to add and approve an exact type-and-version catalog entry first.", }, } basis = plan_basis(context) plan_sha256 = canonical_sha256(basis) proposed_run = { "subcommand": "run", "delivery_id": context["delivery_id"], "report_key": context["report_key"], "validator_id": context["validator_id"], "plan_sha256": plan_sha256, } return { "phase": "plan", "status": "READY", "delivery_id": context["delivery_id"], "delivery_type": context["delivery_type"], "schema_version": context["schema_version"], "selected_validator_id": context["validator_id"], "dispatcher_sha256": context["provenance"]["dispatcher_sha256"], "provenance": context["provenance"], "plan_basis": basis, "plan_sha256": plan_sha256, "proposed_run": proposed_run, "proposed_run_argv": [ "python3", "/delivery-rules/accept_delivery.py", "run", "--delivery-id", context["delivery_id"], "--report-key", context["report_key"], "--validator-id", context["validator_id"], "--plan-sha256", plan_sha256, ], "report_object_name": context["report_path"].name, "mount_output_path": str(context["report_path"]), "failed_checks": [], "next_action": { "code": "REVIEW_AND_APPROVE_RUN", "summary": "Review the selected validator, pinned hashes, and proposed run fields before allowing execution.", }, } def next_action_for(status): if status == "PASS": return { "code": "REVIEW_FOR_DOWNSTREAM_IMPORT", "summary": "Review the evidence and decide whether the batch may enter the downstream system; no import was performed.", } return { "code": "REQUEST_CORRECTED_DELIVERY", "summary": "Ask the supplier to correct the failed checks and resubmit under a new delivery ID; no downstream action was performed.", } def provenance_sha256(context): return { name: context["provenance"][f"{name}_sha256"] for name in ("dispatcher", "catalog", "manifest", "data", "validator", "policy") } def read_verified_source(path, expected_sha256, label): value = read_bytes(path, label) if sha256_bytes(value) != expected_sha256: raise ContractError(f"{label} changed after inspection") return value def write_private_snapshot(path, value): try: with path.open("xb") as target: target.write(value) except OSError as error: raise ContractError(f"unable to create private validation snapshot: {error}") from error def invoke_validator_snapshot(context): expected = context["provenance"] read_verified_source(context["dispatcher_path"], expected["dispatcher_sha256"], "dispatcher") read_verified_source(context["catalog_path"], expected["catalog_sha256"], "validator catalog") validator_bytes = read_verified_source( context["validator_path"], expected["validator_sha256"], "validator", ) manifest_bytes = read_verified_source( context["manifest_path"], expected["manifest_sha256"], "manifest", ) data_bytes = read_verified_source(context["data_path"], expected["data_sha256"], "data file") policy_bytes = read_verified_source(context["policy_path"], expected["policy_sha256"], "policy") try: temporary_directory = tempfile.TemporaryDirectory(prefix="agentworks-delivery-validation-") except OSError as error: raise ContractError(f"unable to create private validation directory: {error}") from error with temporary_directory as temporary_path: snapshot_root = Path(temporary_path) validator_snapshot = snapshot_root / "validate_csv.py" manifest_snapshot = snapshot_root / "manifest.json" data_snapshot = snapshot_root / context["data_path"].name policy_snapshot = snapshot_root / context["policy_path"].name for path, value in ( (validator_snapshot, validator_bytes), (manifest_snapshot, manifest_bytes), (data_snapshot, data_bytes), (policy_snapshot, policy_bytes), ): write_private_snapshot(path, value) return subprocess.run( [ sys.executable, str(validator_snapshot), "--manifest", str(manifest_snapshot), "--data-file", str(data_snapshot), "--policy", str(policy_snapshot), "--validator-id", context["validator_id"], ], check=False, capture_output=True, text=True, timeout=30, ) def run_validation(context, supplied_validator_id, supplied_plan_sha256): validate_token(supplied_validator_id, "validator-id") if supplied_validator_id != context["validator_id"]: raise ContractError("--validator-id does not match the validator selected from the approved catalog") supplied_plan_sha256 = require_sha256(supplied_plan_sha256, "plan-sha256") actual_plan_sha256 = canonical_sha256(plan_basis(context)) if supplied_plan_sha256 != actual_plan_sha256: raise ContractError("--plan-sha256 does not match the current delivery, catalog, validator, policy, data, or dispatcher state") completed = invoke_validator_snapshot(context) if completed.returncode != 0: raise ContractError(f"validator failed to complete: {completed.stdout.strip() or completed.stderr.strip()}") try: result = json.loads(completed.stdout) except json.JSONDecodeError as error: raise ContractError("validator output is not JSON") from error if not isinstance(result, dict) or result.get("status") not in {"PASS", "FAIL"}: raise ContractError("validator output does not contain a supported status") if result.get("validator_id") != context["validator_id"]: raise ContractError("validator output validator_id does not match the approved selection") failed_checks = result.get("failed_checks") checks = result.get("checks") if not isinstance(failed_checks, list) or not all(isinstance(item, str) for item in failed_checks): raise ContractError("validator output failed_checks must be an array of strings") if not isinstance(checks, list): raise ContractError("validator output checks must be an array") status = result["status"] if (status == "PASS") != (len(failed_checks) == 0): raise ContractError("validator status and failed_checks disagree") next_action = next_action_for(status) dispatcher_sha256 = context["provenance"]["dispatcher_sha256"] report = { "report_schema_version": "2.0", "acceptance_runner_version": ACCEPTANCE_RUNNER_VERSION, "validator_engine_version": result.get("validator_engine_version"), "delivery_id": context["delivery_id"], "delivery_type": context["delivery_type"], "schema_version": context["schema_version"], "validator_id": context["validator_id"], "dispatcher_sha256": dispatcher_sha256, "plan_sha256": actual_plan_sha256, "provenance_sha256": provenance_sha256(context), "provenance": context["provenance"], "status": status, "failed_check_count": len(failed_checks), "failed_checks": failed_checks, "checks": checks, "next_action": next_action, "downstream_action_performed": False, "report_key": context["report_key"], "report_object_name": context["report_path"].name, "mount_output_path": str(context["report_path"]), } try: with context["report_path"].open("x", encoding="utf-8") as target: json.dump(report, target, ensure_ascii=False, indent=2) target.write("\n") except FileExistsError as error: raise ContractError("report object already exists; choose a new report-key") from error return { "phase": "run", "status": status, "delivery_id": context["delivery_id"], "delivery_type": context["delivery_type"], "schema_version": context["schema_version"], "validator_id": context["validator_id"], "dispatcher_sha256": dispatcher_sha256, "plan_sha256": actual_plan_sha256, "provenance_sha256": provenance_sha256(context), "failed_check_count": len(failed_checks), "failed_checks": failed_checks, "next_action": next_action, "report_object_name": context["report_path"].name, "mount_output_path": str(context["report_path"]), "downstream_action_performed": False, } def parse_args(argv=None): parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="phase", required=True) plan_parser = subparsers.add_parser("plan") plan_parser.add_argument("--delivery-id", required=True) plan_parser.add_argument("--report-key", required=True) run_parser = subparsers.add_parser("run") run_parser.add_argument("--delivery-id", required=True) run_parser.add_argument("--report-key", required=True) run_parser.add_argument("--validator-id", required=True) run_parser.add_argument("--plan-sha256", required=True) return parser.parse_args(argv) def execute(argv=None, roots=None): args = parse_args(argv) runtime_roots = roots or RuntimeRoots() context = inspect_delivery( args.delivery_id, args.report_key, runtime_roots, allow_no_match=args.phase == "plan", ) if args.phase == "plan": return build_plan(context) return run_validation(context, args.validator_id, args.plan_sha256) def main(argv=None, roots=None): try: result = execute(argv, roots) except (ContractError, subprocess.TimeoutExpired) as error: print(json.dumps({"status": "ERROR", "error": str(error)}, ensure_ascii=False)) return 2 print(json.dumps(result, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())