85 lines
3.8 KiB
Python
85 lines
3.8 KiB
Python
"""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
|