67 lines
2.6 KiB
Python
67 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Integrator: retain manifests in RE and select a validated source-bound baseline."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
from campaign import Campaign, ControlError, atomic_json
|
|
from dashboard import gate_valid, replay_valid
|
|
|
|
|
|
def retain(root, source, kind):
|
|
data = source.read_bytes()
|
|
digest = hashlib.sha256(data).hexdigest()
|
|
value = json.loads(data)
|
|
relative = f"campaign/evidence/{digest}-{kind}.json"
|
|
path = root / relative
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
with path.open("xb") as file:
|
|
file.write(data)
|
|
except FileExistsError:
|
|
if path.read_bytes() != data:
|
|
raise ValueError("retained artifact differs from its content-addressed name")
|
|
return value, {"path": relative, "sha256": digest}
|
|
|
|
|
|
def select(root, gate_path, replay_path, next_action):
|
|
campaign = Campaign(str(root.resolve()))
|
|
# Check before copying. Evidence selection does not create or change contract acceptance.
|
|
gate = json.loads(gate_path.read_bytes())
|
|
gate_valid(gate, gate.get("source"))
|
|
binary = Path(gate["binary"]["path"])
|
|
if not binary.is_file() or hashlib.sha256(binary.read_bytes()).hexdigest() != gate["binary"]["sha256"]:
|
|
raise ValueError("gate executable missing or hash mismatch")
|
|
with campaign.lock():
|
|
gate, pointer = retain(root, gate_path, "gate")
|
|
current = {"schema": "sots-current/1", "source": gate["source"], "gate": pointer,
|
|
"replay": None, "next_action": next_action}
|
|
if replay_path:
|
|
replay = json.loads(replay_path.read_bytes())
|
|
replay_valid(root, replay, current, gate)
|
|
_, current["replay"] = retain(root, replay_path, "replay")
|
|
atomic_json(root / "campaign/current.json", current)
|
|
return current
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--state-root", required=True, type=Path)
|
|
parser.add_argument("--gate", required=True, type=Path)
|
|
parser.add_argument("--replay", type=Path)
|
|
parser.add_argument("--next-action", required=True)
|
|
args = parser.parse_args()
|
|
try:
|
|
selected = select(args.state_root.resolve(), args.gate.resolve(),
|
|
args.replay.resolve() if args.replay else None, args.next_action)
|
|
print(json.dumps({"gate": selected["gate"], "replay": selected["replay"]}, indent=2))
|
|
except (ControlError, ValueError, OSError, KeyError, TypeError) as exc:
|
|
print(f"select_evidence: {exc}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|