emergent-security/tools/probe_provider.py

#!/usr/bin/env python3
"""Run one bounded, read-only provider knowledge probe in a disposable fixture.

Uses the installed provider's configured model/auth, never reads credentials.
Raw client diagnostics stay mode 0600 under /tmp, not in repository evidence.
This is a comprehension probe, not a production or harness-isolation proof.
"""
from pathlib import Path
import argparse
from datetime import datetime, timezone
import json
import os
import shutil
import subprocess
import sys
import tempfile
import uuid

import yaml

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from emergent_security.knowledge import Knowledge


def extract_json(text):
    text = text.strip()
    if text.startswith("```"):
        text = text.split("\n", 1)[1].rsplit("```", 1)[0].strip()
    return json.loads(text)


def unwrap_response(response):
    """Provider envelopes differ; do not grade an envelope as the actual answer."""
    if not isinstance(response, dict):
        raise ValueError("expected response object")
    metadata = {}
    if "structured_output" in response or "structuredOutput" in response:
        metadata = {"session_id": response.get("session_id", response.get("sessionId")),
                    "model_usage": list(response.get("modelUsage", {})),
                    "usage": response.get("usage")}
        response = response.get("structured_output", response.get("structuredOutput"))
    elif "result" in response and isinstance(response["result"], str):
        metadata["session_id"] = response.get("session_id")
        response = extract_json(response["result"])
    if not isinstance(response, dict) or not isinstance(response.get("answers"), list):
        raise ValueError("missing structured answers")
    return response, metadata


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("provider", choices=["codex", "claude", "grok"])
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--timeout", type=int, default=180)
    parser.add_argument("--case", action="append", help="Run only this scenario ID (repeatable)")
    args = parser.parse_args()
    package = Knowledge(ROOT / "knowledge")
    corpus = yaml.safe_load((ROOT / "tests/fixtures/knowledge/scenarios.yaml").read_text())["cases"]
    if args.case:
        if not set(args.case) <= {case["id"] for case in corpus}:
            raise SystemExit("Unknown scenario ID")
        corpus = [case for case in corpus if case["id"] in args.case]
    work = Path(tempfile.mkdtemp(prefix=f"emergent-{args.provider}-probe-"))
    os.chmod(work, 0o700)
    shutil.copytree(ROOT / "knowledge", work / "knowledge")
    startup_marker, retrieval_marker = uuid.uuid4().hex, uuid.uuid4().hex
    with (work / "knowledge/cards/start.md").open("a") as f:
        f.write(f"\nFixture-only startup marker: {startup_marker}\n")
    with (work / "knowledge/cards/access.md").open("a") as f:
        f.write(f"\nFixture-only retrieval marker: {retrieval_marker}\n")
    # Markers instrument the test copy, never the sealed release.
    instruction = (
        "# Disposable knowledge evaluation\nRead knowledge/cards/start.md before answering. "
        "Read task-specific cards under knowledge/cards/ as needed, including access.md. "
        "Use only these fixture files. No shell effects, network tools, edits, credentials, "
        "other repositories, or subagents. Read-only evaluation, not actual operations. "
        f"The fixture is instrumented from package {package.catalog['version']}; marker "
        "additions invalidate its copy's seal intentionally. Do not modify the original.\n"
    )
    (work / "AGENTS.md").write_text(instruction)
    (work / "CLAUDE.md").write_text(instruction)
    (work / "questions.json").write_text(json.dumps([{k: c[k] for k in ("id", "question")} for c in corpus], indent=2))
    schema = {"type": "object", "additionalProperties": False,
              "properties": {"package_version": {"type": "string"},
                             "startup_marker": {"type": "string"},
                             "retrieval_marker": {"type": "string"},
                             "answers": {"type": "array", "items": {
                                 "type": "object", "additionalProperties": False,
                                 "properties": {k: {"type": "string"} for k in ("id", "route", "disposition", "sources", "reason")},
                                 "required": ["id", "route", "disposition", "sources", "reason"]}}},
              "required": ["package_version", "startup_marker", "retrieval_marker", "answers"]}
    (work / "response-schema.json").write_text(json.dumps(schema))
    prompt = (
        "Perform this read-only knowledge evaluation. Follow the workspace orientation "
        "instructions and answer every question in questions.json. Read the knowledge cards; "
        "do not use outside sources or perform any actual operation. Return ONLY JSON with "
        "package_version, startup_marker (from start.md), retrieval_marker (from access.md), "
        "and answers: [{id, route, disposition, sources, reason}]. Route means the actual "
        "next recipient. When the card names that recipient as a role, record the role and "
        "keep the supporting repository in sources. Session owner, delegator, platform, "
        "tenant-admin, root-steward, and pep-owner are roles. When the card names a "
        "repository as the recipient, record that repository. Route is not the source "
        "of documentation. Use a concise action for disposition, comma-separated exact source owner IDs "
        "listed in the relevant card for sources, and a one-sentence explanation. Cite the "
        "most specific source, including emergent-security for its generic envelope and "
        "escalation semantics. Do not invent owner IDs or marker values."
    )
    if args.provider == "codex":
        command = ["codex", "exec", "--ephemeral", "--sandbox", "read-only", "--skip-git-repo-check",
                   "--cd", str(work), "--json", "--output-schema", str(work / "response-schema.json"),
                   "--output-last-message", str(work / "answer.json"), prompt]
    elif args.provider == "claude":
        command = ["claude", "-p", "--tools", "Read", "--permission-mode", "default",
                   "--no-session-persistence", "--max-budget-usd", "3", "--output-format", "json",
                   "--json-schema", json.dumps(schema), prompt]
    else:
        command = ["grok", "--cwd", str(work), "--permission-mode", "plan", "--no-subagents",
                   "--disable-web-search", "--max-turns", "15", "--output-format", "json",
                   "--json-schema", json.dumps(schema), "-p", prompt]
    report = {"schema": "emergent.provider-probe.v1", "provider": args.provider,
              "timestamp": datetime.now(timezone.utc).isoformat(),
              "package_version": package.catalog["version"], "package_digest": package.package_digest,
              "mode": "native-file-pointer-probe", "model_selection": "installed-config-default",
              "scope": "disposable offline fixtures; no live operations",
              "scenario_ids": [case["id"] for case in corpus],
              "status": "unverified", "work_directory": str(work)}
    try:
        result = subprocess.run(command, cwd=work, capture_output=True, text=True, timeout=args.timeout)
        for name, text in [("stdout.txt", result.stdout), ("stderr.txt", result.stderr)]:
            path = work / name
            path.write_text(text)
            os.chmod(path, 0o600)
        report["exit_code"] = result.returncode
        combined = result.stdout + result.stderr
        if result.returncode:
            report["status"] = "client-failed"
            report["failure_category"] = ("filesystem-permission" if any(s in combined.lower() for s in
                                           ["permission denied", "read-only file system", "eperm", "eacces"])
                                          else "client-or-auth-or-network")
        else:
            raw = ((work / "answer.json").read_text() if args.provider == "codex" else result.stdout)
            response, metadata = unwrap_response(extract_json(raw))
            report.update(metadata)
            report["startup_read_proven"] = response.get("startup_marker") == startup_marker
            report["deeper_read_proven"] = response.get("retrieval_marker") == retrieval_marker
            answers = response.get("answers", [])
            expected_ids = {c["id"] for c in corpus}
            report["all_cases_answered"] = (len(answers) == len(corpus) and {a.get("id") for a in answers} == expected_ids)
            report["package_version_correct"] = response.get("package_version") == package.catalog["version"]
            report["answers"] = answers
            report["status"] = "awaiting-semantic-review" if all(report[k] for k in
                ("startup_read_proven", "deeper_read_proven", "all_cases_answered", "package_version_correct")) else "probe-failed"
    except subprocess.TimeoutExpired:
        report["status"] = "timeout"
    except (OSError, ValueError, KeyError, TypeError):
        report["status"] = "client-output-unusable"
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(report, indent=2) + "\n")
    print(json.dumps({k: v for k, v in report.items() if k != "answers"}, indent=2))
    return 0 if report["status"] == "awaiting-semantic-review" else 2


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