harness-tools/scripts/nono-evidence.py

#!/usr/bin/env python3
"""Trusted-snapshot entry point for Nono evidence production.

The checkout copy is source only.  A future maintenance installer must promote
this file and its harness into an immutable snapshot, add the marker file, and
invoke this command with that snapshot's digest.  Running this source directly
therefore fails closed.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import platform
import re
import shutil
import stat
import subprocess
import sys
import uuid
from pathlib import Path
from typing import Any, Callable

REPOSITORY_ROOT = Path(__file__).resolve().parents[1]

_MARKER_CONTENT = "pi-sbx-trusted-snapshot-v1\n"
_DIGEST = re.compile(r"^[0-9a-f]{64}$")
_RAW_OUTPUT_FIELDS = frozenset({"stdout", "stderr", "argv", "env", "environment", "exception", "output"})
_LIVE_TESTS = frozenset({"git-live", "github-cli-live"})


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


def _regular_file_identity(path: Path, *, max_size: int) -> tuple[int, int, int, int, str]:
    if path.is_symlink():
        raise EvidenceCliError(f"identity input cannot be a symlink: {path}")
    flags = (
        os.O_RDONLY
        | getattr(os, "O_NOFOLLOW", 0)
        | getattr(os, "O_NONBLOCK", 0)
    )
    try:
        descriptor = os.open(path, flags)
    except OSError as error:
        raise EvidenceCliError(f"identity input is unavailable: {path}: {error}") from error
    try:
        file_stat = os.fstat(descriptor)
        if not stat.S_ISREG(file_stat.st_mode) or file_stat.st_size > max_size:
            raise EvidenceCliError(
                f"identity input must be a bounded regular file: {path}"
            )
        digest = hashlib.sha256()
        remaining = file_stat.st_size
        while remaining:
            chunk = os.read(descriptor, min(65536, remaining))
            if not chunk:
                break
            digest.update(chunk)
            remaining -= len(chunk)
        final_stat = os.fstat(descriptor)
        if remaining or final_stat.st_size != file_stat.st_size:
            raise EvidenceCliError(f"identity input changed while reading: {path}")
        return (
            file_stat.st_dev,
            file_stat.st_ino,
            file_stat.st_size,
            file_stat.st_mtime_ns,
            digest.hexdigest(),
        )
    finally:
        os.close(descriptor)


def _git_head(git: str, repository_root: Path) -> str:
    try:
        result = subprocess.run(
            [git, "-C", str(repository_root), "rev-parse", "HEAD"],
            check=True,
            capture_output=True,
            text=True,
            timeout=10,
            env={"PATH": "/usr/bin:/bin", "LANG": "C", "LC_ALL": "C"},
        )
    except (OSError, subprocess.SubprocessError) as error:
        raise EvidenceCliError(f"could not resolve repository commit: {error}") from error
    return result.stdout.strip()


class TrustedSnapshotError(RuntimeError):
    """The source has not been authorized as an installed producer."""


class EvidenceCliError(RuntimeError):
    """Invalid evidence CLI request."""


def _snapshot_digest(snapshot: Path) -> str:
    if snapshot.is_file():
        return sha256_bytes(snapshot.read_bytes())
    if not snapshot.is_dir():
        raise TrustedSnapshotError(f"trusted snapshot is not a file or directory: {snapshot}")
    records: list[bytes] = []
    for path in sorted(snapshot.rglob("*")):
        relative = path.relative_to(snapshot).as_posix()
        if relative == ".pi-sbx-trusted-snapshot":
            continue
        if path.is_symlink():
            raise TrustedSnapshotError(f"trusted snapshot contains symlink: {relative}")
        if path.is_file():
            records.append(
                relative.encode("utf-8")
                + b"\0"
                + _sha256_bytes(path.read_bytes()).encode("ascii")
                + b"\n"
            )
    return _sha256_bytes(b"".join(records))


def verify_trusted_snapshot(snapshot: str | os.PathLike[str], expected_digest: str) -> Path:
    """Verify an explicit marker and immutable-content digest.

    A path inside this source checkout is rejected even if a marker is added to
    it.  This prevents a repository write from becoming a trusted producer.
    """
    path = Path(snapshot)
    if not path.is_absolute():
        raise TrustedSnapshotError("--trusted-snapshot must be an absolute path")
    if _DIGEST.fullmatch(expected_digest) is None:
        raise TrustedSnapshotError("trusted snapshot digest must be lowercase SHA-256")
    if path.is_symlink():
        raise TrustedSnapshotError("trusted snapshot path itself must not be a symlink")
    try:
        resolved = path.resolve(strict=True)
    except OSError as error:
        raise TrustedSnapshotError(f"cannot resolve trusted snapshot: {error}") from error
    source = REPOSITORY_ROOT.resolve()
    producer = Path(__file__).resolve()
    source_is_checkout = (source / ".git").exists()
    if source_is_checkout:
        raise TrustedSnapshotError("source checkout cannot be a trusted production snapshot")
    if not resolved.is_dir() or resolved != source or resolved not in producer.parents:
        raise TrustedSnapshotError(
            "trusted snapshot must equal the executing producer's repository root"
        )
    if resolved.is_dir():
        if resolved.stat().st_uid != os.geteuid() or resolved.stat().st_mode & 0o022:
            raise TrustedSnapshotError("trusted snapshot directory must be caller-owned and private")
        for candidate in resolved.rglob("*"):
            if candidate.is_symlink():
                raise TrustedSnapshotError(f"trusted snapshot contains symlink: {candidate}")
            if not candidate.is_dir() and not candidate.is_file():
                raise TrustedSnapshotError(f"trusted snapshot contains a non-file entry: {candidate}")
            candidate_stat = candidate.stat()
            if candidate_stat.st_uid != os.geteuid() or candidate_stat.st_mode & 0o022:
                raise TrustedSnapshotError(
                    f"trusted snapshot entries must be caller-owned and not group/other writable: {candidate}"
                )
    marker = resolved / ".pi-sbx-trusted-snapshot"
    try:
        marker_contents = marker.read_text(encoding="utf-8")
    except OSError as error:
        raise TrustedSnapshotError(f"trusted snapshot marker is missing: {marker}") from error
    if marker_contents != _MARKER_CONTENT:
        raise TrustedSnapshotError("trusted snapshot marker has an invalid contract")
    observed = _snapshot_digest(resolved)
    if observed != expected_digest:
        raise TrustedSnapshotError(
            f"trusted snapshot digest mismatch: expected {expected_digest}, observed {observed}"
        )
    return resolved


def _load_trusted_modules() -> None:
    """Import producer code only after the executing snapshot is verified."""
    if str(REPOSITORY_ROOT) not in sys.path:
        sys.path.insert(0, str(REPOSITORY_ROOT))
    global EvidenceStore, EvidenceValidationError, create_summary
    global sha256_bytes, sha256_json, utc_timestamp, redact
    global isolated_environment, HostRequirementError, NonoAdapter
    global run_bounded_command

    from tests.nono.harness.evidence import (
        EvidenceStore,
        EvidenceValidationError,
        create_summary,
        sha256_bytes,
        sha256_json,
        utc_timestamp,
    )
    from tests.nono.harness.fixtures import isolated_environment
    from tests.nono.harness.nono import (
        HostRequirementError,
        NonoAdapter,
        run_bounded_command,
    )
    from tests.nono.harness.redaction import redact


def _observation(
    observation_id: str,
    capability: str,
    expected: str,
    actual: str,
    status: str,
) -> dict[str, str]:
    body = {
        "id": observation_id,
        "capability": capability,
        "expected": expected,
        "actual": actual,
        "status": status,
    }
    body["hash"] = sha256_json(body)
    return body


def _run_fixed_harness(arguments: list[str]) -> dict[str, Any]:
    if arguments:
        command = [
            sys.executable,
            str(REPOSITORY_ROOT / "scripts" / "test_nono.py"),
            *arguments,
        ]
    else:
        command = [
            sys.executable,
            "-m",
            "unittest",
            "discover",
            "-s",
            "tests/nono/harness",
            "-p",
            "test_*.py",
            "-t",
            ".",
        ]
    with isolated_environment(prefix="nono-evidence-runner-") as fixture:
        result = run_bounded_command(
            command,
            cwd=REPOSITORY_ROOT,
            environment=fixture.env,
            timeout=180,
            max_output=65536,
        )
    return {
        "returncode": result.returncode,
        "stdout": result.stdout,
        "stderr": result.stderr,
        "timedOut": result.timed_out,
    }


def _hermetic_core(_: argparse.Namespace) -> dict[str, Any]:
    execution = _run_fixed_harness([])
    passed = execution["returncode"] == 0 and not execution["timedOut"]
    status = "passed" if passed else "failed"
    actual = "allowed" if passed else "unknown"
    return {
        "status": status,
        "observations": [
            _observation("hermetic-suite", "harness", "allowed", actual, status)
        ],
        "hashes": {},
        "stdout": execution["stdout"],
        "stderr": execution["stderr"],
    }


def _negative_matrix(_: argparse.Namespace) -> dict[str, Any]:
    capabilities = (
        "tmp", "runtime-parent", "sibling-worktree", "direct-ssh", "cross-context-key",
        "unrelated-pi-state", "trusted-instructions-write", "docker", "browser",
        "dictation", "update", "production-credentials", "nonselected-provider",
    )
    observations = [
        _observation(f"negative-{name}", name, "denied", "not-run", "blocked")
        for name in capabilities
    ]
    return {"status": "blocked", "observations": observations, "hashes": {}}


def _positive_contract(_: argparse.Namespace) -> dict[str, Any]:
    capabilities = ("git", "gh-cli", "package-cache", "provider-port", "unix-socket")
    observations = [
        _observation(f"positive-{name}", name, "allowed", "not-run", "blocked")
        for name in capabilities
    ]
    return {"status": "blocked", "observations": observations, "hashes": {}}


def _broad_fixture(args: argparse.Namespace) -> dict[str, Any]:
    if args.suite != "host":
        return {
            "status": "blocked",
            "observations": [
                _observation(
                    "broad-workdir",
                    "workdir",
                    "denied",
                    "not-run",
                    "blocked",
                )
            ],
            "hashes": {},
        }
    execution = _run_fixed_harness(["--host", "--nono", args.nono])
    passed = execution["returncode"] == 0 and not execution["timedOut"]
    status = "passed" if passed else "failed"
    return {
        "status": status,
        "observations": [
            _observation(
                "minimal-cross-session",
                "cross-session-filesystem",
                "denied",
                "denied" if passed else "unknown",
                status,
            ),
            _observation(
                "broad-workdir",
                "workdir",
                "allowed",
                "allowed" if passed else "unknown",
                status,
            ),
        ],
        "hashes": {},
        "stdout": execution["stdout"],
        "stderr": execution["stderr"],
    }


def _live_result(_: argparse.Namespace) -> dict[str, Any]:
    # Credentialed live work is intentionally not attempted by source code.  A
    # host procedure must replace this fixed blocked observation in a future
    # installed snapshot after a human has approved it.
    return {
        "status": "blocked",
        "observations": [
            _observation("live-manual-check", "live-credentials", "allowed", "not-run", "blocked")
        ],
        "hashes": {},
    }


BUILTIN_TESTS: dict[str, Callable[[argparse.Namespace], dict[str, Any]]] = {
    "hermetic-core": _hermetic_core,
    "negative-capability-matrix": _negative_matrix,
    "positive-contract-matrix": _positive_contract,
    "broad-fixture": _broad_fixture,
    "git-live": _live_result,
    "github-cli-live": _live_result,
}


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--trusted-snapshot", required=True, help="installed snapshot directory")
    parser.add_argument(
        "--trusted-snapshot-digest",
        "--snapshot-digest",
        dest="trusted_snapshot_digest",
        required=True,
        help="SHA-256 digest of the installed snapshot manifest",
    )
    parser.add_argument(
        "--state-root", required=True, help="injected XDG_STATE_HOME-compatible evidence root"
    )
    parser.add_argument(
        "--summary-root", required=True, help="injected root for redacted summaries"
    )
    subparsers = parser.add_subparsers(dest="action", required=True)
    run = subparsers.add_parser("run", help="run one fixed registry test")
    run.add_argument("test_id")
    run.add_argument("--suite", choices=("hermetic", "host", "live"), default="hermetic")
    run.add_argument("--approver")
    run.add_argument("--approve-live", action="store_true")
    run.add_argument("--repository-root", required=True)
    run.add_argument("--repository-commit", required=True)
    run.add_argument("--policy-file", required=True)
    run.add_argument("--policy-digest", required=True)
    run.add_argument("--nono", required=True, help="absolute Nono 0.74 executable")
    run.add_argument("--canary", action="append", default=[])
    summarize = subparsers.add_parser("summarize", help="summarize retained raw evidence")
    summarize.add_argument("run_id")
    summarize.add_argument("--repository-id", default="fixture-repository")
    summarize.add_argument("--canary", action="append", default=[])
    summarize.add_argument(
        "--credential-name-pattern",
        action="append",
        default=[],
        help="trusted credential-name regex that must be redacted from the summary",
    )
    return parser


def _run(args: argparse.Namespace, snapshot_digest: str) -> int:
    if args.test_id not in BUILTIN_TESTS:
        raise EvidenceCliError(f"unknown test ID {args.test_id!r}; arbitrary commands are forbidden")
    selected_nono = Path(args.nono)
    if not selected_nono.is_absolute():
        raise EvidenceCliError("Nono executable must be an absolute path")
    nono_identity = _regular_file_identity(selected_nono, max_size=256 * 1024 * 1024)
    try:
        with isolated_environment(prefix="nono-evidence-host-") as fixture:
            session = fixture.session("one")
            adapter = NonoAdapter(
                selected_nono,
                cwd=session.repo,
                environment=session.environment(fixture.env),
            )
            args.nono_version = adapter.require_host_0_74()
            if args.suite == "host":
                profile = (
                    REPOSITORY_ROOT
                    / "tests"
                    / "nono"
                    / "fixtures"
                    / "core"
                    / "minimal.json"
                )
                if not adapter.validate(profile).ok or not adapter.run(
                    profile,
                    ["/usr/bin/true"],
                ).ok:
                    raise EvidenceCliError(
                        "host suite fixture validation or execution failed"
                    )
    except (HostRequirementError, OSError, ValueError) as error:
        raise EvidenceCliError(f"Nono 0.74 verification failed: {error}") from error
    if args.suite == "live":
        if args.test_id not in _LIVE_TESTS:
            raise EvidenceCliError("live mode is available only for the fixed live test registry")
        if not args.approve_live or not isinstance(args.approver, str) or not args.approver.strip():
            raise EvidenceCliError("live mode requires explicit approval via --approve-live and a named --approver")
    elif args.test_id in _LIVE_TESTS:
        raise EvidenceCliError("credentialed tests require explicit --suite live approval")
    repository_root = Path(args.repository_root)
    if not repository_root.is_absolute() or repository_root.is_symlink():
        raise EvidenceCliError("repository root must be an absolute non-symlink path")
    try:
        repository_root = repository_root.resolve(strict=True)
    except OSError as error:
        raise EvidenceCliError(f"repository root is unavailable: {error}") from error
    if not (repository_root / ".git").exists():
        raise EvidenceCliError("repository root is not a Git checkout")
    git = shutil.which("git", path="/usr/bin:/bin")
    if git is None:
        raise EvidenceCliError("Git is unavailable for repository identity verification")
    observed_commit = _git_head(git, repository_root)
    if observed_commit != args.repository_commit:
        raise EvidenceCliError("repository commit does not match the tested checkout")

    policy_file = Path(args.policy_file)
    if not policy_file.is_absolute() or policy_file.is_symlink():
        raise EvidenceCliError("policy file must be an absolute non-symlink path")
    try:
        policy_file = policy_file.resolve(strict=True)
    except OSError as error:
        raise EvidenceCliError(f"policy file is unavailable: {error}") from error
    policy_identity = _regular_file_identity(policy_file, max_size=1024 * 1024)
    policy_digest = policy_identity[-1]
    if policy_digest != args.policy_digest:
        raise EvidenceCliError("policy digest does not match the tested policy file")
    run_id = "run-" + uuid.uuid4().hex
    started = utc_timestamp()
    result = BUILTIN_TESTS[args.test_id](args)
    if _regular_file_identity(
        selected_nono,
        max_size=256 * 1024 * 1024,
    ) != nono_identity:
        raise EvidenceCliError("Nono executable changed during evidence execution")
    if _git_head(git, repository_root) != observed_commit:
        raise EvidenceCliError("repository HEAD changed during evidence execution")
    if _regular_file_identity(policy_file, max_size=1024 * 1024) != policy_identity:
        raise EvidenceCliError("policy file changed during evidence execution")
    finished = utc_timestamp()
    payload = {
        "schemaVersion": 1,
        "runId": run_id,
        "testId": args.test_id,
        "suite": args.suite,
        "metadata": {
            "repositoryCommit": args.repository_commit,
            "policySnapshotDigest": policy_digest,
            "installedSnapshotDigest": snapshot_digest,
            "nonoVersion": args.nono_version,
            "approver": args.approver if args.suite == "live" else None,
            "startedAt": started,
            "finishedAt": finished,
        },
        "result": result,
    }
    store = EvidenceStore(args.state_root, args.summary_root)
    store.purge_expired()
    reference = store.write_raw_evidence(run_id, payload)
    print(
        json.dumps(
            {
                "runId": run_id,
                "status": result["status"],
                "rawEvidenceSha256": reference.sha256,
                "rawEvidence": str(reference.path),
            },
            sort_keys=True,
        )
    )
    return {"passed": 0, "failed": 1, "blocked": 2, "not-run": 2}.get(
        str(result["status"]), 2
    )


def _summarize(args: argparse.Namespace, snapshot_digest: str) -> int:
    store = EvidenceStore(args.state_root, args.summary_root)
    payload, raw_metadata = store.read_raw_evidence(args.run_id)
    if not isinstance(payload, dict) or not isinstance(payload.get("metadata"), dict):
        raise EvidenceCliError("retained evidence has no trusted metadata")
    metadata = payload["metadata"]
    if payload.get("runId") != args.run_id:
        raise EvidenceCliError("retained evidence run ID mismatch")
    if metadata.get("installedSnapshotDigest") != snapshot_digest:
        raise EvidenceCliError("evidence was produced by a different installed snapshot")
    result = payload.get("result")
    if not isinstance(result, dict):
        raise EvidenceCliError("retained evidence result is not an object")
    cleaned = redact(
        payload,
        canaries=args.canary,
        credential_name_patterns=args.credential_name_pattern,
    )
    cleaned_result = cleaned.get("result")
    if not isinstance(cleaned_result, dict):
        raise EvidenceCliError("redacted result is not an object")
    result_hashes: dict[str, str] = {}
    raw_hashes = cleaned_result.get("hashes", {})
    if not isinstance(raw_hashes, dict):
        raise EvidenceCliError("result hashes are not an object")
    for key, digest in raw_hashes.items():
        if not isinstance(key, str) or key in _RAW_OUTPUT_FIELDS or _DIGEST.fullmatch(str(digest)) is None:
            raise EvidenceCliError("result contains an invalid or raw hash field")
        result_hashes[key] = str(digest)
    # Output fields are represented only by their digest in a summary.
    for key, value in result.items():
        if key in _RAW_OUTPUT_FIELDS:
            result_hashes[key + "Hash"] = sha256_json(value)
    observations = cleaned_result.get("observations", [])
    if not isinstance(observations, list):
        raise EvidenceCliError("observations are not an array")
    summary = create_summary(
        run_id=args.run_id,
        test_id=str(payload.get("testId", "")),
        suite=str(payload.get("suite", "")),
        repository_commit=str(metadata.get("repositoryCommit", "")),
        policy_snapshot_digest=str(metadata.get("policySnapshotDigest", "")),
        installed_snapshot_digest=snapshot_digest,
        nono_version=str(metadata.get("nonoVersion", "")),
        status=str(cleaned_result.get("status", "")),
        approver=metadata.get("approver"),
        started_at=str(metadata.get("startedAt", "")),
        finished_at=str(metadata.get("finishedAt", "")),
        raw_evidence_sha256=str(raw_metadata.get("sha256", "")),
        raw_evidence_retained_until=str(raw_metadata.get("retainedUntil", "")),
        result_hashes=result_hashes,
        observations=observations,
        canaries=args.canary,
        credential_name_patterns=args.credential_name_pattern,
    )
    policy_digest = str(metadata["policySnapshotDigest"])
    path = store.write_summary(
        summary,
        repository_id=args.repository_id,
        policy_digest=policy_digest,
    )
    print(json.dumps({"runId": args.run_id, "summary": str(path), "status": summary["status"]}, sort_keys=True))
    return {"passed": 0, "failed": 1, "blocked": 2, "not-run": 2}.get(
        str(summary["status"]), 2
    )


def main(argv: list[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    snapshot = verify_trusted_snapshot(args.trusted_snapshot, args.trusted_snapshot_digest)
    del snapshot
    sys.dont_write_bytecode = True
    _load_trusted_modules()
    if args.action == "run":
        return _run(args, args.trusted_snapshot_digest)
    if args.action == "summarize":
        return _summarize(args, args.trusted_snapshot_digest)
    raise EvidenceCliError("unknown action")


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (EvidenceCliError, TrustedSnapshotError, ValueError) as error:
        print(f"nono-evidence: {error}", file=sys.stderr)
        raise SystemExit(2)