#!/usr/bin/env python3
"""Strict, offline provenance primitives for installed Golden Path runtimes."""
from __future__ import annotations

import base64
import csv
import hashlib
import io
import json
import os
import stat
import sys
import zipfile
from email.parser import BytesParser
from email.policy import compat32
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence, cast


def canonical(value: object) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        allow_nan=False,
    ).encode()


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _normal(name: str) -> str:
    return "-".join(name.lower().replace("_", "-").split("-"))


def _within(path: Path, roots: Iterable[Path]) -> bool:
    resolved = path.resolve(strict=True)
    return any(resolved == root.resolve(strict=True) or resolved.is_relative_to(root.resolve(strict=True)) for root in roots)


def _regular(path: Path, label: str, *, private: bool = False) -> None:
    info = path.lstat()
    if stat.S_ISLNK(info.st_mode):
        raise ValueError(f"{label} is a symlink")
    if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.geteuid():
        raise ValueError(f"{label} is not an owned regular file")
    if private and stat.S_IMODE(info.st_mode) not in {0o600, 0o400}:
        raise ValueError(f"{label} permissions are not private")


def _private_directory(path: Path, label: str) -> None:
    info = path.lstat()
    if stat.S_ISLNK(info.st_mode):
        raise ValueError(f"{label} is a symlink")
    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid():
        raise ValueError(f"{label} is not an owned directory")
    if stat.S_IMODE(info.st_mode) not in {0o700, 0o500}:
        raise ValueError(f"{label} permissions are not private")


def read_private_json(path: Path) -> dict[str, Any]:
    _regular(path, "provenance", private=True)
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
        raise ValueError("provenance is not an object")
    return cast(dict[str, Any], value)


def _decode_record_hash(value: str, label: str) -> str:
    algorithm, separator, encoded = value.partition("=")
    if algorithm != "sha256" or not separator or not encoded:
        raise ValueError(f"{label} RECORD hash is invalid")
    try:
        raw = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
    except Exception as exc:
        raise ValueError(f"{label} RECORD hash is invalid") from exc
    if len(raw) != hashlib.sha256().digest_size:
        raise ValueError(f"{label} RECORD hash is invalid")
    return raw.hex()


def _record_rows(data: bytes, record_relative: str, label: str) -> list[tuple[str, str, int | None]]:
    try:
        decoded = data.decode("utf-8")
        parsed = list(csv.reader(io.StringIO(decoded, newline="")))
    except (UnicodeDecodeError, csv.Error) as exc:
        raise ValueError(f"{label} RECORD is invalid") from exc
    rows: list[tuple[str, str, int | None]] = []
    seen: set[str] = set()
    for row in parsed:
        if len(row) != 3:
            raise ValueError(f"{label} RECORD row is invalid")
        relative, encoded_hash, raw_size = row
        candidate = Path(relative)
        if not relative or candidate.is_absolute() or relative in seen:
            raise ValueError(f"{label} RECORD path is invalid")
        seen.add(relative)
        if relative == record_relative:
            if encoded_hash or raw_size:
                raise ValueError(f"{label} RECORD self-row is invalid")
            rows.append((relative, "", None))
            continue
        if not encoded_hash or not raw_size.isdecimal():
            raise ValueError(f"{label} RECORD hash or size is missing")
        rows.append((relative, _decode_record_hash(encoded_hash, label), int(raw_size)))
    if record_relative not in seen:
        raise ValueError(f"{label} RECORD self-row is missing")
    return rows


def _wheel_inventory(
    wheel: Path,
    dist_info_name: str,
    installed_rows: Sequence[tuple[str, str, int | None]],
    label: str,
    *,
    site_packages: Path,
    venv: Path,
) -> str:
    _regular(wheel, f"{label} wheel", private=True)
    try:
        with zipfile.ZipFile(wheel) as archive:
            names = archive.namelist()
            if len(names) != len(set(names)) or any(name.startswith("/") or ".." in Path(name).parts for name in names):
                raise ValueError(f"{label} wheel members are invalid")
            record_relative = f"{dist_info_name}/RECORD"
            wheel_record = archive.read(record_relative)
            rows = _record_rows(wheel_record, record_relative, label)
            installed_by_path = {row[0]: row for row in installed_rows}
            if set(names) != {row[0] for row in rows}:
                raise ValueError(f"{label} wheel inventory differs from wheel RECORD")
            data_prefix = f"{dist_info_name.removesuffix('.dist-info')}.data/"
            for relative, expected_hash, expected_size in rows:
                if relative == record_relative:
                    continue
                try:
                    payload = archive.read(relative)
                except KeyError as exc:
                    raise ValueError(f"{label} wheel RECORD member is missing: {relative}") from exc
                if len(payload) != expected_size:
                    raise ValueError(f"{label} wheel RECORD size mismatch: {relative}")
                if hashlib.sha256(payload).hexdigest() != expected_hash:
                    raise ValueError(f"{label} wheel RECORD hash mismatch: {relative}")
                installed_relative = relative
                if relative.startswith(data_prefix):
                    scheme, separator, suffix = relative[len(data_prefix):].partition("/")
                    if not separator or not suffix:
                        raise ValueError(f"{label} wheel data member is invalid: {relative}")
                    if scheme in {"purelib", "platlib"}:
                        installed_relative = suffix
                    elif scheme == "data":
                        installed_relative = Path(os.path.relpath(venv / suffix, site_packages)).as_posix()
                    else:
                        raise ValueError(f"{label} unsupported wheel data scheme: {scheme}")
                installed = installed_by_path.get(installed_relative)
                if installed != (installed_relative, expected_hash, expected_size):
                    raise ValueError(f"{label} installed RECORD differs for wheel member: {relative}")
            return hashlib.sha256(wheel_record).hexdigest()
    except zipfile.BadZipFile as exc:
        raise ValueError(f"{label} wheel is invalid") from exc


def inspect_installed_distribution(
    *,
    venv: Path,
    site_packages: Path,
    wheel: Path,
    distribution_name: str,
    import_packages: Sequence[str],
) -> dict[str, object]:
    """Recompute one wheel's installed inventory from METADATA and RECORD."""
    venv = venv.absolute()
    site_packages = site_packages.absolute()
    wheel = wheel.absolute()
    _private_directory(venv, "installed venv")
    if not site_packages.is_dir() or not _within(site_packages, (venv,)):
        raise ValueError("site-packages is outside the installed venv")
    matches: list[tuple[Path, bytes, str, str]] = []
    for dist_info in sorted(site_packages.glob("*.dist-info")):
        if dist_info.is_symlink() or not dist_info.is_dir():
            continue
        metadata_path = dist_info / "METADATA"
        if not metadata_path.is_file() or metadata_path.is_symlink():
            continue
        metadata = metadata_path.read_bytes()
        message = BytesParser(policy=compat32).parsebytes(metadata)
        name = str(message.get("Name", ""))
        version = str(message.get("Version", ""))
        if _normal(name) == _normal(distribution_name):
            matches.append((dist_info, metadata, name, version))
    if len(matches) != 1:
        raise ValueError(f"installed distribution cardinality is invalid: {distribution_name}")
    dist_info, metadata, declared_name, version = matches[0]
    if not version:
        raise ValueError(f"installed distribution version is unavailable: {distribution_name}")
    record = dist_info / "RECORD"
    _regular(record, f"{distribution_name} RECORD")
    record_data = record.read_bytes()
    record_relative = record.relative_to(site_packages).as_posix()
    rows = _record_rows(record_data, record_relative, distribution_name)
    inventory: list[dict[str, object]] = []
    for relative, expected_hash, expected_size in rows:
        target = site_packages / relative
        try:
            target.resolve(strict=True)
        except OSError as exc:
            raise ValueError(f"{distribution_name} RECORD file is unavailable: {relative}") from exc
        if not _within(target, (venv,)):
            raise ValueError(f"{distribution_name} RECORD path escapes the installed venv")
        _regular(target, f"{distribution_name} RECORD file")
        if relative == record_relative:
            actual_hash = hashlib.sha256(record_data).hexdigest()
            actual_size = len(record_data)
        else:
            actual_size = target.stat().st_size
            if actual_size != expected_size:
                raise ValueError(f"{distribution_name} RECORD size mismatch: {relative}")
            actual_hash = sha256_file(target)
            if actual_hash != expected_hash:
                raise ValueError(f"{distribution_name} RECORD hash mismatch: {relative}")
        inventory.append({"path": relative, "sha256": actual_hash, "size": actual_size})
    wheel_record_sha256 = _wheel_inventory(
        wheel,
        dist_info.name,
        rows,
        distribution_name,
        site_packages=site_packages,
        venv=venv,
    )
    package_roots: list[str] = []
    for package in import_packages:
        if not package or "." in package or "/" in package:
            raise ValueError("import package name is invalid")
        root = site_packages / package
        if root.is_symlink() or not root.is_dir():
            raise ValueError(f"installed package root is unavailable: {package}")
        package_roots.append(str(root.resolve(strict=True)))
    return {
        "distribution_name": declared_name,
        "distribution_version": version,
        "dist_info": str(dist_info.resolve(strict=True)),
        "metadata_sha256": hashlib.sha256(metadata).hexdigest(),
        "package_roots": package_roots,
        "record_path": str(record.resolve(strict=True)),
        "record_sha256": hashlib.sha256(record_data).hexdigest(),
        "wheel_filename": wheel.name,
        "wheel_path": str(wheel),
        "wheel_sha256": sha256_file(wheel),
        "wheel_record_sha256": wheel_record_sha256,
        "installed_inventory": inventory,
        "installed_inventory_sha256": hashlib.sha256(canonical(inventory)).hexdigest(),
    }


def validate_import_environment(
    *,
    venv: Path,
    site_packages: Path,
    import_packages: Sequence[str],
    search_path: Sequence[str] | None = None,
    executable: Path | None = None,
) -> dict[str, object]:
    """Reject another import authority ahead of the declared site-packages."""
    venv = venv.absolute()
    site_packages = site_packages.resolve(strict=True)
    declared_python = venv / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
    if not declared_python.exists():
        raise ValueError("installed interpreter is unavailable")
    actual = (executable or Path(sys.executable)).absolute()
    try:
        same = os.path.samefile(actual, declared_python)
    except OSError:
        same = False
    if not same:
        raise ValueError("running interpreter differs from declared installed interpreter")
    entries = tuple(sys.path if search_path is None else search_path)
    resolved_entries: list[Path] = []
    for entry in entries:
        candidate = Path(entry or os.getcwd()).absolute()
        if candidate.exists():
            resolved_entries.append(candidate.resolve(strict=True))
    try:
        site_index = resolved_entries.index(site_packages)
    except ValueError as exc:
        raise ValueError("declared site-packages is absent from interpreter search path") from exc
    for entry in resolved_entries[:site_index]:
        if entry == site_packages:
            continue
        if any((entry / package).exists() for package in import_packages):
            raise ValueError(f"import authority precedes declared site-packages: {entry}")
    interpreter = declared_python.resolve(strict=True)
    _regular(interpreter, "installed interpreter")
    return {
        "declared_path": str(declared_python.absolute()),
        "resolved_path": str(interpreter),
        "sha256": sha256_file(interpreter),
        "version": list(sys.version_info[:3]),
        "site_packages": str(site_packages),
    }


def loaded_module_receipts(
    modules: Mapping[str, Path], *, allowed_roots: Sequence[Path]
) -> list[dict[str, object]]:
    roots = tuple(root.resolve(strict=True) for root in allowed_roots)
    receipts: list[dict[str, object]] = []
    for name, raw_path in sorted(modules.items()):
        path = raw_path.absolute()
        if not _within(path, roots):
            raise ValueError(f"loaded module is outside declared installed roots: {name}")
        _regular(path, f"loaded module {name}")
        receipts.append(
            {
                "module": name,
                "path": str(path.resolve(strict=True)),
                "sha256": sha256_file(path),
                "size": path.stat().st_size,
            }
        )
    return receipts


def collect_installed_runtime(
    *,
    venv: Path,
    site_packages: Path,
    profile_wheel: Path,
    hermes_wheel: Path,
) -> dict[str, object]:
    interpreter = validate_import_environment(
        venv=venv,
        site_packages=site_packages,
        import_packages=("checkin_cli", "gateway"),
    )
    profile = inspect_installed_distribution(
        venv=venv,
        site_packages=site_packages,
        wheel=profile_wheel,
        distribution_name="physique-checkin-cli",
        import_packages=("checkin_cli",),
    )
    hermes = inspect_installed_distribution(
        venv=venv,
        site_packages=site_packages,
        wheel=hermes_wheel,
        distribution_name="hermes-agent",
        import_packages=("gateway",),
    )
    return {
        "schema": "installed-golden-runtime-v1",
        "venv": str(venv.absolute()),
        "site_packages": str(site_packages.resolve(strict=True)),
        "interpreter": interpreter,
        "distributions": {"profile": profile, "hermes": hermes},
    }