railiance-platform/scripts/wp0024-owner-review.py

#!/usr/bin/env python3
"""Direct, value-safe owner review interface for RAILIANCE-WP-0024.

The executable surface is intentionally closed: contract files select named
read-only checks, never shell commands.  Approvals are bound to the canonical
contract digest and the SHA-256 digests of the owner's reviewed artifacts.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Callable


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONTRACT = ROOT / "interfaces" / "RAILIANCE-WP-0024-owner-reviews.json"
DEFAULT_API_BASE = os.environ.get("STATE_HUB_URL", "http://127.0.0.1:8000")
RECEIPT_PREFIX = "WP0024-OWNER-RECEIPT"
KNOWN_CHECKS = {
    "focused-unit-tests",
    "database-lease-preflight",
    "node-reboot-preflight",
}


class ReviewError(RuntimeError):
    pass


def canonical_json(value: Any) -> str:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def load_contract(path: Path = DEFAULT_CONTRACT) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ReviewError(f"review contract is unavailable or invalid: {path}") from exc
    if not isinstance(value, dict):
        raise ReviewError("review contract must be a JSON object")
    if value.get("interface") != "railiance.owner-review" or value.get("version") != 1:
        raise ReviewError("unsupported owner-review interface/version")
    owners = value.get("owners")
    hashes = value.get("artifact_sha256")
    task_owners = value.get("task_owners")
    if (
        not isinstance(owners, dict)
        or not owners
        or not isinstance(hashes, dict)
        or not isinstance(task_owners, dict)
        or not task_owners
    ):
        raise ReviewError(
            "review contract requires owners, task_owners and artifact_sha256 maps"
        )
    for owner, review in owners.items():
        if not isinstance(owner, str) or not isinstance(review, dict):
            raise ReviewError("invalid owner review entry")
        artifacts = review.get("artifacts")
        checks = review.get("checks")
        assertions = review.get("assertions")
        if not all(isinstance(item, str) for item in artifacts or []):
            raise ReviewError(f"{owner}: artifacts must be strings")
        if not artifacts or any(item not in hashes for item in artifacts):
            raise ReviewError(f"{owner}: every artifact requires a pinned digest")
        if not checks or any(item not in KNOWN_CHECKS for item in checks):
            raise ReviewError(f"{owner}: unknown or missing read-only check")
        if not assertions or not all(isinstance(item, str) for item in assertions):
            raise ReviewError(f"{owner}: assertions must be non-empty strings")
        expected_tasks = sorted(
            task for task, required in task_owners.items() if owner in required
        )
        if sorted(review.get("tasks") or []) != expected_tasks:
            raise ReviewError(f"{owner}: tasks disagree with task_owners")
    for task, required in task_owners.items():
        if not isinstance(task, str) or not isinstance(required, list) or not required:
            raise ReviewError("task_owners entries must be non-empty owner lists")
        if len(set(required)) != len(required) or any(owner not in owners for owner in required):
            raise ReviewError(f"{task}: task_owners contains unknown or duplicate owners")
    return value


def contract_digest(contract: dict[str, Any]) -> str:
    return sha256_bytes(canonical_json(contract).encode("utf-8"))


def owner_review(contract: dict[str, Any], owner: str) -> dict[str, Any]:
    review = contract["owners"].get(owner)
    if not isinstance(review, dict):
        choices = ", ".join(sorted(contract["owners"]))
        raise ReviewError(f"unknown owner {owner!r}; choose one of: {choices}")
    return review


def artifact_state(
    contract: dict[str, Any], owner: str, *, root: Path = ROOT
) -> dict[str, dict[str, Any]]:
    review = owner_review(contract, owner)
    result: dict[str, dict[str, Any]] = {}
    for relative in review["artifacts"]:
        expected = contract["artifact_sha256"][relative]
        target = (root / relative).resolve()
        try:
            target.relative_to(root.resolve())
        except ValueError as exc:
            raise ReviewError(f"artifact escapes repository root: {relative}") from exc
        try:
            observed = file_sha256(target)
        except OSError:
            observed = None
        result[relative] = {
            "expected_sha256": expected,
            "observed_sha256": observed,
            "matches": observed == expected,
        }
    return result


def check_command(check_id: str) -> list[str]:
    if check_id == "focused-unit-tests":
        return [
            sys.executable,
            "-m",
            "unittest",
            "tests/test_audit_core_recovery_preflight.py",
            "tests/test_audit_core_database_lease_recovery.py",
            "tests/test_wp0024_owner_review.py",
        ]
    if check_id == "database-lease-preflight":
        return [sys.executable, "scripts/audit-core-recovery-preflight.py", "database-lease"]
    if check_id == "node-reboot-preflight":
        return [sys.executable, "scripts/audit-core-recovery-preflight.py", "node-reboot"]
    raise ReviewError(f"unknown check id: {check_id}")


def run_named_check(
    check_id: str,
    *,
    runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
) -> dict[str, Any]:
    completed = runner(
        check_command(check_id),
        cwd=ROOT,
        text=True,
        capture_output=True,
        check=False,
    )
    result: dict[str, Any] = {
        "id": check_id,
        "passed": completed.returncode == 0,
        "exit_code": completed.returncode,
        "read_only": True,
    }
    if check_id.endswith("-preflight") and completed.returncode == 0:
        try:
            payload = json.loads(completed.stdout)
        except json.JSONDecodeError:
            result["passed"] = False
            result["detail"] = "preflight returned invalid JSON"
        else:
            safe = (
                isinstance(payload, dict)
                and payload.get("automated_checks_passed") is True
                and payload.get("secret_values_observed") is False
            )
            result["passed"] = bool(safe)
            result["automated_checks_passed"] = payload.get("automated_checks_passed")
            result["ready_for_live_execution"] = payload.get("ready_for_live_execution")
            result["secret_values_observed"] = payload.get("secret_values_observed")
            if not safe:
                result["detail"] = "preflight safety/automated gate did not pass"
    elif completed.returncode != 0:
        result["detail"] = "check failed; run the named interface verify command locally for diagnosis"
    return result


def verify(
    contract: dict[str, Any],
    owner: str,
    *,
    root: Path = ROOT,
    check_runner: Callable[[str], dict[str, Any]] = run_named_check,
) -> dict[str, Any]:
    artifacts = artifact_state(contract, owner, root=root)
    checks: list[dict[str, Any]] = []
    if all(item["matches"] for item in artifacts.values()):
        checks = [check_runner(item) for item in owner_review(contract, owner)["checks"]]
    return {
        "interface": contract["interface"],
        "version": contract["version"],
        "workplan_id": contract["workplan_id"],
        "owner": owner,
        "contract_digest": contract_digest(contract),
        "artifacts": artifacts,
        "checks": checks,
        "passed": bool(checks)
        and all(item["matches"] for item in artifacts.values())
        and all(item["passed"] for item in checks),
        "secret_values_observed": False,
    }


def validate_reviewer(value: str) -> str:
    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:@/+\-]{1,127}", value):
        raise ReviewError("reviewer must be a stable 2-128 character identifier")
    return value


def validate_note(value: str) -> str:
    note = value.strip()
    if not note or len(note) > 2000:
        raise ReviewError("request-changes requires a note of 1-2000 characters")
    forbidden = ("BEGIN PRIVATE KEY", "BEGIN OPENSSH PRIVATE KEY", "AGE-SECRET-KEY-1", "hvs.")
    if any(marker in note for marker in forbidden):
        raise ReviewError("note appears to contain credential material; do not submit it")
    return note


def receipt_subject(receipt: dict[str, Any]) -> str:
    return "/".join(
        (
            RECEIPT_PREFIX,
            f"v{receipt['version']}",
            receipt["owner"],
            receipt["decision"],
            receipt["contract_digest"],
        )
    )


def build_receipt(
    contract: dict[str, Any],
    owner: str,
    decision: str,
    reviewer: str,
    *,
    verification: dict[str, Any] | None = None,
    note: str | None = None,
    now: datetime | None = None,
) -> dict[str, Any]:
    if decision not in contract["decisions"]:
        raise ReviewError(f"unsupported decision: {decision}")
    if decision == "approve" and (not verification or verification.get("passed") is not True):
        raise ReviewError("approval requires a passing direct verification")
    if decision == "request-changes":
        note = validate_note(note or "")
    review = owner_review(contract, owner)
    timestamp = (now or datetime.now(UTC)).astimezone(UTC).isoformat().replace("+00:00", "Z")
    receipt: dict[str, Any] = {
        "interface": contract["interface"],
        "version": contract["version"],
        "workplan_id": contract["workplan_id"],
        "workplan_uuid": contract["workplan_uuid"],
        "owner": owner,
        "reviewer": validate_reviewer(reviewer),
        "decision": decision,
        "created_at": timestamp,
        "contract_digest": contract_digest(contract),
        "artifact_sha256": {
            path: contract["artifact_sha256"][path] for path in review["artifacts"]
        },
        "assertion_count": len(review["assertions"]),
        "secret_values_observed": False,
    }
    if verification:
        receipt["checks"] = [
            {"id": item["id"], "passed": item["passed"], "read_only": True}
            for item in verification["checks"]
        ]
    if note is not None:
        receipt["note"] = note
    return receipt


def http_json(
    method: str,
    url: str,
    payload: dict[str, Any] | None = None,
    *,
    opener: Callable[..., Any] = urllib.request.urlopen,
) -> Any:
    data = canonical_json(payload).encode("utf-8") if payload is not None else None
    request = urllib.request.Request(
        url,
        data=data,
        method=method,
        headers={"Content-Type": "application/json"} if data else {},
    )
    try:
        with opener(request, timeout=10) as response:
            return json.loads(response.read().decode("utf-8"))
    except (OSError, urllib.error.URLError, json.JSONDecodeError) as exc:
        raise ReviewError(f"State Hub request failed: {method} {url}") from exc


def post_receipt(receipt: dict[str, Any], api_base: str) -> dict[str, Any]:
    payload = {
        "from_agent": receipt["owner"],
        "to_agent": "railiance-platform",
        "subject": receipt_subject(receipt),
        "body": canonical_json(receipt),
    }
    response = http_json("POST", f"{api_base.rstrip('/')}/messages/", payload)
    if not isinstance(response, dict):
        raise ReviewError("State Hub returned an invalid receipt response")
    return response


def parse_receipt_message(message: dict[str, Any]) -> dict[str, Any] | None:
    subject = message.get("subject")
    body = message.get("body")
    if not isinstance(subject, str) or not subject.startswith(RECEIPT_PREFIX + "/"):
        return None
    if not isinstance(body, str):
        return None
    try:
        receipt = json.loads(body)
    except json.JSONDecodeError:
        return None
    if not isinstance(receipt, dict):
        return None
    required = {
        "interface": str,
        "version": int,
        "workplan_id": str,
        "workplan_uuid": str,
        "owner": str,
        "reviewer": str,
        "decision": str,
        "created_at": str,
        "contract_digest": str,
        "artifact_sha256": dict,
        "assertion_count": int,
        "secret_values_observed": bool,
    }
    if any(not isinstance(receipt.get(key), kind) for key, kind in required.items()):
        return None
    if (
        receipt["interface"] != "railiance.owner-review"
        or receipt["version"] != 1
        or receipt["workplan_id"] != "RAILIANCE-WP-0024"
        or receipt["decision"] not in {"approve", "request-changes"}
        or receipt["secret_values_observed"] is not False
        or not re.fullmatch(r"[0-9a-f]{64}", receipt["contract_digest"])
    ):
        return None
    try:
        datetime.fromisoformat(receipt["created_at"].replace("Z", "+00:00"))
        expected_subject = receipt_subject(receipt)
    except (KeyError, TypeError, ValueError):
        return None
    if expected_subject != subject:
        return None
    if message.get("from_agent") != receipt.get("owner"):
        return None
    return receipt


def aggregate_status(
    contract: dict[str, Any], messages: list[dict[str, Any]]
) -> dict[str, Any]:
    digest = contract_digest(contract)
    candidates: dict[str, list[dict[str, Any]]] = {
        owner: [] for owner in contract["owners"]
    }
    stale: dict[str, int] = {owner: 0 for owner in contract["owners"]}
    for message in messages:
        receipt = parse_receipt_message(message)
        if not receipt or receipt.get("owner") not in candidates:
            continue
        owner = receipt["owner"]
        if receipt.get("contract_digest") != digest:
            stale[owner] += 1
            continue
        expected = {
            path: contract["artifact_sha256"][path]
            for path in contract["owners"][owner]["artifacts"]
        }
        review = contract["owners"][owner]
        try:
            validate_reviewer(receipt["reviewer"])
        except ReviewError:
            stale[owner] += 1
            continue
        if (
            receipt.get("interface") != contract["interface"]
            or receipt.get("version") != contract["version"]
            or receipt.get("workplan_id") != contract["workplan_id"]
            or receipt.get("workplan_uuid") != contract["workplan_uuid"]
            or receipt.get("artifact_sha256") != expected
            or receipt.get("assertion_count") != len(review["assertions"])
        ):
            stale[owner] += 1
            continue
        if receipt["decision"] == "approve":
            checks = receipt.get("checks")
            if (
                not isinstance(checks, list)
                or [item.get("id") for item in checks if isinstance(item, dict)]
                != review["checks"]
                or any(
                    not isinstance(item, dict)
                    or item.get("passed") is not True
                    or item.get("read_only") is not True
                    for item in checks
                )
            ):
                stale[owner] += 1
                continue
        candidates[owner].append(receipt)
    owners: dict[str, Any] = {}
    for owner, receipts in candidates.items():
        latest = max(receipts, key=lambda item: item.get("created_at", ""), default=None)
        owners[owner] = {
            "decision": latest.get("decision") if latest else "missing",
            "reviewer": latest.get("reviewer") if latest else None,
            "created_at": latest.get("created_at") if latest else None,
            "stale_receipt_count": stale[owner],
        }
    tasks = {
        task: {
            "required_owners": required,
            "decisions": {owner: owners[owner]["decision"] for owner in required},
            "all_approved": all(owners[owner]["decision"] == "approve" for owner in required),
        }
        for task, required in contract["task_owners"].items()
    }
    try:
        artifact_current = all(
            file_sha256(ROOT / path) == expected
            for path, expected in contract["artifact_sha256"].items()
        )
    except OSError:
        artifact_current = False
    return {
        "interface": contract["interface"],
        "version": contract["version"],
        "workplan_id": contract["workplan_id"],
        "contract_digest": digest,
        "contract_artifacts_current": artifact_current,
        "owners": owners,
        "tasks": tasks,
        "all_approved": artifact_current
        and all(item["all_approved"] for item in tasks.values()),
    }


def fetch_status(contract: dict[str, Any], api_base: str) -> dict[str, Any]:
    query = urllib.parse.urlencode({"to_agent": "railiance-platform", "limit": 500})
    value = http_json("GET", f"{api_base.rstrip('/')}/messages/?{query}")
    if not isinstance(value, list):
        raise ReviewError("State Hub messages response is not a list")
    return aggregate_status(contract, value)


def parser() -> argparse.ArgumentParser:
    result = argparse.ArgumentParser(description=__doc__)
    result.add_argument("--contract", type=Path, default=DEFAULT_CONTRACT)
    result.add_argument("--api-base", default=DEFAULT_API_BASE)
    subparsers = result.add_subparsers(dest="command", required=True)
    for command in ("show", "verify"):
        child = subparsers.add_parser(command)
        child.add_argument("--owner", required=True)
    approve = subparsers.add_parser("approve")
    approve.add_argument("--owner", required=True)
    approve.add_argument("--reviewer", required=True)
    changes = subparsers.add_parser("request-changes")
    changes.add_argument("--owner", required=True)
    changes.add_argument("--reviewer", required=True)
    changes.add_argument("--note", required=True)
    subparsers.add_parser("status")
    return result


def main(argv: list[str] | None = None) -> int:
    args = parser().parse_args(argv)
    try:
        contract = load_contract(args.contract)
        if args.command == "show":
            review = owner_review(contract, args.owner)
            output = {
                "interface": contract["interface"],
                "version": contract["version"],
                "workplan_id": contract["workplan_id"],
                "owner": args.owner,
                "contract_digest": contract_digest(contract),
                **review,
                "artifacts_state": artifact_state(contract, args.owner),
                "decisions": contract["decisions"],
            }
        elif args.command == "verify":
            output = verify(contract, args.owner)
        elif args.command == "approve":
            verification = verify(contract, args.owner)
            receipt = build_receipt(
                contract, args.owner, "approve", args.reviewer, verification=verification
            )
            posted = post_receipt(receipt, args.api_base)
            output = {
                "submitted": True,
                "message_id": posted.get("id"),
                "receipt": receipt,
            }
        elif args.command == "request-changes":
            receipt = build_receipt(
                contract,
                args.owner,
                "request-changes",
                args.reviewer,
                note=args.note,
            )
            posted = post_receipt(receipt, args.api_base)
            output = {
                "submitted": True,
                "message_id": posted.get("id"),
                "receipt": receipt,
            }
        else:
            output = fetch_status(contract, args.api_base)
    except ReviewError as exc:
        print(canonical_json({"ok": False, "error": str(exc)}), file=sys.stderr)
        return 2
    print(json.dumps(output, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())