#!/usr/bin/env python3
"""One-shot descriptor-bound permission inventory and bounded mode repair."""
from __future__ import annotations

import datetime as dt
import fnmatch
import hashlib
import json
import os
import stat
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any

PROFILE = Path("/home/cube/.hermes/profiles/dualcoachtest")
EVIDENCE = Path("/home/cube/projects/richard/traning coach/.omo/evidence/task26/reset-controller-st_01a0054d")
CONTRACT_PATH = EVIDENCE / "schema-contract.json"
INVENTORY_RECEIPT = EVIDENCE / "live-scope-permission-inventory.json"
REPAIR_RECEIPT = EVIDENCE / "live-scope-batch-permission-repair-receipt.json"
FLAGS = os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0)


class RepairError(RuntimeError):
    pass


@dataclass
class Held:
    path: str
    fd: int
    before: os.stat_result
    kind: str
    content_sha256: str
    original_mode: int


def canonical(value: Any) -> bytes:
    return (json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + "\n").encode()


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


def identity(info: os.stat_result) -> tuple[int, ...]:
    return (info.st_dev, info.st_ino, info.st_uid, info.st_gid, info.st_nlink, info.st_size,
            info.st_mtime_ns, info.st_mode)


def invariant(info: os.stat_result) -> tuple[int, ...]:
    return (info.st_dev, info.st_ino, info.st_uid, info.st_gid, info.st_nlink, info.st_size,
            info.st_mtime_ns)


def stat_value(info: os.stat_result) -> dict[str, int | str]:
    return {"dev": info.st_dev, "inode": info.st_ino, "uid": info.st_uid, "gid": info.st_gid,
            "mode": format(stat.S_IMODE(info.st_mode), "04o"), "nlink": info.st_nlink,
            "size": info.st_size, "mtime_ns": info.st_mtime_ns}


def atomic_write(path: Path, value: dict[str, Any]) -> str:
    if path.exists() or path.is_symlink():
        raise RepairError(f"receipt already exists: {path.name}")
    raw = canonical(value)
    fd, temporary = tempfile.mkstemp(prefix=".batch-permission-", dir=path.parent)
    try:
        os.fchmod(fd, 0o600)
        offset = 0
        while offset < len(raw):
            offset += os.write(fd, raw[offset:])
        os.fsync(fd)
        os.close(fd)
        fd = -1
        os.replace(temporary, path)
        parent = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | FLAGS)
        try:
            os.fsync(parent)
        finally:
            os.close(parent)
    finally:
        if fd >= 0:
            os.close(fd)
        try:
            os.unlink(temporary)
        except FileNotFoundError:
            pass
    return sha(raw)


def open_parent(root_fd: int, relative: str) -> tuple[int, str]:
    parts = relative.split("/")
    fd = os.dup(root_fd)
    try:
        for part in parts[:-1]:
            nxt = os.open(part, os.O_RDONLY | os.O_DIRECTORY | FLAGS, dir_fd=fd)
            info = os.fstat(nxt)
            if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid():
                os.close(nxt)
                raise RepairError(f"unsafe path component: {relative}")
            os.close(fd)
            fd = nxt
        return fd, parts[-1]
    except Exception:
        os.close(fd)
        raise


def path_stat(root_fd: int, relative: str) -> os.stat_result | None:
    parent, name = open_parent(root_fd, relative)
    try:
        try:
            return os.stat(name, dir_fd=parent, follow_symlinks=False)
        except FileNotFoundError:
            return None
    finally:
        os.close(parent)


def hash_fd(fd: int, kind: str) -> str:
    if kind == "file":
        os.lseek(fd, 0, os.SEEK_SET)
        value = hashlib.sha256()
        while chunk := os.read(fd, 1024 * 1024):
            value.update(chunk)
        return value.hexdigest()
    names = sorted(os.listdir(fd))
    return sha(b"\0".join(os.fsencode(name) for name in names))


def read_json_fd(fd: int, relative: str) -> object:
    os.lseek(fd, 0, os.SEEK_SET)
    chunks: list[bytes] = []
    while chunk := os.read(fd, 1024 * 1024):
        chunks.append(chunk)
    try:
        return json.loads(b"".join(chunks))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise RepairError(f"invalid JSON: {relative}") from exc


def hold_path(root_fd: int, relative: str, device: int, held: list[Held]) -> None:
    parent, name = open_parent(root_fd, relative)
    try:
        info = os.stat(name, dir_fd=parent, follow_symlinks=False)
        if stat.S_ISLNK(info.st_mode):
            raise RepairError(f"symlink in approved scope: {relative}")
        if stat.S_ISREG(info.st_mode):
            fd = os.open(name, os.O_RDONLY | FLAGS, dir_fd=parent)
            kind = "file"
        elif stat.S_ISDIR(info.st_mode):
            fd = os.open(name, os.O_RDONLY | os.O_DIRECTORY | FLAGS, dir_fd=parent)
            kind = "directory"
        else:
            raise RepairError(f"unsupported object in approved scope: {relative}")
    finally:
        os.close(parent)
    try:
        before = os.fstat(fd)
        if before.st_dev != device:
            raise RepairError(f"device/path escape: {relative}")
        if before.st_uid != os.getuid():
            raise RepairError(f"wrong owner: {relative}")
        if kind == "file" and before.st_nlink != 1:
            raise RepairError(f"hardlinked file: {relative}")
        if kind == "file" and stat.S_IMODE(before.st_mode) & 0o111:
            raise RepairError(f"executable file in approved scope: {relative}")
        content = hash_fd(fd, kind)
        after_read = os.fstat(fd)
        path_after = path_stat(root_fd, relative)
        if path_after is None or identity(before) != identity(after_read) or identity(after_read) != identity(path_after):
            raise RepairError(f"concurrent/path change during inventory: {relative}")
        held.append(Held(relative, fd, before, kind, content, stat.S_IMODE(before.st_mode)))
        fd = -1
        if kind == "directory":
            directory_fd = held[-1].fd
            for child in sorted(os.listdir(directory_fd)):
                hold_path(root_fd, f"{relative}/{child}", device, held)
    finally:
        if fd >= 0:
            os.close(fd)


def validate_schema(held: list[Held], contract: dict[str, Any]) -> int:
    schemas = contract["json_schemas"]
    bootstrap_count = 0
    for item in held:
        if item.kind != "file" or not item.path.endswith(".json"):
            continue
        if item.path not in schemas:
            raise RepairError(f"unknown JSON schema: {item.path}")
        value = read_json_fd(item.fd, item.path)
        rule = schemas[item.path]
        if rule.get("type") == "object" and not isinstance(value, dict):
            raise RepairError(f"wrong JSON root: {item.path}")
        if not isinstance(value, dict) or not set(rule.get("required_keys", [])).issubset(value):
            raise RepairError(f"incomplete JSON schema: {item.path}")
        if "schema_value" in rule and value.get("schema") != rule["schema_value"]:
            raise RepairError(f"unknown schema value: {item.path}")
        if item.path.endswith("telegram-customer-bootstrap-v1/ledger.json"):
            sessions = value.get("sessions")
            terminal = set(contract["bootstrap_terminal_states"])
            if not isinstance(sessions, list):
                raise RepairError("bootstrap sessions is not a list")
            for session in sessions:
                if (not isinstance(session, dict) or session.get("state") not in terminal
                        or session.get("role_claims") or session.get("recovery_attempts")):
                    raise RepairError("bootstrap ledger contains nonterminal authority")
            bootstrap_count = len(sessions)
    return bootstrap_count


def effective_scopes(root_fd: int, contract: dict[str, Any]) -> list[str]:
    scopes = list(contract["approved_clear_scopes"])
    data_fd = os.open("data", os.O_RDONLY | os.O_DIRECTORY | FLAGS, dir_fd=root_fd)
    try:
        children = sorted(os.listdir(data_fd))
    finally:
        os.close(data_fd)
    protected = set(contract["protected_data_roots"])
    approved_top = {path.split("/", 1)[1].split("/", 1)[0] for path in scopes if path.startswith("data/")}
    patterns = contract["dynamic_clear_patterns"]
    for child in children:
        relative = f"data/{child}"
        if child in protected or child in approved_top:
            continue
        if any(fnmatch.fnmatchcase(relative, pattern) for pattern in patterns):
            scopes.append(relative)
            continue
        raise RepairError(f"unknown data authority root: {relative}")
    present = [scope for scope in scopes if path_stat(root_fd, scope) is not None]
    outer: list[str] = []
    for scope in sorted(set(present), key=lambda value: (value.count("/"), value)):
        if not any(scope == parent or scope.startswith(parent + "/") for parent in outer):
            outer.append(scope)
    return sorted(outer)


def verify_held(root_fd: int, held: list[Held], repaired: set[str], post: bool) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for item in held:
        current = os.fstat(item.fd)
        path_current = path_stat(root_fd, item.path)
        expected_mode = (0o600 if item.kind == "file" else 0o700) if item.path in repaired else item.original_mode
        if (path_current is None or invariant(current) != invariant(item.before)
                or invariant(path_current) != invariant(item.before)
                or (current.st_dev, current.st_ino, current.st_mode) !=
                   (path_current.st_dev, path_current.st_ino, path_current.st_mode)
                or stat.S_IMODE(current.st_mode) != expected_mode
                or hash_fd(item.fd, item.kind) != item.content_sha256):
            phase = "post-repair" if post else "pre-repair"
            raise RepairError(f"{phase} concurrent/invariant mismatch: {item.path}")
        rows.append({"path": item.path, "kind": item.kind, "content_sha256": item.content_sha256,
                     "stats": stat_value(current)})
    return rows


def main() -> int:
    for receipt in (INVENTORY_RECEIPT, REPAIR_RECEIPT):
        if receipt.exists() or receipt.is_symlink():
            raise RepairError(f"one-shot receipt already exists: {receipt.name}")
    contract_raw = CONTRACT_PATH.read_bytes()
    contract = json.loads(contract_raw)
    if contract.get("schema") != "task26-profile-reset-contract-v1":
        raise RepairError("unknown contract")
    root_fd = os.open(PROFILE, os.O_RDONLY | os.O_DIRECTORY | FLAGS)
    held: list[Held] = []
    changed: list[Held] = []
    try:
        root_info = os.fstat(root_fd)
        if root_info.st_uid != os.getuid() or stat.S_IMODE(root_info.st_mode) != 0o700:
            raise RepairError("unsafe profile root")
        scopes = effective_scopes(root_fd, contract)
        for scope in scopes:
            hold_path(root_fd, scope, root_info.st_dev, held)
        held.sort(key=lambda item: item.path)
        required = set(contract["required_current_scopes"])
        files = {item.path for item in held if item.kind == "file"}
        if missing := sorted(required - files):
            raise RepairError(f"missing required authority: {missing}")
        bootstrap_count = validate_schema(held, contract)
        repair = [item for item in held if
                  (item.kind == "file" and item.original_mode & ~0o600)
                  or (item.kind == "directory" and item.original_mode & ~0o700)]
        inventory = {
            "schema": "task26-live-scope-permission-inventory-v1", "status": "PASS",
            "recorded_at_utc": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
            "profile": str(PROFILE), "contract_sha256": sha(contract_raw), "approved_scopes": scopes,
            "excluded_protected_roots": [f"data/{name}" for name in contract["protected_data_roots"]],
            "entry_count": len(held), "bootstrap_session_count": bootstrap_count,
            "entries": [{"path": item.path, "kind": item.kind, "content_sha256": item.content_sha256,
                         "stats": stat_value(item.before)} for item in held],
            "repair_set": [{"path": item.path, "kind": item.kind,
                            "before_mode": format(item.original_mode, "04o"),
                            "after_mode": "0600" if item.kind == "file" else "0700"} for item in repair],
            "fatal_violations": [],
        }
        inventory_sha = atomic_write(INVENTORY_RECEIPT, inventory)
        repair_paths = {item.path for item in repair}
        verify_held(root_fd, held, set(), post=False)
        try:
            for item in repair:
                os.fchmod(item.fd, 0o600 if item.kind == "file" else 0o700)
                os.fsync(item.fd)
                changed.append(item)
            parent_paths = sorted({item.path.rpartition("/")[0] for item in repair})
            for parent_path in parent_paths:
                parent = root_fd if not parent_path else os.open(parent_path, os.O_RDONLY | os.O_DIRECTORY | FLAGS,
                                                                  dir_fd=root_fd)
                try:
                    os.fsync(parent)
                finally:
                    if parent != root_fd:
                        os.close(parent)
            post_rows = verify_held(root_fd, held, repair_paths, post=True)
        except Exception:
            for item in reversed(changed):
                current = path_stat(root_fd, item.path)
                if current is not None and (current.st_dev, current.st_ino) == (item.before.st_dev, item.before.st_ino):
                    os.fchmod(item.fd, item.original_mode)
                    os.fsync(item.fd)
            os.fsync(root_fd)
            raise
        changed_post = {row["path"]: row for row in post_rows}
        aggregate = {
            "schema": "task26-live-scope-batch-permission-repair-v1", "status": "PASS",
            "recorded_at_utc": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
            "inventory_receipt_sha256": inventory_sha, "contract_sha256": sha(contract_raw),
            "repair_count": len(repair), "bootstrap_session_count": bootstrap_count,
            "repairs": [{"path": item.path, "kind": item.kind, "content_sha256": item.content_sha256,
                         "before": stat_value(item.before), "after": changed_post[item.path]["stats"],
                         "proof": {"content_hash_unchanged": True, "size_unchanged": True,
                                   "inode_dev_unchanged": True, "uid_gid_unchanged": True,
                                   "mtime_unchanged": True, "name_unchanged": True,
                                   "descriptor_path_match": True}}
                        for item in repair],
            "rollback_boundary": "all changed modes roll back through retained descriptors before receipt on mismatch",
            "protected_paths_touched": 0, "bytes_changed": 0,
            "remaining_permission_violations": [],
        }
        receipt_sha = atomic_write(REPAIR_RECEIPT, aggregate)
        print(json.dumps({"status": "PASS", "inventory_entries": len(held), "repair_count": len(repair),
                          "repair_paths": sorted(repair_paths), "inventory_receipt_sha256": inventory_sha,
                          "repair_receipt_sha256": receipt_sha}, sort_keys=True))
        return 0
    finally:
        for item in held:
            os.close(item.fd)
        os.close(root_fd)


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