8ccd6827cb
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29
42 lines
1.9 KiB
Python
42 lines
1.9 KiB
Python
"""Offline CAS payload preparation; this module never calls OpenBao.
|
|
|
|
The caller supplies the reviewed NetKingdom validator. Its deployment must pin
|
|
both validator and schema. Returning a payload does not approve a custody action.
|
|
"""
|
|
import copy
|
|
import json
|
|
|
|
RECORD_ID = "nk-custody:platform-root"
|
|
OWNER_SYSTEM = "net-kingdom-custody"
|
|
MAX_BYTES = 262144
|
|
|
|
|
|
def prepare_payload(candidate, previous, previous_kv_version, validator):
|
|
"""Require current authoritative baseline; CAS also fences concurrent writes.
|
|
|
|
Bootstrap must use cas=0 and no previous document. Existing records require
|
|
both the last successful owner-read document and its exact KV version.
|
|
The caller must obtain those together and retain the high-water baseline.
|
|
Errors intentionally disclose neither documents nor rejected values.
|
|
"""
|
|
if type(previous_kv_version) is not int or previous_kv_version < 0:
|
|
raise ValueError("invalid KV version")
|
|
if (previous is None) != (previous_kv_version == 0):
|
|
raise ValueError("missing or inconsistent publication baseline")
|
|
documents = [candidate] if previous is None else [previous, candidate]
|
|
for document in documents:
|
|
try:
|
|
encoded = json.dumps(document, allow_nan=False).encode("utf-8")
|
|
if len(encoded) > MAX_BYTES or validator.validate(document):
|
|
raise ValueError
|
|
if document["record_id"] != RECORD_ID or document["owner_system"] != OWNER_SYSTEM:
|
|
raise ValueError
|
|
except Exception:
|
|
raise ValueError("invalid custody document") from None
|
|
if previous is not None:
|
|
try:
|
|
if candidate["revision"] <= previous["revision"] or validator.successor_errors(previous, candidate):
|
|
raise ValueError
|
|
except Exception:
|
|
raise ValueError("invalid custody transition") from None
|
|
return {"options": {"cas": previous_kv_version}, "data": {"document": copy.deepcopy(candidate)}}
|