profile.py

"""
Profile runtime secret oracle adapter.

Provides layered file-operation oracle surfaces:
- caller-supplied probe execution for one profile
- IR-driven probe generation for one profile
- source vs roundtrip comparison for two profiles
"""

from __future__ import annotations

import hashlib
import shutil
import tempfile
from pathlib import Path
from typing import Any

from runtime.oracle.client import ensure_runner_built, run_sandboxed_runner
from runtime.oracle.probe_types import SecretProbe

SECRET_MARKER = "PAWL_PROFILE_ORACLE_SECRET_7f3a9c"
WRITE_MARKER = "PAWL_WRITE_VERIFY_8e2b4d"
WORKSPACE_VERSION = "v1"

DEFAULT_PARAM_BINDINGS = {
    "HOME": "/tmp/pawl_test_home",
    "_HOME": "/tmp/pawl_test_home",
    "TMPDIR": "/tmp",
    "DARWIN_CACHE_DIR": "/tmp/pawl_darwin_cache",
    "DARWIN_USER_DIR": "/tmp/pawl_darwin_user",
    "DARWIN_USER_CACHE_DIR": "/tmp/pawl_darwin_user_cache",
    "DARWIN_USER_TEMP_DIR": "/tmp/pawl_darwin_user_temp",
    "application_bundle": "/tmp/pawl_app_bundle",
    "application_container": "/tmp/pawl_app_container",
    "application_darwin_temp_dir": "/tmp/pawl_darwin_temp",
    "application_darwin_user_dir": "/tmp/pawl_darwin_user",
    "application_group": "/tmp/pawl_app_group",
    "application_library": "/tmp/pawl_app_library",
}

FILE_READ_OPS = {
    "file-read-data",
    "file-read-metadata",
    "file-read-xattr",
}

FILE_WRITE_OPS = {
    "file-write-data",
    "file-write-create",
    "file-write-flags",
    "file-write-mode",
    "file-write-owner",
    "file-write-setugid",
    "file-write-times",
    "file-write-xattr",
    "file-write-unlink",
}

FILE_LINK_OPS = {
    "file-link",
}

WILDCARD_OPS = {
    "file-read*": FILE_READ_OPS,
    "file-write*": FILE_WRITE_OPS,
    "file*": FILE_READ_OPS | FILE_WRITE_OPS | FILE_LINK_OPS,
}


def _probe_id(op_name: str, idx: int) -> str:
    h = hashlib.sha256(f"{op_name}:{idx}".encode()).hexdigest()[:8]
    return f"probe_{op_name.replace('-', '_').replace('*', 'star')}_{h}"


def _op_to_probe_kind_arg(op_name: str) -> tuple[str, str] | None:
    if op_name in FILE_READ_OPS:
        return "file_read_secret", SECRET_MARKER
    if op_name in FILE_WRITE_OPS:
        return "file_write_verify", WRITE_MARKER
    if op_name in FILE_LINK_OPS:
        return "file_link_stat", ""
    return None


def generate_file_probes(
    ir: dict[str, Any],
    workspace: Path,
) -> list[SecretProbe]:
    """Generate file operation probes from Profile IR explicit ops."""
    probes: list[SecretProbe] = []
    seen_ops: set[str] = set()

    slots = ir.get("op_table", {}).get("slots", [])
    for slot in slots:
        if slot.get("assignment_class") != "explicit":
            continue
        op_name = slot.get("op_name", "")

        ops_to_probe: set[str] = set()
        if op_name in WILDCARD_OPS:
            ops_to_probe = WILDCARD_OPS[op_name]
        elif op_name in FILE_READ_OPS or op_name in FILE_WRITE_OPS or op_name in FILE_LINK_OPS:
            ops_to_probe = {op_name}

        for op in sorted(ops_to_probe):
            if op in seen_ops:
                continue
            seen_ops.add(op)
            kind_arg = _op_to_probe_kind_arg(op)
            if not kind_arg:
                continue
            kind, arg = kind_arg
            probe_id = _probe_id(op, len(probes))
            filename = f"{probe_id}.txt"
            probes.append(
                SecretProbe(
                    probe_id=probe_id,
                    kind=kind,
                    path=workspace / filename,
                    arg=arg,
                )
            )
    return probes


def setup_probe_workspace(workspace: Path, probes: list[SecretProbe]) -> None:
    """Create probe files for the secret runner."""
    workspace.mkdir(parents=True, exist_ok=True)
    for probe in probes:
        probe_path = Path(probe.path)
        if probe.kind == "file_read_secret":
            probe_path.parent.mkdir(parents=True, exist_ok=True)
            probe_path.write_text(f"{SECRET_MARKER}\nprobe_id={probe.probe_id}\n")
        elif probe.kind == "file_write_verify":
            probe_path.parent.mkdir(parents=True, exist_ok=True)
            probe_path.touch()
        elif probe.kind == "file_link_stat":
            probe_path.parent.mkdir(parents=True, exist_ok=True)
            probe_path.write_text(f"link_source={probe.probe_id}\n")


def _single_profile_skip_result(reason: str) -> dict[str, Any]:
    return {
        "probes": 0,
        "skipped": reason,
        "verdicts": {},
        "workspace_version": WORKSPACE_VERSION,
    }


def compare_binary_verdicts(
    source_results: dict[str, dict[str, Any]],
    roundtrip_results: dict[str, dict[str, Any]],
) -> tuple[bool | None, list[dict[str, Any]]]:
    """
    Compare verdict dicts from source and roundtrip runs.

    Returns (eq, mismatches), where:
    - eq=True when all verdicts match
    - eq=False when verdicts diverge
    - eq=None when baseline is unavailable (source apply error)
    """
    mismatches: list[dict[str, Any]] = []

    if "APPLY_ERROR" in source_results:
        mismatches.append(
            {
                "probe_id": "APPLY_ERROR",
                "source": source_results.get("APPLY_ERROR"),
                "roundtrip": roundtrip_results.get("APPLY_ERROR"),
            }
        )
        return None, mismatches

    if "APPLY_ERROR" in roundtrip_results:
        mismatches.append(
            {
                "probe_id": "APPLY_ERROR",
                "source": None,
                "roundtrip": roundtrip_results["APPLY_ERROR"],
            }
        )
        return False, mismatches

    if "RUN_ERROR" in source_results or "RUN_ERROR" in roundtrip_results:
        mismatches.append(
            {
                "probe_id": "RUN_ERROR",
                "source": source_results.get("RUN_ERROR"),
                "roundtrip": roundtrip_results.get("RUN_ERROR"),
            }
        )
        return False, mismatches

    all_probes = set(source_results.keys()) | set(roundtrip_results.keys())
    for probe_id in sorted(all_probes):
        src = source_results.get(probe_id, {})
        rt = roundtrip_results.get(probe_id, {})
        src_verdict = src.get("verdict", "missing")
        rt_verdict = rt.get("verdict", "missing")
        if src_verdict != rt_verdict:
            mismatches.append(
                {
                    "probe_id": probe_id,
                    "source_verdict": src_verdict,
                    "roundtrip_verdict": rt_verdict,
                    "source_errno": src.get("errno"),
                    "roundtrip_errno": rt.get("errno"),
                }
            )

    return len(mismatches) == 0, mismatches


def run_probe_set(
    profile: Path,
    probes: list[SecretProbe],
    workspace: Path,
    param_bindings: dict[str, str] | None = None,
) -> dict[str, Any]:
    """Execute a caller-supplied probe set against a single profile."""
    ensure_runner_built()
    setup_probe_workspace(workspace, probes)
    verdicts = run_sandboxed_runner(
        profile,
        probes,
        workspace,
        param_bindings=param_bindings,
    )
    return {
        "probes": len(probes),
        "skipped": None,
        "verdicts": verdicts,
        "workspace_version": WORKSPACE_VERSION,
    }


def run_single_oracle(
    profile: Path,
    ir: dict[str, Any],
    workspace: Path | None = None,
    param_bindings: dict[str, str] | None = None,
) -> dict[str, Any]:
    """Generate file-op probes from IR and execute them against one profile."""
    cleanup_workspace = False
    if workspace is None:
        workspace = Path(tempfile.mkdtemp(prefix="pawl_oracle_single_"))
        cleanup_workspace = True

    try:
        probes = generate_file_probes(ir, workspace)
        if not probes:
            return _single_profile_skip_result("no_file_ops")
        return run_probe_set(
            profile,
            probes,
            workspace,
            param_bindings=param_bindings,
        )
    finally:
        if cleanup_workspace and workspace.exists():
            shutil.rmtree(workspace, ignore_errors=True)


def run_binary_oracle(
    source_profile: Path,
    roundtrip_profile: Path,
    ir: dict[str, Any],
    workspace: Path | None = None,
    param_bindings: dict[str, str] | None = None,
) -> dict[str, Any]:
    """Run full source vs roundtrip profile runtime comparison.

    When *param_bindings* is provided, the runner passes them to
    ``sandbox_init_with_parameters`` so that ``(param ...)`` references in
    the profile text and its ``(import ...)`` targets resolve correctly.
    """
    cleanup_workspace = False
    if workspace is None:
        workspace = Path(tempfile.mkdtemp(prefix="pawl_oracle_"))
        cleanup_workspace = True

    try:
        source_result = run_single_oracle(
            source_profile,
            ir,
            workspace=workspace / "source",
            param_bindings=param_bindings,
        )
        roundtrip_result = run_single_oracle(
            roundtrip_profile,
            ir,
            workspace=workspace / "roundtrip",
            param_bindings=param_bindings,
        )

        if (
            source_result.get("skipped") == "no_file_ops"
            and roundtrip_result.get("skipped") == "no_file_ops"
        ):
            return {
                "eq": True,
                "skipped": "no_file_ops",
                "probes": 0,
                "source_verdicts": {},
                "roundtrip_verdicts": {},
                "mismatches": None,
                "workspace_version": WORKSPACE_VERSION,
            }

        source_results = source_result.get("verdicts", {})
        roundtrip_results = roundtrip_result.get("verdicts", {})
        eq, mismatches = compare_binary_verdicts(source_results, roundtrip_results)
        return {
            "eq": eq,
            "probes": int(source_result.get("probes", 0)),
            "skipped": source_result.get("skipped"),
            "source_verdicts": source_results,
            "roundtrip_verdicts": roundtrip_results,
            "mismatches": mismatches if mismatches else None,
            "workspace_version": str(
                source_result.get("workspace_version")
                or roundtrip_result.get("workspace_version")
                or WORKSPACE_VERSION
            ),
        }
    finally:
        if cleanup_workspace and workspace.exists():
            shutil.rmtree(workspace, ignore_errors=True)