"""Three findings closed:

* ``//etc`` (and any leading ``//``) bypassed the protected-
  path blocklist because ``os.path.normpath`` preserves a leading double
  slash on POSIX, while the kernel collapses ``//`` to ``/`` at lookup. Now
  ``_normalize_host_path`` collapses leading-slash runs before the
  blocklist comparison.

* ``_is_allowed_image`` accepted shell metacharacters in
  the tag (``gr00t:;rm``, ``gr00t:$(rm)``) because the allowlist match
  was ``startswith(prefix)`` only. Now the image name must also pass a
  ``[A-Za-z0-9._:/@-]`` charset gate.

* ``_ke_matches`` treated ``**`` literally, so the default
  subscribe allowlist (``**/presence``, ``**/health``, ``**/safety/**``)
  matched no real Zenoh keys. Now leading ``**/``, trailing ``/**``, and
  both-ends double-star patterns are expanded properly.
"""

from __future__ import annotations

import importlib
from typing import Any

import pytest

gi: Any = importlib.import_module("strands_robots.tools.gr00t_inference")
rm: Any = importlib.import_module("strands_robots.tools.robot_mesh")


# ─── Finding A: leading '//' bypass ──────────────────────────────────────────


@pytest.mark.parametrize(
    "host_path",
    [
        "//etc",
        "//etc/shadow",
        "//root/.ssh",
        "//root/.ssh/id_rsa",
        "//var/run/docker.sock",
        "//run/docker.sock",
        "//proc/1/environ",
        "//home/user/.aws/credentials",
        "///etc",  # triple slash too
        "//etc/../etc/shadow",
    ],
)
def test_double_slash_paths_blocked_by_volume_safety(host_path: str) -> None:
    """A leading ``//`` (or more) must NOT bypass the protected-path blocklist."""
    assert gi._check_volume_safety({host_path: "/x"}) is not None, (
        f"{host_path!r} should be blocked but slipped through _check_volume_safety"
    )


@pytest.mark.parametrize(
    "host_path",
    [
        "//etc",
        "//root/.ssh",
        "//var/run/docker.sock",
        "///etc/shadow",
    ],
)
def test_double_slash_paths_blocked_by_hf_local_dir_safety(host_path: str) -> None:
    """``hf_local_dir`` reaches host fs directly; same blocklist must apply."""
    assert gi._check_hf_local_dir_safety(host_path) is not None, (
        f"{host_path!r} should be blocked but slipped through _check_hf_local_dir_safety"
    )


def test_legitimate_paths_still_allowed() -> None:
    """Regression: don't over-block legitimate operator mounts."""
    assert gi._check_volume_safety({"/data/checkpoints": "/data/checkpoints"}) is None
    assert gi._check_volume_safety({"/opt/models": "/models"}) is None
    assert gi._check_hf_local_dir_safety("/data/checkpoints") is None
    assert gi._check_hf_local_dir_safety(None) is None


# ─── Finding B: image-name shell-meta in tag ─────────────────────────────────


@pytest.mark.parametrize(
    "img",
    [
        "gr00t:;rm -rf /",
        "gr00t:$(rm)",
        "gr00t:; whoami",
        "gr00t:$IFS",
        "gr00t:`whoami`",
        "gr00t: foo",  # whitespace
        "gr00t:tag\nrm",
        "gr00t:tag|cat",
        "gr00t:tag&ls",
    ],
)
def test_image_name_with_shell_metachars_rejected(img: str) -> None:
    """Even though argv-mode subprocess prevents shell exec, reject defence-in-depth."""
    assert gi._is_allowed_image(img) is False, f"{img!r} contains shell metacharacters and must not be allowlisted"


@pytest.mark.parametrize(
    "img",
    [
        "gr00t:latest",
        "gr00t:1.7-rc2",
        "gr00t:abc123",
        "nvcr.io/nvidia/isaac-gr00t:25.10",
        "nvcr.io/nvidia/isaac-gr00t:latest",
    ],
)
def test_legitimate_image_names_still_allowed(img: str) -> None:
    """Regression: real GR00T tags must continue to pass."""
    assert gi._is_allowed_image(img) is True


# ─── Finding C: '**/' default allowlist matches reality ──────────────────────


@pytest.mark.parametrize(
    "key",
    [
        # Leading ``**/`` should match zero or more leading segments
        "presence",
        "robot1/presence",
        "a/b/c/presence",
        "health",
        "robot1/health",
        # Trailing ``/**`` AND leading ``**/`` (``**/safety/**``) should match
        # safety segment anywhere in the key
        "robot1/safety/event",
        "a/b/safety/estop",
        "safety/heartbeat",
        "x/y/z/safety/foo/bar",
    ],
)
def test_default_subscribe_allowlist_matches_real_keys(monkeypatch, key: str) -> None:
    """Default allowlist (``**/presence``, ``**/health``, ``**/safety/**``)
    must match realistic Zenoh key expressions, not just the patterns
    themselves."""
    monkeypatch.delenv("STRANDS_MESH_SUBSCRIBE_ALLOW", raising=False)
    rm._reset_subscribe_allowlist_cache()
    assert rm._is_allowed_subscribe_target(key) is True, f"key {key!r} should match the default subscribe allowlist"


@pytest.mark.parametrize(
    "key",
    [
        # Random keys that don't match presence/health/safety classes
        "robot1/cmd_vel",
        "robot1/joint_states",
        "robot1/camera/raw",
        "fleet/policy/lora_weights",
        "secrets/apikey",
    ],
)
def test_default_subscribe_allowlist_still_rejects_other_keys(monkeypatch, key: str) -> None:
    """Regression: default allowlist must NOT match arbitrary keys."""
    monkeypatch.delenv("STRANDS_MESH_SUBSCRIBE_ALLOW", raising=False)
    rm._reset_subscribe_allowlist_cache()
    assert rm._is_allowed_subscribe_target(key) is False, (
        f"key {key!r} should NOT match the default subscribe allowlist"
    )


def test_ke_matches_specific_shapes() -> None:
    """Direct unit tests for the shapes we explicitly support."""
    # exact equality
    assert rm._ke_matches("foo/bar", "foo/bar") is True
    # leading **/ - zero or more segments before
    assert rm._ke_matches("**/presence", "presence") is True
    assert rm._ke_matches("**/presence", "a/presence") is True
    assert rm._ke_matches("**/presence", "a/b/c/presence") is True
    assert rm._ke_matches("**/presence", "presence/x") is False
    # trailing /** - prefix or any deeper segments
    assert rm._ke_matches("a/safety/**", "a/safety") is True
    assert rm._ke_matches("a/safety/**", "a/safety/event") is True
    assert rm._ke_matches("a/safety/**", "a/safety/x/y") is True
    assert rm._ke_matches("a/safety/**", "b/safety") is False
    # both ends - middle anywhere in the key
    assert rm._ke_matches("**/safety/**", "safety") is True
    assert rm._ke_matches("**/safety/**", "a/safety") is True
    assert rm._ke_matches("**/safety/**", "safety/x") is True
    assert rm._ke_matches("**/safety/**", "a/b/safety/x/y") is True
    assert rm._ke_matches("**/safety/**", "a/safetyfoo") is False
    # segment-level *
    assert rm._ke_matches("strands/*/stream", "strands/peer-b/stream") is True
    assert rm._ke_matches("strands/*/stream", "strands/a/b/stream") is False
    # empty target
    assert rm._ke_matches("**/x", "") is False
