#!/usr/bin/env python3
"""Offline identity-consistency lab for a README-directed install review.

Requires Python 3.10+ and Git. Uses only the Python standard library and local
Git commands. It does not fetch URLs, run candidate files, install software,
scan arbitrary repositories, or determine whether reviewed bytes are benign.

Run from a repository root:
  python3 static/downloads/readme-zip-review-lab.py \
    --work-dir .agent-cache/readme-zip-review-lab

All temporary files and the results.json output stay under --work-dir. The
default is .agent-cache/readme-zip-review-lab under the current directory.
The lab creates its own harmless repositories. The expected receipt is held
outside the candidate repository and installation directory. A real system
would also need to protect its receipt and verifier from the candidate.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path
import platform
import shutil
import subprocess
import tempfile
import zipfile


SKILL = """---
name: offline-review-example
description: A harmless text-only fixture for an identity-consistency lab.
---
Read the example notes when explaining this lab. No commands are required.
"""
README = """# Offline review example

This is synthetic review material, not an installer.
Repository artifact: [example notes](archives/notes.zip)
External artifact label: https://artifacts.example.invalid/notes.zip
The external label is never requested. Its bytes are modeled by a local file.
"""
ORIGINAL_NOTES = b"Harmless example notes, revision one.\n"
CHANGED_NOTES = b"Harmless example notes, revision two.\n"
FIXED_DATE = "2026-11-05T00:00:00+00:00"
CASES = (
    "unchanged",
    "readme_destination_changed",
    "committed_zip_changed",
    "external_url_bytes_changed",
    "copied_inventory_changed",
    "benign_zip_repack",
)
LIMITS = [
    "Identity consistency is not a malware verdict. Equal hashes do not make bytes benign.",
    "Every candidate file contains harmless synthetic text; no candidate is executed.",
    "External URL cases are hypothetical extensions, modeled with local files and no network.",
    "A Git commit binds a committed archive in that snapshot, not bytes served by an external URL.",
    "The trusted receipt is generated from the lab baseline, not from a real security review.",
    "Inventories cover regular file paths and content hashes, not permissions, symlinks or runtime behavior.",
    "The installed-file check follows a safe copy of known text; it does not authorize running an unknown installer first.",
    "The lab does not test agents, approvals, sandbox enforcement, signatures, update systems or TOCTOU defenses.",
    "A benign repack changes archive bytes without changing member contents; a mismatch requires explanation, not a malware label.",
]


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


def inventory(directory: Path) -> dict[str, str]:
    return {
        path.relative_to(directory).as_posix(): sha256(path.read_bytes())
        for path in sorted(directory.rglob("*"))
        if path.is_file() and ".git" not in path.relative_to(directory).parts
    }


def write_zip(path: Path, notes: bytes, *, repack: bool = False) -> None:
    """Write a deterministic text-only ZIP; repack changes only its comment."""
    path.parent.mkdir(parents=True, exist_ok=True)
    with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_STORED) as archive:
        member = zipfile.ZipInfo("notes.txt", date_time=(2026, 11, 5, 0, 0, 0))
        member.create_system = 3
        member.external_attr = 0o100644 << 16
        archive.writestr(member, notes)
        archive.comment = b"benign packaging revision two" if repack else b""


def zip_inventory(path: Path) -> dict[str, str]:
    with zipfile.ZipFile(path) as archive:
        return {
            member.filename: sha256(archive.read(member))
            for member in sorted(archive.infolist(), key=lambda item: item.filename)
            if not member.is_dir()
        }


def git(repo: Path, *args: str) -> str:
    env = os.environ.copy()
    env.update({
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_CONFIG_GLOBAL": os.devnull,
        "GIT_TERMINAL_PROMPT": "0",
        "GIT_AUTHOR_DATE": FIXED_DATE,
        "GIT_COMMITTER_DATE": FIXED_DATE,
        "GIT_AUTHOR_NAME": "Offline Review Lab",
        "GIT_COMMITTER_NAME": "Offline Review Lab",
        "GIT_AUTHOR_EMAIL": "lab@example.invalid",
        "GIT_COMMITTER_EMAIL": "lab@example.invalid",
    })
    return subprocess.run(
        ["git", "-C", str(repo), "-c", "commit.gpgsign=false", *args],
        env=env, check=True, text=True, capture_output=True,
    ).stdout.strip()


def seed(root: Path) -> tuple[Path, Path, Path]:
    repo = root / "candidate"
    installed = root / "installed-copy"
    external = root / "external-artifact" / "notes.zip"
    empty_template = root / "empty-git-template"
    for directory in (repo, installed, empty_template):
        directory.mkdir(parents=True)
    (repo / "SKILL.md").write_text(SKILL, encoding="utf-8")
    (repo / "README.md").write_text(README, encoding="utf-8")
    write_zip(repo / "archives" / "notes.zip", ORIGINAL_NOTES)
    write_zip(repo / "archives" / "alternate-notes.zip", ORIGINAL_NOTES)
    write_zip(external, ORIGINAL_NOTES)
    # Copy a single known member from the generated ZIP. Never extract an
    # arbitrary archive, run its contents, or treat this as a real installer.
    with zipfile.ZipFile(repo / "archives" / "notes.zip") as archive:
        (installed / "notes.txt").write_bytes(archive.read("notes.txt"))
    git(repo, "init", "--quiet", "--initial-branch=main", "--object-format=sha1",
        f"--template={empty_template}")
    git(repo, "config", "core.hooksPath", str(empty_template))
    git(repo, "config", "core.autocrlf", "false")
    git(repo, "add", "SKILL.md", "README.md", "archives/notes.zip", "archives/alternate-notes.zip")
    git(repo, "commit", "--quiet", "-m", "Harmless reviewed baseline")
    return repo, external, installed


def receipt(repo: Path, external: Path, installed: Path) -> dict:
    committed = repo / "archives" / "notes.zip"
    return {
        "git_commit": git(repo, "rev-parse", "HEAD"),
        "source_inventory": inventory(repo),
        "skill_sha256": sha256((repo / "SKILL.md").read_bytes()),
        "readme_sha256": sha256((repo / "README.md").read_bytes()),
        "committed_archive_sha256": sha256(committed.read_bytes()),
        "committed_member_inventory": zip_inventory(committed),
        "external_archive_sha256": sha256(external.read_bytes()),
        "external_member_inventory": zip_inventory(external),
        "installed_inventory": inventory(installed),
    }


def mutate(case: str, repo: Path, external: Path, installed: Path) -> None:
    if case == "readme_destination_changed":
        (repo / "README.md").write_text(
            README.replace("(archives/notes.zip)", "(archives/alternate-notes.zip)"),
            encoding="utf-8",
        )
        git(repo, "add", "README.md")
        git(repo, "commit", "--quiet", "-m", "Change only README destination")
    elif case == "committed_zip_changed":
        write_zip(repo / "archives" / "notes.zip", CHANGED_NOTES)
        git(repo, "add", "archives/notes.zip")
        git(repo, "commit", "--quiet", "-m", "Change committed text archive")
    elif case == "external_url_bytes_changed":
        write_zip(external, CHANGED_NOTES)
    elif case == "copied_inventory_changed":
        (installed / "notes.txt").write_bytes(CHANGED_NOTES)
    elif case == "benign_zip_repack":
        write_zip(external, ORIGINAL_NOTES, repack=True)


EXPECTED_CHANGED = {
    "unchanged": [],
    "readme_destination_changed": ["git_commit", "source_inventory", "readme_sha256"],
    "committed_zip_changed": ["git_commit", "source_inventory", "committed_archive_sha256", "committed_member_inventory"],
    "external_url_bytes_changed": ["external_archive_sha256", "external_member_inventory"],
    "copied_inventory_changed": ["installed_inventory"],
    "benign_zip_repack": ["external_archive_sha256"],
}


def run(work_dir: Path) -> dict:
    checks = []

    def check(name: str, condition: bool) -> None:
        if not condition:
            raise RuntimeError(f"Lab regression failed: {name}")
        checks.append(name)

    with tempfile.TemporaryDirectory(prefix="run-", dir=work_dir) as temporary:
        root = Path(temporary)
        baseline = receipt(*seed(root / "baseline"))
        trusted_dir = root / "trusted-review"
        trusted_dir.mkdir()
        trusted_path = trusted_dir / "expected-receipt.json"
        trusted_path.write_text(json.dumps(baseline, indent=2) + "\n", encoding="utf-8")
        trusted_hash = sha256(trusted_path.read_bytes())
        results = []
        for case in CASES:
            repo, external, installed = seed(root / case)
            mutate(case, repo, external, installed)
            expected = json.loads(trusted_path.read_text(encoding="utf-8"))
            actual = receipt(repo, external, installed)
            changed = [key for key in expected if actual[key] != expected[key]]
            controls = {
                "skill_only_matches": actual["skill_sha256"] == expected["skill_sha256"],
                "commit_only_matches": actual["git_commit"] == expected["git_commit"],
                "full_receipt_matches": actual == expected,
            }
            expected_controls = {
                "skill_only_matches": True,
                "commit_only_matches": case not in ("readme_destination_changed", "committed_zip_changed"),
                "full_receipt_matches": case == "unchanged",
            }
            check(f"{case}: exact changed fields", changed == EXPECTED_CHANGED[case])
            check(f"{case}: observed control outcomes", controls == expected_controls)
            check(f"{case}: trusted receipt unchanged", sha256(trusted_path.read_bytes()) == trusted_hash)
            check(f"{case}: committed source tree clean", git(repo, "status", "--porcelain") == "")
            results.append({
                "case": case,
                "status": "pass",
                "hypothetical_external_artifact_case": case in ("external_url_bytes_changed", "benign_zip_repack"),
                "expected_controls": expected_controls,
                "observed_controls": controls,
                "expected_changed_fields": EXPECTED_CHANGED[case],
                "actual_changed_fields": changed,
                "observed_receipt": actual,
            })
        return {
            "schema_version": 1,
            "experiment": "Harmless offline README, archive and copied-file identity-consistency lab",
            "scope": "Six synthetic cases with real local Git commits and ZIP byte reads. No network or candidate execution.",
            "python_version": platform.python_version(),
            "git_version": subprocess.run(["git", "--version"], check=True, capture_output=True, text=True).stdout.strip(),
            "lab_source_sha256": sha256(Path(__file__).read_bytes()),
            "scenario_count": len(results),
            "recorded_check_count": len(checks),
            "all_checks_passed": True,
            "trusted_expected_receipt": baseline,
            "scenarios": results,
            "checks": checks,
            "limits": LIMITS,
        }


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--work-dir", type=Path, default=Path.cwd() / ".agent-cache" / "readme-zip-review-lab")
    args = parser.parse_args()
    work_dir = args.work_dir.resolve()
    work_dir.mkdir(parents=True, exist_ok=True)
    output = work_dir / "results.json"
    output.unlink(missing_ok=True)  # A failed rerun must not leave stale success evidence.
    if shutil.which("git") is None:
        raise SystemExit("Git is required. No result was written.")
    result = run(work_dir)
    pending = work_dir / "results.pending.json"
    pending.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
    pending.replace(output)
    for scenario in result["scenarios"]:
        controls = scenario["observed_controls"]
        print(f"PASS {scenario['case']}: SKILL={controls['skill_only_matches']} "
              f"commit={controls['commit_only_matches']} full={controls['full_receipt_matches']}")
    print(f"{result['scenario_count']} scenarios; {result['recorded_check_count']} recorded checks passed.")
    print(f"Results: {output}")
    print("These results measure identity consistency, not whether reviewed files are benign.")


if __name__ == "__main__":
    main()
