railiance-platform/scripts/wp0024-t02-driver-candidate.py

#!/usr/bin/env python3
"""Register or inspect a revision-pinned T02 synthetic-load driver candidate.

Registration reads and hashes committed source.  It never executes the driver,
reads a bearer, or authorizes the live database-lease exercise.
"""

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


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CONTRACT = ROOT / "interfaces" / "RAILIANCE-WP-0024-T02-synthetic-load-driver.json"
DEFAULT_AUDIT_REPO = ROOT.parent / "audit-core"
DEFAULT_API_BASE = os.environ.get("STATE_HUB_URL", "http://127.0.0.1:8000")
PREFIX = "WP0024-T02-DRIVER-CANDIDATE"


class CandidateError(RuntimeError):
    pass


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


def digest(value: bytes) -> str:
    return hashlib.sha256(value).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 CandidateError("synthetic-load contract is unavailable or invalid") from exc
    if (
        not isinstance(value, dict)
        or value.get("interface") != "railiance.synthetic-load-driver"
        or value.get("version") != 1
        or value.get("task_id") != "RAILIANCE-WP-0024-T02"
        or value.get("owner") != "audit-core"
    ):
        raise CandidateError("unsupported synthetic-load contract")
    phases = value.get("phases")
    if not isinstance(phases, dict) or set(phases) != {
        "baseline", "expect-unavailable", "expect-recovered", "cleanup"
    }:
        raise CandidateError("synthetic-load contract has the wrong phase set")
    for phase, record in phases.items():
        if not isinstance(record, dict) or not record.get("keys") or not record.get("status"):
            raise CandidateError(f"synthetic-load phase is incomplete: {phase}")
    return value


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


def run_git(repo: Path, args: list[str]) -> str:
    completed = subprocess.run(
        ["git", "-C", str(repo), *args], text=True, capture_output=True, check=False
    )
    if completed.returncode != 0:
        raise CandidateError("cannot resolve committed driver source")
    return completed.stdout.strip()


def inspect_driver(repo: Path, source_path: str) -> dict[str, str]:
    repo = repo.resolve()
    target = (repo / source_path).resolve()
    try:
        relative = target.relative_to(repo).as_posix()
    except ValueError as exc:
        raise CandidateError("driver path escapes the audit-core repository") from exc
    if not target.is_file() or not os.access(target, os.X_OK):
        raise CandidateError("driver candidate must be an executable file")
    if run_git(repo, ["status", "--porcelain", "--", relative]):
        raise CandidateError("driver candidate must be committed with no path-local changes")
    tracked = run_git(repo, ["ls-files", "--error-unmatch", "--", relative])
    if tracked != relative:
        raise CandidateError("driver candidate is not tracked at the requested path")
    revision = run_git(repo, ["log", "-1", "--format=%H", "--", relative])
    if not re.fullmatch(r"[0-9a-f]{40}", revision):
        raise CandidateError("driver source revision is not a full Git commit")
    return {
        "source_repo": "audit-core",
        "source_revision": revision,
        "source_path": relative,
        "driver_revision": "sha256:" + digest(target.read_bytes()),
    }


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


def build_receipt(
    contract: dict[str, Any], source: dict[str, str], reviewer_id: str,
    *, now: datetime | None = None,
) -> dict[str, Any]:
    return {
        "interface": "railiance.synthetic-load-driver-candidate",
        "version": 1,
        "workplan_id": contract["workplan_id"],
        "task_id": contract["task_id"],
        "contract_id": contract["contract_id"],
        "contract_digest": contract_digest(contract),
        "owner": "audit-core",
        "reviewer": reviewer(reviewer_id),
        **source,
        "created_at": (now or datetime.now(UTC)).astimezone(UTC).isoformat().replace("+00:00", "Z"),
        "registration_executed_driver": False,
        "secret_values_observed": False,
    }


def subject(receipt: dict[str, Any]) -> str:
    return "/".join((PREFIX, "v1", receipt["contract_digest"], receipt["driver_revision"][7:]))


def http_json(method: str, url: str, payload: dict[str, Any] | None = None) -> Any:
    data = canonical(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 urllib.request.urlopen(request, timeout=10) as response:
            return json.loads(response.read().decode("utf-8"))
    except (OSError, urllib.error.URLError, json.JSONDecodeError) as exc:
        raise CandidateError(f"State Hub request failed: {method} {url}") from exc


def post(receipt: dict[str, Any], api_base: str) -> dict[str, Any]:
    value = http_json(
        "POST",
        f"{api_base.rstrip('/')}/messages/",
        {
            "from_agent": "audit-core",
            "to_agent": "railiance-platform",
            "subject": subject(receipt),
            "body": canonical(receipt),
        },
    )
    if not isinstance(value, dict):
        raise CandidateError("State Hub returned an invalid candidate response")
    return value


def parse_message(message: dict[str, Any], contract: dict[str, Any]) -> dict[str, Any] | None:
    if message.get("from_agent") != "audit-core":
        return None
    try:
        receipt = json.loads(message["body"])
    except (KeyError, TypeError, json.JSONDecodeError):
        return None
    if not isinstance(receipt, dict):
        return None
    try:
        valid = (
            message.get("subject") == subject(receipt)
            and receipt.get("interface") == "railiance.synthetic-load-driver-candidate"
            and receipt.get("version") == 1
            and receipt.get("task_id") == contract["task_id"]
            and receipt.get("contract_id") == contract["contract_id"]
            and receipt.get("contract_digest") == contract_digest(contract)
            and receipt.get("owner") == "audit-core"
            and receipt.get("source_repo") == "audit-core"
            and re.fullmatch(r"[0-9a-f]{40}", receipt.get("source_revision", ""))
            and re.fullmatch(r"sha256:[0-9a-f]{64}", receipt.get("driver_revision", ""))
            and receipt.get("registration_executed_driver") is False
            and receipt.get("secret_values_observed") is False
        )
    except (KeyError, TypeError):
        return None
    return receipt if valid else None


def 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 CandidateError("State Hub messages response is not a list")
    candidates = [item for message in value if (item := parse_message(message, contract))]
    latest = max(candidates, key=lambda item: item.get("created_at", ""), default=None)
    return {
        "interface": contract["interface"],
        "version": contract["version"],
        "task_id": contract["task_id"],
        "contract_id": contract["contract_id"],
        "contract_digest": contract_digest(contract),
        "candidate_registered": latest is not None,
        "candidate": latest,
    }


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)
    sub = result.add_subparsers(dest="command", required=True)
    sub.add_parser("show")
    register = sub.add_parser("register")
    register.add_argument("--repo", type=Path, default=DEFAULT_AUDIT_REPO)
    register.add_argument("--driver", required=True, help="path relative to the audit-core repo")
    register.add_argument("--reviewer", required=True)
    sub.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":
            output = {**contract, "contract_digest": contract_digest(contract)}
        elif args.command == "register":
            source = inspect_driver(args.repo, args.driver)
            receipt = build_receipt(contract, source, args.reviewer)
            response = post(receipt, args.api_base)
            output = {"submitted": True, "message_id": response.get("id"), "receipt": receipt}
        else:
            output = status(contract, args.api_base)
    except CandidateError as exc:
        print(canonical({"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())