From c84984efc8933297ff090ca8687b3696ac320129 Mon Sep 17 00:00:00 2001 From: agent-56a7678c6cd71659 Date: Sat, 25 Apr 2026 21:36:43 +0200 Subject: [PATCH] build(agent): jabba#56a767 iteration --- gravityweave/validator.py | 84 +++++++++++++++++++++++++++++++++++++++ tests/test_validator.py | 46 +++++++++++++++++++++ 2 files changed, 130 insertions(+) create mode 100644 gravityweave/validator.py create mode 100644 tests/test_validator.py diff --git a/gravityweave/validator.py b/gravityweave/validator.py new file mode 100644 index 0000000..12193ca --- /dev/null +++ b/gravityweave/validator.py @@ -0,0 +1,84 @@ +"""Lightweight witness validator. + +Provides quick local validation of a feasibility witness against a node's +resource envelope. Intended to be cheap and conservative so it can run on +constrained nodes and during short custody windows. + +Verdicts: +- "OK": witness comfortably within resource margins +- "needs-review": within a configurable margin (conservative) +- "unstable": witness exceeds available resources (infeasible) + +If a secret_key is provided, make_signed_verdict will attach an HMAC-SHA256 +signature over the verdict (short hex). +""" +from typing import Dict, Any, Tuple +import json +import hmac +import hashlib + + +def validate_witness(witness: Dict[str, Any], resources: Dict[str, Any], *, fuel_margin: float = 0.1, power_margin: float = 0.1) -> Tuple[str, Dict[str, Any]]: + """Validate a witness against resources. + + Parameters: + - witness: output of generate_feasibility_witness + - resources: dict with keys 'available_fuel', 'available_power', 'max_payload_bytes' + - fuel_margin/power_margin: fractional margin; if witness consumption is + above available*(1 - margin) but <= available it's 'needs-review'. + + Returns (verdict, details) + """ + details: Dict[str, Any] = {"checks": []} + + # helpers + def _check(name: str, required: float, available: float, margin: float): + if required <= available * (1.0 - margin): + return "ok", f"{name} within comfortable margin ({required} <= {available*(1.0-margin)})" + if required <= available: + return "review", f"{name} within hard limit but near margin ({required} <= {available})" + return "unstable", f"{name} exceeds available ({required} > {available})" + + total_fuel = float(witness.get("total_fuel", 0.0)) + total_power = float(witness.get("total_power", 0.0)) + max_payload = int(witness.get("max_payload_bytes", 0)) + + avail_fuel = float(resources.get("available_fuel", 0.0)) + avail_power = float(resources.get("available_power", 0.0)) + avail_payload = int(resources.get("max_payload_bytes", 0)) + + f_status, f_msg = _check("fuel", total_fuel, avail_fuel, fuel_margin) + p_status, p_msg = _check("power", total_power, avail_power, power_margin) + pl_status, pl_msg = _check("payload", float(max_payload), float(avail_payload), 0.0) + + details["checks"].append({"fuel": {"status": f_status, "msg": f_msg, "required": total_fuel, "available": avail_fuel}}) + details["checks"].append({"power": {"status": p_status, "msg": p_msg, "required": total_power, "available": avail_power}}) + details["checks"].append({"payload": {"status": pl_status, "msg": pl_msg, "required": max_payload, "available": avail_payload}}) + + # Decide overall verdict: unstable > needs-review > OK + statuses = {f_status, p_status, pl_status} + if "unstable" in statuses: + verdict = "unstable" + elif "review" in statuses: + verdict = "needs-review" + else: + verdict = "OK" + + details["verdict_reason"] = f"derived from checks: {sorted(list(statuses))}" + return verdict, details + + +def make_signed_verdict(witness: Dict[str, Any], resources: Dict[str, Any], secret_key: bytes = None, **validate_kwargs) -> Dict[str, Any]: + """Run validation and optionally attach an HMAC signature. + + The returned dict contains: verdict, details, and if secret_key provided, + a short signature hex string (first 32 chars of HMAC-SHA256). + """ + verdict, details = validate_witness(witness, resources, **validate_kwargs) + out = {"verdict": verdict, "details": details} + # canonicalize and sign + if secret_key is not None: + payload = json.dumps({"verdict": verdict, "details": details}, sort_keys=True, separators=(",", ":")).encode("utf-8") + sig = hmac.new(secret_key, payload, hashlib.sha256).hexdigest()[:32] + out["signature"] = sig + return out diff --git a/tests/test_validator.py b/tests/test_validator.py new file mode 100644 index 0000000..52192ff --- /dev/null +++ b/tests/test_validator.py @@ -0,0 +1,46 @@ +from gravityweave.plan_delta import PlanDelta +from gravityweave.witness import generate_feasibility_witness +from gravityweave.validator import validate_witness, make_signed_verdict + + +def test_validate_witness_ok(): + pd = PlanDelta(origin="node1") + pd.commitments.add({"type": "burn", "fuel": 5.0, "power": 1.0, "payload_bytes": 100}) + w = generate_feasibility_witness(pd) + + resources = {"available_fuel": 10.0, "available_power": 5.0, "max_payload_bytes": 200} + verdict, details = validate_witness(w, resources, fuel_margin=0.1, power_margin=0.1) + assert verdict == "OK" + assert "checks" in details + + +def test_validate_witness_needs_review(): + pd = PlanDelta(origin="node2") + pd.commitments.add({"type": "burn", "fuel": 9.0, "power": 1.0}) + w = generate_feasibility_witness(pd) + + resources = {"available_fuel": 10.0, "available_power": 5.0, "max_payload_bytes": 0} + verdict, _ = validate_witness(w, resources, fuel_margin=0.2, power_margin=0.1) + assert verdict == "needs-review" + + +def test_validate_witness_unstable(): + pd = PlanDelta(origin="node3") + pd.commitments.add({"type": "burn", "fuel": 11.0, "power": 1.0}) + w = generate_feasibility_witness(pd) + + resources = {"available_fuel": 10.0, "available_power": 5.0, "max_payload_bytes": 0} + verdict, _ = validate_witness(w, resources, fuel_margin=0.1, power_margin=0.1) + assert verdict == "unstable" + + +def test_make_signed_verdict(): + pd = PlanDelta(origin="node4") + pd.commitments.add({"type": "burn", "fuel": 1.0, "power": 0.5}) + w = generate_feasibility_witness(pd) + + resources = {"available_fuel": 5.0, "available_power": 5.0, "max_payload_bytes": 0} + secret = b"testkey" + out = make_signed_verdict(w, resources, secret_key=secret) + assert out["verdict"] == "OK" + assert "signature" in out and isinstance(out["signature"], str) and len(out["signature"]) == 32