informed-decision/history/20260909-initial-exploration/canonicalize.py

#!/usr/bin/env python3
"""Canonicalization for informed-decision view_hash and awareness_hash.

Profile: JCS-inspired restricted JSON (RFC 8785 subset).
- UTF-8
- objects with keys sorted by UTF-8 code units (equivalent to UTF-16 for our ASCII keys)
- no insignificant whitespace
- integers only (no floats)
- strings escaped as in RFC 8259 / JCS
"""

from __future__ import annotations

import hashlib
import json
from typing import Any


ALLOWED_BINDING_KEYS = (
    "awareness_promoted",
    "binding",
    "binding_level",
    "brief",
    "highlights",
    "locale",
    "memo_id",
    "memo_version",
    "packet",
    "question",
    "requested_act",
    "ui_release",
)

ALLOWED_AWARENESS_KEYS = (
    "available_hats",
    "available_scopes",
    "last_session",
    "locale",
    "memo_id",
    "memo_version",
    "proposed_hat",
    "proposed_hat_source",
    "situation_note",
    "ui_release",
)

BINDING_SLICE_KEYS = (
    "available_bind_scopes",
    "available_identities",
    "blast_radius",
    "granted_at_bind",
    "justification",
    "principal",
    "target",
    "terms",
)


def _is_int(n: Any) -> bool:
    return isinstance(n, int) and not isinstance(n, bool)


def escape_string(s: str) -> str:
    out = ['"']
    for ch in s:
        o = ord(ch)
        if ch == '"':
            out.append('\\"')
        elif ch == "\\":
            out.append("\\\\")
        elif ch == "\b":
            out.append("\\b")
        elif ch == "\f":
            out.append("\\f")
        elif ch == "\n":
            out.append("\\n")
        elif ch == "\r":
            out.append("\\r")
        elif ch == "\t":
            out.append("\\t")
        elif o < 0x20:
            out.append(f"\\u{o:04x}")
        else:
            out.append(ch)
    out.append('"')
    return "".join(out)


def dumps(value: Any) -> str:
    if value is None:
        return "null"
    if value is True:
        return "true"
    if value is False:
        return "false"
    if _is_int(value):
        return str(value)
    if isinstance(value, str):
        return escape_string(value)
    if isinstance(value, list):
        return "[" + ",".join(dumps(v) for v in value) + "]"
    if isinstance(value, dict):
        items = []
        for key in sorted(value.keys()):
            if not isinstance(key, str):
                raise TypeError("object keys must be strings")
            items.append(escape_string(key) + ":" + dumps(value[key]))
        return "{" + ",".join(items) + "}"
    raise TypeError(f"unsupported type for canonicalization: {type(value)!r}")


def pick(obj: dict, allowed: tuple[str, ...]) -> dict:
    return {k: obj[k] for k in allowed if k in obj}


def strip_nulls(value: Any) -> Any:
    if isinstance(value, dict):
        return {k: strip_nulls(v) for k, v in value.items() if v is not None}
    if isinstance(value, list):
        return [strip_nulls(v) for v in value]
    return value


def normalize_hash(h: Any) -> str:
    if isinstance(h, str):
        return h if ":" in h else f"sha256:{h}"
    if isinstance(h, dict):
        return f"{h['alg']}:{h['hex']}"
    raise TypeError("hash must be string or {alg,hex}")


def normalize_packet(packet: list) -> list:
    items = []
    for item in packet:
        items.append(
            {
                "hash": normalize_hash(item["hash"]),
                "item_id": item["item_id"],
            }
        )
    items.sort(key=lambda i: i["item_id"])
    return items


def normalize_highlights(highlights: list) -> list:
    items = []
    for hl in highlights:
        loc = hl.get("locator") or {}
        items.append(
            {
                "id": hl["id"],
                "item_id": hl["item_id"],
                "locator": strip_nulls(loc),
                "required_ack": bool(hl.get("required_ack", False)),
                "severity": hl.get("severity", "informational"),
            }
        )
    items.sort(key=lambda i: i["id"])
    return items


def normalize_party(party: dict) -> dict:
    out = {
        "display_name": party["display_name"],
        "id": party["id"],
        "kind": party["kind"],
    }
    if party.get("identifiers"):
        ids = []
        for ident in party["identifiers"]:
            ids.append({"scheme": ident["scheme"], "value": ident["value"]})
        ids.sort(key=lambda i: (i["scheme"], i["value"]))
        out["identifiers"] = ids
    if party.get("role"):
        out["role"] = party["role"]
    return out


def normalize_scope(scope: dict) -> dict:
    out = {
        "id": scope["id"],
        "kind": scope["kind"],
        "label": scope["label"],
    }
    if "environment" in scope:
        out["environment"] = scope["environment"]
    if "requires_new_bind" in scope:
        out["requires_new_bind"] = bool(scope["requires_new_bind"])
    return out


def normalize_hat(hat: dict) -> dict:
    out = {"id": hat["id"], "label": hat["label"]}
    if hat.get("kind"):
        out["kind"] = hat["kind"]
    if "elevates" in hat:
        out["elevates"] = bool(hat["elevates"])
    if hat.get("permissions_preview"):
        out["permissions_preview"] = sorted(hat["permissions_preview"])
    if hat.get("scope_id"):
        out["scope_id"] = hat["scope_id"]
    return out


def normalize_binding(binding: dict) -> dict:
    raw = pick(binding, BINDING_SLICE_KEYS)
    out: dict[str, Any] = {}
    if "principal" in raw:
        out["principal"] = normalize_party(raw["principal"])
    if raw.get("available_identities"):
        ids = [normalize_party(p) for p in raw["available_identities"]]
        ids.sort(key=lambda p: p["id"])
        out["available_identities"] = ids
    if "target" in raw:
        out["target"] = normalize_scope(raw["target"])
    if raw.get("available_bind_scopes"):
        scopes = [normalize_scope(s) for s in raw["available_bind_scopes"]]
        scopes.sort(key=lambda s: s["id"])
        out["available_bind_scopes"] = scopes
    if raw.get("granted_at_bind"):
        g = dict(raw["granted_at_bind"])
        if g.get("roles"):
            g["roles"] = sorted(g["roles"])
        if g.get("permissions"):
            g["permissions"] = sorted(g["permissions"])
        out["granted_at_bind"] = strip_nulls(g)
    for k in ("justification", "blast_radius", "terms"):
        if k in raw:
            out[k] = strip_nulls(raw[k])
    return out


def binding_document(src: dict) -> dict:
    doc = pick(src, ALLOWED_BINDING_KEYS)
    if "packet" in doc:
        doc["packet"] = normalize_packet(doc["packet"])
    if "highlights" in doc:
        doc["highlights"] = normalize_highlights(doc["highlights"])
    if "binding" in doc:
        doc["binding"] = normalize_binding(doc["binding"])
    if "awareness_promoted" in doc:
        doc["awareness_promoted"] = strip_nulls(doc["awareness_promoted"])
    return strip_nulls(doc)


def awareness_document(src: dict) -> dict:
    doc = pick(src, ALLOWED_AWARENESS_KEYS)
    if doc.get("proposed_hat"):
        doc["proposed_hat"] = normalize_hat(doc["proposed_hat"])
    if doc.get("available_hats"):
        hats = [normalize_hat(h) for h in doc["available_hats"]]
        hats.sort(key=lambda h: h["id"])
        doc["available_hats"] = hats
    if doc.get("available_scopes"):
        scopes = [normalize_scope(s) for s in doc["available_scopes"]]
        scopes.sort(key=lambda s: s["id"])
        doc["available_scopes"] = scopes
    return strip_nulls(doc)


def sha256_hex(canonical: str) -> str:
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def view_hash(src: dict) -> dict:
    canonical = dumps(binding_document(src))
    return {
        "alg": "sha256",
        "hex": sha256_hex(canonical),
        "canonical": canonical,
    }


def awareness_hash(src: dict) -> dict:
    canonical = dumps(awareness_document(src))
    return {
        "alg": "sha256",
        "hex": sha256_hex(canonical),
        "canonical": canonical,
    }


if __name__ == "__main__":
    import pathlib
    import sys

    path = pathlib.Path(sys.argv[1])
    data = json.loads(path.read_text())
    kind = sys.argv[2] if len(sys.argv) > 2 else "view"
    result = view_hash(data) if kind == "view" else awareness_hash(data)
    print(result["canonical"])
    print(result["hex"])