File size: 4,978 Bytes
b0951d6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | """
Reference SSZ-V1 encoder/merkleizer for KAY Sentinel's SemanticStateRecord.
This is a REFERENCE implementation of the Layer 2 serialization + merkleization
rules in docs/ssz_profile.md. It is NOT a Geth/Reth extractor -- it takes the
already-normalized `expected_normalized_ssr` block from a validation_vectors/
fixture and encodes it, so it validates the encoding/merkleization layer only.
Client-fidelity (does a real trace actually normalize to this SSR) is still
untested; that's Phase 2/3 of the implementation roadmap.
Depends on `remerkleable` (https://pypi.org/project/remerkleable/), an
existing, independently-maintained SSZ implementation used in Ethereum
consensus-layer tooling -- this script does not reimplement SSZ from scratch.
Usage:
pip install remerkleable
python ssz_reference.py path/to/validation_vectors/KS-ST-001.json
python ssz_reference.py path/to/validation_vectors/*.json --write-manifest schemas/ssz_v1_conformance_manifest.json
"""
import argparse
import glob
import json
import sys
from remerkleable.basic import uint8, uint64, uint256
from remerkleable.byte_arrays import ByteVector, Bytes32
from remerkleable.complex import Container, List
Bytes20 = ByteVector[20]
MAX_STORAGE_UPDATES = 65536 # 2^16
MAX_ACCOUNTS = 16384 # 2^14
class StorageUpdate(Container):
slot: Bytes32
original: Bytes32
current: Bytes32
class AccountDelta(Container):
address: Bytes20
original_nonce: uint64
current_nonce: uint64
original_balance: uint256
current_balance: uint256
original_code_hash: Bytes32
current_code_hash: Bytes32
storage_updates: List[StorageUpdate, MAX_STORAGE_UPDATES]
class SemanticStateRecord(Container):
consensus_outcome: uint8 # 0 = Included, 1 = Rejected
execution_outcome: uint8 # 0 = Success, 1 = Revert, 2 = ExceptionalHalt
account_deltas: List[AccountDelta, MAX_ACCOUNTS]
CONSENSUS_MAP = {"Included": 0, "Rejected": 1}
EXECUTION_MAP = {"Success": 0, "Revert": 1, "ExceptionalHalt": 2}
def build_ssr(vector: dict) -> SemanticStateRecord:
ssr_data = vector["expected_normalized_ssr"]
account_views = []
for acc in ssr_data["account_deltas"]:
su_views = [
StorageUpdate(
slot=Bytes32(bytes.fromhex(su["slot"][2:])),
original=Bytes32(bytes.fromhex(su["original"][2:])),
current=Bytes32(bytes.fromhex(su["current"][2:])),
)
for su in acc["storage_updates"]
]
account_views.append(
AccountDelta(
address=Bytes20(bytes.fromhex(acc["address"][2:])),
original_nonce=uint64(acc["nonce"]["original"]),
current_nonce=uint64(acc["nonce"]["current"]),
original_balance=uint256(int(acc["balance"]["original"])),
current_balance=uint256(int(acc["balance"]["current"])),
original_code_hash=Bytes32(bytes.fromhex(acc["code_hash"]["original"][2:])),
current_code_hash=Bytes32(bytes.fromhex(acc["code_hash"]["current"][2:])),
storage_updates=List[StorageUpdate, MAX_STORAGE_UPDATES](*su_views),
)
)
return SemanticStateRecord(
consensus_outcome=uint8(CONSENSUS_MAP[ssr_data["consensus_outcome"]]),
execution_outcome=uint8(EXECUTION_MAP[ssr_data["execution_outcome"]]),
account_deltas=List[AccountDelta, MAX_ACCOUNTS](*account_views),
)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("vector_files", nargs="+", help="validation_vectors/*.json files")
parser.add_argument("--write-manifest", help="path to write a Layer 2 conformance manifest JSON")
args = parser.parse_args()
files = []
for pattern in args.vector_files:
files.extend(sorted(glob.glob(pattern)) or [pattern])
manifest_targets = []
for path in files:
vector = json.load(open(path))
ssr = build_ssr(vector)
encoded = ssr.encode_bytes()
root = "0x" + ssr.hash_tree_root().hex()
length = len(encoded)
vid = vector["vector_id"]
print(f"{vid}: serialized_length={length} expected_root_hash={root}")
manifest_targets.append({
"vector_id": vid,
"serialized_length": length,
"expected_root_hash": root,
})
if args.write_manifest:
manifest = {
"profile_id": "SSZ-V1",
"profile_target_version": "1.0.0",
"status": "computed by scripts/ssz_reference.py against remerkleable; "
"validates encoding/merkleization only, not client fidelity",
"vector_targets": manifest_targets,
}
with open(args.write_manifest, "w") as f:
json.dump(manifest, f, indent=2)
f.write("\n")
print(f"\nWrote manifest to {args.write_manifest}")
if __name__ == "__main__":
sys.exit(main())
|