amethyst/cli/tests/sno/refdriver.py

#!/usr/bin/env python3
"""Adapter over the cyberspace project's own reference implementations.

Speaks one JSON object per line on stdout so `sno-conformance.sh` can diff it
against `amy sno … --json` with jq. Nothing here reimplements anything: it
imports `decks/sno-reference.py` from the cyberspace spec repo (the §1.9
arbiter) and `cyberspace_core/avatar.py` from cyberspace-cli (the §8.10 work
and payment reference), and just re-emits what they answer.

  refdriver.py cases                 the deck's own rejection table, + Appendix A
  refdriver.py vectors               cyberspace-cli's avatar-work golden vectors
  refdriver.py verdict < payload     {"valid":…, "rule":…}
  refdriver.py work    < payload     {"required":…}
  refdriver.py verify  < event       {"ok":…, "required":…, "committed":…, "zeros":…, "reason":…}
  refdriver.py region COORD HEIGHT   {"key":…, "lookup_id":…, "x":…, "y":…, "z":…, "plane":…}
  refdriver.py hints                 §7.7's golden hint vectors, one per line
  refdriver.py encrypt COORD HEIGHT  seal stdin into a kind-33330 bag at that region
  refdriver.py box COORD HX HY HZ    the §7.7 hint tag for the box around that coord

Paths come from $CYBERSPACE_DIR and $CYBERSPACE_CLI_DIR.
"""
import importlib.util
import json
import os
import signal
import sys

# Used in pipelines (`| head`), where dying on SIGPIPE is the right answer
# rather than a traceback.
signal.signal(signal.SIGPIPE, signal.SIG_DFL)


def _load(path, name):
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None:
        raise SystemExit(f"cannot load {path}")
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod
    spec.loader.exec_module(mod)
    return mod


def _sno():
    root = os.environ.get("CYBERSPACE_DIR")
    if not root:
        raise SystemExit("CYBERSPACE_DIR is not set")
    return _load(os.path.join(root, "decks", "sno-reference.py"), "sno_reference")


def _avatar():
    root = os.environ.get("CYBERSPACE_CLI_DIR")
    if not root:
        raise SystemExit("CYBERSPACE_CLI_DIR is not set")
    sys.path.insert(0, os.path.join(root, "src"))
    return _load(os.path.join(root, "src", "cyberspace_core", "avatar.py"), "ref_avatar")


def emit(obj):
    print(json.dumps(obj, separators=(",", ":"), sort_keys=True))


def cmd_cases():
    sno = _sno()
    emit({"name": "appendix A", "rule": None, "payload": dict(sno.APPENDIX_A)})
    for label, payload in sno._rejections():
        emit({"name": label, "rule": label.replace("rule ", ""), "payload": payload})


def cmd_vectors():
    root = os.environ.get("CYBERSPACE_CLI_DIR")
    if not root:
        raise SystemExit("CYBERSPACE_CLI_DIR is not set")
    with open(os.path.join(root, "tests", "fixtures", "avatar_work.json")) as fh:
        for case in json.load(fh):
            emit({"name": case["name"], "required": case["required"], "payload": case["payload"]})


def cmd_verdict():
    sno = _sno()
    payload = json.load(sys.stdin)
    try:
        sno.validate(dict(payload))
        emit({"valid": True, "rule": None})
    except sno.SnoError as err:
        text = str(err)
        rule = text.split(":")[0].replace("rule ", "").strip() if text.startswith("rule ") else "?"
        emit({"valid": False, "rule": rule})


def cmd_work():
    avatar = _avatar()
    emit({"required": avatar.avatar_work(json.load(sys.stdin))})


def cmd_verify():
    avatar = _avatar()
    result = avatar.verify_avatar_work(json.load(sys.stdin))
    emit({k: result[k] for k in ("ok", "required", "committed", "zeros", "reason")})


def cmd_region():
    """§2.2 decode plus §7.2 derivation, from cyberspace-cli's own modules.

    `location_encryption.py` imports AESGCM at module scope and that binding is
    not always present; `cantor` and `movement` are the whole of what a region
    key needs, and going through them keeps this an adapter rather than a
    reimplementation.
    """
    sys.path.insert(0, os.path.join(os.environ["CYBERSPACE_CLI_DIR"], "src"))
    from cyberspace_core.coords import coord_to_xyz
    from cyberspace_core.cantor import cantor_pair, int_to_bytes_be_min, sha256
    from cyberspace_core.movement import compute_subtree_cantor

    coord_hex, height = sys.argv[2], int(sys.argv[3])
    x, y, z, plane = coord_to_xyz(int(coord_hex, 16))
    base = lambda v: (v >> height) << height if height > 0 else v
    region_n = cantor_pair(
        cantor_pair(
            compute_subtree_cantor(base(x), height),
            compute_subtree_cantor(base(y), height),
        ),
        compute_subtree_cantor(base(z), height),
    )
    key = sha256(int_to_bytes_be_min(region_n))
    emit({
        "key": key.hex(),
        "lookup_id": sha256(key).hex(),
        "x": str(x), "y": str(y), "z": str(z),
        "plane": "ideaspace" if plane else "dataspace",
    })


def cmd_hints():
    """§7.7's golden vectors, straight out of the spec's own hint-reference.py."""
    root = os.environ["CYBERSPACE_DIR"]
    # It sits at the repo root, beside CYBERSPACE_V2.md, rather than under
    # decks/ where the DECK references live; accept either in case it moves.
    path = next(
        (p for p in (os.path.join(root, "hint-reference.py"), os.path.join(root, "decks", "hint-reference.py")) if os.path.exists(p)),
        None,
    )
    if path is None:
        raise SystemExit("hint-reference.py not found under $CYBERSPACE_DIR")
    hint_ref = _load(path, "hint_reference")
    for name, vector in hint_ref.vectors().items():
        emit({"name": name, **vector})


def cmd_encrypt():
    """Seal stdin into a §8.6 bag, using cyberspace-cli's own cipher and event builder.

    The point of the round trip is that nothing on this side is ours: the key
    comes from `location_encryption`, the sealing from `encrypt_with_location_key`
    and the tags from `make_encrypted_content_event`. What `amy cyberspace open`
    then has to do is derive the same key from the same coordinate and read a
    ciphertext it never saw made.

    A fixed nonce keeps the fixture reproducible; §7.6 wants a fresh one per
    bag, which matters for a hider and not for a test that seals once.
    """
    sys.path.insert(0, os.path.join(os.environ["CYBERSPACE_CLI_DIR"], "src"))
    import base64

    from cyberspace_core.coords import coord_to_xyz
    from cyberspace_cli.nostr_event import make_encrypted_content_event
    from cyberspace_core.location_encryption import (
        derive_region_key_material_for_height,
        encrypt_with_location_key,
    )

    coord_hex, height = sys.argv[2], int(sys.argv[3])
    x, y, z, _plane = coord_to_xyz(int(coord_hex, 16))
    material = derive_region_key_material_for_height(x=x, y=y, z=z, height=height)
    payload = encrypt_with_location_key(
        sys.stdin.buffer.read(),
        location_decryption_key=material.location_decryption_key,
        nonce=bytes(range(12)),
    )
    event = make_encrypted_content_event(
        pubkey_hex="b" * 64,
        created_at=1,
        lookup_id_hex=material.lookup_id_hex,
        algorithm="aes-256-gcm",
        ciphertext_b64=base64.b64encode(payload).decode("ascii"),
        height_hint=height,
        content="",
        kind=33330,
    )
    emit({
        "event": event,
        "key": material.location_decryption_key.hex(),
        "lookup_id": material.lookup_id_hex,
    })


def cmd_box():
    """The §7.7 `hint` tag naming the aligned box around a coordinate.

    Built from the reference's own interleave so a sweep test is not marking
    its own homework: the box comes from their `coord_to_xyz`/`xyz_to_coord`,
    and if our alignment disagreed the bag would simply not be in the box we
    were handed.
    """
    sys.path.insert(0, os.path.join(os.environ["CYBERSPACE_CLI_DIR"], "src"))
    from cyberspace_core.coords import coord_to_xyz, xyz_to_coord

    coord_hex = sys.argv[2]
    hx, hy, hz = (int(v) for v in sys.argv[3:6])
    x, y, z, plane = coord_to_xyz(int(coord_hex, 16))
    base = xyz_to_coord((x >> hx) << hx, (y >> hy) << hy, (z >> hz) << hz, plane)
    emit({"tag": ["hint", f"{base:064x}", str(hx), str(hy), str(hz)]})


COMMANDS = {
    "box": cmd_box,
    "cases": cmd_cases,
    "encrypt": cmd_encrypt,
    "region": cmd_region,
    "hints": cmd_hints,
    "vectors": cmd_vectors,
    "verdict": cmd_verdict,
    "work": cmd_work,
    "verify": cmd_verify,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
        raise SystemExit(f"usage: refdriver.py <{'|'.join(COMMANDS)}>")
    COMMANDS[sys.argv[1]]()