fa958cc051
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29
255 lines
12 KiB
Python
255 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Silent attended OpenBao admission/publisher; never emits server responses."""
|
|
import argparse
|
|
import datetime as dt
|
|
import fcntl
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
from root_grant_publish import MAX_BYTES, RECORD_ID, prepare_payload
|
|
from openbao_platform_admin_check import require_attended
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PATH = "platform/data/custody/net-kingdom/platform-root"
|
|
META = "platform/metadata/custody/net-kingdom/platform-root"
|
|
ROLE_PATH = "auth/kubernetes/role/user-engine-root-custody-reader"
|
|
POLICY = "user-engine-root-custody-reader"
|
|
ROLE = {"bound_service_account_names": ["user-engine"],
|
|
"bound_service_account_namespaces": ["user-engine"], "audience": "openbao",
|
|
"token_policies": [POLICY], "token_no_default_policy": True,
|
|
"token_ttl": 60, "token_max_ttl": 60, "token_explicit_max_ttl": 60}
|
|
|
|
|
|
class Refused(Exception):
|
|
pass
|
|
|
|
|
|
def digest(document):
|
|
return hashlib.sha256(json.dumps(document, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()).hexdigest()
|
|
|
|
|
|
def atomic_json(path, document):
|
|
path = Path(path)
|
|
if path.is_symlink():
|
|
raise Refused("unsafe_state_path")
|
|
with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as out:
|
|
temporary = Path(out.name)
|
|
os.chmod(temporary, 0o600)
|
|
out.write(json.dumps(document, sort_keys=True).encode())
|
|
out.flush()
|
|
os.fsync(out.fileno())
|
|
try:
|
|
os.replace(temporary, path)
|
|
fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
class Bao:
|
|
def call(self, command, payload=None, optional=False):
|
|
result = subprocess.run(["bao", *command], input=None if payload is None else json.dumps(payload).encode(),
|
|
capture_output=True, timeout=15)
|
|
if result.returncode:
|
|
# Distinguish a documented missing-path message from auth/outage.
|
|
if optional and result.returncode == 2 and result.stderr.strip() == ("No value found at " + command[-1]).encode():
|
|
return None
|
|
raise Refused("openbao_request_failed")
|
|
if len(result.stdout) > MAX_BYTES * 2:
|
|
raise Refused("oversize_owner_response")
|
|
return json.loads(result.stdout) if result.stdout.strip() else {}
|
|
|
|
def read(self, path, optional=False):
|
|
response = self.call(["read", "-format=json", path], optional=optional)
|
|
return None if response is None else response["data"]
|
|
|
|
def write(self, path, payload):
|
|
return self.call(["write", "-format=json", path, "-"], payload)
|
|
|
|
|
|
def check_baseline(actual, state, validator):
|
|
if state is None:
|
|
if actual is not None:
|
|
raise Refused("controlled_baseline_import_required")
|
|
return
|
|
previous = state["document"]
|
|
if validator.validate(previous) or previous["record_id"] != RECORD_ID or state["digest"] != digest(previous):
|
|
raise Refused("invalid_retained_baseline")
|
|
if actual is None or validator.validate(actual["data"]["document"]):
|
|
raise Refused("missing_or_invalid_source")
|
|
document = actual["data"]["document"]
|
|
if digest(document) != state["digest"] or actual["metadata"]["version"] != state["kv_version"]:
|
|
raise Refused("source_differs_from_retained_baseline")
|
|
|
|
|
|
def retain_history(state_path, receipt):
|
|
history = state_path.parent / "history"
|
|
history.mkdir(mode=0o700, exist_ok=True)
|
|
target = history / (str(receipt["revision"]) + "-" + receipt["digest"] + ".json")
|
|
if target.exists():
|
|
if json.loads(target.read_text()) != receipt:
|
|
raise Refused("history_receipt_differs")
|
|
return
|
|
with target.open("x") as out:
|
|
os.chmod(target, 0o600)
|
|
json.dump(receipt, out, sort_keys=True)
|
|
out.flush()
|
|
os.fsync(out.fileno())
|
|
fd = os.open(history, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def publish(bao, candidate, state_path, validator, provenance):
|
|
state_path = Path(state_path)
|
|
if not isinstance(provenance, str) or not provenance.strip() or len(provenance) > 300 or validator.SECRET_VALUE.search(provenance):
|
|
raise Refused("nonsecret_human_provenance_required")
|
|
if state_path.is_symlink():
|
|
raise Refused("unsafe_state_path")
|
|
state = json.loads(state_path.read_text()) if state_path.exists() else None
|
|
actual = bao.read(PATH, optional=True)
|
|
# Pending state is fsynced before network mutation. Recover only exact readback;
|
|
# ambiguous network failures are never retried or cleared automatically.
|
|
if state is not None and state.get("pending"):
|
|
pending = state["pending"]
|
|
if actual is None or digest(actual["data"]["document"]) != pending["digest"] or actual["metadata"]["version"] != pending["expected_kv_version"]:
|
|
raise Refused("pending_publication_requires_owner_review")
|
|
if validator.validate(pending["document"]) or pending["document"]["record_id"] != RECORD_ID:
|
|
raise Refused("invalid_pending_document")
|
|
state = {key: value for key, value in pending.items() if key != "expected_kv_version"}
|
|
retain_history(state_path, state)
|
|
atomic_json(state_path, state)
|
|
check_baseline(actual, state, validator)
|
|
previous = None if state is None else state["document"]
|
|
version = 0 if actual is None else actual["metadata"]["version"]
|
|
if previous is not None and digest(candidate) == state["digest"]:
|
|
# Exact replay is an idempotent readback, not a new custody/KV revision.
|
|
return {**{key: value for key, value in state.items() if key != "document"}, "unchanged": True}
|
|
payload = prepare_payload(candidate, previous, version, validator)
|
|
now = dt.datetime.now(dt.timezone.utc)
|
|
times = [candidate["updated_at"]]
|
|
for entry in candidate["entries"]:
|
|
times.append(entry["granted"]["at"])
|
|
if entry.get("revoked"):
|
|
times.append(entry["revoked"]["at"])
|
|
if any(validator.parse_time(value) > now for value in times):
|
|
raise Refused("future_custody_action")
|
|
receipt = {"record_id": RECORD_ID, "revision": candidate["revision"], "digest": digest(candidate),
|
|
"kv_version": version + 1, "document": candidate, "provenance_ref": provenance,
|
|
"recorded_at": dt.datetime.now(dt.timezone.utc).isoformat()}
|
|
pending = dict(receipt, expected_kv_version=version + 1)
|
|
atomic_json(state_path, {**(state or {}), "pending": pending})
|
|
bao.write(PATH, payload)
|
|
returned = bao.read(PATH)
|
|
if returned["metadata"]["version"] != version + 1 or digest(returned["data"]["document"]) != receipt["digest"]:
|
|
raise Refused("publication_readback_failed")
|
|
# Independent immutable receipts precede advancing local state. They must be
|
|
# backed up off-host; restoring only this directory is not independent proof.
|
|
retain_history(state_path, receipt)
|
|
atomic_json(state_path, receipt)
|
|
return {key: value for key, value in receipt.items() if key != "document"}
|
|
|
|
|
|
def provision(bao):
|
|
# Require existing Kubernetes auth; no auth mount or cluster reviewer mutation.
|
|
if bao.read("auth/kubernetes/config") is None:
|
|
raise Refused("kubernetes_auth_unavailable")
|
|
audit = bao.read("sys/audit")
|
|
if not audit:
|
|
raise Refused("audit_required")
|
|
policy = (ROOT / "openbao/root-grant/reader.hcl").read_text()
|
|
existing = bao.read("sys/policies/acl/" + POLICY, optional=True)
|
|
if existing is not None and existing.get("policy", existing.get("rules", "")).strip() != policy.strip():
|
|
raise Refused("existing_reader_policy_differs")
|
|
role = bao.read(ROLE_PATH, optional=True)
|
|
if role is not None and any(role.get(key) != value for key, value in ROLE.items()):
|
|
raise Refused("existing_reader_role_differs")
|
|
if existing is None:
|
|
bao.write("sys/policies/acl/" + POLICY, {"policy": policy})
|
|
if role is None:
|
|
bao.write(ROLE_PATH, ROLE)
|
|
fresh = bao.read(ROLE_PATH)
|
|
if any(fresh.get(key) != value for key, value in ROLE.items()):
|
|
raise Refused("role_readback_failed")
|
|
fresh_policy = bao.read("sys/policies/acl/" + POLICY)
|
|
if fresh_policy.get("policy", fresh_policy.get("rules", "")).strip() != policy.strip():
|
|
raise Refused("policy_readback_failed")
|
|
# Require CAS on this key without altering retention or destroying versions.
|
|
metadata = bao.read(META, optional=True)
|
|
if not metadata or not metadata.get("cas_required"):
|
|
bao.write(META, {"cas_required": True})
|
|
if bao.read(META).get("cas_required") is not True:
|
|
raise Refused("cas_required_readback_failed")
|
|
return {"reader_role": ROLE_PATH, "reader_namespace": "user-engine", "reader_service_account": "user-engine",
|
|
"audience": "openbao", "record_id": RECORD_ID, "cas_required": True, "audit_enabled": True}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("action", choices=["provision", "publish", "restore-check"])
|
|
parser.add_argument("--receipt", type=Path, required=True)
|
|
parser.add_argument("--state", type=Path)
|
|
parser.add_argument("--candidate", type=Path)
|
|
parser.add_argument("--net-kingdom", type=Path)
|
|
parser.add_argument("--provenance-ref")
|
|
parser.add_argument("--validator-digest")
|
|
parser.add_argument("--schema-digest")
|
|
args = parser.parse_args()
|
|
try:
|
|
if args.receipt.exists() or args.receipt.is_symlink():
|
|
raise Refused("unique_run_receipt_required")
|
|
require_attended()
|
|
if os.getenv("BAO_ADDR") != "http://127.0.0.1:18200":
|
|
raise Refused("approved_operator_tunnel_required")
|
|
bao = Bao()
|
|
if args.action == "provision":
|
|
receipt = provision(bao)
|
|
else:
|
|
if not args.state or not args.net_kingdom or not args.validator_digest or not args.schema_digest:
|
|
raise Refused("pinned_source_and_state_required")
|
|
source = args.net_kingdom / "tools/custody-record/validate.py"
|
|
schema = args.net_kingdom / "canon/schemas/custody-record_v0.1.schema.json"
|
|
for path, expected in [(source, args.validator_digest), (schema, args.schema_digest)]:
|
|
if hashlib.sha256(path.read_bytes()).hexdigest() != expected:
|
|
raise Refused("canon_pin_mismatch")
|
|
spec = importlib.util.spec_from_file_location("custody_validator", source)
|
|
validator = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(validator)
|
|
if not args.state.parent.is_dir():
|
|
raise Refused("owner_state_directory_required")
|
|
# Serialize this publisher, including reads and pending-journal recovery.
|
|
with (args.state.parent / "publisher.lock").open("a") as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX)
|
|
if args.action == "restore-check":
|
|
if not args.state.exists():
|
|
raise Refused("retained_baseline_required")
|
|
state = json.loads(args.state.read_text())
|
|
if state.get("pending"):
|
|
raise Refused("pending_publication_requires_owner_review")
|
|
check_baseline(bao.read(PATH, optional=True), state, validator)
|
|
receipt = {"record_id": RECORD_ID, "baseline_matches": True, "revision": state["document"]["revision"]}
|
|
else:
|
|
if not args.candidate or not args.provenance_ref:
|
|
raise Refused("candidate_and_human_provenance_required")
|
|
candidate = json.loads(args.candidate.read_bytes()[:MAX_BYTES + 1])
|
|
receipt = publish(bao, candidate, args.state, validator, args.provenance_ref)
|
|
atomic_json(args.receipt, {"ok": True, **receipt})
|
|
return 0
|
|
except Exception:
|
|
# CLI/process/validation errors may embed secrets. Never emit them.
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|