"""Approval-gated read-only MacBook access over SSH.

The tool exposes fixed actions only. It never accepts or forwards raw shell
commands from the model.
"""

from __future__ import annotations

import json
import logging
import os
import shlex
import subprocess
import time
from pathlib import PurePosixPath
from typing import Any

from hermes_cli.config import cfg_get, load_config
from hermes_constants import get_hermes_home
from tools.registry import registry

logger = logging.getLogger(__name__)

_DEFAULT_BLOCKED_SUBPATHS = (
    ".ssh",
    ".aws",
    ".config",
    ".gnupg",
    "Library/Keychains",
    "Library/Messages",
    "Library/Mail",
    "Library/Cookies",
    "Library/Containers",
    "Library/Group Containers",
)
_DEFAULT_BLOCKED_NAME_FRAGMENTS = (
    ".env",
    "token",
    "secret",
    "credential",
    "password",
    "passwd",
    "apikey",
    "api_key",
    "private",
    "id_rsa",
    "id_ed25519",
    ".pem",
    ".key",
    ".p12",
    ".mobileprovision",
)
_TEXT_EXTENSIONS = {
    ".txt", ".md", ".markdown", ".rst", ".log", ".csv", ".tsv",
    ".json", ".jsonl", ".yaml", ".yml", ".toml", ".ini", ".cfg",
    ".py", ".js", ".ts", ".tsx", ".jsx", ".sh", ".zsh", ".bash",
    ".css", ".html", ".xml", ".sql",
}
_ACTIONS_REQUIRING_APPROVAL = {"list_dir", "search_files", "read_text_file", "recent_files"}


class MacbookRemoteError(ValueError):
    """Expected user-facing validation failure."""


def _config() -> dict[str, Any]:
    config = load_config()
    return dict(config.get("macbook_remote", {}) or {})


def _enabled_config(cfg: dict[str, Any] | None = None) -> bool:
    cfg = _config() if cfg is None else cfg
    return bool(cfg.get("enabled")) and bool(cfg.get("host")) and bool(cfg.get("user"))


def _check_requirements() -> bool:
    try:
        return _enabled_config()
    except Exception:
        return False


def _as_int(cfg: dict[str, Any], key: str, default: int, *, minimum: int, maximum: int) -> int:
    try:
        value = int(cfg.get(key, default))
    except (TypeError, ValueError):
        value = default
    return max(minimum, min(maximum, value))


def _normalize_abs_posix(path: str, cfg: dict[str, Any] | None = None) -> str:
    raw = (path or "").strip()
    if not raw:
        raise MacbookRemoteError("path is required")
    if "\x00" in raw:
        raise MacbookRemoteError("path contains a NUL byte")
    if raw == "~" or raw.startswith("~/"):
        user = str((cfg or {}).get("user") or "").strip()
        if not user:
            raise MacbookRemoteError("macbook_remote.user is required to expand '~'")
        raw = f"/Users/{user}" + (raw[1:] if raw != "~" else "")
    elif not raw.startswith("/"):
        user = str((cfg or {}).get("user") or "").strip()
        if not user:
            raise MacbookRemoteError("path must be absolute")
        raw = f"/Users/{user}/{raw}"
    p = PurePosixPath(raw)
    if not p.is_absolute():
        raise MacbookRemoteError("path must be absolute")
    if any(part in {"..", ""} for part in p.parts[1:]):
        raise MacbookRemoteError("path traversal is not allowed")
    return str(p)


def _allowed_roots(cfg: dict[str, Any]) -> list[str]:
    roots = cfg.get("allowed_roots") or []
    normalized = []
    for root in roots:
        try:
            normalized.append(_normalize_abs_posix(str(root), cfg))
        except MacbookRemoteError:
            logger.warning("Ignoring invalid macbook_remote allowed root: %r", root)
    return sorted(set(normalized), key=len, reverse=True)


def _is_under(path: str, root: str) -> bool:
    return path == root or path.startswith(root.rstrip("/") + "/")


def _blocked_subpaths(cfg: dict[str, Any]) -> list[str]:
    values = cfg.get("blocked_subpaths") or _DEFAULT_BLOCKED_SUBPATHS
    return [str(v).strip("/") for v in values if str(v).strip("/")]


def _blocked_name_fragments(cfg: dict[str, Any]) -> list[str]:
    values = cfg.get("blocked_name_fragments") or _DEFAULT_BLOCKED_NAME_FRAGMENTS
    return [str(v).lower() for v in values if str(v)]


def _validate_path(path: str, cfg: dict[str, Any], *, for_read: bool = False) -> str:
    normalized = _normalize_abs_posix(path, cfg)
    roots = _allowed_roots(cfg)
    if not roots:
        raise MacbookRemoteError("macbook_remote.allowed_roots is empty")
    if not any(_is_under(normalized, root) for root in roots):
        raise MacbookRemoteError("path is outside configured MacBook allowed_roots")

    lower = normalized.lower()
    for root in roots:
        if not _is_under(normalized, root):
            continue
        rel = normalized[len(root):].strip("/")
        rel_lower = rel.lower()
        for blocked in _blocked_subpaths(cfg):
            blocked_lower = blocked.lower().strip("/")
            if rel_lower == blocked_lower or rel_lower.startswith(blocked_lower + "/"):
                raise MacbookRemoteError("path is under a blocked MacBook directory")

    if any(fragment in lower for fragment in _blocked_name_fragments(cfg)):
        raise MacbookRemoteError("path or filename matches a blocked secret pattern")
    if for_read and PurePosixPath(normalized).suffix.lower() not in _TEXT_EXTENSIONS:
        raise MacbookRemoteError("read_text_file only allows known text file extensions")
    return normalized


def _ssh_base(cfg: dict[str, Any]) -> list[str]:
    host = str(cfg.get("host") or "").strip()
    user = str(cfg.get("user") or "").strip()
    if not host or not user:
        raise MacbookRemoteError("macbook_remote.host and macbook_remote.user are required")
    target = f"{user}@{host}"
    cmd = [
        "ssh",
        "-o", "BatchMode=yes",
        "-o", f"ConnectTimeout={_as_int(cfg, 'connect_timeout_seconds', 8, minimum=1, maximum=60)}",
    ]
    key = str(cfg.get("ssh_key") or "").strip()
    if key and key.lower() != "default":
        cmd.extend(["-i", os.path.expanduser(key)])
    port = cfg.get("port")
    if port:
        cmd.extend(["-p", str(port)])
    cmd.append(target)
    return cmd


def _run_remote_python(cfg: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]:
    script = r'''
import fnmatch, json, os, sys, time
from pathlib import Path

payload = json.loads(sys.argv[1])
action = payload.get("action")
target = payload.get("path") or payload.get("root") or ""
allowed_roots = [Path(p).expanduser().resolve(strict=False) for p in payload.get("allowed_roots", [])]
blocked_subpaths = [str(p).strip("/").lower() for p in payload.get("blocked_subpaths", [])]
blocked_fragments = [str(p).lower() for p in payload.get("blocked_name_fragments", [])]
max_depth = int(payload.get("max_depth", 3))
max_results = int(payload.get("max_results", 100))
read_limit = int(payload.get("read_text_limit_bytes", 8000))


def fail(message):
    print(json.dumps({"ok": False, "error": message}))
    raise SystemExit(0)


def check_path(value, *, require_file=False):
    try:
        p = Path(value).expanduser().resolve(strict=False)
    except Exception as exc:
        fail(f"invalid path: {exc}")
    if not any(p == root or root in p.parents for root in allowed_roots):
        fail("resolved path is outside allowed roots")
    low = str(p).lower()
    if any(fragment in low for fragment in blocked_fragments):
        fail("resolved path matches blocked secret pattern")
    for root in allowed_roots:
        if not (p == root or root in p.parents):
            continue
        try:
            rel = p.relative_to(root)
        except ValueError:
            continue
        rel_low = str(rel).lower()
        for blocked in blocked_subpaths:
            if rel_low == blocked or rel_low.startswith(blocked + "/"):
                fail("resolved path is under a blocked directory")
    if require_file and not p.is_file():
        fail("path is not a regular file")
    return p


def item(path):
    st = path.stat()
    return {
        "path": str(path),
        "name": path.name,
        "type": "dir" if path.is_dir() else "file" if path.is_file() else "other",
        "size": st.st_size,
        "mtime": int(st.st_mtime),
        "mtime_iso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(st.st_mtime)),
    }

try:
    if action == "health":
        print(json.dumps({
            "ok": True,
            "hostname": os.uname().nodename,
            "whoami": os.getenv("USER") or "",
            "cwd": os.getcwd(),
        }))
    elif action == "list_dir":
        p = check_path(target)
        if not p.is_dir():
            fail("path is not a directory")
        rows = []
        for child in sorted(p.iterdir(), key=lambda c: c.name.lower()):
            try:
                checked = check_path(str(child))
                rows.append(item(checked))
            except SystemExit:
                continue
            except Exception:
                continue
            if len(rows) >= max_results:
                break
        print(json.dumps({"ok": True, "items": rows, "truncated": len(rows) >= max_results}))
    elif action == "search_files":
        root = check_path(payload.get("root") or target)
        if not root.is_dir():
            fail("root is not a directory")
        pattern = payload.get("pattern") or "*"
        rows = []
        base_depth = len(root.parts)
        for current, dirs, files in os.walk(root):
            current_path = check_path(current)
            depth = len(current_path.parts) - base_depth
            if depth >= max_depth:
                dirs[:] = []
            kept_dirs = []
            for d in dirs:
                try:
                    check_path(str(current_path / d))
                    kept_dirs.append(d)
                except SystemExit:
                    pass
                except Exception:
                    pass
            dirs[:] = kept_dirs
            for name in files:
                candidate = current_path / name
                try:
                    checked = check_path(str(candidate), require_file=True)
                except SystemExit:
                    continue
                except Exception:
                    continue
                if fnmatch.fnmatch(name.lower(), str(pattern).lower()):
                    rows.append(item(checked))
                    if len(rows) >= max_results:
                        print(json.dumps({"ok": True, "items": rows, "truncated": True}))
                        raise SystemExit(0)
        rows.sort(key=lambda row: row["mtime"], reverse=True)
        print(json.dumps({"ok": True, "items": rows[:max_results], "truncated": len(rows) > max_results}))
    elif action == "recent_files":
        root = check_path(payload.get("root") or target)
        if not root.is_dir():
            fail("root is not a directory")
        rows = []
        base_depth = len(root.parts)
        for current, dirs, files in os.walk(root):
            try:
                current_path = check_path(current)
            except SystemExit:
                dirs[:] = []
                continue
            depth = len(current_path.parts) - base_depth
            if depth >= max_depth:
                dirs[:] = []
            kept_dirs = []
            for d in dirs:
                try:
                    check_path(str(current_path / d))
                    kept_dirs.append(d)
                except SystemExit:
                    pass
                except Exception:
                    pass
            dirs[:] = kept_dirs
            for name in files:
                candidate = current_path / name
                try:
                    checked = check_path(str(candidate), require_file=True)
                    rows.append(item(checked))
                except Exception:
                    continue
        rows.sort(key=lambda row: row["mtime"], reverse=True)
        print(json.dumps({"ok": True, "items": rows[:max_results], "truncated": len(rows) > max_results}))
    elif action == "read_text_file":
        p = check_path(target, require_file=True)
        data = p.read_bytes()[:read_limit]
        print(json.dumps({"ok": True, "path": str(p), "bytes_returned": len(data), "text": data.decode("utf-8", errors="replace")}))
    else:
        fail("unsupported action")
except SystemExit:
    raise
except Exception as exc:
    fail(str(exc))
'''
    remote = "python3 -c " + shlex.quote(script) + " " + shlex.quote(json.dumps(payload))
    cmd = _ssh_base(cfg) + [remote]
    timeout = _as_int(cfg, "command_timeout_seconds", 20, minimum=1, maximum=120)
    result = subprocess.run(cmd, text=True, capture_output=True, timeout=timeout, check=False)
    if result.returncode != 0:
        return {"ok": False, "error": result.stderr.strip() or result.stdout.strip() or f"ssh exited {result.returncode}"}
    try:
        return json.loads(result.stdout.strip().splitlines()[-1])
    except (json.JSONDecodeError, IndexError) as exc:
        return {"ok": False, "error": f"invalid remote response: {exc}"}


def _audit(action: str, path: str, ok: bool, error: str = "") -> None:
    try:
        log_dir = get_hermes_home() / "logs"
        log_dir.mkdir(parents=True, exist_ok=True)
        record = {
            "ts": int(time.time()),
            "action": action,
            "path": path,
            "ok": ok,
            "error": error[:500],
        }
        with (log_dir / "macbook_remote_audit.jsonl").open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(record, ensure_ascii=False) + "\n")
    except Exception:
        logger.debug("Failed to write MacBook remote audit log", exc_info=True)


def _approval_description(action: str, path: str, cfg: dict[str, Any]) -> str:
    return (
        f"MacBook {action} for {path} under allowed roots "
        f"{', '.join(_allowed_roots(cfg))}; max_depth={_as_int(cfg, 'max_depth', 3, minimum=0, maximum=8)}, "
        f"max_results={_as_int(cfg, 'max_results', 100, minimum=1, maximum=500)}. "
        "Blocked secret/key/mail/message paths remain denied."
    )


def _approval_pattern_key(path: str, cfg: dict[str, Any]) -> str:
    # Approve at the configured root boundary.  The remote tool remains read-only
    # and still enforces blocked secret/key/mail/message paths inside that root.
    for root in _allowed_roots(cfg):
        if _is_under(path, root):
            return f"macbook_remote:read:{root}"
    return f"macbook_remote:read:{path}"


def macbook_remote(
    action: str,
    path: str = "",
    root: str = "",
    pattern: str = "*",
    task_id: str | None = None,
) -> str:
    """Run one fixed read-only MacBook action over SSH."""
    cfg = _config()
    if not _enabled_config(cfg):
        return json.dumps({"ok": False, "error": "macbook_remote is not enabled or is missing host/user config"})

    action = (action or "").strip()
    if action not in {"health", "list_dir", "search_files", "read_text_file", "recent_files"}:
        return json.dumps({"ok": False, "error": "unsupported macbook_remote action"})

    max_depth = _as_int(cfg, "max_depth", 3, minimum=0, maximum=8)
    max_results = _as_int(cfg, "max_results", 100, minimum=1, maximum=500)
    read_limit = _as_int(cfg, "read_text_limit_bytes", 8000, minimum=1, maximum=50000)

    try:
        if action == "health":
            target_path = ""
        elif action == "search_files":
            target_path = _validate_path(root or path, cfg)
        else:
            target_path = _validate_path(path or root, cfg, for_read=(action == "read_text_file"))
    except MacbookRemoteError as exc:
        _audit(action, path or root, False, str(exc))
        return json.dumps({"ok": False, "error": str(exc)})

    if action in _ACTIONS_REQUIRING_APPROVAL:
        from tools.approval import request_tool_approval

        approval = request_tool_approval(
            action=f"macbook_remote:{action}:{target_path}",
            description=_approval_description(action, target_path, cfg),
            pattern_key=_approval_pattern_key(target_path, cfg),
            allow_permanent=True,
            allow_session=True,
        )
        if not approval.get("approved"):
            _audit(action, target_path, False, approval.get("message", "approval denied"))
            return json.dumps({"ok": False, "error": approval.get("message") or "approval denied"})

    payload = {
        "action": action,
        "path": target_path,
        "root": target_path,
        "pattern": pattern or "*",
        "allowed_roots": _allowed_roots(cfg),
        "blocked_subpaths": _blocked_subpaths(cfg),
        "blocked_name_fragments": _blocked_name_fragments(cfg),
        "max_depth": max_depth,
        "max_results": max_results,
        "read_text_limit_bytes": read_limit,
    }
    result = _run_remote_python(cfg, payload)
    _audit(action, target_path, bool(result.get("ok")), str(result.get("error") or ""))
    return json.dumps(result, ensure_ascii=False)


registry.register(
    name="macbook_remote",
    toolset="macbook_remote",
    schema={
        "name": "macbook_remote",
        "description": (
            "Approval-gated read-only MacBook SSH tool. Use only fixed actions; "
            "never for arbitrary shell commands. Requires explicit user approval "
            "for file and directory access."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "action": {
                    "type": "string",
                    "enum": ["health", "list_dir", "search_files", "read_text_file", "recent_files"],
                    "description": "Fixed read-only action to run on the MacBook.",
                },
                "path": {
                    "type": "string",
                    "description": (
                        "MacBook path for list_dir/read_text_file/recent_files. "
                        "Absolute /Users/... paths are preferred; ~/Downloads or Downloads are expanded to the configured MacBook user home."
                    ),
                },
                "root": {
                    "type": "string",
                    "description": (
                        "MacBook directory root for search_files or recent_files. "
                        "Absolute /Users/... paths are preferred; ~/Downloads or Downloads are expanded to the configured MacBook user home."
                    ),
                },
                "pattern": {
                    "type": "string",
                    "description": "Filename glob for search_files, e.g. '*.pdf'.",
                },
            },
            "required": ["action"],
        },
    },
    handler=lambda args, **kw: macbook_remote(
        action=args.get("action", ""),
        path=args.get("path", ""),
        root=args.get("root", ""),
        pattern=args.get("pattern", "*"),
        task_id=kw.get("task_id"),
    ),
    check_fn=_check_requirements,
    description="Approval-gated read-only MacBook SSH access",
    emoji="💻",
    max_result_size_chars=20000,
)