87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
"""PlanDelta synopsis and priority utils.
|
|
|
|
Creates compact synopses for PlanDelta objects used by DTN custody holders to
|
|
decide whether to fetch full payloads during short windows. The synopsis is
|
|
kept intentionally small and includes a short HMAC-based signature.
|
|
"""
|
|
from typing import Dict, Any, List, Optional
|
|
import json
|
|
import hashlib
|
|
import hmac
|
|
import uuid
|
|
|
|
|
|
def _serialize_delta(delta) -> bytes:
|
|
# serialize commitments and priority into a deterministic JSON blob
|
|
payload = {
|
|
"commitments": [c for c in delta.commitments.value()],
|
|
"priority": delta.priority.read(),
|
|
"origin": getattr(delta, "origin", ""),
|
|
}
|
|
return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
|
|
|
|
def build_priority_key(urgency_score: int, impact_score: int) -> int:
|
|
"""Combine urgency (0-7) and impact (0-15) into a 0-127 key as (urgency<<4)|impact."""
|
|
if not (0 <= urgency_score <= 7):
|
|
raise ValueError("urgency_score must be 0..7")
|
|
if not (0 <= impact_score <= 15):
|
|
raise ValueError("impact_score must be 0..15")
|
|
return (urgency_score << 4) | (impact_score & 0x0F)
|
|
|
|
|
|
def build_synopsis(delta, parent_version: str = "", urgency_score: int = 0, impact_score: int = 0,
|
|
affected_assets: Optional[List[str]] = None, signer: str = "", key: bytes = b"",
|
|
delta_id: Optional[str] = None) -> Dict[str, Any]:
|
|
"""Build a compact synopsis for a PlanDelta.
|
|
|
|
- delta: PlanDelta instance
|
|
- parent_version: reference to parent PlanDelta/version
|
|
- urgency_score, impact_score: small integers used to compute priority_key
|
|
- affected_assets: optional list of affected node ids
|
|
- signer: identifier of signer
|
|
- key: optional bytes key used to produce a short HMAC signature
|
|
"""
|
|
payload = _serialize_delta(delta)
|
|
summary_hash = hashlib.sha256(payload).hexdigest()
|
|
size_estimate = len(payload)
|
|
priority_key = build_priority_key(urgency_score, impact_score)
|
|
delta_id = delta_id or uuid.uuid4().hex
|
|
|
|
synopsis = {
|
|
"delta_id": delta_id,
|
|
"parent_version": parent_version,
|
|
"priority_key": priority_key,
|
|
"affected_assets": affected_assets or [],
|
|
"summary_hash": summary_hash,
|
|
"size_estimate": size_estimate,
|
|
"signer": signer,
|
|
}
|
|
|
|
# short signature: HMAC of summary_hash with provided key if key present
|
|
if key:
|
|
short_sig = hmac.new(key, summary_hash.encode("utf-8"), hashlib.sha256).hexdigest()
|
|
else:
|
|
short_sig = ""
|
|
|
|
synopsis["short_sig"] = short_sig
|
|
return synopsis
|
|
|
|
|
|
def serialize_synopsis(syn: Dict[str, Any]) -> bytes:
|
|
return json.dumps(syn, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
|
|
|
|
def deserialize_synopsis(b: bytes) -> Dict[str, Any]:
|
|
return json.loads(b.decode("utf-8"))
|
|
|
|
|
|
def verify_synopsis(syn: Dict[str, Any], key: bytes = b"") -> bool:
|
|
summary_hash = syn.get("summary_hash", "")
|
|
if not summary_hash:
|
|
return False
|
|
if key:
|
|
expected = hmac.new(key, summary_hash.encode("utf-8"), hashlib.sha256).hexdigest()
|
|
return hmac.compare_digest(expected, syn.get("short_sig", ""))
|
|
return syn.get("short_sig", "") == ""
|