"""Shared file safety rules used by both tools and ACP shims.

Every guard here is defense-in-depth, NOT a security boundary: the terminal
tool runs as the same OS user and can read/write anything. The value is a
clear denial for models that respect tool errors plus a visible audit trail.
"""

from __future__ import annotations

import os
from pathlib import Path
from contextlib import suppress
from typing import Optional


def _constants_path(getter_name: str) -> Path:
    """Call ``hermes_constants.<getter_name>()`` (local import avoids cycles); ``~/.hermes`` on any failure."""
    try:
        import hermes_constants

        return getattr(hermes_constants, getter_name)()
    except Exception:
        return Path(os.path.expanduser("~/.hermes"))


def _hermes_home_path() -> Path:
    """Active HERMES_HOME (profile-aware). Tests monkeypatch this name."""
    return _constants_path("get_hermes_home")


def _hermes_root_path() -> Path:
    """Hermes root dir (parent of any profile, never per-profile)."""
    return _constants_path("get_default_hermes_root")


def _hermes_dirs() -> list[Path]:
    """Resolved active HERMES_HOME and global root, deduplicated.

    Both are checked so credential stores at <root>/... stay guarded when
    running under a profile (HERMES_HOME = <root>/profiles/<name>).
    """
    return list(dict.fromkeys(_resolve_each((_hermes_home_path(), _hermes_root_path()))))


def _resolve_each(paths) -> list[Path]:
    """``p.resolve()`` for each path, skipping ones that fail to resolve."""
    out: list[Path] = []
    for p in paths:
        with suppress(Exception):
            out.append(p.resolve())
    return out


def _is_under(resolved: str | Path, base: str | Path) -> bool:
    """True when ``resolved`` equals ``base`` or lies below it (both already resolved);
    ``Path`` inputs use ``relative_to`` (platform semantics), ``str`` a realpath prefix test."""
    if isinstance(resolved, Path):
        try:
            return resolved.relative_to(base) is not None
        except ValueError:
            return False
    return resolved == base or resolved.startswith(str(base) + os.sep)


def _resolve_target(path: str) -> Optional[Path]:
    """``Path(expanduser(path)).resolve()``, or None when resolution fails."""
    with suppress(OSError, RuntimeError):
        return Path(os.path.expanduser(str(path))).resolve()
    return None


def _guard_homes(path: str = "") -> set[str]:
    """Every home the write guards must cover. Process ``~`` alone is wrong whenever the
    process HOME is not the OS user's real home — ``TERMINAL_HOME_MODE=profile``,
    containers, and spawned workers pin ``HOME`` to ``{HERMES_HOME}/home``, which leaves
    the real home's credential paths unguarded against absolute-path writes while file
    tools happily write there (and ``_expand_tilde`` may route ``~`` to yet another
    home). Deny/approval lists are built over the union: process home, real home,
    subprocess home, and the profile home. A ``~name/...`` input resolves to a named
    account's home, which joins the set so ``~root/.ssh/authorized_keys`` stays denied."""
    homes = {os.path.expanduser("~")}
    with suppress(Exception):
        from hermes_constants import get_real_home, get_subprocess_home, _profile_home_path

        for candidate in (get_real_home(), get_subprocess_home(), _profile_home_path()):
            if candidate:
                homes.add(str(candidate))
    raw = str(path)
    if len(raw) > 1 and raw.startswith("~") and raw[1] not in "/\\":
        name = raw[1:].split("/", 1)[0].split("\\", 1)[0]
        expanded = os.path.expanduser(f"~{name}")
        if not expanded.startswith("~"):
            homes.add(expanded)
    return {os.path.realpath(h) for h in homes}


def _homes_and_resolved(path: str) -> tuple[set[str], str]:
    """``(guard homes, realpath(expanduser(path)))`` — the write-guard coordinate pair."""
    return _guard_homes(path), os.path.realpath(os.path.expanduser(str(path)))


# ---------------------------------------------------------------------------
# Windows NT-namespace path guard
#
# Pre-approval file accesses reject Windows NT-namespace (``\??\``) paths so
# the remaining unguarded path touches cannot be turned into an NTLM
# credential leak.
#
# The vector: on Windows, merely *resolving or touching* a path such as
# ``\\??\\UNC\\attacker.example\\share\\x`` (or the ``\\\\?\\UNC\\`` /
# ``GLOBALROOT`` re-entry forms) makes the OS initiate SMB authentication to
# the remote host, leaking the user's NTLM hash — even when the read itself
# fails or would later be denied. NT object-namespace paths also bypass
# normal Win32 path normalization, which lets them dodge deny-prefix checks
# built on ``realpath()`` string comparison. A model tricked by injected
# content into "reading" such a path leaks credentials before any denylist
# built on resolved paths can fire.
#
# Consequently this check MUST run on the *raw* path string before any
# ``Path.resolve()`` / ``os.path.realpath()`` call, and it does — it is the
# first check in both :func:`get_read_block_error` and the write-denial
# classifier.
#
# Scope (deliberately narrow to avoid false positives):
#   * ``\\??\\...``            — NT object-namespace paths. Never legitimate
#                                tool input on any platform.
#   * ``\\\\.\\...``           — Win32 device namespace (``\\\\.\\pipe\\``,
#                                ``\\\\.\\PhysicalDrive0``, ...). Not a file
#                                read/write target for agent tools.
#   * ``\\\\?\\UNC\\...``      — extended-length UNC form (remote host).
#   * ``\\\\?\\GLOBALROOT...`` — re-entry into the NT namespace.
#
# Plain drive-letter extended-length paths (``\\\\?\\C:\\...``) stay ALLOWED:
# they are a routine local form (see hermes_cli/windows_ssh_runtime.py) and
# carry no remote-auth trigger. Plain UNC shares (``\\\\server\\share``) are
# also unchanged here — blocking ordinary UNC reads is a policy question,
# not part of this namespace-bypass guard.
#
# The guard runs on every platform: these prefixes are never legitimate
# inputs on POSIX either, and path strings can be relayed toward Windows
# hosts (remote terminal backends, desktop bridges).
# ---------------------------------------------------------------------------

def is_nt_namespace_path(path: str) -> bool:
    """Return True if ``path`` is a Windows NT-/device-namespace path.

    Checks the raw string only — never resolves the path (resolving is the
    credential-leak trigger this guard exists to prevent).
    """
    s = str(path).replace("/", "\\")
    if s.startswith("\\??\\"):
        return True
    if s.startswith("\\\\.\\"):
        return True
    if s.startswith("\\\\?\\"):
        rest = s[4:]
        upper = rest.upper()
        if upper.startswith("UNC\\") or upper.startswith("GLOBALROOT\\"):
            return True
    return False


def get_nt_namespace_error(path: str, *, verb: str = "Access") -> Optional[str]:
    """Return an error message when ``path`` uses the NT/device namespace."""
    if not is_nt_namespace_path(path):
        return None
    return (
        f"{verb} denied: '{path}' uses a Windows NT/device namespace prefix "
        "(\\??\\, \\\\.\\, \\\\?\\UNC\\, or GLOBALROOT). These paths bypass "
        "normal path normalization and can trigger outbound SMB "
        "authentication (NTLM credential leak) merely by being resolved. "
        "Use a normal absolute path instead."
    )


def build_write_denied_paths(home: str) -> set[str]:
    """Return exact sensitive paths that must never be written."""
    # ``~/.ssh/config`` is deliberately NOT hard-denied: no key bytes, and editing
    # it is routine. It can carry ProxyCommand / Match exec, so it goes through the
    # approval gate instead (build_write_approval_paths).
    home_files = (
        (".ssh", "authorized_keys"), (".ssh", "id_rsa"), (".ssh", "id_ed25519"),
        (".netrc",), (".pgpass",), (".npmrc",), (".pypirc",), (".git-credentials",),
    )
    # Secret material under HERMES_HOME, on both the active profile and the global
    # root: overwriting the root .env leaks credentials across every profile that
    # inherits it, and the root Anthropic PKCE store is still read by default /
    # non-profile sessions when a profile is active. google_oauth.json is an OAuth
    # token store; both Bitwarden caches hold Secrets Manager material.
    #
    # auth.json, auth.lock, config.yaml and webhook_subscriptions.json are
    # deliberately NOT here: #45947 freed those control files on purpose
    # ("true containment belongs in Docker/remote backends and OS permissions,
    # not an expanding hardcoded denylist"). They stay read-denied, not write-denied.
    hermes_files = (
        ".env", ".anthropic_oauth.json",
        os.path.join("auth", "google_oauth.json"),
        os.path.join("cache", "bws_cache.json"),
        os.path.join("cache", "bws_cache.enc.json"),
    )
    paths = [
        *(os.path.join(home, *f) for f in home_files),
        *(str(base / f) for f in hermes_files for base in (_hermes_home_path(), _hermes_root_path())),
        "/etc/sudoers", "/etc/passwd", "/etc/shadow",
    ]
    return {os.path.realpath(p) for p in paths}


def build_write_denied_prefixes(home: str) -> list[str]:
    """Return sensitive directory prefixes that must never be written."""
    paths = [
        *(os.path.join(home, d) for d in (".ssh", ".aws", ".gnupg", ".kube")),
        "/etc/sudoers.d", "/etc/systemd",
        *(os.path.join(home, *d) for d in ((".docker",), (".azure",), (".config", "gh"), (".config", "gcloud"))),
    ]
    return [os.path.realpath(p) + os.sep for p in paths]


def get_safe_write_roots() -> set[str]:
    """Resolved HERMES_WRITE_SAFE_ROOT paths (``os.pathsep``-separated list)."""
    roots: set[str] = set()
    for path in filter(None, os.getenv("HERMES_WRITE_SAFE_ROOT", "").split(os.pathsep)):
        with suppress(OSError, ValueError):
            roots.add(os.path.realpath(os.path.expanduser(path)))
    return roots


def build_write_approval_paths(home: str) -> set[str]:
    """Paths that need human APPROVAL to write but are not hard-denied credentials.

    ``~/.ssh/config`` is routine to edit and holds no key bytes, but can carry
    ``ProxyCommand`` / ``Match exec``. Interactive file tools prompt
    (approve-once/session/always, like the terminal tool's ``~/.ssh`` gate);
    non-interactive callers (ACP shims, background jobs) fail closed.
    """
    return {os.path.realpath(os.path.join(home, ".ssh", "config"))}


# HERMES_HOME / root subpaths that the agent's generic file tools must not
# rewrite. Session transcripts (state.db, sessions/) are application-owned
# state whose rewrite can falsify history and break resume/compression;
# mcp-tokens/, pairing/, vault/ (key + ciphertext side by side) and
# browser-profile/ (copied cookies / Login Data) hold credential material.
# Control files (auth.json, config.yaml, webhook_subscriptions.json) are
# deliberately NOT here (#45947): read-denied, but the user may ask to edit them.
_HERMES_PROTECTED_SUBPATHS = ("state.db", "sessions", "mcp-tokens", "pairing", "vault", "browser-profile")


def _classify_write_denial(path: str, *, entry: bool = False) -> Optional[str]:
    """Return ``'credential'``, ``'safe_root'``, ``'nt_namespace'``, or ``None`` if writes are allowed.

    ``entry=True`` is for ops that unlink/rename the directory entry itself (a
    symlink, not its target): the entry — parent realpath'd, final component
    kept — is vetted as well as the target it resolves to."""
    # NT/device-namespace check runs on the RAW string, before realpath():
    # resolving such a path is itself the NTLM-leak trigger, and namespace
    # prefixes defeat string-prefix denylist comparison after normalization.
    if is_nt_namespace_path(path):
        return "nt_namespace"
    homes, resolved = _homes_and_resolved(path)

    # The runtime's own interpreter/venv is never agent-writable (an overwrite
    # bricks the next start exactly like a delete, #58748) — and this must fire
    # BEFORE the approval-gated allow so ~/.ssh-style gating cannot re-open it.
    from agent.runtime_self_protection import is_protected_path, split_entry

    if is_protected_path(path) or (entry and is_protected_path(path, follow=False)):
        return "credential"
    denial = _classify_resolved_write_denial(homes, resolved)
    if denial or not entry:
        return denial
    parent, leaf = split_entry(os.path.expanduser(str(path)))
    entry_path = os.path.join(os.path.realpath(parent or "."), leaf)
    return _classify_resolved_write_denial(homes, entry_path)


def _classify_resolved_write_denial(homes: set[str], resolved: str) -> Optional[str]:
    """Credential / protected-subpath / safe-root verdict for an already-resolved path."""
    # Approval-gated paths are allowed at this layer so interactive tools can
    # prompt; checked first so the ``.ssh/`` prefix deny doesn't swallow them.
    if any(resolved in build_write_approval_paths(home) for home in homes):
        return None

    if any(
        resolved in build_write_denied_paths(home)
        or any(resolved.startswith(prefix) for prefix in build_write_denied_prefixes(home))
        for home in homes
    ):
        return "credential"

    for base in _hermes_dirs():
        for sub in _HERMES_PROTECTED_SUBPATHS:
            with suppress(Exception):
                if _is_under(resolved, os.path.realpath(os.path.join(str(base), sub))):
                    return "credential"

    safe_roots = get_safe_write_roots()
    if safe_roots and not any(_is_under(resolved, root) for root in safe_roots):
        return "safe_root"

    return None


def is_write_denied(path: str) -> bool:
    """Return True if path is blocked by the write denylist or safe root."""
    return _classify_write_denial(path) is not None


def get_write_denied_error(path: str, *, verb: str = "Write", entry: bool = False) -> Optional[str]:
    """Return a user/model-facing error when writes to ``path`` are blocked
    (``entry``: see :func:`_classify_write_denial`)."""
    denial = _classify_write_denial(path, entry=entry)
    if denial == "safe_root":
        roots_display = os.pathsep.join(sorted(get_safe_write_roots()))
        return (
            f"{verb} denied: '{path}' is outside HERMES_WRITE_SAFE_ROOT "
            f"({roots_display}). Unset the variable or add this path's directory prefix."
        )
    if denial == "nt_namespace":
        return get_nt_namespace_error(path, verb=verb)
    return f"{verb} denied: '{path}' is a protected system/credential file." if denial else None


def is_write_approval_required(path: str) -> bool:
    """True if ``path`` is approval-gated (``~/.ssh/config``): interactive callers
    prompt, callers without a channel treat it as a block (fail closed)."""
    homes, resolved = _homes_and_resolved(path)
    return any(resolved in build_write_approval_paths(home) for home in homes)


# Secret-bearing project-local env file basenames, blocked anywhere on disk.
_BLOCKED_PROJECT_ENV_BASENAMES: set[str] = {
    ".env", ".env.local", ".env.development", ".env.production", ".env.test", ".env.staging", ".envrc",
}

_DID_SUFFIX = (
    " (Defense-in-depth — not a security boundary; the terminal tool can still bypass.)"
)

# Exact-file credential stores under HERMES_HOME / <root>. The agent never
# needs these directly — provider tools consume them through internal channels.
# bws_cache.json is the Bitwarden Secrets Manager disk cache: plaintext secret values.
_CREDENTIAL_FILE_NAMES = (
    "auth.json", "auth.lock", ".anthropic_oauth.json", ".env", "webhook_subscriptions.json",
    os.path.join("auth", "google_oauth.json"), os.path.join("cache", "bws_cache.json"),
)

# Directory-prefix read denies under HERMES_HOME / <root>: (subdir, message for
# the directory itself, message for a file inside). browser-profile/ is a copy
# of the user's Cookies / Login Data — the same credential class as auth.json.
_READ_DENIED_DIRS = (
    ("mcp-tokens",
     "is the Hermes MCP token directory and cannot be read directly.",
     "is a Hermes MCP token file and cannot be read directly."),
    ("browser-profile",
     "is the Hermes real-profile browser snapshot directory (copied cookies/logins) and cannot be read directly.",
     "is inside the Hermes real-profile browser snapshot (copied cookies/logins) and cannot be read directly."),
    # vault.key + vault.json.enc sit side by side; key + ciphertext = plaintext, so the whole dir is one credential.
    ("vault",
     "is the Hermes credential vault directory and cannot be read directly (secrets are filled server-side by browser_vault_fill).",
     "is inside the Hermes credential vault (encrypted secrets + local key) and cannot be read directly (browser_vault_fill resolves them server-side)."),
)


def get_read_block_error(path: str) -> Optional[str]:
    """Return an error message when a read targets a denied Hermes path.

    Blocked: internal skill-hub caches (prompt-injection carriers), credential
    stores under HERMES_HOME and the global root (exact files, plus anything
    under ``mcp-tokens/`` and ``browser-profile/``), and project-local ``.env``
    files anywhere on disk (``.env.example`` is the documented-shape substitute).

    Callers that resolve relative paths against a non-process cwd (e.g.
    ``TERMINAL_CWD``) MUST pass an absolute path: ``resolve()`` here anchors at
    the process cwd, so a relative ``"auth.json"`` would miss the denylist.
    """
    # NT/device-namespace check runs on the RAW string, before resolve():
    # on Windows, resolving \??\UNC\host\share (or \\?\UNC\, GLOBALROOT)
    # already triggers outbound SMB auth — the NTLM leak happens before any
    # resolved-path denylist could fire. Namespace prefixes also bypass
    # normal path normalization, defeating prefix-comparison denylists.
    nt_error = get_nt_namespace_error(path, verb="Read")
    if nt_error:
        return nt_error
    resolved = Path(path).expanduser().resolve()
    hermes_dirs = _hermes_dirs()
    reason = None
    if any(_is_under(resolved, hd / "skills" / ".hub") for hd in hermes_dirs):
        reason = (
            "is an internal Hermes cache file and cannot be read directly to prevent "
            "prompt injection. Use the skills_list or skill_view tools instead."
        )
    elif any(resolved in _resolve_each(hd / name for hd in hermes_dirs) for name in _CREDENTIAL_FILE_NAMES):
        reason = (
            "is a Hermes credential store and cannot be read directly. Provider tools "
            "consume these credentials through internal channels." + _DID_SUFFIX
        )
    else:
        for subdir, dir_msg, file_msg in _READ_DENIED_DIRS:
            for blocked_dir in _resolve_each(hd / subdir for hd in hermes_dirs):
                if _is_under(resolved, blocked_dir):
                    reason = (dir_msg if resolved == blocked_dir else file_msg) + _DID_SUFFIX
                    break
            if reason:
                break
        if reason is None and resolved.name.lower() in _BLOCKED_PROJECT_ENV_BASENAMES:
            reason = (
                "is a secret-bearing environment file and cannot be read to prevent credential "
                "leakage. If you need to check the file structure, read .env.example instead." + _DID_SUFFIX
            )
    return f"Access denied: {path} {reason}" if reason else None


def raise_if_read_blocked(path: str) -> None:
    """Raise ``ValueError`` if ``path`` is a denied Hermes read (see ``get_read_block_error``).

    Shared chokepoint for provider input-loading sites (e.g. image-gen local
    paths). Best-effort: unexpected internal errors no-op rather than break
    local-file loading; a real block still propagates.
    """
    try:
        blocked = get_read_block_error(path)
    except Exception:  # noqa: BLE001 - guard must never break local-file loading
        return
    if blocked:
        raise ValueError(blocked)


def _resolve_active_profile_name() -> str:
    """Active profile name from HERMES_HOME: ``~/.hermes`` -> ``"default"``,
    ``~/.hermes/profiles/X`` -> ``"X"``; ``"default"`` on any resolution failure."""
    try:
        parts = _hermes_home_path().resolve().relative_to(_hermes_root_path().resolve() / "profiles").parts
    except (OSError, RuntimeError, ValueError):
        return "default"
    return parts[0] if parts else "default"


# --- Sandbox-mirror write guard ---
# Non-local terminal backends bind a sandbox-local dir to the container's $HOME:
#   <HERMES_HOME>/profiles/<name>/sandboxes/<backend>/<task>/home/.hermes/...
# A host-side write there lands on a mirror the host never reads: silent success,
# divergent copies. Path-shape-only detection, independent of the active profile;
# the inner-container case (bind mount strips the prefix) is classify_container_mirror_target.

_SANDBOX_MIRROR_WARNING = (
    "Sandbox-mirror write blocked by soft guard: {target_path} "
    "sits under {mirror_root!r}, which is {body} "
    "Use the host-side tool for authoritative state (e.g. ``memory`` for memories), "
    "or address the host path directly. To bypass {bypass} with ``cross_profile=True``. "
    "(Defense-in-depth — not a security boundary; the terminal tool can still bypass.)"
)


def _mirror_info(target: Path, mirror_root: Path, inner_path: str) -> dict:
    """Common ``classify_*_mirror_target`` result shape."""
    return {"target_path": str(target), "mirror_root": str(mirror_root), "inner_path": inner_path}


def classify_sandbox_mirror_target(path: str) -> Optional[dict]:
    """Classify a write target as a sandbox-mirror of authoritative Hermes state: ``None``
    for non-mirror paths, else ``target_path`` (resolved), ``mirror_root`` (the
    ``…/home/.hermes`` prefix) and ``inner_path`` (what the agent meant on the host)."""
    target = _resolve_target(path)
    parts = target.parts if target is not None else ()
    # Need at least: sandboxes / <backend> / <task> / home / .hermes / <thing>; inner_idx = the .hermes part.
    inner_idx = next(
        (i + 4 for i, part in enumerate(parts)
         if part == "sandboxes" and i + 5 < len(parts) and parts[i + 3] == "home" and parts[i + 4] == ".hermes"),
        None,
    )
    if inner_idx is None:
        return None
    inner = str(Path(*parts[inner_idx + 1:])) if inner_idx + 1 < len(parts) else ""
    return _mirror_info(target, Path(*parts[: inner_idx + 1]), inner)


def _mirror_warning(info: Optional[dict], body: str, bypass: str) -> Optional[str]:
    """Render ``_SANDBOX_MIRROR_WARNING`` for a classify_* result (``body`` may use ``{inner_path}``)."""
    if info is None:
        return None
    return _SANDBOX_MIRROR_WARNING.format(**info, body=body.format(inner_path=info["inner_path"]), bypass=bypass)


def get_sandbox_mirror_warning(path: str) -> Optional[str]:
    """Model-facing soft-guard warning when ``path`` lands in a sandbox mirror, else ``None``;
    the caller surfaces it as a tool-result error and ``cross_profile=True`` bypasses."""
    return _mirror_warning(
        classify_sandbox_mirror_target(path),
        "a per-task mirror created by a non-local terminal backend (docker/daytona/etc.). "
        "Writes here land on a copy that the host Hermes process never reads — the "
        "authoritative file is likely {inner_path!r} under the real HERMES_HOME.",
        "this guard after explicit user direction, retry the call",
    )


def classify_container_mirror_target(path: str, mirror_prefix: str | None = None) -> Optional[dict]:
    """Classify a write target as a container-side sandbox mirror. Inside the container
    the bind mount strips the ``sandboxes/`` prefix (the agent sees plain ``/root/.hermes/…``),
    so the caller supplies ``mirror_prefix`` once it knows file tools run in a docker sandbox.
    ``None`` without a prefix or outside it, else ``target_path``/``mirror_root``/``inner_path``."""
    target, mirror = _resolve_target(path), _resolve_target(mirror_prefix) if mirror_prefix else None
    if target is None or mirror is None or not _is_under(target, mirror):
        return None
    return _mirror_info(target, mirror, target.relative_to(mirror).as_posix())


def get_container_mirror_warning(path: str, mirror_prefix: str | None = None) -> Optional[str]:
    """Model-facing soft-guard warning when ``path`` lands in the container's mirror, else ``None``."""
    return _mirror_warning(
        classify_container_mirror_target(path, mirror_prefix),
        "the container's bind-mounted home — a per-task mirror that the host Hermes "
        "process never reads. The authoritative file is {inner_path!r} under "
        "the real HERMES_HOME.",
        "after explicit user direction, retry",
    )